Gabriel Vidal
b07e93246e
The list rebuilt every card after each transcript write, and every client the SSE bus woke up missed the cache at once and ran the same full build in parallel on one GIL; FastAPI then encoded the big bodies on the event loop. Live, that pinned a core with /api/conversations at 16-80 s and even /api/health at 43 s. - _Memo: version-keyed, single-flight memo for the card list, agent runs and agents-by-conversation (cards also key on a 15 s time bucket, since a card's running/finished state reads the clock) - lru_cache on _conv_id / _parent_path_of / _is_sidechain_path: pathlib.relative_to was 0.24 ms a call, half the list's cost - MetaStore.peek no longer takes the lock (convoyed thousands of peeks per build); MetaStore.items() replaces the 1.4 MB deep copy in _build_agent_runs - @_json_in_worker: the eleven heaviest GETs return a ready JSONResponse from their worker thread instead of leaving jsonable_encoder (5-10x slower than json.dumps) to run on the event loop Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
241 lines
9.9 KiB
Python
241 lines
9.9 KiB
Python
"""
|
|
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}, … ],
|
|
"worker": "<workerId>"|null, # paired worker the hub launched
|
|
# the run on (workers.py)
|
|
"workerSource": "<workerId>"|null, # worker the transcript was
|
|
# mirrored from (remotefeed.py) —
|
|
# a terminal session there
|
|
"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,
|
|
# {"markdown": …, "at": iso} — the conversation's published summary,
|
|
# written by the `complete` skill's end-of-task pass (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.
|
|
|
|
Lock-free on purpose: writers only ever *replace* ``_data`` or one of
|
|
its entries (never mutate an entry in place), and a dict lookup is
|
|
atomic under the GIL. Taking the lock here made every list build
|
|
(thousands of peeks) convoy behind any thread holding it."""
|
|
return self._data.get(cid) or {}
|
|
|
|
def items(self) -> list[tuple[str, dict]]:
|
|
"""Shallow ``(cid, entry)`` snapshot — :meth:`all` without the deep
|
|
copy, for read-only sweeps. Same contract as :meth:`peek`."""
|
|
return list(self._data.items())
|
|
|
|
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."""
|
|
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)
|
|
cur.update({k: v for k, v in patch.items() if v is not None})
|
|
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 stamp_many(self, patches: dict[str, dict]) -> int:
|
|
"""Set plain fields on many conversations in **one** save (the remote
|
|
mirror's first sync stamps hundreds of backfilled sessions — a save
|
|
per session rewrites this whole file each time). Only keys whose value
|
|
differs are written; returns how many conversations changed."""
|
|
n = 0
|
|
with self._lock:
|
|
for cid, patch in patches.items():
|
|
cur = self._data.get(cid) or {}
|
|
diff = {k: v for k, v in patch.items()
|
|
if v is not None and cur.get(k) != v}
|
|
if diff:
|
|
self._data[cid] = {**empty(), **cur, **diff}
|
|
n += 1
|
|
if n:
|
|
self._save()
|
|
self.version += 1
|
|
return n
|
|
|
|
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
|