"""Progress tracker - persists chunk status across sessions.""" import json from dataclasses import asdict from pathlib import Path from .decomposer import TaskChunk TRACKER_DIR = ".claude-chunks" def get_project_dir(project_root: Path) -> Path: d = project_root / TRACKER_DIR d.mkdir(exist_ok=True) return d def save_plan(project_root: Path, task: str, chunks: list[TaskChunk], codebase_summary: str) -> Path: """Save a decomposed plan to disk.""" d = get_project_dir(project_root) plan = { "task": task, "codebase_summary": codebase_summary, "chunks": [asdict(c) for c in chunks], } plan_file = d / "plan.json" plan_file.write_text(json.dumps(plan, indent=2)) # Also write individual prompt files for easy copy-paste prompts_dir = d / "prompts" prompts_dir.mkdir(exist_ok=True) for chunk in chunks: prompt_file = prompts_dir / f"chunk_{chunk.id:02d}.md" prompt_file.write_text(chunk.prompt(codebase_summary, len(chunks))) return plan_file def load_plan(project_root: Path) -> dict | None: """Load an existing plan from disk.""" plan_file = project_root / TRACKER_DIR / "plan.json" if not plan_file.exists(): return None return json.loads(plan_file.read_text()) def mark_done(project_root: Path, chunk_id: int) -> bool: """Mark a chunk as completed.""" plan = load_plan(project_root) if not plan: return False for chunk in plan["chunks"]: if chunk["id"] == chunk_id: chunk["status"] = "done" plan_file = project_root / TRACKER_DIR / "plan.json" plan_file.write_text(json.dumps(plan, indent=2)) return True return False def get_next_chunk(project_root: Path) -> dict | None: """Get the next pending chunk whose dependencies are satisfied.""" plan = load_plan(project_root) if not plan: return None done_ids = {c["id"] for c in plan["chunks"] if c["status"] == "done"} for chunk in plan["chunks"]: if chunk["status"] != "pending": continue deps = set(chunk.get("depends_on", [])) if deps.issubset(done_ids): return chunk return None def get_status_summary(project_root: Path) -> str: """Get a human-readable status summary.""" plan = load_plan(project_root) if not plan: return "No active plan found. Run `ctc plan` first." chunks = plan["chunks"] done = sum(1 for c in chunks if c["status"] == "done") total = len(chunks) task = plan["task"] lines = [ f"Task: {task}", f"Progress: {done}/{total} chunks complete", "", ] for c in chunks: status_icon = {"done": "[x]", "pending": "[ ]", "skipped": "[-]"}.get(c["status"], "[ ]") deps = "" if c.get("depends_on"): deps = f" (after: {', '.join(str(d) for d in c['depends_on'])})" lines.append(f" {status_icon} {c['id']}. {c['title']}{deps}") if done < total: nxt = get_next_chunk(project_root) if nxt: lines.extend(["", f"Next up: Chunk {nxt['id']} - {nxt['title']}", f" Prompt file: {TRACKER_DIR}/prompts/chunk_{nxt['id']:02d}.md"]) return "\n".join(lines)