92002f2090
The `plan` skill now expresses a plan's open questions, propositions and clarifications as an ask-form field spec in a fenced json block under `## Questions`. The backend parses it (reusing forms.valid_fields), the plan page renders it with the same controls the conversation forms use, answers are kept in localStorage per plan, and Submit & work drops the implement-this-plan prompt into the composer with the decisions appended.
294 lines
10 KiB
Python
294 lines
10 KiB
Python
"""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)
|
|
## Questions -> a ```json fence holding an ask-form field spec
|
|
|
|
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 json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
|
|
import forms
|
|
|
|
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
|
|
|
|
|
|
_JSON_FENCE_RE = re.compile(r"(?ms)^[ \t]*```(?:json[a-z0-9]*)?[ \t]*\n(.*?)^[ \t]*```")
|
|
|
|
|
|
def _parse_questions(body: str) -> dict | None:
|
|
"""The ``## Questions`` section's fenced JSON block → an ask-form spec.
|
|
|
|
The `plan` skill writes a plan's open questions / propositions /
|
|
clarifications as a form spec (the same field shape the ``ask-form`` skill
|
|
posts to ``/api/forms``) inside a ```json fence under ``## Questions``. The
|
|
plan page renders it as a real form whose answers are folded into the
|
|
"implement this plan" prompt — so a plan's unknowns get decided in one
|
|
click instead of in prose.
|
|
|
|
Returns ``{title, description, fields: [...]}`` or ``None`` when the
|
|
section is missing, isn't valid JSON, or carries no usable field."""
|
|
section = _section(body, "Questions")
|
|
if not section.strip():
|
|
return None
|
|
m = _JSON_FENCE_RE.search(section)
|
|
if not m:
|
|
return None
|
|
try:
|
|
raw = json.loads(m.group(1))
|
|
except (ValueError, TypeError):
|
|
return None
|
|
if isinstance(raw, list): # a bare field list is allowed
|
|
raw = {"fields": raw}
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
try:
|
|
fields = forms.valid_fields(raw.get("fields") or [])
|
|
except ValueError:
|
|
return None
|
|
return {
|
|
"title": str(raw.get("title") or "Open questions").strip(),
|
|
"description": str(raw.get("description") or "").strip(),
|
|
"fields": fields,
|
|
}
|
|
|
|
|
|
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)
|
|
questions = _parse_questions(body) # needs the fences — the spec IS one
|
|
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),
|
|
"questions": len(questions["fields"]) if questions else 0,
|
|
}
|
|
if with_body:
|
|
rec["fileList"] = files
|
|
rec["stepList"] = steps
|
|
rec["questionForm"] = questions
|
|
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
|