#!/usr/bin/env python3 """ Context Snapshot Generator - Process Refinement System

Generate minimal context snapshots (<500 tokens) for efficient session handoffs and state preservation.

Usage: python context_snapshot.py python context_snapshot.py --save python context_snapshot.py --token-budget 300

Author: CODITECT Process Refinement Version: 1.0.0 Created: 2026-01-02 """

import argparse import json import re from datetime import datetime, timezone from pathlib import Path from typing import Optional

def get_project_root() -> Path: """Get the project root directory.""" current = Path(file).resolve().parent.parent while current != current.parent: if (current / "CLAUDE.md").exists(): return current current = current.parent return Path(file).resolve().parent.parent

PROJECT_ROOT = get_project_root()

ADR-114 & ADR-118: Use centralized path discovery

import sys sys.path.insert(0, str(Path(file).parent / "core")) try: from paths import get_context_storage_dir, CONTEXT_STORAGE except ImportError: # Fallback for backward compatibility _user_data = Path.home() / "PROJECTS" / ".coditect-data" / "context-storage" if _user_data.exists(): CONTEXT_STORAGE = _user_data else: CONTEXT_STORAGE = PROJECT_ROOT / "context-storage"

SNAPSHOTS_DIR = CONTEXT_STORAGE / "snapshots"

def ensure_directories(): """Ensure required directories exist.""" SNAPSHOTS_DIR.mkdir(parents=True, exist_ok=True)

def estimate_tokens(text: str) -> int: """Estimate token count (rough: ~4 chars per token).""" return len(text) // 4

def get_track_summary() -> dict: """Get compressed summary of track progress.""" pilot_plan = PROJECT_ROOT / "internal" / "project" / "plans" / "PILOT-PARALLEL-EXECUTION-PLAN.md"

summary = {}

if pilot_plan.exists():
    content = pilot_plan.read_text()

    # Extract track progress
    for track in "ABCDEFG":
        completed = len(re.findall(rf'\[x\]\s*{track}\.\d+', content, re.IGNORECASE))
        total = len(re.findall(rf'\[.\]\s*{track}\.\d+', content))
        if total > 0:
            summary[track] = {
                "done": completed,
                "total": total,
                "pct": round(completed / total * 100)
            }

return summary

def get_recent_completions(limit: int = 5) -> list[str]: """Get most recent completed tasks.""" pilot_plan = PROJECT_ROOT / "internal" / "project" / "plans" / "PILOT-PARALLEL-EXECUTION-PLAN.md" completions = []

if pilot_plan.exists():
    content = pilot_plan.read_text()
    # Find completed tasks
    matches = re.findall(r'\[x\]\s*([A-G]\.\d+\.?\d*)\s*[-–]\s*([^\n]+)', content, re.IGNORECASE)
    completions = [f"{m[0]}: {m[1][:40]}" for m in matches[-limit:]]

return completions

def get_next_priorities(limit: int = 3) -> list[str]: """Get next priority tasks.""" pilot_plan = PROJECT_ROOT / "internal" / "project" / "plans" / "PILOT-PARALLEL-EXECUTION-PLAN.md" priorities = []

if pilot_plan.exists():
    content = pilot_plan.read_text()
    # Find incomplete tasks
    matches = re.findall(r'\[ \]\s*([A-G]\.\d+\.?\d*)\s*[-–]\s*([^\n]+)', content)
    priorities = [f"{m[0]}: {m[1][:40]}" for m in matches[:limit]]

return priorities

def get_active_blockers() -> list[str]: """Get any documented blockers.""" # Would integrate with blocker tracking return []

def generate_snapshot(token_budget: int = 500) -> dict: """Generate context snapshot within token budget.""" snapshot = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M"), "tracks": get_track_summary(), "recent": [], "next": [], "blockers": [] }

# Add items until we approach budget
current_tokens = estimate_tokens(json.dumps(snapshot))

# Add recent completions
for item in get_recent_completions(5):
    test_snapshot = snapshot.copy()
    test_snapshot["recent"] = snapshot["recent"] + [item]
    if estimate_tokens(json.dumps(test_snapshot)) < token_budget * 0.8:
        snapshot["recent"].append(item)
    else:
        break

# Add priorities
for item in get_next_priorities(5):
    test_snapshot = snapshot.copy()
    test_snapshot["next"] = snapshot["next"] + [item]
    if estimate_tokens(json.dumps(test_snapshot)) < token_budget:
        snapshot["next"].append(item)
    else:
        break

# Add blockers if room
blockers = get_active_blockers()
for blocker in blockers:
    test_snapshot = snapshot.copy()
    test_snapshot["blockers"] = snapshot["blockers"] + [blocker]
    if estimate_tokens(json.dumps(test_snapshot)) < token_budget:
        snapshot["blockers"].append(blocker)
    else:
        break

snapshot["_tokens"] = estimate_tokens(json.dumps(snapshot))

return snapshot

def format_markdown(snapshot: dict) -> str: """Format snapshot as markdown.""" md = f"""# Context Snapshot

Generated: {snapshot['ts']} Tokens: ~{snapshot['_tokens']} (target: <500)

Track Progress

""" for track, data in snapshot.get("tracks", {}).items(): md += f"- {track}: {data['pct']}% ({data['done']}/{data['total']})\n"

if snapshot.get("recent"):
    md += "\n## Recent Completions\n\n"
    for item in snapshot["recent"]:
        md += f"- {item}\n"

if snapshot.get("next"):
    md += "\n## Next Priorities\n\n"
    for i, item in enumerate(snapshot["next"], 1):
        md += f"{i}. {item}\n"

if snapshot.get("blockers"):
    md += "\n## Blockers\n\n"
    for blocker in snapshot["blockers"]:
        md += f"- ⚠️ {blocker}\n"

return md

def save_snapshot(snapshot: dict) -> Path: """Save snapshot to file.""" timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") filename = SNAPSHOTS_DIR / f"snapshot-{timestamp}.json"

with open(filename, "w") as f:
    json.dump(snapshot, f, indent=2)

return filename

def main(): parser = argparse.ArgumentParser(description="Context Snapshot Generator") parser.add_argument("--token-budget", type=int, default=500, help="Target token limit") parser.add_argument("--save", action="store_true", help="Save snapshot to file") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--markdown", action="store_true", help="Output as markdown")

args = parser.parse_args()

ensure_directories()

snapshot = generate_snapshot(args.token_budget)

if args.save:
    filepath = save_snapshot(snapshot)
    print(f"Snapshot saved to: {filepath}")

if args.json:
    print(json.dumps(snapshot, indent=2))
elif args.markdown:
    print(format_markdown(snapshot))
else:
    # Default: markdown format
    print(format_markdown(snapshot))

if name == "main": main()

Track Progress​

Track Progress