ai-agent/backend/meta.py

"""
Per-conversation *action* metadata, kept in a single JSON sidecar
(``/data/conversations-meta.json``) keyed by Claude Code session id:

    { "<sessionId>": {
        "title":      "…"|null,               # self-chosen title, overrides the
                                              # parsed one (first user message /
                                              # Claude's ai-title)
        "projects":   ["foo", "bar"],         # projects worked on (manual)
        "state":      "running"|"finished"|…, # explicit override (optional)
        "exitedAt":   iso|null,               # when the run's process died
                                              # (exit watcher; guards the
                                              # exit-time "finished" stamp)
        "committed":  iso|null,               # last time work was committed
        "pushed":     iso|null,
        "merged":     iso|null,
        "deployed":   iso|null,
        "notified":   iso|null,
        "notifications": [ {title, body, type, url, at}, … ],
        "hooks":      { "<hookId>": iso },    # conversation-done hooks that
                                              # already fired on this session
                                              # (the once-per-hook loop guard —
                                              # see hooks.py)
    } }

This file is the source of truth the skill scripts edit (``conv-meta.sh`` /
``select-project.sh``); the backend reloads it on change and merges it into the
conversation list/detail responses. Parsed-from-transcript fields (cost, tokens,
``byTool``, inferred projects, auto state) live in the SQLite summary instead —
the two are merged at request time.
"""

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

# Action fields a conversation's metadata may carry (besides projects/state).
STAMP_FIELDS = ("committed", "pushed", "merged", "deployed", "notified")

# Two notifications recorded within this many seconds with identical
# title/body/type/url are treated as the same push (a double-send / retry in a
# single turn) and collapsed. Genuine later repeats fall outside the window.
NOTIF_DEDUP_WINDOW_S = 120


def _notif_key(n: dict) -> tuple:
    """Content identity of a notification, ignoring its timestamp."""
    return (n.get("title") or "", n.get("body") or "",
            n.get("type") or "", n.get("url") or "")


def _is_dupe(note: dict, existing: list[dict]) -> bool:
    """True if ``note`` duplicates a recent entry in ``existing`` — same content
    (title/body/type/url) within ``NOTIF_DEDUP_WINDOW_S`` seconds."""
    key = _notif_key(note)
    at = note.get("at") or ""
    for prev in reversed(existing):
        if _notif_key(prev) != key:
            continue
        # Identical content: dupe if timestamps are missing or close together.
        pat = prev.get("at") or ""
        if not at or not pat:
            return True
        try:
            delta = abs(
                (datetime.fromisoformat(at) - datetime.fromisoformat(pat)).total_seconds()
            )
            if delta <= NOTIF_DEDUP_WINDOW_S:
                return True
        except ValueError:
            return True
    return False


def empty() -> dict:
    return {
        "title": None,
        "projects": [],
        "state": None,
        "committed": None, "pushed": None, "merged": None,
        "deployed": None, "notified": None,
        "notifications": [],
        "worktrees": [],
        "archived": False,
        # {ctaId: iso} — which end-of-conversation CTAs have been run (ctas.py).
        "ctas": {},
        # {"markdown": …, "at": iso} — the conversation's published summary,
        # written by the Complete CTA (see scaffold.py).
        "summary": None,
    }


def apply_worktree_event(worktrees: list[dict], ev: dict) -> list[dict]:
    """Fold a single worktree create/remove event into ``worktrees`` in place.

    Entries are keyed by directory basename (``dir``); a create fills ``name`` +
    ``createdAt`` (earliest wins), a remove sets ``removedAt``. Returns the list."""
    key = ev.get("dir") or ev.get("name")
    if not key:
        return worktrees
    wt = next((w for w in worktrees
               if (w.get("dir") or w.get("name")) == key), None)
    if wt is None:
        wt = {"name": ev.get("name") or key, "dir": ev.get("dir") or key,
              "createdAt": None, "removedAt": None}
        worktrees.append(wt)
    if ev.get("name"):
        wt["name"] = ev["name"]
    if ev.get("createdAt") and not wt.get("createdAt"):
        wt["createdAt"] = ev["createdAt"]
    if ev.get("removedAt"):
        wt["removedAt"] = ev["removedAt"]
    return worktrees


class MetaStore:
    def __init__(self, path: str):
        self.path = pathlib.Path(path)
        self._lock = threading.Lock()
        self._mtime: float | None = None
        self._data: dict[str, dict] = {}
        # Bumped on every reload/update so request-level caches keyed on the
        # sidecar's state (e.g. project cost rollups) know when to recompute.
        self.version = 0
        self._load()

    def _load(self) -> None:
        try:
            self._data = json.loads(self.path.read_text(encoding="utf-8")) or {}
            self._mtime = self.path.stat().st_mtime
        except (OSError, ValueError):
            self._data = {}
            self._mtime = None
        self.version += 1

    def reload_if_changed(self) -> bool:
        """Pick up edits made to the file out-of-band (by the skill scripts)."""
        try:
            m = self.path.stat().st_mtime
        except OSError:
            return False
        if m != self._mtime:
            with self._lock:
                self._load()
            return True
        return False

    def all(self) -> dict[str, dict]:
        with self._lock:
            return json.loads(json.dumps(self._data))  # deep copy

    def get(self, cid: str) -> dict:
        with self._lock:
            return json.loads(json.dumps(self._data.get(cid) or {}))

    def peek(self, cid: str) -> dict:
        """Read-only, no-copy lookup for hot per-card paths — the deep copy in
        :meth:`get` costs a JSON round-trip per call, which adds up when a list
        request merges metadata for hundreds of conversations. Callers must
        not mutate the returned dict."""
        with self._lock:
            return self._data.get(cid) or {}

    def update(self, cid: str, patch: dict) -> dict:
        """Shallow-merge ``patch`` into the conversation's metadata and persist.

        ``projects`` is unioned (preserving order); ``notifications`` is appended
        to (the patch supplies one entry under ``notification``); ``worktree`` is
        a single create/remove event folded into the ``worktrees`` list; and
        ``ctas`` is **merged** key-by-key rather than replaced, so running one
        CTA doesn't erase the record of the others (see ctas.py)."""
        with self._lock:
            cur = {**empty(), **(self._data.get(cid) or {})}
            note = patch.pop("notification", None)
            projects = patch.pop("projects", None)
            wt_event = patch.pop("worktree", None)
            cta_stamps = patch.pop("ctas", None)
            cur.update({k: v for k, v in patch.items() if v is not None})
            if cta_stamps:
                cur["ctas"] = {**(cur.get("ctas") or {}), **cta_stamps}
            if wt_event:
                cur["worktrees"] = apply_worktree_event(
                    list(cur.get("worktrees") or []), wt_event)
            if projects:
                seen = list(cur.get("projects") or [])
                for p in (projects if isinstance(projects, list) else [projects]):
                    if p and p not in seen:
                        seen.append(p)
                cur["projects"] = seen
            if note:
                notes = list(cur.get("notifications") or [])
                if not _is_dupe(note, notes):
                    notes.append(note)
                    cur["notifications"] = notes[-20:]  # keep the last 20
                # Always advance the "notified" stamp (a duplicate push still
                # means the conversation just notified again).
                cur["notified"] = note.get("at") or cur.get("notified")
            self._data[cid] = cur
            self._save()
            self.version += 1
            return json.loads(json.dumps(cur))

    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(self._data, indent=2, sort_keys=True),
                       encoding="utf-8")
        os.replace(tmp, self.path)
        try:
            self._mtime = self.path.stat().st_mtime
        except OSError:
            pass