From 302a101ea7db6aa1a747fc8e5fd6d5eb065aee9e Mon Sep 17 00:00:00 2001 From: 0xcardinal <0xcardinal@gmail.com> Date: Thu, 22 Oct 2026 01:32:12 +0530 Subject: [PATCH 1/2] perf(import): watch transcripts, stream large files, overlap parse and write Watcher (app/watch.py): with passive mode on, a watchdog observer (FSEvents on macOS, inotify on Linux) watches the Claude, Cursor and Codex transcript folders. A burst of .jsonl writes is debounced (1.5 s) and the changed, enabled agents are imported in-process with live sessions included, so a running session shows up in about two seconds instead of after the schedule and the 10-minute hold-back. In-process passes now close only imported sessions quiet for 10 minutes (complete-imported takes quiet_seconds; an older bridge's {} still closes all). The schedule keeps running for analysis, OpenCode/Hermes, and anything a dropped event missed. Without watchdog, or with COT_PASSIVE_WATCH=0, nothing is watched. Large files: new bytes are read 1 MB at a time with parser state carried across chunks, so the bridge peaks at ~140 MB instead of ~720 MB, and transcripts over 50 MB are no longer skipped. A line longer than a chunk is read whole up to 64 MB; a longer one is skipped with a warning. Lines split on "\n" only, offsets are cut on bytes (invalid UTF-8 no longer shifts them). Fixes Cursor appends being dropped: import line numbers restarted at 0 on every pass, so an appended line got the dedup key of the file's first line. Line numbers are now file-wide (kept in the identity sidecar, counted once for older entries); keys of already-imported lines are unchanged. Overlap: the import poster sends batches from a background thread while the next batch parses (SQLite and HTTP release the GIL). --- backend/app/db.py | 12 +- backend/app/main.py | 17 +- backend/app/passive.py | 29 ++- backend/app/watch.py | 138 ++++++++++++ backend/requirements.txt | 1 + backend/tests/conftest.py | 4 + backend/tests/test_import_batching.py | 12 + backend/tests/test_import_chunks.py | 167 ++++++++++++++ backend/tests/test_watch.py | 102 +++++++++ bridge/cot | 305 ++++++++++++++++++-------- desktop/packaging/cot-collector.spec | 2 + 11 files changed, 689 insertions(+), 100 deletions(-) create mode 100644 backend/app/watch.py create mode 100644 backend/tests/test_import_chunks.py create mode 100644 backend/tests/test_watch.py diff --git a/backend/app/db.py b/backend/app/db.py index 0a328c9..cc86ca1 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -4786,13 +4786,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 +4811,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 = ?", diff --git a/backend/app/main.py b/backend/app/main.py index 37d8090..9277e3a 100644 --- a/backend/app/main.py +++ b/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__) @@ -318,6 +318,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 +1382,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") diff --git a/backend/app/passive.py b/backend/app/passive.py index 0e20e74..211e147 100644 --- a/backend/app/passive.py +++ b/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(): diff --git a/backend/app/watch.py b/backend/app/watch.py new file mode 100644 index 0000000..f8ead47 --- /dev/null +++ b/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 diff --git a/backend/requirements.txt b/backend/requirements.txt index 425171a..a9fe12f 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -1,2 +1,3 @@ fastapi==0.141.1 uvicorn==0.54.0 +watchdog==6.0.0 diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 2583278..50373aa 100644 --- a/backend/tests/conftest.py +++ b/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: diff --git a/backend/tests/test_import_batching.py b/backend/tests/test_import_batching.py index 2485ac8..56483ae 100644 --- a/backend/tests/test_import_batching.py +++ b/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 diff --git a/backend/tests/test_import_chunks.py b/backend/tests/test_import_chunks.py new file mode 100644 index 0000000..4cc78e2 --- /dev/null +++ b/backend/tests/test_import_chunks.py @@ -0,0 +1,167 @@ +"""Large transcripts import a chunk at a time with the same result as one +read, and line-keyed dedup stays file-wide across appends and chunks.""" + +from __future__ import annotations + +import importlib.machinery +import importlib.util +import json +from pathlib import Path + +import pytest + +_REPO = Path(__file__).resolve().parents[2] + + +@pytest.fixture +def bridge(monkeypatch, tmp_path): + monkeypatch.setenv("HOME", str(tmp_path)) + loader = importlib.machinery.SourceFileLoader("cot_bridge_chunks_under_test", str(_REPO / "bridge" / "cot")) + spec = importlib.util.spec_from_loader("cot_bridge_chunks_under_test", loader) + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + monkeypatch.setenv("COT_IMPORT_HOME", str(tmp_path)) + return module + + +def _capture(bridge, monkeypatch) -> list[dict]: + sent: list[dict] = [] + + class Poster: + def __init__(self, url): + pass + + def add(self, payload): + sent.append(payload) + + def flush(self): + pass + + monkeypatch.setattr(bridge, "_ImportPoster", Poster) + return sent + + +def _cursor_line(text: str) -> str: + return json.dumps({"role": "user", "message": {"content": [ + {"type": "text", "text": f"\n{text}\n"}]}}) + "\n" + + +def _cursor_file(home: Path) -> Path: + d = home / ".cursor" / "projects" / "p" / "agent-transcripts" + d.mkdir(parents=True) + return d / "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa.jsonl" + + +def _claude_file(home: Path, turns: int) -> Path: + sid = "44444444-4444-4444-8444-444444444444" + d = home / ".claude" / "projects" / "-p" + d.mkdir(parents=True) + lines = [{"type": "user", "uuid": "p", "sessionId": sid, "timestamp": "2026-06-01T00:00:00Z", + "message": {"role": "user", "content": "go"}}] + for t in range(turns): + lines.append({"type": "assistant", "uuid": f"a{t}", "sessionId": sid, + "timestamp": "2026-06-01T00:01:00Z", + "message": {"role": "assistant", "content": [ + {"type": "text", "text": f"step {t} " + "x" * 200}, + {"type": "tool_use", "id": f"t{t}", "name": "Bash", "input": {"command": "ls"}}]}}) + lines.append({"type": "user", "uuid": f"r{t}", "sessionId": sid, "timestamp": "2026-06-01T00:01:30Z", + "message": {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": f"t{t}", "content": "ok " * 50}]}}) + f = d / f"{sid}.jsonl" + f.write_text("".join(json.dumps(x) + "\n" for x in lines)) + return f + + +def _keys(sent: list[dict]) -> list: + return [(p.get("hook_event_name"), p.get("_dedup_key"), p.get("tool_name"), p.get("prompt"), + json.dumps(p.get("tool_response"), sort_keys=True)) for p in sent] + + +def test_chunked_import_matches_one_read(bridge, monkeypatch, tmp_path): + f = _claude_file(tmp_path, turns=200) + assert f.stat().st_size > 100_000 + sent = _capture(bridge, monkeypatch) + bridge._import_agent("claude", None, {}, {}, verbose=False) + whole = _keys(sent) + sent.clear() + bridge._FILE_STATE.clear() + monkeypatch.setattr(bridge, "_IMPORT_CHUNK_BYTES", 4096) + offsets: dict = {} + bridge._import_agent("claude", None, offsets, {}, verbose=False) + assert _keys(sent) == whole and len(whole) > 400 + assert offsets[str(f)] == f.stat().st_size + + +def test_cursor_append_keeps_file_wide_line_keys(bridge, monkeypatch, tmp_path): + f = _cursor_file(tmp_path) + f.write_text(_cursor_line("first") + _cursor_line("second")) + sent = _capture(bridge, monkeypatch) + offsets: dict = {} + bridge._import_agent("cursor", None, offsets, {}, verbose=False) + with f.open("a") as fh: + fh.write(_cursor_line("third")) + bridge._import_agent("cursor", None, offsets, {}, verbose=False) + keys = {p["prompt"]: p["_dedup_key"] for p in sent} + assert keys["third"].endswith(":2:prompt"), keys + assert len(set(keys.values())) == 3 + + +def test_line_numbers_count_from_the_file_for_offsets_without_them(bridge, monkeypatch, tmp_path): + """Offsets written before line counts were kept are counted once.""" + f = _cursor_file(tmp_path) + f.write_text(_cursor_line("first") + "\n" + _cursor_line("second")) + offsets = {str(f): f.stat().st_size} # an older bridge read both prompts + with f.open("a") as fh: + fh.write(_cursor_line("third")) + sent = _capture(bridge, monkeypatch) + bridge._import_agent("cursor", None, offsets, {}, verbose=False) + assert [p["_dedup_key"].rsplit(":", 2)[-2] for p in sent] == ["2"] + + +def test_line_longer_than_a_chunk_is_read_whole(bridge, monkeypatch, tmp_path): + f = _cursor_file(tmp_path) + f.write_text(_cursor_line("before") + _cursor_line("big " + "z" * 20_000) + _cursor_line("after")) + monkeypatch.setattr(bridge, "_IMPORT_CHUNK_BYTES", 4096) + sent = _capture(bridge, monkeypatch) + offsets: dict = {} + lines, offset, _, more = bridge._import_read_chunk(f, offsets) + assert lines and more # the long line comes back alone, with more behind it + bridge._import_agent("cursor", None, {}, {}, verbose=False) + assert [p["prompt"][:3] for p in sent] == ["bef", "big", "aft"] + assert sent[2]["_dedup_key"].endswith(":2:prompt") + + +def test_line_past_the_line_cap_is_skipped_not_wedged(bridge, monkeypatch, tmp_path): + f = _cursor_file(tmp_path) + f.write_text(_cursor_line("before") + json.dumps({"blob": "z" * 3_000_000}) + "\n" + _cursor_line("after")) + monkeypatch.setattr(bridge, "_IMPORT_CHUNK_BYTES", 4096) + monkeypatch.setattr(bridge, "_MAX_LINE_BYTES", 1024 * 1024) + sent = _capture(bridge, monkeypatch) + offsets: dict = {} + bridge._import_agent("cursor", None, offsets, {}, verbose=False) + assert [p["prompt"] for p in sent] == ["before", "after"] + assert sent[1]["_dedup_key"].endswith(":2:prompt") + assert offsets[str(f)] == f.stat().st_size + + +def test_invalid_utf8_does_not_shift_the_offset(bridge, tmp_path): + f = tmp_path / "t.jsonl" + f.write_bytes(b'{"a": "\xff\xfe"}\n{"b": 1}\n') + offsets: dict = {} + lines, offset, first, more = bridge._import_read_chunk(f, offsets) + assert len(lines) == 2 and offset == f.stat().st_size and first == 0 and not more + + +def test_large_files_are_no_longer_skipped(bridge, monkeypatch, tmp_path): + f = _claude_file(tmp_path, turns=50) + monkeypatch.setattr(bridge, "_MAX_TRANSCRIPT_BYTES", 1024) # the old whole-file cap + sent = _capture(bridge, monkeypatch) + bridge._import_agent("claude", None, {}, {}, verbose=False) + assert len(sent) > 100 + + +def test_unicode_line_separator_inside_a_string_stays_one_line(bridge, tmp_path): + f = tmp_path / "t.jsonl" + f.write_text(json.dumps({"text": "a\u2028b"}, ensure_ascii=False) + "\n" + '{"n": 2}\n') + lines, _, _, _ = bridge._import_read_chunk(f, {}) + assert [json.loads(x) for x in lines] == [{"text": "a\u2028b"}, {"n": 2}] diff --git a/backend/tests/test_watch.py b/backend/tests/test_watch.py new file mode 100644 index 0000000..6355e0c --- /dev/null +++ b/backend/tests/test_watch.py @@ -0,0 +1,102 @@ +"""The transcript watcher debounces writes into one import per burst, imports +only enabled agents while passive mode is on, retries when a pass is running, +and imported sessions close only once they've gone quiet.""" + +from __future__ import annotations + +import json +import threading +import time +from datetime import datetime, timedelta, timezone + +import pytest + +from app import db, passive, store, watch + + +@pytest.fixture +def passive_on(fresh_db): + passive.update_config(enabled=True, agents=["claude", "codex"]) + return fresh_db + + +def test_burst_of_writes_is_one_import(passive_on): + calls: list[list[str]] = [] + w = watch.Watcher(importer=lambda agents: calls.append(sorted(agents)) or True, debounce_s=0.05) + for _ in range(20): + w.notify("claude") + w.notify("codex") + w.run_once(threading.Event()) + assert calls == [["claude", "codex"]] + + +def test_waits_for_writes_to_go_quiet(passive_on): + calls: list[float] = [] + w = watch.Watcher(importer=lambda agents: calls.append(time.monotonic()) or True, debounce_s=0.2) + w.notify("claude") + started = time.monotonic() + threading.Timer(0.1, w.notify, args=("claude",)).start() # still writing + w.run_once(threading.Event()) + assert calls and calls[0] - started >= 0.28 + + +def test_nothing_imports_while_passive_mode_is_off(fresh_db): + passive.update_config(enabled=False, agents=["claude"]) + calls: list = [] + w = watch.Watcher(importer=lambda agents: calls.append(agents) or True, debounce_s=0.01) + w.notify("claude") + w.run_once(threading.Event()) + assert calls == [] + + +def test_disabled_agent_is_not_imported(passive_on): + calls: list = [] + w = watch.Watcher(importer=lambda agents: calls.append(agents) or True, debounce_s=0.01) + w.notify("cursor") + w.run_once(threading.Event()) + assert calls == [] + + +def test_busy_lock_retries(passive_on, monkeypatch): + monkeypatch.setattr(watch, "BUSY_RETRY_S", 0.01) + results = iter([False, True]) + calls: list = [] + w = watch.Watcher(importer=lambda agents: calls.append(agents) or next(results), debounce_s=0.01) + w.notify("claude") + w.run_once(threading.Event()) + w.run_once(threading.Event()) + assert calls == [["claude"], ["claude"]] + + +def test_real_filesystem_events_trigger_an_import(passive_on, monkeypatch, tmp_path): + pytest.importorskip("watchdog") + root = tmp_path / ".claude" / "projects" / "-p" + root.mkdir(parents=True) + monkeypatch.setenv("COT_IMPORT_HOME", str(tmp_path)) + seen = threading.Event() + w = watch.Watcher(importer=lambda agents: seen.set() or True, debounce_s=0.1) + stop = threading.Event() + try: + assert w.start(stop) and set(w.roots) == {"claude"} + time.sleep(0.5) # let the stream start + (root / "s.jsonl").write_text(json.dumps({"type": "user"}) + "\n") + assert seen.wait(10) + finally: + stop.set() + w.stop() + + +def _session_with_last_event(sid: str, minutes_ago: float) -> None: + ts = (datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)).isoformat() + db.record_ingests([("claude", {"session_id": sid, "hook_event_name": "UserPromptSubmit", + "prompt": "hi", "timestamp": ts, "_import": True})]) + + +def test_completion_leaves_sessions_still_being_written(fresh_db): + _session_with_last_event("quiet", minutes_ago=30) + _session_with_last_event("live", minutes_ago=1) + assert db.complete_imported_sessions(quiet_seconds=600) == {"completed": 1} + with store.read() as conn: + status = dict(conn.execute("SELECT id, status FROM sessions").fetchall()) + assert status == {"quiet": "completed", "live": "active"} + assert db.complete_imported_sessions() == {"completed": 1} # an older bridge closes all diff --git a/bridge/cot b/bridge/cot index 6ce7baf..cb94976 100755 --- a/bridge/cot +++ b/bridge/cot @@ -413,6 +413,8 @@ def _post_async(url: str, payload: dict) -> None: # one write transaction, so this also bounds how long an import holds the write # lock that live hooks wait on. _IMPORT_POST_BATCH = 200 +# Batches queued for the poster's sender thread before parsing waits on it. +_IMPORT_SEND_QUEUE = 2 # Set by a collector that runs the import in its own process: called with # (url, body) instead of POSTing, where body is one payload or a list of them. @@ -438,11 +440,19 @@ class _ImportPoster: that can't be delivered are spooled, as ``_post`` does, so the caller can advance its offsets either way. Attachment bodies go alone: the collector folds them onto a prompt only when they arrive as a single object. - Call ``flush()`` before reading the results back or exiting.""" + + Batches are sent from one background thread while the caller parses the + next: the send (an HTTP round trip, or SQLite writes, which release the + GIL) overlaps the parsing. At most ``_IMPORT_SEND_QUEUE`` batches wait. + Call ``flush()`` before reading the results back or exiting; it returns + once everything is delivered and re-raises a send's error.""" def __init__(self, url: str) -> None: self.url = url self.pending: list[dict] = [] + self._queue = None + self._thread: threading.Thread | None = None + self._error: BaseException | None = None self.sink = _INGEST_SINK if self.sink is not None: self.batched = True @@ -455,17 +465,51 @@ class _ImportPoster: def add(self, payload: dict) -> None: if not self.batched or payload.get("_attach_to_prompt"): - self.flush() - self._send([payload]) + self._ship() + self._submit([payload]) return self.pending.append(payload) if len(self.pending) >= _IMPORT_POST_BATCH: - self.flush() + self._ship() def flush(self) -> None: + self._ship() + if self._thread is not None: + self._queue.put(None) + self._thread.join() + self._thread = self._queue = None + self._raise() + + def _ship(self) -> None: if self.pending: group, self.pending = self.pending, [] - self._send(group) + self._submit(group) + + def _raise(self) -> None: + if self._error is not None: + error, self._error = self._error, None + raise error + + def _submit(self, group: list[dict]) -> None: + self._raise() + if self._thread is None: + import queue + + self._queue = queue.Queue(maxsize=_IMPORT_SEND_QUEUE) + self._thread = threading.Thread(target=self._sender, daemon=True, name="cot-import-send") + self._thread.start() + self._queue.put(group) + + def _sender(self) -> None: + while True: + group = self._queue.get() + if group is None: + return + if self._error is None: # after a failure, drop the rest; flush raises + try: + self._send(group) + except BaseException as exc: # noqa: BLE001 - re-raised on the caller's thread + self._error = exc def _send(self, group: list[dict]) -> None: if self.sink is not None: @@ -2767,67 +2811,135 @@ def _tail_digest(fh, offset: int) -> str: return hashlib.sha256(fh.read(min(offset, _TAIL_PROBE_BYTES))).hexdigest()[:32] -def _remember_file(key: str, st: os.stat_result, offset: int, tail: str) -> None: - _FILE_STATE[key] = {"offset": offset, "ino": st.st_ino, "mtime_ns": st.st_mtime_ns, "tail": tail} +def _remember_file(key: str, st: os.stat_result, offset: int, tail: str, lines: int) -> None: + _FILE_STATE[key] = {"offset": offset, "ino": st.st_ino, "mtime_ns": st.st_mtime_ns, + "tail": tail, "lines": lines} -def _import_read_new_lines(path: Path, offsets: dict[str, int]) -> tuple[list[str], int]: - """Read new complete lines from *path* past the stored offset. +# New bytes read per step. A large transcript is imported a chunk at a time so +# memory stays flat however big the file is; parser state carries across. 1 MB +# measured as fast as 8 MB with a third of the peak memory. +_IMPORT_CHUNK_BYTES = 1024 * 1024 +# A single line longer than a chunk is read whole up to this size (a large +# inlined tool output); a longer one is skipped rather than held in memory. +_MAX_LINE_BYTES = 64 * 1024 * 1024 + - Returns (lines, new_offset). Incomplete trailing lines (writer mid-flush) - are not consumed — the offset stops at the last newline. Reads in binary so - the stored byte offset seeks exactly; opens with O_NOFOLLOW and enforces a - size cap. +def _count_lines_before(fh, offset: int) -> int: + """Non-empty lines in the first ``offset`` bytes: the line number the next + read starts at, for transcripts tracked before line counts were kept.""" + fh.seek(0) + count, read = 0, 0 + while read < offset: + raw = fh.readline(offset - read) + if not raw: + break + read += len(raw) + if raw.strip(): + count += 1 + return count + + +def _import_read_chunk(path: Path, offsets: dict[str, int], + max_bytes: int | None = None) -> tuple[list[str], int, int, bool]: + """Read up to ``max_bytes`` of new complete lines from *path* past the + stored offset. + + Returns (lines, new_offset, first_lineno, more). ``first_lineno`` is the + file-wide index (counting non-empty lines) of the first returned line, so + line-keyed dedup stays stable across appends and chunks. ``more`` is True + when the read stopped at ``max_bytes`` with more data behind it. An + incomplete trailing line (writer mid-flush) is not consumed; a single line + longer than ``max_bytes`` is read whole, and one past ``_MAX_LINE_BYTES`` + is skipped. Reads in binary so the stored byte + offset seeks exactly; opens with O_NOFOLLOW. A file whose inode, size and mtime match what was recorded is skipped without opening it. One whose inode changed (replaced at the same path) or whose bytes before the offset changed (rewritten) is read from the start; dedup keys keep the re-read idempotent. """ + max_bytes = max_bytes or _IMPORT_CHUNK_BYTES key = str(path) start = offsets.get(key, 0) try: st = path.stat() except OSError: - return [], start + return [], start, 0, False size = st.st_size - if size > _MAX_TRANSCRIPT_BYTES: - # Don't advance the offset; an oversized file is skipped, not consumed. - return [], start known = _FILE_STATE.get(key) if known is not None and known.get("offset") != start: known = None # an older bridge moved the offset; the identity is stale if known is not None: if known.get("ino") != st.st_ino: - start = 0 + start, known = 0, None elif size == start and known.get("mtime_ns") == st.st_mtime_ns: - return [], start + return [], start, int(known.get("lines") or 0), False if size < start: - start = 0 + start, known = 0, None fh = _open_rb_nofollow(path) if fh is None: - return [], start + return [], start, 0, False try: if start and known is not None and known.get("tail") and _tail_digest(fh, start) != known["tail"]: - start = 0 + start, known = 0, None + if start == 0: + base = 0 + elif known is not None and isinstance(known.get("lines"), int): + base = known["lines"] + else: + base = _count_lines_before(fh, start) if size == start: # Unchanged content (a touch, or no identity recorded yet): note it. - _remember_file(key, st, start, _tail_digest(fh, start)) - return [], start + _remember_file(key, st, start, _tail_digest(fh, start), base) + return [], start, base, False fh.seek(start) - chunk = fh.read().decode("utf-8", errors="replace") - # Only consume up to the last newline so partial writes are retried. - last_nl = chunk.rfind("\n") + raw = fh.read(max_bytes) + last_nl = raw.rfind(b"\n") if last_nl < 0: - return [], start - consumed = chunk[: last_nl + 1] - new_offset = start + len(consumed.encode("utf-8")) - _remember_file(key, st, new_offset, _tail_digest(fh, new_offset)) + if len(raw) < max_bytes: + return [], start, base, False # a partial line still being written + # One line longer than a whole chunk: read on to its end. + parts, length, keep = [raw], len(raw), True + while True: + block = fh.read(1024 * 1024) + if not block: + return [], start, base, False # still unterminated + nl = block.find(b"\n") + if nl >= 0: + length += nl + 1 + if keep: + parts.append(block[: nl + 1]) + break + length += len(block) + if keep: + parts.append(block) + if length > _MAX_LINE_BYTES: + keep, parts = False, [] + if not keep: + new_offset = start + length + print(f"cot import: skipped a {length // (1024 * 1024)} MB line in {path.name}", file=sys.stderr) + _remember_file(key, st, new_offset, _tail_digest(fh, new_offset), base + 1) + return [], new_offset, base + 1, True + raw = b"".join(parts) + last_nl = len(raw) - 1 + consumed = raw[: last_nl + 1] + new_offset = start + len(consumed) + more = len(raw) >= max_bytes and new_offset < size + # JSONL splits on "\n" only: splitlines() would also break on U+2028 and + # friends, which JSON allows unescaped inside strings. + lines = [ln for ln in consumed.decode("utf-8", errors="replace").split("\n") if ln.strip()] + _remember_file(key, st, new_offset, _tail_digest(fh, new_offset), base + len(lines)) except OSError: - return [], start + return [], start, 0, False finally: fh.close() - lines = [ln for ln in consumed.splitlines() if ln.strip()] + return lines, new_offset, base, more + + +def _import_read_new_lines(path: Path, offsets: dict[str, int]) -> tuple[list[str], int]: + """Every new complete line past the stored offset, in one read.""" + lines, new_offset, _, _ = _import_read_chunk(path, offsets, max_bytes=_MAX_TRANSCRIPT_BYTES) return lines, new_offset @@ -4535,67 +4647,77 @@ def _import_agent(agent: str, since: str | None, offsets: dict[str, int], except (OSError, ValueError): pass - lines, new_offset = _import_read_new_lines(path, offsets) - if not lines: - offsets[str(path)] = new_offset - continue - - if verbose: - print(f"cot import: {agent} {path.name} ({len(lines)} new lines)") - + key = str(path) + # Per-file state lets parsers correlate across lines (Claude/Codex pair a + # tool call with its result; reset per file so ids never bleed across). + # It carries across chunks of the same file. + state: dict = {} try: mtime = path.stat().st_mtime except OSError: mtime = None - # Per-file state lets parsers correlate across lines (Claude/Codex pair a - # tool call with its result; reset per file so ids never bleed across). - state: dict = {} - path_str = str(path) - parsed_events: list[dict] = [] - n_lines = len(lines) - for lineno, line in enumerate(lines): - # Parsing is half a file's work, posting the other half. - report(done, frac=0.5 * lineno / n_lines) - try: - obj = json.loads(line) - except json.JSONDecodeError: - continue - parsed_events.extend( - line_parser( - obj, - sid, - lineno=lineno, - mtime=mtime, - state=state, - path=path_str, - origin="import", - ) - ) - parsed_events.extend(_codex_flush_pending_ingest_events(state)) - events = _prepare_ingest_events(agent, parsed_events, origin="import") - n_events = max(1, len(events)) - for i_event, event in enumerate(events): - # Subagent transcripts: live hooks already hold the tool calls, so - # import only the conversation text the hooks never deliver. - if responses_only and event.get("kind") == "tool_call": - continue - # Hook-supplemented sessions: hooks already cover responses, - # thoughts, and hooked tool calls. Only import tool_call events - # for tools that the hook system never fires (the same set that - # _keep_live_event preserves during live mode). - if hook_supplement: - if event.get("kind") != "tool_call": - continue - if not _keep_live_event(source, event): + begin = offsets.get(key, 0) + span = max(1, (sizes[done] if done < len(sizes) else 0) - begin) + + def emit(parsed_events: list[dict]) -> None: + nonlocal posted + for event in _prepare_ingest_events(agent, parsed_events, origin="import"): + # Subagent transcripts: live hooks already hold the tool calls, so + # import only the conversation text the hooks never deliver. + if responses_only and event.get("kind") == "tool_call": continue - payload = _ingest_event_to_hook_payload(event) - if payload.get("_dedup_key") is None: - payload.pop("_dedup_key", None) - poster.add(payload) - posted += 1 - report(done, frac=0.5 + 0.5 * (i_event + 1) / n_events) # throttled + # Hook-supplemented sessions: hooks already cover responses, + # thoughts, and hooked tool calls. Only import tool_call events + # for tools that the hook system never fires (the same set that + # _keep_live_event preserves during live mode). + if hook_supplement: + if event.get("kind") != "tool_call": + continue + if not _keep_live_event(source, event): + continue + payload = _ingest_event_to_hook_payload(event) + if payload.get("_dedup_key") is None: + payload.pop("_dedup_key", None) + poster.add(payload) + posted += 1 - offsets[str(path)] = new_offset + while True: + lines, new_offset, first_lineno, more = _import_read_chunk(path, offsets) + if not lines: + offsets[key] = new_offset + if more: + continue # skipped an oversized line; keep going + # Codex holds its last message until the end of what's there. + emit(_codex_flush_pending_ingest_events(state)) + break + if verbose: + print(f"cot import: {agent} {path.name} ({len(lines)} new lines)") + parsed_events: list[dict] = [] + for i, line in enumerate(lines): + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + parsed_events.extend( + line_parser( + obj, + sid, + lineno=first_lineno + i, + mtime=mtime, + state=state, + path=key, + origin="import", + ) + ) + if not more: + # Codex holds its last message until the end of what's there. + parsed_events.extend(_codex_flush_pending_ingest_events(state)) + emit(parsed_events) + del parsed_events + offsets[key] = new_offset + report(done, frac=min(1.0, (new_offset - begin) / span)) # throttled + if not more: + break poster.flush() report(len(transcripts), force=True) @@ -4603,11 +4725,12 @@ def _import_agent(agent: str, since: str | None, offsets: dict[str, int], def _run_import(agents: list[str], *, since: str | None = None, skip_active_s: float = 0, - verbose: bool = True) -> tuple[int, list[str]]: + verbose: bool = True, complete_quiet_s: float | None = None) -> tuple[int, list[str]]: """One import pass over ``agents``: events, then session completion, question recovery and subagent links. Returns (events posted, held-back transcripts). ``cot import`` runs it over HTTP; the collector runs it in - its own process with ``_INGEST_SINK`` set.""" + its own process with ``_INGEST_SINK`` set. ``complete_quiet_s`` closes only + imported sessions quiet that long, for a pass that reads live transcripts.""" say = print if verbose else (lambda *a, **k: None) _reindex_if_parser_changed() origins = _fetch_session_origins() @@ -4633,7 +4756,7 @@ def _run_import(agents: list[str], *, since: str | None = None, skip_active_s: f f"{COT_ENDPOINT}/v1/sessions/complete-imported", method="POST", headers={"Content-Type": "application/json"}, - data=b"{}", + data=json.dumps({"quiet_seconds": complete_quiet_s} if complete_quiet_s else {}).encode(), ) with urllib.request.urlopen(req, timeout=10) as resp: result = json.loads(resp.read().decode("utf-8")) diff --git a/desktop/packaging/cot-collector.spec b/desktop/packaging/cot-collector.spec index 3af5478..27dbec6 100644 --- a/desktop/packaging/cot-collector.spec +++ b/desktop/packaging/cot-collector.spec @@ -31,6 +31,8 @@ a = Analysis( # stdlib modules it needs that the collector itself may not pull in. "argparse", "base64", "fcntl", "hashlib", "mimetypes", "shutil", "socket", "struct", "urllib.request", "urllib.error", + # watchdog picks its platform observer at runtime (app/watch.py). + "watchdog.observers.fsevents", "_watchdog_fsevents", ], hookspath=[], runtime_hooks=[], From d4f22879c66368455e9de6e4614805b32b09391b Mon Sep 17 00:00:00 2001 From: 0xcardinal <0xcardinal@gmail.com> Date: Thu, 22 Oct 2026 01:37:47 +0530 Subject: [PATCH 2/2] perf(storage): keep no raw copy of imported events that projected The raw ingest ledger stored every body twice over: raw_payload (cut at 64 KB) next to events.payload (uncapped). For an imported event that projected, the event row and the transcript on disk both hold the body, so the ledger row now keeps its metadata, hash and status but no payload. Hook bodies keep theirs (cleanup migrations can delete hook events while their ledger rows stay), as do failures, ignored rows and duplicates. Existing DBs are pruned once, in batches of 2,000 per write, at the end of the background startup work. Freed pages are reused by later writes; shrinking the file needs a VACUUM, left to the user. A fresh import of 611 MB of Claude history: 437 MB -> 320 MB, same speed. --- backend/app/db.py | 40 ++++++++++++++++++++++++-- backend/app/main.py | 5 ++++ backend/tests/test_raw_ingest_drift.py | 31 ++++++++++++++++++++ 3 files changed, 74 insertions(+), 2 deletions(-) diff --git a/backend/app/db.py b/backend/app/db.py index cc86ca1..50d86dd 100644 --- a/backend/app/db.py +++ b/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, diff --git a/backend/app/main.py b/backend/app/main.py index 9277e3a..20d4f73 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -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 diff --git a/backend/tests/test_raw_ingest_drift.py b/backend/tests/test_raw_ingest_drift.py index c5e3338..8c7603d 100644 --- a/backend/tests/test_raw_ingest_drift.py +++ b/backend/tests/test_raw_ingest_drift.py @@ -352,3 +352,34 @@ def test_drift_report_groups_weekly_unknown_rates_and_raw_status_counts(): ][0] assert cursor["total_events"] == 0, cursor assert cursor["raw_status_counts"]["ignored"] == 1, cursor + + +def test_projected_import_keeps_no_raw_copy_but_hooks_and_failures_do(monkeypatch): + body = {"session_id": "prune-s1", "hook_event_name": "UserPromptSubmit", "prompt": "hi", + "timestamp": "2026-07-06T09:00:00Z"} + db.record_ingest("claude", {**body, "_import": True, "_dedup_key": "t:0:prompt"}) + db.record_ingest("claude", {**body, "_import": True, "_dedup_key": "t:0:prompt"}) # duplicate + db.record_ingest("claude", {**body, "prompt": "live", "session_id": "prune-s2"}) + with store.read() as conn: + rows = conn.execute( + "SELECT origin, status, raw_payload IS NULL AS dropped, raw_hash FROM raw_ingest_events ORDER BY id" + ).fetchall() + payload = conn.execute("SELECT payload FROM events WHERE session_id = 'prune-s1'").fetchone()[0] + assert [(r["origin"], r["status"], r["dropped"]) for r in rows] == [ + ("import", "projected", 1), ("import", "duplicate", 0), ("hook", "projected", 0), + ] + assert all(r["raw_hash"] for r in rows) + assert '"prompt": "hi"' in payload # the event row still holds the body + + +def test_prune_clears_existing_projected_import_copies_once(monkeypatch): + monkeypatch.setattr(db, "_PRUNE_BATCH", 2) + for i in range(5): + db.record_ingest("claude", {"session_id": f"old-{i}", "hook_event_name": "UserPromptSubmit", + "prompt": "x", "timestamp": "2026-07-06T09:00:00Z", "_import": True}) + with store.write() as conn: # as an older collector left them + conn.execute("UPDATE raw_ingest_events SET raw_payload = '{}'") + assert db.prune_import_raw_payloads() == 5 + assert db.prune_import_raw_payloads() == 0 + with store.read() as conn: + assert conn.execute("SELECT count(*) FROM raw_ingest_events WHERE raw_payload IS NOT NULL").fetchone()[0] == 0