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_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..c5e3338 100644 --- a/backend/tests/test_raw_ingest_drift.py +++ b/backend/tests/test_raw_ingest_drift.py @@ -55,6 +55,235 @@ 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_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", 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.