perf(storage): keep no raw copy of imported events that projected

backend/app/db.py

@@ -2174,14 +2174,47 @@ def _mark_raw_ingest(
     *,
     event_id: int | None = None,
     projection_error: str | None = None,
+    drop_payload: bool = False,
 ) -> None:
     conn.execute(
         "UPDATE raw_ingest_events SET status = ?, event_id = ?, projection_error = ?"
-        " WHERE id = ?",
+        + (", raw_payload = NULL" if drop_payload else "")
+        + " WHERE id = ?",
         (status, event_id, projection_error, raw_ingest_id),
     )
 
 
+# Batch size for clearing the raw copy of already-projected imports.
+_PRUNE_BATCH = 2000
+
+
+def prune_import_raw_payloads() -> int:
+    """Clear ``raw_payload`` on imported rows that projected cleanly, a batch
+    per write so ingest gets in between. Their event row holds the full body
+    (``events.payload`` is uncapped; the raw copy is cut at 64 KB) and the
+    transcript on disk is the source, so the copy only took space: about a
+    third of an import-heavy DB. Hook rows, failures and duplicates keep
+    theirs. The freed pages are reused by later writes; reclaiming them on
+    disk needs a VACUUM, which is left to the user. Returns rows cleared."""
+    if get_setting("raw_import_pruned") == "1":
+        return 0
+    cleared = 0
+    while True:
+        with store.write() as conn:
+            n = conn.execute(
+                "UPDATE raw_ingest_events SET raw_payload = NULL WHERE id IN ("
+                " SELECT id FROM raw_ingest_events"
+                " WHERE status = 'projected' AND origin = 'import' AND raw_payload IS NOT NULL"
+                " LIMIT ?)",
+                (_PRUNE_BATCH,),
+            ).rowcount
+        cleared += n
+        if n < _PRUNE_BATCH:
+            break
+    set_setting("raw_import_pruned", "1")
+    return cleared
+
+
 def raw_ingest_events(limit: int = 100) -> list[dict[str, Any]]:
     limit = max(1, min(limit, 500))
     with store.read() as conn:
@@ -2308,7 +2341,10 @@ def _record_ingest(conn: sqlite3.Connection, source: str, raw: dict[str, Any]) -
         }
 
     raw_status = "projected" if inserted else "duplicate"
-    _mark_raw_ingest(conn, raw_id, raw_status, event_id=event_id)
+    # An imported body that projected lives on in its event row (uncapped) and
+    # in the transcript; the raw copy would only double it.
+    _mark_raw_ingest(conn, raw_id, raw_status, event_id=event_id,
+                     drop_payload=inserted and bool(raw.get("_import")))
     return {
         "ok": True,
         "raw_ingest_id": raw_id,
@@ -4786,13 +4822,19 @@ def pct(part: Any, whole: Any) -> float:
     }
 
 
-def complete_imported_sessions() -> dict[str, Any]:
+def complete_imported_sessions(quiet_seconds: float | None = None) -> dict[str, Any]:
     """Mark import-only sessions as completed.
 
     Transcripts don't carry lifecycle-end events so imported sessions
     stay 'active' forever.  This closes them using the timestamp of
-    their most recent event as ended_at.
+    their most recent event as ended_at. With ``quiet_seconds``, only
+    sessions whose last event is at least that old: an import that reads
+    transcripts while they're still being written (the watcher) must not
+    close a session that's still running.
     """
+    cutoff = (
+        datetime.now(timezone.utc) - timedelta(seconds=quiet_seconds) if quiet_seconds else None
+    )
     with store.write() as conn:
         rows = conn.execute(
             "SELECT s.id, MAX(e.ts) AS last_ts"
@@ -4805,6 +4847,8 @@ def complete_imported_sessions() -> dict[str, Any]:
             "   )"
             " GROUP BY s.id"
         ).fetchall()
