ai-agent/backend/plans.py

"""Claude *plans* — structured implementation plans the assistant writes with the
``plan`` skill, surfaced in the dashboard's files view.

Plans live in ``PLANS_DIR`` (the repo's root ``data/plans`` tree, mounted
read-only). Each file is a markdown doc with a flat YAML-ish frontmatter block::

    ---
    title: <one line>
    created: <YYYY-MM-DD>
    status: proposed | approved | in-progress | done
    sessionId: <planning session id>
    conversationIds: <comma-separated session ids, optional>
    projects: <comma list, optional>
    services: <comma list, optional>
    estCost: <usd>            # estimated cost to IMPLEMENT (forward estimate)
    estTimeMinutes: <int>
    estTokens: <int>
    files: <count>
    steps: <count>
    ---

    # <title>
    ## Summary … ## Files changed (table) … ## Steps (numbered list)

The plan files are intentionally **not** git-tracked — they live only on the
machine (same as the rest of the root ``data/`` tree). We parse the frontmatter
(no PyYAML dependency — the shape is flat and simple, mirroring ``memories.py``),
pull the *Files changed* table and *Steps* list out of the body for a clean
summary, and expose everything to the frontend. The conversation-metric **join**
(estimated-vs-actual cost/time) is done in ``main.py`` where the transcript
summaries live; this module stays pure parsing.
"""
from __future__ import annotations

import datetime
import os
import pathlib
import re

PLANS_DIR = pathlib.Path(os.environ.get("PLANS_DIR", "/workspace/data/plans")).resolve()

# `projects/<slug>` / `services/<slug>` mentions in the body → related repos,
# same convention the conversation + memory parsers use.
_PROJECT_RE = re.compile(r"(?:^|[/\s(`\"'])projects/([A-Za-z0-9._][\w.-]*)")
_SERVICE_RE = re.compile(r"(?:^|[/\s(`\"'])services/([A-Za-z0-9._][\w.-]*)")

MAX_BODY = 60_000  # cap a plan body so detail responses stay bounded


def _iso(epoch: float) -> str:
    return (datetime.datetime.fromtimestamp(epoch, datetime.timezone.utc)
            .isoformat().replace("+00:00", "Z"))


def _parse_frontmatter(text: str) -> tuple[dict, str]:
    """Split a ``---`` frontmatter block from the body. Returns (data, body).

    Supports one level of nesting (a ``metadata:`` map), which is all these files
    use. Values are unquoted; missing frontmatter yields ({}, text)."""
    if not text.startswith("---"):
        return {}, text
    end = text.find("\n---", 3)
    if end == -1:
        return {}, text
    block = text[3:end].strip("\n")
    body = text[end + 4:].lstrip("\n")
    data: dict = {}
    cur = data
    for line in block.splitlines():
        if not line.strip():
            continue
        indent = len(line) - len(line.lstrip())
        key, sep, val = line.strip().partition(":")
        if not sep:
            continue
        key = key.strip()
        val = val.strip().strip('"').strip("'")
        if indent == 0:
            if val == "":
                cur = data.setdefault(key, {})  # start a nested map
            else:
                data[key] = val
                cur = data
        elif isinstance(cur, dict):
            cur[key] = val
    return data, body


def _list_field(fm: dict, key: str) -> list[str]:
    """A comma/space-separated frontmatter value → a clean list of tokens."""
    raw = fm.get(key)
    if not raw:
        return []
    return [t.strip() for t in re.split(r"[,\s]+", str(raw)) if t.strip()]


def _num(fm: dict, key: str, cast):
    v = fm.get(key)
    if v is None or v == "":
        return None
    try:
        return cast(str(v).replace("$", "").replace(",", "").strip())
    except (TypeError, ValueError):
        return None


def _strip_fences(body: str) -> str:
    """Drop fenced ``` code blocks so a template/example inside a plan doesn't
    get mistaken for the real Files-changed table / Steps list."""
    return re.sub(r"(?ms)^[ \t]*```.*?^[ \t]*```[ \t]*$", "", body)


def _section(body: str, heading: str) -> str:
    """The text under a ``## <heading>`` up to the next ``## `` heading."""
    m = re.search(rf"(?im)^\s*#{{1,6}}\s*{re.escape(heading)}\s*$", body)
    if not m:
        return ""
    rest = body[m.end():]
    nxt = re.search(r"(?im)^\s*#{1,6}\s+\S", rest)
    return rest[:nxt.start()] if nxt else rest


