Gabriel Vidal
81c455defa
A guest sees the conversations they started or were shared (list, detail, summary, search index, subagents, raw meta, diffs), the forms, asks and pushes of those, and the projects they were granted. Their ui-state lives under u.<id>.<key>; the SSE hub filters per subscriber. Forms record answeredBy. Pushes from a guest's session skip the owner's webhooks and go to the guest's channels (in-app, email). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
118 lines
3.9 KiB
Python
118 lines
3.9 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 typing
|
|
|
|
import fsscan
|
|
|
|
|
|
class Hub:
|
|
"""Each subscriber may bring a ``filter(event) -> event | None``: a guest's
|
|
stream (``core/routes.py``) drops what isn't theirs and trims the rest."""
|
|
|
|
def __init__(self) -> None:
|
|
self._subs: dict[queue.Queue, typing.Callable | None] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def subscribe(self, filter: typing.Callable | None = None) -> queue.Queue:
|
|
q: queue.Queue = queue.Queue(maxsize=100)
|
|
with self._lock:
|
|
self._subs[q] = filter
|
|
return q
|
|
|
|
def unsubscribe(self, q: queue.Queue) -> None:
|
|
with self._lock:
|
|
self._subs.pop(q, None)
|
|
|
|
def publish(self, event: dict) -> None:
|
|
data = json.dumps(event)
|
|
with self._lock:
|
|
subs = list(self._subs.items())
|
|
for q, flt in subs:
|
|
out = data
|
|
if flt is not None:
|
|
try:
|
|
ev = flt(event)
|
|
except Exception:
|
|
ev = None
|
|
if ev is None:
|
|
continue
|
|
out = data if ev is event else json.dumps(ev)
|
|
try:
|
|
q.put_nowait(out)
|
|
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})
|