"""Markdown artifact writer for session documents."""
from pathlib import Path
from pathfinder_mcp.session import SessionManager
RESEARCH_TEMPLATE = """# Research: {task}
## Task Description
{task}
## Findings
"""
PLAN_TEMPLATE = """---
name: {name}
overview: {overview}
todos: []
---
# {name}
## Architecture Overview
<!-- High-level description of the system/workflow -->
## Core Components
```
<!-- File structure diagram -->
```
## Phase 1: Setup & Scaffold
### 1.1 Tasks
- [ ] Task 1
- [ ] Task 2
## Phase 2: Core Implementation
### 2.1 Tasks
- [ ] Task 1
- [ ] Task 2
## Implementation Details
<!-- Technical specifications, data structures -->
## Success Criteria Checklist
- [ ] Criterion 1
- [ ] Criterion 2
"""
PROGRESS_TEMPLATE = """# Implementation Progress
## Completed Phases
## Current Phase
## Notes
"""
class ArtifactWriter:
"""Writes markdown artifacts to session directory."""
def __init__(self, session_manager: SessionManager | None = None):
self.session_manager = session_manager or SessionManager()
def _get_artifact_path(self, session_id: str, filename: str) -> Path:
"""Get path for an artifact file."""
session_path = self.session_manager.get_session_path(session_id)
session_path.mkdir(parents=True, exist_ok=True)
return session_path / filename
def write_research(self, session_id: str, content: str, task: str = "") -> Path:
"""Write or append to research.md."""
path = self._get_artifact_path(session_id, "research.md")
if not path.exists():
# Initialize with template
initial = RESEARCH_TEMPLATE.format(task=task or "Research Task")
path.write_text(initial + content)
else:
# Append to existing
existing = path.read_text()
path.write_text(existing + "\n" + content)
return path
def write_plan(self, session_id: str, content: str) -> Path:
"""Write plan.md content."""
path = self._get_artifact_path(session_id, "plan.md")
path.write_text(content)
return path
def write_progress(self, session_id: str, content: str) -> Path:
"""Write or append to progress.md."""
path = self._get_artifact_path(session_id, "progress.md")
if not path.exists():
path.write_text(PROGRESS_TEMPLATE + content)
else:
existing = path.read_text()
path.write_text(existing + "\n" + content)
return path
def read_artifact(self, session_id: str, artifact_name: str) -> str | None:
"""Read an artifact file. Returns None if not found."""
path = self._get_artifact_path(session_id, artifact_name)
if not path.exists():
return None
return path.read_text()
def get_plan_template(self, name: str = "Plan", overview: str = "") -> str:
"""Return structured plan.md template."""
return PLAN_TEMPLATE.format(
name=name, overview=overview or "Implementation plan"
)
def artifact_exists(self, session_id: str, artifact_name: str) -> bool:
"""Check if artifact exists."""
path = self._get_artifact_path(session_id, artifact_name)
return path.exists()