From d60845dc0e83205c57e6da91a26deae130795988 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:08:41 +0200 Subject: [PATCH 1/5] fix: recover native parser evidence and read outcomes (#128) --- CHANGELOG.md | 6 ++ agent_observer/adapters/claude.py | 6 +- agent_observer/adapters/codex.py | 80 +++++++++++++-- agent_observer/adapters/grok.py | 67 +++++++++--- agent_observer/analysis.py | 4 +- agent_observer/identity.py | 12 ++- agent_observer/ingest.py | 4 +- agent_observer/privacy.py | 18 ++-- tests/test_analysis.py | 2 + tests/test_codex_fork.py | 6 +- tests/test_codex_user_identity.py | 2 +- tests/test_native_events.py | 4 +- tests/test_parser_gaps.py | 162 ++++++++++++++++++++++++++++++ 13 files changed, 332 insertions(+), 41 deletions(-) create mode 100644 tests/test_parser_gaps.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 54a16a3..9023306 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## [Unreleased] + +### Fixed + +- Parse sectioned AgentsMD direction identity, Codex native durations and Read calls, and Grok MCP timing and outcomes. Retain read success state and count only successful procedure reads; replay older sources to repair stored evidence (#128). + ## [1.3.0] - 2026-10-19 ### Added diff --git a/agent_observer/adapters/claude.py b/agent_observer/adapters/claude.py index 953b24d..3fd7bdb 100644 --- a/agent_observer/adapters/claude.py +++ b/agent_observer/adapters/claude.py @@ -934,7 +934,7 @@ def _user(r: _Reader, obj: dict, ordinal: int, ts) -> None: skill = privacy.filter_target(raw_skill, family="skill_read") r.event("skill_read", obj.get("uuid") or f"ordinal:{ordinal}", ordinal, ts, - name=skill, target=skill, size_bytes=len(text), + name=skill, target=skill, status="ok", size_bytes=len(text), detail={"skill": raw_skill, "skill_path": skill_file} if (raw_skill or skill_file) else None) native_kind = _native_user_kind(obj, text, agent_id=r.agent_id, @@ -997,13 +997,13 @@ def _tool_result(r: _Reader, obj: dict, result: dict, ordinal: int, ts) -> None: lines = (file_info or {}).get("numLines") if family == "skill_read": r.event(family, f"{call_id}:read", ordinal, ts, name=os.path.basename(path), - target=safe_skill, size_bytes=size, + target=safe_skill, status="ok", size_bytes=size, fingerprint=fingerprint(path, start, lines), detail={"start_line": start, "num_lines": lines, "skill": raw_skill, "skill_path": path}) else: r.event(family, f"{call_id}:read", ordinal, ts, name=os.path.basename(path), - target=path, size_bytes=size, + target=path, status="ok", size_bytes=size, fingerprint=fingerprint(path, start, lines), detail={"start_line": start, "num_lines": lines}) diff --git a/agent_observer/adapters/codex.py b/agent_observer/adapters/codex.py index 4bb13c4..ea9c758 100644 --- a/agent_observer/adapters/codex.py +++ b/agent_observer/adapters/codex.py @@ -14,6 +14,7 @@ import glob import json +import math import os import sqlite3 @@ -78,11 +79,11 @@ def _occurrences_ready(row) -> bool: ("lifecycle_task", True, "event_msg task_started/task_complete with turn timing; turn_aborted marks cancelled turns"), ("lifecycle_dispatch", True, "collaboration.spawn_agent call plus SubAgentActivity started; child join needs explicit delegation capture"), ("file_change", True, "FileChange items with path, change kind, content size/hash; contents never stored"), - ("read_evidence", True, "CommandExecution parsed_cmd entries of type read; target/size/owner/window preserved"), + ("read_evidence", True, "CommandExecution parsed_cmd reads and Read function calls; attempted/pending/failed/ok state retained"), ("skill_file_reads", True, "read events under an installed Skill directory; mentions never count"), ("skill_invocation", False, "no native skill-invocation event observed in Codex sample; stays unknown"), - ("timing_tool", True, "CommandExecution duration when reported; otherwise unknown, never invented"), - ("instruction_identity", True, "AgentsMD direction block in injected messages and versioned plugin paths read"), + ("timing_tool", True, "CommandExecution numeric and {secs,nanos} durations; otherwise unknown"), + ("instruction_identity", True, "AgentsMD legacy and nonce-section direction blocks; successfully read versioned plugin paths"), ("quota", False, "rate-limit snapshots are account evidence, not task attribution; stays unknown at task level"), ] @@ -1850,6 +1851,11 @@ def _ingest_response_item(r: _Reader, obj: dict, native_ordinal, position: int) args, custom=ptype == "custom_tool_call"), status=p.get("status"), fingerprint=fingerprint(name, str(args)[:2000])) + if name == "Read": + target = _call_target(args) + if target: + _function_read(r, obj, call_id or p.get("id"), target, "attempted", turn_id) + elif ptype in ("function_call_output", "custom_tool_call_output"): call_id = p.get("call_id") out = p.get("output") @@ -1860,6 +1866,19 @@ def _ingest_response_item(r: _Reader, obj: dict, native_ordinal, position: int) size = len(str(out or "")) r.event(obj, "tool_result", call_id or p.get("id"), turn_id=turn_id, name=ptype, size_bytes=size) + prior = r.con.execute( + "SELECT family,target,detail_json FROM events WHERE session_key=? AND native_id=?" + " AND family IN ('read','skill_read')", (r.session_key, call_id)).fetchone() + if prior: + state = "failed" if p.get("is_error") is True else _read_output_state(out) + if state == "ok": + path = (json.loads(prior["detail_json"] or "{}").get("skill_path") + if prior["family"] == "skill_read" else prior["target"]) + if path: + r.identity.observe_path(path) + r.con.execute("UPDATE events SET status=?,size_bytes=? WHERE session_key=?" + " AND native_id=? AND family IN ('read','skill_read')", + (state, size, r.session_key, call_id)) _PATH_KEYS = ("file_path", "path", "filePath", "target_file", "filename") @@ -2075,7 +2094,8 @@ def _ingest_command(r: _Reader, obj: dict, item: dict, turn_id) -> None: target = entry.get("name") if not isinstance(target, str) or not target: continue - r.identity.observe_path(target) + if _command_read_state(item) == "ok": + r.identity.observe_path(target) raw_skill = skill_from_path(target) safe_skill = privacy.filter_target(raw_skill, family="skill_read") \ if raw_skill else None @@ -2088,7 +2108,7 @@ def _ingest_command(r: _Reader, obj: dict, item: dict, turn_id) -> None: # only in detail skill_path. r.event(obj, fam, f"{item.get('id')}:{target}", turn_id=turn_id, name=os.path.basename(target), target=safe_skill, - status=item.get("status"), + status=_command_read_state(item), duration_ms=_duration(item), size_bytes=size, fingerprint=fingerprint(target), detail={"cmd": entry.get("cmd"), "skill": raw_skill, @@ -2096,7 +2116,7 @@ def _ingest_command(r: _Reader, obj: dict, item: dict, turn_id) -> None: else: r.event(obj, fam, f"{item.get('id')}:{target}", turn_id=turn_id, name=os.path.basename(target), target=target, - status=item.get("status"), + status=_command_read_state(item), duration_ms=_duration(item), size_bytes=size, fingerprint=fingerprint(target), detail={"cmd": entry.get("cmd")}) @@ -2120,7 +2140,51 @@ def _ingest_compacted(r: _Reader, obj: dict) -> None: def _duration(item: dict): - d = item.get("duration") or item.get("duration_ms") or item.get("elapsed") - if isinstance(d, (int, float)): + d = next((item[k] for k in ("duration", "duration_ms", "elapsed") + if item.get(k) is not None), None) + if isinstance(d, dict): + secs, nanos = d.get("secs"), d.get("nanos") + if type(secs) is int and type(nanos) is int and secs >= 0 and 0 <= nanos < 10**9: + return secs * 1000 + nanos // 10**6 + return None + if type(d) in (int, float) and math.isfinite(d) and d >= 0: return int(d * 1000) if d < 100000 and d != int(d) else int(d) return None + + +def _command_read_state(item): + code = item.get("exit_code") + if type(code) is int: + return "ok" if code == 0 else "failed" + status = item.get("status") + if status in ("failed", "error", "cancelled", "denied"): + return "failed" + if status in ("in_progress", "running", "pending"): + return "pending" + return "attempted" + + +def _read_output_state(output): + if isinstance(output, str): + try: + output = json.loads(output) + except ValueError: + # A terminal native Read result is file text unless the tool + # explicitly reports an error. Never retain its contents. + return "failed" if output.startswith(("Error:", "Error reading file:")) else "ok" + if isinstance(output, dict): + if output.get("is_error") is True or output.get("error") or output.get("status") in ("failed", "error"): + return "failed" + if output.get("is_error") is False or output.get("status") in ("success", "completed", "ok") or "content" in output: + return "ok" + if isinstance(output, list) and output and all(isinstance(x, dict) and x.get("type") == "text" for x in output): + return "ok" + return "attempted" + + +def _function_read(r, obj, call_id, path, state, turn_id): + skill = skill_from_path(path) + family = "skill_read" if skill or path.endswith("SKILL.md") or "/skills/" in path else "read" + r.event(obj, family, call_id, turn_id=turn_id, name=os.path.basename(path), + target=skill if family == "skill_read" else path, status=state, + detail={"skill": skill, "skill_path": path} if family == "skill_read" else None) diff --git a/agent_observer/adapters/grok.py b/agent_observer/adapters/grok.py index dff5cdc..ce591a6 100644 --- a/agent_observer/adapters/grok.py +++ b/agent_observer/adapters/grok.py @@ -37,8 +37,8 @@ CAPABILITIES = [ ("model_usage", True, "one responses row per turn_completed prompt_id; input includes cached reads, output includes reasoning"), ("tool_calls", True, "tool_call records with validated native tool name, safe target and argument fingerprint; raw input never stored"), - ("tool_results", True, "tool_call_update status joined with events.jsonl tool_completed duration and outcome on toolCallId"), - ("read_evidence", True, "tool_call_update kind=read locations as read events with paths and line metadata"), + ("tool_results", True, "tool_call_update and tool_completed plus MCP start/completion records joined by call ID; duration, success, timeout and retries retained"), + ("read_evidence", True, "tool_call_update kind=read locations with attempted/pending/failed/ok state"), ("skill_file_reads", True, "reads under an installed Skill directory as skill_read with the skill name"), ("skill_invocation", False, "no native skill-invocation event observed in Grok records; stays unknown"), ("compaction", True, "auto_compact_started and auto_compact_completed as compaction events"), @@ -2625,8 +2625,9 @@ def _tool_call_update(r: _Reader, src: JsonlSource, update: dict, continue paths.append(path) if paths: - for path in paths: - r.identity.observe_path(path) + if _read_state(update.get("status")) == "ok": + for path in paths: + r.identity.observe_path(path) raw_skill = skill_from_path(paths[0]) skill_path = paths[0] # A second location in the same call may name the Skill while the @@ -2651,7 +2652,7 @@ def _tool_call_update(r: _Reader, src: JsonlSource, update: dict, r.event(src, family, call_id, ordinal, ts, turn_id=turn_id, name=safe_skill or os.path.basename(paths[0]), - target=safe_skill, + target=safe_skill, status=_read_state(update.get("status")), fingerprint=fingerprint(call_id, paths), detail={"skill": raw_skill, "skill_path": skill_path}) else: @@ -2660,10 +2661,15 @@ def _tool_call_update(r: _Reader, src: JsonlSource, update: dict, r.event(src, family, call_id, ordinal, ts, turn_id=turn_id, name=os.path.basename(paths[0]), - target=paths[0], + target=paths[0], status=_read_state(update.get("status")), fingerprint=fingerprint(call_id, paths), detail=None) status = update.get("status") + # A terminal update often omits locations. Join the prior read by call ID. + if status is not None: + r.con.execute("UPDATE events SET status=? WHERE session_key=? AND native_id=?" + " AND family IN ('read','skill_read')", + (_read_state(status), r.session_key, call_id)) if isinstance(status, str) and status in TERMINAL_TOOL_STATUS: # Tool results keep no detail under privacy.py rule 6; the mapped # terminal status is the ledger column. @@ -2724,6 +2730,21 @@ def _ingest_event(r: _Reader, src: JsonlSource, obj: dict, outcome = _valid_enum(obj.get("outcome"), _OUTCOME_VALUES) r.event(src, "lifecycle", f"turn_ended:{ordinal}", ordinal, ts, name="turn_ended", status=outcome) + elif kind == "mcp_tool_call_started": + call_id = _safe_token(obj.get("call_id")) + if call_id is None: + raise _AdapterError("schema_error") + prior = r.con.execute("SELECT id FROM events WHERE session_key=? AND family='tool_call' AND native_id=?", + (r.session_key, call_id)).fetchone() + timeout = obj.get("timeout_sec") + detail = {"timeout_sec": timeout} if type(timeout) is int and timeout >= 0 else {} + if prior: + r.con.execute("UPDATE events SET detail_json=? WHERE id=?", (json.dumps(detail) if detail else None, prior['id'])) + else: + r.event(src, "tool_call", call_id, ordinal, ts, + name=_safe_token(obj.get("tool_name")) or "unknown", detail=detail) + elif kind == "mcp_tool_call_completed": + _tool_completed(r, src, obj, ordinal, ts) elif kind == "tool_started": tool_s = _safe_token(obj.get("tool_name")) or "unknown" r.event(src, "lifecycle", f"tool_started:{ordinal}", ordinal, ts, @@ -2761,14 +2782,24 @@ def _ingest_event(r: _Reader, src: JsonlSource, obj: dict, def _tool_completed(r: _Reader, src: JsonlSource, obj: dict, ordinal: int, ts) -> None: - raw_call = obj.get("tool_call_id") + is_mcp = obj.get("type") == "mcp_tool_call_completed" + raw_call = obj.get("call_id") if is_mcp else obj.get("tool_call_id") if not isinstance(raw_call, str) or _safe_token(raw_call) is None: raise _AdapterError("schema_error") call_id = raw_call outcome = obj.get("outcome") - status = "ok" if outcome in ("success", "completed") else "error" + if is_mcp and type(obj.get("success")) is bool: + status = "ok" if obj["success"] else "error" + elif outcome in ("success", "completed"): + status = "ok" + elif outcome in ("error", "failed", "failure", "cancelled"): + status = "error" + else: + status = None + mcp_detail = {k: obj[k] for k in ("is_timeout", "auth_retry_attempted", "reconnect_attempted") + if type(obj.get(k)) is bool} duration = obj.get("duration_ms") - if isinstance(duration, bool) or not isinstance(duration, int): + if isinstance(duration, bool) or not isinstance(duration, int) or duration < 0: duration = None row = r.con.execute( "SELECT id, status, duration_ms, detail_json FROM events" @@ -2777,7 +2808,7 @@ def _tool_completed(r: _Reader, src: JsonlSource, obj: dict, ordinal: int, if row is None: tool_s = _safe_token(obj.get("tool_name")) or "unknown" r.event(src, "tool_result", call_id, ordinal, ts, - name=tool_s, status=status, duration_ms=duration) + name=tool_s, status=status, duration_ms=duration, detail=mcp_detail) return try: loaded = json.loads(row["detail_json"]) if row["detail_json"] else {} @@ -2786,16 +2817,24 @@ def _tool_completed(r: _Reader, src: JsonlSource, obj: dict, ordinal: int, # Scrub any legacy values through privacy.py rule 6: only allowlisted # tool_result keys with correctly typed values survive, so unknown keys # and free text are dropped, never stringified. - detail = privacy.filter_detail("tool_result", loaded) + detail = privacy.filter_detail("tool_result", {**loaded, **mcp_detail}) payload = json.dumps(detail, sort_keys=True) if detail else None changed = payload != row["detail_json"] if row["duration_ms"] is None and duration is not None: changed = True - if row["status"] is None and status: + if status is not None and (row["status"] is None or is_mcp and row["status"] != status): changed = True if changed: r.con.execute( "UPDATE events SET duration_ms=COALESCE(duration_ms, ?)," - " status=COALESCE(status, ?), detail_json=? WHERE id=?", - (duration, status, payload, + " status=?, detail_json=? WHERE id=?", + (duration, status if is_mcp and status is not None else row["status"] or status, payload, row["id"])) + + +def _read_state(status): + if status in ("completed", "success"): + return "ok" + if status in ("failed", "error", "cancelled"): + return "failed" + return "pending" if status in ("in_progress", "pending", "running") else "attempted" diff --git a/agent_observer/analysis.py b/agent_observer/analysis.py index 19b8177..53c79e9 100644 --- a/agent_observer/analysis.py +++ b/agent_observer/analysis.py @@ -353,6 +353,8 @@ def _repeated_skills(session, events) -> list: del seen[key] if e["family"] not in ("skill_read", "skill_invoke"): continue + if e["family"] == "skill_read" and e["status"] not in ("ok", "success", "completed"): + continue name = (e["name"] or e["target"] or "").lower() if e["family"] == "skill_read": name = (_detail(e).get("skill") or name).lower() @@ -605,7 +607,7 @@ def _session_metrics(con, s) -> dict: "SELECT kind, COUNT(*) n FROM user_messages WHERE session_key=? GROUP BY kind", (key,))} reading = con.execute( "SELECT COUNT(*) n, COALESCE(SUM(size_bytes), 0) b FROM events WHERE session_key=?" - " AND family IN ('read', 'skill_read')" + " AND family IN ('read', 'skill_read') AND status IN ('ok','success','completed')" " AND (target LIKE ? OR detail_json LIKE ?)", (key, f"%{AGENTSMD_PATH}%", f"%{AGENTSMD_PATH}%") ).fetchone() diff --git a/agent_observer/identity.py b/agent_observer/identity.py index 33bbdf9..5f7c434 100644 --- a/agent_observer/identity.py +++ b/agent_observer/identity.py @@ -22,7 +22,7 @@ import subprocess BLOCK_RE = re.compile( - r"<<>>\s*(\{.*?\})\s*<<>>", + r"<<>>\s*(\{.*?\})(?=\s*<<<(?:END_AGENTSMD_PROJECT_DIRECTION_V1>>>|AGENTSMD_SECTION ))", re.S) VERSION_PATH_RE = re.compile(r"/agentsmd/(\d+\.\d+\.\d+)/") SKILL_PATH_RE = re.compile(r"(?:/|^)skills/([A-Za-z0-9_.-]+)/") @@ -124,6 +124,16 @@ def parse_direction_block(text: str) -> dict | None: return {"status": "unparsed"} if not isinstance(block, dict): return {"status": "unparsed"} + suffix = text[match.end():].lstrip() + nonce = block.get("sections") + if nonce is not None: + if not isinstance(nonce, str) or not re.fullmatch(r"[0-9a-f]{16}", nonce): + return {"status": "unparsed"} + if not suffix.startswith(f"<<>>" not in suffix: + return {"status": "unparsed"} + elif not suffix.startswith("<<>>"): + return {"status": "unparsed"} result: dict = {} status = block.get("status") if isinstance(status, str) and status in DIRECTION_STATUSES: diff --git a/agent_observer/ingest.py b/agent_observer/ingest.py index e3cd76e..0182ff4 100644 --- a/agent_observer/ingest.py +++ b/agent_observer/ingest.py @@ -358,8 +358,8 @@ def insert_event(con: sqlite3.Connection, stats: dict, *, source_id: int, truncated, fingerprint, payload)) if cur.rowcount == 0: if update: - fields = "name=?, target=?, status=?, detail_json=?" - values = [safe_name, safe_target, safe_status, payload] + fields = "name=?, target=?, status=?, detail_json=?, duration_ms=?" + values = [safe_name, safe_target, safe_status, payload, duration_ms] if migrate_ordinal: fields += ", ordinal_num=?" values.append(ordinal) diff --git a/agent_observer/privacy.py b/agent_observer/privacy.py index df9c391..9d2e3f7 100644 --- a/agent_observer/privacy.py +++ b/agent_observer/privacy.py @@ -28,7 +28,8 @@ # Rule 3: bump when any rule in this module changes meaning. A stored source # version that differs forces a full re-import with in-place correction. -PRIVACY_VERSION = 9 +# Re-derive native parser evidence and successful-read state on old sources. +PRIVACY_VERSION = 10 SUBMISSION_EXCERPT_CHARS = 300 ASSISTANT_EXCERPT_CHARS = 400 @@ -188,12 +189,14 @@ def line_excerpt(line: object) -> str: # or other free text. EVENT_DETAIL_ALLOWLIST: dict[str, dict[str, str]] = { "file_change": {"paths": "paths"}, - "tool_result": {"exit_code": "int", "exitCode": "int"}, + "tool_result": {"exit_code": "int", "exitCode": "int", + "is_timeout": "bool", "auth_retry_attempted": "bool", + "reconnect_attempted": "bool"}, "read": {"start_line": "int", "num_lines": "int", "cmd": "command"}, "skill_read": {"start_line": "int", "num_lines": "int", "cmd": "command", "skill": "identifier", "skill_path": "path"}, "assistant_message": {"excerpt": "excerpt"}, - "tool_call": {}, + "tool_call": {"timeout_sec": "int"}, "skill_invoke": {"skill_path": "path"}, "compaction": {}, "lifecycle": {}, @@ -612,7 +615,10 @@ def filter_detail(family: str, detail: object) -> dict: if key not in detail: continue value = detail[key] - if kind == "int": + if kind == "bool": + if type(value) is bool: + out[key] = value + elif kind == "int": if _valid_int(value): out[key] = value elif kind == "identifier": @@ -728,8 +734,8 @@ def filter_event_name(family: str, value: object) -> str | None: EVENT_STATUS_ENUMS: dict[str, frozenset] = { "tool_call": _COMMON_STATUS | frozenset({"cancelled"}), "tool_result": _COMMON_STATUS | frozenset({"cancelled"}), - "read": _COMMON_STATUS | frozenset({"cancelled"}), - "skill_read": _COMMON_STATUS | frozenset({"cancelled"}), + "read": _COMMON_STATUS | frozenset({"cancelled", "attempted", "pending"}), + "skill_read": _COMMON_STATUS | frozenset({"cancelled", "attempted", "pending"}), "lifecycle": frozenset( {"completed", "cancelled", "interrupted", "denied", "ok", "error", "success", "failed"}), diff --git a/tests/test_analysis.py b/tests/test_analysis.py index 2627418..8322ce2 100644 --- a/tests/test_analysis.py +++ b/tests/test_analysis.py @@ -16,6 +16,8 @@ def session(self, key, version="12.1.0", project="/p/app", started=1000.0): def event(self, key, family, name=None, target=None, status=None, ts=None, detail=None): + if status is None and family == "skill_read": + status = "ok" self.seq += 1 self.con.execute( "INSERT INTO events(session_key, ts, family, native_id, name, target," diff --git a/tests/test_codex_fork.py b/tests/test_codex_fork.py index a46846a..fe52cbc 100644 --- a/tests/test_codex_fork.py +++ b/tests/test_codex_fork.py @@ -517,7 +517,7 @@ def test_privacy8_sparse_fork_moves_copied_history_before_replay(self): "SELECT COUNT(*) FROM events WHERE native_id='parent-text'" ).fetchone()[0], 1) self.assertEqual(self.con.execute( - "SELECT privacy_version FROM sources").fetchone()[0], 9) + "SELECT privacy_version FROM sources").fetchone()[0], 10) self.assert_child_owns_its_rows() @@ -809,7 +809,7 @@ def test_shared_raw_ordinal_keeps_each_explicit_identity(self): self.assertEqual( index.records[found[key]]["payload"]["id"], bare, key) self.assertEqual(con.execute( - "SELECT privacy_version FROM sources").fetchone()[0], 9) + "SELECT privacy_version FROM sources").fetchone()[0], 10) finally: con.close() @@ -1047,7 +1047,7 @@ def test_base_and_current_seeds_keep_answer_and_abort(self): "codex:c-" + seed) self.assertEqual(con.execute( "SELECT occurrence_backfill, privacy_version" - " FROM sources").fetchone()[:], (1, 9)) + " FROM sources").fetchone()[:], (1, 10)) self.assertEqual(con.execute( "SELECT COUNT(*) FROM import_errors").fetchone()[0], 0) current_ids = [row["id"] for row in rows] diff --git a/tests/test_codex_user_identity.py b/tests/test_codex_user_identity.py index 335c04c..443e0bb 100644 --- a/tests/test_codex_user_identity.py +++ b/tests/test_codex_user_identity.py @@ -102,7 +102,7 @@ def _resync(self, path, conflicts=0): result = codex.sync(self.con, source=path, full=full) self.assertEqual(result["failed"], []) self.assertEqual(result["user_messages_inserted"], 0) - self.assertEqual(self.query("SELECT privacy_version FROM sources")[0][0], 9) + self.assertEqual(self.query("SELECT privacy_version FROM sources")[0][0], 10) self.assertEqual(self.query("PRAGMA foreign_key_check"), []) self.assertEqual([tuple(r) for r in self.query( "SELECT error, COUNT(*) FROM import_errors GROUP BY error")], diff --git a/tests/test_native_events.py b/tests/test_native_events.py index b432289..3d94fb5 100644 --- a/tests/test_native_events.py +++ b/tests/test_native_events.py @@ -970,7 +970,7 @@ def activity(turns): self.assertEqual(ledger.execute( "PRAGMA foreign_key_check").fetchall(), []) self.assertEqual(ledger.execute( - "SELECT privacy_version FROM sources").fetchone()[0], 9) + "SELECT privacy_version FROM sources").fetchone()[0], 10) finally: ledger.close() @@ -1150,7 +1150,7 @@ def test_privacy8_archive_move_keeps_original_ownership(self): "PRAGMA foreign_key_check").fetchall(), []) self.assertEqual(ledger.execute( "SELECT privacy_version FROM sources WHERE id=?", - (moved_source,)).fetchone()[0], 9) + (moved_source,)).fetchone()[0], 10) finally: ledger.close() diff --git a/tests/test_parser_gaps.py b/tests/test_parser_gaps.py new file mode 100644 index 0000000..7d23aa4 --- /dev/null +++ b/tests/test_parser_gaps.py @@ -0,0 +1,162 @@ +"""Native fixture regressions for the five capture gaps in #128.""" +import json +import os + +from agent_observer import analysis, identity, privacy +from agent_observer.adapters import codex, grok +from tests.helpers import LedgerCase + + +class ParserGapsTest(LedgerCase): + def write(self, name, records): + path = os.path.join(self.tmp.name, name) + with open(path, 'w') as f: + for record in records: + f.write(json.dumps(record) + '\n') + return path + + def codex(self, items): + records = [{'type': 'session_meta', 'payload': {'id': 'gaps', 'source': 'cli'}}] + records += [{'type': 'event_msg', 'payload': {'type': 'item_completed', 'item': x}} + for x in items] + return self.write('rollout-gaps.jsonl', records) + + def command(self, id='cmd', exit_code=0, **extra): + return dict(type='CommandExecution', id=id, command=['cat', '/repo/agentsmd/skills/operations/SKILL.md'], + parsed_cmd=[{'type': 'read', 'path': '/repo/agentsmd/skills/operations/SKILL.md'}], + status='completed', exit_code=exit_code, **extra) + + def test_sectioned_direction_and_legacy_fail_closed(self): + header = {'status': 'ready', 'instructions': {'sha256': 'a' * 64}, + 'sections': '0123456789abcdef'} + body = ('<<>>\n' + json.dumps(header) + + '\n<<>>\n' + 'private text {"instructions": {"sha256": "bad"}}\n' + '<<>>') + self.assertEqual(identity.parse_direction_block(body)['instructions_sha256'], 'a' * 64) + self.assertNotIn('private text', json.dumps(identity.parse_direction_block(body))) + self.assertEqual(identity.parse_direction_block(body.replace('abcdef>>>', 'abcdee>>>')), + {'status': 'unparsed'}) + legacy = '<<>>\n' + json.dumps({'status': 'ready'}) + '\n<<>>' + self.assertEqual(identity.parse_direction_block(legacy), {'status': 'ready'}) + self.assertEqual(identity.parse_direction_block(body.replace(json.dumps(header), 'free text')), + {'status': 'unparsed'}) + + def test_codex_native_duration_and_replay(self): + path = self.codex([self.command(duration={'secs': 2, 'nanos': 345000000})]) + codex.sync(self.con, source=path) + self.assertEqual(self.query("SELECT duration_ms FROM events WHERE family='tool_result'")[0][0], 2345) + self.con.execute('UPDATE events SET duration_ms=NULL') + self.con.execute('UPDATE sources SET privacy_version=9') + self.con.commit() + for _ in range(2): + codex.sync(self.con, source=path) + self.assertEqual(self.query("SELECT duration_ms FROM events WHERE family='tool_result'")[0][0], 2345) + self.assertEqual(self.query('SELECT COUNT(*) FROM events')[0][0], 2) + self.assertEqual(codex._duration({'duration': 1.5}), 1500) + self.assertEqual(codex._duration({'duration_ms': 0}), 0) + for value in ({'secs': True, 'nanos': 0}, {'secs': -1, 'nanos': 0}, + {'secs': 1, 'nanos': 1000000000}): + self.assertIsNone(codex._duration({'duration': value})) + + def test_codex_read_states_and_success_report(self): + path = self.codex([self.command('ok'), self.command('bad', 1), + dict(self.command('pending'), status='in_progress', exit_code=None)]) + codex.sync(self.con, source=path) + states = {r['native_id'].split(':')[0]: r['status'] for r in + self.query("SELECT * FROM events WHERE family='skill_read'")} + self.assertEqual(states, {'ok': 'ok', 'bad': 'failed', 'pending': 'pending'}) + session = self.query('SELECT * FROM sessions')[0] + self.assertEqual(analysis._session_metrics(self.con, session)['agentsmd_reads'], 1) + # Repeated failed/pending procedure reads must not look like repeated loads. + self.assertEqual(analysis._repeated_skills(session, self.query('SELECT * FROM events')), []) + + def test_codex_read_function_result_and_pending(self): + records = [{'type': 'session_meta', 'payload': {'id': 'reads', 'source': 'cli'}}] + for call, path, output in [('skill', '/repo/agentsmd/skills/operations/SKILL.md', {'content': 'fixture'}), + ('file', '/repo/a.py', 'fixture text'), + ('bad', '/repo/b.py', {'is_error': True}), + ('pending', '/repo/c.py', None)]: + records.append({'type': 'response_item', 'payload': {'type': 'function_call', + 'name': 'Read', 'call_id': call, 'arguments': json.dumps({'file_path': path})}}) + if output is not None: + records.append({'type': 'response_item', 'payload': {'type': 'function_call_output', + 'call_id': call, 'output': output if isinstance(output, str) else json.dumps(output)}}) + path = self.write('rollout-read.jsonl', records) + codex.sync(self.con, source=path) + rows = {r['native_id']: (r['family'], r['status']) for r in self.query( + "SELECT * FROM events WHERE family IN ('read','skill_read')")} + self.assertEqual(rows, {'skill': ('skill_read', 'ok'), 'file': ('read', 'ok'), + 'bad': ('read', 'failed'), 'pending': ('read', 'attempted')}) + # Incremental result joins the persisted call without needing an in-memory pending map. + with open(path, 'a') as f: + f.write(json.dumps({'type': 'response_item', 'payload': {'type': 'function_call_output', + 'call_id': 'pending', 'output': '{"content":"fixture"}'}}) + '\n') + codex.sync(self.con, source=path) + self.assertEqual(self.query("SELECT status FROM events WHERE family='read' AND native_id='pending'")[0][0], 'ok') + + def grok_fixture(self, updates, events): + path = os.path.join(self.tmp.name, 'grok-gaps'); os.mkdir(path) + with open(os.path.join(path, 'summary.json'), 'w') as f: + json.dump({'info': {'id': 'grok-gaps', 'cwd': '/repo'}}, f) + self.write('grok-gaps/updates.jsonl', [ + {'method': 'session/update', 'params': {'sessionId': 'grok-gaps', + '_meta': {'eventId': str(i), 'promptId': 'p1'}, 'update': x}} + for i, x in enumerate(updates)]) + self.write('grok-gaps/events.jsonl', events) + return path + + def test_grok_mcp_joins_and_outcomes(self): + path = self.grok_fixture([ + {'sessionUpdate': 'tool_call', 'toolCallId': 'm1', '_meta': {'x.ai/tool': {'name': 'lookup'}}}, + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'm1', 'status': 'completed'}], [ + {'type': 'mcp_tool_call_started', 'call_id': 'm1', 'tool_name': 'lookup', 'timeout_sec': 30}, + {'type': 'mcp_tool_call_completed', 'call_id': 'm1', 'tool_name': 'lookup', + 'duration_ms': 234, 'success': True, 'is_timeout': False, + 'auth_retry_attempted': False, 'reconnect_attempted': True}]) + for full in (False, True, False): + grok.sync(self.con, source=path, full=full) + self.assertEqual(self.query('SELECT COUNT(*) FROM import_errors')[0][0], 0) + self.assertEqual(self.query("SELECT COUNT(*) FROM events WHERE family='tool_call'")[0][0], 1) + row = self.query("SELECT * FROM events WHERE family='tool_result'")[0] + self.assertEqual((row['duration_ms'], row['status']), (234, 'ok')) + self.assertEqual(json.loads(row['detail_json'])['reconnect_attempted'], True) + + def test_grok_read_state(self): + path = self.grok_fixture([ + {'sessionUpdate': 'tool_call_update', 'toolCallId': call, 'kind': 'read', + 'status': status, 'locations': [{'path': '/repo/agentsmd/skills/operations/SKILL.md'}]} + for call, status in [('ok', 'completed'), ('bad', 'failed'), ('pending', 'in_progress')]], []) + grok.sync(self.con, source=path) + self.assertEqual({r['native_id']: r['status'] for r in self.query( + "SELECT * FROM events WHERE family='skill_read'")}, + {'ok': 'ok', 'bad': 'failed', 'pending': 'pending'}) + + def test_grok_completion_without_locations_updates_prior_read(self): + path = self.grok_fixture([ + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'r1', 'kind': 'read', + 'status': 'in_progress', 'locations': [{'path': '/repo/a.py'}]}], []) + grok.sync(self.con, source=path) + with open(os.path.join(path, 'updates.jsonl'), 'a') as f: + f.write(json.dumps({'method': 'session/update', 'params': {'sessionId': 'grok-gaps', + '_meta': {'eventId': 'late', 'promptId': 'p1'}, 'update': { + 'sessionUpdate': 'tool_call_update', 'toolCallId': 'r1', 'status': 'failed'}}}) + '\n') + grok.sync(self.con, source=path) + self.assertEqual(self.query("SELECT status FROM events WHERE family='read'")[0][0], 'failed') + + def test_grok_mcp_timeout_and_privacy_repair(self): + path = self.grok_fixture([], [ + {'type': 'mcp_tool_call_started', 'call_id': 'm2', 'tool_name': 'lookup', 'timeout_sec': 30}, + {'type': 'mcp_tool_call_completed', 'call_id': 'm2', 'tool_name': 'lookup', + 'duration_ms': 30000, 'success': False, 'is_timeout': True, + 'auth_retry_attempted': True, 'reconnect_attempted': False, 'error': 'private error'}]) + grok.sync(self.con, source=path) + self.con.execute('UPDATE sources SET privacy_version=9') + self.con.commit() + for _ in range(2): + grok.sync(self.con, source=path) + rows = self.query("SELECT * FROM events WHERE family='tool_result'") + self.assertEqual(len(rows), 1) + self.assertEqual((rows[0]['status'], rows[0]['duration_ms']), ('error', 30000)) + self.assertTrue(json.loads(rows[0]['detail_json'])['is_timeout']) + self.assertNotIn('private error', str(dict(rows[0]))) From d9b717c558c461993d10c59af006eaec511fd49f Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:13:14 +0200 Subject: [PATCH 2/5] fix: preserve opaque Read content and late read identity --- agent_observer/adapters/codex.py | 12 +++++------- agent_observer/adapters/grok.py | 9 +++++++++ tests/test_parser_gaps.py | 16 +++++++++++++++- 3 files changed, 29 insertions(+), 8 deletions(-) diff --git a/agent_observer/adapters/codex.py b/agent_observer/adapters/codex.py index ea9c758..a70d78c 100644 --- a/agent_observer/adapters/codex.py +++ b/agent_observer/adapters/codex.py @@ -1870,7 +1870,8 @@ def _ingest_response_item(r: _Reader, obj: dict, native_ordinal, position: int) "SELECT family,target,detail_json FROM events WHERE session_key=? AND native_id=?" " AND family IN ('read','skill_read')", (r.session_key, call_id)).fetchone() if prior: - state = "failed" if p.get("is_error") is True else _read_output_state(out) + state = "failed" if p.get("is_error") is True or p.get("status") in ( + "failed", "error", "cancelled", "denied") else _read_output_state(out) if state == "ok": path = (json.loads(prior["detail_json"] or "{}").get("skill_path") if prior["family"] == "skill_read" else prior["target"]) @@ -2166,12 +2167,9 @@ def _command_read_state(item): def _read_output_state(output): if isinstance(output, str): - try: - output = json.loads(output) - except ValueError: - # A terminal native Read result is file text unless the tool - # explicitly reports an error. Never retain its contents. - return "failed" if output.startswith(("Error:", "Error reading file:")) else "ok" + # File contents are opaque, including JSON and error-shaped prose. + # Only native result metadata can distinguish a failed Read. + return "ok" if isinstance(output, dict): if output.get("is_error") is True or output.get("error") or output.get("status") in ("failed", "error"): return "failed" diff --git a/agent_observer/adapters/grok.py b/agent_observer/adapters/grok.py index ce591a6..d434014 100644 --- a/agent_observer/adapters/grok.py +++ b/agent_observer/adapters/grok.py @@ -2666,6 +2666,15 @@ def _tool_call_update(r: _Reader, src: JsonlSource, update: dict, detail=None) status = update.get("status") # A terminal update often omits locations. Join the prior read by call ID. + if _read_state(status) == "ok": + for read in r.con.execute( + "SELECT family,target,detail_json FROM events WHERE session_key=?" + " AND native_id=? AND family IN ('read','skill_read')", + (r.session_key, call_id)): + path = (json.loads(read["detail_json"] or "{}").get("skill_path") + if read["family"] == "skill_read" else read["target"]) + if path: + r.identity.observe_path(path) if status is not None: r.con.execute("UPDATE events SET status=? WHERE session_key=? AND native_id=?" " AND family IN ('read','skill_read')", diff --git a/tests/test_parser_gaps.py b/tests/test_parser_gaps.py index 7d23aa4..d3d7c09 100644 --- a/tests/test_parser_gaps.py +++ b/tests/test_parser_gaps.py @@ -81,7 +81,7 @@ def test_codex_read_function_result_and_pending(self): 'name': 'Read', 'call_id': call, 'arguments': json.dumps({'file_path': path})}}) if output is not None: records.append({'type': 'response_item', 'payload': {'type': 'function_call_output', - 'call_id': call, 'output': output if isinstance(output, str) else json.dumps(output)}}) + 'call_id': call, 'output': output}}) path = self.write('rollout-read.jsonl', records) codex.sync(self.con, source=path) rows = {r['native_id']: (r['family'], r['status']) for r in self.query( @@ -160,3 +160,17 @@ def test_grok_mcp_timeout_and_privacy_repair(self): self.assertEqual((rows[0]['status'], rows[0]['duration_ms']), ('error', 30000)) self.assertTrue(json.loads(rows[0]['detail_json'])['is_timeout']) self.assertNotIn('private error', str(dict(rows[0]))) + + def test_codex_read_json_file_contents_are_not_result_metadata(self): + for text in ('{"version":1}', '[]', '123', '{"error":"example"}', 'Error: example from a file'): + self.assertEqual(codex._read_output_state(text), 'ok') + + def test_grok_locationless_success_retains_version_evidence(self): + path = self.grok_fixture([ + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'r2', 'kind': 'read', + 'status': 'in_progress', 'locations': [ + {'path': '/plugins/agentsmd/17.3.1/skills/operations/SKILL.md'}]}, + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'r2', 'status': 'completed'}], []) + grok.sync(self.con, source=path) + self.assertEqual(self.query("SELECT agentsmd_version FROM sessions")[0][0], '17.3.1') + self.assertEqual(self.query("SELECT status FROM events WHERE family='skill_read'")[0][0], 'ok') From 634fb2671ece0522a40b7b45d3bf9827fedac4c0 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:18:23 +0200 Subject: [PATCH 3/5] fix: count each successful Grok read identity once --- agent_observer/adapters/grok.py | 11 +++++++++-- tests/test_parser_gaps.py | 19 +++++++++++++++++++ 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/agent_observer/adapters/grok.py b/agent_observer/adapters/grok.py index d434014..57a35ed 100644 --- a/agent_observer/adapters/grok.py +++ b/agent_observer/adapters/grok.py @@ -435,6 +435,7 @@ def __init__(self, con, session_key, native_sid, stats, sync_ctx=None): self.native_sid = native_sid self.stats = stats self.identity = SessionIdentity() + self.read_identity_seen: set[tuple[str, str]] = set() self.first_ts = None self.last_ts = None self.meta: dict = {} @@ -449,6 +450,12 @@ def note_ts(self, ts) -> None: self.first_ts = ts if self.first_ts is None else min(self.first_ts, ts) self.last_ts = ts if self.last_ts is None else max(self.last_ts, ts) + def observe_read_path(self, call_id: str, path: str) -> None: + key = (call_id, path) + if key not in self.read_identity_seen: + self.read_identity_seen.add(key) + self.identity.observe_path(path) + def event(self, src: JsonlSource, family, native_id, ordinal, ts, **kw): # Every target and detail mapping passes through # agent_observer/privacy.py rule 6 in insert_event. On a privacy @@ -2627,7 +2634,7 @@ def _tool_call_update(r: _Reader, src: JsonlSource, update: dict, if paths: if _read_state(update.get("status")) == "ok": for path in paths: - r.identity.observe_path(path) + r.observe_read_path(call_id, path) raw_skill = skill_from_path(paths[0]) skill_path = paths[0] # A second location in the same call may name the Skill while the @@ -2674,7 +2681,7 @@ def _tool_call_update(r: _Reader, src: JsonlSource, update: dict, path = (json.loads(read["detail_json"] or "{}").get("skill_path") if read["family"] == "skill_read" else read["target"]) if path: - r.identity.observe_path(path) + r.observe_read_path(call_id, path) if status is not None: r.con.execute("UPDATE events SET status=? WHERE session_key=? AND native_id=?" " AND family IN ('read','skill_read')", diff --git a/tests/test_parser_gaps.py b/tests/test_parser_gaps.py index d3d7c09..e7e64ed 100644 --- a/tests/test_parser_gaps.py +++ b/tests/test_parser_gaps.py @@ -174,3 +174,22 @@ def test_grok_locationless_success_retains_version_evidence(self): grok.sync(self.con, source=path) self.assertEqual(self.query("SELECT agentsmd_version FROM sessions")[0][0], '17.3.1') self.assertEqual(self.query("SELECT status FROM events WHERE family='skill_read'")[0][0], 'ok') + + def test_grok_each_successful_read_contributes_version_once(self): + updates = [] + for call, version, with_locations in [('a', '17.1.0', True), + ('b1', '17.3.1', False), + ('b2', '17.3.1', False)]: + location = {'path': f'/plugins/agentsmd/{version}/skills/operations/SKILL.md'} + updates.append({'sessionUpdate': 'tool_call_update', 'toolCallId': call, + 'kind': 'read', 'status': 'completed' if with_locations else 'in_progress', + 'locations': [location]}) + # Repeated witnesses of the same successful read must count once. + updates.append({'sessionUpdate': 'tool_call_update', 'toolCallId': call, + 'status': 'completed'}) + path = self.grok_fixture(updates, []) + grok.sync(self.con, source=path) + session = self.query('SELECT agentsmd_version,identity_json FROM sessions')[0] + self.assertEqual(session['agentsmd_version'], '17.3.1') + self.assertEqual(json.loads(session['identity_json'])['plugin_paths'], + {'17.1.0': 1, '17.3.1': 2}) From 1ed08473af56237cdeb4a93a48cedaa5a55e3260 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:47:06 +0200 Subject: [PATCH 4/5] fix: retain pending Grok read paths and Codex result identity --- agent_observer/adapters/codex.py | 4 +- agent_observer/adapters/grok.py | 143 ++++++++++++++++--------------- tests/test_parser_gaps.py | 41 +++++++++ 3 files changed, 119 insertions(+), 69 deletions(-) diff --git a/agent_observer/adapters/codex.py b/agent_observer/adapters/codex.py index a70d78c..ed9889e 100644 --- a/agent_observer/adapters/codex.py +++ b/agent_observer/adapters/codex.py @@ -1857,14 +1857,14 @@ def _ingest_response_item(r: _Reader, obj: dict, native_ordinal, position: int) _function_read(r, obj, call_id or p.get("id"), target, "attempted", turn_id) elif ptype in ("function_call_output", "custom_tool_call_output"): - call_id = p.get("call_id") + call_id = p.get("call_id") or p.get("id") out = p.get("output") if isinstance(out, list): size = sum(len(str(c.get("text", ""))) for c in out if isinstance(c, dict)) else: size = len(str(out or "")) - r.event(obj, "tool_result", call_id or p.get("id"), turn_id=turn_id, + r.event(obj, "tool_result", call_id, turn_id=turn_id, name=ptype, size_bytes=size) prior = r.con.execute( "SELECT family,target,detail_json FROM events WHERE session_key=? AND native_id=?" diff --git a/agent_observer/adapters/grok.py b/agent_observer/adapters/grok.py index 57a35ed..06d962f 100644 --- a/agent_observer/adapters/grok.py +++ b/agent_observer/adapters/grok.py @@ -436,6 +436,7 @@ def __init__(self, con, session_key, native_sid, stats, sync_ctx=None): self.stats = stats self.identity = SessionIdentity() self.read_identity_seen: set[tuple[str, str]] = set() + self.read_paths: dict[str, dict[str, None]] = {} self.first_ts = None self.last_ts = None self.meta: dict = {} @@ -456,6 +457,27 @@ def observe_read_path(self, call_id: str, path: str) -> None: self.read_identity_seen.add(key) self.identity.observe_path(path) + def remember_read_update(self, update: dict) -> list[str]: + call_id = update.get("toolCallId") + if not isinstance(call_id, str) or _safe_token(call_id) is None: + return [] + paths = [] + locations = update.get("locations") + if update.get("kind") == "read" and isinstance(locations, list): + for loc in locations: + if not isinstance(loc, dict) or any(k in loc for k in _FREE_TEXT_LOCATION_KEYS): + continue + path = loc.get("path") + if isinstance(path, str) and path: + paths.append(path) + if paths: + remembered = self.read_paths.setdefault(call_id, {}) + remembered.update(dict.fromkeys(paths)) + if _read_state(update.get("status")) == "ok": + for path in self.read_paths.get(call_id, {}): + self.observe_read_path(call_id, path) + return paths + def event(self, src: JsonlSource, family, native_id, ordinal, ts, **kw): # Every target and detail mapping passes through # agent_observer/privacy.py rule 6 in insert_event. On a privacy @@ -1736,6 +1758,20 @@ def _replay_error(ordinal: int, category: str, line: str = "") -> None: updates_records, events_records, (_prompts, _order, _completions, _turn_models_cached)) _advance_source_to_cached(updates_src, updates_path, updates_records) + # The complete cached prefix is already loaded for usage replay. + # Restore pending locations and successful identity evidence before + # ingesting an appended locationless completion, without new I/O or + # widening the stored event detail allowlist. + if updates_src.incremental: + for ordinal, obj, _line in updates_records: + if ordinal > updates_src.row["ordinal_max"]: + break + if not isinstance(obj, dict) or obj.get("method") not in METHODS: + continue + params = obj.get("params") + update = params.get("update") if isinstance(params, dict) else None + if isinstance(update, dict) and update.get("sessionUpdate") == "tool_call_update": + r.remember_read_update(update) for ordinal, obj, line in _delta_records( updates_records, updates_src): stats["lines"] += 1 @@ -2614,74 +2650,47 @@ def _tool_call_update(r: _Reader, src: JsonlSource, update: dict, raise _AdapterError("schema_error") call_id = raw_call turn_id = _turn_id(r, params) - locations = update.get("locations") - if update.get("kind") == "read" and isinstance(locations, list) \ - and locations: - # Fail closed: every location must be a plain {"path": str} shape - # with an optional integer line. A location bearing message/output - # style free-text keys, a non-string path, or a non-integer line is - # dropped entirely and never stringified into any ledger column. - paths: list[str] = [] - for loc in locations: - if not isinstance(loc, dict): - continue - if any(k in loc for k in _FREE_TEXT_LOCATION_KEYS): - continue - path = loc.get("path") - if not isinstance(path, str) or not path: - continue - paths.append(path) - if paths: - if _read_state(update.get("status")) == "ok": - for path in paths: - r.observe_read_path(call_id, path) - raw_skill = skill_from_path(paths[0]) - skill_path = paths[0] - # A second location in the same call may name the Skill while the - # first does not; the Skill read is the identity evidence. - if raw_skill is None: - for path in paths[1:]: - raw_skill = skill_from_path(path) - if raw_skill is not None: - skill_path = path - break - if raw_skill is not None and _safe_token(raw_skill) is None: - raw_skill = None - safe_skill = privacy.filter_target(raw_skill, family="skill_read") \ - if raw_skill else None - looks_like_skill = any( - p.endswith("SKILL.md") or "/skills/" in p for p in paths) - family = "skill_read" if (safe_skill or looks_like_skill) else "read" - if family == "skill_read": - # Rule 6: skill_read target holds only the validated skill - # identifier, never the installed path. The file path lives - # only in detail skill_path. - r.event(src, family, call_id, - ordinal, ts, turn_id=turn_id, - name=safe_skill or os.path.basename(paths[0]), - target=safe_skill, status=_read_state(update.get("status")), - fingerprint=fingerprint(call_id, paths), - detail={"skill": raw_skill, "skill_path": skill_path}) - else: - # Read events keep no detail under privacy.py rule 6; the - # first validated path is the target. - r.event(src, family, call_id, - ordinal, ts, turn_id=turn_id, - name=os.path.basename(paths[0]), - target=paths[0], status=_read_state(update.get("status")), - fingerprint=fingerprint(call_id, paths), - detail=None) + # Keep every validated location until a later terminal update, rather + # than relying on the single representative path stored on the read row. + paths = r.remember_read_update(update) + if paths: + raw_skill = skill_from_path(paths[0]) + skill_path = paths[0] + # A second location in the same call may name the Skill while the + # first does not; the Skill read is the identity evidence. + if raw_skill is None: + for path in paths[1:]: + raw_skill = skill_from_path(path) + if raw_skill is not None: + skill_path = path + break + if raw_skill is not None and _safe_token(raw_skill) is None: + raw_skill = None + safe_skill = privacy.filter_target(raw_skill, family="skill_read") \ + if raw_skill else None + looks_like_skill = any( + p.endswith("SKILL.md") or "/skills/" in p for p in paths) + family = "skill_read" if (safe_skill or looks_like_skill) else "read" + if family == "skill_read": + # Rule 6: skill_read target holds only the validated skill + # identifier, never the installed path. The file path lives + # only in detail skill_path. + r.event(src, family, call_id, + ordinal, ts, turn_id=turn_id, + name=safe_skill or os.path.basename(paths[0]), + target=safe_skill, status=_read_state(update.get("status")), + fingerprint=fingerprint(call_id, paths), + detail={"skill": raw_skill, "skill_path": skill_path}) + else: + # Read events keep no detail under privacy.py rule 6; the + # first validated path is the target. + r.event(src, family, call_id, + ordinal, ts, turn_id=turn_id, + name=os.path.basename(paths[0]), + target=paths[0], status=_read_state(update.get("status")), + fingerprint=fingerprint(call_id, paths), + detail=None) status = update.get("status") - # A terminal update often omits locations. Join the prior read by call ID. - if _read_state(status) == "ok": - for read in r.con.execute( - "SELECT family,target,detail_json FROM events WHERE session_key=?" - " AND native_id=? AND family IN ('read','skill_read')", - (r.session_key, call_id)): - path = (json.loads(read["detail_json"] or "{}").get("skill_path") - if read["family"] == "skill_read" else read["target"]) - if path: - r.observe_read_path(call_id, path) if status is not None: r.con.execute("UPDATE events SET status=? WHERE session_key=? AND native_id=?" " AND family IN ('read','skill_read')", diff --git a/tests/test_parser_gaps.py b/tests/test_parser_gaps.py index e7e64ed..f2d377b 100644 --- a/tests/test_parser_gaps.py +++ b/tests/test_parser_gaps.py @@ -193,3 +193,44 @@ def test_grok_each_successful_read_contributes_version_once(self): self.assertEqual(session['agentsmd_version'], '17.3.1') self.assertEqual(json.loads(session['identity_json'])['plugin_paths'], {'17.1.0': 1, '17.3.1': 2}) + + def test_grok_pending_multiple_paths_survive_locationless_success(self): + path = self.grok_fixture([ + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'multi', 'kind': 'read', + 'status': 'in_progress', 'locations': [ + {'path': '/plugins/agentsmd/17.1.0/README.md'}, + {'path': '/plugins/agentsmd/17.3.1/skills/operations/SKILL.md'}]}, + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'multi', 'status': 'completed'}, + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'multi', 'status': 'completed'}], []) + for full in (False, True, False): + grok.sync(self.con, source=path, full=full) + session = self.query('SELECT identity_json FROM sessions')[0] + self.assertEqual(json.loads(session['identity_json'])['plugin_paths'], + {'17.1.0': 1, '17.3.1': 1}) + + def test_grok_pending_multiple_paths_survive_incremental_completion(self): + path = self.grok_fixture([ + {'sessionUpdate': 'tool_call_update', 'toolCallId': 'multi', 'kind': 'read', + 'status': 'in_progress', 'locations': [ + {'path': '/plugins/agentsmd/17.1.0/README.md'}, + {'path': '/plugins/agentsmd/17.3.1/skills/operations/SKILL.md'}]}], []) + grok.sync(self.con, source=path) + with open(os.path.join(path, 'updates.jsonl'), 'a') as f: + f.write(json.dumps({'method': 'session/update', 'params': {'sessionId': 'grok-gaps', + '_meta': {'eventId': 'late', 'promptId': 'p1'}, 'update': { + 'sessionUpdate': 'tool_call_update', 'toolCallId': 'multi', 'status': 'completed'}}}) + '\n') + grok.sync(self.con, source=path) + session = self.query('SELECT identity_json FROM sessions')[0] + self.assertEqual(json.loads(session['identity_json'])['plugin_paths'], + {'17.1.0': 1, '17.3.1': 1}) + + def test_codex_read_output_id_fallback_matches_event_identity(self): + path = self.write('rollout-id-only.jsonl', [ + {'type': 'session_meta', 'payload': {'id': 'id-only', 'source': 'cli'}}, + {'type': 'response_item', 'payload': {'type': 'function_call', 'id': 'read-id', + 'name': 'Read', 'arguments': '{"file_path":"/repo/a.py"}'}}, + {'type': 'response_item', 'payload': {'type': 'function_call_output', + 'id': 'read-id', 'output': 'fixture'}}]) + codex.sync(self.con, source=path) + self.assertEqual(self.query("SELECT native_id,status FROM events WHERE family='read'")[0][:], + ('read-id', 'ok')) From 8647f156de05bc506f574ae94f28024fd5d3d5d7 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:51:04 +0200 Subject: [PATCH 5/5] fix: refresh reconstructed Grok read identity on append --- agent_observer/adapters/grok.py | 20 ++++++++++++++++++++ tests/test_parser_gaps.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/agent_observer/adapters/grok.py b/agent_observer/adapters/grok.py index 06d962f..a5394cf 100644 --- a/agent_observer/adapters/grok.py +++ b/agent_observer/adapters/grok.py @@ -1837,6 +1837,26 @@ def _replay_error(ordinal: int, category: str, line: str = "") -> None: updates_src.source_id if updates_src is not None else (events_src.source_id if events_src is not None else None), **fields) + if replayed_updates: + # This reader rebuilt read evidence from the whole cached source. + # Replace derived path counts instead of retaining the first sync's + # counts through upsert_session's fill-unknown policy. Other identity + # evidence stays intact; a resolved instruction hash still wins. + stored = con.execute("SELECT identity_json FROM sessions WHERE session_key=?", + (r.session_key,)).fetchone() + evidence = json.loads(stored["identity_json"] or "{}") + if r.identity.path_versions or "plugin_paths" in evidence: + if r.identity.path_versions: + evidence["plugin_paths"] = r.identity.path_versions + else: + evidence.pop("plugin_paths", None) + version = max(r.identity.path_versions, key=r.identity.path_versions.get) \ + if r.identity.path_versions else None + con.execute( + "UPDATE sessions SET identity_json=?, agentsmd_version=COALESCE(" + "(SELECT version FROM agentsmd_versions WHERE sha256=sessions.instructions_sha256), ?)" + " WHERE session_key=?", + (json.dumps(evidence, sort_keys=True) if evidence else None, version, r.session_key)) _store_meta(con, r.session_key, summary_fp, chat_fp_current, chat) if updates_src is not None: stats.update(updates_src.finish(session_id=r.native_sid)) diff --git a/tests/test_parser_gaps.py b/tests/test_parser_gaps.py index f2d377b..07ae2ab 100644 --- a/tests/test_parser_gaps.py +++ b/tests/test_parser_gaps.py @@ -234,3 +234,34 @@ def test_codex_read_output_id_fallback_matches_event_identity(self): codex.sync(self.con, source=path) self.assertEqual(self.query("SELECT native_id,status FROM events WHERE family='read'")[0][:], ('read-id', 'ok')) + + def test_grok_late_success_refreshes_existing_version_evidence(self): + updates = [] + for call, version, status in [('a', '17.1.0', 'completed'), + ('b1', '17.3.1', 'in_progress'), + ('b2', '17.3.1', 'in_progress')]: + updates.append({'sessionUpdate': 'tool_call_update', 'toolCallId': call, + 'kind': 'read', 'status': status, 'locations': [ + {'path': f'/plugins/agentsmd/{version}/skills/operations/SKILL.md'}]}) + path = self.grok_fixture(updates, []) + grok.sync(self.con, source=path) + self.assertEqual(self.query('SELECT agentsmd_version FROM sessions')[0][0], '17.1.0') + def complete(call, event): + with open(os.path.join(path, 'updates.jsonl'), 'a') as f: + f.write(json.dumps({'method': 'session/update', 'params': {'sessionId': 'grok-gaps', + '_meta': {'eventId': event, 'promptId': 'p1'}, 'update': { + 'sessionUpdate': 'tool_call_update', 'toolCallId': call, 'status': 'completed'}}}) + '\n') + complete('b1', 'late-1'); complete('b2', 'late-2') + grok.sync(self.con, source=path) + row = self.query('SELECT agentsmd_version,identity_json FROM sessions')[0] + self.assertEqual(row['agentsmd_version'], '17.3.1') + self.assertEqual(json.loads(row['identity_json'])['plugin_paths'], + {'17.1.0': 1, '17.3.1': 2}) + # An instruction hash resolved to a release outranks path evidence. + self.con.execute('INSERT INTO agentsmd_versions(sha256,version) VALUES(?,?)', + ('a' * 64, '16.0.0')) + self.con.execute('UPDATE sessions SET instructions_sha256=?', ('a' * 64,)) + self.con.commit() + complete('b1', 'repeat') + grok.sync(self.con, source=path) + self.assertEqual(self.query('SELECT agentsmd_version FROM sessions')[0][0], '16.0.0')