From 5f6b7b4d9983814788c870c08cba668b30775060 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 23:48:19 +0200 Subject: [PATCH 1/5] feat: import Muse native evidence and report unmeasured usage (#127) Add a read-only Muse adapter: one tool_call and at most one tool_result per call_id, read and skill_invoke events, run lifecycle and child links, with session-level MCP versus built-in identity. Changed files are parsed whole and reconciled in place. Report owned response-less sessions' activity, unknown or lower-bound usage for harnesses that import no usage, partial or unavailable activity coverage per native harness, unowned linked children, and linked children without logs as ledger-only. Co-Authored-By: Claude Opus 5.5 --- .claude-plugin/plugin.json | 2 +- .codex-plugin/plugin.json | 2 +- .grok-plugin/plugin.json | 2 +- CHANGELOG.md | 6 + README.md | 5 +- VERSION | 2 +- agent_observer/adapters/__init__.py | 2 +- agent_observer/adapters/muse.py | 398 +++++++++++++ agent_observer/adapters/t3.py | 4 +- agent_observer/cli.py | 19 +- agent_observer/db.py | 2 +- agent_observer/ingest.py | 28 +- agent_observer/lookup.py | 14 +- agent_observer/privacy.py | 16 +- agent_observer/publish.py | 26 +- agent_observer/report.py | 177 +++++- docs/contracts.md | 75 ++- skills/agent-observer/SKILL.md | 5 +- .../session.jsonl | 27 + .../session.jsonl | 5 + tests/test_cli.py | 4 +- tests/test_muse.py | 537 ++++++++++++++++++ tests/test_pr_checks_pending.py | 2 +- 23 files changed, 1312 insertions(+), 48 deletions(-) create mode 100644 agent_observer/adapters/muse.py create mode 100644 tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/session.jsonl create mode 100644 tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/subagent/0a000000-0000-7000-8000-0000000000c1/session.jsonl create mode 100644 tests/test_muse.py diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index b7c3275..119f2a7 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agent-observer", - "version": "1.4.2", + "version": "1.5.0", "description": "Measure agent work across Codex, Claude Code, OpenCode and Grok Build from native records, and compare behavior by model and AgentsMD version.", "author": { "name": "toolboxmd", diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index 24d0aa3..20be2e5 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agent-observer", - "version": "1.4.2", + "version": "1.5.0", "description": "Measure agent work across Codex, Claude Code, OpenCode and Grok Build from native records, and compare behavior by model and AgentsMD version.", "author": { "name": "toolboxmd", diff --git a/.grok-plugin/plugin.json b/.grok-plugin/plugin.json index 5bbb92e..9ad993a 100644 --- a/.grok-plugin/plugin.json +++ b/.grok-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agent-observer", - "version": "1.4.2", + "version": "1.5.0", "description": "Measure agent work across Codex, Claude Code, OpenCode and Grok Build from native records, and compare behavior by model and AgentsMD version.", "author": { "name": "toolboxmd", diff --git a/CHANGELOG.md b/CHANGELOG.md index d882686..e344af8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## [1.5.0] - 2026-10-19 + +### Added + +- Import Muse native tool, read and skill evidence; report unmeasured usage and partial activity coverage instead of zeros + ## [1.4.2] - 2026-10-19 ### Changed diff --git a/README.md b/README.md index 5a81e36..ee6187a 100644 --- a/README.md +++ b/README.md @@ -2,8 +2,8 @@ Local evidence about how coding agents use resources and produce outcomes. -Agent Observer reads the records Codex, Claude Code, OpenCode and Grok Build -already keep on this Mac, plus Model Router's job ledger, into one private +Agent Observer reads the records Codex, Claude Code, OpenCode, Grok Build and +Muse already keep on this machine, plus Model Router's job ledger, into one private SQLite ledger. It reports usage, timelines, repeated work and behavior incidents by session, task, model, project and AgentsMD version. It never changes the harnesses' records, never calls a model, and posts nothing to @@ -52,6 +52,7 @@ names another. | Claude Code | `~/.claude/projects/*/*.jsonl` and subagent transcripts | per message, counted once across streamed blocks | | OpenCode | `~/.local/share/opencode/opencode.db` (read-only) | per completed assistant message | | Grok Build | `~/.grok/sessions/*/*/` | per prompt completion | +| Muse | `~/.local/share/muse/sessions/*/*/*/*/session.jsonl` and `subagent/*/session.jsonl` | not imported: tool, read and skill events only; usage reads unknown | | Chromeria bare Codex and Grok | each `bareReceipt.nativeRecord.root` (`/sessions`, plus Codex's sibling `archived_sessions`) in the T3 state database the T3 adapter reads | as for that harness | | Model Router | `DURABLE_RUNNER_STATE_DIR`, `~/.local/share/durable-runner`, and legacy `~/.local/state/model-router` (read-only, deduplicated) | explicit task ownership, jobs and invocations | diff --git a/VERSION b/VERSION index 9df886c..bc80560 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.4.2 +1.5.0 diff --git a/agent_observer/adapters/__init__.py b/agent_observer/adapters/__init__.py index 224179c..a55274e 100644 --- a/agent_observer/adapters/__init__.py +++ b/agent_observer/adapters/__init__.py @@ -11,7 +11,7 @@ import importlib -ORDER = ("t3", "codex", "claude", "opencode", "grok", "router") +ORDER = ("t3", "codex", "claude", "opencode", "grok", "muse", "router") def available() -> dict: diff --git a/agent_observer/adapters/muse.py b/agent_observer/adapters/muse.py new file mode 100644 index 0000000..1cf8ae0 --- /dev/null +++ b/agent_observer/adapters/muse.py @@ -0,0 +1,398 @@ +"""Muse Code native sessions (#127). + +Muse writes one append-only ``session.jsonl`` per session under +``~/.local/share/muse/sessions/YYYY/MM/DD//`` and internal +child sessions under ``/subagent//session.jsonl``. +Each line is one frame: ``record_type``, ``payload_type``, +``schema_version``, ``payload_schema_version``, native ``sequence``, frame +``id``, ``recorded_at`` (microseconds) and ``stream {id, kind: session}``. + +One native tool call is observed several times: the committed call +(``assistant_tool_calls_committed``), a started and a terminal +``tool_batch`` effect, and the result committed to the model +(``tool_result_batch_committed``). They join only on ``call_id`` and +become one ``tool_call`` and at most one ``tool_result``; effects never +add calls. The terminal outcome gives the status, the committed text +gives the UTF-8 size; either may be missing and then stays NULL. + +Because those observations span lines, a changed file is parsed whole +(complete lines only) and each event is written with its final values +through ``insert_event(reconcile=True)``. No join state is kept between +syncs. + +MCP identity is a session-level rule: a name in the session's +``mcp_tool_identity_catalog`` is ``mcp`` (with server and raw tool), a +name only in the active toolset is ``builtin``, anything else (no +catalog or toolset, or conflicting catalog entries) is ``unknown``. A +name prefix is never evidence. + +Usage (``model_completed``) is not imported: whether output includes +reasoning is not established, so Muse sessions are reported as +unmeasured rather than zero. Prompts, results, reasons, hooks and +memory text are never stored. +""" + +from __future__ import annotations + +import glob +import json +import os +import re +import sqlite3 + +from .. import db +from ..identity import SessionIdentity +from ..ingest import JsonlSource, MissingNativeId, fingerprint, insert_event + +HARNESS = "muse" + +DEFAULT_ROOT = os.path.join(os.path.expanduser("~"), ".local", "share", + "muse", "sessions") + +CAPABILITIES = [ + ("model_usage", False, "model_completed usage is not imported; output/reasoning counter semantics are unverified, so sessions read unmeasured"), + ("tool_calls", True, "assistant_tool_calls_committed, one event per call_id"), + ("tool_results", True, "terminal effect status and committed result UTF-8 size, joined on call_id"), + ("mcp_identity", True, "tool_source from the session's MCP tool identity catalog and active toolset; never from a name prefix"), + ("read_evidence", True, "read_file calls as read attempts"), + ("skill_invocation", True, "read_skill calls"), + ("skill_file_reads", False, "skill file paths are not recorded"), + ("file_change", False, "edit and write calls stay tool calls; no file_change events"), + ("compaction", False, "no compaction record observed"), + ("lifecycle_task", True, "run start and terminal"), + ("subagents", True, "memory reminder child links; child logs under subagent/"), + ("human_input", False, "prompts are not imported"), + ("instruction_identity", True, "AgentsMD version from read_file paths only"), +] + +_SESSION_ID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") +_TARGET_ARGS = {"read_file": "path", "edit_file": "path", "write_file": "path", + "bash": "command"} +_RUN_KINDS = frozenset({ + "started", "terminal", "model_request_configured", + "assistant_tool_calls_committed", "tool_result_batch_committed", + "memory_reminder_child_session_linked"}) +_STATUS = {"completed": "ok", "failed": "error"} + + +def _session_id(value) -> str | None: + if isinstance(value, str) and _SESSION_ID_RE.fullmatch(value): + return value + return None + + +def _ts(value) -> float | None: + """Muse ``recorded_at`` is integer microseconds since the epoch.""" + if type(value) is int and value > 0: + return value / 1_000_000 + return None + + +def _str(value) -> str | None: + return value if isinstance(value, str) and value else None + + +def discover(root: str | None = None) -> list[str]: + root = root or DEFAULT_ROOT + main = glob.glob(os.path.join(root, "*", "*", "*", "*", "session.jsonl")) + children = glob.glob(os.path.join(root, "*", "*", "*", "*", "subagent", + "*", "session.jsonl")) + return sorted(main + children) + + +def _layout_parent(path: str) -> tuple[str | None, str | None]: + """(child id, parent id) when the file sits at /subagent//.""" + child_dir = os.path.dirname(os.path.abspath(path)) + sub_dir = os.path.dirname(child_dir) + if os.path.basename(sub_dir) != "subagent": + return None, None + child = _session_id(os.path.basename(child_dir)) + parent = _session_id(os.path.basename(os.path.dirname(sub_dir))) + if child is None or parent is None: + return None, None + return child, parent + + +class _State: + """Everything one complete parse of a session file establishes.""" + + def __init__(self): + self.session = None + self.catalog: dict[str, set] = {} + self.catalog_seen = False + self.active: set = set() + self.toolset_seen = False + self.calls: dict[str, dict] = {} + self.terminals: dict[str, dict] = {} + self.results: dict[str, dict] = {} + self.runs: dict[str, dict] = {} + self.links: dict[str, dict] = {} + self.project_dir = None + self.branch = None + self.model = None + self.started_at = None + self.ended_at = None + + def identity(self, name: str) -> dict: + entries = self.catalog.get(name) + if entries: + if len(entries) == 1: + server, tool = next(iter(entries)) + return {"tool_source": "mcp", "mcp_server": server, "mcp_tool": tool} + return {"tool_source": "unknown"} + if self.catalog_seen and self.toolset_seen and name in self.active: + return {"tool_source": "builtin"} + return {"tool_source": "unknown"} + + +def _parse(src: JsonlSource, totals: dict) -> _State: + state = _State() + + def bad(ordinal, category, raw): + src.error(ordinal, category, raw) + totals["malformed"] += 1 + + for ordinal, obj, raw in src.records(): + if obj is None: + bad(ordinal, "malformed_json", raw) + continue + if not isinstance(obj, dict) or obj.get("record_type") != "event": + continue # The retained envelope and non-event lines carry nothing consumed. + ptype = obj.get("payload_type") + payload = obj.get("payload") if isinstance(obj.get("payload"), dict) else {} + event = payload.get("event") if isinstance(payload.get("event"), dict) else {} + run_kind = ptype == "runtime.session" and payload.get("kind") == "run" \ + and event.get("kind") in _RUN_KINDS + consumed = run_kind or ptype in ( + "runtime.mcp_tool_identity_catalog", "runtime.session.metadata", + "session.workspace_branch.observed", "session.end", + "tool_batch.effect.terminal") + ts = _ts(obj.get("recorded_at")) + if ts is not None: + state.started_at = ts if state.started_at is None else min(state.started_at, ts) + state.ended_at = ts if state.ended_at is None else max(state.ended_at, ts) + if not consumed: + continue + if obj.get("schema_version") != 1 or obj.get("payload_schema_version") != 1: + bad(ordinal, "unsupported_schema", raw) + continue + stream = obj.get("stream") if isinstance(obj.get("stream"), dict) else {} + sid = _session_id(stream.get("id")) + if sid is None: + bad(ordinal, "missing_id", raw) + continue + if state.session is None: + state.session = sid + elif sid != state.session: + bad(ordinal, "identity_conflict", raw) + continue + seq = obj.get("sequence") if type(obj.get("sequence")) is int else None + at = {"seq": seq, "ts": ts} + record = payload.get("record") if isinstance(payload.get("record"), dict) else {} + if ptype == "runtime.mcp_tool_identity_catalog": + entries = payload.get("entries") + if not isinstance(entries, list): + bad(ordinal, "schema_error", raw) + continue + state.catalog_seen = True + for entry in entries: + if not isinstance(entry, dict): + continue + name, server, tool = (_str(entry.get("canonical_id")), + _str(entry.get("server_name")), + _str(entry.get("raw_tool_name"))) + if name and server and tool: + state.catalog.setdefault(name, set()).add((server, tool)) + elif ptype == "runtime.session.metadata": + state.project_dir = state.project_dir or _str(record.get("workspace_root")) + state.model = state.model or _str(record.get("model_id")) + elif ptype == "session.workspace_branch.observed": + ref = record.get("reference") if isinstance(record.get("reference"), dict) else {} + if ref.get("kind") == "branch" and _str(ref.get("name")): + state.branch = ref["name"] + elif ptype == "session.end": + pass # recorded_at already extends ended_at. + elif ptype == "tool_batch.effect.terminal": + call = _str(record.get("call_id")) + outcome = record.get("outcome") if isinstance(record.get("outcome"), dict) else {} + if call is None: + bad(ordinal, "missing_id", raw) + continue + state.terminals[call] = {**at, "status": _STATUS.get(outcome.get("kind"))} + else: + run_id = _str(payload.get("run_id")) + kind = event["kind"] + if run_id is None: + bad(ordinal, "missing_id", raw) + continue + at["run"] = run_id + if kind == "started": + frame = _str(obj.get("id")) + if frame is None: + bad(ordinal, "missing_id", raw) + continue + state.runs.setdefault(run_id, {})["start"] = {**at, "frame": frame} + elif kind == "terminal": + frame = _str(obj.get("id")) + if frame is None: + bad(ordinal, "missing_id", raw) + continue + state.runs.setdefault(run_id, {})["end"] = { + **at, "frame": frame, "terminal": _str(event.get("terminal"))} + elif kind == "model_request_configured": + toolset = event.get("toolset") if isinstance(event.get("toolset"), dict) else {} + active = toolset.get("active_tools") + if isinstance(active, list): + state.toolset_seen = True + state.active |= {t for t in active if isinstance(t, str)} + elif kind == "assistant_tool_calls_committed": + items = event.get("tool_calls") + if not isinstance(items, list): + bad(ordinal, "schema_error", raw) + continue + for item in items: + call = _str(item.get("call_id")) if isinstance(item, dict) else None + name = _str(item.get("name")) if isinstance(item, dict) else None + if call is None or name is None: + bad(ordinal, "missing_id", raw) + continue + args = None + if isinstance(item.get("args"), str): + try: + decoded = json.loads(item["args"]) + except ValueError: + decoded = None + args = decoded if isinstance(decoded, dict) else None + state.calls.setdefault(call, {**at, "name": name, "args": args}) + elif kind == "tool_result_batch_committed": + items = event.get("results") + if not isinstance(items, list): + bad(ordinal, "schema_error", raw) + continue + for item in items: + call = _str(item.get("tool_call_id")) if isinstance(item, dict) else None + if call is None: + bad(ordinal, "missing_id", raw) + continue + text = item.get("text") + state.results[call] = { + **at, "size": len(text.encode("utf-8")) + if isinstance(text, str) else None} + elif kind == "memory_reminder_child_session_linked": + child = _session_id(event.get("child_session_id")) + if child is None: + bad(ordinal, "missing_id", raw) + continue + state.links.setdefault(child, at) + return state + + +def _write(con: sqlite3.Connection, src: JsonlSource, state: _State, + stats: dict, parent: str | None) -> None: + key = f"{HARNESS}:{state.session}" + identity = SessionIdentity() + + def event(family, native_id, at, **kw): + try: + insert_event(con, stats, source_id=src.source_id, session_key=key, + family=family, native_id=native_id, ordinal=at["seq"], + ts=at["ts"], turn_id=f"{HARNESS}:{at['run']}" + if at.get("run") else None, reconcile=True, **kw) + except MissingNativeId: + src.error(None, "missing_id") + + for call, info in state.calls.items(): + name, args = info["name"], info["args"] or {} + detail = state.identity(name) + arg_key = _TARGET_ARGS.get(name) + target = args.get(arg_key) if arg_key else None + target = target if isinstance(target, str) else None + event("tool_call", call, info, name=name, target=target, detail=detail, + fingerprint=fingerprint(json.dumps(info["args"], sort_keys=True)) + if info["args"] is not None else None) + terminal, result = state.terminals.get(call), state.results.get(call) + status = terminal["status"] if terminal else None + if terminal or result: + first = min((x for x in (terminal, result) if x), + key=lambda x: (x["seq"] is None, x["seq"] or 0)) + event("tool_result", call, {**first, "run": info.get("run")}, + name=name, status=status, detail=detail, + size_bytes=result["size"] if result else None) + if name == "read_file" and target: + identity.observe_path(target) + event("read", call, info, name=name, target=target, status=status) + if name == "read_skill": + event("skill_invoke", call, info, name=name, target=args.get("name")) + for run_id, run in state.runs.items(): + start, end = run.get("start"), run.get("end") + if start: + event("lifecycle", start["frame"], start, name="turn_started") + if end: + event("lifecycle", end["frame"], end, name="turn_ended", + status=end["terminal"]) + con.execute( + "INSERT INTO turns(turn_id, source_id, session_key, session_id," + " model_observed, started_at, completed_at, duration_ms, state)" + " VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(turn_id) DO UPDATE SET" + " model_observed=excluded.model_observed," + " started_at=excluded.started_at, completed_at=excluded.completed_at," + " duration_ms=excluded.duration_ms, state=excluded.state", + (f"{HARNESS}:{run_id}", src.source_id, key, state.session, state.model, + start["ts"] if start else None, end["ts"] if end else None, + int((end["ts"] - start["ts"]) * 1000) + if start and end and start["ts"] is not None and end["ts"] is not None else None, + (end["terminal"] if end else "active"))) + for child, at in state.links.items(): + event("lifecycle", child, at, name="subagent_spawned") + db.upsert_session(con, f"{HARNESS}:{child}", HARNESS, child, None, + parent_session_key=key, role="subagent") + fields = dict(project_dir=state.project_dir, git_branch=state.branch, + started_at=state.started_at, ended_at=state.ended_at, + **identity.fields(con)) + if parent: + fields.update(parent_session_key=f"{HARNESS}:{parent}", role="subagent") + db.upsert_session(con, key, HARNESS, state.session, src.source_id, **fields) + + +def import_muse_file(con: sqlite3.Connection, path: str, full: bool = False) -> dict: + stats = {"events_inserted": 0, "malformed": 0} + src = JsonlSource(con, HARNESS, path, full=full) + if src.unchanged: + return {"unchanged": True} + # A grown file still needs earlier lines: a result or catalog can follow + # the call it completes across an append boundary. + src.start_offset = 0 + state = _parse(src, stats) + if state.session is None: + src.finish() + return stats + layout_child, parent = _layout_parent(path) + if layout_child is not None and layout_child != state.session: + src.error(None, "identity_conflict") + stats["malformed"] += 1 + parent = None + _write(con, src, state, stats, parent) + src.finish(session_id=state.session) + return stats + + +def sync(con: sqlite3.Connection, root: str | None = None, full: bool = False, + source: str | None = None) -> dict: + paths = [source] if source else discover(root) + totals = {"harness": HARNESS, "sources": 0, "unchanged": 0, + "responses_inserted": 0, "events_inserted": 0, + "events_updated": 0, "user_messages_inserted": 0, + "malformed": 0, "failed": []} + for path in paths: + try: + stats = import_muse_file(con, path, full=full) + con.commit() + except (OSError, sqlite3.DatabaseError, UnicodeDecodeError) as exc: + con.rollback() + totals["failed"].append({"path": path, "error": str(exc)}) + continue + totals["sources"] += 1 + totals["unchanged"] += 1 if stats.get("unchanged") else 0 + for key in ("events_inserted", "events_updated", "malformed"): + totals[key] += stats.get(key, 0) + return totals diff --git a/agent_observer/adapters/t3.py b/agent_observer/adapters/t3.py index ce09a6a..2ed9efc 100644 --- a/agent_observer/adapters/t3.py +++ b/agent_observer/adapters/t3.py @@ -89,7 +89,7 @@ V2_RUNS = "orchestration_v2_projection_runs" V2_LEGACY_IMPORTS = "orchestration_v2_legacy_imports" # V2 drivers whose nativeThreadRef is the ledger's native session id. -DRIVER_HARNESSES = {"claudeAgent": "claude", "codex": "codex", "grok": "grok", +DRIVER_HARNESSES = {"claudeAgent": "claude", "codex": "codex", "grok": "grok", "muse": "muse", "opencode": "opencode", "opencode2": "opencode"} V2_DRIVERS = tuple(DRIVER_HARNESSES) V2_PROVIDER_TURNS = "orchestration_v2_projection_provider_turns" @@ -1164,7 +1164,7 @@ def _tree_sessions(con: sqlite3.Connection, threads: dict, native_id == native_session or native_id.startswith(native_session + ":")): keys.add(sess_row["session_key"]) - elif harness in ("codex", "opencode", "grok") \ + elif harness in ("codex", "opencode", "grok", "muse") \ and native_id == native_session: keys.add(sess_row["session_key"]) keys.update(claude_turn_sessions(con, thread_id)) diff --git a/agent_observer/cli.py b/agent_observer/cli.py index 1aa43ee..8339fbf 100644 --- a/agent_observer/cli.py +++ b/agent_observer/cli.py @@ -61,12 +61,17 @@ def _fmt_section_total(section: dict) -> str: Sections over mixed counter semantics carry no top-level raw total; the text view names that instead of inventing a sum. """ + unmeasured = section.get("unmeasured_sessions") or [] + suffix = f"; {len(unmeasured)} unmeasured sessions" if unmeasured else "" if "total_tokens" in section: - return _fmt_tokens(section.get("total_tokens"), - _bucket_unknown(section)) + # A session-level lower bound (unmeasured sessions beside measured + # responses) renders like a response-level one. + lower = _bucket_unknown(section) or ( + 1 if section.get("usage_status") == "lower_bound" else 0) + return _fmt_tokens(section.get("total_tokens"), lower) + suffix if "by_semantics" in section: - return "mixed semantics (see --json)" - return "unknown" + return "mixed semantics (see --json)" + suffix + return "unknown" + suffix def _fmt_session_total(session: dict) -> str: @@ -184,7 +189,9 @@ def _text(payload) -> str: for phase in r.get("phases", []): lines.append(f" phase {phase['phase']}: responses={phase['usage']['responses']} " f"total={_fmt_section_total(phase['usage'])}; " - f"activity={phase['activity']}") + f"activity={phase['activity']}" + + (f" coverage={phase['activity_coverage']}" + if phase.get('activity_coverage') else "")) for model in phase['models']: g=model['generation'] lines.append(f" {model['harness']}/{model['model']}: " @@ -1409,6 +1416,8 @@ def _sessions(con, ns) -> int: r.pop("identity_json", None) usage = _report.scope_totals(con, {r["session_key"]}) r["responses"] = usage.get("responses", 0) + if usage.get("usage_status"): + r["usage_status"] = usage["usage_status"] if "total_tokens" in usage: r["total_tokens"] = usage.get("total_tokens") r["unknown_tokens"] = (usage.get("unknown_counts") or {}).get( diff --git a/agent_observer/db.py b/agent_observer/db.py index f81a9f0..7aea5b5 100644 --- a/agent_observer/db.py +++ b/agent_observer/db.py @@ -401,7 +401,7 @@ class LedgerBusy(Exception): "unknown", ) -HARNESSES = ("t3", "codex", "claude", "opencode", "grok", "router") +HARNESSES = ("t3", "codex", "claude", "opencode", "grok", "muse", "router") def default_path() -> str: diff --git a/agent_observer/ingest.py b/agent_observer/ingest.py index 6c5e0c3..a6eb05d 100644 --- a/agent_observer/ingest.py +++ b/agent_observer/ingest.py @@ -312,7 +312,8 @@ def insert_event(con: sqlite3.Connection, stats: dict, *, source_id: int, ts=None, turn_id=None, name=None, target=None, status=None, duration_ms=None, size_bytes=None, truncated=None, fingerprint=None, detail=None, update: bool = False, - migrate_ordinal: bool = False) -> None: + migrate_ordinal: bool = False, + reconcile: bool = False) -> None: """Insert one event under its natural key; a repeat is a no-op. Every protected field passes through agent_observer/privacy.py rule 6: @@ -326,6 +327,12 @@ def insert_event(con: sqlite3.Connection, stats: dict, *, source_id: int, migrate_ordinal. Codex does that for the first sight of an event this source already stored. The native id and source are not rewritten here. + + When reconcile is true (an adapter that rebuilds each event from a + whole changed source, such as Muse), an existing row takes every given + evidence value (ordinal, time, turn, name, target, status, duration, + size, truncation, fingerprint, detail), so evidence that arrived later + fills the same row. It counts as updated only when a value changed. """ native = privacy.filter_native_id(family, native_id) if not native: @@ -360,7 +367,24 @@ def insert_event(con: sqlite3.Connection, stats: dict, *, source_id: int, safe_name, safe_target, safe_status, duration_ms, size_bytes, truncated, fingerprint, payload)) if cur.rowcount == 0: - if update: + if reconcile: + columns = ("ordinal_num", "ts", "turn_id", "name", "target", "status", + "duration_ms", "size_bytes", "truncated", "fingerprint", + "detail_json") + wanted = (ordinal, ts, turn_id, safe_name, safe_target, safe_status, + duration_ms, size_bytes, truncated, fingerprint, payload) + row = con.execute( + f"SELECT {', '.join(columns)} FROM events WHERE session_key=?" + " AND family=? AND native_id=?", (session_key, family, native)).fetchone() + if tuple(row) != wanted: + con.execute( + f"UPDATE events SET {', '.join(c + '=?' for c in columns)}" + " WHERE session_key=? AND family=? AND native_id=?", + wanted + (session_key, family, native)) + stats["events_updated"] = stats.get("events_updated", 0) + 1 + else: + stats["events_duplicate"] = stats.get("events_duplicate", 0) + 1 + elif update: fields = "name=?, target=?, status=?, detail_json=?, duration_ms=?" values = [safe_name, safe_target, safe_status, payload, duration_ms] if migrate_ordinal: diff --git a/agent_observer/lookup.py b/agent_observer/lookup.py index 671c961..9dc48e1 100644 --- a/agent_observer/lookup.py +++ b/agent_observer/lookup.py @@ -531,7 +531,10 @@ def _backing_without_events(con, event_rows, *, session_key, session_keys, """Sources that back a session but have no event rows. A response-only session still has a native source. Event counts miss it, - and a default of available would hide that the source is gone. + and a default of available would hide that the source is gone. A linked + child session (parent_session_key set) with no backing source at all + yields one row with a NULL source, which reads as no_source: its log was + never read, and a parent's readable log is no evidence for it (#127). """ selects = [] args: list = [] @@ -569,6 +572,10 @@ def _backing_without_events(con, event_rows, *, session_key, session_keys, " AND src.session_id = sess.native_id" f" WHERE src.id IS NOT NULL{clause}") args.extend(key_args) + selects.append( + "SELECT sess.session_key, NULL FROM sessions sess" + f" WHERE sess.parent_session_key IS NOT NULL{clause}") + args.extend(key_args) harness_sql = "" if harness is not None: harness_sql = " WHERE harness=?" @@ -585,8 +592,11 @@ def _backing_without_events(con, event_rows, *, session_key, session_keys, " LEFT JOIN sessions sess ON sess.session_key = b.session_key" f") backed{harness_sql}", args).fetchall() seen = {(row["session_key"], row["source_id"]) for row in event_rows} + backed = {row["session_key"] for row in event_rows} | { + row["session_key"] for row in found if row["source_id"] is not None} return [row for row in found - if (row["session_key"], row["source_id"]) not in seen] + if (row["session_key"], row["source_id"]) not in seen + and (row["source_id"] is not None or row["session_key"] not in backed)] def availability(con: sqlite3.Connection, session_key: str | None = None, diff --git a/agent_observer/privacy.py b/agent_observer/privacy.py index 0abc4f7..a4afe9a 100644 --- a/agent_observer/privacy.py +++ b/agent_observer/privacy.py @@ -182,6 +182,8 @@ def line_excerpt(line: object) -> str: # - skill_invoke.skill_path: installed SKILL.md path when the native # invocation supplies a directory, otherwise omitted # - assistant_message.excerpt: analysis._permission_seeking +# - tool_call/tool_result.tool_source/mcp_server/mcp_tool: report activity +# MCP counts (closed tool_source set; server and tool are identifiers) # Every other family keeps no detail. Values may be numbers, booleans, # fixed-length hex hashes, native identifiers of the expected type, file # paths, shell commands, or closed status/kind strings; anything else is @@ -191,12 +193,15 @@ def line_excerpt(line: object) -> str: "file_change": {"paths": "paths"}, "tool_result": {"exit_code": "int", "exitCode": "int", "is_timeout": "bool", "auth_retry_attempted": "bool", - "reconnect_attempted": "bool"}, + "reconnect_attempted": "bool", + "tool_source": "tool_source", "mcp_server": "identifier", + "mcp_tool": "identifier"}, "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": {"timeout_sec": "int"}, + "tool_call": {"timeout_sec": "int", "tool_source": "tool_source", + "mcp_server": "identifier", "mcp_tool": "identifier"}, "skill_invoke": {"skill_path": "path"}, "compaction": {}, "lifecycle": {}, @@ -205,6 +210,10 @@ def line_excerpt(line: object) -> str: _TARGET_CHARS = 500 +# tool_call and tool_result detail tool_source: closed set. mcp only from +# a harness's MCP identity record, never from a name prefix (#127). +TOOL_SOURCES = frozenset({"mcp", "builtin", "unknown"}) + # Rule 8: one target argument (a path or a full shell command, including a # Codex code-mode exec program) keeps up to 64 KiB. The longest real values # measured on 2026-09-29 over 14 days of local sessions were 42,102 chars @@ -621,6 +630,9 @@ def filter_detail(family: str, detail: object) -> dict: elif kind == "int": if _valid_int(value): out[key] = value + elif kind == "tool_source": + if isinstance(value, str) and value in TOOL_SOURCES: + out[key] = value elif kind == "identifier": kept = _valid_identifier(value) if kept is not None: diff --git a/agent_observer/publish.py b/agent_observer/publish.py index d7769fd..9830c8d 100644 --- a/agent_observer/publish.py +++ b/agent_observer/publish.py @@ -120,6 +120,15 @@ def _fmt_tokens(value, unknown: int = 0) -> str: return text +def _activity_cell(value, coverage: str | None) -> str: + """An activity count that never turns missing support into zero.""" + if value is None or coverage == "unavailable": + return "unavailable" + if coverage == "partial": + return f">={int(value):,} (partial)" + return str(value) + + def summarize(con, session_keys: set, label: str, task_id: str | None = None, schedule: dict | None = None) -> dict: rep = report.task_report(con, task_id, schedule=schedule) if task_id else None @@ -190,6 +199,7 @@ def summarize(con, session_keys: set, label: str, task_id: str | None = None, "scope": rep["scope"], "phases": rep["phases"], "activity_session_context": rep["activity_session_context"], + "activity_session_context_coverage": rep["activity_session_context_coverage"], "outcome": rep["outcome"], "acceptance_state": rep["acceptance_state"], "task_turns": rep["task_turns"], @@ -555,13 +565,21 @@ def render(summary: dict, target: str | None = None) -> str: cost = _fmt_cost(c.get('estimated_cost_usd_partial') if c.get('priced_responses') else None) lines.append(f"| {_safe_field(phase['phase'])} | {_model_cell(m)} " f"| {m['responses']} | {buckets} | {cost} |") + unmeasured = phase["usage"].get("unmeasured_sessions") or [] + if unmeasured: + # Native usage not imported: unknown, never a zero row. + buckets = " | ".join("unknown" for _ in report.BUCKETS) + lines.append(f"| {_safe_field(phase['phase'])} | unmeasured sessions " + f"({len(unmeasured)}) | 0 | {buckets} | unknown |") lines += ["", "Recorded activity by phase (observations, not separate token bills):", "", - "| Phase | Tool calls | MCP results | Reads | File changes | Failed tool results |", - "| --- | ---: | ---: | ---: | ---: | ---: |"] + "| Phase | Tool calls | MCP results | Reads | Skill invocations | File changes | Failed tool results |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: |"] for phase in summary['phases']: a = phase['activity'] - counts = " | ".join(str(a.get(k, 0)) for k in - ('tool_calls', 'mcp_results', 'reads', 'file_changes', 'failed_tool_results')) + coverage = phase.get('activity_coverage') or {} + counts = " | ".join(_activity_cell(a.get(k, 0), coverage.get(k)) for k in + ('tool_calls', 'mcp_results', 'reads', 'skill_invocations', + 'file_changes', 'failed_tool_results')) lines.append(f"| {_safe_field(phase['phase'])} | {counts} |") lines += _price_evidence_lines(summary) lines += ["", "Local measurement from native records; usage totals are not billing. " diff --git a/agent_observer/report.py b/agent_observer/report.py index 2674082..50a9c8f 100644 --- a/agent_observer/report.py +++ b/agent_observer/report.py @@ -96,6 +96,62 @@ def _usage_totals(rows, include_responses: bool = True) -> dict: return _bucket_totals(rows, include_responses=include_responses) +def _capabilities() -> dict: + """harness -> {capability family: declared support}.""" + from .adapters import capabilities as adapter_capabilities + out: dict = {} + for row in adapter_capabilities(): + out.setdefault(row["harness"], {})[row["family"]] = row["supported"] + return out + + +def _session_harness(con: sqlite3.Connection, key: str) -> str | None: + row = con.execute("SELECT harness FROM sessions WHERE session_key=?", + (key,)).fetchone() + return row["harness"] if row is not None else None + + +def unmeasured_sessions(con: sqlite3.Connection, session_keys) -> set: + """Sessions whose usage is not imported at all (#127). + + Only a session whose native harness declares model_usage=False and + that holds no responses: its consumption is unknown, not zero. A + response-less session of a harness that imports usage keeps the + existing exact rendering. + """ + caps = _capabilities() + out = set() + for key in sorted(session_keys or ()): + harness = _session_harness(con, key) + if harness is None or caps.get(harness, {}).get("model_usage") is not False: + continue + if con.execute("SELECT 1 FROM responses WHERE session_key=? LIMIT 1", + (key,)).fetchone() is None: + out.add(key) + return out + + +def _mark_unmeasured(totals: dict, sessions) -> dict: + """Mark a usage total that misses unmeasured sessions' consumption. + + With no responses every raw bucket is unknown (None); with responses + the numbers are measured subtotals, a lower bound. Mixed semantics keep + by_semantics only. Response counts and unknown_counts keep their + response-level meaning. + """ + if not sessions: + return totals + totals["unmeasured_sessions"] = sorted(sessions) + if not totals.get("responses") and "by_semantics" not in totals: + for bucket in BUCKETS: + if bucket in totals: + totals[bucket] = None + totals["usage_status"] = "unknown" + else: + totals["usage_status"] = "lower_bound" + return totals + + def model_usage(rows) -> list[dict]: """Model buckets over an already selected response scope. @@ -141,6 +197,8 @@ def scope_totals(con: sqlite3.Connection, session_keys=None) -> dict: totals = _usage_totals(live) | { "overlap_responses": sum(1 for r in rows if r["is_overlap"]), } + if session_keys is not None: + _mark_unmeasured(totals, unmeasured_sessions(con, session_keys)) return totals @@ -305,7 +363,42 @@ def _native_cost(rows: list) -> dict: "estimates and subscription billing."} -def _phase_report(con, task_id, rows, whole, attempts, schedule): +# Activity counter -> the capability family that must support it (#127). +ACTIVITY_CAPABILITIES = { + "tool_calls": "tool_calls", "mcp_results": "tool_results", + "reads": "read_evidence", "file_changes": "file_change", + "failed_tool_results": "tool_results", "skill_invocations": "skill_invocation", +} + + +def _activity_coverage(cell: dict, harnesses: list, caps: dict) -> dict: + """Per counter: unavailable when no contributing session supports it, + partial when only some do. Unavailable counters become None.""" + coverage = {} + for counter, family in ACTIVITY_CAPABILITIES.items(): + supported = [caps.get(h, {}).get(family) is not False for h in harnesses] + if supported and not any(supported): + cell[counter] = None + coverage[counter] = "unavailable" + elif not all(supported): + coverage[counter] = "partial" + return coverage + + +def _is_mcp_result(event) -> bool: + """An MCP result by recorded identity; the mcp. name rule only for + records that carry no tool_source (Codex).""" + try: + detail = json.loads(event["detail_json"] or "{}") + except (TypeError, ValueError): + detail = {} + if isinstance(detail, dict) and "tool_source" in detail: + return detail["tool_source"] == "mcp" + return (event["name"] or "").startswith("mcp.") + + +def _phase_report(con, task_id, rows, whole, attempts, schedule, + unmeasured=frozenset()): from . import pricing, analysis by_turn, by_session = {}, {} @@ -337,27 +430,52 @@ def selected(row): for row in rows: groups.setdefault(selected(row), []).append(row) activities = {} + contributors: dict[str, dict[str, str | None]] = {} turns = {r["turn_id"] for r in rows if r["turn_id"]} - for key in sorted({r["session_key"] for r in rows}): + # Exclusively owned whole sessions count even without responses (#127). + for key in sorted({r["session_key"] for r in rows} | set(whole)): + harness = _session_harness(con, key) for event in con.execute("SELECT * FROM events WHERE session_key=?", (key,)): label = ("session context" if key not in whole and event["turn_id"] not in turns else selected(event)) + # Capability comes from the event's own native session, never + # from the importer that attributed it (Router). + contributors.setdefault(label, {})[key] = harness cell = activities.setdefault(label, {"tool_calls": 0, "mcp_results": 0, - "reads": 0, "file_changes": 0, "failed_tool_results": 0}) + "reads": 0, "file_changes": 0, "failed_tool_results": 0, + "skill_invocations": 0}) if event["family"] == "tool_call": cell["tool_calls"] += 1 if event["family"] == "tool_result": - if (event["name"] or "").startswith("mcp."): cell["mcp_results"] += 1 + if _is_mcp_result(event): cell["mcp_results"] += 1 if analysis._failed(event): cell["failed_tool_results"] += 1 if event["family"] == "read": cell["reads"] += 1 if event["family"] == "file_change": cell["file_changes"] += 1 + if event["family"] == "skill_invoke": cell["skill_invocations"] += 1 + caps = _capabilities() + coverage = {label: _activity_coverage(cell, list(contributors[label].values()), caps) + for label, cell in activities.items()} context = activities.pop("session context", {}) - phases = [{"phase": name, "usage": _usage_totals(groups.get(name, [])), - "models": model_usage(groups.get(name, [])), - "estimated_cost": pricing.price_scope(groups.get(name, []), schedule), - "activity": activities.get(name, {}), - "activity_note": "Recorded observations, not separately billed token costs; coverage varies by harness."} - for name in sorted(set(groups) | set(activities))] - return phases, context + context_coverage = coverage.pop("session context", {}) + + def phase_entry(name): + group = groups.get(name, []) + missing = sorted(k for k in contributors.get(name, {}) if k in unmeasured) + usage = _mark_unmeasured(_usage_totals(group), missing) + estimate = pricing.price_scope(group, schedule) + if missing: + # Same rule as the task-level gap: no complete total, any priced + # subtotal kept. + estimate.update(complete=False, estimated_cost_usd_total=None, + status="partial" if estimate.get("priced_responses") else "unknown", + coverage_note="Unmeasured sessions: native usage is not imported") + return {"phase": name, "usage": usage, "models": model_usage(group), + "estimated_cost": estimate, + "activity": activities.get(name, {}), + "activity_coverage": coverage.get(name, {}), + "activity_note": "Recorded observations, not separately billed token costs; coverage varies by harness."} + + phases = [phase_entry(name) for name in sorted(set(groups) | set(activities))] + return phases, context, context_coverage def task_report(con: sqlite3.Connection, task_id: str, @@ -501,16 +619,41 @@ def child_session(identity): empty_sessions = sorted(key for key in scope_keys if key not in missing_sessions and not con.execute("SELECT 1 FROM responses WHERE session_key=?", (key,)).fetchone()) + # Children linked to an owned session but not owned by this task are a + # gap; the link never assigns ownership (#127). + owned_scope = sorted(owned_sessions & scope_keys) + linked_children = {r["session_key"] for r in con.execute( + "SELECT session_key FROM sessions WHERE parent_session_key IN" + f" ({','.join('?' * len(owned_scope))})", owned_scope)} if owned_scope else set() unbound_children = sorted({a["session_key"] for a in task_turn_rows if a.get("harness") == "router" and a.get("session_key") - and a["session_key"] not in whole}) + and a["session_key"] not in whole} + | (linked_children - owned_sessions)) + # Unmeasured sessions join the partition their responses would take. + unmeasured = unmeasured_sessions(con, scope_keys) + prompt_sessions: dict[str, set] = {} + for row in con.execute( + "SELECT s.session_key, a.user_message_id FROM assignments a JOIN user_messages s" + " ON s.native_id=a.user_message_id WHERE a.task_id=?", (task_id,)): + prompt_sessions.setdefault(row["session_key"], set()).add(row["user_message_id"]) + shared_owned = (set(whole) & whole_shared) | whole_conflicts + unmeasured_attributed, unmeasured_shared = set(), set() + for key in unmeasured: + subs = prompt_sessions.get(key, set()) + if key in shared_owned: + unmeasured_shared.add(key) + elif key in whole or any(s not in joint for s in subs): + unmeasured_attributed.add(key) + elif subs: + unmeasured_shared.add(key) unavailable_usage = bool(not scope_keys or missing_sessions or empty_sessions or unbound_children or unbound_delegations or whole_conflicts) cutoff = _source_cutoff(con, scope_keys) task_turns = _task_turn_ids(con, task_id, turn_sub, whole) diagnostics = _task_diagnostics(con, scope_keys, whole, task_turns) - phases, activity_context = _phase_report(con, task_id, attributed, set(whole)-whole_shared, - task_turn_rows, schedule) + phases, activity_context, activity_context_coverage = _phase_report( + con, task_id, attributed, set(whole)-whole_shared, task_turn_rows, schedule, + unmeasured_attributed) def _resp_tuple(r) -> list: return [r.get("response_id"), r.get("harness"), r.get("model"), @@ -532,6 +675,7 @@ def _resp_tuple(r) -> list: "price_schedule": schedule, "phases": phases, "activity_session_context": activity_context, + "activity_session_context_coverage": activity_context_coverage, "coverage_evidence": {"missing_submissions": sorted(missing_submissions), "conflicting_sessions": sorted(whole_conflicts), "joint": joint, "whole_shared": sorted(whole_shared), @@ -582,11 +726,12 @@ def _resp_tuple(r) -> list: active_task_turns = [a for a in task_turn_rows if a.get("state") == "active"] report = { "task": dict(task), - "attributed": total(attributed), + "attributed": _mark_unmeasured(total(attributed), unmeasured_attributed), "models": model_usage(attributed), "phases": phases, "activity_session_context": activity_context, - "shared_joint": total(shared), + "activity_session_context_coverage": activity_context_coverage, + "shared_joint": _mark_unmeasured(total(shared), unmeasured_shared), "shared_models": model_usage(shared), "unassigned_in_scope": total(unassigned), "unplaced_shared": total(unplaced), diff --git a/docs/contracts.md b/docs/contracts.md index 093cdfe..484f025 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -42,7 +42,7 @@ are committed. | `import_errors` | none | Quarantined records; prior valid data is kept. | Every natural key carries its harness prefix (`t3:`, `codex:`, `claude:`, -`opencode:`, `grok:`, `router:`) except event native ids, which are unique +`opencode:`, `grok:`, `muse:`, `router:`) except event native ids, which are unique within their session and family. Rows are keyed by session, not by source file, so re-reading a file, a grown log, or a copied file never adds usage twice. @@ -152,6 +152,7 @@ An adapter is `agent_observer/adapters/.py` with `HARNESS`, | Claude Code | `claude:input_excludes_cache,output_includes_thinking` | input + cache_creation + cache_read + output | | OpenCode | `opencode:input_excludes_cache,reasoning_separate` | input + output + reasoning + cache read + cache write | | Grok Build | `grok:` | as reported per prompt completion | +| Muse | none: usage is not imported (`model_usage` unsupported) | sessions read unmeasured, never zero | ### Event families @@ -171,6 +172,17 @@ An adapter is `agent_observer/adapters/.py` with `HARNESS`, A tool call joins its result only on an equal native call id. Nothing is guessed into a join. +`tool_call` and `tool_result` may carry detail `tool_source` (closed set +`mcp`, `builtin`, `unknown`) with `mcp_server` and `mcp_tool` identifiers +for `mcp`. Only a harness's own MCP identity record sets `mcp`; a name +prefix never does. Today only Muse writes these keys. + +`insert_event(..., reconcile=True)` is for an adapter that rebuilds every +event from a whole changed source (Muse): an existing row takes every given +evidence value, so evidence that arrived later fills the same row, and it +counts as updated only when a value changed. The default and `update` paths +are unchanged. + ## Instruction identity `identity.SessionIdentity` records, where evidence exists: @@ -439,7 +451,31 @@ An explicit `mixed` label is appropriate when a long turn spans phases. Shared and unassigned responses remain outside phase-attributed costs. Each phase exposes native buckets, per-model estimates and recorded tool calls, -MCP results, reads, file changes and failed tool results. These activity +MCP results, reads, skill invocations, file changes and failed tool results. +Phases include events of exclusively owned sessions that have no responses. +An MCP result is one whose detail `tool_source` is `mcp`; a result without +`tool_source` keeps the older `mcp.` name rule (Codex). Each counter depends +on one capability (`tool_calls`, `tool_results`, `read_evidence`, +`file_change`, `skill_invocation`) of each contributing session's own native +harness, never of the importer that attributed it (Router). When no +contributor supports it the counter is `null` and `activity_coverage` names it +`unavailable`; when only some do, the measured count stays and is `partial`. +The session-context cell reports the same under +`activity_session_context_coverage`. + +An unmeasured session exists in the ledger, has no responses, and its native +harness declares `model_usage` unsupported (today only Muse). It joins the +partition its responses would take (`attributed`, `shared_joint`, and always +`scope`) and every phase its events are labeled into. That total gains +`unmeasured_sessions` and `usage_status`: `unknown` with `null` raw buckets +when the partition has no responses, `lower_bound` when its numbers are a +measured subtotal (mixed semantics keep `by_semantics` only). The phase +estimate then has no complete total. A partition with no unmeasured session +is unchanged, including exact zeros. Task text prints `>=N (lower bound)` and +`; N unmeasured sessions`; publication adds an `unmeasured sessions (N)` row +with unknown buckets. Children whose `parent_session_key` is an owned session +but which the task does not own are `unbound_child_sessions`; a link never +assigns ownership. These activity observations can overlap and depend on harness coverage. They are not a split of the token bill: a later model input can include many earlier tool results. Reasoning and other generated output form a separate dimension inside every @@ -524,6 +560,41 @@ re-read every sync; workload bindings are durable ledger rows. sender's child records the same fields. `thread-misuse` reads these two tables only and follows the recorded parents for chains. +## Muse (`sync --harness muse`) + +Default root `~/.local/share/muse/sessions`: each +`YYYY/MM/DD//session.jsonl` and its internal child logs at +`/subagent//session.jsonl`, read-only. Frames carry +`payload_type`, `schema_version`, native `sequence`, frame `id`, +`recorded_at` (microseconds) and `stream.id` (the session id). + +- A changed file is parsed whole (complete lines only), then each event is + written with its final values (`reconcile=True`). No join state is kept. + Event `ordinal_num` is native `sequence`; import errors use line numbers. +- One native call is one `tool_call` and at most one `tool_result`, keyed by + `call_id`. The terminal `tool_batch` effect gives `ok`/`error`; the + committed result gives `size_bytes` as UTF-8 bytes. Either may be missing + and stays NULL. Effects never add calls. +- `read_file` also writes a `read` attempt with its path and the result + status; `read_skill` writes one `skill_invoke`. `skill_read_observed` + adds nothing. +- `tool_source` is a session-level rule: in the session's + `mcp_tool_identity_catalog` (one server and raw tool) is `mcp`; in the + active toolset and in no catalog is `builtin`; otherwise `unknown`. +- Run frames (`payload.kind=run`) only: `turn_started` and `turn_ended` keyed + by their frame ids, turn `muse:` (the id Chromeria's + `nativeTurnRef` carries). Internal `kind=task` frames are skipped. +- A child link writes `subagent_spawned` and the child session row + (`parent_session_key`, role `subagent`). A child log takes its parent only + from its own `/subagent//` location when the directory + matches its `stream.id`; a mismatch is an `identity_conflict`. Record-supplied + log paths are never opened or stored. A linked child with no backing source + reads `ledger_only` (`no_source`) in `sessions availability`. +- Consumed variants need `schema_version` and `payload_schema_version` 1, + else `unsupported_schema`; unconsumed kinds are skipped. Prompts, result + text, outcome reasons, hooks, memory and usage are never stored. +- T3 maps `driver: muse` strong `nativeThreadRef` to `muse:`. + ## Codex model context across imports Each source preserves the native model/effort context at its imported byte diff --git a/skills/agent-observer/SKILL.md b/skills/agent-observer/SKILL.md index 9aa9be8..a5404c9 100644 --- a/skills/agent-observer/SKILL.md +++ b/skills/agent-observer/SKILL.md @@ -5,8 +5,9 @@ description: Measure what agents actually did on this machine. Use when someone # Agent Observer -Agent Observer reads native Codex, Claude Code, OpenCode and Grok Build -records, Model Router's ledger and Chromeria/T3 Code state into one +Agent Observer reads native Codex, Claude Code, OpenCode, Grok Build and +Muse records (Muse: tool, read and skill events only; its token usage reads +unknown, never zero), Model Router's ledger and Chromeria/T3 Code state into one private local ledger. It never changes those records, never calls a model, and posts only its cost comment, on a PR being delivered or when asked. diff --git a/tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/session.jsonl b/tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/session.jsonl new file mode 100644 index 0000000..08a1b64 --- /dev/null +++ b/tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/session.jsonl @@ -0,0 +1,27 @@ +{"children": [], "content_sha256": "0000000000000000000000000000000000000000000000000000000000000000", "frame_schema_version": 1, "outer_log_ordinal": 0, "retained_frame": "synthetic", "transaction_id": null} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000001", "payload": {"background_tasks": "none", "command_id": "x", "kind": "session_started", "previous_session_stream": null}, "payload_schema_version": 1, "payload_type": "session.started", "record_type": "event", "recorded_at": 1792442401000000, "schema_version": 1, "sequence": 1, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000002", "payload": {"kind": "metadata", "record": {"model_id": "muse-synthetic-1", "provider_id": "synthetic", "tool_surface_version": "2", "workflow_authoring_profile": "default", "workspace_root": "/work/project"}}, "payload_schema_version": 1, "payload_type": "runtime.session.metadata", "record_type": "event", "recorded_at": 1792442402000000, "schema_version": 1, "sequence": 2, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000003", "payload": {"kind": "workspace_branch", "record": {"reference": {"kind": "branch", "name": "feature/synthetic"}}}, "payload_schema_version": 1, "payload_type": "session.workspace_branch.observed", "record_type": "event", "recorded_at": 1792442403000000, "schema_version": 1, "sequence": 3, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000004", "payload": {"event": {"kind": "started", "prompt": "synthetic prompt"}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442404000000, "schema_version": 1, "sequence": 4, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000005", "payload": {"event": {"kind": "model_request_configured", "toolset": {"active_tools": ["read_file", "read_skill", "bash", "edit_file", "write_file", "web_search", "mcp.lookalike", "mcp__t3_code__web_read"], "mode": "all", "source": "default"}}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442405000000, "schema_version": 1, "sequence": 5, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000006", "payload": {"event": {"kind": "started"}, "kind": "task", "run_id": "0a000000-0000-7000-8000-0000000000a1", "task_id": "0c000000-0000-7000-8000-000000000001"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442406000000, "schema_version": 1, "sequence": 6, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000007", "payload": {"event": {"kind": "assistant_tool_calls_committed", "tool_calls": [{"args": "{\"path\": \"/work/project/missing.md\"}", "call_id": "call_r1", "id": "fc_l_r1", "name": "read_file"}, {"args": "{\"url\": \"https://example.invalid/\", \"maxChars\": 10}", "call_id": "call_m1", "id": "fc_l_m1", "name": "mcp__t3_code__web_read"}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442407000000, "schema_version": 1, "sequence": 7, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000008", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_r1", "effect_id": "0dall_r1", "kind": "started", "model_call_index": 0, "parallel_profile": {"kind": "file_read"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}, "tool_name": "read_file"}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.started", "record_type": "event", "recorded_at": 1792442408000000, "schema_version": 1, "sequence": 8, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000009", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_r1", "effect_id": "0dall_r1", "kind": "terminal", "model_call_index": 0, "outcome": {"kind": "failed", "reason": "synthetic reason text"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.terminal", "record_type": "event", "recorded_at": 1792442409000000, "schema_version": 1, "sequence": 9, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000000a", "payload": {"entries": [{"canonical_id": "mcp__t3_code__web_read", "raw_tool_name": "web_read", "server_name": "t3-code", "surface": {"name": "web_read", "namespace": "mcp__t3_code"}}], "kind": "mcp_tool_identity_catalog", "schema_version": 1}, "payload_schema_version": 1, "payload_type": "runtime.mcp_tool_identity_catalog", "record_type": "event", "recorded_at": 1792442410000000, "schema_version": 1, "sequence": 10, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000000b", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_m1", "effect_id": "0dall_m1", "kind": "started", "model_call_index": 0, "parallel_profile": {"kind": "file_read"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}, "tool_name": "mcp__t3_code__web_read"}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.started", "record_type": "event", "recorded_at": 1792442411000000, "schema_version": 1, "sequence": 11, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000000c", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_m1", "effect_id": "0dall_m1", "kind": "terminal", "model_call_index": 0, "outcome": {"kind": "completed"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.terminal", "record_type": "event", "recorded_at": 1792442412000000, "schema_version": 1, "sequence": 12, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000000d", "payload": {"event": {"kind": "tool_result_batch_committed", "results": [{"text": "missing", "tool_call_id": "call_r1", "tool_call_index": 0}, {"text": "caf\u00e9 \u2014 r\u00e9sum\u00e9", "tool_call_id": "call_m1", "tool_call_index": 1}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442413000000, "schema_version": 1, "sequence": 13, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000000e", "payload": {"event": {"child_session_id": "0a000000-0000-7000-8000-0000000000c1", "child_session_log_path": "subagent/0a000000-0000-7000-8000-0000000000c1/session.jsonl", "kind": "memory_reminder_child_session_linked", "parent_session_id": "0a000000-0000-7000-8000-000000000001", "reminder_agent_id": "goal-reminder"}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442414000000, "schema_version": 1, "sequence": 14, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000000f", "payload": {"event": {"child_session_id": "0a000000-0000-7000-8000-0000000000c2", "kind": "memory_reminder_child_session_linked", "parent_session_id": "0a000000-0000-7000-8000-000000000001", "reminder_agent_id": "skill-reminder"}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442415000000, "schema_version": 1, "sequence": 15, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000010", "payload": {"event": {"kind": "assistant_tool_calls_committed", "tool_calls": [{"args": "{\"path\": \"/work/project/README.md\"}", "call_id": "call_r2", "id": "fc_l_r2", "name": "read_file"}, {"args": "{\"name\": \"plugin:agentsmd:operations\"}", "call_id": "call_s1", "id": "fc_l_s1", "name": "read_skill"}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442416000000, "schema_version": 1, "sequence": 16, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000011", "payload": {"event": {"kind": "tool_result_batch_committed", "results": [{"text": "\u00fcber", "tool_call_id": "call_r2", "tool_call_index": 0}, {"text": "skill body", "tool_call_id": "call_s1", "tool_call_index": 1}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442417000000, "schema_version": 1, "sequence": 17, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000012", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_r2", "effect_id": "0dall_r2", "kind": "terminal", "model_call_index": 0, "outcome": {"kind": "completed"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.terminal", "record_type": "event", "recorded_at": 1792442418000000, "schema_version": 1, "sequence": 18, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000013", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_s1", "effect_id": "0dall_s1", "kind": "terminal", "model_call_index": 0, "outcome": {"kind": "completed"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.terminal", "record_type": "event", "recorded_at": 1792442419000000, "schema_version": 1, "sequence": 19, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000014", "payload": {"event": {"kind": "model_completed", "model": "muse-synthetic-1", "response_id": "resp_1", "usage": {"input_tokens": 10, "output_tokens": 2}}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442420000000, "schema_version": 1, "sequence": 20, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000015", "payload": {"event": {"kind": "assistant_tool_calls_committed", "tool_calls": [{"args": "{\"q\": 1}", "call_id": "call_u1", "id": "fc_l_u1", "name": "mcp__ghost__tool"}, {"args": "{\"q\": 2}", "call_id": "call_b1", "id": "fc_l_b1", "name": "mcp.lookalike"}, {"args": "{\"command\": \"ls -la\"}", "call_id": "call_p1", "id": "fc_l_p1", "name": "bash"}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442421000000, "schema_version": 1, "sequence": 21, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000016", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_u1", "effect_id": "0dall_u1", "kind": "terminal", "model_call_index": 0, "outcome": {"kind": "completed"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.terminal", "record_type": "event", "recorded_at": 1792442422000000, "schema_version": 1, "sequence": 22, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000017", "payload": {"event": {"kind": "tool_result_batch_committed", "results": [{"text": "ok", "tool_call_id": "call_b1", "tool_call_index": 0}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442424000000, "schema_version": 1, "sequence": 24, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000018", "payload": {"kind": "tool_batch_effect", "record": {"call_id": "call_b1", "effect_id": "0dall_b1", "kind": "terminal", "model_call_index": 0, "outcome": {"kind": "completed"}, "task_id": "0c000000-0000-7000-8000-000000000002", "task_stream": {"id": "0c000000-0000-7000-8000-000000000002", "kind": "task"}}, "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "tool_batch.effect.terminal", "record_type": "event", "recorded_at": 1792442423000000, "schema_version": 1, "sequence": 23, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-000000000019", "payload": {"event": {"kind": "terminal", "reason": null, "terminal": "completed"}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000a1"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442425000000, "schema_version": 1, "sequence": 25, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000001a", "payload": {"kind": "session_end", "record": {"exit_reason": "clean", "session_id": "0a000000-0000-7000-8000-000000000001", "uptime_ms": 5}}, "payload_schema_version": 1, "payload_type": "session.end", "record_type": "event", "recorded_at": 1792442426000000, "schema_version": 1, "sequence": 26, "stream": {"id": "0a000000-0000-7000-8000-000000000001", "kind": "session"}} diff --git a/tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/subagent/0a000000-0000-7000-8000-0000000000c1/session.jsonl b/tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/subagent/0a000000-0000-7000-8000-0000000000c1/session.jsonl new file mode 100644 index 0000000..7a705c0 --- /dev/null +++ b/tests/fixtures/muse/sessions/2026/10/19/0a000000-0000-7000-8000-000000000001/subagent/0a000000-0000-7000-8000-0000000000c1/session.jsonl @@ -0,0 +1,5 @@ +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000001b", "payload": {"kind": "metadata", "record": {"model_id": "muse-synthetic-1", "provider_id": "synthetic", "workspace_root": "/work/project"}}, "payload_schema_version": 1, "payload_type": "runtime.session.metadata", "record_type": "event", "recorded_at": 1792442401000000, "schema_version": 1, "sequence": 1, "stream": {"id": "0a000000-0000-7000-8000-0000000000c1", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000001c", "payload": {"event": {"kind": "started"}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000c9"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442402000000, "schema_version": 1, "sequence": 2, "stream": {"id": "0a000000-0000-7000-8000-0000000000c1", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000001d", "payload": {"event": {"kind": "assistant_tool_calls_committed", "tool_calls": [{"args": "{\"decision\": \"none\"}", "call_id": "call_c1", "id": "fc_l_c1", "name": "submit_reminder_decision"}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000c9"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442403000000, "schema_version": 1, "sequence": 3, "stream": {"id": "0a000000-0000-7000-8000-0000000000c1", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000001e", "payload": {"event": {"kind": "tool_result_batch_committed", "results": [{"text": "{}", "tool_call_id": "call_c1", "tool_call_index": 0}]}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000c9"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442404000000, "schema_version": 1, "sequence": 4, "stream": {"id": "0a000000-0000-7000-8000-0000000000c1", "kind": "session"}} +{"causation_id": null, "durability": "durable", "id": "0b000000-0000-7000-8000-00000000001f", "payload": {"event": {"kind": "terminal", "terminal": "completed"}, "kind": "run", "run_id": "0a000000-0000-7000-8000-0000000000c9"}, "payload_schema_version": 1, "payload_type": "runtime.session", "record_type": "event", "recorded_at": 1792442405000000, "schema_version": 1, "sequence": 5, "stream": {"id": "0a000000-0000-7000-8000-0000000000c1", "kind": "session"}} diff --git a/tests/test_cli.py b/tests/test_cli.py index 44453cf..dea739f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -125,9 +125,9 @@ def test_runtime_never_touches_ccusage(self): f"ccusage import/call in runtime file {name}") r = run(self.db, "trace", "--capabilities", "--json") self.assertEqual(r.returncode, 0, r.stderr) - caps = {c["family"]: c["supported"] + caps = {(c["harness"], c["family"]): c["supported"] for c in json.loads(r.stdout)["capabilities"]} - self.assertTrue(caps["model_usage"]) + self.assertTrue(caps[("codex", "model_usage")]) class UnknownCountersCliTest(unittest.TestCase): diff --git a/tests/test_muse.py b/tests/test_muse.py new file mode 100644 index 0000000..ee5f12b --- /dev/null +++ b/tests/test_muse.py @@ -0,0 +1,537 @@ +"""Muse native sessions (#127): import, reconciliation and reporting. + +The fixture tree under tests/fixtures/muse mirrors real Muse 1.4.11 frame +shapes with invented ids and text. Expected values below are literal: +counts, statuses, UTF-8 sizes and identities, never the output of the +code under test. +""" + +import json +import os +import shutil +import sqlite3 +import subprocess +import sys +import tempfile +import unittest + +from agent_observer import db, lookup +from agent_observer.ingest import insert_event + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +TREE = os.path.join(REPO, "tests", "fixtures", "muse", "sessions") +ROOT = "0a000000-0000-7000-8000-000000000001" +RUN = "0a000000-0000-7000-8000-0000000000a1" +C1 = "0a000000-0000-7000-8000-0000000000c1" +C2 = "0a000000-0000-7000-8000-0000000000c2" +ROOT_KEY = "muse:" + ROOT +DAY = os.path.join("2026", "10", "19") +ROOT_LOG = os.path.join(TREE, DAY, ROOT, "session.jsonl") +CHILD_LOG = os.path.join(TREE, DAY, ROOT, "subagent", C1, "session.jsonl") +MINI = os.path.join(REPO, "tests", "fixtures", "codex-mini.jsonl") +CLAUDE = os.path.join(REPO, "tests", "fixtures", "claude-shapes.jsonl") + +# (family, native id) -> (name, target, status, size_bytes, tool_source, +# mcp_server, mcp_tool). Sizes are UTF-8 bytes of the committed text: +# "missing" 7, "café — résumé" 18 (13 characters), "über" 5, +# "skill body" 10, "ok" 2. +EXPECTED = { + ("tool_call", "call_r1"): ("read_file", "/work/project/missing.md", None, None, "builtin", None, None), + ("tool_call", "call_m1"): ("mcp__t3_code__web_read", None, None, None, "mcp", "t3-code", "web_read"), + ("tool_call", "call_r2"): ("read_file", "/work/project/README.md", None, None, "builtin", None, None), + ("tool_call", "call_s1"): ("read_skill", None, None, None, "builtin", None, None), + ("tool_call", "call_u1"): ("mcp__ghost__tool", None, None, None, "unknown", None, None), + ("tool_call", "call_b1"): ("mcp.lookalike", None, None, None, "builtin", None, None), + ("tool_call", "call_p1"): ("bash", "ls -la", None, None, "builtin", None, None), + ("tool_result", "call_r1"): ("read_file", None, "error", 7, "builtin", None, None), + ("tool_result", "call_m1"): ("mcp__t3_code__web_read", None, "ok", 18, "mcp", "t3-code", "web_read"), + ("tool_result", "call_r2"): ("read_file", None, "ok", 5, "builtin", None, None), + ("tool_result", "call_s1"): ("read_skill", None, "ok", 10, "builtin", None, None), + ("tool_result", "call_u1"): ("mcp__ghost__tool", None, "ok", None, "unknown", None, None), + ("tool_result", "call_b1"): ("mcp.lookalike", None, "ok", 2, "builtin", None, None), + ("read", "call_r1"): ("read_file", "/work/project/missing.md", "error", None, None, None, None), + ("read", "call_r2"): ("read_file", "/work/project/README.md", "ok", None, None, None, None), + ("skill_invoke", "call_s1"): ("read_skill", "plugin:agentsmd:operations", None, None, None, None, None), + ("lifecycle", C1): ("subagent_spawned", None, None, None, None, None, None), + ("lifecycle", C2): ("subagent_spawned", None, None, None, None, None, None), + ("lifecycle", "0b000000-0000-7000-8000-000000000004"): ("turn_started", None, None, None, None, None, None), + ("lifecycle", "0b000000-0000-7000-8000-000000000019"): ("turn_ended", None, "completed", None, None, None, None), +} + + +def run(db_path, *args): + env = dict(os.environ, AGENT_OBSERVER_DB=db_path) + return subprocess.run([sys.executable, "-m", "agent_observer", *args], + cwd=REPO, capture_output=True, text=True, env=env) + + +def ok(db_path, *args): + result = run(db_path, *args) + if result.returncode not in (0, 3): + raise AssertionError(f"{args}: {result.returncode} {result.stderr}") + return result + + +def root_events(db_path): + con = sqlite3.connect(db_path) + con.row_factory = sqlite3.Row + out = {} + for r in con.execute("SELECT * FROM events WHERE session_key=?", (ROOT_KEY,)): + detail = json.loads(r["detail_json"] or "{}") + out[(r["family"], r["native_id"])] = ( + r["name"], r["target"], r["status"], r["size_bytes"], + detail.get("tool_source"), detail.get("mcp_server"), detail.get("mcp_tool")) + con.close() + return out + + +def comparable(db_path): + """Ledger content under the order-test equality (#127 Spec).""" + con = sqlite3.connect(db_path) + con.row_factory = sqlite3.Row + events = sorted(tuple(r) for r in con.execute( + "SELECT session_key, family, native_id, ordinal_num, ts, turn_id, name," + " target, status, duration_ms, size_bytes, truncated, fingerprint," + " detail_json FROM events")) + sessions = sorted(tuple(r) for r in con.execute( + "SELECT session_key, harness, native_id, parent_session_key, role," + " project_dir, git_branch, started_at, ended_at FROM sessions")) + turns = sorted(tuple(r) for r in con.execute( + "SELECT turn_id, session_key, model_observed, started_at, completed_at," + " duration_ms, state FROM turns")) + errors = sorted(tuple(r) for r in con.execute( + "SELECT harness, error, ordinal_num, line_excerpt FROM import_errors")) + con.close() + return events, sessions, turns, errors + + +class MuseCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = os.path.join(self.tmp.name, "muse.db") + + def tearDown(self): + self.tmp.cleanup() + + def copy_tree(self, name="tree"): + dest = os.path.join(self.tmp.name, name) + shutil.copytree(TREE, dest) + return dest + + def sync_tree(self, tree=TREE, db_path=None): + return ok(db_path or self.db, "sync", "--harness", "muse", "--root", tree) + + def own_root(self, task="T-MUSE", db_path=None): + db_path = db_path or self.db + ok(db_path, "capture", "create-task", "--task", task) + ok(db_path, "capture", "assign-session", "--task", task, + "--session", ROOT_KEY, "--exclusive", "--evidence", "fixture") + + def task(self, task="T-MUSE"): + return json.loads(ok(self.db, "task", "show", "--task", task, "--json").stdout) + + def publish(self, task="T-MUSE"): + return json.loads(ok(self.db, "publish", "--task", task, "--dry-run", "--json").stdout) + + +class ImportTest(MuseCase): + def test_each_call_is_one_call_and_one_result_with_literal_evidence(self): + result = self.sync_tree() + self.assertIn("muse: 2 sources (0 unchanged), 0 new responses, 24 new events", + result.stdout) + self.assertEqual(root_events(self.db), EXPECTED) + con = sqlite3.connect(self.db) + self.assertEqual(con.execute("SELECT COUNT(*) FROM responses").fetchone()[0], 0) + turns = con.execute("SELECT turn_id, state, duration_ms FROM turns" + " WHERE session_key=?", (ROOT_KEY,)).fetchall() + self.assertEqual(turns, [("muse:" + RUN, "completed", 21000)]) + # Internal task starts add no turn; every root event carries the run. + self.assertEqual(con.execute( + "SELECT DISTINCT turn_id FROM events WHERE session_key=?", + (ROOT_KEY,)).fetchall(), [("muse:" + RUN,)]) + con.close() + + def test_unchanged_resync_writes_nothing(self): + self.sync_tree() + before = comparable(self.db) + again = json.loads(ok(self.db, "sync", "--harness", "muse", "--root", TREE, + "--json").stdout) + row = next(h for h in again["harnesses"] if h["harness"] == "muse") + self.assertEqual((row["unchanged"], row["events_inserted"], row["events_updated"]), + (2, 0, 0)) + self.assertEqual(comparable(self.db), before) + + def test_split_appends_reach_the_same_literal_result(self): + tree = self.copy_tree() + path = os.path.join(tree, DAY, ROOT, "session.jsonl") + with open(ROOT_LOG, "rb") as fh: + lines = fh.readlines() + split_db = os.path.join(self.tmp.name, "split.db") + with open(path, "wb") as fh: + fh.truncate() + statuses = [] + for index, line in enumerate(lines): + with open(path, "ab") as fh: + fh.write(line[:len(line) // 2]) + # A partial trailing line is left for the next sync. + ok(split_db, "sync", "--harness", "muse", "--source", path) + with open(path, "ab") as fh: + fh.write(line[len(line) // 2:]) + ok(split_db, "sync", "--harness", "muse", "--source", path) + events = root_events(split_db) + statuses.append(events.get(("tool_result", "call_r2"), (None,) * 4)[2:4]) + # call_r2: result committed (size 5, no status) before its terminal. + self.assertIn((None, 5), statuses) + self.assertEqual(statuses[-1], ("ok", 5)) + self.assertEqual(root_events(split_db), EXPECTED) + ok(self.db, "sync", "--harness", "muse", "--source", ROOT_LOG) + self.assertEqual(comparable(split_db), comparable(self.db)) + + def test_partial_line_is_not_imported(self): + tree = self.copy_tree() + path = os.path.join(tree, DAY, ROOT, "session.jsonl") + with open(ROOT_LOG, "rb") as fh: + lines = fh.readlines() + with open(path, "wb") as fh: + fh.writelines(lines[:-1]) + fh.write(lines[-1][:20]) + ok(self.db, "sync", "--harness", "muse", "--source", path) + con = sqlite3.connect(self.db) + self.assertEqual(con.execute("SELECT COUNT(*) FROM import_errors").fetchone()[0], 0) + con.close() + + def test_consumed_unsupported_schema_is_quarantined_by_key_names(self): + tree = self.copy_tree() + path = os.path.join(tree, DAY, ROOT, "session.jsonl") + with open(path, "a") as fh: + for ptype, payload in ( + ("tool_batch.effect.terminal", {"kind": "tool_batch_effect", "run_id": RUN, + "record": {"call_id": "call_p1", + "outcome": {"kind": "completed"}}}), + ("runtime.unconsumed.kind", {"kind": "other"})): + fh.write(json.dumps({ + "id": "0b000000-0000-7000-8000-0000000000ff", "payload": payload, + "payload_schema_version": 1, "payload_type": ptype, + "record_type": "event", "recorded_at": 1792442499000000, + "schema_version": 2, "sequence": 99, + "stream": {"id": ROOT, "kind": "session"}}) + "\n") + ok(self.db, "sync", "--harness", "muse", "--source", path) + con = sqlite3.connect(self.db) + errors = con.execute("SELECT error, ordinal_num, line_excerpt FROM import_errors").fetchall() + self.assertEqual(errors, [("unsupported_schema", 27, "id,payload,payload_schema_version," + "payload_type,record_type,recorded_at,schema_version,sequence,stream")]) + # The unsupported terminal did not answer call_p1. + self.assertNotIn(("tool_result", "call_p1"), root_events(self.db)) + con.close() + + def test_no_free_text_reaches_the_ledger(self): + self.sync_tree() + con = sqlite3.connect(self.db) + dump = "\n".join(con.iterdump()) + con.close() + for text in ("synthetic reason text", "synthetic prompt", "café", + "skill body", "https://example.invalid", "\"decision\""): + self.assertNotIn(text, dump) + + def test_identity_detail_is_closed_and_validated(self): + from agent_observer import privacy + self.assertEqual(privacy.filter_detail("tool_result", { + "tool_source": "mcp", "mcp_server": "t3-code", "mcp_tool": "web_read"}), + {"tool_source": "mcp", "mcp_server": "t3-code", "mcp_tool": "web_read"}) + self.assertEqual(privacy.filter_detail("tool_call", { + "tool_source": "MCP", "mcp_server": "server with spaces", + "mcp_tool": "tool"}), {}) + self.assertEqual(privacy.filter_detail("read", {"tool_source": "mcp"}), {}) + + def test_existing_lookup_reports_not_found_without_an_indexer(self): + self.sync_tree() + con = db.connect(self.db) + event_id = con.execute("SELECT id FROM events WHERE family='tool_call'" + " AND native_id='call_m1'").fetchone()[0] + self.assertEqual(lookup.show(con, [event_id])[0]["status"], "not_found") + con.close() + + +class ChildTest(MuseCase): + def availability(self, db_path, session=None): + args = ["sessions", "availability", "--json"] + if session: + args += ["--session", session] + return json.loads(ok(db_path, *args).stdout) + + def test_children_report_their_own_availability(self): + self.sync_tree() + machine = self.availability(self.db) + self.assertEqual(machine["sessions"], {"available": 2, "partial": 0, "ledger_only": 1}) + self.assertEqual(self.availability(self.db, "muse:" + C2)["status"], "ledger_only") + self.assertEqual(self.availability(self.db, "muse:" + C1)["status"], "available") + con = sqlite3.connect(self.db) + parents = dict(con.execute("SELECT session_key, parent_session_key FROM sessions" + " WHERE role='subagent'").fetchall()) + con.close() + self.assertEqual(parents, {"muse:" + C1: ROOT_KEY, "muse:" + C2: ROOT_KEY}) + + def test_parent_first_and_child_first_converge(self): + other = os.path.join(self.tmp.name, "child-first.db") + ok(self.db, "sync", "--harness", "muse", "--source", ROOT_LOG) + ok(self.db, "sync", "--harness", "muse", "--source", CHILD_LOG) + ok(other, "sync", "--harness", "muse", "--source", CHILD_LOG) + ok(other, "sync", "--harness", "muse", "--source", ROOT_LOG) + self.assertEqual(comparable(self.db), comparable(other)) + self.assertEqual(self.availability(self.db), self.availability(other)) + + def test_a_late_child_log_becomes_available_without_duplicates(self): + tree = self.copy_tree() + self.sync_tree(tree) + before = comparable(self.db) + late = os.path.join(tree, DAY, ROOT, "subagent", C2) + os.makedirs(late) + with open(CHILD_LOG) as src, open(os.path.join(late, "session.jsonl"), "w") as dst: + dst.write(src.read().replace(C1, C2).replace("call_c1", "call_c2")) + self.sync_tree(tree) + self.assertEqual(self.availability(self.db, "muse:" + C2)["status"], "available") + con = sqlite3.connect(self.db) + self.assertEqual(con.execute("SELECT COUNT(*) FROM sessions").fetchone()[0], 3) + self.assertEqual(con.execute( + "SELECT COUNT(*) FROM events WHERE family='lifecycle' AND name='subagent_spawned'" + ).fetchone()[0], 2) + con.close() + self.assertEqual(len(comparable(self.db)[0]), len(before[0]) + 4) + + def test_directory_and_stream_mismatch_links_no_parent(self): + tree = self.copy_tree() + wrong = os.path.join(tree, DAY, ROOT, "subagent", C2) + os.rename(os.path.join(tree, DAY, ROOT, "subagent", C1), wrong) + ok(self.db, "sync", "--harness", "muse", "--source", os.path.join(wrong, "session.jsonl")) + con = sqlite3.connect(self.db) + self.assertEqual(con.execute("SELECT session_key, parent_session_key FROM sessions").fetchall(), + [("muse:" + C1, None)]) + self.assertEqual(con.execute("SELECT error FROM import_errors").fetchall(), + [("identity_conflict",)]) + con.close() + + def test_grok_child_without_log_is_not_available(self): + con = db.connect(self.db) + db.init_db(con) + db.upsert_session(con, "grok:child-1", "grok", "child-1", None, + parent_session_key="grok:parent-1", role="subagent") + con.commit() + con.close() + self.assertEqual(self.availability(self.db, "grok:child-1")["status"], "ledger_only") + + +class ReportTest(MuseCase): + def test_muse_only_task_reports_activity_and_unknown_usage(self): + self.sync_tree() + self.own_root() + rep = self.task() + nulls = {b: None for b in ("input_tokens", "cached_input_tokens", + "cache_write_input_tokens", "output_tokens", + "reasoning_output_tokens", "total_tokens")} + for section in (rep["attributed"], rep["scope"]): + self.assertEqual({k: section[k] for k in nulls}, nulls) + self.assertEqual(section["usage_status"], "unknown") + self.assertEqual(section["unmeasured_sessions"], [ROOT_KEY]) + # Nothing is shared: that partition stays a known zero. + self.assertEqual(rep["shared_joint"]["total_tokens"], 0) + self.assertNotIn("usage_status", rep["shared_joint"]) + [phase] = rep["phases"] + self.assertEqual(phase["usage"]["total_tokens"], None) + self.assertEqual(phase["usage"]["usage_status"], "unknown") + self.assertFalse(phase["estimated_cost"]["complete"]) + self.assertIsNone(phase["estimated_cost"]["estimated_cost_usd_total"]) + self.assertEqual(phase["estimated_cost"]["status"], "unknown") + self.assertEqual(phase["activity"], { + "tool_calls": 7, "mcp_results": 1, "reads": 2, "file_changes": None, + "failed_tool_results": 1, "skill_invocations": 1}) + self.assertEqual(phase["activity_coverage"], {"file_changes": "unavailable"}) + self.assertEqual(rep["coverage"]["unbound_child_sessions"], ["muse:" + C1, "muse:" + C2]) + self.assertFalse(rep["complete"]) + self.assertEqual(rep["total_cost"]["text"], "unknown") + text = ok(self.db, "task", "show", "--task", "T-MUSE").stdout + self.assertIn("attributed responses: 0 total=unknown; 1 unmeasured sessions", text) + self.assertIn("phase unclassified: responses=0 total=unknown; 1 unmeasured sessions", text) + published = self.publish() + self.assertEqual(published["summary"]["usage"]["usage_status"], "unknown") + self.assertEqual(published["summary"]["phases"], rep["phases"]) + body = published["body"] + self.assertIn("| unclassified | unmeasured sessions (1) | 0 | unknown | unknown | unknown" + " | unknown | unknown | unknown | unknown |", body) + self.assertIn("| unclassified | 7 | 1 | 2 | 1 | unavailable | 1 |", body) + self.assertIn("2 worker sessions without an owner", body) + self.assertNotIn("$0.00", body) + + def test_muse_beside_measured_codex_is_a_lower_bound(self): + self.sync_tree() + ok(self.db, "sync", "--source", MINI) + self.own_root() + ok(self.db, "capture", "assign-session", "--task", "T-MUSE", "--session", + "codex:sess-fixture-mini-01", "--exclusive", "--evidence", "fixture") + rep = self.task() + self.assertEqual(rep["attributed"]["total_tokens"], 5500) + self.assertEqual(rep["attributed"]["usage_status"], "lower_bound") + [phase] = rep["phases"] + self.assertEqual(phase["usage"]["total_tokens"], 5500) + self.assertEqual(phase["usage"]["usage_status"], "lower_bound") + self.assertFalse(phase["estimated_cost"]["complete"]) + self.assertIsNone(phase["estimated_cost"]["estimated_cost_usd_total"]) + # Codex records skill invocations as unsupported and Muse records + # no file changes: the measured counts stay, marked partial. + self.assertEqual(phase["activity_coverage"], + {"file_changes": "partial", "skill_invocations": "partial"}) + self.assertEqual(phase["activity"]["skill_invocations"], 1) + text = ok(self.db, "task", "show", "--task", "T-MUSE").stdout + self.assertIn("total=>=5,500 (lower bound); 1 unmeasured sessions", text) + body = self.publish()["body"] + self.assertIn("unmeasured sessions (1)", body) + self.assertIn(">=1 (partial)", body) + + def test_mixed_semantics_keep_by_semantics_only(self): + self.sync_tree() + ok(self.db, "sync", "--source", MINI) + ok(self.db, "sync", "--harness", "claude", "--source", CLAUDE) + self.own_root() + con = sqlite3.connect(self.db) + keys = [r[0] for r in con.execute( + "SELECT DISTINCT session_key FROM responses WHERE session_key LIKE 'codex:%'" + " OR session_key LIKE 'claude:%'")] + con.close() + for key in keys: + ok(self.db, "capture", "assign-session", "--task", "T-MUSE", "--session", key, + "--exclusive", "--evidence", "fixture") + rep = self.task() + self.assertNotIn("total_tokens", rep["attributed"]) + self.assertIn("by_semantics", rep["attributed"]) + self.assertEqual(rep["attributed"]["usage_status"], "lower_bound") + text = ok(self.db, "task", "show", "--task", "T-MUSE").stdout + self.assertIn("total=mixed semantics (see --json); 1 unmeasured sessions", text) + + def test_measured_only_and_known_zero_controls_are_unchanged(self): + ok(self.db, "sync", "--source", MINI) + ok(self.db, "capture", "create-task", "--task", "T-CX") + ok(self.db, "capture", "assign-session", "--task", "T-CX", "--session", + "codex:sess-fixture-mini-01", "--exclusive", "--evidence", "fixture") + rep = self.task("T-CX") + self.assertEqual(rep["attributed"]["total_tokens"], 5500) + self.assertNotIn("usage_status", rep["attributed"]) + self.assertEqual(rep["shared_joint"]["total_tokens"], 0) + for phase in rep["phases"]: + self.assertNotIn("usage_status", phase["usage"]) + self.assertNotIn("file_changes", phase["activity_coverage"]) + # A response-less session of a harness that imports usage keeps + # today's exact zero. + con = db.connect(self.db) + source = con.execute("INSERT INTO sources(harness, path, sha256, imported_at)" + " VALUES('codex','fixture:empty','x',1.0)").lastrowid + db.upsert_session(con, "codex:empty", "codex", "empty", source) + con.commit() + con.close() + ok(self.db, "capture", "create-task", "--task", "T-EMPTY") + ok(self.db, "capture", "assign-session", "--task", "T-EMPTY", "--session", + "codex:empty", "--exclusive", "--evidence", "fixture") + empty = self.task("T-EMPTY") + self.assertEqual(empty["attributed"]["total_tokens"], 0) + self.assertNotIn("usage_status", empty["attributed"]) + listing = ok(self.db, "sessions", "list").stdout + self.assertIn("codex:empty", listing) + self.assertIn("responses=0 total=0 ", listing) + + def test_sessions_list_and_show_keep_unknown_usage(self): + self.sync_tree() + listing = ok(self.db, "sessions", "list", "--harness", "muse").stdout + self.assertIn(f"{ROOT_KEY} /work/project agentsmd=unknown responses=0 total=unknown", + listing) + rows = {r["session_key"]: r for r in json.loads( + ok(self.db, "sessions", "list", "--harness", "muse", "--json").stdout)["sessions"]} + self.assertEqual(rows[ROOT_KEY]["usage_status"], "unknown") + self.assertIsNone(rows[ROOT_KEY]["total_tokens"]) + show = json.loads(ok(self.db, "sessions", "show", "--session", ROOT_KEY).stdout) + self.assertEqual(show["usage"]["usage_status"], "unknown") + self.assertEqual(show["usage"]["unmeasured_sessions"], [ROOT_KEY]) + + def ledger_cell(self, sessions): + """One task whole-owning sessions built directly in the ledger.""" + con = db.connect(self.db) + db.init_db(con) + for key, harness, family in sessions: + source = con.execute("INSERT INTO sources(harness, path, sha256, imported_at)" + " VALUES(?,?,?,?)", (harness, "fixture:" + key, "x", 1.0)).lastrowid + db.upsert_session(con, key, harness, key.split(":", 1)[1], source) + if family: + insert_event(con, {}, source_id=source, session_key=key, family=family, + native_id="e-" + key, name="Edit" if family == "file_change" else "skill", + target="/work/a.py" if family == "file_change" else "agentsmd:operations") + con.execute("INSERT INTO tasks(task_id, created_at) VALUES('T-CELL', 1.0)") + for key, _, _ in sessions: + con.execute("INSERT INTO session_assignments VALUES(?,?,?,?)", + (key, "T-CELL", "fixture", 1.0)) + con.commit() + con.close() + + def test_partial_cells_keep_measured_counts(self): + self.sync_tree() + self.ledger_cell([("opencode:oc-1", "opencode", "file_change")]) + con = db.connect(self.db) + con.execute("INSERT INTO session_assignments VALUES(?,?,?,?)", + (ROOT_KEY, "T-CELL", "fixture", 1.0)) + con.commit() + con.close() + [phase] = self.task("T-CELL")["phases"] + self.assertEqual(phase["activity"]["file_changes"], 1) + self.assertEqual(phase["activity_coverage"]["file_changes"], "partial") + body = self.publish("T-CELL")["body"] + self.assertIn("| unclassified | 7 | 1 | 2 | 1 | >=1 (partial) | 1 |", body) + + def test_claude_skill_beside_codex_is_partial(self): + # A session contributes to a cell through its events. + self.ledger_cell([("claude:cl-1", "claude", "skill_invoke"), + ("codex:cx-1", "codex", "tool_call")]) + [phase] = self.task("T-CELL")["phases"] + self.assertEqual(phase["activity"]["skill_invocations"], 1) + self.assertEqual(phase["activity_coverage"], {"skill_invocations": "partial"}) + + def test_router_owned_native_session_keeps_native_capabilities(self): + self.ledger_cell([("codex:cx-r", "codex", "file_change")]) + con = db.connect(self.db) + con.execute("INSERT INTO task_turns(task_id, turn_id, role, harness, session_key, stage)" + " VALUES('T-CELL','router:inv-1','child','router','codex:cx-r','implementation')") + con.commit() + con.close() + phases = {p["phase"]: p for p in self.task("T-CELL")["phases"]} + [phase] = phases.values() + self.assertEqual(phase["activity"]["tool_calls"], 0) + self.assertEqual(phase["activity"]["file_changes"], 1) + self.assertEqual(phase["activity_coverage"], {"skill_invocations": "unavailable"}) + + +class T3Test(MuseCase): + def test_strong_muse_thread_binds_the_root_and_weak_binds_nothing(self): + from agent_observer.adapters import t3 as t3_adapter + from tests.test_t3_v2 import PR5, ROOT as THREAD, write_v2 + state = write_v2(os.path.join(self.tmp.name, "chromeria-v2.sqlite")) + native = sqlite3.connect(state) + for tid, native_id, strength in ((THREAD, ROOT, "strong"), + ("thread:weak-muse", C2, "weak")): + native.execute( + "INSERT INTO orchestration_v2_projection_provider_threads(" + " provider_thread_id, thread_id, provider, status, updated_at," + " payload_json, driver) VALUES(?,?,?,?,?,?,?)", + (f"provider-thread:{tid}:{native_id}", tid, "muse", "idle", + "2026-10-19T20:00:00Z", + json.dumps({"appThreadId": tid, "driver": "muse", + "nativeThreadRef": {"driver": "muse", "nativeId": native_id, + "strength": strength}}), "muse")) + native.commit() + native.close() + self.sync_tree() + con = db.connect(self.db) + t3_adapter.sync(con, source=state) + bound = {(r[0], r[1]) for r in con.execute( + "SELECT session_key, task_id FROM session_assignments WHERE session_key LIKE 'muse:%'")} + con.close() + self.assertEqual(bound, {(ROOT_KEY, "example/beta#5")}) + self.assertTrue(PR5.endswith("/5")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pr_checks_pending.py b/tests/test_pr_checks_pending.py index e9fb13a..1f4f793 100644 --- a/tests/test_pr_checks_pending.py +++ b/tests/test_pr_checks_pending.py @@ -40,7 +40,7 @@ def test_real_exit8_is_pending_and_not_a_failure(self): self.sync("codex-mini.jsonl") usage = dict(self.query("SELECT * FROM responses LIMIT 1")[0]) usage["session_key"] = "codex:pr-checks-pending" - phases, _ = report._phase_report( + phases, _, _ = report._phase_report( self.con, "unbound-task", [usage], {usage["session_key"]}, [], pricing.load_schedule()) self.assertEqual(phases[0]["activity"]["failed_tool_results"], 0) From 4bc4f706a8610c7633f0123ca0a43f3fd48c96db Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Tue, 20 Oct 2026 00:10:42 +0200 Subject: [PATCH 2/5] fix: quarantine malformed consumed Muse records (#127) Check run discriminators and terminal outcomes for type before use, so a malformed record becomes a per-line schema_error instead of aborting the import, and an unusable terminal leaves the result status unknown. Version-gate effect-started and session-opened frames like the other consumed variants. Unconsumed kinds stay silently skipped. Co-Authored-By: Claude Opus 5.5 --- agent_observer/adapters/muse.py | 35 +++++++++++++------- docs/contracts.md | 10 ++++-- tests/test_muse.py | 57 +++++++++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 13 deletions(-) diff --git a/agent_observer/adapters/muse.py b/agent_observer/adapters/muse.py index 1cf8ae0..6144850 100644 --- a/agent_observer/adapters/muse.py +++ b/agent_observer/adapters/muse.py @@ -73,6 +73,13 @@ "assistant_tool_calls_committed", "tool_result_batch_committed", "memory_reminder_child_session_linked"}) _STATUS = {"completed": "ok", "failed": "error"} +# Consumed non-run payload types; each needs schema and payload version 1. +# Effect-started and session-opened carry nothing imported but are version +# gated like the rest. +_CONSUMED = ( + "runtime.mcp_tool_identity_catalog", "runtime.session.metadata", + "session.opened.observed", "session.workspace_branch.observed", + "session.end", "tool_batch.effect.started", "tool_batch.effect.terminal") def _session_id(value) -> str | None: @@ -161,18 +168,19 @@ def bad(ordinal, category, raw): ptype = obj.get("payload_type") payload = obj.get("payload") if isinstance(obj.get("payload"), dict) else {} event = payload.get("event") if isinstance(payload.get("event"), dict) else {} - run_kind = ptype == "runtime.session" and payload.get("kind") == "run" \ - and event.get("kind") in _RUN_KINDS - consumed = run_kind or ptype in ( - "runtime.mcp_tool_identity_catalog", "runtime.session.metadata", - "session.workspace_branch.observed", "session.end", - "tool_batch.effect.terminal") ts = _ts(obj.get("recorded_at")) if ts is not None: state.started_at = ts if state.started_at is None else min(state.started_at, ts) state.ended_at = ts if state.ended_at is None else max(state.ended_at, ts) - if not consumed: + # Discriminators are checked for type before any membership test, + # so a malformed consumed record is quarantined, never a crash. + run_frame = ptype == "runtime.session" and payload.get("kind") == "run" + if run_frame and not isinstance(event.get("kind"), str): + bad(ordinal, "schema_error", raw) continue + consumed = (run_frame and event["kind"] in _RUN_KINDS) or ptype in _CONSUMED + if not consumed: + continue # Deliberately unconsumed: envelope, kind=task, usage, others. if obj.get("schema_version") != 1 or obj.get("payload_schema_version") != 1: bad(ordinal, "unsupported_schema", raw) continue @@ -210,15 +218,20 @@ def bad(ordinal, category, raw): ref = record.get("reference") if isinstance(record.get("reference"), dict) else {} if ref.get("kind") == "branch" and _str(ref.get("name")): state.branch = ref["name"] - elif ptype == "session.end": - pass # recorded_at already extends ended_at. + elif ptype in ("session.end", "session.opened.observed", + "tool_batch.effect.started"): + pass # Version gated only; recorded_at already extends the span. elif ptype == "tool_batch.effect.terminal": call = _str(record.get("call_id")) - outcome = record.get("outcome") if isinstance(record.get("outcome"), dict) else {} + outcome = record.get("outcome") if call is None: bad(ordinal, "missing_id", raw) continue - state.terminals[call] = {**at, "status": _STATUS.get(outcome.get("kind"))} + if not isinstance(outcome, dict) or not isinstance(outcome.get("kind"), str): + # No usable outcome: the result's status stays unknown. + bad(ordinal, "schema_error", raw) + continue + state.terminals[call] = {**at, "status": _STATUS.get(outcome["kind"])} else: run_id = _str(payload.get("run_id")) kind = event["kind"] diff --git a/docs/contracts.md b/docs/contracts.md index 484f025..80babfc 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -590,8 +590,14 @@ Default root `~/.local/share/muse/sessions`: each matches its `stream.id`; a mismatch is an `identity_conflict`. Record-supplied log paths are never opened or stored. A linked child with no backing source reads `ledger_only` (`no_source`) in `sessions availability`. -- Consumed variants need `schema_version` and `payload_schema_version` 1, - else `unsupported_schema`; unconsumed kinds are skipped. Prompts, result +- Consumed variants (catalog, session metadata, opened, branch and end, + effect started and terminal, and the listed `kind=run` events) need + `schema_version` and `payload_schema_version` 1, else `unsupported_schema`. + A run frame whose `event.kind` is not a string, or a terminal without an + outcome `kind` string, is a `schema_error`; that terminal's result keeps + an unknown status. Each bad record is quarantined alone at its line number + with key names only, and the rest of the file still imports. Unconsumed + kinds (the envelope, `kind=task`, usage) are skipped silently. Prompts, result text, outcome reasons, hooks, memory and usage are never stored. - T3 maps `driver: muse` strong `nativeThreadRef` to `muse:`. diff --git a/tests/test_muse.py b/tests/test_muse.py index ee5f12b..6e1523e 100644 --- a/tests/test_muse.py +++ b/tests/test_muse.py @@ -224,6 +224,63 @@ def test_consumed_unsupported_schema_is_quarantined_by_key_names(self): self.assertNotIn(("tool_result", "call_p1"), root_events(self.db)) con.close() + def edited_import(self, line, edit): + """Sync a copy of the root log whose 0-based line was edited.""" + tree = self.copy_tree() + path = os.path.join(tree, DAY, ROOT, "session.jsonl") + with open(path) as fh: + records = [json.loads(text) for text in fh] + edit(records[line]) + with open(path, "w") as fh: + fh.writelines(json.dumps(r, sort_keys=True) + "\n" for r in records) + result = run(self.db, "sync", "--harness", "muse", "--source", path) + self.assertEqual(result.returncode, 0, result.stderr) + con = sqlite3.connect(self.db) + errors = con.execute("SELECT error, ordinal_num, line_excerpt FROM import_errors").fetchall() + con.close() + return errors, root_events(self.db) + + FRAME_KEYS = ("causation_id,durability,id,payload,payload_schema_version,payload_type," + "record_type,recorded_at,schema_version,sequence,stream") + + def test_wrong_type_terminal_outcome_is_quarantined(self): + # Line 9 is call_r1's failed terminal. + errors, events = self.edited_import( + 9, lambda r: r["payload"]["record"]["outcome"].update(kind=[])) + self.assertEqual(errors, [("schema_error", 9, self.FRAME_KEYS)]) + expected = dict(EXPECTED) + expected[("tool_result", "call_r1")] = ("read_file", None, None, 7, "builtin", None, None) + expected[("read", "call_r1")] = ("read_file", "/work/project/missing.md", None, None, + None, None, None) + self.assertEqual(events, expected) + + def test_missing_terminal_outcome_leaves_status_unknown(self): + errors, events = self.edited_import( + 9, lambda r: r["payload"]["record"].update(outcome={})) + self.assertEqual(errors, [("schema_error", 9, self.FRAME_KEYS)]) + self.assertEqual(events[("tool_result", "call_r1")][2:4], (None, 7)) + self.assertEqual(events[("tool_result", "call_m1")][2:4], ("ok", 18)) + + def test_wrong_type_run_discriminator_is_quarantined(self): + # Line 4 is the run's start frame; every other event survives. + errors, events = self.edited_import(4, lambda r: r["payload"]["event"].update(kind=[])) + self.assertEqual(errors, [("schema_error", 4, self.FRAME_KEYS)]) + expected = {k: v for k, v in EXPECTED.items() + if k != ("lifecycle", "0b000000-0000-7000-8000-000000000004")} + self.assertEqual(events, expected) + + def test_unconsumed_task_frame_stays_silent_even_when_odd(self): + # Line 6 is an internal kind=task start: deliberately not consumed. + errors, events = self.edited_import(6, lambda r: r["payload"]["event"].update(kind=[])) + self.assertEqual(errors, []) + self.assertEqual(events, EXPECTED) + + def test_effect_started_is_version_gated(self): + # Line 8 is call_r1's started effect; it adds no evidence itself. + errors, events = self.edited_import(8, lambda r: r.update(schema_version=2)) + self.assertEqual(errors, [("unsupported_schema", 8, self.FRAME_KEYS)]) + self.assertEqual(events, EXPECTED) + def test_no_free_text_reaches_the_ledger(self): self.sync_tree() con = sqlite3.connect(self.db) From 9463a7a7e8d72b3f1f00ff9bb2dd0184b9a8fd2a Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Tue, 20 Oct 2026 00:24:42 +0200 Subject: [PATCH 3/5] fix: let only accepted Muse records set identity and span (#127) A consumed record now establishes the session id and widens the session span only after its validation passes. A rejected first record from another stream can no longer claim the source, and a rejected record's timestamp no longer moves started_at. Co-Authored-By: Claude Opus 5.5 --- agent_observer/adapters/muse.py | 32 +++++++++++++++++++++------- docs/contracts.md | 3 ++- tests/test_muse.py | 37 +++++++++++++++++++++++++++++++++ 3 files changed, 64 insertions(+), 8 deletions(-) diff --git a/agent_observer/adapters/muse.py b/agent_observer/adapters/muse.py index 6144850..91380fb 100644 --- a/agent_observer/adapters/muse.py +++ b/agent_observer/adapters/muse.py @@ -159,6 +159,18 @@ def bad(ordinal, category, raw): src.error(ordinal, category, raw) totals["malformed"] += 1 + def extend(ts): + if ts is not None: + state.started_at = ts if state.started_at is None else min(state.started_at, ts) + state.ended_at = ts if state.ended_at is None else max(state.ended_at, ts) + + def accept(sid, ts): + """A consumed record passed validation: only now may it establish + the session identity or widen the span. A rejected one does neither.""" + if state.session is None: + state.session = sid + extend(ts) + for ordinal, obj, raw in src.records(): if obj is None: bad(ordinal, "malformed_json", raw) @@ -169,9 +181,6 @@ def bad(ordinal, category, raw): payload = obj.get("payload") if isinstance(obj.get("payload"), dict) else {} event = payload.get("event") if isinstance(payload.get("event"), dict) else {} ts = _ts(obj.get("recorded_at")) - if ts is not None: - state.started_at = ts if state.started_at is None else min(state.started_at, ts) - state.ended_at = ts if state.ended_at is None else max(state.ended_at, ts) # Discriminators are checked for type before any membership test, # so a malformed consumed record is quarantined, never a crash. run_frame = ptype == "runtime.session" and payload.get("kind") == "run" @@ -180,6 +189,7 @@ def bad(ordinal, category, raw): continue consumed = (run_frame and event["kind"] in _RUN_KINDS) or ptype in _CONSUMED if not consumed: + extend(ts) # Unconsumed but well-formed frames still date the session. continue # Deliberately unconsumed: envelope, kind=task, usage, others. if obj.get("schema_version") != 1 or obj.get("payload_schema_version") != 1: bad(ordinal, "unsupported_schema", raw) @@ -189,9 +199,7 @@ def bad(ordinal, category, raw): if sid is None: bad(ordinal, "missing_id", raw) continue - if state.session is None: - state.session = sid - elif sid != state.session: + if state.session is not None and sid != state.session: bad(ordinal, "identity_conflict", raw) continue seq = obj.get("sequence") if type(obj.get("sequence")) is int else None @@ -202,6 +210,7 @@ def bad(ordinal, category, raw): if not isinstance(entries, list): bad(ordinal, "schema_error", raw) continue + accept(sid, ts) state.catalog_seen = True for entry in entries: if not isinstance(entry, dict): @@ -212,15 +221,17 @@ def bad(ordinal, category, raw): if name and server and tool: state.catalog.setdefault(name, set()).add((server, tool)) elif ptype == "runtime.session.metadata": + accept(sid, ts) state.project_dir = state.project_dir or _str(record.get("workspace_root")) state.model = state.model or _str(record.get("model_id")) elif ptype == "session.workspace_branch.observed": + accept(sid, ts) ref = record.get("reference") if isinstance(record.get("reference"), dict) else {} if ref.get("kind") == "branch" and _str(ref.get("name")): state.branch = ref["name"] elif ptype in ("session.end", "session.opened.observed", "tool_batch.effect.started"): - pass # Version gated only; recorded_at already extends the span. + accept(sid, ts) # Version gated only; adds no evidence beyond the span. elif ptype == "tool_batch.effect.terminal": call = _str(record.get("call_id")) outcome = record.get("outcome") @@ -231,6 +242,7 @@ def bad(ordinal, category, raw): # No usable outcome: the result's status stays unknown. bad(ordinal, "schema_error", raw) continue + accept(sid, ts) state.terminals[call] = {**at, "status": _STATUS.get(outcome["kind"])} else: run_id = _str(payload.get("run_id")) @@ -244,15 +256,18 @@ def bad(ordinal, category, raw): if frame is None: bad(ordinal, "missing_id", raw) continue + accept(sid, ts) state.runs.setdefault(run_id, {})["start"] = {**at, "frame": frame} elif kind == "terminal": frame = _str(obj.get("id")) if frame is None: bad(ordinal, "missing_id", raw) continue + accept(sid, ts) state.runs.setdefault(run_id, {})["end"] = { **at, "frame": frame, "terminal": _str(event.get("terminal"))} elif kind == "model_request_configured": + accept(sid, ts) toolset = event.get("toolset") if isinstance(event.get("toolset"), dict) else {} active = toolset.get("active_tools") if isinstance(active, list): @@ -263,6 +278,7 @@ def bad(ordinal, category, raw): if not isinstance(items, list): bad(ordinal, "schema_error", raw) continue + accept(sid, ts) for item in items: call = _str(item.get("call_id")) if isinstance(item, dict) else None name = _str(item.get("name")) if isinstance(item, dict) else None @@ -282,6 +298,7 @@ def bad(ordinal, category, raw): if not isinstance(items, list): bad(ordinal, "schema_error", raw) continue + accept(sid, ts) for item in items: call = _str(item.get("tool_call_id")) if isinstance(item, dict) else None if call is None: @@ -296,6 +313,7 @@ def bad(ordinal, category, raw): if child is None: bad(ordinal, "missing_id", raw) continue + accept(sid, ts) state.links.setdefault(child, at) return state diff --git a/docs/contracts.md b/docs/contracts.md index 80babfc..d305f4b 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -596,7 +596,8 @@ Default root `~/.local/share/muse/sessions`: each A run frame whose `event.kind` is not a string, or a terminal without an outcome `kind` string, is a `schema_error`; that terminal's result keeps an unknown status. Each bad record is quarantined alone at its line number - with key names only, and the rest of the file still imports. Unconsumed + with key names only: it neither sets the session identity nor widens the + session span, and the rest of the file still imports. Unconsumed kinds (the envelope, `kind=task`, usage) are skipped silently. Prompts, result text, outcome reasons, hooks, memory and usage are never stored. - T3 maps `driver: muse` strong `nativeThreadRef` to `muse:`. diff --git a/tests/test_muse.py b/tests/test_muse.py index 6e1523e..5f98465 100644 --- a/tests/test_muse.py +++ b/tests/test_muse.py @@ -269,6 +269,43 @@ def test_wrong_type_run_discriminator_is_quarantined(self): if k != ("lifecycle", "0b000000-0000-7000-8000-000000000004")} self.assertEqual(events, expected) + def test_rejected_first_record_cannot_claim_the_session(self): + # A malformed terminal from another invented stream, inserted before + # every valid consumed frame, is quarantined alone. + tree = self.copy_tree() + path = os.path.join(tree, DAY, ROOT, "session.jsonl") + with open(path) as fh: + records = [json.loads(text) for text in fh] + bad = json.loads(json.dumps(records[9])) + bad["payload"]["record"]["outcome"]["kind"] = [] + bad["stream"]["id"] = "0a000000-0000-7000-8000-0000000000ee" + records.insert(1, bad) + with open(path, "w") as fh: + fh.writelines(json.dumps(r, sort_keys=True) + "\n" for r in records) + result = run(self.db, "sync", "--harness", "muse", "--source", path) + self.assertEqual(result.returncode, 0, result.stderr) + con = sqlite3.connect(self.db) + errors = con.execute("SELECT error, ordinal_num, line_excerpt FROM import_errors").fetchall() + sessions = [r[0] for r in con.execute("SELECT session_key FROM sessions ORDER BY 1")] + source_session = con.execute("SELECT session_id FROM sources").fetchone()[0] + con.close() + self.assertEqual(errors, [("schema_error", 1, self.FRAME_KEYS)]) + self.assertEqual(sessions, [ROOT_KEY, "muse:" + C1, "muse:" + C2]) + self.assertEqual(source_session, ROOT) + self.assertEqual(root_events(self.db), EXPECTED) + + def test_rejected_record_does_not_move_the_span(self): + errors, events = self.edited_import(9, lambda r: ( + r["payload"]["record"]["outcome"].update(kind=[]), r.update(recorded_at=1))) + self.assertEqual(errors, [("schema_error", 9, self.FRAME_KEYS)]) + self.assertEqual(events[("tool_result", "call_r1")][2:4], (None, 7)) + con = sqlite3.connect(self.db) + span = con.execute("SELECT started_at, ended_at FROM sessions WHERE session_key=?", + (ROOT_KEY,)).fetchone() + con.close() + # The valid frames span sequence 1 to 26 at one second each. + self.assertEqual(span, (1792442401.0, 1792442426.0)) + def test_unconsumed_task_frame_stays_silent_even_when_odd(self): # Line 6 is an internal kind=task start: deliberately not consumed. errors, events = self.edited_import(6, lambda r: r["payload"]["event"].update(kind=[])) From 9885e42a780afba0a07c4112f4f50513abc893ee Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Tue, 20 Oct 2026 00:41:03 +0200 Subject: [PATCH 4/5] fix: validate whole Muse batches before accepting them (#127) Call and result batches now validate every item before the record may set the session identity, widen the span or store any item. A batch with any item missing its ids is rejected whole as one missing_id, so a rejected batch leaves no prefix and cannot claim the source. Co-Authored-By: Claude Opus 5.5 --- agent_observer/adapters/muse.py | 31 +++++++++----- docs/contracts.md | 4 +- tests/test_muse.py | 72 +++++++++++++++++++++++++++++++++ 3 files changed, 97 insertions(+), 10 deletions(-) diff --git a/agent_observer/adapters/muse.py b/agent_observer/adapters/muse.py index 91380fb..cd5ac0e 100644 --- a/agent_observer/adapters/muse.py +++ b/agent_observer/adapters/muse.py @@ -278,13 +278,15 @@ def accept(sid, ts): if not isinstance(items, list): bad(ordinal, "schema_error", raw) continue - accept(sid, ts) + # Every item is validated before the record may touch state: + # one bad item rejects the whole batch, leaving no prefix. + calls = [] for item in items: call = _str(item.get("call_id")) if isinstance(item, dict) else None name = _str(item.get("name")) if isinstance(item, dict) else None if call is None or name is None: - bad(ordinal, "missing_id", raw) - continue + calls = None + break args = None if isinstance(item.get("args"), str): try: @@ -292,22 +294,33 @@ def accept(sid, ts): except ValueError: decoded = None args = decoded if isinstance(decoded, dict) else None + calls.append((call, name, args)) + if calls is None: + bad(ordinal, "missing_id", raw) + continue + accept(sid, ts) + for call, name, args in calls: state.calls.setdefault(call, {**at, "name": name, "args": args}) elif kind == "tool_result_batch_committed": items = event.get("results") if not isinstance(items, list): bad(ordinal, "schema_error", raw) continue - accept(sid, ts) + results = [] for item in items: call = _str(item.get("tool_call_id")) if isinstance(item, dict) else None if call is None: - bad(ordinal, "missing_id", raw) - continue + results = None + break text = item.get("text") - state.results[call] = { - **at, "size": len(text.encode("utf-8")) - if isinstance(text, str) else None} + results.append((call, len(text.encode("utf-8")) + if isinstance(text, str) else None)) + if results is None: + bad(ordinal, "missing_id", raw) + continue + accept(sid, ts) + for call, size in results: + state.results[call] = {**at, "size": size} elif kind == "memory_reminder_child_session_linked": child = _session_id(event.get("child_session_id")) if child is None: diff --git a/docs/contracts.md b/docs/contracts.md index d305f4b..1a05962 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -594,7 +594,9 @@ Default root `~/.local/share/muse/sessions`: each effect started and terminal, and the listed `kind=run` events) need `schema_version` and `payload_schema_version` 1, else `unsupported_schema`. A run frame whose `event.kind` is not a string, or a terminal without an - outcome `kind` string, is a `schema_error`; that terminal's result keeps + outcome `kind` string, is a `schema_error`; a call or result batch with any + item lacking its `call_id`/`name` or `tool_call_id` is a `missing_id` and is + rejected whole, keeping none of its items; that terminal's result keeps an unknown status. Each bad record is quarantined alone at its line number with key names only: it neither sets the session identity nor widens the session span, and the rest of the file still imports. Unconsumed diff --git a/tests/test_muse.py b/tests/test_muse.py index 5f98465..793d22a 100644 --- a/tests/test_muse.py +++ b/tests/test_muse.py @@ -306,6 +306,78 @@ def test_rejected_record_does_not_move_the_span(self): # The valid frames span sequence 1 to 26 at one second each. self.assertEqual(span, (1792442401.0, 1792442426.0)) + def inserted_import(self, line, edit): + """Sync the root log with an edited copy of line inserted at line 1.""" + tree = self.copy_tree() + path = os.path.join(tree, DAY, ROOT, "session.jsonl") + with open(path) as fh: + records = [json.loads(text) for text in fh] + bad = json.loads(json.dumps(records[line])) + edit(bad) + records.insert(1, bad) + with open(path, "w") as fh: + fh.writelines(json.dumps(r, sort_keys=True) + "\n" for r in records) + result = run(self.db, "sync", "--harness", "muse", "--source", path) + self.assertEqual(result.returncode, 0, result.stderr) + con = sqlite3.connect(self.db) + state = { + "errors": con.execute("SELECT error, ordinal_num, line_excerpt FROM import_errors").fetchall(), + "sessions": [r[0] for r in con.execute("SELECT session_key FROM sessions ORDER BY 1")], + "source": con.execute("SELECT session_id FROM sources").fetchone()[0], + "span": con.execute("SELECT started_at, ended_at FROM sessions WHERE session_key=?", + (ROOT_KEY,)).fetchone(), + } + con.close() + state["events"] = root_events(self.db) + return state + + # (fixture line, batch list key) for the call batch and the result batch. + BATCHES = ((7, "tool_calls"), (13, "results")) + + def test_rejected_batch_from_another_stream_cannot_claim_the_session(self): + for line, key in self.BATCHES: + with self.subTest(key=key): + self.tearDown() + self.setUp() + got = self.inserted_import(line, lambda r: ( + r["payload"]["event"].update({key: [{}]}), + r["stream"].update(id="0a000000-0000-7000-8000-0000000000ee"))) + self.assertEqual(got["errors"], [("missing_id", 1, self.FRAME_KEYS)]) + self.assertEqual(got["sessions"], [ROOT_KEY, "muse:" + C1, "muse:" + C2]) + self.assertEqual(got["source"], ROOT) + self.assertEqual(got["span"], (1792442401.0, 1792442426.0)) + self.assertEqual(got["events"], EXPECTED) + + def test_rejected_batch_does_not_move_the_span(self): + for line, key in self.BATCHES: + with self.subTest(key=key): + self.tearDown() + self.setUp() + got = self.inserted_import(line, lambda r: ( + r["payload"]["event"].update({key: [{}]}), r.update(recorded_at=1))) + self.assertEqual(got["errors"], [("missing_id", 1, self.FRAME_KEYS)]) + self.assertEqual(got["span"], (1792442401.0, 1792442426.0)) + self.assertEqual(got["events"], EXPECTED) + + def test_mixed_call_batch_is_rejected_whole(self): + # Line 7 commits call_r1 and call_m1; a bad second item rejects both. + errors, events = self.edited_import(7, lambda r: r["payload"]["event"].update( + tool_calls=[r["payload"]["event"]["tool_calls"][0], {"name": "read_file"}])) + self.assertEqual(errors, [("missing_id", 7, self.FRAME_KEYS)]) + expected = {k: v for k, v in EXPECTED.items() if k[1] not in ("call_r1", "call_m1")} + self.assertEqual(events, expected) + + def test_mixed_result_batch_is_rejected_whole(self): + # Line 13 commits results for call_r1 and call_m1; the terminals stay. + errors, events = self.edited_import(13, lambda r: r["payload"]["event"].update( + results=[r["payload"]["event"]["results"][0], {"text": "x"}])) + self.assertEqual(errors, [("missing_id", 13, self.FRAME_KEYS)]) + expected = dict(EXPECTED) + expected[("tool_result", "call_r1")] = ("read_file", None, "error", None, "builtin", None, None) + expected[("tool_result", "call_m1")] = ("mcp__t3_code__web_read", None, "ok", None, + "mcp", "t3-code", "web_read") + self.assertEqual(events, expected) + def test_unconsumed_task_frame_stays_silent_even_when_odd(self): # Line 6 is an internal kind=task start: deliberately not consumed. errors, events = self.edited_import(6, lambda r: r["payload"]["event"].update(kind=[])) From 8288a92d970af2e09e4341ba4cf012b479f9a14c Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Tue, 20 Oct 2026 00:57:14 +0200 Subject: [PATCH 5/5] fix: recheck a Muse source before skipping it as unchanged (#127) Call JsonlSource.recheck_unchanged() immediately before the unchanged fast path, as the Claude, Codex and Grok adapters do, so an append that lands after the initial check is imported in the same sync. Co-Authored-By: Claude Opus 5.5 --- agent_observer/adapters/muse.py | 4 ++- tests/test_muse.py | 47 +++++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/agent_observer/adapters/muse.py b/agent_observer/adapters/muse.py index cd5ac0e..5eaf97a 100644 --- a/agent_observer/adapters/muse.py +++ b/agent_observer/adapters/muse.py @@ -401,7 +401,9 @@ def event(family, native_id, at, **kw): def import_muse_file(con: sqlite3.Connection, path: str, full: bool = False) -> dict: stats = {"events_inserted": 0, "malformed": 0} src = JsonlSource(con, HARNESS, path, full=full) - if src.unchanged: + # Re-stat immediately before the fast path so an append racing the + # first check is imported in this sync (ingest contract). + if src.unchanged and src.recheck_unchanged(): return {"unchanged": True} # A grown file still needs earlier lines: a result or catalog can follow # the call it completes across an append boundary. diff --git a/tests/test_muse.py b/tests/test_muse.py index 793d22a..affb0bb 100644 --- a/tests/test_muse.py +++ b/tests/test_muse.py @@ -161,6 +161,53 @@ def test_unchanged_resync_writes_nothing(self): (2, 0, 0)) self.assertEqual(comparable(self.db), before) + def test_append_after_the_initial_check_is_imported_same_sync(self): + import io + from contextlib import redirect_stdout + from unittest import mock + from agent_observer import cli + from agent_observer.ingest import JsonlSource + tree = self.copy_tree() + path = os.path.join(tree, DAY, ROOT, "session.jsonl") + with open(ROOT_LOG, "rb") as fh: + lines = fh.readlines() + with open(path, "wb") as fh: + fh.writelines(lines[:8]) + + def sync(): + out = io.StringIO() + with redirect_stdout(out): + code = cli.main(["--db", self.db, "sync", "--harness", "muse", + "--source", path, "--json"]) + self.assertEqual(code, 0) + row = next(h for h in json.loads(out.getvalue())["harnesses"] + if h["harness"] == "muse") + return row["unchanged"], row["events_inserted"], row["events_updated"] + + self.assertEqual(sync(), (0, 4, 0)) + self.assertEqual(len(root_events(self.db)), 4) + original = JsonlSource.__init__ + appended = [] + + def raced(src, *args, **kwargs): + # The rest of the log lands right after the real initial check + # found the source unchanged, before the adapter's fast path. + original(src, *args, **kwargs) + if src.path == path and src.unchanged and not appended: + with open(path, "ab") as fh: + fh.writelines(lines[8:]) + appended.append(True) + + with mock.patch.object(JsonlSource, "__init__", raced): + self.assertEqual(sync(), (0, 16, 3)) + self.assertEqual(appended, [True]) + self.assertEqual(root_events(self.db), EXPECTED) + con = sqlite3.connect(self.db) + offsets = con.execute("SELECT read_offset, size_bytes FROM sources").fetchone() + con.close() + self.assertEqual(offsets, (len(b"".join(lines)), len(b"".join(lines)))) + self.assertEqual(sync(), (1, 0, 0)) + def test_split_appends_reach_the_same_literal_result(self): tree = self.copy_tree() path = os.path.join(tree, DAY, ROOT, "session.jsonl")