+        if cutoff is not None:
+            rows = [r for r in rows if (last := timeutil.parse_ts(r["last_ts"])) is not None and last <= cutoff]
         for r in rows:
             conn.execute(
                 "UPDATE sessions SET status = 'completed', ended_at = ? WHERE id = ?",

backend/app/main.py

@@ -42,7 +42,7 @@
 
 from . import (
     __version__, activity, ai_insights, context_cost, cron, db, extension_usage, insights, otlp,
-    passive, rule_drafts, session_markdown, store, timeutil, user_rules,
+    passive, rule_drafts, session_markdown, store, timeutil, user_rules, watch,
 )
 
 app = FastAPI(title="cot collector", version=__version__)
@@ -293,6 +293,11 @@ def _build_search_index() -> None:
         insights.warm_shared_insights(30)
     except Exception:  # noqa: BLE001 - the first request computes it instead
         pass
+    # Last, so it never delays a read: a one-time batched pass on existing DBs.
+    try:
+        db.prune_import_raw_payloads()
+    except sqlite3.Error:
+        pass  # resumes next start
 
 
 # A findings pass is seconds of pure-Python work, and every SQLite call drops
@@ -318,6 +323,9 @@ async def _startup() -> None:
     # and degrades silently when offline/air-gapped.
     asyncio.create_task(_telemetry_loop())
     passive.start_scheduler()
+    # Watching sets up an FSEvents/inotify stream per transcript folder (inotify
+    # walks the tree), so it starts off the startup path too.
+    threading.Thread(target=watch.start, name="transcript-watch-start", daemon=True).start()
 
 
 @app.get("/health")
@@ -1379,8 +1387,16 @@ def get_drift_report() -> dict[str, Any]:
 
 
 @app.post("/v1/sessions/complete-imported")
-def complete_imported_sessions() -> dict[str, Any]:
-    return db.complete_imported_sessions()
+async def complete_imported_sessions(request: Request) -> dict[str, Any]:
+    """``{"quiet_seconds": N}`` closes only sessions quiet that long; an older
+    bridge sends ``{}`` and closes every import-only active session."""
+    try:
+        body = json.loads(await request.body() or b"{}")
+    except ValueError:
+        body = {}
+    quiet = body.get("quiet_seconds") if isinstance(body, dict) else None
+    quiet = float(quiet) if isinstance(quiet, (int, float)) and quiet > 0 else None
+    return await asyncio.to_thread(db.complete_imported_sessions, quiet)
 
 
 @app.post("/v1/import/reset")

backend/app/passive.py

@@ -460,10 +460,12 @@ def _ingest(url: str, body: dict[str, Any] | list[dict[str, Any]]) -> None:
         on_ingest(result.get("session_id"))
 
 
-def _import_agent(agent: str) -> dict[str, int]:
+def _import_agent(agent: str, *, hold_back: bool = True, progress: bool = True) -> dict[str, int]:
     """Import one agent's transcripts in this process, with the bridge's own
     parsers, writing events straight to the store. Transcripts of sessions
-    still in progress are held back. Returns events stored and how many
+    still in progress are held back unless ``hold_back`` is off (the watcher,
+    which imports them as they're written). Either way only sessions quiet for
+    ACTIVE_WINDOW_MIN are marked completed. Returns events stored and how many
     transcripts were held back.
 
     The bridge still reaches this collector over HTTP for the few per-pass
@@ -474,10 +476,11 @@ def _import_agent(agent: str) -> dict[str, int]:
     bridge = _load_bridge()
     bridge.COT_ENDPOINT = self_url
     bridge._INGEST_SINK = _ingest  # noqa: SLF001
-    bridge._PROGRESS_SINK = _set_progress  # noqa: SLF001
+    bridge._PROGRESS_SINK = _set_progress if progress else (lambda *_: None)  # noqa: SLF001
     try:
         posted, held_back = bridge._run_import(  # noqa: SLF001
-            [agent], skip_active_s=ACTIVE_WINDOW_MIN * 60, verbose=False
+            [agent], skip_active_s=ACTIVE_WINDOW_MIN * 60 if hold_back else 0, verbose=False,
+            complete_quiet_s=ACTIVE_WINDOW_MIN * 60,
         )
     finally:
         bridge._INGEST_SINK = None  # noqa: SLF001
@@ -571,6 +574,24 @@ def run(
     return {"id": run_id, "trigger": trigger, "started_at": started, "finished_at": finished, "status": status, **detail}
 
 
+def import_now(agents: list[str]) -> bool:
+    """Import ``agents`` straight away, live sessions included, without the
+    analysis pass or a run record: the file watcher's path. False when another
+    pass holds the lock (the caller retries)."""
+    if not _lock.acquire(blocking=False):
+        return False
+    try:
+        for agent in agents:
+            try:
+                _import_agent(agent, hold_back=False, progress=False)
+            except Exception:  # noqa: BLE001 - one agent failing shouldn't stop the rest
+                pass
+        _pending_cache.clear()
+    finally:
+        _lock.release()
+    return True
+
+
 def run_in_background(trigger: str = "manual") -> bool:
     """Start a pass on a worker thread; False if one is already running."""
     if _lock.locked():

backend/app/watch.py

@@ -0,0 +1,138 @@
+"""Import transcripts as they're written, instead of waiting for the schedule.
+
+When passive mode is on, a filesystem watcher (FSEvents on macOS, inotify on
+Linux, via ``watchdog``) sees transcript writes under each agent's folder. A
+burst of writes is debounced, then the changed agents are imported in-process
+with live sessions included (``passive.import_now``). The scheduled pass still
+runs as before: it does the analysis, covers agents that keep a database rather
+than transcripts (OpenCode, Hermes), and catches anything a dropped event missed.
+
+Without ``watchdog`` installed, or with COT_PASSIVE_WATCH=0, nothing is
+watched and the schedule alone imports, as before.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+from pathlib import Path
+from typing import Any, Callable
+
+from . import passive
+
+# Writes this close together are one burst: an agent appends many lines per turn.
+DEBOUNCE_S = 1.5
+# How long to wait before retrying when a scheduled pass holds the import lock.
+BUSY_RETRY_S = 5.0
+# Transcript folders under each agent's root. Agents missing here keep a
+# database (OpenCode, Hermes) and are left to the schedule.
+_WATCH_DIRS = {"claude": "projects", "cursor": "projects", "codex": "sessions"}
+
+
+class Watcher:
+    """Collects changed agents from filesystem events and imports them once
+    writes have been quiet for ``debounce_s``."""
+
+    def __init__(self, importer: Callable[[list[str]], bool] | None = None,
+                 debounce_s: float = DEBOUNCE_S) -> None:
+        self._import = importer or passive.import_now
+        self.debounce_s = debounce_s
+        self._pending: set[str] = set()
+        self._last_event = 0.0
+        self._cond = threading.Condition()
+        self._observer: Any = None
+        self.roots: dict[str, Path] = {}
+
+    def notify(self, agent: str) -> None:
+        with self._cond:
+            self._pending.add(agent)
+            self._last_event = time.monotonic()
+            self._cond.notify()
+
+    def _take_quiet_batch(self, stop: threading.Event) -> set[str]:
+        """Block until some agents changed and writes have gone quiet."""
+        with self._cond:
+            while not stop.is_set():
+                if self._pending:
+                    wait = self._last_event + self.debounce_s - time.monotonic()
+                    if wait <= 0:
+                        batch, self._pending = self._pending, set()
+                        return batch
+                    self._cond.wait(wait)
+                else:
+                    self._cond.wait(1.0)
+        return set()
+
+    def run_once(self, stop: threading.Event) -> None:
+        batch = self._take_quiet_batch(stop)
+        if not batch:
+            return
+        cfg = passive.config()
+        agents = [a for a in cfg["agents"] if a in batch] if cfg["enabled"] else []
+        if agents and not self._import(agents):
+            # A scheduled pass is running; try again once it's had a moment.
+            stop.wait(BUSY_RETRY_S)
+            for agent in agents:
+                self.notify(agent)
+
+    def start(self, stop: threading.Event | None = None) -> bool:
+        """Watch every readable transcript folder. False when ``watchdog`` is
+        missing or nothing could be watched."""
+        try:
+            from watchdog.events import FileSystemEventHandler
+            from watchdog.observers import Observer
+        except ImportError:
+            return False
+
+        watcher = self
+
+        class _Handler(FileSystemEventHandler):
+            def __init__(self, agent: str) -> None:
+                self.agent = agent
+
+            def on_any_event(self, event: Any) -> None:
+                if event.is_directory or event.event_type not in ("created", "modified", "moved"):
+                    return
+                path = getattr(event, "dest_path", "") or event.src_path
+                if str(path).endswith(".jsonl"):
+                    watcher.notify(self.agent)
+
+        observer = Observer()
+        for agent, sub in _WATCH_DIRS.items():
+            root = passive.transcript_root(agent) / sub
+            try:
+                resolved = root.resolve(strict=True)
+                resolved.relative_to(passive._import_home().resolve(strict=True))  # noqa: SLF001
+            except (OSError, ValueError):
+                continue
+            observer.schedule(_Handler(agent), str(resolved), recursive=True)
+            self.roots[agent] = resolved
+        if not self.roots:
+            return False
+        observer.daemon = True
+        observer.start()
+        self._observer = observer
+        stop = stop or threading.Event()
+
+        def loop() -> None:
+            while not stop.is_set():
+                try:
+                    self.run_once(stop)
+                except Exception:  # noqa: BLE001 - keep the watcher alive
+                    stop.wait(BUSY_RETRY_S)
+
+        threading.Thread(target=loop, daemon=True, name="cot-transcript-watch").start()
+        return True
+
+    def stop(self) -> None:
+        if self._observer is not None:
+            self._observer.stop()
+
+
+def start() -> Watcher | None:
+    """Start the watcher unless COT_PASSIVE_WATCH=0 (tests) or COT_PASSIVE_SCHEDULER=0."""
+    if os.environ.get("COT_PASSIVE_WATCH", "1") == "0" or os.environ.get("COT_PASSIVE_SCHEDULER", "1") == "0":
+        return None
+    watcher = Watcher()
+    return watcher if watcher.start() else None

backend/requirements.txt

@@ -1,2 +1,3 @@
 fastapi==0.141.1
 uvicorn==0.54.0
+watchdog==6.0.0

backend/tests/conftest.py

@@ -1,10 +1,14 @@
 from __future__ import annotations
 
+import os
 import sys
 from pathlib import Path
 
 import pytest
 
+# Tests start the app; never watch the real transcript folders from them.
+os.environ.setdefault("COT_PASSIVE_WATCH", "0")
+
 
 BACKEND = Path(__file__).resolve().parents[1]
 if str(BACKEND) not in sys.path:

backend/tests/test_import_batching.py

@@ -162,3 +162,15 @@ def test_batched_import_stores_what_single_posts_store(monkeypatch, tmp_path):
     single = _import_into(tmp_path / "single.db", home, monkeypatch, batched=False)
     batched = _import_into(tmp_path / "batched.db", home, monkeypatch, batched=True)
     assert single and batched == single
+
+
+def test_sink_error_is_raised_on_the_callers_thread(monkeypatch):
+    def broken(url, body):
+        raise RuntimeError("store is gone")
+
+    monkeypatch.setattr(bridge, "_INGEST_SINK", broken)
+    poster = bridge._ImportPoster(URL)
+    poster.add({"n": 0})
+    with pytest.raises(RuntimeError, match="store is gone"):
+        poster.flush()
+    assert poster._thread is None  # the sender thread was stopped