diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index f713329..b7c3275 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agent-observer", - "version": "1.4.1", + "version": "1.4.2", "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 2b98f8b..24d0aa3 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agent-observer", - "version": "1.4.1", + "version": "1.4.2", "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 e1a9590..5bbb92e 100644 --- a/.grok-plugin/plugin.json +++ b/.grok-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agent-observer", - "version": "1.4.1", + "version": "1.4.2", "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 011697a..d882686 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## [1.4.2] - 2026-10-19 + +### Changed + +- Credit intervention senders, preserve grouped verdicts and retain unknown actor reasons + ## [1.4.1] - 2026-10-19 ### Changed diff --git a/README.md b/README.md index 0908f07..5a81e36 100644 --- a/README.md +++ b/README.md @@ -411,6 +411,45 @@ person. They are never interventions and are counted as `policy_denials`. Claude writes a rejected tool call twice, as the denial and as an interrupt in the same second; that is one intervention. +T3 steer envelopes identify their sender. A steer is one `interrupt`, keyed +by its follow-up prompt; `related_ids` links the native interrupt marker and +tool denials it replaces in this view. The raw messages and events remain. +Delegated-task terminal-state notices count as `agent`, including when the +harness labels their input as user-authored. Unknown actors carry an +`actor_reason`; native import stores the reason without the message body. +`native_sdk_prompt_without_matching_origin` means a Claude SDK prompt has +no matching Chromeria origin. A root thread alone cannot establish its author, +and a provider reference to an assistant UUID cannot identify the user prompt. +The next authorized sync replays older Claude and Codex sources once to +index this evidence; reporting itself remains read-only. + +Grouped ids remain valid: `intervention show` resolves them to their canonical +steer and its follow-up text. `verdict list` retains every stored verdict under +its original id and adds `canonical_intervention`. Recurrence and verdict groups +count one judgment per canonical intervention and judge. A `yes` always wins +over `no` or `unclear`, regardless of timestamp or which id holds it. Within +that priority, the latest judgment wins, then the lexically greatest original +id, with no canonical-id preference. `verdict list` exposes the winning +`canonical_verdict`, `canonical_problem` and `canonical_group` beside every +original row. Recurrence reports displaced records, including their problem, +group and timestamp, in `superseded_verdicts` and names their count in text +output. Stored verdicts are +never rewritten or deleted by this grouping, and an alias verdict also marks +the canonical intervention as judged for `--unjudged`. + +Grok imports the same transport metadata, including first-input +`` envelopes and notices immediately followed by ``. +Its import-version change replays older session metadata once. + +Completion recognition consumes the entire input: only trailing whitespace +and a complete `` block may follow the notice. Pasting a notice +and adding an instruction remains ordinary user input. Grok separates its +complete native runtime/linking/tool-use context before parsing; prose after those +blocks is never discarded. Steers require the complete T3 interruption note, +and explicit native actor evidence takes precedence over steer senders. Complete +terminal notices remain agent inputs even if Codex retains a preceding human +turn id. + ```sh agent-observer interventions --since 2026-09-14 --unjudged correction --json # default --created-by user --created-by agent agent-observer intervention show sub:claude: ev:codex:/lifecycle/turn_aborted: --json diff --git a/VERSION b/VERSION index 347f583..9df886c 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.4.1 +1.4.2 diff --git a/agent_observer/adapters/claude.py b/agent_observer/adapters/claude.py index 3fd7bdb..685341a 100644 --- a/agent_observer/adapters/claude.py +++ b/agent_observer/adapters/claude.py @@ -18,7 +18,7 @@ import os import sqlite3 -from .. import db, privacy +from .. import db, privacy, intervention_evidence from ..identity import SessionIdentity, skill_from_path from ..ingest import (JsonlSource, MissingNativeId, fingerprint, insert_event, iso_ts, text_hash) @@ -105,7 +105,7 @@ def _valid_counter(value) -> bool: _FINALITY_TABLE = "claude_response_finality" _NATIVE_KINDS_TABLE = "claude_native_kinds" -_IMPORT_VERSION = 2 +_IMPORT_VERSION = 4 def _ensure_native_kinds(con) -> None: @@ -333,6 +333,7 @@ def event(self, family, native_id, ordinal, ts, **kw): def import_claude_file(con: sqlite3.Connection, path: str, full: bool = False) -> dict: + intervention_evidence.ensure_table(con) stats = {"lines": 0, "responses_inserted": 0, "responses_duplicate": 0, "responses_updated": 0, "user_messages_inserted": 0, "events_inserted": 0, @@ -716,6 +717,10 @@ def _submission(r: _Reader, native_id, ordinal, ts, kind: str, text: str, r.con.execute(f"INSERT INTO {_NATIVE_KINDS_TABLE}(native_id,kind,origin_allowed) VALUES(?,?,?)" " ON CONFLICT(native_id) DO UPDATE SET kind=excluded.kind,origin_allowed=excluded.origin_allowed", (f"{HARNESS}:{native_id}", native_kind or kind, int(origin_allowed))) + intervention_evidence.record( + r.con, f"{HARNESS}:{native_id}", text, + sdk=r.meta.get('entrypoint') == 'sdk-ts', + allowed=origin_allowed) def _seed_queue_identity(r: _Reader) -> None: diff --git a/agent_observer/adapters/codex.py b/agent_observer/adapters/codex.py index ed9889e..f8c0410 100644 --- a/agent_observer/adapters/codex.py +++ b/agent_observer/adapters/codex.py @@ -752,13 +752,15 @@ def import_codex_file(con: sqlite3.Connection, path: str, Malformed lines are recorded in import_errors and skipped. Previously valid data is never destroyed. """ + from .. import intervention_evidence + intervention_evidence.ensure_table(con) stats = { "lines": 0, "responses_inserted": 0, "responses_duplicate": 0, "user_messages_inserted": 0, "events_inserted": 0, "events_duplicate": 0, "compactions": 0, "malformed": 0, } src = JsonlSource(con, HARNESS, path, full=full) - if (src.row["codex_context_version"] != 3 + if (src.row["codex_context_version"] != 5 or src.row["codex_context_offset"] != src.row["read_offset"]): # Replay older indexes once, including search extensions previously # discarded from otherwise unchanged sources (#94). @@ -958,7 +960,7 @@ def import_codex_file(con: sqlite3.Connection, path: str, thread_source=reader.thread_source)) backfill = 1 if _backfill_complete( con, src.source_id, getattr(src, "last_ordinal", None)) else 0 - con.execute("UPDATE sources SET codex_context_version=3,codex_context_offset=?," + con.execute("UPDATE sources SET codex_context_version=5,codex_context_offset=?," "codex_model=?,codex_effort=?,occurrence_backfill=? WHERE id=?", (src.end_offset, reader.model, reader.effort, backfill, src.source_id)) @@ -1825,6 +1827,9 @@ def _ingest_response_item(r: _Reader, obj: dict, native_ordinal, position: int) values + [native_key]) r.stats["user_messages_updated"] = \ r.stats.get("user_messages_updated", 0) + 1 + if first and (same_source or claims): + from .. import intervention_evidence + intervention_evidence.record(r.con, native_key, text) _note_occurrence(r, "user", native_key, obj) return if ptype == "message" and p.get("role") == "assistant": diff --git a/agent_observer/adapters/grok.py b/agent_observer/adapters/grok.py index a5394cf..0e98c60 100644 --- a/agent_observer/adapters/grok.py +++ b/agent_observer/adapters/grok.py @@ -550,6 +550,7 @@ def _has_any_counter(counters: dict | None) -> bool: _GROK_META_TABLE = "grok_session_meta" +_IMPORT_VERSION = 2 # Replay older transport-envelope parsing once. def _ensure_grok_meta_table(con: sqlite3.Connection) -> None: @@ -590,7 +591,7 @@ def _summary_fingerprint(summary: dict) -> str: canonical = json.dumps(summary, sort_keys=True, default=str) except (TypeError, ValueError): canonical = repr(sorted(summary.keys())) - return hashlib.sha256(canonical.encode("utf-8", "replace")).hexdigest() + return hashlib.sha256(f"{_IMPORT_VERSION}:{canonical}".encode("utf-8", "replace")).hexdigest() def _stored_meta(con: sqlite3.Connection, session_key: str): @@ -1426,6 +1427,8 @@ def import_grok_session(con: sqlite3.Connection, session_dir: str, already imported, using their resolved source spelling. Metadata still comes from session_dir; sharing one file never redirects the other. """ + from .. import intervention_evidence + intervention_evidence.ensure_table(con) stats = {"lines": 0, "responses_inserted": 0, "responses_duplicate": 0, "user_messages_inserted": 0, "events_inserted": 0, "events_duplicate": 0, "compactions": 0, "malformed": 0} @@ -2240,6 +2243,24 @@ def _upsert_submission(con, r: _Reader, src, chat: dict, key: str, stale values. Turn bindings only fill unknowns. """ full_text = "".join(entry["texts"]) + # Strip only the complete native context suffix, never prose after it. + # These sibling blocks are Grok's machine-added runtime context; + # they are not part of the offered user input. + context = re.fullmatch( + r"(.*?)\s*(?:(?!).)*\s*" + r"(?:(?!).)*\s*" + r"(?:(?:(?!).)*\s*)?", + full_text, re.DOTALL) + offered_text = context[1] if context else full_text + # Grok's native first-input envelope contains the actual input in + # user_request, after Chromeria instructions. Do not search arbitrary + # prose for a notice: only unwrap this complete, anchored native shape. + wrapped = re.fullmatch( + r".*?\s*" + r"(.*?)\s*" + r"(?:(?:(?!).)*\s*)?", + offered_text, re.DOTALL) + transport_text = wrapped[1].strip() if wrapped else offered_text is_main = r.is_main_session kind, is_genuine = _classify_prompt(key, chat, is_main) excerpt = privacy.submission_excerpt( @@ -2261,6 +2282,8 @@ def _upsert_submission(con, r: _Reader, src, chat: dict, key: str, digest, excerpt, is_genuine)) if cur.rowcount: r.stats["user_messages_inserted"] += 1 + from .. import intervention_evidence + intervention_evidence.record(con, native_id, transport_text, sdk=True, allowed=is_main) return if digest != existing["text_hash"] \ or (existing["text_excerpt"] or "") != excerpt \ @@ -2275,6 +2298,8 @@ def _upsert_submission(con, r: _Reader, src, chat: dict, key: str, if turn_id and not existing["turn_id"]: con.execute("UPDATE user_messages SET turn_id=? WHERE native_id=?", (turn_id, native_id)) + from .. import intervention_evidence + intervention_evidence.record(con, native_id, transport_text, sdk=True, allowed=is_main) def _reclassify_existing(con, r: _Reader, session_dir: str, diff --git a/agent_observer/adapters/t3.py b/agent_observer/adapters/t3.py index 15d88b7..ce09a6a 100644 --- a/agent_observer/adapters/t3.py +++ b/agent_observer/adapters/t3.py @@ -360,6 +360,8 @@ def sync(con: sqlite3.Connection, root: str | None = None, _import_messages(con, native, totals, lineage) _import_prompt_origins(con, native, totals) totals["user_messages_reclassified"] = _backfill_prompt_origins(con) + from .. import intervention_evidence + intervention_evidence.backfill_root_reasons(con) totals["responses_retouched"] = _retouch_response_turns(con) bound = _bind_trees(con, threads, links, lineage) totals["tasks"] = bound["tasks"] diff --git a/agent_observer/cli.py b/agent_observer/cli.py index 62f0470..1aa43ee 100644 --- a/agent_observer/cli.py +++ b/agent_observer/cli.py @@ -1149,7 +1149,8 @@ def _judge(con, ns) -> int: + ", ".join(f"{k} {v}" for k, v in payload["policy_denials"].items())) for m in payload["interventions"]: print(f"{m['id']} {m['kind']} {m['created_by']} {m['at']} {m['project'] or ''}" - f" opening={json.dumps(m['opening'])}") + f" opening={json.dumps(m['opening'])}" + + (f" actor_reason={m['actor_reason']}" if m.get('actor_reason') else '')) return 0 if ns.cmd == "intervention": ids = list(ns.ids) @@ -1180,7 +1181,11 @@ def _judge(con, ns) -> int: else: for v in listed: print(f"{v['intervention']} {v['verdict']} {v['group'] or '-'}" - f" {v['problem'] or ''}") + f" {v['problem'] or ''}" + + (f" canonical={v['canonical_intervention']}" + if v['canonical_intervention'] not in (None, v['intervention']) else '') + + (f" canonical_verdict={v['canonical_verdict']}" + if v['canonical_verdict'] not in (None, v['verdict']) else '')) return 0 if ns.cmd == "verdict": groups = _verdicts.groups(con, ns.behavior, judge=ns.judge, diff --git a/agent_observer/intervention_evidence.py b/agent_observer/intervention_evidence.py new file mode 100644 index 0000000..97bccd6 --- /dev/null +++ b/agent_observer/intervention_evidence.py @@ -0,0 +1,67 @@ +"""Closed transport metadata from native inputs, never their private bodies.""" + +import re + + +_STEER = re.compile( + r"\ANote from T3 Code: the (user|agent in T3 thread [^\s()]+) " + r"interrupted your previous turn, if it was still running, to deliver the message below\. " + r"Tool calls stopped for this message were interrupted, not refused, even if their results " + r"say the user declined them or does not want them\. " + r"Follow the message below before deciding whether to retry a stopped action\. " + r"Stopped actions may have partly run, other running work may have stopped too, " + r"and actions that finished keep their effects\.\n\n[^\s].*\Z", re.DOTALL) +_RUNTIME_INFO = r"(?:(?!).)*" +_RUNTIME_SUFFIX = r"\s*(?:" + _RUNTIME_INFO + r"\s*)?\Z" +_COMPLETION = re.compile( + r"\ADelegated task (?P[^\s]+) reached a terminal state\. " + r"Use task_status with taskId (?P=id) to read the result\." + _RUNTIME_SUFFIX, re.DOTALL) +_COMPLETIONS = re.compile( + r"\ADelegated tasks [^\n]+ reached terminal states\. " + r"Use task_status with each taskId to read the results\." + _RUNTIME_SUFFIX, re.DOTALL) + + +def ensure_table(con): + con.execute( + "CREATE TABLE IF NOT EXISTS intervention_prompt_evidence(" + " native_id TEXT PRIMARY KEY REFERENCES user_messages(native_id)" + " ON UPDATE CASCADE ON DELETE CASCADE," + " transport_kind TEXT CHECK(transport_kind IN ('steer','completion'))," + " actor_kind TEXT CHECK(actor_kind IN ('client','server'))," + " actor_reason TEXT NOT NULL)") + + +def record(con, native_id, text, *, sdk=False, allowed=True): + """Parse only anchored transport envelopes, excluding Claude scaffolding/children. + + The envelope is sender evidence even when a native provider mislabeled the + input as user-authored. Ordinary SDK inputs need a matching origin, never + an inference from the thread opener or a preceding conversational turn. + """ + kind, actor = None, None + reason = ('native_sdk_prompt_without_matching_origin' if native_id.startswith('claude:') + else 'native_prompt_without_matching_origin') if sdk else 'actor_evidence_missing' + if allowed: + steer = _STEER.match(text) if sdk else None + if steer: + kind = 'steer' + actor = 'client' if steer[1] == 'user' else 'server' + reason = 't3_steer_sender' + elif _COMPLETION.match(text) or _COMPLETIONS.match(text): + kind, actor, reason = 'completion', 'server', 'delegated_task_notice' + con.execute( + "INSERT INTO intervention_prompt_evidence VALUES(?,?,?,?)" + " ON CONFLICT(native_id) DO UPDATE SET transport_kind=excluded.transport_kind," + " actor_kind=excluded.actor_kind,actor_reason=excluded.actor_reason", + (native_id, kind, actor, reason)) + + +def backfill_root_reasons(con): + """Keep a stored gap even for harnesses without a transport-envelope parser.""" + ensure_table(con) + con.execute( + "INSERT OR IGNORE INTO intervention_prompt_evidence" + " SELECT u.native_id,NULL,NULL,'native_prompt_without_matching_origin'" + " FROM user_messages u JOIN sessions s ON s.session_key=u.session_key" + " JOIN t3_threads t ON t.native_session=s.native_id" + " WHERE t.thread_id=t.root_thread_id AND t.opened_by='client'") diff --git a/agent_observer/interventions.py b/agent_observer/interventions.py index 5a2037c..46d235c 100644 --- a/agent_observer/interventions.py +++ b/agent_observer/interventions.py @@ -77,6 +77,12 @@ def __init__(self, con: sqlite3.Connection): con, "SELECT session_key, native_id, actor_kind FROM t3_native_prompt_origins")} self.restricted_origins = {r[0] for r in self._rows( con, "SELECT native_id FROM claude_native_kinds WHERE origin_allowed=0")} + self.native = {r[0]: (r[1], r[2], r[3]) for r in self._rows( + con, "SELECT native_id,transport_kind,actor_kind,actor_reason" + " FROM intervention_prompt_evidence")} + self.related: dict[str, list[str]] = {} + self.grouped: dict[str, str] = {} + self._group_steers(con) # Thread position: root (a top-level thread a person opened), # launched (top-level, opened by an agent or without evidence), or # child. A session seen in several threads takes the least @@ -96,6 +102,34 @@ def __init__(self, con: sqlite3.Connection): if rank[position] >= rank.get(self.threads.get(session), -1): self.threads[session] = position + def _group_steers(self, con): + # The next substantive native input after Claude's interrupt marker + # must be an explicit T3 steer. Same-source positions establish the + # link; proximity alone must never turn a real refusal into a steer. + previous = {} + for row in con.execute( + "SELECT native_id,source_id,session_key,ordinal_num,kind,ts" + " FROM user_messages WHERE kind NOT IN ('scaffolding','command')" + " ORDER BY source_id,ordinal_num"): + prior = previous.get(row['source_id']) + previous[row['source_id']] = row + if self.native.get(row['native_id'], (None,))[0] != 'steer' or prior is None \ + or row['source_id'] is None or row['ordinal_num'] is None \ + or prior['ordinal_num'] is None or prior['kind'] != 'interrupt' \ + or prior['session_key'] != row['session_key']: + continue + canonical = 'sub:' + row['native_id'] + related = ['sub:' + prior['native_id']] + for event in con.execute( + "SELECT session_key,native_id FROM events WHERE source_id=?" + " AND session_key=? AND family='permission' AND status IN ('denied','deny')" + " AND json_extract(detail_json,'$.denial')='user-rejected'" + " AND ? - ts BETWEEN 0 AND 1.0 AND ordinal_num < ?", + (row['source_id'], row['session_key'], prior['ts'], prior['ordinal_num'])): + related.append(f"ev:{event['session_key']}/permission/{event['native_id']}") + self.related[canonical] = related + self.grouped.update((identity, canonical) for identity in related) + @staticmethod def _rows(con, sql) -> list: try: @@ -173,31 +207,44 @@ def _scope(harness=None, project=None, attrs=None) -> tuple[str, list]: def _base(row, intervention_id: str, kind: str, ts, created_by: str, thread: str) -> dict: - return {"id": intervention_id, "kind": kind, "created_by": created_by, "thread": thread, - "ts": ts, "at": _iso(ts), "session": row["session_key"], - "harness": row["harness"], "project": row["project_dir"]} + out = {"id": intervention_id, "kind": kind, "created_by": created_by, "thread": thread, + "ts": ts, "at": _iso(ts), "session": row["session_key"], + "harness": row["harness"], "project": row["project_dir"]} + if created_by == 'unknown': + out['actor_reason'] = 'actor_evidence_missing' + return out def _from_submission(row, ev: _Evidence) -> dict | None: """An intervention for a genuine prompt, an interrupt, or a message another agent sent (a synthetic user message Chromeria records as agent-sent); None for other synthetic input such as reminders.""" - kind = "interrupt" if row["kind"] == "interrupt" else "prompt" + transport, sender, reason = ev.native.get( + row['native_id'], (None, None, 'origin_evidence_not_imported')) + kind = "interrupt" if row["kind"] == "interrupt" or transport == 'steer' else "prompt" thread = ev.thread(row["session_native"]) identity = row["native_id"] if row["harness"] == "claude": from .adapters.claude import prompt_native_id identity = prompt_native_id(row) actor = ev.prompt_actor(row["harness"], identity, - row["session_native"], row["turn_id"]) \ - if kind == "prompt" else None + row["session_native"], row["turn_id"]) or sender \ + if kind == "prompt" or transport == 'steer' else None + if transport == 'completion': + # A complete notice contains no user instruction. Codex can attach + # its preceding human turn id to this separate system input. + actor = sender if row["kind"] == "synthetic" and ( actor != "server" or ev.origin_restricted(row["harness"], row["native_id"])): return None created_by = "unknown" if row["kind"] == "unknown" and not actor \ else _created_by(kind, actor, row["entrypoint"], thread) - return _base(row, f"sub:{row['native_id']}", kind, row["ts"], - created_by, thread) + out = _base(row, f"sub:{row['native_id']}", kind, row["ts"], created_by, thread) + if created_by == 'unknown': + out['actor_reason'] = reason + if out['id'] in ev.related: + out['related_ids'] = ev.related[out['id']] + return out def _denial(row): @@ -271,12 +318,14 @@ def _all(con, harness=None, project=None, attrs=None): first = _first_activity(con) for row in con.execute(f"{_SUB_SQL} WHERE u.kind IN {_SUB_KINDS}" f" AND {where}", args): - intervention = None if row["native_id"] in echoes \ + intervention = None if row["native_id"] in echoes or 'sub:' + row['native_id'] in ev.grouped \ else _from_submission(row, ev) if intervention is not None: found.append(_mark_opening(intervention, first, row)) for row in con.execute(f"{_EVENT_SQL} WHERE {_EVENT_WHERE} AND {where}", args): + if f"ev:{row['session_key']}/{row['family']}/{row['native_id']}" in ev.grouped: + continue kind = _event_kind(row) if kind is None: continue @@ -346,7 +395,8 @@ def judged_ids(con, behavior: str, judge=None) -> set: if judge: sql += " AND judge=?" args.append(judge) - return {r[0] for r in con.execute(sql, args)} + ev = _Evidence(con) + return {ev.grouped.get(r[0], r[0]) for r in con.execute(sql, args)} def select(con: sqlite3.Connection, since=None, until=None, harness=None, @@ -389,6 +439,7 @@ def resolve(con: sqlite3.Connection, intervention_id, ev=None) -> dict | None: if not isinstance(intervention_id, str): return None ev = ev or _Evidence(con) + intervention_id = ev.grouped.get(intervention_id, intervention_id) if intervention_id.startswith("sub:"): row = con.execute(f"{_SUB_SQL} WHERE u.native_id = ? AND u.kind IN" f" {_SUB_KINDS}", (intervention_id[4:],)).fetchone() @@ -590,7 +641,7 @@ def _timeline(con, session_key: str, ev: _Evidence) -> list: for row in con.execute( f"{_SUB_SQL} WHERE u.session_key = ? AND u.kind IN {_SUB_KINDS}", (session_key,)): - intervention = None if row["native_id"] in echoes \ + intervention = None if row["native_id"] in echoes or 'sub:' + row['native_id'] in ev.grouped \ else _from_submission(row, ev) if intervention is None: continue @@ -756,6 +807,7 @@ def show(con: sqlite3.Connection, intervention_ids: list, if intervention is None: out.append({"id": intervention_id, "status": "no_such_intervention"}) continue + intervention_id = intervention['id'] session = intervention["session"] if session not in timelines: timelines[session] = _timeline(con, session, ev) @@ -822,6 +874,8 @@ def render(items: list[dict]) -> str: lines.append(f"{item['kind']} by {item['created_by']} in {item['session']}" f" ({item['harness']}, thread {item['thread']}) at" f" {item['at']} opening={json.dumps(item['opening'])}") + if item.get('actor_reason'): + lines.append(f"actor_reason={item['actor_reason']}") for turn in item.get("before", []): if turn["prompt"]: lines.append(_render_prompt(turn["prompt"])) diff --git a/agent_observer/verdicts.py b/agent_observer/verdicts.py index deaa7cb..1084692 100644 --- a/agent_observer/verdicts.py +++ b/agent_observer/verdicts.py @@ -164,7 +164,7 @@ def add_session_attributes(con: sqlite3.Connection, records: list) -> dict: def _rows(con, behavior: str, judge=None) -> list: - sql = ("SELECT intervention_id, judge, verdict, problem, group_id FROM verdicts" + sql = ("SELECT intervention_id, judge, verdict, problem, group_id, recorded_at FROM verdicts" " WHERE behavior=?") args = [behavior] if judge: @@ -173,6 +173,38 @@ def _rows(con, behavior: str, judge=None) -> list: return con.execute(sql, args).fetchall() +def _aliases(scope): + return {identity: m['id'] for m in scope.values() + for identity in m.get('related_ids', [])} + + +def _canonical_rows(con, behavior, judge, scope): + """One current judgment per canonical intervention and judge, with provenance. + + Stored rows are never rewritten. A yes always wins over a no or unclear. + Within that priority, latest recorded_at wins, then lexical id, without a + canonical-id preference. Every displaced row retains its audit details. + """ + aliases = _aliases(scope) + cohorts = {} + for r in _rows(con, behavior, judge): + canonical = aliases.get(r['intervention_id'], r['intervention_id']) + if canonical in scope: + cohorts.setdefault((canonical, r['judge']), []).append(r) + current, superseded = [], [] + for (canonical, name), rows in cohorts.items(): + rows.sort(key=lambda r: (r['verdict'] == 'yes', r['recorded_at'], r['intervention_id'])) + winner = dict(rows[-1]) + winner['intervention_id'] = canonical + current.append(winner) + superseded.extend({'intervention': r['intervention_id'], + 'canonical_intervention': canonical, 'judge': name, + 'verdict': r['verdict'], 'problem': r['problem'], + 'group': r['group_id'], 'recorded_at': r['recorded_at']} + for r in rows[:-1]) + return current, superseded + + def _group_entries(rows, scope: dict) -> list: """Groups of yes verdicts, largest first, from interventions in scope.""" groups: dict[str, list] = {} @@ -210,7 +242,8 @@ def groups(con: sqlite3.Connection, behavior: str, judge=None, out = [] judges = sorted({r["judge"] for r in _rows(con, behavior, judge)}) for name in judges: - for entry in _group_entries(_rows(con, behavior, name), scope): + rows, _ = _canonical_rows(con, behavior, name, scope) + for entry in _group_entries(rows, scope): entry["judge"] = name out.append(entry) return out @@ -222,12 +255,21 @@ def listing(con: sqlite3.Connection, behavior: str, judge=None, and time, oldest first; the grouping pass reads its problem lines here.""" found, _ = _interventions._all(con) scope = {m["id"]: m for m in found} + aliases = _aliases(scope) + current, _ = _canonical_rows(con, behavior, judge, scope) + canonical_verdicts = {(r['intervention_id'], r['judge']): r for r in current} out = [] for r in _rows(con, behavior, judge): if verdict and r["verdict"] != verdict: continue - intervention = scope.get(r["intervention_id"]) + canonical = aliases.get(r['intervention_id'], r['intervention_id']) + intervention = scope.get(canonical) + chosen = canonical_verdicts.get((canonical, r['judge'])) out.append({"intervention": r["intervention_id"], "judge": r["judge"], + "canonical_intervention": canonical if intervention else None, + "canonical_verdict": chosen['verdict'] if chosen else None, + "canonical_problem": chosen['problem'] if chosen else None, + "canonical_group": chosen['group_id'] if chosen else None, "verdict": r["verdict"], "problem": r["problem"], "group": r["group_id"], "created_by": intervention["created_by"] if intervention else None, @@ -248,7 +290,7 @@ def recurrence(con: sqlite3.Connection, behavior: str, judge=None, harness=harness, project=project, created_by=created_by, attrs=attrs) scope = {m["id"]: m for m in selected["interventions"]} - rows = [r for r in _rows(con, behavior, judge) if r["intervention_id"] in scope] + rows, superseded = _canonical_rows(con, behavior, judge, scope) judges = sorted({r["judge"] for r in rows}) if judge is None and len(judges) > 1: raise VerdictError( @@ -282,7 +324,8 @@ def recurrence(con: sqlite3.Connection, behavior: str, judge=None, every = {m["id"]: m for m in _interventions.select( con, since=since, until=until, harness=harness, project=project, created_by=None, attrs=attrs)["interventions"]} - for r in _rows(con, behavior, chosen): + other_rows, _ = _canonical_rows(con, behavior, chosen, every) + for r in other_rows: intervention = every.get(r["intervention_id"]) if intervention is None or intervention["created_by"] in created_by \ or r["verdict"] != "yes": @@ -304,6 +347,7 @@ def recurrence(con: sqlite3.Connection, behavior: str, judge=None, "by_created_by": selected["counts"]["by_created_by"], "unjudged_by_created_by": unjudged}, "verdicts": verdict_counts, + "superseded_verdicts": [r for r in superseded if r['judge'] == chosen], "occurrences": verdict_counts["yes"], "ungrouped": sum(1 for r in rows if r["verdict"] == "yes" and not r["group_id"]), @@ -340,6 +384,9 @@ def render(report: dict) -> str: for origin, entry in sorted(report.get("other_created_by", {}).items()): lines.append(f"also {entry['occurrences']} yes on {origin} interventions" f" ({entry['problems']} problems), not counted above") + if report.get('superseded_verdicts'): + lines.append(f"{len(report['superseded_verdicts'])} stored verdicts superseded by" + " current canonical judgments; verdict list retains them") for g in report["groups"]: lines.append(f"- {g['group']}: {g['occurrences']}x in {g['sessions']}" f" sessions ({', '.join(g['hosts'])}), {g['first_seen']}" diff --git a/tests/test_intervention_actors.py b/tests/test_intervention_actors.py new file mode 100644 index 0000000..4cdc093 --- /dev/null +++ b/tests/test_intervention_actors.py @@ -0,0 +1,338 @@ +"""Sanitized shapes from #133's real Claude steers and Codex notices. + +Only transport boilerplate survives; ids, commands and message bodies are fake. +""" +import json +import io +import os +import sqlite3 +import tempfile +import unittest +from unittest import mock +from contextlib import redirect_stdout, closing + +from agent_observer import db, interventions, intervention_evidence, cli, verdicts +from agent_observer.adapters import claude, codex, grok, t3 +from tests.test_interventions import _line, _tool_use, _tool_result, _write +from tests.test_t3_v2 import write_v2, ROOT, CL2, P1 + + +NOTE = ("Note from T3 Code: the {sender} interrupted your previous turn, if it was still " + "running, to deliver the message below. Tool calls stopped for this message were " + "interrupted, not refused, even if their results say the user declined them or " + "does not want them. Follow the message below before deciding whether to retry " + "a stopped action. Stopped actions may have partly run, other running work may " + "have stopped too, and actions that finished keep their effects.\n\nSanitized message.") + + +class ActorRegressionTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.con = db.connect(os.path.join(self.tmp.name, 'ledger.db')) + self.addCleanup(self.con.close) + db.init_db(self.con) + t3._ensure_tables(self.con) + + def test_real_steer_sequences_count_once_and_keep_raw_evidence(self): + for sender, actor in [('user', 'user'), ('agent in T3 thread sender-1', 'agent')]: + with self.subTest(sender=sender): + session = actor + path = os.path.join(self.tmp.name, session + '.jsonl') + records = [ + _line(session, 'assistant', 'a-'+actor, + _tool_use('call-'+actor, 'Bash', {'command': 'true'}), 0, + entrypoint='sdk-ts'), + _line(session, 'user', 'r-'+actor, + _tool_result('call-'+actor, 'Stopped', True), 1, + entrypoint='sdk-ts', toolDenialKind='user-rejected'), + _line(session, 'user', 'echo-'+actor, + '[Request interrupted by user for tool use]', 1, + entrypoint='sdk-ts'), + _line(session, 'user', 'steer-'+actor, NOTE.format(sender=sender), 5, + entrypoint='sdk-ts', promptSource='sdk'), + ] + _write(path, records) + for full in (False, True, False): + claude.import_claude_file(self.con, path, full=full) + rows = interventions.select(self.con, created_by=None)['interventions'] + rows = [r for r in rows if r['session'] == 'claude:'+session] + self.assertEqual(len(rows), 1) + [row] = rows + self.assertEqual((row['id'], row['kind'], row['created_by']), + ('sub:claude:steer-'+actor, 'interrupt', actor)) + self.assertIn('ev:claude:'+session+'/permission/call-'+actor, + row['related_ids']) + self.assertEqual(interventions.resolve(self.con, row['id']), row) + for identity in row['related_ids']: + self.assertEqual(interventions.resolve(self.con, identity)['id'], row['id']) + [alias] = interventions.show(self.con, [identity]) + self.assertEqual((alias['id'], alias['status']), (row['id'], 'ok')) + [shown] = interventions.show(self.con, [row['id']]) + self.assertEqual(shown['status'], 'ok') + self.assertIn('Sanitized message.', shown['text']) + self.assertEqual(self.con.execute( + 'SELECT count(*) FROM events WHERE session_key=? AND family="permission"', + ('claude:'+session,)).fetchone()[0], 1) + + def test_completion_notice_is_agent_even_with_human_native_marker(self): + path = os.path.join(self.tmp.name, 'codex.jsonl') + text = ('Delegated task node:delegated-task:fixture reached a terminal state. ' + 'Use task_status with taskId node:delegated-task:fixture to read the result.') + _write(path, [ + {'type': 'session_meta', 'payload': {'id': 'cx', 'originator': 'codex-tui'}}, + {'type': 'response_item', 'payload': {'type': 'message', 'role': 'user', + 'id': 'notice', 'content': [{'type': 'input_text', 'text': text}], + 'internal_chat_message_metadata_passthrough': {'content_item_kinds': ['user.text']}}}, + ]) + for full in (False, True, False): + codex.import_codex_file(self.con, path, full=full) + row = interventions.resolve(self.con, 'sub:codex:notice') + self.assertEqual(row['created_by'], 'agent') + + # Native notices can inherit a preceding human conversational turn. + self.con.execute("UPDATE user_messages SET turn_id='codex:previous-human'" + " WHERE native_id='codex:notice'") + self.con.execute("INSERT INTO t3_native_prompt_origins" + "(session_key,native_id,actor_kind) VALUES(?,?,?)", + ('codex:cx', 'codex:previous-human', 'client')) + self.assertEqual(interventions.resolve(self.con, 'sub:codex:notice') + ['created_by'], 'agent') + + def test_pasted_notice_with_new_instruction_stays_user(self): + notice = ('Delegated task node:fixture reached a terminal state. ' + 'Use task_status with taskId node:fixture to read the result.') + for suffix in ('\nNow implement my request.', + 'Context\nNow implement my request.', + 'Context\nMy words'): + with self.subTest(suffix=suffix): + path = os.path.join(self.tmp.name, 'pasted.jsonl') + _write(path, [ + {'type': 'session_meta', 'payload': {'id': 'pasted', 'originator': 'codex-tui'}}, + {'type': 'response_item', 'payload': {'type': 'message', 'role': 'user', + 'id': 'pasted', 'content': [{'type': 'input_text', 'text': notice+suffix}], + 'internal_chat_message_metadata_passthrough': { + 'content_item_kinds': ['user.text']}}}]) + codex.import_codex_file(self.con, path, full=True) + self.assertEqual(interventions.resolve(self.con, 'sub:codex:pasted') + ['created_by'], 'user') + + def test_partial_or_human_authored_agent_steer_does_not_claim_agent(self): + path = os.path.join(self.tmp.name, 'pasted-steer.jsonl') + header = NOTE.format(sender='agent in T3 thread sender-1') + partial = header.partition(' Follow the message below')[0]+'\nMy own instruction.' + _write(path, [ + _line('pasted-steer', 'user', 'partial', partial, 1, + entrypoint='sdk-ts', promptSource='sdk'), + _line('pasted-steer', 'user', 'human', header, 2, + entrypoint='sdk-ts', promptSource='sdk')]) + for identity in ('partial', 'human'): + self.con.execute('INSERT INTO t3_turn_origins VALUES(?,?,?,?)', + (identity, 'root', 'client', None)) + claude.import_claude_file(self.con, path) + for identity in ('partial', 'human'): + self.assertEqual(interventions.resolve(self.con, 'sub:claude:'+identity) + ['created_by'], 'user') + self.assertIsNone(self.con.execute('SELECT transport_kind FROM intervention_prompt_evidence ' + 'WHERE native_id=?', ('claude:partial',)).fetchone()[0]) + + def test_unknown_sdk_root_keeps_stored_reason_without_guessing_from_root(self): + path = os.path.join(self.tmp.name, 'unknown.jsonl') + _write(path, [_line('root-session', 'user', 'unmatched', 'Sanitized input', 0, + entrypoint='sdk-ts', promptSource='sdk')]) + self.con.execute('INSERT INTO t3_threads VALUES(?,?,?,?,?,?)', + ('root', 'root', 'claude', None, 'root-session', 'client')) + claude.import_claude_file(self.con, path) + row = interventions.resolve(self.con, 'sub:claude:unmatched') + self.assertEqual((row['created_by'], row['thread']), ('unknown', 'root')) + self.assertEqual(row['actor_reason'], 'native_sdk_prompt_without_matching_origin') + # Reporting must not rely on rereading a transcript or writing the ledger. + os.remove(path) + self.assertEqual(interventions.resolve(self.con, row['id'])['actor_reason'], + row['actor_reason']) + stored = self.con.execute('SELECT actor_reason FROM intervention_prompt_evidence ' + 'WHERE native_id=?', ('claude:unmatched',)).fetchone() + self.assertEqual(stored[0], row['actor_reason']) + self.con.commit() + output = io.StringIO() + with redirect_stdout(output): + result = cli.main(['--db', os.path.join(self.tmp.name, 'ledger.db'), + 'interventions', '--created-by', 'unknown']) + self.assertEqual(result, 0) + self.assertIn('actor_reason='+row['actor_reason'], output.getvalue()) + + def test_weak_assistant_reference_cannot_credit_a_root_prompt(self): + # Real root-thread shape: Chromeria points to its last assistant UUID, + # not the native user UUID. Matching text or the opener cannot repair it. + state = os.path.join(self.tmp.name, 't3.sqlite') + write_v2(state) + with closing(sqlite3.connect(state)) as native: + native.execute('DELETE FROM orchestration_v2_projection_provider_threads ' + 'WHERE thread_id=? AND provider=?', (ROOT, 'codex')) + native.execute( + 'UPDATE orchestration_v2_projection_provider_turns SET payload_json=? ' + 'WHERE run_attempt_id=?', + (json.dumps({'nativeTurnRef': {'driver': 'claudeAgent', + 'nativeId': 'assistant-uuid', 'strength': 'weak'}}), 'attempt:vm1')) + native.commit() + path = os.path.join(self.tmp.name, 'weak.jsonl') + _write(path, [ + _line(CL2, 'user', P1, 'Sanitized input', 0, + entrypoint='sdk-ts', promptSource='sdk'), + _line(CL2, 'assistant', 'assistant-uuid', 'Sanitized response', 1, + entrypoint='sdk-ts')]) + claude.import_claude_file(self.con, path) + t3.sync(self.con, source=state) + row = interventions.resolve(self.con, 'sub:claude:'+P1) + self.assertEqual((row['created_by'], row['thread']), ('unknown', 'root')) + self.assertEqual(row['actor_reason'], 'native_sdk_prompt_without_matching_origin') + + def test_unrelated_refusal_before_an_ordinary_prompt_stays_a_rejection(self): + path = os.path.join(self.tmp.name, 'refusal.jsonl') + _write(path, [ + _line('plain', 'user', 'rejected', _tool_result('call', 'No', True), 1, + toolDenialKind='user-rejected'), + _line('plain', 'user', 'echo', '[Request interrupted by user for tool use]', 1), + _line('plain', 'user', 'next', 'Ordinary input', 5)]) + claude.import_claude_file(self.con, path) + rows = interventions.select(self.con, created_by=('user',))['interventions'] + self.assertEqual({r['kind'] for r in rows}, {'rejection', 'prompt'}) + + def test_quoted_or_typed_transport_text_does_not_claim_a_steer(self): + path = os.path.join(self.tmp.name, 'typed.jsonl') + _write(path, [ + _line('typed', 'user', 'typed', NOTE.format(sender='user'), 0), + _line('typed', 'user', 'quoted', 'Quote:\n'+NOTE.format(sender='user'), 1, + entrypoint='sdk-ts', promptSource='sdk')]) + claude.import_claude_file(self.con, path) + for identity in ('typed', 'quoted'): + self.assertEqual(interventions.resolve(self.con, 'sub:claude:'+identity)['kind'], + 'prompt') + + def test_root_mirror_backfills_a_reason_for_an_unparsed_harness(self): + self.test_unknown_sdk_root_keeps_stored_reason_without_guessing_from_root() + self.con.execute('DELETE FROM intervention_prompt_evidence') + intervention_evidence.backfill_root_reasons(self.con) + row = interventions.resolve(self.con, 'sub:claude:unmatched') + self.assertEqual(row['actor_reason'], 'native_prompt_without_matching_origin') + self.assertEqual(self.con.execute('SELECT actor_reason FROM intervention_prompt_evidence') + .fetchone()[0], row['actor_reason']) + + def test_stored_grouped_verdicts_resolve_show_and_recur_once(self): + path = os.path.join(self.tmp.name, 'stored.jsonl') + records = [ + _line('stored', 'user', 'echo', '[Request interrupted by user]', 1, + entrypoint='sdk-ts'), + ] + _write(path, records) + self.con.execute('INSERT INTO t3_threads VALUES(?,?,?,?,?,?)', + ('root', 'root', 'claude', None, 'stored', 'client')) + claude.import_claude_file(self.con, path) + verdicts.add(self.con, [{'intervention': 'sub:claude:echo', 'behavior': 'correction', + 'judge': 'spectrum-review', 'verdict': 'yes', 'problem': 'Sanitized problem', + 'group': 'fixture'}]) + records.append(_line('stored', 'user', 'steer', NOTE.format(sender='user'), 5, + entrypoint='sdk-ts', promptSource='sdk')) + _write(path, records) + claude.import_claude_file(self.con, path) + canonical = 'sub:claude:steer' + self.assertEqual(interventions.resolve(self.con, 'sub:claude:echo')['id'], canonical) + [shown] = interventions.show(self.con, ['sub:claude:echo']) + self.assertEqual((shown['id'], shown['status']), (canonical, 'ok')) + self.assertIn('Sanitized message.', shown['text']) + [listed] = verdicts.listing(self.con, 'correction') + self.assertEqual(listed['intervention'], 'sub:claude:echo') + self.assertEqual(listed['canonical_intervention'], canonical) + self.assertEqual(listed['session'], 'claude:stored') + result = verdicts.recurrence(self.con, 'correction') + self.assertEqual((result['occurrences'], result['coverage']['judged']), (1, 1)) + self.assertEqual(result['groups'][0]['interventions'], [canonical]) + self.assertEqual(interventions.select(self.con, unjudged='correction')['interventions'], []) + # The real spectrum-review batch also judged the follow-up itself. + # Equal-time duplicate yes verdicts still describe one intervention. + verdicts.add(self.con, [{'intervention': canonical, 'behavior': 'correction', + 'judge': 'spectrum-review', 'verdict': 'yes', 'problem': 'Sanitized problem', + 'group': 'fixture'}]) + self.con.execute('UPDATE verdicts SET recorded_at=1') + result = verdicts.recurrence(self.con, 'correction') + self.assertEqual((result['occurrences'], result['coverage']['judged']), (1, 1)) + self.assertEqual(result['superseded_verdicts'][0]['intervention'], 'sub:claude:echo') + self.assertEqual(result['superseded_verdicts'][0]['problem'], 'Sanitized problem') + self.assertEqual(result['superseded_verdicts'][0]['group'], 'fixture') + # A canonical no cannot hide the alias yes, whatever their timestamps. + verdicts.add(self.con, [{'intervention': canonical, 'behavior': 'correction', + 'judge': 'spectrum-review', 'verdict': 'no'}]) + self.assertEqual(len(verdicts.listing(self.con, 'correction')), 2) + for alias_time, canonical_time in ((1, 1), (1, 2), (2, 1)): + with self.subTest(alias_time=alias_time, canonical_time=canonical_time): + self.con.execute('UPDATE verdicts SET recorded_at=? WHERE intervention_id=?', + (alias_time, 'sub:claude:echo')) + self.con.execute('UPDATE verdicts SET recorded_at=? WHERE intervention_id=?', + (canonical_time, canonical)) + result = verdicts.recurrence(self.con, 'correction') + self.assertEqual(result['verdicts'], {'yes': 1, 'no': 0, 'unclear': 0}) + self.assertEqual(verdicts.groups(self.con, 'correction')[0]['occurrences'], 1) + for row in verdicts.listing(self.con, 'correction'): + self.assertEqual(row['canonical_verdict'], 'yes') + self.assertEqual(row['canonical_problem'], 'Sanitized problem') + self.assertEqual(row['canonical_group'], 'fixture') + [superseded] = result['superseded_verdicts'] + self.assertEqual(superseded['intervention'], canonical) + self.assertIn('problem', superseded) + self.assertIn('group', superseded) + + def test_grok_records_agent_steer_and_completion_metadata(self): + path = os.path.join(self.tmp.name, 'grok-session') + texts = [NOTE.format(sender='agent in T3 thread sender-1'), + 'Delegated task node:fixture reached a terminal state. ' + 'Use task_status with taskId node:fixture to read the result.' + 'Sanitized runtime'] + texts += ['Sanitized instructions\n' + ''+text+'' for text in texts[:1]] + texts += ['Sanitized instructions\n' + 'Delegated task node:wrapped reached a terminal state. ' + 'Use task_status with taskId node:wrapped to read the result.' + 'Sanitized runtime'] + texts += ['Quoted notice: '+texts[1]] + native_context = 'Sanitized linking' + texts += [texts[1]+native_context, texts[1]+native_context+'\nMy own instruction.'] + tool_context = 'Sanitized tools' + texts += [texts[1]+native_context+tool_context, + texts[1]+native_context+tool_context+'\nMy own instruction.'] + _write(os.path.join(path, 'updates.jsonl'), [ + {'timestamp': 1791367200 + i, 'method': 'session/update', 'params': { + 'sessionId': 'grok-session', 'update': {'sessionUpdate': 'user_message_chunk', + 'content': {'type': 'text', 'text': text}, '_meta': {'promptIndex': i}}}} + for i, text in enumerate(texts)]) + for full in (False, True, False): + grok.import_grok_session(self.con, path, full=full) + for i, kind in enumerate(('interrupt', 'prompt', 'interrupt', 'prompt')): + row = interventions.resolve(self.con, f'sub:grok:grok-session:prompt:{i}') + self.assertEqual((row['created_by'], row['kind']), ('agent', kind)) + quoted = interventions.resolve(self.con, 'sub:grok:grok-session:prompt:4') + self.assertEqual((quoted['created_by'], quoted['kind']), ('unknown', 'prompt')) + self.assertEqual(quoted['actor_reason'], 'native_prompt_without_matching_origin') + self.assertEqual(interventions.resolve(self.con, 'sub:grok:grok-session:prompt:5') + ['created_by'], 'agent') + self.assertEqual(interventions.resolve(self.con, 'sub:grok:grok-session:prompt:6') + ['created_by'], 'unknown') + self.assertEqual(interventions.resolve(self.con, 'sub:grok:grok-session:prompt:7') + ['created_by'], 'agent') + self.assertEqual(interventions.resolve(self.con, 'sub:grok:grok-session:prompt:8') + ['created_by'], 'unknown') + # A ledger from the old importer must replay unchanged native files once. + self.con.execute('DELETE FROM intervention_prompt_evidence') + self.con.execute('UPDATE grok_session_meta SET summary_fp=?', + ('44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a',)) + with mock.patch.object(grok, '_load_jsonl_records', wraps=grok._load_jsonl_records) as load: + grok.import_grok_session(self.con, path) + self.assertTrue(load.called) + self.assertEqual(interventions.resolve(self.con, 'sub:grok:grok-session:prompt:1') + ['created_by'], 'agent') + with mock.patch.object(grok, '_load_jsonl_records', wraps=grok._load_jsonl_records) as load: + grok.import_grok_session(self.con, path) + load.assert_not_called() + + +if __name__ == '__main__': + unittest.main()