ai-agent/backend/ctas.py

"""Conversation CTAs: the prompt buttons a finished conversation offers.

The replacement for the auto-firing conversation hooks (``hooks.py``, removed
in favour of this). A hook was a scheduler that resumed *every* finished run
behind your back, ten seconds after the ``DONE`` marker — which meant paying to
re-create the whole transcript as fresh input tokens for a follow-up nobody
asked for (a resume never hits the finished run's prompt cache: new OS process
⇒ new environment preamble ⇒ cache-prefix mismatch). A CTA is the same prompt,
minus the scheduler: the viewer renders one button per CTA under the last
message, and *you* decide.

    click       → the CTA's prompt is loaded into the conversation's resume box
                  (edit it, add a note, then send) — nothing runs yet.
    long-press  → the prompt is fired straight at the session in the background
                  (``POST /api/ctas/{id}/run``), no round-trip through the UI.

The same CTAs are also selectable as ``cta``-kind tags in the composers, so a
CTA's prompt can be folded into any message you type rather than only into the
one the button builds.

Storage is a single JSON sidecar (``/data/ctas.json``), same discipline as
``cron.py`` / ``meta.py`` — one lock, atomic saves, mtime reload so the other
container during a blue-green cutover is picked up before any decision:

    { "ctas": { "<ctaId>": {
        "id":         "complete",
        "name":       "Complete",
        "label":      "Complete",              # button text (defaults to name)
        "description": "Post-process …",       # tooltip / tag description
        "icon":        "sparkles",             # lucide name, resolved client-side
        "promptFile": ".claude/agents/cta/complete.md",
        "enabled":    true,
        "order":      0,                       # button order, low → left
        "createdAt":  iso,
        "lastStatus": "ok" | "error: …",
        "history":    { iso: sessionId, … },   # background runs, newest last
      } } }

**How a CTA runs is declared in its prompt file's frontmatter** (``harness``,
``model``, ``effort``, ``thinking`` — see ``cron.run_config``), exactly like a
cron job's prompt, so the run config lives next to the prompt it configures.
Unset fields mean "keep whatever the session already ran with", which is what a
follow-up on a finished conversation almost always wants.

Unlike a hook there is **no loop guard and no once-per-session ledger** — a CTA
only ever runs because someone pressed it, so re-running one is a feature. The
firings are still stamped on ``meta.ctas[ctaId]`` (newest wins) so the viewer
can show which CTAs a conversation has already been through.
"""

import json
import os
import pathlib
import threading
from datetime import datetime

import cron as cron_mod

# CTA prompts live in their own subdirectory of the cron prompt dir, so a CTA
# prompt is never mistaken for a schedulable agent and vice versa.
PROMPT_DIR = f"{cron_mod.PROMPT_DIR}/cta"

# Re-exported so main.py has one import for the prompt-file helpers.
strip_frontmatter = cron_mod.strip_frontmatter
run_config = cron_mod.run_config
frontmatter_block = cron_mod.frontmatter_block
slugify = cron_mod.slugify


def _now_iso() -> str:
    return datetime.now().astimezone().strftime("%Y-%m-%dT%H:%M:%S%z")


def valid_prompt_file(rel: str) -> str:
    """Normalize + validate a CTA's prompt-file path: a ``.md`` directly under
    ``.claude/agents/cta/``. Raises ValueError otherwise."""
    rel = (rel or "").strip().lstrip("/")
    p = pathlib.PurePosixPath(rel)
    if ".." in p.parts:
        raise ValueError("promptFile must not contain '..'")
    if p.suffix.lower() != ".md":
        raise ValueError("promptFile must be a .md file")
    if str(p.parent) != PROMPT_DIR:
        raise ValueError(f"promptFile must live in {PROMPT_DIR}/")
    return str(p)