def _parse_files(body: str) -> list[dict]:
    """Rows of the ``## Files changed`` markdown table → [{file, change}]."""
    out: list[dict] = []
    for line in _section(body, "Files changed").splitlines():
        line = line.strip()
        if not line.startswith("|"):
            continue
        cells = [c.strip() for c in line.strip("|").split("|")]
        if len(cells) < 2:
            continue
        head = cells[0].lower()
        if head in ("file", "path") or set(cells[0]) <= {"-", ":", " "}:
            continue  # header row / separator
        out.append({"file": cells[0], "change": cells[1]})
    return out


def _parse_steps(body: str) -> list[dict]:
    """Top-level numbered/bulleted items under ``## Steps`` → [{title, detail}].

    An item spans until the next top-level item: wrapped continuation lines and
    nested sub-bullets are folded into it, so the whole point survives — not
    just its first line. Each item is then split on its first em/en-dash so
    ``**Title** — detail`` becomes {title, detail}; falls back to the whole
    item as the title."""
    items: list[list[str]] = []
    for line in _section(body, "Steps").splitlines():
        m = re.match(r"^\s{0,1}(?:\d+[.)]|[-*])\s+(.*)$", line)
        if m:
            items.append([m.group(1).strip()])
        elif items and line.strip():
            items[-1].append(re.sub(r"^[-*]\s+", "", line.strip()))
    out: list[dict] = []
    for lines in items:
        parts = re.split(r"\s+[—–-]\s+", " ".join(lines), maxsplit=1)
        title = re.sub(r"\*\*|`", "", parts[0]).strip()
        detail = re.sub(r"\*\*|`", "", parts[1]).strip() if len(parts) > 1 else ""
        if title:
            out.append({"title": title, "detail": detail})
    return out


def _record(path: pathlib.Path, with_body: bool = False) -> dict:
    slug = path.stem
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
        mtime = path.stat().st_mtime
    except OSError:
        text, mtime = "", 0.0
    fm, body = _parse_frontmatter(text)

    prose = _strip_fences(body)
    files = _parse_files(prose)
    steps = _parse_steps(prose)
    projects = _list_field(fm, "projects") or sorted(
        {m for m in _PROJECT_RE.findall(body) if not m.startswith("-")})
    services = _list_field(fm, "services") or sorted(
        {m for m in _SERVICE_RE.findall(body) if not m.startswith("-")})

    rec = {
        "slug": slug,
        "title": fm.get("title") or slug,
        "status": (fm.get("status") or "proposed").lower(),
        "created": fm.get("created"),
        "updatedAt": _iso(mtime) if mtime else None,
        "sessionId": fm.get("sessionId"),
        "conversationIds": _list_field(fm, "conversationIds"),
        "projects": projects,
        "services": services,
        "estCost": _num(fm, "estCost", float),
        "estTimeMinutes": _num(fm, "estTimeMinutes", int),
        "estTokens": _num(fm, "estTokens", int),
        "files": _num(fm, "files", int) if fm.get("files") else len(files),
        "steps": _num(fm, "steps", int) if fm.get("steps") else len(steps),
    }
    if with_body:
        rec["fileList"] = files
        rec["stepList"] = steps
        rec["content"] = body[:MAX_BODY] + (
            f"\n\n… [{len(body) - MAX_BODY} more chars truncated]"
            if len(body) > MAX_BODY else "")
    return rec


def _iter_files():
    if not PLANS_DIR.is_dir():
        return
    for p in sorted(PLANS_DIR.glob("*.md")):
        yield p


def list_plans() -> list[dict]:
    """All plan summaries, newest-updated first."""
    items = [_record(p) for p in _iter_files()]
    items.sort(key=lambda m: m.get("updatedAt") or "", reverse=True)
    return items


def _safe_path(slug: str) -> pathlib.Path | None:
    name = pathlib.PurePosixPath(slug).name
    if not name:
        return None
    p = (PLANS_DIR / f"{name}.md").resolve()
    try:
        p.relative_to(PLANS_DIR)
    except ValueError:
        return None
    return p if p.is_file() else None


def plan_detail(slug: str) -> dict | None:
    p = _safe_path(slug)
    if not p:
        return None
    return _record(p, with_body=True)


def linked_session_ids(plan: dict) -> list[str]:
    """Every conversation this plan references (planning session + extra ids)."""
    ids: list[str] = []
    for sid in [plan.get("sessionId"), *plan.get("conversationIds", [])]:
        if sid and sid not in ids:
            ids.append(sid)
    return ids