Gabriel Vidal
ebae75df94
Idle CPU sat at ~58-110% and the catalogs took seconds, because the hot paths all did O(everything) work on a timer or per request: - the indexer os.walk'd the whole workspace every 10s to find ~280 files, and the watcher rglob'd ~2600 transcripts every 2s (pathlib allocates a Path per entry) -> fsscan.walk_files (os.scandir) + a cached discovery walk, refreshed every WALK_INTERVAL_SECS and forced on save; - a live session's growing transcript was re-parsed from byte 0 on every tick (O(n^2) over a session) -> conversations.ParserState is now resumable, and the indexer feeds it only the appended lines from a stored byte offset; - the watcher already knew which transcripts changed but threw it away -> scan_transcripts(only=…) imports + re-parses just those, instead of re-walking both source dirs and the whole archive; - /api/services forked two `git log`s per service per request (11s for 34) -> githist.BucketedHistory walks each repo's log once, buckets commits per dir and caches on the repo HEAD; standalone project repos get a TTL cache; - every list request json.loads'd ~900 summary blobs and deep-copied the meta sidecar per card -> decoded summaries are cached in the store (read-only) with a version counter, and MetaStore.peek() skips the copy on hot paths. Also: SQLite in WAL, file word/char stats stored on change, and the skills rollup no longer SELECT *'s every summary blob to read one column. Verified: parse output is byte-identical to the old parser across all 905 archived transcripts (full and incremental), and idle CPU drops ~58% -> ~7% measured against the live container on the same workload. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
"""
|
|
Realtime fan-out for the ai-agent backend.
|
|
|
|
* ``Hub`` — a tiny thread-safe pub/sub: each SSE client gets a queue, and
|
|
``publish`` drops a JSON event onto every live queue.
|
|
* ``Watcher`` — a daemon thread that polls the metadata sidecar and the live
|
|
transcript files; when either changes it re-indexes (cheap/incremental) and
|
|
publishes a ``meta`` / ``transcript`` event so the frontend can update without
|
|
a manual refresh.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import queue
|
|
import threading
|
|
|
|
import fsscan
|
|
|
|
|
|
class Hub:
|
|
def __init__(self) -> None:
|
|
self._subs: set[queue.Queue] = set()
|
|
self._lock = threading.Lock()
|
|
|
|
def subscribe(self) -> queue.Queue:
|
|
q: queue.Queue = queue.Queue(maxsize=100)
|
|
with self._lock:
|
|
self._subs.add(q)
|
|
return q
|
|
|
|
def unsubscribe(self, q: queue.Queue) -> None:
|
|
with self._lock:
|
|
self._subs.discard(q)
|
|
|
|
def publish(self, event: dict) -> None:
|
|
data = json.dumps(event)
|
|
with self._lock:
|
|
subs = list(self._subs)
|
|
for q in subs:
|
|
try:
|
|
q.put_nowait(data)
|
|
except queue.Full:
|
|
pass
|
|
|
|
|
|
class Watcher(threading.Thread):
|
|
def __init__(self, source_dir, meta_store, indexer, hub, interval: float = 2.0):
|
|
super().__init__(daemon=True)
|
|
# One or many live transcript dirs (Claude Code + the pi runner mirror).
|
|
if source_dir is None:
|
|
self.source_dirs = []
|
|
elif isinstance(source_dir, (list, tuple)):
|
|
self.source_dirs = [d for d in source_dir if d]
|
|
else:
|
|
self.source_dirs = [source_dir]
|
|
self.meta_store = meta_store
|
|
self.indexer = indexer
|
|
self.hub = hub
|
|
self.interval = interval
|
|
self._sig: dict[str, tuple] = {}
|
|
self._stop = threading.Event()
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
|
|
def run(self) -> None:
|
|
self._scan_sources() # prime signatures so we don't fire on boot
|
|
while not self._stop.wait(self.interval):
|
|
try:
|
|
self._tick()
|
|
except Exception:
|
|
pass
|
|
|
|
def _scan_sources(self) -> list[str]:
|
|
"""Return rel paths of transcripts whose (mtime,size) changed.
|
|
|
|
Runs every ``interval`` seconds over ~2 600 files, so it walks with
|
|
``os.scandir`` (see fsscan) rather than ``rglob`` — no Path object per
|
|
entry."""
|
|
changed: list[str] = []
|
|
for source_dir in self.source_dirs:
|
|
if not source_dir.is_dir():
|
|
continue
|
|
base = str(source_dir)
|
|
for path, mtime, size in fsscan.walk_files(source_dir, ".jsonl"):
|
|
rel = os.path.relpath(path, base)
|
|
sig = (mtime, size)
|
|
if self._sig.get(rel) != sig:
|
|
self._sig[rel] = sig
|
|
changed.append(rel)
|
|
return changed
|
|
|
|
def _tick(self) -> None:
|
|
if self.meta_store.reload_if_changed():
|
|
self.hub.publish({"type": "meta"})
|
|
changed = self._scan_sources()
|
|
if changed:
|
|
try:
|
|
# Targeted: import + re-parse only what changed. The indexer's
|
|
# own periodic sweep still does the full pass (discovery+prune).
|
|
self.indexer.scan_transcripts(only=changed)
|
|
except Exception:
|
|
pass
|
|
self.hub.publish({"type": "transcript", "ids": changed})
|