# ── the seeded CTA ───────────────────────────────────────────────────────────
# "Complete" is what the old post-task hook used to do on a timer, now behind a
# button — and doing more, because it can afford to: the algorithmic half of the
# post-processing (metadata, the squashed diff, the merged tool/bash/task
# summaries) is scaffolded by `conv-scaffold` instead of being re-derived by the
# model from the transcript.
DEFAULT_CTAS = {
    "complete": {
        "id": "complete",
        "name": "Complete",
        "label": "Complete",
        "description": ("Post-process this conversation: scaffold its summary "
                        "(metadata, squashed diff, merged tool summaries), "
                        "write the analysis, then tidy up."),
        "icon": "sparkles",
        "promptFile": f"{PROMPT_DIR}/complete.md",
        "enabled": True,
        "order": 0,
        "createdAt": None,   # stamped at seed time
        "lastStatus": None,
        "history": {},
    },
}

# The prompt the seeded CTA sends. Written to the workspace on first run (only
# if the file doesn't exist), so a fresh install has a working Complete button
# with no setup.
#
# The heavy reading is delegated to a **subagent**: the scaffold is already a
# compressed view of the transcript, so the analysis pass doesn't need this
# conversation's context — and keeping it out of the parent means Complete
# costs a fraction of what the old hook's full-transcript resume did.
DEFAULT_COMPLETE_PROMPT = """---
name: complete
description: Post-process this conversation — scaffold the summary (metadata, squashed diff, merged tool/bash/task summaries), write the final analysis, then tidy up.
---

Complete this conversation. Two steps, in order.

**1. Scaffold + summary.** Run `conv-scaffold` — it writes a markdown scaffold
of *this* conversation (metadata and tags, the squashed diff of everything that
changed, merged bash/tool/task summaries) and prints its path. Then spawn **one
subagent** whose whole job is that file: it reads the scaffold, fills in the
`## Summary` and `## Analysis` sections (what was built, the key decisions, what
is worth remembering, what is still open), and publishes it with
`conv-scaffold publish <path>`. Everything it needs is in the scaffold — it
does not need this conversation's history.

**2. Tidy up.** While the subagent works, do the housekeeping this task earned —
only what genuinely applies, and keep it short:

- **Memory.** Anything non-obvious here — a gotcha, a wrong turn, a constraint
  invisible in the code? One fact per file in your memory directory, plus its
  line in `MEMORY.md`. Skip what the repo already records.
- **Docs.** If the change altered how something is used, built or deployed,
  update the affected `README.md` / `CLAUDE.md` / `GOAL.md`. Describe the
  current state — no changelog prose.
- **Artefacts.** Copy anything worth keeping (screenshots, diagrams, reports)
  into the canonical project's `.ai/artefacts/<sessionId>/`. Leave scratch behind.
- **Cleanup.** Remove worktrees you created once their branch is merged
  (`scripts/new-worktree.sh -r <path>`), and kill background processes, dev
  servers or standby containers you no longer need.
- **Changelog.** If the project keeps one, add the user-facing entry.

Do **not** re-notify (no `notify-done`) and do not start new feature work.

End your reply with `COMPLETED` on its own line (not `DONE` — this pass has its
own marker so the viewer can show that the task was completed, not just finished).
"""


