From 6ffb4a5808d9bb69286d19b9eb02c3d75517b90a Mon Sep 17 00:00:00 2001 From: Yash Dave Date: Tue, 14 Jul 2026 21:21:20 +0530 Subject: [PATCH 1/3] Route live scanners through canonical ingest events --- CONTEXT.md | 42 + backend/tests/test_bridge_live.py | 417 ++++++ backend/tests/test_import.py | 47 + backend/tests/test_raw_ingest_drift.py | 34 + bridge/cot | 1181 +++++++++-------- ...e-scanners-emit-canonical-ingest-events.md | 17 + .../live-scanners-canonical-ingest-events.md | 79 ++ docs/validation/live-scanner-replay.md | 67 + 8 files changed, 1307 insertions(+), 577 deletions(-) create mode 100644 CONTEXT.md create mode 100644 backend/tests/test_bridge_live.py create mode 100644 docs/adr/0002-live-scanners-emit-canonical-ingest-events.md create mode 100644 docs/specs/live-scanners-canonical-ingest-events.md create mode 100644 docs/validation/live-scanner-replay.md diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000..575337f --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,42 @@ +# cot + +Self-hosted observability for coding agents: the Bridge captures agent activity on the developer's machine and the Collector turns it into Sessions, Events, and Insights served to the dashboard. + +## Language + +### Ingest + +**Event**: +One normalized record of agent activity (a hook firing, tool call, prompt, or response) belonging to a Session. +_Avoid_: trace, log entry, record + +**Session**: +One agent working session, identified by the agent's own session id; the unit the dashboard lists and the timeline renders. +_Avoid_: conversation, run (a run is a segment within a session's timeline) + +**Origin**: +An Event's provenance — `hook` when delivered live by an agent hook, `import` when recovered from a transcript on disk. +_Avoid_: source (source is the agent: claude, cursor, codex) + +**Hook data wins**: +The reconciliation policy for a Session that has both origins: live hook Events supersede previously imported Events, which are purged. + +**Ingest Event**: +The canonical, kind-keyed record of one piece of agent activity parsed from a transcript (prompt, response, thought, tool_call, plan, attachment, …) — the same shape regardless of Origin, and the form agent activity takes before it becomes an Event. +_Avoid_: hook payload, synthetic event, artifact (for this shape) + +### Storage + +**Store**: +The single seam through which Events and Sessions are read from and written to disk. Callers may read with SQL; connection lifecycle, write discipline, and the Event row shape belong to the Store alone. +_Avoid_: database layer, repository, DAL + +### Components + +**Collector**: +The self-hosted service that ingests Events and serves the dashboard. +_Avoid_: server, backend (as a noun for the component) + +**Bridge**: +The on-machine CLI that wires agent hooks, scans transcripts, and posts to the Collector. +_Avoid_: agent (reserved for the coding agents being observed), client diff --git a/backend/tests/test_bridge_live.py b/backend/tests/test_bridge_live.py new file mode 100644 index 0000000..fa62ac9 --- /dev/null +++ b/backend/tests/test_bridge_live.py @@ -0,0 +1,417 @@ +"""Canonical Ingest Event and live-shell contract tests for the Bridge.""" + +from __future__ import annotations + +import importlib.machinery +import importlib.util +import json +import os +import sys + +_HERE = os.path.dirname(os.path.abspath(__file__)) +_BACKEND = os.path.dirname(_HERE) +_REPO = os.path.dirname(_BACKEND) + +sys.path.insert(0, _BACKEND) + + +def _load_bridge(): + path = os.path.join(_REPO, "bridge", "cot") + loader = importlib.machinery.SourceFileLoader("cot_bridge_live_under_test", path) + spec = importlib.util.spec_from_loader("cot_bridge_live_under_test", loader) + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + return module + + +bridge = _load_bridge() + + +def _write_jsonl(path, *objects): + path.write_text( + "".join(json.dumps(obj, ensure_ascii=False) + "\n" for obj in objects), + encoding="utf-8", + ) + + +def _capture_posts(monkeypatch): + posted = [] + monkeypatch.setattr(bridge, "_post", lambda url, payload: posted.append((url, payload))) + return posted + + +def _is_ingest(post, source): + return post[0].endswith(f"/v1/ingest/{source}") + + +def test_live_kind_filter_folds_usage_onto_next_kept_event(): + events = [ + {"kind": "tool_call", "usage": {"input_tokens": 12}}, + {"kind": "thought", "text": "checking", "usage": {"output_tokens": 3}}, + {"kind": "prompt", "text": "already delivered", "usage": {"input_tokens": 5}}, + {"kind": "response", "text": "done"}, + ] + + prepared = bridge._prepare_ingest_events("claude", events, origin="hook") # noqa: SLF001 + + assert [event["kind"] for event in prepared] == ["thought", "response"] + assert prepared[0]["usage"] == {"input_tokens": 12, "output_tokens": 3} + assert prepared[1]["usage"] == {"input_tokens": 5} + + +def test_all_filtered_usage_folds_onto_hook_body(tmp_path, monkeypatch): + transcript = tmp_path / "claude-tools-only.jsonl" + _write_jsonl(transcript, { + "type": "assistant", + "uuid": "assistant-1", + "message": { + "role": "assistant", + "usage": {"input_tokens": 20, "output_tokens": 4}, + "content": [ + {"type": "tool_use", "id": "tool-1", "name": "Bash", + "input": {"command": "pwd"}}, + ], + }, + }) + monkeypatch.setattr(bridge, "STATE_DIR", tmp_path / "state") + monkeypatch.setattr(bridge, "TRANSCRIPT_OFFSETS", tmp_path / "state" / "offsets.json") + posted = _capture_posts(monkeypatch) + body = { + "hook_event_name": "Stop", + "session_id": "SID", + "transcript_path": str(transcript), + "timestamp": "2026-07-14T12:00:01Z", + } + + bridge._emit_claude_events(body, "claude") # noqa: SLF001 + + assert posted == [] + assert body["usage"] == {"input_tokens": 20, "output_tokens": 4} + + +def test_shared_post_steps_collapse_responses_and_flag_interruption(): + events = [ + {"kind": "response", "text": "Same words"}, + {"kind": "response", "text": "same words"}, + {"kind": "interruption"}, + ] + + prepared = bridge._prepare_ingest_events("cursor", events, origin="import") # noqa: SLF001 + + assert prepared == [{"kind": "response", "text": "Same words", "interrupted": True}] + + +def test_cursor_subagent_import_prepares_the_whole_transcript(tmp_path, monkeypatch): + transcript = tmp_path / "subagent.jsonl" + _write_jsonl( + transcript, + {"role": "assistant", "message": {"content": [ + {"type": "text", "text": "Repeated response"}, + ]}}, + {"role": "assistant", "message": {"content": [ + {"type": "text", "text": " repeated response "}, + ]}}, + {"role": "assistant", "message": {"content": [ + {"type": "text", "text": "Partial response"}, + {"type": "tool_use", "name": "Shell", "input": {"command": "sleep 10"}}, + ]}}, + { + "type": "turn_ended", + "status": "aborted", + "error": "User aborted/interrupted manually.", + }, + ) + posted = _capture_posts(monkeypatch) + + bridge._import_cursor_subagent_responses(transcript, "CHILD") # noqa: SLF001 + + payloads = [payload for _, payload in posted] + assert [payload["response"] for payload in payloads] == [ + "Repeated response", + "Partial response", + ] + assert payloads[-1]["interrupted"] is True + + +def test_cursor_parser_emits_plan_instant_question_and_live_dedup_keys(): + line = { + "role": "assistant", + "message": { + "content": [ + {"type": "thinking", "thinking": "considering"}, + {"type": "tool_use", "name": "AskQuestion", "input": { + "title": "Choose", "questions": [{"id": "q1", "prompt": "Which?"}], + }}, + {"type": "tool_use", "name": "CreatePlan", "input": { + "name": "Ship", "overview": "Overview", "plan": "Body", "todos": ["one"], + }}, + ] + }, + } + + events = bridge._cursor_line_to_ingest_events( # noqa: SLF001 + line, + "SID", + origin="hook", + byte_offset=128, + path="/transcript.jsonl", + ) + + assert [event["kind"] for event in events] == ["thought", "tool_call", "plan"] + assert events[1]["phase"] == "instant" and events[1]["tool_name"] == "AskQuestion" + assert events[2]["name"] == "Ship" + assert all(event["dedup_key"].startswith("live:cursor:128:") for event in events) + + +def test_plan_adapter_uses_collector_plan_wire_shape(): + payload = bridge._ingest_event_to_hook_payload({ # noqa: SLF001 + "kind": "plan", + "origin": "import", + "session_id": "SID", + "name": "Ship", + "overview": "Overview", + "body": "Body", + "todos": ["one"], + "timestamp": "2026-07-14T12:00:00Z", + }) + + assert payload["hook_event_name"] == "createPlan" + assert payload["_synthetic_category"] == "plan" + assert payload["plan_name"] == "Ship" + assert payload["plan_overview"] == "Overview" + assert payload["plan_body"] == "Body" + assert payload["plan_todos"] == ["one"] + assert payload["_import"] is True + + +def test_cursor_live_shell_posts_canonical_sequence_and_pairs_answer(tmp_path, monkeypatch): + transcript = tmp_path / "cursor.jsonl" + _write_jsonl( + transcript, + {"role": "assistant", "message": {"content": [ + {"type": "text", "text": "Working"}, + {"type": "text", "text": " working "}, + {"type": "thinking", "thinking": "Need a choice"}, + {"type": "tool_use", "name": "AskQuestion", "input": { + "title": "Choose", "questions": [{"id": "q1", "prompt": "Which?"}], + }}, + {"type": "tool_use", "name": "CreatePlan", "input": { + "name": "Ship", "overview": "Overview", "plan": "Body", "todos": ["one"], + }}, + ]}}, + ) + monkeypatch.setattr(bridge, "STATE_DIR", tmp_path / "state") + monkeypatch.setattr(bridge, "TRANSCRIPT_OFFSETS", tmp_path / "state" / "offsets.json") + posted = _capture_posts(monkeypatch) + body = { + "hook_event_name": "afterAgentResponse", + "session_id": "SID", + "transcript_path": str(transcript), + "text": "I will use the first choice", + "timestamp": "2026-07-14T12:00:00Z", + "cwd": "/repo", + "model": "cursor-model", + } + + handled = bridge._emit_cursor_events(body, "cursor") # noqa: SLF001 + + assert handled is True + ingest = [payload for post, payload in posted if post.endswith("/v1/ingest/cursor")] + assert [payload.get("_synthetic_category") or payload.get("tool_name") for payload in ingest] == [ + "response", "thought", "AskQuestion", "plan", "response", + ] + assert [payload["response"] for payload in ingest if payload.get("response")] == [ + "Working", "I will use the first choice", + ] + assert all(payload.get("_dedup_key", "").startswith("live:cursor:") for payload in ingest) + assert [payload["timestamp"] for payload in ingest] == sorted( + payload["timestamp"] for payload in ingest + ) + answers = [payload for post, payload in posted if post.endswith("/v1/questions/answer")] + assert answers == [{ + "session_id": "SID", + "title": "Choose", + "qids": ["q1"], + "response_text": "I will use the first choice", + }] + + +def test_cursor_plan_only_turn_keeps_response_fallback(tmp_path, monkeypatch): + transcript = tmp_path / "cursor-plan.jsonl" + _write_jsonl(transcript, {"role": "assistant", "message": {"content": [ + {"type": "tool_use", "name": "CreatePlan", "input": { + "name": "Ship", "overview": "Overview", "plan": "Plan body", "todos": [], + }}, + ]}}) + monkeypatch.setattr(bridge, "STATE_DIR", tmp_path / "state") + monkeypatch.setattr(bridge, "TRANSCRIPT_OFFSETS", tmp_path / "state" / "offsets.json") + posted = _capture_posts(monkeypatch) + + handled = bridge._emit_cursor_events({ # noqa: SLF001 + "hook_event_name": "afterAgentResponse", + "session_id": "SID", + "transcript_path": str(transcript), + "text": "", + "timestamp": "2026-07-14T12:00:00Z", + }, "cursor") + + assert handled is True + ingest = [payload for post, payload in posted if _is_ingest((post, payload), "cursor")] + assert [payload.get("_synthetic_category") for payload in ingest] == ["plan", "response"] + assert ingest[-1]["response"] == "Plan body" + + +def test_claude_live_shell_uses_shared_parser_and_folds_filtered_usage(tmp_path, monkeypatch): + transcript = tmp_path / "claude.jsonl" + _write_jsonl(transcript, { + "type": "assistant", + "uuid": "assistant-1", + "timestamp": "2026-07-14T12:00:00Z", + "message": { + "role": "assistant", + "usage": {"input_tokens": 20, "output_tokens": 4}, + "content": [ + {"type": "tool_use", "id": "tool-1", "name": "Bash", "input": {"command": "pwd"}}, + {"type": "text", "text": "Done"}, + ], + }, + }) + monkeypatch.setattr(bridge, "STATE_DIR", tmp_path / "state") + monkeypatch.setattr(bridge, "TRANSCRIPT_OFFSETS", tmp_path / "state" / "offsets.json") + posted = _capture_posts(monkeypatch) + + bridge._emit_claude_events({ # noqa: SLF001 + "hook_event_name": "Stop", + "session_id": "SID", + "transcript_path": str(transcript), + "timestamp": "2026-07-14T12:00:01Z", + "cwd": "/repo", + "model": "claude-model", + }, "claude") + + ingest = [payload for post, payload in posted if post.endswith("/v1/ingest/claude")] + assert len(ingest) == 1 + assert ingest[0]["response"] == "Done" + assert ingest[0]["usage"] == {"input_tokens": 20, "output_tokens": 4} + assert ingest[0]["_dedup_key"] == "live:claude:assistant-1:resp:1" + assert ingest[0]["cwd"] == "/repo" and ingest[0]["model"] == "claude-model" + + +def test_live_rescan_after_offset_loss_reuses_explicit_keys(tmp_path, monkeypatch): + transcript = tmp_path / "cursor-rescan.jsonl" + _write_jsonl(transcript, {"role": "assistant", "message": {"content": [ + {"type": "text", "text": "Stable"}, + ]}}) + monkeypatch.setattr(bridge, "STATE_DIR", tmp_path / "state") + offsets = tmp_path / "state" / "offsets.json" + monkeypatch.setattr(bridge, "TRANSCRIPT_OFFSETS", offsets) + + first = bridge._scan_live_ingest_events( # noqa: SLF001 + "cursor", str(transcript), "SID", "2026-07-14T12:00:00Z" + ) + offsets.unlink() + second = bridge._scan_live_ingest_events( # noqa: SLF001 + "cursor", str(transcript), "SID", "2026-07-14T12:00:00Z" + ) + + assert [event["dedup_key"] for event in first] == [event["dedup_key"] for event in second] + assert all(event["dedup_key"].startswith("live:cursor:") for event in first) + + +def test_cursor_question_without_hook_text_posts_no_answer(tmp_path, monkeypatch): + transcript = tmp_path / "cursor-question.jsonl" + _write_jsonl(transcript, {"role": "assistant", "message": {"content": [ + {"type": "tool_use", "name": "AskQuestion", "input": { + "title": "Choose", "questions": [{"id": "q1", "prompt": "Which?"}], + }}, + ]}}) + monkeypatch.setattr(bridge, "STATE_DIR", tmp_path / "state") + monkeypatch.setattr(bridge, "TRANSCRIPT_OFFSETS", tmp_path / "state" / "offsets.json") + posted = _capture_posts(monkeypatch) + + handled = bridge._emit_cursor_events({ # noqa: SLF001 + "hook_event_name": "afterAgentResponse", + "session_id": "SID", + "transcript_path": str(transcript), + "text": "", + "timestamp": "2026-07-14T12:00:00Z", + }, "cursor") + + assert handled is False + ingest = [payload for post, payload in posted if post.endswith("/v1/ingest/cursor")] + assert len(ingest) == 1 and ingest[0]["tool_name"] == "AskQuestion" + assert not any(post.endswith("/v1/questions/answer") for post, _ in posted) + + +def test_codex_live_shell_emits_thought_response_usage_and_stable_keys(tmp_path, monkeypatch): + transcript = tmp_path / "codex.jsonl" + _write_jsonl( + transcript, + {"type": "response_item", "id": "thought-1", "timestamp": "2026-07-14T12:00:00Z", + "payload": {"type": "message", "role": "assistant", "phase": "commentary", + "content": [{"type": "output_text", "text": "Checking"}]}}, + {"type": "response_item", "id": "tool-1", "timestamp": "2026-07-14T12:00:01Z", + "payload": {"type": "function_call", "name": "exec_command", "call_id": "call-1", + "arguments": "{\"cmd\": \"pwd\"}"}}, + {"type": "response_item", "id": "response-1", "timestamp": "2026-07-14T12:00:02Z", + "payload": {"type": "message", "role": "assistant", "phase": "final_answer", + "content": [{"type": "output_text", "text": "Done"}]}}, + {"type": "event_msg", "timestamp": "2026-07-14T12:00:03Z", + "payload": {"type": "token_count", "info": {"last_token_usage": { + "input_tokens": 30, "output_tokens": 5, + }}}}, + ) + monkeypatch.setattr(bridge, "STATE_DIR", tmp_path / "state") + monkeypatch.setattr(bridge, "TRANSCRIPT_OFFSETS", tmp_path / "state" / "offsets.json") + posted = _capture_posts(monkeypatch) + + bridge._emit_codex_events({ # noqa: SLF001 + "hook_event_name": "Stop", + "session_id": "SID", + "transcript_path": str(transcript), + "timestamp": "2026-07-14T12:00:04Z", + }, "codex") + + ingest = [payload for post, payload in posted if post.endswith("/v1/ingest/codex")] + assert [payload.get("_synthetic_category") for payload in ingest] == ["thought", "response"] + assert ingest[1]["usage"] == {"input_tokens": 30, "output_tokens": 5} + assert [payload["_dedup_key"] for payload in ingest] == [ + "live:codex:thought-1:think", "live:codex:response-1:resp", + ] + + +def test_codex_stop_falls_back_when_transcript_is_unreadable(monkeypatch): + posted = _capture_posts(monkeypatch) + + bridge._emit_codex_events({ # noqa: SLF001 + "hook_event_name": "Stop", + "session_id": "SID", + "transcript_path": "/missing/transcript.jsonl", + "last_assistant_message": "Fallback answer", + "timestamp": "2026-07-14T12:00:00Z", + }, "codex") + + assert len(posted) == 1 + payload = posted[0][1] + assert payload["response"] == "Fallback answer" + assert payload["_dedup_key"].startswith("live:codex:fallback:") + + +def test_all_shared_parsers_emit_canonical_interruption_markers(): + claude = bridge._claude_line_to_ingest_events({ # noqa: SLF001 + "type": "user", "uuid": "u1", "message": {"role": "user", "content": [ + {"type": "text", "text": "[Request interrupted by user]"}, + ]}, + }, "SID") + cursor = bridge._cursor_line_to_ingest_events({ # noqa: SLF001 + "role": "user", "message": {"content": [ + {"type": "text", "text": "[Request interrupted by user]"}, + ]}, + }, "SID", path="/cursor.jsonl") + codex = bridge._codex_line_to_ingest_events({ # noqa: SLF001 + "type": "event_msg", "payload": {"type": "turn_aborted"}, + }, "SID") + + assert [event["kind"] for event in claude] == ["interruption"] + assert [event["kind"] for event in cursor] == ["interruption"] + assert [event["kind"] for event in codex] == ["interruption"] diff --git a/backend/tests/test_import.py b/backend/tests/test_import.py index 55f4933..389b444 100644 --- a/backend/tests/test_import.py +++ b/backend/tests/test_import.py @@ -214,6 +214,53 @@ def test_claude_tool_result_paired_to_call(): assert post and "file1" in str(post[0].get("tool_response")), hooks +def test_import_keeps_historical_dedup_scope_for_identifierless_records(): + claude = bridge._claude_line_to_ingest_events({ # noqa: SLF001 + "type": "assistant", + "message": {"role": "assistant", "content": "done"}, + }, "SID", path="/claude.jsonl") + codex = bridge._codex_line_to_ingest_events({ # noqa: SLF001 + "type": "response_item", + "payload": {"type": "message", "role": "user", "content": [ + {"type": "input_text", "text": "hello"}, + ]}, + }, "SID", path="/codex.jsonl") + cursor = bridge._cursor_line_to_ingest_events({ # noqa: SLF001 + "role": "user", + "message": {"content": [ + {"type": "text", "text": "see this"}, + {"type": "image", "source": {"media_type": "image/png"}}, + ]}, + }, "SID", path="/cursor.jsonl") + + assert claude[0]["dedup_key"] is None + assert codex[0]["dedup_key"] is None + assert [event["dedup_key"] for event in cursor if event["kind"] == "attachment"] == [None] + + +def test_claude_import_recovers_matching_standalone_file_attachment(): + state: dict = {} + prompt = bridge._claude_line_to_ingest_events({ # noqa: SLF001 + "type": "user", + "uuid": "prompt-1", + "message": {"role": "user", "content": "Review @report.pdf"}, + }, "SID", state=state) + attachment = bridge._claude_line_to_ingest_events({ # noqa: SLF001 + "type": "attachment", + "parentUuid": "prompt-1", + "attachment": { + "type": "file", + "filename": "report.pdf", + "content": {"file": {"filePath": "/tmp/report.pdf"}}, + }, + }, "SID", state=state) + + assert prompt[0]["kind"] == "prompt" + assert attachment[0]["kind"] == "attachment" + assert attachment[0]["text"] == "Review @report.pdf" + assert attachment[0]["dedup_key"] is None + + def test_claude_tool_result_error_status(): lines = [ {"type": "assistant", "uuid": "u1", "timestamp": "2026-06-01T00:00:00Z", diff --git a/backend/tests/test_raw_ingest_drift.py b/backend/tests/test_raw_ingest_drift.py index 3d0f46d..a9b647a 100644 --- a/backend/tests/test_raw_ingest_drift.py +++ b/backend/tests/test_raw_ingest_drift.py @@ -55,6 +55,40 @@ def test_record_ingest_persists_raw_before_projection_and_ignored_rows(): assert events[0]["category"] == "prompt" +def test_namespaced_live_event_supersedes_import_and_retry_is_idempotent(): + sid = "reconcile-s1" + imported = { + "session_id": sid, + "hook_event_name": "afterAgentResponse", + "response": "Imported approximation", + "_synthetic_category": "response", + "_import": True, + "_dedup_key": "/transcript.jsonl:0:resp:0", + "timestamp": "2026-07-14T12:00:00Z", + } + live = { + "session_id": sid, + "hook_event_name": "afterAgentResponse", + "response": "Hook truth", + "_synthetic_category": "response", + "_dedup_key": "live:cursor:0:resp:0", + "timestamp": "2026-07-14T12:00:01Z", + } + + assert db.record_ingest("cursor", imported)["raw_status"] == "projected" + assert db.record_ingest("cursor", live)["raw_status"] == "projected" + assert db.record_ingest("cursor", live)["raw_status"] == "duplicate" + + detail = db.get_session_detail(sid) + assert detail is not None + responses = [ + event["detail"] for event in detail["events"] + if event["category"] == "response" + ] + assert responses == ["Hook truth"] + assert db.session_origins()[sid] == "hook" + + def test_malformed_raw_input_is_evidence_not_a_timeline_event(): result = db.record_malformed_ingest( "codex", diff --git a/bridge/cot b/bridge/cot index 785de4f..49267b5 100755 --- a/bridge/cot +++ b/bridge/cot @@ -37,8 +37,9 @@ DEFAULT_ENDPOINT = "http://127.0.0.1:31337" # reimport. (1 = original; 2 = dedicated cursor parser, tool results, mtime ts, # meta categorization; 3 = codex token usage, cwd/model, interruptions, and a # SessionStart carrying sandbox/approval posture; 4 = codex environment_context -# lifted into its own system event instead of leaking as a prompt.) -IMPORT_PARSER_VERSION = "4" +# lifted into its own system event instead of leaking as a prompt; 5 = shared +# live/import parsers, Cursor plans, interruption markers, and response collapse.) +IMPORT_PARSER_VERSION = "5" # Skip transcript files larger than this. Guards against a runaway/hostile file # exhausting memory; real transcripts are far smaller. @@ -443,6 +444,7 @@ def _is_interruption(text: str | None) -> bool: t = text.lower() return ( "request interrupted by user" in t + or "user aborted" in t or "" in t or "interrupted the previous turn" in t ) @@ -467,482 +469,373 @@ def _merge_usage(acc: dict | None, usage: dict) -> dict: return out -def _read_transcript_delta(transcript_path: str, session_id: str) -> str: - """Return newly appended transcript bytes since the last read for this session.""" +def _response_equality_key(text: object) -> str: + return " ".join(str(text or "").lower().split()) + + +def _read_live_transcript_lines(transcript_path: str, session_id: str) -> list[tuple[int, str]]: + """Read complete new JSONL records with their absolute byte offsets.""" path = Path(transcript_path) - if not path.exists(): - return "" + try: + size = path.stat().st_size + except OSError: + return [] + if size > _MAX_TRANSCRIPT_BYTES: + return [] offsets = _load_offsets() key = f"{session_id}:{transcript_path}" start = offsets.get(key, 0) + if start > size: + start = 0 try: - text = path.read_text(encoding="utf-8", errors="replace") + with path.open("rb") as fh: + fh.seek(start) + chunk = fh.read() except OSError: - return "" - if start > len(text): - start = 0 - new_text = text[start:] - offsets[key] = len(text) + return [] + last_nl = chunk.rfind(b"\n") + if last_nl < 0: + return [] + consumed = chunk[: last_nl + 1] + out: list[tuple[int, str]] = [] + relative = 0 + for raw_line in consumed.splitlines(keepends=True): + absolute = start + relative + relative += len(raw_line) + line = raw_line.rstrip(b"\r\n").decode("utf-8", errors="replace").strip() + if line: + out.append((absolute, line)) + offsets[key] = start + len(consumed) _save_offsets(offsets) - return new_text + return out -def _scan_transcript(transcript_path: str, session_id: str) -> dict: - """Read the transcript delta once; extract assistant responses and the - attachments (images/files) on user messages. Single read so the shared - file offset only advances once. +def _origin_dedup_key(origin: str, source: str, key: str | None) -> str | None: + if not key: + return None + return key if origin == "import" else f"live:{source}:{key}" - Returns {"responses": [(ts, text, model, usage, interrupted)], - "thoughts": [(ts, text, usage, interrupted)], "attachments": [(ts, [att])]}. - """ - empty: dict[str, list] = {"responses": [], "thoughts": [], "attachments": []} - new_text = _read_transcript_delta(transcript_path, session_id) - if not new_text: - return empty - - # Ordered records so a user-stop can flag the message it cut off (the last - # assistant thought/response before the interruption marker). - msgs: list[dict] = [] - attachments: list[tuple] = [] - user_prompts: dict[str, str] = {} - pending_usage: dict | None = None - seen_msg_ids: set[str] = set() - for line in new_text.splitlines(): - line = line.strip() - if not line: + +def _scoped_dedup_key( + origin: str, + source: str, + suffix: str, + *, + import_anchor: str | None, + live_anchor: str | None, + live_only: bool = False, +) -> str | None: + """Apply the shared import/live dedup scope to one Ingest Event.""" + if live_only and origin != "hook": + return None + anchor = import_anchor if origin == "import" else live_anchor + return _origin_dedup_key(origin, source, f"{anchor}:{suffix}" if anchor else None) + + +def _keep_live_event(source: str, event: dict) -> bool: + kind = event.get("kind") + if kind in ("response", "thought", "attachment", "interruption", "plan"): + return True + if kind == "tool_call": + name = event.get("tool_name") + return (source == "cursor" and name == "AskQuestion") or ( + source == "codex" and name == "WebSearch" + ) + return False + + +def _filter_live_events_with_usage(source: str, events: list[dict]) -> list[dict]: + """Drop hook-delivered kinds while preserving every parsed usage delta.""" + kept: list[dict] = [] + pending: dict | None = None + for original in events: + event = dict(original) + usage = event.get("usage") if isinstance(event.get("usage"), dict) else None + if not _keep_live_event(source, event): + if usage: + pending = _merge_usage(pending, usage) continue - try: - obj = json.loads(line) - except json.JSONDecodeError: + if pending: + event["usage"] = _merge_usage(pending, usage or {}) + pending = None + kept.append(event) + if pending and kept: + last = dict(kept[-1]) + last["usage"] = _merge_usage(last.get("usage"), pending) + kept[-1] = last + elif pending: + # No transcript-only Event exists in this delta. Keep the tokens inside + # the canonical list as session metadata; the live shell folds this + # marker onto the hook body before the body is posted. + kept.append({ + "kind": "session_metadata", + "metadata_type": "usage", + "origin": "hook", + "usage": pending, + }) + return kept + + +def _collapse_consecutive_responses(events: list[dict]) -> list[dict]: + out: list[dict] = [] + for event in events: + if ( + event.get("kind") == "response" + and out + and out[-1].get("kind") == "response" + and _response_equality_key(out[-1].get("text")) + == _response_equality_key(event.get("text")) + ): continue - message = obj.get("message") if isinstance(obj.get("message"), dict) else {} - role = obj.get("role") or message.get("role") or obj.get("type") - content = obj.get("content") - if content is None: - content = message.get("content") - ts = obj.get("timestamp") - - if role == "assistant" or obj.get("type") == "assistant": - msg_id = message.get("id") - if isinstance(content, list): - for block in content: - if isinstance(block, dict) and block.get("type") == "thinking": - think = block.get("thinking") - if think and not (msg_id and f"t:{msg_id}" in seen_msg_ids): - msgs.append({"kind": "thought", "ts": ts, "text": str(think), - "model": None, "usage": None, "interrupted": False}) - if msg_id: - seen_msg_ids.add(f"t:{msg_id}") - text_val = _text_from_blocks(content) if isinstance(content, list) else (content or "") - model = message.get("model") or obj.get("model") - usage = message.get("usage") if isinstance(message.get("usage"), dict) else None - if text_val and not (msg_id and f"r:{msg_id}" in seen_msg_ids): - if pending_usage: - usage = _merge_usage(pending_usage, usage) if usage else pending_usage - pending_usage = None - msgs.append({"kind": "response", "ts": ts, "text": str(text_val), - "model": model, "usage": usage, "interrupted": False}) - if msg_id: - seen_msg_ids.add(f"r:{msg_id}") - elif usage: - pending_usage = _merge_usage(pending_usage, usage) - elif role == "user" and isinstance(content, list): - user_text = _text_from_blocks(content, clean_user=True) - uuid = obj.get("uuid") - if uuid and user_text: - user_prompts[str(uuid)] = user_text - if _is_interruption(user_text) and msgs: - msgs[-1]["interrupted"] = True - atts = _message_attachments(content) - if atts: - attachments.append((ts, user_text, atts)) - elif obj.get("type") == "attachment": - att = obj.get("attachment") if isinstance(obj.get("attachment"), dict) else {} - meta = _file_attachment_meta(att) - parent_uuid = obj.get("parentUuid") - parent_text = user_prompts.get(str(parent_uuid)) if parent_uuid else None - if meta and _file_attachment_matches_text(att, parent_text): - attachments.append((ts, parent_text, [meta])) - - if pending_usage and msgs: - last = msgs[-1] - last["usage"] = _merge_usage(pending_usage, last["usage"]) if last["usage"] else pending_usage - - thoughts = [(m["ts"], m["text"], m["usage"], m["interrupted"]) for m in msgs if m["kind"] == "thought"] - responses = [(m["ts"], m["text"], m["model"], m["usage"], m["interrupted"]) for m in msgs if m["kind"] == "response"] - return {"responses": responses, "thoughts": thoughts, "attachments": attachments} - - -def _scan_codex_transcript(transcript_path: str, session_id: str) -> dict: - """Read Codex rollout JSONL deltas. - - Codex mirrors each assistant line twice (``agent_message`` then - ``output_text`` with identical text). Only ingest the ``response_item`` - and split by ``phase``: ``commentary`` → thought, ``final_answer`` → response. - """ - empty: dict[str, list] = { - "thoughts": [], - "responses": [], - "attachments": [], - "web_searches": [], - } - new_text = _read_transcript_delta(transcript_path, session_id) - if not new_text: - return empty - - # Each emitted assistant message gets a mutable record so we can fold the - # per-turn ``token_count`` deltas (which arrive after the message) onto it. - msgs: list[dict] = [] - attachments: list[tuple] = [] - web_searches: list[tuple] = [] - for line in new_text.splitlines(): - line = line.strip() - if not line: + out.append(event) + return out + + +def _apply_interruption_markers(events: list[dict]) -> list[dict]: + out: list[dict] = [] + for event in events: + if event.get("kind") != "interruption": + out.append(event) continue + if out: + prior = dict(out[-1]) + prior["interrupted"] = True + out[-1] = prior + return out + + +def _prepare_ingest_events( + source: str, + events: list[dict], + *, + origin: str, + hook_timestamp: object = None, +) -> list[dict]: + """Apply the shared pure event-list transformations for either Origin.""" + prepared = [dict(event) for event in events] + if origin == "hook": + prepared = _filter_live_events_with_usage(source, prepared) + # Only Cursor is configured for this shared transform: its transcript + # format is the one observed to repeat identical commentary blocks. Both + # Cursor Origins pass through this gate; enabling it for another source is + # a policy choice, not another parser implementation. + if source == "cursor": + prepared = _collapse_consecutive_responses(prepared) + prepared = _apply_interruption_markers(prepared) + if origin == "hook" and source == "cursor": + prepared = [ + {**event, "timestamp": _ordered_ts(hook_timestamp, index)} + for index, event in enumerate(prepared) + ] + return prepared + + +def _scan_live_ingest_events( + source: str, + transcript_path: str, + session_id: str, + hook_timestamp: object, +) -> list[dict]: + """Parse one transcript delta through the same line-parser as import.""" + parser = _ingest_line_parser(source) + state: dict = {} + events: list[dict] = [] + path = str(transcript_path) + for lineno, (byte_offset, line) in enumerate( + _read_live_transcript_lines(transcript_path, session_id) + ): try: obj = json.loads(line) except json.JSONDecodeError: continue - ts = obj.get("timestamp") - kind = obj.get("type") - payload = obj.get("payload") if isinstance(obj.get("payload"), dict) else {} - - if kind == "response_item": - pt = payload.get("type") - if pt == "message" and payload.get("role") == "assistant": - text_val = _text_from_blocks(payload.get("content")) - if not text_val: - continue - model = payload.get("model") or obj.get("model") - phase = payload.get("phase") or "" - # final_answer → response; commentary / unknown phase → thought. - msgs.append({ - "kind": "response" if phase == "final_answer" else "thought", - "ts": ts, - "text": text_val, - "model": model, - "usage": None, - "interrupted": False, - }) - elif pt == "message" and payload.get("role") == "user": - content = payload.get("content") - if isinstance(content, list): - user_text = _text_from_blocks(content, clean_user=True) - if _is_interruption(user_text) and msgs: - msgs[-1]["interrupted"] = True - atts = _message_attachments(content) - if atts: - attachments.append((ts, user_text, atts)) - elif pt == "web_search_call": - action = payload.get("action") if isinstance(payload.get("action"), dict) else {} - query = action.get("query") or payload.get("query") or "" - if query: - web_searches.append((ts, str(query), action, payload.get("status"))) - elif kind == "event_msg" and payload.get("type") == "token_count": - # Attribute the per-turn token delta to the message it followed. - info = payload.get("info") if isinstance(payload.get("info"), dict) else {} - last = info.get("last_token_usage") if isinstance(info.get("last_token_usage"), dict) else None - if last and msgs: - msgs[-1]["usage"] = _merge_usage(msgs[-1].get("usage"), last) + events.extend( + parser( + obj, + session_id, + lineno=lineno, + state=state, + path=path, + origin="hook", + byte_offset=byte_offset, + ) + ) + if source == "codex": + events.extend(_codex_flush_pending_ingest_events(state)) + return _prepare_ingest_events( + source, events, origin="hook", hook_timestamp=hook_timestamp + ) - thoughts = [(m["ts"], m["text"], m["usage"], m["interrupted"]) for m in msgs if m["kind"] == "thought"] - responses = [(m["ts"], m["text"], m["model"], m["usage"], m["interrupted"]) for m in msgs if m["kind"] == "response"] + +def _ingest_line_parser(source: str): return { - "thoughts": thoughts, - "responses": responses, - "attachments": attachments, - "web_searches": web_searches, - } + "claude": _claude_line_to_ingest_events, + "cursor": _cursor_line_to_ingest_events, + "codex": _codex_line_to_ingest_events, + }[source] -def _emit_synthetic_messages( +def _import_line_to_hook_payloads( source: str, - body: dict, - scanned: dict, -) -> None: - session_id = body.get("session_id") - if not session_id: - return - for ts, text, atts in scanned.get("attachments", []): - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "_attach_to_prompt": True, - "text": text, - "attachments": atts, - "timestamp": ts or body.get("timestamp"), - }, - ) - for ts, msg, usage, interrupted in scanned.get("thoughts", []): - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": "afterAgentThought", - "thought": msg, - "_synthetic_category": "thought", - "timestamp": ts or body.get("timestamp"), - "model": body.get("model"), - "usage": usage, - "interrupted": interrupted, - }, - ) - for ts, msg, model, usage, interrupted in scanned.get("responses", []): - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": "afterAgentResponse", - "response": msg, - "_synthetic_category": "response", - "timestamp": ts or body.get("timestamp"), - "model": model or body.get("model"), - "usage": usage, - "interrupted": interrupted, - }, - ) - for ts, query, action, status in scanned.get("web_searches", []): + obj: dict, + session_id: str, + *, + lineno: int = 0, + mtime: float | None = None, + state: dict | None = None, + path: str = "", +) -> list[dict]: + """Run one import line through the authoritative canonical pipeline.""" + events = _ingest_line_parser(source)( + obj, + session_id, + lineno=lineno, + mtime=mtime, + state=state, + path=path, + origin="import", + ) + return _ingest_events_to_hook_payloads( + _prepare_ingest_events(source, events, origin="import") + ) + + +def _enrich_live_events(events: list[dict], body: dict) -> list[dict]: + out: list[dict] = [] + for original in events: + event = dict(original) + event.setdefault("origin", "hook") + if not event.get("session_id"): + event["session_id"] = body.get("session_id") + if not event.get("cwd") and body.get("cwd"): + event["cwd"] = body.get("cwd") + if not event.get("model") and body.get("model"): + event["model"] = body.get("model") + if not event.get("composer_mode") and body.get("composer_mode"): + event["composer_mode"] = body.get("composer_mode") + out.append(event) + return out + + +def _fold_usage_metadata_into_hook_body(events: list[dict], body: dict) -> list[dict]: + """Fold internal usage metadata onto the hook body and remove the marker.""" + out: list[dict] = [] + for event in events: + if event.get("kind") == "session_metadata" and event.get("metadata_type") == "usage": + usage = event.get("usage") if isinstance(event.get("usage"), dict) else {} + if usage: + body["usage"] = _merge_usage(body.get("usage"), usage) + continue + out.append(event) + return out + + +def _post_ingest_events(source: str, events: list[dict]) -> None: + for event in events: _post( f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": "PostToolUse", - "tool_name": "WebSearch", - "tool_input": {"query": query, "queries": action.get("queries", [])}, - "tool_response": {"status": status, "action": action}, - "timestamp": ts or body.get("timestamp"), - "model": body.get("model"), - }, + _ingest_event_to_hook_payload(event), ) -def _emit_claude_responses(body: dict, source: str) -> None: - hook = body.get("hook_event_name") or body.get("hook") or "" - if hook not in ("Stop", "SubagentStop"): - return - transcript = body.get("transcript_path") - session_id = body.get("session_id") - if not transcript or not session_id: - return - scanned = _scan_transcript(transcript, session_id) - _emit_synthetic_messages(source, body, scanned) - - -def _collapse_response_text(text: object) -> str: - return " ".join(str(text or "").lower().split()) +def _shell_dedup_key(source: str, label: str, body: dict, text: str) -> str: + raw = f"{body.get('timestamp') or ''}\0{text}".encode("utf-8", "replace") + return f"live:{source}:{label}:{hashlib.sha1(raw).hexdigest()[:16]}" -def _scan_cursor_artifacts(transcript_path: str, session_id: str, response_text: str = "") -> dict: - """Extract Cursor transcript-only artifacts. +def _shell_response_event( + source: str, + body: dict, + text: str, + *, + label: str | None = None, + timestamp: object = None, + composer_mode: str | None = None, + dedup_key: str | None = None, +) -> dict: + """Build the canonical response shape used by live-shell fallbacks.""" + if dedup_key is None: + if not label: + raise ValueError("shell response requires a label or explicit dedup key") + dedup_key = _shell_dedup_key(source, label, body, text) + return { + "kind": "response", + "origin": "hook", + "session_id": body.get("session_id"), + "cwd": body.get("cwd"), + "model": body.get("model"), + "composer_mode": composer_mode or body.get("composer_mode"), + "text": text, + "timestamp": timestamp if timestamp is not None else body.get("timestamp"), + "dedup_key": dedup_key, + } - In plan mode Cursor generates the plan through an internal ``CreatePlan`` - tool that fires no tool hooks; ``afterAgentResponse`` then arrives with empty - text. Cursor question prompts similarly appear as ``AskQuestion`` tool_use - blocks without pre/post tool hooks. Recover both in one transcript pass so - one artifact scanner does not consume another scanner's offset.""" - new_text = _read_transcript_delta(transcript_path, session_id) - out: dict[str, list] = {"plans": [], "questions": [], "attachments": [], "responses": []} - if not new_text: - return out - plans: list[dict] = [] - questions: list[dict] = [] - attachments: list[tuple] = [] - responses: list[str] = [] - for line in new_text.splitlines(): - line = line.strip() - if not line: - continue - try: - obj = json.loads(line) - except json.JSONDecodeError: - continue - message = obj.get("message") if isinstance(obj.get("message"), dict) else {} - content = obj.get("content") - if content is None: - content = message.get("content") - if not isinstance(content, list): - continue - role = obj.get("role") or message.get("role") or obj.get("type") - if role == "user": - atts = _message_attachments(content) - if atts: - attachments.append((obj.get("timestamp"), _text_from_blocks(content, clean_user=True), atts)) - for block in content: - if not isinstance(block, dict): - continue - # Cursor's afterAgentResponse hook only delivers the final text per - # turn, so the intermediate commentary between tool calls lives only - # here. Collect every assistant text block, in order. - if block.get("type") == "text" and role == "assistant": - t = str(block.get("text") or "").strip() - if t: - responses.append(t) - continue - if block.get("type") != "tool_use": - continue - name = block.get("name") - if name == "CreatePlan": - inp = block.get("input") if isinstance(block.get("input"), dict) else {} - plans.append( - { - "name": inp.get("name") or "Plan", - "overview": inp.get("overview") or "", - "plan": inp.get("plan") or "", - "todos": inp.get("todos") if isinstance(inp.get("todos"), list) else [], - } - ) - elif name == "AskQuestion": - inp = block.get("input") if isinstance(block.get("input"), dict) else {} - questions.append( - { - "input": inp, - "response_text": response_text, - } - ) - out["plans"] = plans - out["questions"] = questions - out["attachments"] = attachments - out["responses"] = responses - return out +def _emit_claude_events(body: dict, source: str) -> None: + hook = body.get("hook_event_name") or body.get("hook") or "" + transcript = body.get("transcript_path") + session_id = body.get("session_id") + if hook not in ("Stop", "SubagentStop") or not transcript or not session_id: + return + events = _scan_live_ingest_events(source, transcript, session_id, body.get("timestamp")) + events = _fold_usage_metadata_into_hook_body(events, body) + _post_ingest_events(source, _enrich_live_events(events, body)) -def _emit_cursor_artifacts(body: dict, source: str) -> bool: - """Recover Cursor artifacts that are present only in transcript JSONL. - Returns True when this was a plan turn and the function has already posted - both the (text-filled) response and a structured plan event, so the caller - skips the default post.""" +def _emit_cursor_events(body: dict, source: str) -> bool: + """Run Cursor's shared parser and retain only trigger-context behavior.""" hook = body.get("hook_event_name") or body.get("hook") or "" - # afterAgentResponse carries the turn-final text; stop is a backstop that - # flushes any transcript text written after the last afterAgentResponse. - if hook not in ("afterAgentResponse", "stop"): - return False transcript = body.get("transcript_path") session_id = body.get("session_id") - if not transcript or not session_id: + if hook not in ("afterAgentResponse", "stop") or not transcript or not session_id: return False - scanned = _scan_cursor_artifacts(transcript, session_id, str(body.get("text") or body.get("response") or "")) - plans = scanned.get("plans", []) - questions = scanned.get("questions", []) - for ts, text, atts in scanned.get("attachments", []): - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "_attach_to_prompt": True, - "text": text, - "attachments": atts, - "timestamp": ts or body.get("timestamp"), - }, - ) - for q in questions: - for hook_name in ("preToolUse", "postToolUse"): - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": hook_name, - "tool_name": "AskQuestion", - "tool_input": q["input"], - "tool_response": {}, - "composer_mode": body.get("composer_mode"), - "model": body.get("model"), - "timestamp": body.get("timestamp"), - }, - ) - if str(q.get("response_text") or "").strip(): - _post( - f"{COT_ENDPOINT}/v1/questions/answer", - _question_answer_payload(session_id, q["input"], q["response_text"]), - ) - # Emit every assistant text block from the transcript delta. Cursor's - # afterAgentResponse hook carries only the turn-final text, so this is the - # only place the intermediate commentary is available. - # Deduplicate consecutive identical texts: Cursor transcripts often repeat - # the same brief commentary across many assistant messages within a turn - # (e.g. 56 identical blocks), and the final response can appear twice. - responses = scanned.get("responses", []) - deduped: list[str] = [] - for r in responses: - if not deduped or _collapse_response_text(deduped[-1]) != _collapse_response_text(r): - deduped.append(r) + events = _scan_live_ingest_events(source, transcript, session_id, body.get("timestamp")) + events = _fold_usage_metadata_into_hook_body(events, body) + events = _enrich_live_events(events, body) body_text = str(body.get("text") or body.get("response") or "").strip() - emit_list = list(deduped) - # Safety net: if the hook's final text wasn't in the delta yet, include it - # (deduped against the last transcript block). - if body_text and (not emit_list or _collapse_response_text(emit_list[-1]) != _collapse_response_text(body_text)): - emit_list.append(body_text) - for i, text in enumerate(emit_list): - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": "afterAgentResponse", - "_synthetic_category": "response", - "response": text, - "composer_mode": body.get("composer_mode"), - "model": body.get("model"), - "timestamp": _ordered_ts(body.get("timestamp"), i), - }, - ) - # Structured plan artifact (name + overview + todo checklist). A distinct - # hook name routes it via _synthetic_category, not afterAgentResponse. - for p in plans: - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": "createPlan", - "_synthetic_category": "plan", - "plan_name": p["name"], - "plan_overview": p["overview"], - "plan_body": p["plan"], - "plan_todos": p["todos"], - "composer_mode": "plan", - "model": body.get("model"), - "timestamp": body.get("timestamp"), - }, - ) - - # Plan-only turn with no text response: surface the plan body so the - # conversation view isn't blank. - if plans and not emit_list: - last = plans[-1] - plan_text = last.get("plan") or last.get("overview") or "" + questions = [ + event for event in events + if event.get("kind") == "tool_call" and event.get("tool_name") == "AskQuestion" + ] + responses = [event for event in events if event.get("kind") == "response"] + plans = [event for event in events if event.get("kind") == "plan"] + if body_text and ( + not responses + or _response_equality_key(responses[-1].get("text")) + != _response_equality_key(body_text) + ): + events.append(_shell_response_event( + source, + body, + body_text, + label="safety", + timestamp=_ordered_ts(body.get("timestamp"), len(events)), + )) + responses.append(events[-1]) + if plans and not responses: + plan = plans[-1] + plan_text = str(plan.get("body") or plan.get("overview") or "").strip() if plan_text: + events.append(_shell_response_event( + source, + body, + plan_text, + composer_mode="plan", + timestamp=_ordered_ts(body.get("timestamp"), len(events)), + dedup_key=f"{plan.get('dedup_key')}:fallback", + )) + responses.append(events[-1]) + + _post_ingest_events(source, events) + if body_text: + for question in questions: _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": "afterAgentResponse", - "_synthetic_category": "response", - "response": plan_text, - "composer_mode": "plan", - "model": body.get("model"), - "timestamp": body.get("timestamp"), - }, + f"{COT_ENDPOINT}/v1/questions/answer", + _question_answer_payload( + session_id, question.get("tool_input") or {}, body_text + ), ) - emit_list.append(plan_text) - - # On 'stop', let the caller still post the lifecycle "Session ended" body — - # we only emitted responses here, which don't duplicate it. - if hook == "stop": - return False - # On afterAgentResponse we've emitted the response(s) ourselves; tell the - # caller to skip the default body post (which would duplicate the final). - # Only when we emitted nothing do we let the caller post the raw body. - return bool(emit_list or plans) + return hook != "stop" and bool(responses or plans) def _find_codex_transcript(session_id: str) -> str | None: @@ -1006,42 +899,26 @@ def _codex_read_session_posture(session_id: str) -> dict: return out -def _emit_codex_responses(body: dict, source: str) -> None: - """Codex stores commentary and assistant text in rollout JSONL transcripts. - - Scan the transcript so intermediate thoughts, responses, web searches, and - token usage appear while the turn is in progress. The transcript is located - from the session id when the hook payload omits ``transcript_path``. Fall - back to ``last_assistant_message`` on Stop when the transcript can't be read. - """ +def _emit_codex_events(body: dict, source: str) -> None: + """Run Codex's shared parser, with its no-transcript Stop fallback.""" session_id = body.get("session_id") transcript = body.get("transcript_path") or _find_codex_transcript(session_id) hook = body.get("hook_event_name") or body.get("hook") or "" - - scanned: dict = {"thoughts": [], "responses": [], "attachments": [], "web_searches": []} - if transcript and session_id: - scanned = _scan_codex_transcript(transcript, session_id) - _emit_synthetic_messages(source, body, scanned) - - # Fallback only when no transcript could be read at all. When a transcript - # was scanned it is authoritative: the final answer is already emitted (here - # or on an earlier hook), so adding last_assistant_message would duplicate it - # — the per-scan ``responses`` check can't see the earlier emit. - if hook in ("Stop", "SubagentStop") and session_id and not transcript: - msg = body.get("last_assistant_message") - if msg: - _post( - f"{COT_ENDPOINT}/v1/ingest/{source}", - { - "session_id": session_id, - "cwd": body.get("cwd"), - "hook_event_name": "afterAgentResponse", - "response": msg, - "_synthetic_category": "response", - "timestamp": body.get("timestamp"), - "model": body.get("model"), - }, - ) + readable = bool(transcript and Path(str(transcript)).is_file()) + if readable and session_id: + events = _scan_live_ingest_events( + source, str(transcript), session_id, body.get("timestamp") + ) + events = _fold_usage_metadata_into_hook_body(events, body) + _post_ingest_events(source, _enrich_live_events(events, body)) + if hook not in ("Stop", "SubagentStop") or not session_id or readable: + return + msg = str(body.get("last_assistant_message") or "").strip() + if not msg: + return + _post_ingest_events(source, [ + _shell_response_event(source, body, msg, label="fallback") + ]) def _hook_response(source: str, body: dict) -> dict | None: @@ -2202,12 +2079,12 @@ def _sync_subagent_links(agents: list[str], *, verbose: bool = False) -> int: def _import_cursor_subagent_responses(path: Path, child_id: str) -> None: - """Post a subagent's prompt/thought/response events from its transcript, - dropping tool_use (the child's live tool hooks already captured those). + """Post a subagent's prompt/thought/response Ingest Events from its transcript, + dropping tool_call Events (the child's live tool hooks already captured those). - Idempotent: the parser stamps each line with a stable ``_dedup_key`` and - ``_import``, so the collector dedups on (session, key) with no time window — - re-running on every stop hook is safe.""" + Transcript-position keyed conversation Ingest Events are idempotent across + stop hooks. Imported attachment Ingest Events intentionally retain their + historical unkeyed, time-windowed dedup behavior.""" try: mtime = path.stat().st_mtime lines = path.read_text(errors="replace").splitlines() @@ -2215,6 +2092,7 @@ def _import_cursor_subagent_responses(path: Path, child_id: str) -> None: return state: dict = {} path_str = str(path) + parsed_events: list[dict] = [] for lineno, raw in enumerate(lines): raw = raw.strip() if not raw: @@ -2223,14 +2101,26 @@ def _import_cursor_subagent_responses(path: Path, child_id: str) -> None: obj = json.loads(raw) except json.JSONDecodeError: continue - for ev in _cursor_line_to_events( - obj, child_id, lineno=lineno, mtime=mtime, state=state, path=path_str - ): - if ev.get("tool_name") is not None or ev.get("hook_event_name") == "PostToolUse": - continue - if ev.get("_dedup_key") is None: - ev.pop("_dedup_key", None) - _post(f"{COT_ENDPOINT}/v1/ingest/cursor", ev) + parsed_events.extend( + _cursor_line_to_ingest_events( + obj, + child_id, + lineno=lineno, + mtime=mtime, + state=state, + path=path_str, + origin="import", + ) + ) + parsed_events = [ + event for event in parsed_events if event.get("kind") != "tool_call" + ] + events = _prepare_ingest_events("cursor", parsed_events, origin="import") + for event in events: + payload = _ingest_event_to_hook_payload(event) + if payload.get("_dedup_key") is None: + payload.pop("_dedup_key", None) + _post(f"{COT_ENDPOINT}/v1/ingest/cursor", payload) def _reconcile_cursor_subagents(body: dict) -> None: @@ -2277,6 +2167,8 @@ def _claude_line_to_ingest_events( mtime: float | None = None, state: dict | None = None, path: str = "", + origin: str = "import", + byte_offset: int | None = None, ) -> list[dict]: """Map a single Claude transcript JSONL line to canonical ingest events. @@ -2297,14 +2189,29 @@ def _claude_line_to_ingest_events( cwd = obj.get("cwd") model = message.get("model") or obj.get("model") usage = message.get("usage") if isinstance(message.get("usage"), dict) else None + fallback_anchor = f"{path}:{byte_offset if byte_offset is not None else lineno}" + + def _dedup(suffix: str, *, live_only: bool = False) -> str | None: + return _scoped_dedup_key( + origin, + "claude", + suffix, + import_anchor=str(uuid) if uuid else None, + live_anchor=str(uuid or fallback_anchor), + live_only=live_only, + ) + base = { "session_id": sid, "cwd": cwd, - "origin": "import", + "origin": origin, "timestamp": ts, } if role == "user": + text = _text_from_blocks(content, clean_user=True) if isinstance(content, list) else _clean_user_text(str(content or "")) + if uuid and text: + state.setdefault("_user_prompts", {})[str(uuid)] = text # Tool results ride in user messages; pair them to the earlier call. if isinstance(content, list): for i, block in enumerate(content): @@ -2329,15 +2236,20 @@ def _claude_line_to_ingest_events( "tool_input": matched_input, "tool_response": result_text, "model": model, - "dedup_key": f"{uuid}:post:{tool_use_id or i}" if uuid else None, + "dedup_key": _dedup(f"post:{tool_use_id or i}"), }) - text = _text_from_blocks(content, clean_user=True) if isinstance(content, list) else _clean_user_text(str(content or "")) - if text and not _is_interruption(text): + if _is_interruption(text): + events.append({ + **base, + "kind": "interruption", + "dedup_key": _dedup("interruption"), + }) + elif text: events.append({ **base, "kind": "prompt", "text": text, - "dedup_key": f"{uuid}:prompt" if uuid else None, + "dedup_key": _dedup("prompt"), }) if isinstance(content, list): atts = _message_attachments(content) @@ -2347,6 +2259,7 @@ def _claude_line_to_ingest_events( "kind": "attachment", "text": text, "attachments": atts, + "dedup_key": _dedup("attachment", live_only=True), }) elif role == "assistant": if isinstance(content, list): @@ -2368,7 +2281,7 @@ def _claude_line_to_ingest_events( "text": str(think), "model": model, "usage": block_usage, - "dedup_key": f"{uuid}:think:{i}" if uuid else None, + "dedup_key": _dedup(f"think:{i}"), }) elif btype == "text": text = block.get("text") @@ -2379,7 +2292,7 @@ def _claude_line_to_ingest_events( "text": str(text), "model": model, "usage": block_usage, - "dedup_key": f"{uuid}:resp:{i}" if uuid else None, + "dedup_key": _dedup(f"resp:{i}"), }) elif btype == "tool_use": tool_name = block.get("name") or "" @@ -2396,7 +2309,7 @@ def _claude_line_to_ingest_events( "tool_input": tool_input, "model": model, "usage": block_usage, - "dedup_key": f"{uuid}:pre:{tool_id or i}" if uuid else None, + "dedup_key": _dedup(f"pre:{tool_id or i}"), }) else: text = str(content or "").strip() @@ -2407,8 +2320,21 @@ def _claude_line_to_ingest_events( "text": text, "model": model, "usage": usage, - "dedup_key": f"{uuid}:resp" if uuid else None, + "dedup_key": _dedup("resp"), }) + elif obj.get("type") == "attachment": + attachment = obj.get("attachment") if isinstance(obj.get("attachment"), dict) else {} + meta = _file_attachment_meta(attachment) + parent_uuid = obj.get("parentUuid") + parent_text = state.get("_user_prompts", {}).get(str(parent_uuid)) if parent_uuid else None + if meta and _file_attachment_matches_text(attachment, parent_text): + events.append({ + **base, + "kind": "attachment", + "text": parent_text, + "attachments": [meta], + "dedup_key": _dedup("attachment", live_only=True), + }) return events @@ -2423,10 +2349,14 @@ def _claude_line_to_events( path: str = "", ) -> list[dict]: """Map a Claude transcript line to current collector hook payloads.""" - return _ingest_events_to_hook_payloads( - _claude_line_to_ingest_events( - obj, session_id, lineno=lineno, mtime=mtime, state=state, path=path - ) + return _import_line_to_hook_payloads( + "claude", + obj, + session_id, + lineno=lineno, + mtime=mtime, + state=state, + path=path, ) @@ -2471,6 +2401,8 @@ def _ingest_event_to_hook_payload(event: dict) -> dict: out["interrupted"] = True if event.get("codex"): out["codex"] = event.get("codex") + if event.get("composer_mode"): + out["composer_mode"] = event.get("composer_mode") kind = event.get("kind") if kind == "prompt": @@ -2484,6 +2416,14 @@ def _ingest_event_to_hook_payload(event: dict) -> dict: out["hook_event_name"] = "afterAgentThought" out["_synthetic_category"] = "thought" out["thought"] = event.get("text") or "" + elif kind == "plan": + out["hook_event_name"] = "createPlan" + out["_synthetic_category"] = "plan" + out["plan_name"] = event.get("name") or "Plan" + out["plan_overview"] = event.get("overview") or "" + out["plan_body"] = event.get("body") or "" + out["plan_todos"] = event.get("todos") or [] + out["composer_mode"] = "plan" elif kind == "tool_call": phase = event.get("phase") or "instant" if phase == "start": @@ -2525,6 +2465,8 @@ def _cursor_line_to_ingest_events( mtime: float | None = None, state: dict | None = None, path: str = "", + origin: str = "import", + byte_offset: int | None = None, ) -> list[dict]: """Map a Cursor transcript line to canonical ingest events. @@ -2541,21 +2483,48 @@ def _cursor_line_to_ingest_events( if content is None: content = message.get("content") + def _dedup(suffix: str, *, live_only: bool = False) -> str | None: + return _scoped_dedup_key( + origin, + "cursor", + suffix, + import_anchor=f"{path}:{lineno}", + live_anchor=str(byte_offset if byte_offset is not None else lineno), + live_only=live_only, + ) + def _base(block_idx: int = 0) -> dict: return { - "origin": "import", + "origin": origin, "session_id": session_id, "timestamp": _cursor_import_ts(mtime, lineno, block_idx), } + if ( + obj.get("type") == "turn_ended" + and obj.get("status") in ("error", "aborted") + and _is_interruption(str(obj.get("error") or "")) + ): + return [{ + **_base(), + "kind": "interruption", + "dedup_key": _dedup("interruption"), + }] + if role == "user": text = _text_from_blocks(content, clean_user=True) if isinstance(content, list) else _clean_user_text(str(content or "")) - if text and not _is_interruption(text): + if _is_interruption(text): + events.append({ + **_base(), + "kind": "interruption", + "dedup_key": _dedup("interruption"), + }) + elif text: events.append({ **_base(), "kind": "prompt", "text": text, - "dedup_key": f"{path}:{lineno}:prompt", + "dedup_key": _dedup("prompt"), }) if isinstance(content, list): atts = _message_attachments(content) @@ -2565,6 +2534,7 @@ def _cursor_line_to_ingest_events( "kind": "attachment", "text": text, "attachments": atts, + "dedup_key": _dedup("attachment", live_only=True), }) return events @@ -2580,7 +2550,7 @@ def _cursor_line_to_ingest_events( **_base(i), "kind": "response", "text": str(text), - "dedup_key": f"{path}:{lineno}:resp:{i}", + "dedup_key": _dedup(f"resp:{i}"), }) elif btype == "thinking": think = block.get("thinking") @@ -2589,7 +2559,7 @@ def _cursor_line_to_ingest_events( **_base(i), "kind": "thought", "text": str(think), - "dedup_key": f"{path}:{lineno}:think:{i}", + "dedup_key": _dedup(f"think:{i}"), }) elif btype == "tool_use": name = block.get("name") or "" @@ -2597,15 +2567,28 @@ def _cursor_line_to_ingest_events( # ApplyPatch sends a raw patch string; keep dict or str as-is so # normalize can pull the target path out of either shape. tool_input = inp if isinstance(inp, (dict, str)) else {} - events.append({ - **_base(i), - "kind": "tool_call", - "phase": "instant", - "status": "ok", - "tool_name": name, - "tool_input": tool_input, - "dedup_key": f"{path}:{lineno}:tool:{i}", - }) + if name == "CreatePlan": + plan_input = tool_input if isinstance(tool_input, dict) else {} + events.append({ + **_base(i), + "kind": "plan", + "name": plan_input.get("name") or "Plan", + "overview": plan_input.get("overview") or "", + "body": plan_input.get("plan") or "", + "todos": plan_input.get("todos") + if isinstance(plan_input.get("todos"), list) else [], + "dedup_key": _dedup(f"plan:{i}"), + }) + else: + events.append({ + **_base(i), + "kind": "tool_call", + "phase": "instant", + "status": "ok", + "tool_name": name, + "tool_input": tool_input, + "dedup_key": _dedup(f"tool:{i}"), + }) return events return events @@ -2621,10 +2604,14 @@ def _cursor_line_to_events( path: str = "", ) -> list[dict]: """Map a Cursor transcript line to current collector hook payloads.""" - return _ingest_events_to_hook_payloads( - _cursor_line_to_ingest_events( - obj, session_id, lineno=lineno, mtime=mtime, state=state, path=path - ) + return _import_line_to_hook_payloads( + "cursor", + obj, + session_id, + lineno=lineno, + mtime=mtime, + state=state, + path=path, ) @@ -2722,17 +2709,19 @@ def _codex_session_config(meta: dict | None, turn: dict | None) -> dict: return cfg -def _codex_session_start_event(session_id: str, ts, state: dict) -> dict: +def _codex_session_start_event( + session_id: str, ts, state: dict, *, origin: str = "import" +) -> dict: """A SessionStart lifecycle event for an imported Codex session, carrying the sandbox/approval posture that Codex only records on its metadata lines.""" ev: dict = { "session_id": session_id, - "origin": "import", + "origin": origin, "timestamp": ts, "kind": "session_metadata", "metadata_type": "session_start", "cwd": state.get("cwd"), - "dedup_key": "session:start", + "dedup_key": _origin_dedup_key(origin, "codex", "session:start"), } cfg = _codex_session_config(state.get("_codex_meta"), state.get("_codex_turn")) if cfg.get("model"): @@ -2750,6 +2739,8 @@ def _codex_line_to_ingest_events( mtime: float | None = None, state: dict | None = None, path: str = "", + origin: str = "import", + byte_offset: int | None = None, ) -> list[dict]: """Map a Codex rollout JSONL line to canonical ingest events. @@ -2764,6 +2755,22 @@ def _codex_line_to_ingest_events( kind = obj.get("type") payload = obj.get("payload") if isinstance(obj.get("payload"), dict) else {} line_id = obj.get("id") or "" + fallback_anchor = str(byte_offset if byte_offset is not None else lineno) + + def _dedup( + suffix: str, + *, + anchor_override: str | None = None, + live_only: bool = False, + ) -> str | None: + return _scoped_dedup_key( + origin, + "codex", + suffix, + import_anchor=anchor_override or (str(line_id) if line_id else None), + live_anchor=anchor_override or str(line_id or fallback_anchor), + live_only=live_only, + ) # Codex never repeats cwd/model on message lines; it records them once on # session_meta and per-turn on turn_context. Capture them in per-file state @@ -2784,7 +2791,7 @@ def _codex_line_to_ingest_events( # emit the SessionStart here, once, with everything gathered so far. if not state.get("_session_started") and isinstance(state.get("_codex_meta"), dict): state["_session_started"] = True - return [_codex_session_start_event(session_id, ts, state)] + return [_codex_session_start_event(session_id, ts, state, origin=origin)] return events # Per-turn token usage and interruptions arrive on event_msg lines *after* # the assistant message, so fold them onto the deferred pending message. @@ -2796,14 +2803,21 @@ def _codex_line_to_ingest_events( last = info.get("last_token_usage") if isinstance(info.get("last_token_usage"), dict) else None if last: pending["usage"] = _merge_usage(pending.get("usage"), last) - elif ptype == "turn_aborted" and isinstance(pending, dict): - pending["interrupted"] = True + elif ptype == "turn_aborted": + events.extend(_codex_flush_pending_ingest_events(state)) + events.append({ + "session_id": session_id, + "origin": origin, + "timestamp": ts, + "kind": "interruption", + "dedup_key": _dedup("interruption"), + }) return events if kind != "response_item": return events - base: dict = {"session_id": session_id, "origin": "import", "timestamp": ts} + base: dict = {"session_id": session_id, "origin": origin, "timestamp": ts} if state.get("cwd"): base["cwd"] = state["cwd"] @@ -2820,7 +2834,7 @@ def _codex_line_to_ingest_events( "kind": "response", "text": text_val, "model": model, - "dedup_key": f"{line_id}:resp" if line_id else None, + "dedup_key": _dedup("resp"), } else: ev = { @@ -2828,7 +2842,7 @@ def _codex_line_to_ingest_events( "kind": "thought", "text": text_val, "model": model, - "dedup_key": f"{line_id}:think" if line_id else None, + "dedup_key": _dedup("think"), } # Defer so a trailing token_count / turn_aborted line can fold usage and # interrupted status onto this message before it is posted. The prior @@ -2836,7 +2850,10 @@ def _codex_line_to_ingest_events( flushed = _codex_flush_pending_ingest_events(state) state["_pending_msg"] = ev return flushed - elif pt == "message" and payload.get("role") == "user": + else: + events.extend(_codex_flush_pending_ingest_events(state)) + + if pt == "message" and payload.get("role") == "user": content = payload.get("content") raw_text = _text_from_blocks(content, clean_user=False) if isinstance(content, list) else str(content or "") env_block, _ = _extract_env_context(raw_text) @@ -2844,16 +2861,18 @@ def _codex_line_to_ingest_events( events.append(_env_context_event(base, env_block)) text_val = _clean_user_text(raw_text) if _is_interruption(text_val): - pending = state.get("_pending_msg") - if isinstance(pending, dict): - pending["interrupted"] = True + events.append({ + **base, + "kind": "interruption", + "dedup_key": _dedup("interruption"), + }) return events if text_val: events.append({ **base, "kind": "prompt", "text": text_val, - "dedup_key": f"{line_id}:prompt" if line_id else None, + "dedup_key": _dedup("prompt"), }) if isinstance(content, list): atts = _message_attachments(content) @@ -2863,6 +2882,7 @@ def _codex_line_to_ingest_events( "kind": "attachment", "text": text_val, "attachments": atts, + "dedup_key": _dedup("attachment", live_only=True), }) elif pt == "reasoning": text_val = _codex_reasoning_text(payload) @@ -2871,7 +2891,7 @@ def _codex_line_to_ingest_events( **base, "kind": "thought", "text": text_val, - "dedup_key": f"{line_id}:reason" if line_id else None, + "dedup_key": _dedup("reason"), }) elif pt == "web_search_call": action = payload.get("action") if isinstance(payload.get("action"), dict) else {} @@ -2885,7 +2905,7 @@ def _codex_line_to_ingest_events( "tool_name": "WebSearch", "tool_input": {"query": str(query)}, "tool_response": {"status": payload.get("status"), "action": action}, - "dedup_key": f"{line_id}:web" if line_id else None, + "dedup_key": _dedup("web"), }) elif pt in ("function_call", "custom_tool_call"): name = payload.get("name") or "" @@ -2894,7 +2914,7 @@ def _codex_line_to_ingest_events( tool_name, tool_input = _codex_map_tool(name, raw_args) if call_id: state[call_id] = (tool_name, tool_input) - dk = f"{call_id}:pre" if call_id else (f"{line_id}:pre" if line_id else None) + dk = _dedup("pre", anchor_override=str(call_id) if call_id else None) events.append({ **base, "kind": "tool_call", @@ -2911,7 +2931,7 @@ def _codex_line_to_ingest_events( tool_name, tool_input = state.get(call_id, ("", {})) status = payload.get("status") is_error = status not in (None, "completed", "success") - dk = f"{call_id}:post" if call_id else (f"{line_id}:post" if line_id else None) + dk = _dedup("post", anchor_override=str(call_id) if call_id else None) events.append({ **base, "kind": "tool_call", @@ -2936,10 +2956,14 @@ def _codex_line_to_events( path: str = "", ) -> list[dict]: """Map a Codex rollout JSONL line to current collector hook payloads.""" - return _ingest_events_to_hook_payloads( - _codex_line_to_ingest_events( - obj, session_id, lineno=lineno, mtime=mtime, state=state, path=path - ) + return _import_line_to_hook_payloads( + "codex", + obj, + session_id, + lineno=lineno, + mtime=mtime, + state=state, + path=path, ) @@ -2995,11 +3019,7 @@ def _import_agent(agent: str, since: str | None, offsets: dict[str, int], print(f"cot import: no {agent} transcripts found", file=sys.stderr) return 0 - line_parser = { - "claude": _claude_line_to_events, - "cursor": _cursor_line_to_events, - "codex": _codex_line_to_events, - }[agent] + line_parser = _ingest_line_parser(agent) source = agent posted = 0 @@ -3057,34 +3077,34 @@ def _import_agent(agent: str, since: str | None, offsets: dict[str, int], # tool call with its result; reset per file so ids never bleed across). state: dict = {} path_str = str(path) + parsed_events: list[dict] = [] for lineno, line in enumerate(lines): try: obj = json.loads(line) except json.JSONDecodeError: continue - events = line_parser( - obj, sid, lineno=lineno, mtime=mtime, state=state, path=path_str + parsed_events.extend( + line_parser( + obj, + sid, + lineno=lineno, + mtime=mtime, + state=state, + path=path_str, + origin="import", + ) ) - for ev in events: - # Subagent transcripts: live hooks already hold the tool calls, - # so import only the conversation text the hooks never deliver. - if responses_only and ( - ev.get("tool_name") is not None - or ev.get("hook_event_name") == "PostToolUse" - ): - continue - # Strip None dedup keys - if ev.get("_dedup_key") is None: - ev.pop("_dedup_key", None) - _post(f"{COT_ENDPOINT}/v1/ingest/{source}", ev) - posted += 1 - - # Release any message held back to absorb its trailing token_count / - # turn_aborted (Codex only; a no-op for the other parsers). - for ev in _codex_flush_pending(state): - if ev.get("_dedup_key") is None: - ev.pop("_dedup_key", None) - _post(f"{COT_ENDPOINT}/v1/ingest/{source}", ev) + parsed_events.extend(_codex_flush_pending_ingest_events(state)) + events = _prepare_ingest_events(agent, parsed_events, origin="import") + for event in 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 + payload = _ingest_event_to_hook_payload(event) + if payload.get("_dedup_key") is None: + payload.pop("_dedup_key", None) + _post(f"{COT_ENDPOINT}/v1/ingest/{source}", payload) posted += 1 offsets[str(path)] = new_offset @@ -3452,32 +3472,39 @@ def main() -> int: if isinstance(prompt, str): env_block, remainder = _extract_env_context(prompt) if env_block: + event = _env_context_event( + { + "session_id": body.get("session_id"), + "cwd": body.get("cwd"), + "origin": "hook", + "timestamp": body.get("timestamp"), + }, + env_block, + ) + # This is hook-derived rather than scanner-recovered, so keep + # the Collector's ordinary five-second payload-hash dedup. + event.pop("dedup_key", None) _post( f"{COT_ENDPOINT}/v1/ingest/{source}", - _env_context_event( - { - "session_id": body.get("session_id"), - "cwd": body.get("cwd"), - "timestamp": body.get("timestamp"), - }, - env_block, - ), + _ingest_event_to_hook_payload(event), ) body["prompt"] = remainder suppress = not remainder redundant = source == "cursor" and hook in CURSOR_REDUNDANT_HOOKS - # Cursor transcript-only artifacts are posted by _emit_cursor_artifacts. For - # plan-mode turns it also posts the filled response, so skip the default - # empty response in that case. - handled = source == "cursor" and _emit_cursor_artifacts(body, source) + handled = False + if source == "cursor": + # Cursor's shell derives whether its canonical response/plan Events + # replace the raw afterAgentResponse hook body. + handled = _emit_cursor_events(body, source) + elif source == "claude": + _emit_claude_events(body, source) + elif source == "codex": + _emit_codex_events(body, source) if not redundant and not handled and not suppress: _post(f"{COT_ENDPOINT}/v1/ingest/{source}", body) if source == "claude": - _emit_claude_responses(body, source) if hook in ("SessionStart", "sessionStart"): _log_launch(body) - elif source == "codex": - _emit_codex_responses(body, source) elif source == "cursor" and hook in ("subagentStop", "stop"): # A subagent just finished (or the turn ended): fold its transcript-only # responses in now, so active sessions don't wait for cot watch/import. diff --git a/docs/adr/0002-live-scanners-emit-canonical-ingest-events.md b/docs/adr/0002-live-scanners-emit-canonical-ingest-events.md new file mode 100644 index 0000000..7e15989 --- /dev/null +++ b/docs/adr/0002-live-scanners-emit-canonical-ingest-events.md @@ -0,0 +1,17 @@ +# Live scanners emit canonical Ingest Events; trigger-context behaviour stays in live shells + +Status: accepted + +The Bridge captures transcripts on two paths — live (hook-triggered delta scans) and import — and they used to encode "transcript → Collector payload" independently, which is where both shipped scanner bugs lived. We decided both Origins share one line-parser per agent, emitting canonical Ingest Events, adapted to the wire format in exactly one place. The live path's genuine differences are handled by design, not by a second parser: generic post-steps on the event list (kind filtering with usage folding, timestamp rewrite, interruption flagging, consecutive-response collapse), plus a thin per-agent live shell for what only the trigger context knows (when to scan, pairing recovered question answers, the suppress-default-post signal, safety-net final text). + +## Considered Options + +- **Separate live scanners that merely return canonical events** — rejected: keeps two parsers per agent, preserving the exact duplication that produced the dropped-response and duplicated-response bugs. +- **Events + sidecar return shape for live extras** — rejected: every sidecar fact (plan-turn flag, questions) is derivable from the event list plus the hook body the shell already holds; a sidecar is a second copy with one authority available. +- **Question turns as synthetic pre/post pairs (old live behaviour)** — rejected in favour of the import parser's single instant tool_call; Cursor transcripts carry no tool results, so the pair was two rows rendering as one item. + +## Consequences + +- Live shells must not grow parsing or payload knowledge; if a shell needs a new fact from the transcript, the shared parser emits it as an Ingest Event (or marker kind, like `interruption`) instead. +- Live scanner Events carry origin-namespaced explicit dedup keys, making re-scans idempotent without colliding with import keys or disturbing hook-data-wins reconciliation. +- Agreed fallback if Cursor's replay diff proves noisy: Cursor keeps a bespoke scanner that still emits canonical Ingest Events — parser sharing is sacrificed, the vocabulary and single adapter are not. diff --git a/docs/specs/live-scanners-canonical-ingest-events.md b/docs/specs/live-scanners-canonical-ingest-events.md new file mode 100644 index 0000000..9d47b06 --- /dev/null +++ b/docs/specs/live-scanners-canonical-ingest-events.md @@ -0,0 +1,79 @@ +# Route the Bridge's live scanners through the canonical Ingest Event seam + +Status: implemented; replay validation pending + +## Problem Statement + +The Bridge captures agent activity twice over: live, when hooks fire during a Session, and by import, when transcripts are recovered from disk. The two paths encode "how a transcript becomes Events" independently. As a user, the same transcript produces materially different Sessions depending on Origin: imported Cursor plan-mode Sessions render blank where live ones show a structured plan; live Cursor Sessions are missing the agent's thoughts that import recovers; repeated commentary blocks are collapsed live but spam imported timelines; and token usage is accounted differently per path. As a maintainer, every transcript-format change and every payload bug must be found and fixed twice — two shipped bugs (dropped assistant responses, duplicated Cursor responses) lived exactly in this duplication, and the live path has no tests because scanning, payload mapping, and posting are fused into single functions. + +## Solution + +The import path already owns a deep seam: per-agent parsers emit canonical Ingest Events (kind-keyed records — prompt, response, thought, tool_call, attachment, session_metadata), and a single adapter maps them onto the Collector's wire format. This change routes the live path through that same seam. Live scanners return canonical Ingest Events from the same shared per-agent line-parsers that import uses; a thin per-agent live shell handles only what is genuinely trigger-context behaviour (when to scan, pairing recovered question answers, the suppress-default-post protocol). The canonical vocabulary grows to absorb Cursor's specials: a `plan` kind and an `interruption` marker. One parser per agent, one adapter, both Origins — a payload fix lands once and covers live and import alike, and the live path becomes testable exactly the way the import path already is. + +## User Stories + +1. As a developer observing my agent sessions, I want a Cursor Session to look the same whether it was captured live or imported from a transcript, so that the dashboard tells one truthful story per Session. +2. As a developer observing my agent sessions, I want imported Cursor plan-mode Sessions to show the structured plan (name, overview, body, todos), so that plan turns no longer render blank in imported history. +3. As a developer observing my agent sessions, I want live Cursor Sessions to include the agent's thoughts, so that live capture is not poorer than import for the same transcript. +4. As a developer observing my agent sessions, I want repeated identical commentary blocks collapsed in imported Sessions the same way they are live, so that imported Cursor timelines are not padded with dozens of duplicate Events. +5. As a developer observing my agent sessions, I want token usage totals preserved when live scanning drops Events that hooks already delivered, so that cost tracking does not silently undercount. +6. As a developer observing my agent sessions, I want a user-interruption to flag the Event it cut off in both Origins' data, so that interrupted turns are identifiable in the timeline. +7. As a developer whose Bridge crashes or loses its offset state, I want re-scanned transcripts to be recognized as already ingested, so that recovery never duplicates my Session history. +8. As a developer whose agent double-fires a hook, I want scanner-recovered Events deduplicated by stable keys rather than a five-second window, so that slow retries cannot create duplicate Events. +9. As a developer with an imported Session that later goes live, I want live Events reconciled by the hook-data-wins policy rather than swallowed as duplicates of imported rows, so that resuming an imported Session works correctly. +10. As a developer using Cursor's AskQuestion flow, I want the question and its recovered answer captured as one coherent Event pairing in both Origins, so that clarification exchanges are visible in every Session. +11. As a maintainer, I want the transcript shape of each agent known to exactly one parser, so that a format change is a one-module edit regardless of Origin. +12. As a maintainer, I want the Collector wire format known only to the single adapter, so that payload bugs concentrate in one place and a fix covers both Origins. +13. As a maintainer, I want the live shells reduced to trigger-context orchestration, so that reading a shell tells me when scanning happens, not how payloads are built. +14. As a maintainer, I want the live path covered by tests at the same seams as the import path, so that the previously untested half of the Bridge stops being where bugs ship. +15. As a maintainer, I want golden-master recordings of today's live behaviour frozen before the refactor, so that unwritten behaviours (plan-only-turn fallback, safety-net final text) cannot be silently lost. +16. As a maintainer, I want a replay diff of the new path against my real transcripts and the raw ingest ledger before merge, so that every behaviour change is either on the expected-differences list or treated as a bug. +17. As a maintainer, I want Claude and Codex unified and gated before Cursor, so that the riskiest agent cannot stall the proven wins. +18. As a maintainer, I want a documented fallback for Cursor (bespoke scanner still emitting canonical Ingest Events), so that a noisy replay diff has a pre-agreed exit that preserves the vocabulary win. +19. As a future agent-source implementer, I want adding a new agent to mean writing one line-parser that emits canonical Ingest Events, so that both live and import capture come for free. +20. As an AI coding agent working on this repo, I want the Ingest Event vocabulary defined in the domain glossary and the seam recorded in an ADR, so that I can navigate to the right module without reverse-engineering two pipelines. +21. As a dashboard user, I want Cursor question turns to appear as a single instant tool Event rather than a synthetic start/end pair, so that question turns read as one item without a fake duration. +22. As a maintainer of the drift ledger, I want live and import to project through the same mapping, so that drift metrics measure agent behaviour rather than pipeline divergence. + +## Implementation Decisions + +- **Scope: all three agents** (Claude, Codex, Cursor). The unification is void without Cursor; its differences are handled by design, not exclusion. +- **The live scanner interface is a list of canonical Ingest Events** — the same type the import parsers return. No sidecar return values: everything the live path needs beyond the events is derivable from the event list plus the hook body the shell already holds. +- **One shared line-parser per agent**, used by both Origins. Live shells become: read transcript delta → feed lines through the shared parser → apply generic post-steps → adapt → post. +- **Four generic post-steps**, each a pure transformation on the event list, written once and shared across agents: + 1. Per-agent kind filter — drop Events the agent's hook surface already delivers live — with usage folding, so tokens on dropped Events land on the next kept Event and totals are preserved. + 2. Timestamp rewrite — Cursor live Events take the hook timestamp plus ordering offsets instead of the import path's mtime synthesis. + 3. Interruption flagging — the parser emits a canonical `interruption` marker Event; the shared post-step flags the preceding Event in both Origins and then removes the marker before adaptation. + 4. Consecutive-identical response collapse — applied to both Origins, so imported Cursor Sessions get the same cleanup live ones do. +- **Vocabulary additions:** a `plan` kind carrying name, overview, body, and todos, mapped by the adapter onto the plan wire shape the Collector's normalize layer already accepts; questions become a single instant tool_call (adopting the import parser's documented choice), retiring the live path's synthetic pre/post pair. +- **Trigger-context behaviour stays in thin per-agent live shells:** which hooks trigger a scan; pairing the hook body's text to recovered question Events and posting answers to the dedicated answer endpoint; the suppress-default-post signal (derived from whether the scan yielded response or plan Events); the safety-net append of the hook's final text; Codex's fallback to the last assistant message when no transcript is readable. +- **Dedup: live scanner Events carry explicit, origin-namespaced dedup keys** (Claude: uuid-derived; Cursor: absolute byte offset plus block index). Namespacing guarantees no collision with import keys, so hook-data-wins reconciliation is untouched, while re-scans and hook retries become idempotent under the Collector's existing explicit-key dedup regime. Non-scanner hook payloads keep the existing five-second window behaviour. +- **Zero Collector changes.** The adapter still emits hook-style payloads; the plan wire shape and the explicit-dedup semantics already exist server-side. +- **Sequencing: two replay-gated changes.** Change one: seam infrastructure, post-steps, and the Claude and Codex live shells. Change two: Cursor (plan kind, instant questions, shared collapse, answer pairing in the shell). If Cursor's replay diff is noisy, change two falls back to a bespoke Cursor scanner that still emits canonical Ingest Events — vocabulary and adapter unification survive even in the fallback. +- **Side effects:** add "Ingest Event" to the domain glossary; record an ADR ("live scanners emit canonical Ingest Events; trigger-context behaviour stays in live shells") so future architecture passes don't re-litigate the shells' existence or re-specialize the scanners. + +## Testing Decisions + +- **Good tests here exercise external behaviour at a seam:** transcript content in, canonical Ingest Events or posted payloads out. No test reaches past an interface to assert on parser internals, offset state, or intermediate shapes. +- **Two test seams, both already existing in the code:** + 1. **The canonical Ingest Event seam** — transcript lines in, canonical Events out. Covers the shared parsers (plan kind, instant questions, interruption marker, dedup key minting) and the pure post-steps (list in, list out). Prior art: the import path's canonical contract tests, which already test the parsers this way. + 2. **The posting seam** — the Bridge's single exit. Tests intercept the post function and assert ordered (endpoint, payload) sequences from a hook body plus a transcript fixture. Covers live-shell orchestration: suppress-default protocol, answer pairing, safety-net text, Codex's no-transcript fallback. +- **Golden masters recorded before any refactor:** today's live emit behaviour captured at the posting seam over fixtures covering Cursor's four hard turns — a plan turn, a question turn with and without a recoverable answer, the repeated-commentary pathology, and a plan-only turn with no text. After the refactor the same fixtures must reproduce the recordings modulo the expected-differences list; the goldens then graduate into the live path's permanent suite. +- **Merge gate: replay diff against real data** — the developer's actual transcripts and the raw ingest ledger's recorded payloads as the oracle. Every diff must appear on the expected-differences list or be treated as a regression. +- **Expected-differences list (seed):** Cursor live gains thought Events; Codex live gains standalone reasoning-summary thought Events already recovered by import; question pre/post pairs become single instant Events; live Events gain dedup keys; Claude usage placement shifts from pending-fold to per-block with folding on filtered Events (totals preserved); imported Cursor Sessions gain plan Events and lose consecutive duplicate responses; imported Claude and Cursor output immediately followed by an interruption marker gains `interrupted: true` (Codex import already carried this signal); imported Claude Sessions recover standalone file-attachment records that match their parent prompt. +- **Final confidence pass:** one live smoke Session per agent against a dev Collector, eyeballing the timeline beside an equivalent old-path Session. + +## Out of Scope + +- Any Collector-side changes: wire format, normalize layer, dedup regimes, the hook-data-wins policy, and the Store seam are all untouched. +- Removing the Collector-side Cursor transcript scan (the question-recovery backfill that duplicates Bridge knowledge). It is a separate deepening candidate; this change may make it redundant, but verifying and deleting it is its own piece of work. +- Typing the Store's read queries (settled by the existing ADR) and any category-vocabulary registry work. +- The Bridge's hook wiring, install, backup, and docker lifecycle concepts — untested today, but not part of this seam. +- Changing what the dashboard renders beyond what falls out of unified Event streams. +- Migrating or rewriting historical Sessions already in the Store; only re-imports and new captures flow through the unified path. + +## Further Notes + +- The riskiest behaviour cluster is Cursor's live shell (plan-only fallback, answer pairing, suppress-default). The design isolates it precisely so the fallback can amputate Cursor's parser sharing without losing the vocabulary and adapter unification. +- The interruption marker and the plan kind slightly widen the canonical vocabulary; both are consumed Bridge-side (marker) or already understood Collector-side (plan), so the wire format's surface does not grow. +- Import automatically gains plan Events by virtue of the shared parser — flagged during design as closing a drift bug, not scope creep; the replay diff quantifies exactly what it changes in existing history. diff --git a/docs/validation/live-scanner-replay.md b/docs/validation/live-scanner-replay.md new file mode 100644 index 0000000..0f3a362 --- /dev/null +++ b/docs/validation/live-scanner-replay.md @@ -0,0 +1,67 @@ +# Live scanner canonical-event validation + +**Status: not merge-ready.** Cursor transcript replay is complete. A current +raw-ingest-ledger diff and disposable live smoke Sessions for all three agents +remain open merge gates. Cursor's documented fallback was not needed; the +shared-parser path matched the expected semantic differences. + +Validated on 2026-07-15 against base `8d035a0` and the current implementation +commit. The replay compared semantic hashes only; transcript contents were not +printed or copied. + +## Posting-seam recordings + +`backend/tests/test_bridge_live.py` freezes ordered endpoint/payload behavior +for Cursor plan turns, question turns with and without a recoverable answer, +consecutive repeated commentary, plan-only fallback text, and the hook-text +safety net. It also covers Claude and Codex live shells at the same posting +seam. + +These recordings were reconstructed from the `8d035a0` behavior and added in +the implementation commit, rather than committed before the refactor. They are +useful permanent regression tests, but they are not independent historical +evidence and do not satisfy the real-data replay gate by themselves. + +Hook-data-wins reconciliation and namespaced-live-key retry idempotency are +covered in the Collector suite by +`backend/tests/test_raw_ingest_drift.py::test_namespaced_live_event_supersedes_import_and_retry_is_idempotent`. + +## Local transcript replay + +The machine had 11 Claude transcripts and 20 readable Codex transcripts. The +old live scanners and new canonical path produced identical response, +attachment, and web-search multisets for every file. Claude Event counts also +matched exactly. Codex gained only standalone reasoning-summary thought Events, +which is listed in the spec's expected differences because import already +recovered them. + +Cursor replay used a purpose-built, non-sensitive fixture Session containing 30 +valid main-transcript records and a three-record subagent transcript. The old +scanner and shared-parser path agreed on one structured plan, one AskQuestion, +and all non-duplicate response semantics. The new path removed exactly two +consecutive-identical responses, matching the expected-differences list. Cursor +stored the attached file as textual mentions rather than a structured +attachment record, so both paths emitted zero attachment Events. + +The replay exposed Cursor's current interruption shape: a `turn_ended` record +with `status` `error` or `aborted` and a user-abort error string. The shared +parser now maps that shape to the canonical interruption marker. Two main-turn +interruptions and one subagent interruption were recovered; the subagent marker +followed a tool call and correctly flagged the preceding retained response after +tool filtering. The real shape is locked at the posting seam in +`backend/tests/test_bridge_live.py`. + +The local `~/.cot/cot.db` still predates `raw_ingest_events`, and no dev +Collector was running during the fixture Session, so no raw-ledger oracle was +available. That input remains required before merge. + +`CONTEXT.md` was supplied as an explicit implementation input and committed +intact. Its broader glossary/component material was therefore not authored as +part of this refactor; the relevant implementation side effect is the included +Ingest Event definition. + +## Smoke status + +No dev Collector was running for disposable live Sessions, so the three-agent +live smoke remains a pre-merge manual gate. The implementation does not mutate +the developer's installed hooks or production Collector as part of validation. From 71b79cc68d62702d7aac22e85ad94d288ca16f1f Mon Sep 17 00:00:00 2001 From: Yash Dave Date: Thu, 16 Jul 2026 10:07:40 +0530 Subject: [PATCH 2/3] Fix Claude Desktop lifecycle noise --- backend/app/db.py | 187 +++++++++++++++++++----- backend/app/normalize.py | 28 +++- backend/tests/test_raw_ingest_drift.py | 195 +++++++++++++++++++++++++ 3 files changed, 370 insertions(+), 40 deletions(-) diff --git a/backend/app/db.py b/backend/app/db.py index 34f6fbe..9ae82c8 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -17,12 +17,10 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from . import __version__, store, timeutil -from .normalize import APPROVAL_REVIEW_PREFIX, categorize, normalize +from .normalize import APPROVAL_REVIEW_PREFIX, categorize, lifecycle_boundary, normalize from .pricing import cost_for, normalize_model from .question_recovery import ANSWER_SOURCE_ASSISTANT_SUMMARY, recover_cursor_question_response -_SESSION_END_HOOKS = {"Stop", "stop", "SessionEnd", "sessionEnd"} -_SESSION_START_HOOKS = {"SessionStart", "sessionStart"} _APPROVAL_REVIEW_RE = re.compile(r"\bReviewed Codex session id:\s*([0-9a-fA-F-]{36})\b") @@ -401,6 +399,85 @@ def _recategorize_subagents(conn: sqlite3.Connection) -> None: ) +def _is_claude_desktop_suggestion( + source: str, hook: str, raw: dict[str, Any], origin: str +) -> bool: + """Identify Claude Desktop's internal follow-up suggestion hook.""" + return ( + source == "claude" + and origin == "hook" + and hook == "SubagentStop" + and not str(raw.get("agent_type") or "").strip() + ) + + +def _repair_claude_desktop_events(conn: sqlite3.Connection) -> None: + """Repair lifecycle and internal-suggestion rows from Claude Desktop.""" + suggestion_rows = conn.execute( + "SELECT id, source, hook, origin, raw_ingest_id, payload FROM events" + " WHERE source = 'claude' AND hook = 'SubagentStop'" + ).fetchall() + for row in suggestion_rows: + try: + raw = json.loads(row["payload"] or "{}") + except json.JSONDecodeError: + continue + if not _is_claude_desktop_suggestion( + row["source"], row["hook"], raw, row["origin"] + ): + continue + if row["raw_ingest_id"] is not None: + conn.execute( + "UPDATE raw_ingest_events SET status = 'ignored', event_id = NULL" + " WHERE id = ?", + (row["raw_ingest_id"],), + ) + else: + _append_raw_ingest( + conn, "claude", raw, origin="hook", status="ignored" + ) + conn.execute("DELETE FROM events WHERE id = ?", (row["id"],)) + + conn.execute( + "UPDATE events SET title = 'Turn ended'" + " WHERE source = 'claude' AND hook IN ('Stop', 'stop')" + ) + + lifecycle_rows = conn.execute( + "SELECT session_id, hook, ts FROM events" + " WHERE source = 'claude' AND origin = 'hook'" + " AND category = 'lifecycle'" + " ORDER BY ts ASC, id ASC" + ).fetchall() + bounds: dict[str, dict[str, tuple[datetime, str] | None]] = {} + for row in lifecycle_rows: + parsed = timeutil.parse_ts(row["ts"]) + if parsed is None: + continue + boundary = lifecycle_boundary("claude", row["hook"]) + if boundary not in ("session_start", "session_end"): + continue + state = bounds.setdefault(row["session_id"], {"start": None, "end": None}) + key = "start" if boundary == "session_start" else "end" + current = state[key] + if current is None or parsed >= current[0]: + state[key] = (parsed, row["ts"]) + + for session_id, state in bounds.items(): + start = state["start"] + end = state["end"] + if end is not None and (start is None or end[0] >= start[0]): + conn.execute( + "UPDATE sessions SET status = 'completed', ended_at = ? WHERE id = ?", + (end[1], session_id), + ) + else: + conn.execute( + "UPDATE sessions SET status = 'active', ended_at = NULL WHERE id = ?", + (session_id,), + ) + + def _recategorize_cursor_tools(conn: sqlite3.Connection) -> None: """Cursor's generic pre/postToolUse events used to fall through to ``other`` (only MCP was rescued). They now categorize like Claude/Codex tool calls — @@ -509,9 +586,23 @@ def _drop_redundant_cursor_hooks(conn: sqlite3.Connection) -> None: ) -def should_ignore_event(norm: dict[str, Any]) -> bool: - """Return True for hook rows intentionally excluded from storage.""" - return norm.get("source") == "cursor" and norm.get("hook") in _CURSOR_GRANULAR_HOOKS +def should_ignore_event( + norm: dict[str, Any], raw: dict[str, Any] | None = None +) -> bool: + """Return True for hook rows intentionally excluded from Event storage.""" + if norm.get("source") == "cursor" and norm.get("hook") in _CURSOR_GRANULAR_HOOKS: + return True + # Claude Desktop uses untyped, ephemeral subagents to generate suggested + # follow-up prompts after a turn. Real Claude Code subagents carry the + # documented agent_type field; keep the internal suggestions in the raw + # ingest ledger without presenting them as user-created subagent Events. + body = raw or {} + return _is_claude_desktop_suggestion( + str(norm.get("source") or ""), + str(norm.get("hook") or ""), + body, + "import" if body.get("_import") else "hook", + ) def _recategorize_web_search(conn: sqlite3.Connection) -> None: @@ -981,7 +1072,7 @@ def _question_response_obj(detail: Any) -> dict[str, Any] | None: return obj if isinstance(obj, dict) else None -_MIGRATIONS_VERSION = "8" +_MIGRATIONS_VERSION = "9" _RAW_PAYLOAD_MAX_BYTES = 64 * 1024 @@ -1006,8 +1097,12 @@ def init_db() -> None: if stored and stored["value"] == _MIGRATIONS_VERSION: return _backfill(conn) + # Older Event rows predate Origin; normalize them before migrations + # whose reconciliation policy depends on distinguishing hook/import. + conn.execute("UPDATE events SET origin = 'hook' WHERE origin IS NULL") _recategorize_network_calls(conn) _recategorize_subagents(conn) + _repair_claude_desktop_events(conn) _recategorize_cursor_tools(conn) _recategorize_web_targets(conn) _recategorize_questions(conn) @@ -1020,9 +1115,6 @@ def init_db() -> None: _backfill_cursor_questions(conn) _recategorize_other_tools(conn) _purge_import_for_hook_sessions(conn) - conn.execute( - "UPDATE events SET origin = 'hook' WHERE origin IS NULL" - ) conn.execute( "INSERT INTO settings (key, value) VALUES ('migrations_version', ?)" " ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -1264,7 +1356,8 @@ def _raw_received_at(raw: Any) -> str: return timeutil.now() -def append_raw_ingest( +def _append_raw_ingest( + conn: sqlite3.Connection, source: str, raw: Any, *, @@ -1275,29 +1368,47 @@ def append_raw_ingest( payload, truncated = _raw_payload_text(raw) now = timeutil.now() received_at = _raw_received_at(raw) + cur = conn.execute( + "INSERT INTO raw_ingest_events (source, origin, received_at, session_id_guess," + " raw_kind, raw_payload, raw_payload_truncated, raw_hash, parser_version," + " agent_version, status, projection_error, created_at)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + source, + origin, + received_at, + _session_id_guess(source, raw), + _raw_kind(raw), + payload, + truncated, + _raw_hash(payload), + __version__, + _agent_version(raw), + status, + projection_error, + now, + ), + ) + return int(cur.lastrowid) + + +def append_raw_ingest( + source: str, + raw: Any, + *, + origin: str = "hook", + status: str = "pending", + projection_error: str | None = None, +) -> int: with store.write() as conn: - cur = conn.execute( - "INSERT INTO raw_ingest_events (source, origin, received_at, session_id_guess," - " raw_kind, raw_payload, raw_payload_truncated, raw_hash, parser_version," - " agent_version, status, projection_error, created_at)" - " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - ( - source, - origin, - received_at, - _session_id_guess(source, raw), - _raw_kind(raw), - payload, - truncated, - _raw_hash(payload), - __version__, - _agent_version(raw), - status, - projection_error, - now, - ), + return _append_raw_ingest( + conn, + source, + raw, + origin=origin, + status=status, + projection_error=projection_error, ) - return int(cur.lastrowid) def mark_raw_ingest( @@ -1352,7 +1463,7 @@ def record_ingest(source: str, raw: dict[str, Any]) -> dict[str, Any]: raw_id = append_raw_ingest(source, raw, origin="import" if raw.get("_import") else "hook") try: norm = normalize(source, raw) - if should_ignore_event(norm): + if should_ignore_event(norm, raw): mark_raw_ingest(raw_id, "ignored") return { "ok": True, @@ -1431,17 +1542,19 @@ def record_event( ) return (sid, event_id, False) if return_status else (sid, event_id) + boundary = lifecycle_boundary(norm["source"], norm["hook"]) row = conn.execute("SELECT id FROM sessions WHERE id = ?", (sid,)).fetchone() - if row is None or norm["hook"] in _SESSION_START_HOOKS: + if row is None or boundary == "session_start": if row is None: conn.execute( "INSERT INTO sessions (id, source, cwd, started_at, status, created_at)" " VALUES (?, ?, ?, ?, 'active', ?)", (sid, norm["source"], norm["cwd"], ts, timeutil.now()), ) - elif norm["hook"] in _SESSION_START_HOOKS: + elif boundary == "session_start": conn.execute( - "UPDATE sessions SET status = 'active', cwd = COALESCE(?, cwd)" + "UPDATE sessions SET status = 'active', ended_at = NULL," + " cwd = COALESCE(?, cwd)" " WHERE id = ?", (norm["cwd"], sid), ) @@ -1460,7 +1573,7 @@ def record_event( (ts, sid, ts), ) - if norm["hook"] in _SESSION_END_HOOKS: + if boundary == "session_end": conn.execute( "UPDATE sessions SET status = 'completed', ended_at = ? WHERE id = ?", (ts, sid), diff --git a/backend/app/normalize.py b/backend/app/normalize.py index 97b217e..cdbf712 100644 --- a/backend/app/normalize.py +++ b/backend/app/normalize.py @@ -4,7 +4,7 @@ import json import re -from typing import Any +from typing import Any, Literal from . import timeutil from .tool_classification import ( @@ -20,6 +20,7 @@ ) Source = str # 'claude' | 'cursor' | 'codex' +LifecycleBoundary = Literal["session_start", "turn_end", "session_end"] APPROVAL_REVIEW_PREFIX = "The following is the Codex agent history" _START_HOOKS = { @@ -56,6 +57,17 @@ def _phase(hook: str) -> str: return "instant" +def lifecycle_boundary(source: Source, hook: str) -> LifecycleBoundary | None: + """Classify lifecycle hooks once for both display and Session state.""" + if hook in ("SessionStart", "sessionStart"): + return "session_start" + if hook in ("SessionEnd", "sessionEnd"): + return "session_end" + if hook in ("Stop", "stop"): + return "turn_end" if source == "claude" else "session_end" + return None + + def _short(text: str | None, limit: int = 80) -> str: if not text: return "" @@ -182,7 +194,8 @@ def categorize(source: Source, hook: str, body: dict[str, Any], tool: str | None } # --- Lifecycle --- - if hook in ("SessionStart", "sessionStart"): + boundary = lifecycle_boundary(source, hook) + if boundary == "session_start": return { "category": "lifecycle", "title": "Session started", @@ -191,7 +204,7 @@ def categorize(source: Source, hook: str, body: dict[str, Any], tool: str | None "status": "ok", "duration_ms": duration_ms, } - if hook in ("SessionEnd", "sessionEnd", "Stop", "stop"): + if boundary == "session_end": return { "category": "lifecycle", "title": "Session ended", @@ -200,6 +213,15 @@ def categorize(source: Source, hook: str, body: dict[str, Any], tool: str | None "status": "ok", "duration_ms": duration_ms, } + if boundary == "turn_end": + return { + "category": "lifecycle", + "title": "Turn ended", + "target": None, + "detail": _json_detail(body), + "status": "ok", + "duration_ms": duration_ms, + } # --- Subagents (before lifecycle used to swallow subagentStop) --- if hook in ("subagentStart", "SubagentStart", "subagentStop", "SubagentStop"): diff --git a/backend/tests/test_raw_ingest_drift.py b/backend/tests/test_raw_ingest_drift.py index a9b647a..c5e3338 100644 --- a/backend/tests/test_raw_ingest_drift.py +++ b/backend/tests/test_raw_ingest_drift.py @@ -89,6 +89,201 @@ def test_namespaced_live_event_supersedes_import_and_retry_is_idempotent(): assert db.session_origins()[sid] == "hook" +def test_claude_stop_ends_turn_session_end_closes_and_session_start_reopens(): + sid = "claude-desktop-lifecycle" + db.record_ingest( + "claude", + { + "hook_event_name": "SessionStart", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:20:29Z", + }, + ) + db.record_ingest( + "claude", + { + "hook_event_name": "Stop", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:20:58Z", + "last_assistant_message": "Check complete.", + }, + ) + + detail = db.get_session_detail(sid) + assert detail is not None + assert detail["summary"]["ended_at"] is None + assert [event["title"] for event in detail["events"]] == [ + "Session started", + "Turn ended", + ] + + db.record_ingest( + "claude", + { + "hook_event_name": "SessionEnd", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:21:30Z", + "reason": "other", + }, + ) + + detail = db.get_session_detail(sid) + assert detail is not None + assert detail["summary"]["ended_at"] == "2026-07-16T03:21:30+00:00" + assert [event["title"] for event in detail["events"]] == [ + "Session started", + "Turn ended", + "Session ended", + ] + + db.record_ingest( + "claude", + { + "hook_event_name": "SessionStart", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:22:00Z", + }, + ) + + detail = db.get_session_detail(sid) + assert detail is not None + assert detail["summary"]["ended_at"] is None + + +def test_claude_desktop_suggestion_subagent_is_evidence_not_a_timeline_event(): + sid = "claude-desktop-suggestions" + db.record_ingest( + "claude", + { + "hook_event_name": "SessionStart", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:20:29Z", + }, + ) + + suggestion = db.record_ingest( + "claude", + { + "hook_event_name": "SubagentStop", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:21:15Z", + "agent_id": "a1ea05adf9a1d9b17", + "agent_type": "", + "agent_transcript_path": "/repo/subagents/agent-a1ea05adf9a1d9b17.jsonl", + "last_assistant_message": "update the readme", + }, + ) + explicit = db.record_ingest( + "claude", + { + "hook_event_name": "SubagentStop", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:22:15Z", + "agent_id": "af7b1027e6e6d8880", + "agent_type": "Explore", + "agent_transcript_path": "/repo/subagents/agent-af7b1027e6e6d8880.jsonl", + "last_assistant_message": "Repository inspection complete.", + }, + ) + + assert suggestion["raw_status"] == "ignored" + assert suggestion["event_id"] is None + assert explicit["raw_status"] == "projected" + detail = db.get_session_detail(sid) + assert detail is not None + subagents = [ + event for event in detail["events"] if event["category"] == "subagent" + ] + assert len(subagents) == 1 + assert subagents[0]["title"] == "Explore" + + +def test_migration_repairs_stored_claude_desktop_lifecycle_and_suggestions(): + sid = "stored-claude-desktop-session" + db.record_ingest( + "claude", + { + "hook_event_name": "SessionStart", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:20:29Z", + }, + ) + stop = db.record_ingest( + "claude", + { + "hook_event_name": "Stop", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:20:58Z", + "last_assistant_message": "Check complete.", + }, + ) + suggestion_payload = { + "hook_event_name": "SubagentStop", + "session_id": sid, + "cwd": "/repo", + "timestamp": "2026-07-16T03:21:15Z", + "agent_id": "a1ea05adf9a1d9b17", + "agent_type": "", + "last_assistant_message": "update the readme", + } + with store.write() as conn: + store.insert_event( + conn, + session_id=sid, + source="claude", + hook="SubagentStop", + phase="end", + ts="2026-07-16T03:21:15Z", + payload=suggestion_payload, + category="subagent", + title="Subagent", + detail="update the readme", + target="a1ea05adf9a1d9b17", + status="ok", + created_at="2026-07-16T03:21:15Z", + ) + conn.execute( + "UPDATE events SET title = 'Session ended' WHERE id = ?", + (stop["event_id"],), + ) + conn.execute( + "UPDATE sessions SET status = 'completed', ended_at = ? WHERE id = ?", + ("2026-07-16T03:20:58Z", sid), + ) + conn.execute( + "UPDATE settings SET value = 'pre-claude-desktop-repair'" + " WHERE key = 'migrations_version'" + ) + conn.execute("UPDATE events SET origin = NULL WHERE session_id = ?", (sid,)) + + db.init_db() + + detail = db.get_session_detail(sid) + assert detail is not None + assert detail["summary"]["ended_at"] is None + assert [event["title"] for event in detail["events"]] == [ + "Session started", + "Turn ended", + ] + suggestion_raw = [ + row + for row in db.raw_ingest_events() + if row["session_id_guess"] == sid and row["raw_kind"] == "SubagentStop" + ] + assert len(suggestion_raw) == 1 + assert suggestion_raw[0]["status"] == "ignored" + assert suggestion_raw[0]["event_id"] is None + + def test_malformed_raw_input_is_evidence_not_a_timeline_event(): result = db.record_malformed_ingest( "codex", From e84e576a0ad25e844540754ebb244e400c62b668 Mon Sep 17 00:00:00 2001 From: Yash Dave Date: Thu, 16 Jul 2026 14:42:05 +0530 Subject: [PATCH 3/3] Keep implementation docs local --- CONTEXT.md | 42 ---------- ...e-scanners-emit-canonical-ingest-events.md | 17 ---- .../live-scanners-canonical-ingest-events.md | 79 ------------------- docs/validation/live-scanner-replay.md | 67 ---------------- 4 files changed, 205 deletions(-) delete mode 100644 CONTEXT.md delete mode 100644 docs/adr/0002-live-scanners-emit-canonical-ingest-events.md delete mode 100644 docs/specs/live-scanners-canonical-ingest-events.md delete mode 100644 docs/validation/live-scanner-replay.md diff --git a/CONTEXT.md b/CONTEXT.md deleted file mode 100644 index 575337f..0000000 --- a/CONTEXT.md +++ /dev/null @@ -1,42 +0,0 @@ -# cot - -Self-hosted observability for coding agents: the Bridge captures agent activity on the developer's machine and the Collector turns it into Sessions, Events, and Insights served to the dashboard. - -## Language - -### Ingest - -**Event**: -One normalized record of agent activity (a hook firing, tool call, prompt, or response) belonging to a Session. -_Avoid_: trace, log entry, record - -**Session**: -One agent working session, identified by the agent's own session id; the unit the dashboard lists and the timeline renders. -_Avoid_: conversation, run (a run is a segment within a session's timeline) - -**Origin**: -An Event's provenance — `hook` when delivered live by an agent hook, `import` when recovered from a transcript on disk. -_Avoid_: source (source is the agent: claude, cursor, codex) - -**Hook data wins**: -The reconciliation policy for a Session that has both origins: live hook Events supersede previously imported Events, which are purged. - -**Ingest Event**: -The canonical, kind-keyed record of one piece of agent activity parsed from a transcript (prompt, response, thought, tool_call, plan, attachment, …) — the same shape regardless of Origin, and the form agent activity takes before it becomes an Event. -_Avoid_: hook payload, synthetic event, artifact (for this shape) - -### Storage - -**Store**: -The single seam through which Events and Sessions are read from and written to disk. Callers may read with SQL; connection lifecycle, write discipline, and the Event row shape belong to the Store alone. -_Avoid_: database layer, repository, DAL - -### Components - -**Collector**: -The self-hosted service that ingests Events and serves the dashboard. -_Avoid_: server, backend (as a noun for the component) - -**Bridge**: -The on-machine CLI that wires agent hooks, scans transcripts, and posts to the Collector. -_Avoid_: agent (reserved for the coding agents being observed), client diff --git a/docs/adr/0002-live-scanners-emit-canonical-ingest-events.md b/docs/adr/0002-live-scanners-emit-canonical-ingest-events.md deleted file mode 100644 index 7e15989..0000000 --- a/docs/adr/0002-live-scanners-emit-canonical-ingest-events.md +++ /dev/null @@ -1,17 +0,0 @@ -# Live scanners emit canonical Ingest Events; trigger-context behaviour stays in live shells - -Status: accepted - -The Bridge captures transcripts on two paths — live (hook-triggered delta scans) and import — and they used to encode "transcript → Collector payload" independently, which is where both shipped scanner bugs lived. We decided both Origins share one line-parser per agent, emitting canonical Ingest Events, adapted to the wire format in exactly one place. The live path's genuine differences are handled by design, not by a second parser: generic post-steps on the event list (kind filtering with usage folding, timestamp rewrite, interruption flagging, consecutive-response collapse), plus a thin per-agent live shell for what only the trigger context knows (when to scan, pairing recovered question answers, the suppress-default-post signal, safety-net final text). - -## Considered Options - -- **Separate live scanners that merely return canonical events** — rejected: keeps two parsers per agent, preserving the exact duplication that produced the dropped-response and duplicated-response bugs. -- **Events + sidecar return shape for live extras** — rejected: every sidecar fact (plan-turn flag, questions) is derivable from the event list plus the hook body the shell already holds; a sidecar is a second copy with one authority available. -- **Question turns as synthetic pre/post pairs (old live behaviour)** — rejected in favour of the import parser's single instant tool_call; Cursor transcripts carry no tool results, so the pair was two rows rendering as one item. - -## Consequences - -- Live shells must not grow parsing or payload knowledge; if a shell needs a new fact from the transcript, the shared parser emits it as an Ingest Event (or marker kind, like `interruption`) instead. -- Live scanner Events carry origin-namespaced explicit dedup keys, making re-scans idempotent without colliding with import keys or disturbing hook-data-wins reconciliation. -- Agreed fallback if Cursor's replay diff proves noisy: Cursor keeps a bespoke scanner that still emits canonical Ingest Events — parser sharing is sacrificed, the vocabulary and single adapter are not. diff --git a/docs/specs/live-scanners-canonical-ingest-events.md b/docs/specs/live-scanners-canonical-ingest-events.md deleted file mode 100644 index 9d47b06..0000000 --- a/docs/specs/live-scanners-canonical-ingest-events.md +++ /dev/null @@ -1,79 +0,0 @@ -# Route the Bridge's live scanners through the canonical Ingest Event seam - -Status: implemented; replay validation pending - -## Problem Statement - -The Bridge captures agent activity twice over: live, when hooks fire during a Session, and by import, when transcripts are recovered from disk. The two paths encode "how a transcript becomes Events" independently. As a user, the same transcript produces materially different Sessions depending on Origin: imported Cursor plan-mode Sessions render blank where live ones show a structured plan; live Cursor Sessions are missing the agent's thoughts that import recovers; repeated commentary blocks are collapsed live but spam imported timelines; and token usage is accounted differently per path. As a maintainer, every transcript-format change and every payload bug must be found and fixed twice — two shipped bugs (dropped assistant responses, duplicated Cursor responses) lived exactly in this duplication, and the live path has no tests because scanning, payload mapping, and posting are fused into single functions. - -## Solution - -The import path already owns a deep seam: per-agent parsers emit canonical Ingest Events (kind-keyed records — prompt, response, thought, tool_call, attachment, session_metadata), and a single adapter maps them onto the Collector's wire format. This change routes the live path through that same seam. Live scanners return canonical Ingest Events from the same shared per-agent line-parsers that import uses; a thin per-agent live shell handles only what is genuinely trigger-context behaviour (when to scan, pairing recovered question answers, the suppress-default-post protocol). The canonical vocabulary grows to absorb Cursor's specials: a `plan` kind and an `interruption` marker. One parser per agent, one adapter, both Origins — a payload fix lands once and covers live and import alike, and the live path becomes testable exactly the way the import path already is. - -## User Stories - -1. As a developer observing my agent sessions, I want a Cursor Session to look the same whether it was captured live or imported from a transcript, so that the dashboard tells one truthful story per Session. -2. As a developer observing my agent sessions, I want imported Cursor plan-mode Sessions to show the structured plan (name, overview, body, todos), so that plan turns no longer render blank in imported history. -3. As a developer observing my agent sessions, I want live Cursor Sessions to include the agent's thoughts, so that live capture is not poorer than import for the same transcript. -4. As a developer observing my agent sessions, I want repeated identical commentary blocks collapsed in imported Sessions the same way they are live, so that imported Cursor timelines are not padded with dozens of duplicate Events. -5. As a developer observing my agent sessions, I want token usage totals preserved when live scanning drops Events that hooks already delivered, so that cost tracking does not silently undercount. -6. As a developer observing my agent sessions, I want a user-interruption to flag the Event it cut off in both Origins' data, so that interrupted turns are identifiable in the timeline. -7. As a developer whose Bridge crashes or loses its offset state, I want re-scanned transcripts to be recognized as already ingested, so that recovery never duplicates my Session history. -8. As a developer whose agent double-fires a hook, I want scanner-recovered Events deduplicated by stable keys rather than a five-second window, so that slow retries cannot create duplicate Events. -9. As a developer with an imported Session that later goes live, I want live Events reconciled by the hook-data-wins policy rather than swallowed as duplicates of imported rows, so that resuming an imported Session works correctly. -10. As a developer using Cursor's AskQuestion flow, I want the question and its recovered answer captured as one coherent Event pairing in both Origins, so that clarification exchanges are visible in every Session. -11. As a maintainer, I want the transcript shape of each agent known to exactly one parser, so that a format change is a one-module edit regardless of Origin. -12. As a maintainer, I want the Collector wire format known only to the single adapter, so that payload bugs concentrate in one place and a fix covers both Origins. -13. As a maintainer, I want the live shells reduced to trigger-context orchestration, so that reading a shell tells me when scanning happens, not how payloads are built. -14. As a maintainer, I want the live path covered by tests at the same seams as the import path, so that the previously untested half of the Bridge stops being where bugs ship. -15. As a maintainer, I want golden-master recordings of today's live behaviour frozen before the refactor, so that unwritten behaviours (plan-only-turn fallback, safety-net final text) cannot be silently lost. -16. As a maintainer, I want a replay diff of the new path against my real transcripts and the raw ingest ledger before merge, so that every behaviour change is either on the expected-differences list or treated as a bug. -17. As a maintainer, I want Claude and Codex unified and gated before Cursor, so that the riskiest agent cannot stall the proven wins. -18. As a maintainer, I want a documented fallback for Cursor (bespoke scanner still emitting canonical Ingest Events), so that a noisy replay diff has a pre-agreed exit that preserves the vocabulary win. -19. As a future agent-source implementer, I want adding a new agent to mean writing one line-parser that emits canonical Ingest Events, so that both live and import capture come for free. -20. As an AI coding agent working on this repo, I want the Ingest Event vocabulary defined in the domain glossary and the seam recorded in an ADR, so that I can navigate to the right module without reverse-engineering two pipelines. -21. As a dashboard user, I want Cursor question turns to appear as a single instant tool Event rather than a synthetic start/end pair, so that question turns read as one item without a fake duration. -22. As a maintainer of the drift ledger, I want live and import to project through the same mapping, so that drift metrics measure agent behaviour rather than pipeline divergence. - -## Implementation Decisions - -- **Scope: all three agents** (Claude, Codex, Cursor). The unification is void without Cursor; its differences are handled by design, not exclusion. -- **The live scanner interface is a list of canonical Ingest Events** — the same type the import parsers return. No sidecar return values: everything the live path needs beyond the events is derivable from the event list plus the hook body the shell already holds. -- **One shared line-parser per agent**, used by both Origins. Live shells become: read transcript delta → feed lines through the shared parser → apply generic post-steps → adapt → post. -- **Four generic post-steps**, each a pure transformation on the event list, written once and shared across agents: - 1. Per-agent kind filter — drop Events the agent's hook surface already delivers live — with usage folding, so tokens on dropped Events land on the next kept Event and totals are preserved. - 2. Timestamp rewrite — Cursor live Events take the hook timestamp plus ordering offsets instead of the import path's mtime synthesis. - 3. Interruption flagging — the parser emits a canonical `interruption` marker Event; the shared post-step flags the preceding Event in both Origins and then removes the marker before adaptation. - 4. Consecutive-identical response collapse — applied to both Origins, so imported Cursor Sessions get the same cleanup live ones do. -- **Vocabulary additions:** a `plan` kind carrying name, overview, body, and todos, mapped by the adapter onto the plan wire shape the Collector's normalize layer already accepts; questions become a single instant tool_call (adopting the import parser's documented choice), retiring the live path's synthetic pre/post pair. -- **Trigger-context behaviour stays in thin per-agent live shells:** which hooks trigger a scan; pairing the hook body's text to recovered question Events and posting answers to the dedicated answer endpoint; the suppress-default-post signal (derived from whether the scan yielded response or plan Events); the safety-net append of the hook's final text; Codex's fallback to the last assistant message when no transcript is readable. -- **Dedup: live scanner Events carry explicit, origin-namespaced dedup keys** (Claude: uuid-derived; Cursor: absolute byte offset plus block index). Namespacing guarantees no collision with import keys, so hook-data-wins reconciliation is untouched, while re-scans and hook retries become idempotent under the Collector's existing explicit-key dedup regime. Non-scanner hook payloads keep the existing five-second window behaviour. -- **Zero Collector changes.** The adapter still emits hook-style payloads; the plan wire shape and the explicit-dedup semantics already exist server-side. -- **Sequencing: two replay-gated changes.** Change one: seam infrastructure, post-steps, and the Claude and Codex live shells. Change two: Cursor (plan kind, instant questions, shared collapse, answer pairing in the shell). If Cursor's replay diff is noisy, change two falls back to a bespoke Cursor scanner that still emits canonical Ingest Events — vocabulary and adapter unification survive even in the fallback. -- **Side effects:** add "Ingest Event" to the domain glossary; record an ADR ("live scanners emit canonical Ingest Events; trigger-context behaviour stays in live shells") so future architecture passes don't re-litigate the shells' existence or re-specialize the scanners. - -## Testing Decisions - -- **Good tests here exercise external behaviour at a seam:** transcript content in, canonical Ingest Events or posted payloads out. No test reaches past an interface to assert on parser internals, offset state, or intermediate shapes. -- **Two test seams, both already existing in the code:** - 1. **The canonical Ingest Event seam** — transcript lines in, canonical Events out. Covers the shared parsers (plan kind, instant questions, interruption marker, dedup key minting) and the pure post-steps (list in, list out). Prior art: the import path's canonical contract tests, which already test the parsers this way. - 2. **The posting seam** — the Bridge's single exit. Tests intercept the post function and assert ordered (endpoint, payload) sequences from a hook body plus a transcript fixture. Covers live-shell orchestration: suppress-default protocol, answer pairing, safety-net text, Codex's no-transcript fallback. -- **Golden masters recorded before any refactor:** today's live emit behaviour captured at the posting seam over fixtures covering Cursor's four hard turns — a plan turn, a question turn with and without a recoverable answer, the repeated-commentary pathology, and a plan-only turn with no text. After the refactor the same fixtures must reproduce the recordings modulo the expected-differences list; the goldens then graduate into the live path's permanent suite. -- **Merge gate: replay diff against real data** — the developer's actual transcripts and the raw ingest ledger's recorded payloads as the oracle. Every diff must appear on the expected-differences list or be treated as a regression. -- **Expected-differences list (seed):** Cursor live gains thought Events; Codex live gains standalone reasoning-summary thought Events already recovered by import; question pre/post pairs become single instant Events; live Events gain dedup keys; Claude usage placement shifts from pending-fold to per-block with folding on filtered Events (totals preserved); imported Cursor Sessions gain plan Events and lose consecutive duplicate responses; imported Claude and Cursor output immediately followed by an interruption marker gains `interrupted: true` (Codex import already carried this signal); imported Claude Sessions recover standalone file-attachment records that match their parent prompt. -- **Final confidence pass:** one live smoke Session per agent against a dev Collector, eyeballing the timeline beside an equivalent old-path Session. - -## Out of Scope - -- Any Collector-side changes: wire format, normalize layer, dedup regimes, the hook-data-wins policy, and the Store seam are all untouched. -- Removing the Collector-side Cursor transcript scan (the question-recovery backfill that duplicates Bridge knowledge). It is a separate deepening candidate; this change may make it redundant, but verifying and deleting it is its own piece of work. -- Typing the Store's read queries (settled by the existing ADR) and any category-vocabulary registry work. -- The Bridge's hook wiring, install, backup, and docker lifecycle concepts — untested today, but not part of this seam. -- Changing what the dashboard renders beyond what falls out of unified Event streams. -- Migrating or rewriting historical Sessions already in the Store; only re-imports and new captures flow through the unified path. - -## Further Notes - -- The riskiest behaviour cluster is Cursor's live shell (plan-only fallback, answer pairing, suppress-default). The design isolates it precisely so the fallback can amputate Cursor's parser sharing without losing the vocabulary and adapter unification. -- The interruption marker and the plan kind slightly widen the canonical vocabulary; both are consumed Bridge-side (marker) or already understood Collector-side (plan), so the wire format's surface does not grow. -- Import automatically gains plan Events by virtue of the shared parser — flagged during design as closing a drift bug, not scope creep; the replay diff quantifies exactly what it changes in existing history. diff --git a/docs/validation/live-scanner-replay.md b/docs/validation/live-scanner-replay.md deleted file mode 100644 index 0f3a362..0000000 --- a/docs/validation/live-scanner-replay.md +++ /dev/null @@ -1,67 +0,0 @@ -# Live scanner canonical-event validation - -**Status: not merge-ready.** Cursor transcript replay is complete. A current -raw-ingest-ledger diff and disposable live smoke Sessions for all three agents -remain open merge gates. Cursor's documented fallback was not needed; the -shared-parser path matched the expected semantic differences. - -Validated on 2026-07-15 against base `8d035a0` and the current implementation -commit. The replay compared semantic hashes only; transcript contents were not -printed or copied. - -## Posting-seam recordings - -`backend/tests/test_bridge_live.py` freezes ordered endpoint/payload behavior -for Cursor plan turns, question turns with and without a recoverable answer, -consecutive repeated commentary, plan-only fallback text, and the hook-text -safety net. It also covers Claude and Codex live shells at the same posting -seam. - -These recordings were reconstructed from the `8d035a0` behavior and added in -the implementation commit, rather than committed before the refactor. They are -useful permanent regression tests, but they are not independent historical -evidence and do not satisfy the real-data replay gate by themselves. - -Hook-data-wins reconciliation and namespaced-live-key retry idempotency are -covered in the Collector suite by -`backend/tests/test_raw_ingest_drift.py::test_namespaced_live_event_supersedes_import_and_retry_is_idempotent`. - -## Local transcript replay - -The machine had 11 Claude transcripts and 20 readable Codex transcripts. The -old live scanners and new canonical path produced identical response, -attachment, and web-search multisets for every file. Claude Event counts also -matched exactly. Codex gained only standalone reasoning-summary thought Events, -which is listed in the spec's expected differences because import already -recovered them. - -Cursor replay used a purpose-built, non-sensitive fixture Session containing 30 -valid main-transcript records and a three-record subagent transcript. The old -scanner and shared-parser path agreed on one structured plan, one AskQuestion, -and all non-duplicate response semantics. The new path removed exactly two -consecutive-identical responses, matching the expected-differences list. Cursor -stored the attached file as textual mentions rather than a structured -attachment record, so both paths emitted zero attachment Events. - -The replay exposed Cursor's current interruption shape: a `turn_ended` record -with `status` `error` or `aborted` and a user-abort error string. The shared -parser now maps that shape to the canonical interruption marker. Two main-turn -interruptions and one subagent interruption were recovered; the subagent marker -followed a tool call and correctly flagged the preceding retained response after -tool filtering. The real shape is locked at the posting seam in -`backend/tests/test_bridge_live.py`. - -The local `~/.cot/cot.db` still predates `raw_ingest_events`, and no dev -Collector was running during the fixture Session, so no raw-ledger oracle was -available. That input remains required before merge. - -`CONTEXT.md` was supplied as an explicit implementation input and committed -intact. Its broader glossary/component material was therefore not authored as -part of this refactor; the relevant implementation side effect is the included -Ingest Event definition. - -## Smoke status - -No dev Collector was running for disposable live Sessions, so the three-agent -live smoke remains a pre-merge manual gate. The implementation does not mutate -the developer's installed hooks or production Collector as part of validation.