class CtaStore:
    """The JSON store. Mirrors ``CronStore``/``HookStore`` minus the queue —
    a CTA has no schedule and nothing pending, it fires when pressed."""

    def __init__(self, path: str):
        self.path = pathlib.Path(path)
        self._lock = threading.Lock()
        self._mtime: float | None = None
        self._ctas: dict[str, dict] = {}
        self._load()
        if self._mtime is None:  # no file yet → seed the defaults
            with self._lock:
                self._ctas = json.loads(json.dumps(DEFAULT_CTAS))
                for c in self._ctas.values():
                    c["createdAt"] = _now_iso()
                self._save()

    # ── persistence ──────────────────────────────────────────────────────────
    def _load(self) -> None:
        try:
            raw = json.loads(self.path.read_text(encoding="utf-8")) or {}
            self._ctas = raw.get("ctas") or {}
            self._mtime = self.path.stat().st_mtime
        except (OSError, ValueError):
            self._ctas = {}
            self._mtime = None

    def _save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".json.tmp")
        tmp.write_text(json.dumps({"ctas": self._ctas}, indent=2, sort_keys=True),
                       encoding="utf-8")
        os.replace(tmp, self.path)
        try:
            self._mtime = self.path.stat().st_mtime
        except OSError:
            pass

    def _reload_if_changed(self) -> None:
        try:
            m = self.path.stat().st_mtime
        except OSError:
            return
        if m != self._mtime:
            self._load()

    # ── CRUD ─────────────────────────────────────────────────────────────────
    def list(self) -> "list[dict]":
        """Every CTA, in button order (``order`` asc, then name)."""
        with self._lock:
            self._reload_if_changed()
            out = json.loads(json.dumps(list(self._ctas.values())))
        return sorted(out, key=lambda c: (int(c.get("order") or 0),
                                          (c.get("name") or "").lower()))

    def get(self, cta_id: str) -> dict | None:
        with self._lock:
            self._reload_if_changed()
            c = self._ctas.get(cta_id)
            return json.loads(json.dumps(c)) if c else None

    def create(self, name: str, *, label: str = "", description: str = "",
               icon: str = "", prompt_file: str, enabled: bool = True,
               order: int | None = None) -> dict:
        prompt_file = valid_prompt_file(prompt_file)
        with self._lock:
            self._reload_if_changed()
            base = slugify(name)
            cta_id, n = base, 2
            while cta_id in self._ctas:
                cta_id, n = f"{base}-{n}", n + 1
            if order is None:
                order = 1 + max((int(c.get("order") or 0)
                                 for c in self._ctas.values()), default=-1)
            cta = {"id": cta_id, "name": name or cta_id,
                   "label": label or name or cta_id,
                   "description": description or "", "icon": icon or "",
                   "promptFile": prompt_file, "enabled": bool(enabled),
                   "order": int(order), "createdAt": _now_iso(),
                   "lastStatus": None, "history": {}}
            self._ctas[cta_id] = cta
            self._save()
            return json.loads(json.dumps(cta))

    def update(self, cta_id: str, patch: dict) -> dict | None:
        if patch.get("promptFile") is not None:
            patch["promptFile"] = valid_prompt_file(patch["promptFile"])
        if patch.get("order") is not None:
            patch["order"] = int(patch["order"])
        with self._lock:
            self._reload_if_changed()
            cta = self._ctas.get(cta_id)
            if not cta:
                return None
            for k in ("name", "label", "description", "icon", "promptFile",
                      "enabled", "order"):
                if patch.get(k) is not None:
                    cta[k] = patch[k]
            self._save()
            return json.loads(json.dumps(cta))

    # NB: both annotations are strings — the ``list`` method above shadows the
    # builtin inside the class body, so a bare ``list[str]`` would evaluate
    # against it and raise at import time.
    def reorder(self, ids: "list[str]") -> "list[dict]":
        """Renumber ``order`` to match the given id sequence (unknown ids are
        ignored, CTAs left out keep their relative order after the listed ones)."""
        with self._lock:
            self._reload_if_changed()
            rank = {cid: i for i, cid in enumerate(ids) if cid in self._ctas}
            tail = len(rank)
            for cta in sorted(self._ctas.values(),
                              key=lambda c: int(c.get("order") or 0)):
                if cta["id"] in rank:
                    cta["order"] = rank[cta["id"]]
                else:
                    cta["order"] = tail
                    tail += 1
            self._save()
        return self.list()

    def delete(self, cta_id: str) -> bool:
        with self._lock:
            self._reload_if_changed()
            if cta_id not in self._ctas:
                return False
            del self._ctas[cta_id]
            self._save()
            return True

    def record_run(self, cta_id: str, session_id: str,
                   status: str = "ok") -> None:
        """Append ``{now: sessionId}`` to the CTA's history and stamp the
        outcome (what the settings card's status chip shows)."""
        with self._lock:
            self._reload_if_changed()
            cta = self._ctas.get(cta_id)
            if not cta:
                return
            if session_id:
                hist = dict(cta.get("history") or {})
                hist[_now_iso()] = session_id
                if len(hist) > 200:      # keep the most recent 200 firings
                    for k in sorted(hist)[:len(hist) - 200]:
                        del hist[k]
                cta["history"] = hist
            cta["lastStatus"] = status
            self._save()