From 066ea5a3e62e8add7ad16aa8bf59b91b2d8e8a88 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:10:27 +0200 Subject: [PATCH 1/7] feat(cue): import decisions and report procedure reads (#129) --- CHANGELOG.md | 4 + README.md | 42 +++++ agent_observer/cli.py | 35 +++- agent_observer/cue.py | 375 +++++++++++++++++++++++++++++++++++++++ agent_observer/fleet.py | 3 +- agent_observer/health.py | 3 +- docs/contracts.md | 3 + tests/test_cue.py | 216 ++++++++++++++++++++++ tests/test_health.py | 3 +- 9 files changed, 679 insertions(+), 5 deletions(-) create mode 100644 agent_observer/cue.py create mode 100644 tests/test_cue.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9023306..4edbe72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- Import Cue decisions and settled ContextMD reads read-only; report procedure reads by entry, harness, machine and AgentsMD version against session-wide holdout, with source and target gaps. + ### Fixed - Parse sectioned AgentsMD direction identity, Codex native durations and Read calls, and Grok MCP timing and outcomes. Retain read success state and count only successful procedure reads; replay older sources to repair stored evidence (#128). diff --git a/README.md b/README.md index c8b141a..2dd2487 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,48 @@ from AgentsMD's Project Direction hook, the instruction text a host embedded, or the versioned plugin path of a Skill it loaded, resolved to a release through the AgentsMD repository's tags. +## Cue procedure reads + +`sync` (all harnesses, or `--harness t3`) also imports metadata from +`$T3CODE_HOME/userdata/contextmd.sqlite`, default `~/.t3/userdata/contextmd.sqlite`. +The source opens read-only. To refresh just this evidence or choose another source: + +```sh +agent-observer cue import --source /path/to/contextmd.sqlite +agent-observer cue report +agent-observer cue report --json +``` + +Reports group Cue entries by harness, machine and observed AgentsMD version +(`unknown` when the ledger cannot identify it). Each arm shows candidate +opportunities, distinct sessions, target reads, full versus partial reads, +and reads already present in that context epoch. A read must be settled and +successful in ContextMD; pending calls are excluded. Matching native read events +are linked by exact session, call id and path, without inserting duplicate events. +ContextMD-only and native-matched read counts remain visible. + +The observation window begins strictly after the moment and ends before the next +user message, context epoch or 30 minutes, whichever comes first. It is clipped +to the imported source watermark; incomplete windows are counted as pending, so +read rates are lower bounds while evidence is still arriving. Full wins over +partial when both occur in a window. Earlier reads are separate and can overlap +a later reread. Repeated candidates produce repeated opportunities, not independent +samples; distinct session counts show the comparison group's size. + +Holdout withholds all Cue entries for the session. Delivery outcomes are not +proof the model saw a rule, and reads do not prove it followed the procedure. +This observational comparison does not establish a per-procedure causal effect +or label a serve `ignored`, as ContextMD's outcome judge does. + +Targets are resolved from version-matching Library files named by the moment's +owners, with optional extra directories supplied by `cue import --library DIR`. +Changed or missing entry versions, ambiguous targets and entries without targets +are explicit gaps, not unread procedures. Missing databases, incompatible schemas +and machines without successful Cue moments show gaps in both `health` and the +report. `cue` exits 4 for unavailable evidence; `health` also exits 4 for a Cue gap. +No prompt, window, entry body or served text is imported. `export` carries only +this report's aggregates, without read paths, session ids or individual moments. + ## Evaluating AgentsMD releases `compare --by agentsmd` groups sessions from every project by the AgentsMD diff --git a/agent_observer/cli.py b/agent_observer/cli.py index a8a3388..4b0d6ef 100644 --- a/agent_observer/cli.py +++ b/agent_observer/cli.py @@ -120,6 +120,8 @@ def _text(payload) -> str: f" failed {h['harness']} {item['path']}: {item['error']}") if h.get("bare_roots"): lines += _bare_lines(h["harness"], h["bare_roots"]) + if "cue" in payload: + lines.append(f"Cue evidence: {payload['cue']['status']}") if payload.get("agentsmd_versions") is not None: lines.append(f"agentsmd versions mapped: {payload['agentsmd_versions']}") return "\n".join(lines) @@ -236,6 +238,16 @@ def build_parser() -> argparse.ArgumentParser: help="AgentsMD repository for the version map") s.add_argument("--json", action="store_true", dest="as_json", default=argparse.SUPPRESS) + cue = sub.add_parser("cue", help="import Cue evidence or report procedure reads") + cues = cue.add_subparsers(dest="op", required=True) + ci = cues.add_parser("import", help="refresh Cue metadata and settled reads, source read-only") + ci.add_argument("--source", default=None, help="ContextMD database") + ci.add_argument("--library", action="append", default=[], help="additional Library directory") + cr = cues.add_parser("report", help="aggregate target reads after Cue decisions") + cr.add_argument("--source", default=None, help="ContextMD database for availability check") + for cp in (ci, cr): + cp.add_argument("--json", action="store_true", dest="as_json", default=argparse.SUPPRESS) + ss = sub.add_parser("sessions", help="list or show imported sessions") ssub = ss.add_subparsers(dest="op", required=True) sl = ssub.add_parser("list") @@ -513,7 +525,7 @@ def _intervention_scope(parser): def _writes(ns) -> bool: - return ns.cmd in WRITERS or (ns.cmd, getattr(ns, "op", None)) == ( + return ns.cmd in WRITERS or (ns.cmd == "cue" and ns.op == "import") or (ns.cmd, getattr(ns, "op", None)) == ( "verdict", "add") @@ -528,6 +540,17 @@ def _run_command(con, ns) -> int: if ns.cmd == "sync": return _sync(con, ns) + if ns.cmd == "cue": + from . import cue + if ns.op == "import": + result = cue.import_source(con, source=ns.source, libraries=ns.library) + con.commit() + _emit(result, ns.as_json) + else: + result = cue.report(con, source=ns.source) + print(json.dumps(result, indent=2, sort_keys=True) if ns.as_json else cue.render(result)) + return 0 if result["status"] == "ok" else 4 + if ns.cmd == "sessions": return _sessions(con, ns) @@ -720,6 +743,7 @@ def _health_check(ns) -> int: for gap in result["unavailable"]: subject = f"{gap['host']} pid {gap['pid']}: " if "pid" in gap else "" print(f"- {subject}{gap['check']} unavailable: {gap['reason']}") + print(f"Cue evidence: {result['cue']['status']}") conflicts = result.get("identity_conflicts") if conflicts is None: print("ledger not read: identity conflicts unknown") @@ -731,7 +755,7 @@ def _health_check(ns) -> int: for harness, roots in result["bare_roots"].items(): for line in _bare_lines(harness, roots): print(f"- {harness} {line.strip()}") - return 4 if result["findings"] or result["unavailable"] else 0 + return 4 if result["findings"] or result["unavailable"] or result["cue"]["status"] != "ok" else 0 def _capabilities(ns) -> int: @@ -1036,6 +1060,11 @@ def _sync(con, ns) -> int: if row.get("harness") == "t3" and "skipped" not in row: row["shared_sessions"] = placement.record_shared_checkouts(con) break + cue_result = None + if not ns.source and "t3" in wanted: + from . import cue + cue_source = os.path.join(ns.root, "userdata", "contextmd.sqlite") if ns.root else None + cue_result = cue.import_source(con, source=cue_source) _identity.refresh_session_versions(con) con.commit() payload = {"_view": "sync", "harnesses": results, @@ -1043,6 +1072,8 @@ def _sync(con, ns) -> int: "contract": {"schema": SCHEMA_VERSION, "event": EVENT_CONTRACT_VERSION, "capture": CAPTURE_CONTRACT_VERSION}} + if cue_result is not None: + payload["cue"] = cue_result _emit(payload, ns.as_json) return 1 if any(r.get("failed") for r in results) else 0 diff --git a/agent_observer/cue.py b/agent_observer/cue.py new file mode 100644 index 0000000..1b275c7 --- /dev/null +++ b/agent_observer/cue.py @@ -0,0 +1,375 @@ +"""Cue decisions and successful ContextMD reads, without serving or prompt text. + +The source is a read-only snapshot. Local mirror tables are replaced atomically +on import because pending reads and delivery outcomes can change in place. +Native events are linked, never duplicated into the events table. +""" +from __future__ import annotations + +from bisect import bisect_left, bisect_right +from collections import Counter, defaultdict +from contextlib import closing +import hashlib +import json +import math +import os +from pathlib import Path +import re +import sqlite3 +import time + +from . import privacy + +WINDOW_SECONDS = 1800 +SERVED = frozenset({'injected', 'truncated', 'unverified', 'awaiting_evidence', 'hidden'}) +FULL = frozenset({'read_tool', 'shell_full'}) +KINDS = FULL | {'read_partial', 'shell_partial'} +IDENT = re.compile(r'^[A-Za-z0-9_.:/@+-]{1,512}$') +ENTRY_ID = re.compile(r'^[A-Za-z0-9][A-Za-z0-9_.-]{0,119}$') +REQUIRED = { + 'moments': {'id', 'received_at', 'harness', 'session', 'epoch', 'holdout', + 'kind', 'tool_use_id', 'owners_json', 'cue_status', 'delivery'}, + 'candidates': {'id', 'moment_id', 'matcher', 'entry', 'version', 'score', + 'outcome', 'stage', 'served_at'}, + 'reads': {'harness', 'session', 'epoch', 'path', 'kind', 'moment_id', + 'read_at', 'pending_tool_use_id'}, + 'deliveries': {'delivery', 'stage'}, +} + + +def source_path(source=None): + return Path(source or Path(os.environ.get('T3CODE_HOME', Path.home() / '.t3')) + / 'userdata/contextmd.sqlite').expanduser().absolute() + + +def _open(source): + con = sqlite3.connect(source_path(source).as_uri() + '?mode=ro', uri=True) + con.row_factory = sqlite3.Row + return con + + +def _schema(con): + for table, required in REQUIRED.items(): + if not required <= {r[1] for r in con.execute(f'PRAGMA table_info({table})')}: + return 'schema_drift' + return None + + +def source_status(source=None): + if not source_path(source).is_file(): + return 'missing_database' + try: + with closing(_open(source)) as con: + gap = _schema(con) + if gap: + return gap + return 'ok' if con.execute( + "SELECT 1 FROM moments WHERE cue_status='ok' LIMIT 1").fetchone() else 'no_cue' + except (sqlite3.Error, OSError): + return 'unreadable_database' + + +def _ensure(con): + # No global ledger migration or native parser changes are needed. + con.execute('CREATE TABLE IF NOT EXISTS cue_import(status TEXT, imported_at REAL, watermark REAL)') + con.execute('''CREATE TABLE IF NOT EXISTS cue_moments( + id INTEGER PRIMARY KEY, at REAL, harness TEXT, session TEXT, epoch INTEGER, + holdout INTEGER, kind TEXT, tool_use_id TEXT, cue_status TEXT)''') + con.execute('''CREATE TABLE IF NOT EXISTS cue_candidates( + id INTEGER PRIMARY KEY, moment_id INTEGER, entry TEXT, version TEXT, + score REAL, outcome TEXT, stage TEXT, served_at REAL, delivery_stage TEXT, + target TEXT, target_status TEXT)''') + con.execute('''CREATE TABLE IF NOT EXISTS cue_reads( + harness TEXT, session TEXT, epoch INTEGER, path TEXT, kind TEXT, + moment_id INTEGER, at REAL, native_event_id INTEGER, + PRIMARY KEY(harness, session, epoch, path, kind, moment_id, at))''') + con.execute('CREATE INDEX IF NOT EXISTS cue_read_path ON cue_reads(harness, session, epoch, path, at)') + + +def _id(value): + return value if isinstance(value, str) and IDENT.fullmatch(value) else None + + +def _path(value): + if (not isinstance(value, str) or not value.startswith('/') or len(value) > 8192 + or any(ord(c) < 32 for c in value)): + return None + safe = privacy.filter_target(value) + return safe if safe == value else None + + +def _number(value): + return type(value) in (int, float) and math.isfinite(value) + + +def _targets(roots): + """Resolve only version-matching files, never infer an old target from a new file. + + Chromeria hashes the original bytes and resolves target relative to the + entry file (or the home directory). Only scalar id/target header fields + are needed; no body or arbitrary YAML data reaches the ledger. + """ + found = {} + for root in roots: + try: + paths = Path(root).glob('*.md') + for file in paths: + if file.name.lower() == 'readme.md': + continue + raw = file.read_bytes() + text = raw.decode('utf-8-sig').replace('\r\n', '\n') + lines = text.splitlines() + if not lines or lines[0].strip() != '---': + continue + fields = {} + for line in lines[1:]: + if line.strip() == '---': + break + key, sep, val = line.partition(':') + if sep and key in ('id', 'target'): + val = val.strip() + if val.startswith('"'): + val = json.loads(val) + elif val.startswith("'") and val.endswith("'"): + val = val[1:-1].replace("''", "'") + fields[key] = val + entry = _id(fields.get('id', file.stem)) + if not entry: + continue + target = fields.get('target') + if isinstance(target, str) and target: + target = Path(target).expanduser() + target = str((file.parent / target).resolve()) + else: + target = None + key = (entry, hashlib.sha256(raw).hexdigest()[:12]) + value = (_path(target), 'resolved' if _path(target) else 'no_target') + if key in found and found[key] != value: + found[key] = (None, 'ambiguous_target') + else: + found[key] = value + except (OSError, ValueError, UnicodeError): + continue + return found + + +def _native_reads(con): + """Exact call id and path only; timestamp proximity is not identity.""" + found = defaultdict(set) + for r in con.execute("SELECT id,session_key,native_id,target,detail_json FROM events" + " WHERE family IN ('read','skill_read')"): + try: + detail = json.loads(r['detail_json'] or '{}') + except ValueError: + continue + path = detail.get('skill_path') if isinstance(detail, dict) else None + path = path or r['target'] + if not _path(path): + continue + native = r['native_id'] + # Native adapters suffix a read's call id with :read or :. + for call in (native, native.removesuffix(':read'), native.removesuffix(':' + path)): + found[(r['session_key'], call, path)].add(r['id']) + return found + + +def import_source(con, source=None, libraries=()): + """Import one consistent source snapshot. A failed refresh preserves prior data.""" + _ensure(con) + status = source_status(source) + if status != 'ok': + con.execute('DELETE FROM cue_import') + con.execute('INSERT INTO cue_import VALUES(?,?,NULL)', (status, time.time())) + return {'status': status} + with closing(_open(source)) as native: + native.execute('BEGIN') + moments = native.execute('SELECT id,received_at,harness,session,epoch,holdout,kind,' + 'tool_use_id,owners_json,cue_status FROM moments').fetchall() + roots = set(libraries) + for m in moments: + try: + owners = json.loads(m['owners_json'] or '{}') + if isinstance(owners, dict): + roots.update(k for k in owners if _path(k)) + except ValueError: + pass + targets_by_root = {root: _targets([root]) for root in roots} + target_cache = {} + moment_targets = {} + for m in moments: + try: + owners = json.loads(m['owners_json'] or '{}') + except ValueError: + owners = {} + active = set(owners) if isinstance(owners, dict) else set() + active.update(libraries) + root_key = tuple(sorted(active)) + if root_key not in target_cache: + mapping = {} + for root in active: + for key, value in targets_by_root.get(root, {}).items(): + if key in mapping and mapping[key] != value: + mapping[key] = (None, 'ambiguous_target') + else: + mapping[key] = value + target_cache[root_key] = mapping + moment_targets[m['id']] = target_cache[root_key] + matches = _native_reads(con) + con.execute('SAVEPOINT cue_refresh') + try: + for table in ('cue_candidates', 'cue_reads', 'cue_moments', 'cue_import'): + con.execute('DELETE FROM ' + table) + for m in moments: + if not (_number(m['received_at']) and _id(m['harness']) and _id(m['session']) + and type(m['epoch']) is int and m['holdout'] in (0, 1)): + continue + con.execute('INSERT INTO cue_moments VALUES(?,?,?,?,?,?,?,?,?)', + (m['id'],m['received_at'],m['harness'],m['session'],m['epoch'], + m['holdout'],_id(m['kind']),_id(m['tool_use_id']),_id(m['cue_status']))) + ids = {r[0] for r in con.execute('SELECT id FROM cue_moments')} + columns = {r[1] for r in native.execute('PRAGMA table_info(candidates)')} + delivery = 'COALESCE(c.delivery,m.delivery)' if 'delivery' in columns else 'm.delivery' + for c in native.execute(f'''SELECT c.id,c.moment_id,c.entry,c.version,c.score,c.outcome, + c.stage,c.served_at,d.stage AS delivery_stage FROM candidates c + JOIN moments m ON m.id=c.moment_id + LEFT JOIN deliveries d ON d.delivery={delivery} WHERE c.matcher='intent' '''): + if (c['moment_id'] not in ids or not isinstance(c['entry'], str) + or not ENTRY_ID.fullmatch(c['entry']) or not _id(c['version'])): + continue + target, target_status = moment_targets[c['moment_id']].get((c['entry'], c['version']), + (None, 'version_unavailable')) + con.execute('INSERT INTO cue_candidates VALUES(?,?,?,?,?,?,?,?,?,?,?)', + (c['id'],c['moment_id'],c['entry'],c['version'], + c['score'] if _number(c['score']) else None, + _id(c['outcome']),_id(c['stage']),c['served_at'], + _id(c['delivery_stage']),target,target_status)) + calls = {m['id']: m['tool_use_id'] for m in moments} + for r in native.execute('SELECT harness,session,epoch,path,kind,moment_id,read_at' + ' FROM reads WHERE pending_tool_use_id IS NULL'): + if (r['moment_id'] not in ids or not _path(r['path']) or r['kind'] not in KINDS + or not _number(r['read_at'])): + continue + key = (f"{r['harness']}:{r['session']}", calls.get(r['moment_id']), r['path']) + evidence = matches.get(key, set()) + event_id = next(iter(evidence)) if len(evidence) == 1 else None + con.execute('INSERT OR IGNORE INTO cue_reads VALUES(?,?,?,?,?,?,?,?)', + (*tuple(r), event_id)) + watermark = max((m['received_at'] for m in moments if _number(m['received_at'])), default=0) + con.execute('INSERT INTO cue_import VALUES(?,?,?)', ('ok',time.time(),watermark)) + con.execute('RELEASE cue_refresh') + except BaseException: + con.execute('ROLLBACK TO cue_refresh') + con.execute('RELEASE cue_refresh') + raise + return {'status': 'ok', **{t: con.execute('SELECT count(*) FROM cue_' + t).fetchone()[0] + for t in ('moments','candidates','reads')}} + + +def health(con=None, source=None): + status = source_status(source) + if status != 'ok': + return {'status': status} + if con is None: + return {'status': 'not_imported'} + if not con.execute("SELECT 1 FROM sqlite_master WHERE name='cue_import'").fetchone(): + return {'status': 'not_imported'} + row = con.execute('SELECT * FROM cue_import').fetchone() + return dict(row) if row else {'status': 'not_imported'} + + +def report(con, machine=None, check_source=True, source=None): + from .fleet import machine_name + state = health(con, source) if check_source else _stored_health(con) + result = {'status': state['status'], 'import': state, 'window_seconds': WINDOW_SECONDS, + 'holdout_unit': 'session', + 'interpretation': 'Observational session-wide Cue holdout; read evidence is not proof of following a procedure or a per-procedure causal effect.', + 'window': 'After the moment, before the next user message or context epoch, capped at 30 minutes and the imported source watermark. Earlier reads are separate; incomplete windows are pending.', + 'groups': [], 'gaps': {}} + if state['status'] != 'ok': + return result + moments = {r['id']: dict(r) for r in con.execute('SELECT * FROM cue_moments')} + boundaries = defaultdict(list) + by_session = defaultdict(list) + for m in moments.values(): + by_session[(m['harness'],m['session'])].append(m) + for key, ms in by_session.items(): + prior_epoch = None + for m in sorted(ms, key=lambda m: (m['at'], m['id'])): + if m['kind'] == 'user_message' or (prior_epoch is not None and m['epoch'] != prior_epoch): + boundaries[key].append(m['at']) + prior_epoch = m['epoch'] + reads = defaultdict(list) + provenance = Counter() + for r in con.execute('SELECT * FROM cue_reads ORDER BY at'): + reads[(r['harness'],r['session'],r['epoch'],r['path'])].append(dict(r)) + provenance['native_and_contextmd' if r['native_event_id'] else 'contextmd_only'] += 1 + versions = {r[0]: r[1] or 'unknown' for r in con.execute('SELECT session_key,agentsmd_version FROM sessions')} + groups = {} + gaps = Counter() + for c in con.execute('SELECT * FROM cue_candidates'): + m = moments[c['moment_id']] + if not c['target']: + gaps[c['target_status']] += 1 + continue + key = (c['entry'],m['harness'],machine or machine_name(), + versions.get(f"{m['harness']}:{m['session']}", 'unknown')) + if key not in groups: + groups[key] = {'entry': key[0], 'harness': key[1], 'machine': key[2], + 'agentsmd_version': key[3], 'outcomes': Counter(), 'arms': {}} + g = groups[key] + g['outcomes'][c['outcome']] += 1 + arm = 'holdout' if c['outcome'] == 'holdout' and m['holdout'] == 1 else ( + 'served' if c['outcome'] in SERVED and m['holdout'] == 0 else None) + if arm is None: + continue + a = g['arms'].setdefault(arm, {'opportunities': 0, 'sessions': set(), 'read': 0, + 'full': 0, 'partial': 0, 'already_read': 0, 'already_full': 0, 'already_partial': 0, + 'pending_window': 0, 'native_matched_reads': 0}) + a['opportunities'] += 1 + a['sessions'].add(m['session']) + bs = boundaries[(m['harness'],m['session'])] + index = bisect_right(bs, m['at']) + end = min(m['at'] + WINDOW_SECONDS, bs[index] if index < len(bs) else math.inf) + watermark = state.get('watermark') or 0 + a['pending_window'] += end > watermark + end = min(end, math.nextafter(watermark, math.inf)) + rs = reads[(m['harness'],m['session'],m['epoch'],c['target'])] + times = [r['at'] for r in rs] + earlier = rs[:bisect_right(times,m['at'])] + if earlier: + a['already_read'] += 1 + a['already_full' if any(r['kind'] in FULL for r in earlier) else 'already_partial'] += 1 + after = rs[bisect_right(times,m['at']):bisect_left(times,end)] + if after: + a['read'] += 1 + a['full' if any(r['kind'] in FULL for r in after) else 'partial'] += 1 + a['native_matched_reads'] += sum(r['native_event_id'] is not None for r in after) + for g in groups.values(): + for a in g['arms'].values(): + a['sessions'] = len(a['sessions']) + a['read_rate'] = a['read'] / a['opportunities'] + g['outcomes'] = dict(g['outcomes']) + result.update(groups=sorted(groups.values(),key=lambda g:(g['entry'],g['harness'],g['agentsmd_version'])), + gaps=dict(gaps), read_sources=dict(provenance), + cue_statuses=dict(Counter(m['cue_status'] or 'unknown' for m in moments.values()))) + return result + + +def _stored_health(con): + if not con.execute("SELECT 1 FROM sqlite_master WHERE name='cue_import'").fetchone(): + return {'status': 'not_imported'} + r = con.execute('SELECT * FROM cue_import').fetchone() + return dict(r) if r else {'status': 'not_imported'} + + +def render(result): + lines = [f"Cue procedure reads: {result['status']}", result['interpretation'], result['window']] + for g in result['groups']: + lines.append(f"{g['entry']} | {g['harness']} | {g['machine']} | AgentsMD {g['agentsmd_version']}") + for arm,a in g['arms'].items(): + lines.append(f" {arm}: {a['read']}/{a['opportunities']} read, {a['sessions']} sessions; " + f"full {a['full']}, partial {a['partial']}, earlier {a['already_read']}, " + f"pending windows {a['pending_window']}") + if result['gaps']: + lines.append('Target gaps: ' + json.dumps(result['gaps'],sort_keys=True)) + return '\n'.join(lines) diff --git a/agent_observer/fleet.py b/agent_observer/fleet.py index e11585e..101e8a6 100644 --- a/agent_observer/fleet.py +++ b/agent_observer/fleet.py @@ -132,7 +132,8 @@ def export(con: sqlite3.Connection, machine: str | None = None) -> dict: "sessions": sorted(report.task_sessions(con, task_id))}) synced = con.execute("SELECT MAX(imported_at) FROM sources").fetchone()[0] counts, judgments = _judgments(con) - return {"schema": SCHEMA, "machine": machine, + from . import cue + return {"cue": cue.report(con, machine=machine), "schema": SCHEMA, "machine": machine, "observer_version": __version__, "exported_at": _now(), "ledger_synced_at": _iso(synced), "sessions": sessions, "tasks": tasks, "withheld_tasks": withheld, diff --git a/agent_observer/health.py b/agent_observer/health.py index 1b2689a..c64a4c2 100644 --- a/agent_observer/health.py +++ b/agent_observer/health.py @@ -218,7 +218,8 @@ def check(processes=None, registry=None, con=None, bare=None) -> dict: "pid": proc["pid"], "reason": proc["cwd_error"]}) registry = claude_registry() if registry is None else registry findings = check_claude(registry) + check_opencode(processes) + check_codex(processes) - return {"findings": findings, + from . import cue + return {"cue": cue.health(con), "findings": findings, "unavailable": unavailable, "identity_conflicts": identity_conflicts(con) if con is not None else None, "bare_roots": bare_roots() if bare is None else bare, diff --git a/docs/contracts.md b/docs/contracts.md index 6ac196c..2d88b93 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -36,6 +36,9 @@ are committed. | `t3_outcome_provenance` (`router_outcome_provenance`) | task id | Last adapter-written outcome snapshot; a row that differs is user-owned and survives re-import. | | `router_jobs`, `router_invocations`, `router_readings` | router ids | Model Router ledger rows, copied read-only. | | `agentsmd_versions` | AGENTS.md SHA-256 | Release map from the local AgentsMD tags. | +| `cue_import` | one snapshot | ContextMD import status, time and source watermark; unavailable evidence is a gap. | +| `cue_moments`, `cue_candidates` | source moment id, candidate id | Cue decision metadata, scores, all candidate outcomes and delivery stages; version-matched targets, never prompt, window or served text. | +| `cue_reads` | harness, session, epoch, path, kind, moment, time | Settled successful ContextMD reads with optional exact native event linkage; no duplicate native events are inserted. | | `import_errors` | none | Quarantined records; prior valid data is kept. | Every natural key carries its harness prefix (`t3:`, `codex:`, `claude:`, diff --git a/tests/test_cue.py b/tests/test_cue.py new file mode 100644 index 0000000..896f216 --- /dev/null +++ b/tests/test_cue.py @@ -0,0 +1,216 @@ +"""Cue's real SQLite boundary and aggregate report contract.""" +import contextlib +import hashlib +import io +import json +from pathlib import Path +import sqlite3 +from unittest.mock import patch + +from helpers import LedgerCase +from agent_observer import cli, cue, fleet + + +class CueTests(LedgerCase): + def setUp(self): + super().setUp() + self.source = Path(self.tmp.name) / 'contextmd.sqlite' + self.library = Path(self.tmp.name) / 'library' + self.library.mkdir() + self.target = str(Path(self.tmp.name) / 'operations/procedures/research/index.md') + self.raw = ('---\nid: research\nname: Research\napplies: research work\n' + f'target: {self.target}\n---\nPRIVATE SERVED TEXT\n').encode() + (self.library / 'research.md').write_bytes(self.raw) + self.version = hashlib.sha256(self.raw).hexdigest()[:12] + self.native = sqlite3.connect(self.source) + self.native.executescript(''' + CREATE TABLE moments(id INTEGER PRIMARY KEY, received_at REAL, harness TEXT, + session TEXT, epoch INTEGER, holdout INTEGER, kind TEXT, tool_use_id TEXT, + owners_json TEXT, cue_status TEXT, delivery TEXT, window TEXT); + CREATE TABLE candidates(id INTEGER PRIMARY KEY, moment_id INTEGER, matcher TEXT, + entry TEXT, version TEXT, score REAL, outcome TEXT, stage TEXT, served_at REAL); + CREATE TABLE reads(harness TEXT,session TEXT,epoch INTEGER,path TEXT,kind TEXT, + moment_id INTEGER,read_at REAL,pending_tool_use_id TEXT); + CREATE TABLE deliveries(delivery TEXT,stage TEXT,body TEXT); + ''') + self.moment(1, 100) + self.candidate(1, 1) + self.moment(2, 110, kind='tool_result', call='read-call') + self.read(2, 110) + self.moment(3, 2000, kind='before_tool') + self.native.commit() + + def tearDown(self): + self.native.close() + super().tearDown() + + def moment(self, id, at, session='s', epoch=0, holdout=0, kind='before_tool', call=None): + self.native.execute('INSERT INTO moments VALUES(?,?,?,?,?,?,?,?,?,?,?,?)', + (id,at,'codex',session,epoch,holdout,kind,call, + json.dumps({str(self.library): 'abc'}),'ok',None,'PRIVATE WINDOW TEXT')) + + def candidate(self, id, moment, outcome='injected', version=None, entry='research'): + self.native.execute('INSERT INTO candidates VALUES(?,?,?,?,?,?,?,?,?)', + (id,moment,'intent',entry,version or self.version,0.9,outcome,'settled',100)) + + def read(self, moment, at, session='s', epoch=0, kind='shell_full', pending=None): + self.native.execute('INSERT INTO reads VALUES(?,?,?,?,?,?,?,?)', + ('codex',session,epoch,self.target,kind,moment,at,pending)) + + def refresh(self): + self.native.commit() + result = cue.import_source(self.con, self.source) + self.assertEqual(result['status'], 'ok') + return cue.report(self.con, check_source=False, machine='test') + + def arm(self, result, arm='served'): + return result['groups'][0]['arms'][arm] + + def test_full_partial_and_no_text(self): + result = self.refresh() + arm = self.arm(result) + self.assertEqual((arm['read'],arm['full'],arm['partial'],arm['sessions']), (1,1,0,1)) + self.assertEqual(arm['pending_window'],0) + text = '\n'.join(self.con.iterdump()) + self.assertNotIn('PRIVATE',text) + self.assertEqual(result['read_sources'], {'contextmd_only':1}) + self.native.execute("UPDATE reads SET kind='shell_partial'") + self.assertEqual(self.arm(self.refresh())['partial'],1) + + def test_holdout_is_session_wide_and_counts_sessions(self): + self.moment(4,100,session='h',holdout=1) + self.candidate(4,4,'holdout') + self.moment(5,110,session='h',holdout=1) + self.read(5,110,session='h',kind='read_partial') + self.candidate(5,4,'holdout') + result = self.refresh() + self.assertEqual(result['holdout_unit'],'session') + arm = self.arm(result,'holdout') + self.assertEqual((arm['opportunities'],arm['sessions'],arm['partial']),(2,1,2)) + self.assertIn('causal', result['interpretation']) + + def test_pending_failed_and_duplicate_reads_not_counted(self): + self.native.execute("UPDATE reads SET pending_tool_use_id='read-call'") + self.assertEqual(self.arm(self.refresh())['read'],0) + self.native.execute('UPDATE reads SET pending_tool_use_id=NULL') + self.read(2,110) + result = self.refresh() + self.assertEqual(self.arm(result)['read'],1) + self.assertEqual(self.con.execute('SELECT count(*) FROM cue_reads').fetchone()[0],1) + self.native.execute('DELETE FROM reads') # failed pending reads are deleted by Chromeria + self.assertEqual(self.arm(self.refresh())['read'],0) + + def test_native_identity_is_linked_without_duplicate_events(self): + self.con.execute("INSERT INTO events(session_key,family,native_id,target,detail_json) VALUES(?,?,?,?,?)", + ('codex:s','skill_read','read-call:'+self.target,'operations',json.dumps({'skill_path':self.target}))) + result = self.refresh() + self.assertEqual(result['read_sources'],{'native_and_contextmd':1}) + self.assertEqual(self.arm(result)['native_matched_reads'],1) + self.assertEqual(self.con.execute('SELECT count(*) FROM events').fetchone()[0],1) + self.con.execute("UPDATE events SET native_id='different-call:read'") + self.assertEqual(self.refresh()['read_sources'],{'contextmd_only':1}) + + def test_earlier_read_is_separate_and_epoch_scoped(self): + self.native.execute('DELETE FROM reads') + self.moment(4,90) + self.read(4,90,kind='read_partial') + arm = self.arm(self.refresh()) + self.assertEqual((arm['read'],arm['already_read'],arm['already_partial']),(0,1,1)) + self.native.execute('UPDATE reads SET epoch=1') + self.assertEqual(self.arm(self.refresh())['already_read'],0) + + def test_window_stops_at_user_message_epoch_or_time_cap(self): + for kind,epoch,at in [('user_message',0,105),('compaction',1,105),('before_tool',0,1900)]: + with self.subTest(kind=kind): + self.native.execute('DELETE FROM moments WHERE id=4') + self.moment(4,at,epoch=epoch,kind=kind) + self.native.execute('UPDATE reads SET read_at=?',(at,)) + self.assertEqual(self.arm(self.refresh())['read'],0) + + def test_pending_window_and_watermark_read(self): + self.native.execute('DELETE FROM moments WHERE id=3') + arm = self.arm(self.refresh()) + self.assertEqual((arm['read'],arm['pending_window']),(1,1)) + + def test_outcomes_and_unavailable_versions_are_not_misclassified(self): + self.candidate(4,1,'below_threshold') + self.candidate(5,1,'delivery_unverified') + self.candidate(6,1,'holdout') # inconsistent source arm, never added to holdout denominator + self.candidate(7,1,version='old-version') + result = self.refresh() + self.assertEqual(self.arm(result)['opportunities'],1) + self.assertNotIn('holdout',result['groups'][0]['arms']) + self.assertEqual(result['gaps'],{'version_unavailable':1}) + self.assertEqual(result['groups'][0]['outcomes']['below_threshold'],1) + + def test_source_is_read_only_and_refresh_idempotent(self): + before = self.source.read_bytes() + self.refresh() + self.refresh() + self.assertEqual(self.source.read_bytes(),before) + with contextlib.closing(cue._open(self.source)) as con: + with self.assertRaises(sqlite3.OperationalError): + con.execute('DELETE FROM moments') + self.assertEqual(self.con.execute('SELECT count(*) FROM cue_candidates').fetchone()[0],1) + + def test_missing_schema_drift_no_cue_and_not_imported_are_gaps(self): + self.assertEqual(cue.health(self.con,self.source)['status'],'not_imported') + self.native.execute("UPDATE moments SET cue_status='unavailable'") + self.native.commit() + self.assertEqual(cue.import_source(self.con,self.source)['status'],'no_cue') + self.native.execute('DROP TABLE reads') + self.native.commit() + self.assertEqual(cue.import_source(self.con,self.source)['status'],'schema_drift') + self.assertEqual(cue.import_source(self.con, str(self.source)+'missing')['status'],'missing_database') + self.assertEqual(cue.report(self.con,check_source=False)['groups'],[]) + + def test_version_matching_target_relative_to_library(self): + self.native.execute('UPDATE candidates SET version=?',('stale',)) + result = self.refresh() + self.assertEqual(result['groups'],[]) + self.assertEqual(result['gaps'],{'version_unavailable':1}) + + def test_cli_and_export_aggregates_only(self): + self.con.commit() + output = io.StringIO() + with contextlib.redirect_stdout(output): + self.assertEqual(cli.main(['--db',self.db_path,'cue','import','--source',str(self.source),'--json']),0) + self.assertEqual(cli.main(['--db',self.db_path,'cue','report','--source',str(self.source),'--json']),0) + self.assertIn('opportunities',output.getvalue()) + with patch.object(cue,'source_path',return_value=self.source): + doc = fleet.export(self.con,machine='test') + encoded = json.dumps(doc['cue']) + for secret in ('PRIVATE',self.target,'codex:s','moment_id','native_event_id'): + self.assertNotIn(secret, encoded) + self.assertIn('research',encoded) + self.assertEqual(cue.render(doc['cue']).splitlines()[0],'Cue procedure reads: ok') + + def test_targets_use_each_moments_own_library_roots(self): + other = Path(self.tmp.name) / 'other/library' + other.mkdir(parents=True) + relative = self.raw.replace(self.target.encode(), b'../procedure.md') + (self.library / 'research.md').write_bytes(relative) + (other / 'research.md').write_bytes(relative) + self.native.execute('UPDATE candidates SET version=?', + (hashlib.sha256(relative).hexdigest()[:12],)) + self.moment(4,2001,session='other') + self.native.execute('UPDATE moments SET owners_json=? WHERE id=4', + (json.dumps({str(other): 'abc'}),)) + result = self.refresh() + self.assertEqual(result['gaps'],{}) + target = self.con.execute('SELECT target FROM cue_candidates').fetchone()[0] + self.assertEqual(target,str((self.library / '../procedure.md').resolve())) + + def test_candidate_delivery_tracks_carrys_own_delivery(self): + self.native.execute('ALTER TABLE candidates ADD COLUMN delivery TEXT') + self.native.execute("INSERT INTO deliveries VALUES('original','released','PRIVATE')") + self.native.execute("INSERT INTO deliveries VALUES('carry','settled','PRIVATE')") + self.native.execute("UPDATE moments SET delivery='original'") + self.native.execute("UPDATE candidates SET delivery='carry'") + self.refresh() + self.assertEqual(self.con.execute('SELECT delivery_stage FROM cue_candidates').fetchone()[0],'settled') + + def test_entry_ids_cannot_export_paths(self): + self.candidate(2,1,entry='/private/path') + self.refresh() + self.assertEqual(self.con.execute('SELECT count(*) FROM cue_candidates').fetchone()[0],1) diff --git a/tests/test_health.py b/tests/test_health.py index 40fb25c..5a9ff5b 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -30,7 +30,8 @@ def tool(self, name, body): def check_unavailable(self, reason): with patch.object(health, "check_claude", return_value=[{"host": "claude"}]) as claude, \ - patch.object(health, "identity_conflicts", return_value=[]) as identity: + patch.object(health, "identity_conflicts", return_value=[]) as identity, \ + patch("agent_observer.cue.health", return_value={"status": "not_imported"}): result = health.check(registry=[], con=object()) self.assertEqual(result["findings"], [{"host": "claude"}]) self.assertEqual(result["identity_conflicts"], []) From 9fbeb3bb1be69cec99963f2504b0aef917e0d05c Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:16:43 +0200 Subject: [PATCH 2/7] fix(cue): preserve and validate optional fleet aggregates --- README.md | 4 +-- agent_observer/cue.py | 54 +++++++++++++++++++++++++++++++++++++++++ agent_observer/fleet.py | 9 ++++++- tests/test_cue.py | 7 +++++- 4 files changed, 70 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 2dd2487..3c8f69b 100644 --- a/README.md +++ b/README.md @@ -98,8 +98,8 @@ The source opens read-only. To refresh just this evidence or choose another sour ```sh agent-observer cue import --source /path/to/contextmd.sqlite -agent-observer cue report -agent-observer cue report --json +agent-observer cue report --source /path/to/contextmd.sqlite +agent-observer cue report --source /path/to/contextmd.sqlite --json ``` Reports group Cue entries by harness, machine and observed AgentsMD version diff --git a/agent_observer/cue.py b/agent_observer/cue.py index 1b275c7..ff712d3 100644 --- a/agent_observer/cue.py +++ b/agent_observer/cue.py @@ -373,3 +373,57 @@ def render(result): if result['gaps']: lines.append('Target gaps: ' + json.dumps(result['gaps'],sort_keys=True)) return '\n'.join(lines) + + +EXPORT_FIELDS = frozenset({'status', 'window_seconds', 'holdout_unit', 'groups', + 'gaps', 'read_sources', 'cue_statuses'}) +ARM_COUNTS = frozenset({'opportunities', 'sessions', 'read', 'full', 'partial', + 'already_read', 'already_full', 'already_partial', 'pending_window', 'native_matched_reads'}) + + +def export(con, machine=None): + """Optional aggregate extension; old ledgers retain their existing export shape.""" + if _stored_health(con)['status'] == 'not_imported': + return None + value = report(con, machine=machine) + return {k: value.get(k, {}) for k in EXPORT_FIELDS} + + +def validate_export(value): + """Fail closed at the fleet boundary: no paths, moments or free-text extras.""" + def require(ok): + if not ok: + raise ValueError('invalid Cue aggregate export') + + def counts(mapping, allowed=None): + require(isinstance(mapping, dict)) + for k, v in mapping.items(): + require(isinstance(k, str) and re.fullmatch(r'[a-z_]{1,64}', k) + and (allowed is None or k in allowed) + and type(v) is int and 0 <= v <= 2**53) + + require(isinstance(value, dict) and set(value) == EXPORT_FIELDS) + require(isinstance(value['status'], str) and value['status'] in {'ok','missing_database','schema_drift','no_cue', + 'unreadable_database','not_imported'}) + require(value['window_seconds'] == WINDOW_SECONDS and value['holdout_unit'] == 'session') + counts(value['gaps'], {'version_unavailable','no_target','ambiguous_target'}) + counts(value['read_sources'], {'native_and_contextmd','contextmd_only'}) + counts(value['cue_statuses']) + require(isinstance(value['groups'], list)) + for g in value['groups']: + require(isinstance(g, dict) and set(g) == {'entry','harness','machine', + 'agentsmd_version','outcomes','arms'}) + require(isinstance(g['entry'], str) and ENTRY_ID.fullmatch(g['entry'])) + require(isinstance(g['harness'], str) and g['harness'] in {'codex','claude','grok','opencode'}) + require(isinstance(g['machine'], str) and re.fullmatch(r'[a-z0-9][a-z0-9-]{0,62}', g['machine'])) + require(isinstance(g['agentsmd_version'], str) + and re.fullmatch(r'(?:[0-9]+\.[0-9]+\.[0-9]+|unknown)', g['agentsmd_version'])) + counts(g['outcomes']) + require(isinstance(g['arms'], dict) and set(g['arms']) <= {'served','holdout'}) + for arm in g['arms'].values(): + require(isinstance(arm, dict) and set(arm) == ARM_COUNTS | {'read_rate'}) + counts({k: arm[k] for k in ARM_COUNTS}, ARM_COUNTS) + require(_number(arm['read_rate']) and 0 <= arm['read_rate'] <= 1) + require(arm['read'] == arm['full'] + arm['partial'] <= arm['opportunities']) + require(arm['already_read'] == arm['already_full'] + arm['already_partial'] <= arm['opportunities']) + return value diff --git a/agent_observer/fleet.py b/agent_observer/fleet.py index 101e8a6..af9537e 100644 --- a/agent_observer/fleet.py +++ b/agent_observer/fleet.py @@ -133,7 +133,9 @@ def export(con: sqlite3.Connection, machine: str | None = None) -> dict: synced = con.execute("SELECT MAX(imported_at) FROM sources").fetchone()[0] counts, judgments = _judgments(con) from . import cue - return {"cue": cue.report(con, machine=machine), "schema": SCHEMA, "machine": machine, + cue_data = cue.export(con, machine=machine) + return {**({"cue": cue_data} if cue_data is not None else {}), + "schema": SCHEMA, "machine": machine, "observer_version": __version__, "exported_at": _now(), "ledger_synced_at": _iso(synced), "sessions": sessions, "tasks": tasks, "withheld_tasks": withheld, @@ -447,6 +449,9 @@ def validate(doc) -> dict: if not isinstance(judgments, list) \ or not all(_valid_judgment(j) for j in judgments): raise ValueError("invalid judgments") + if "cue" in doc: + from . import cue + cue.validate_export(doc["cue"]) return doc @@ -459,6 +464,8 @@ def validate(doc) -> dict: def _schema_only(doc: dict) -> dict: """A validated export reduced to its schema fields; extras never stored.""" out = {k: doc.get(k) for k in HEADER_FIELDS if k in doc} + if "cue" in doc: + out["cue"] = doc["cue"] out["sessions"] = [ {**{k: s.get(k) for k in SESSION_FIELDS if k in s}, "usage": [{k: g.get(k) for k in USAGE_FIELDS if k in g} diff --git a/tests/test_cue.py b/tests/test_cue.py index 896f216..f6f0797 100644 --- a/tests/test_cue.py +++ b/tests/test_cue.py @@ -183,7 +183,12 @@ def test_cli_and_export_aggregates_only(self): for secret in ('PRIVATE',self.target,'codex:s','moment_id','native_event_id'): self.assertNotIn(secret, encoded) self.assertIn('research',encoded) - self.assertEqual(cue.render(doc['cue']).splitlines()[0],'Cue procedure reads: ok') + self.assertEqual(cue.render(cue.report(self.con,check_source=False)).splitlines()[0], + 'Cue procedure reads: ok') + self.assertEqual(fleet._schema_only(fleet.validate(doc))['cue'],doc['cue']) + doc['cue']['groups'][0]['prompt'] = 'PRIVATE' + with self.assertRaisesRegex(ValueError,'Cue aggregate'): + fleet.validate(doc) def test_targets_use_each_moments_own_library_roots(self): other = Path(self.tmp.name) / 'other/library' From 5bd3f9aeae0c38c9f91bacb83d62aa5095e06335 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 14:28:59 +0200 Subject: [PATCH 3/7] fix(cue): export stored snapshots and reject oversized rates --- README.md | 3 ++- agent_observer/cue.py | 5 +++-- tests/test_cue.py | 18 ++++++++++++++++++ 3 files changed, 23 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 3c8f69b..81ccbee 100644 --- a/README.md +++ b/README.md @@ -130,7 +130,8 @@ are explicit gaps, not unread procedures. Missing databases, incompatible schema and machines without successful Cue moments show gaps in both `health` and the report. `cue` exits 4 for unavailable evidence; `health` also exits 4 for a Cue gap. No prompt, window, entry body or served text is imported. `export` carries only -this report's aggregates, without read paths, session ids or individual moments. +the imported snapshot's aggregates, without read paths, session ids or individual +moments; it does not require the original database to remain available. ## Evaluating AgentsMD releases diff --git a/agent_observer/cue.py b/agent_observer/cue.py index ff712d3..d09fe9c 100644 --- a/agent_observer/cue.py +++ b/agent_observer/cue.py @@ -99,7 +99,8 @@ def _path(value): def _number(value): - return type(value) in (int, float) and math.isfinite(value) + # Integers are finite without a potentially overflowing float conversion. + return type(value) is int or (type(value) is float and math.isfinite(value)) def _targets(roots): @@ -385,7 +386,7 @@ def export(con, machine=None): """Optional aggregate extension; old ledgers retain their existing export shape.""" if _stored_health(con)['status'] == 'not_imported': return None - value = report(con, machine=machine) + value = report(con, machine=machine, check_source=False) return {k: value.get(k, {}) for k in EXPORT_FIELDS} diff --git a/tests/test_cue.py b/tests/test_cue.py index f6f0797..f2a7ee0 100644 --- a/tests/test_cue.py +++ b/tests/test_cue.py @@ -219,3 +219,21 @@ def test_entry_ids_cannot_export_paths(self): self.candidate(2,1,entry='/private/path') self.refresh() self.assertEqual(self.con.execute('SELECT count(*) FROM cue_candidates').fetchone()[0],1) + + def test_export_keeps_custom_source_snapshot_without_default_database(self): + self.refresh() + missing = Path(self.tmp.name) / 'absent-default.sqlite' + with patch.object(cue, 'source_path', return_value=missing): + self.assertEqual(cue.report(self.con)['status'], 'missing_database') + exported = fleet.export(self.con, machine='test')['cue'] + self.assertEqual(exported['status'], 'ok') + self.assertEqual(exported['groups'][0]['arms']['served']['read'], 1) + + def test_oversized_json_rate_is_a_validation_error(self): + self.refresh() + with patch.object(cue, 'source_path', return_value=self.source): + exported = fleet.export(self.con, machine='test') + exported['cue']['groups'][0]['arms']['served']['read_rate'] = 10**400 + # Exercise JSON's real integer decoding before the fleet trust boundary. + with self.assertRaisesRegex(ValueError, 'Cue aggregate'): + fleet.validate(json.loads(json.dumps(exported))) From 3119acf059ca51a7486897322a2b416abddc5479 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 15:00:26 +0200 Subject: [PATCH 4/7] fix(cue): compare first judgments and retain all message boundaries --- CHANGELOG.md | 4 + README.md | 58 +++++++++---- agent_observer/cue.py | 195 +++++++++++++++++++++++++++++++----------- docs/contracts.md | 2 +- tests/test_cue.py | 115 ++++++++++++++++++++++++- 5 files changed, 303 insertions(+), 71 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4edbe72..2f0df71 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Compare first above-threshold Cue judgments per session, entry and epoch; separate repeats and hidden deliveries, retain all user-message window boundaries and report session-cluster intervals. + ### Added - Import Cue decisions and settled ContextMD reads read-only; report procedure reads by entry, harness, machine and AgentsMD version against session-wide holdout, with source and target gaps. diff --git a/README.md b/README.md index 81ccbee..196a7ae 100644 --- a/README.md +++ b/README.md @@ -102,26 +102,48 @@ agent-observer cue report --source /path/to/contextmd.sqlite agent-observer cue report --source /path/to/contextmd.sqlite --json ``` -Reports group Cue entries by harness, machine and observed AgentsMD version -(`unknown` when the ledger cannot identify it). Each arm shows candidate -opportunities, distinct sessions, target reads, full versus partial reads, -and reads already present in that context epoch. A read must be settled and -successful in ContextMD; pending calls are excluded. Matching native read events -are linked by exact session, call id and path, without inserting duplicate events. +The headline compares the **first above-threshold judgment per harness session, +entry and context epoch** in each arm. It does not count repeated injections in +the serving arm against every repeated score in holdout. The first judgment is +chosen in source order, before filtering delivery outcomes or target availability; +a later delivery or entry version cannot replace it. `below_threshold` and +`not_scored` rows do not consume the first judgment. + +Only delivered first judgments enter the served denominator. `hidden` first +judgments have their own excluded group, and other undelivered first judgments +are excluded as `not_served`. Every later above-threshold judgment is reported +under repeats, including serving-enabled `deduped` rows, never in headline rates. +Repeat arm labels describe the session's assignment, not proof of a fresh delivery. +All source outcome counts remain visible. + +The summary gives distinct harness-session counts per arm across all entries, +both observed assignment coverage (`sessions_by_arm`) and sessions contributing +to each first-judgment denominator. Excluded first deliveries can make these differ; +groups split by entry, harness, machine and observed AgentsMD version (`unknown` +when unidentifiable). Each group and summary separates full/partial target reads, +reads already present in that epoch and pending observation windows. Reads must +be settled and successful in ContextMD; pending calls are excluded. Native reads +link by exact session, call id and path without inserting duplicate events. ContextMD-only and native-matched read counts remain visible. -The observation window begins strictly after the moment and ends before the next -user message, context epoch or 30 minutes, whichever comes first. It is clipped -to the imported source watermark; incomplete windows are counted as pending, so -read rates are lower bounds while evidence is still arriving. Full wins over -partial when both occur in a window. Earlier reads are separate and can overlap -a later reread. Repeated candidates produce repeated opportunities, not independent -samples; distinct session counts show the comparison group's size. - -Holdout withholds all Cue entries for the session. Delivery outcomes are not -proof the model saw a rule, and reads do not prove it followed the procedure. -This observational comparison does not establish a per-procedure causal effect -or label a serve `ignored`, as ContextMD's outcome judge does. +The observation window begins strictly after the judgment and ends before the +next user message, known context epoch change or 30 minutes, whichever is first. +Every user message ends it, including rows with missing epoch or holdout metadata; +`boundary_without_cue_metadata` counts windows that needed such a boundary. Missing +epochs do not create invented epoch changes. The window is clipped to the imported +watermark; pending windows make current rates lower bounds. Full wins over partial +within a window; earlier reads are separate and can overlap a later reread. + +95% percentile intervals resample whole observed harness sessions 2,000 times with +a fixed seed, keeping all their entry/epoch opportunities together. They are +unavailable with fewer than two sessions; small samples or all-equal observed +outcomes can yield unstable or degenerate intervals. Distinct session counts must +accompany the rates, particularly for the small holdout sample. + +Holdout withholds all Cue entries for the session. Delivery outcomes do not prove +model receipt, and reads do not prove the agent followed a procedure. These +observed first-judgment rates and intervals do not establish a per-procedure +causal effect or assign the ContextMD outcome judge's `ignored` label. Targets are resolved from version-matching Library files named by the moment's owners, with optional extra directories supplied by `cue import --library DIR`. diff --git a/agent_observer/cue.py b/agent_observer/cue.py index d09fe9c..c4f8d4c 100644 --- a/agent_observer/cue.py +++ b/agent_observer/cue.py @@ -15,13 +15,16 @@ import os from pathlib import Path import re +import random import sqlite3 import time from . import privacy WINDOW_SECONDS = 1800 -SERVED = frozenset({'injected', 'truncated', 'unverified', 'awaiting_evidence', 'hidden'}) +SERVED = frozenset({'injected', 'truncated', 'unverified', 'awaiting_evidence'}) +COMPARISON = 'first_above_threshold_per_session_entry_epoch' +INTERVAL_METHOD = 'session_cluster_bootstrap_percentile_95_2000' FULL = frozenset({'read_tool', 'shell_full'}) KINDS = FULL | {'read_partial', 'shell_partial'} IDENT = re.compile(r'^[A-Za-z0-9_.:/@+-]{1,512}$') @@ -221,20 +224,22 @@ def import_source(con, source=None, libraries=()): for table in ('cue_candidates', 'cue_reads', 'cue_moments', 'cue_import'): con.execute('DELETE FROM ' + table) for m in moments: - if not (_number(m['received_at']) and _id(m['harness']) and _id(m['session']) - and type(m['epoch']) is int and m['holdout'] in (0, 1)): + if not (_number(m['received_at']) and _id(m['harness']) and _id(m['session'])): continue con.execute('INSERT INTO cue_moments VALUES(?,?,?,?,?,?,?,?,?)', - (m['id'],m['received_at'],m['harness'],m['session'],m['epoch'], - m['holdout'],_id(m['kind']),_id(m['tool_use_id']),_id(m['cue_status']))) + (m['id'],m['received_at'],m['harness'],m['session'], + m['epoch'] if type(m['epoch']) is int else None, + m['holdout'] if m['holdout'] in (0, 1) else None, + _id(m['kind']),_id(m['tool_use_id']),_id(m['cue_status']))) ids = {r[0] for r in con.execute('SELECT id FROM cue_moments')} + eligible = {m['id'] for m in moments if type(m['epoch']) is int and m['holdout'] in (0, 1)} columns = {r[1] for r in native.execute('PRAGMA table_info(candidates)')} delivery = 'COALESCE(c.delivery,m.delivery)' if 'delivery' in columns else 'm.delivery' for c in native.execute(f'''SELECT c.id,c.moment_id,c.entry,c.version,c.score,c.outcome, c.stage,c.served_at,d.stage AS delivery_stage FROM candidates c JOIN moments m ON m.id=c.moment_id LEFT JOIN deliveries d ON d.delivery={delivery} WHERE c.matcher='intent' '''): - if (c['moment_id'] not in ids or not isinstance(c['entry'], str) + if (c['moment_id'] not in ids or c['moment_id'] not in eligible or not isinstance(c['entry'], str) or not ENTRY_ID.fullmatch(c['entry']) or not _id(c['version'])): continue target, target_status = moment_targets[c['moment_id']].get((c['entry'], c['version']), @@ -278,37 +283,88 @@ def health(con=None, source=None): return dict(row) if row else {'status': 'not_imported'} +def _cluster_interval(sessions): + """Resample whole harness sessions, keeping their entry/epoch opportunities together.""" + clusters = sorted(sessions.values()) + if len(clusters) < 2: + return None + rng = random.Random(0) + rates = [] + for _ in range(2000): + sample = rng.choices(clusters, k=len(clusters)) + rates.append(sum(r for r, n in sample) / sum(n for r, n in sample)) + rates.sort() + return [rates[49], rates[1949]] + + +def _bucket(): + return {**{k: 0 for k in ARM_COUNTS if k != 'sessions'}, 'sessions': {}} + + +def _add(bucket, session, metrics): + for k, value in metrics.items(): + bucket[k] += value + cluster = bucket['sessions'].setdefault(session, [0, 0]) + cluster[0] += metrics['read'] + cluster[1] += 1 + + +def _finish_buckets(container): + for section in ('arms', 'repeats', 'excluded'): + for bucket in container[section].values(): + bucket['interval95'] = _cluster_interval(bucket['sessions']) + bucket['sessions'] = len(bucket['sessions']) + bucket['read_rate'] = bucket['read'] / bucket['opportunities'] + + def report(con, machine=None, check_source=True, source=None): from .fleet import machine_name state = health(con, source) if check_source else _stored_health(con) result = {'status': state['status'], 'import': state, 'window_seconds': WINDOW_SECONDS, - 'holdout_unit': 'session', - 'interpretation': 'Observational session-wide Cue holdout; read evidence is not proof of following a procedure or a per-procedure causal effect.', - 'window': 'After the moment, before the next user message or context epoch, capped at 30 minutes and the imported source watermark. Earlier reads are separate; incomplete windows are pending.', - 'groups': [], 'gaps': {}} + 'holdout_unit': 'session', 'comparison': COMPARISON, 'interval_method': INTERVAL_METHOD, + 'interpretation': 'Compare first above-threshold judgments per harness session, entry and context epoch. Hidden and other undelivered first judgments are excluded; repeats are separate. Holdout withholds all Cue entries for a session. Read evidence proves neither following a procedure nor a per-procedure causal effect. Small holdout session samples limit precision.', + 'window': 'After the judgment, before every next user message (even with missing Cue metadata) or known context epoch change, capped at 30 minutes and the imported watermark. Earlier reads are separate; incomplete windows are pending.', + 'interval_note': '95% percentile intervals resample whole observed harness sessions 2000 times with a fixed seed. Fewer than two sessions gives no interval; small samples and all-equal outcomes can give unstable or degenerate intervals. These are descriptive, not causal.', + 'summary': {'arms': {}, 'repeats': {}, 'excluded': {}}, + 'sessions_by_arm': {}, 'groups': [], 'gaps': {}} if state['status'] != 'ok': return result moments = {r['id']: dict(r) for r in con.execute('SELECT * FROM cue_moments')} - boundaries = defaultdict(list) + boundaries = {} by_session = defaultdict(list) for m in moments.values(): by_session[(m['harness'],m['session'])].append(m) for key, ms in by_session.items(): prior_epoch = None + flags = {} for m in sorted(ms, key=lambda m: (m['at'], m['id'])): - if m['kind'] == 'user_message' or (prior_epoch is not None and m['epoch'] != prior_epoch): - boundaries[key].append(m['at']) - prior_epoch = m['epoch'] + known_epoch = type(m['epoch']) is int + epoch_changed = known_epoch and prior_epoch is not None and m['epoch'] != prior_epoch + if m['kind'] == 'user_message' or epoch_changed: + missing = not known_epoch or m['holdout'] not in (0, 1) + # If another valid boundary has this timestamp, missing metadata was not needed. + flags[m['at']] = flags.get(m['at'], True) and missing + if known_epoch: + prior_epoch = m['epoch'] + boundaries[key] = (sorted(flags), flags) reads = defaultdict(list) provenance = Counter() for r in con.execute('SELECT * FROM cue_reads ORDER BY at'): reads[(r['harness'],r['session'],r['epoch'],r['path'])].append(dict(r)) provenance['native_and_contextmd' if r['native_event_id'] else 'contextmd_only'] += 1 versions = {r[0]: r[1] or 'unknown' for r in con.execute('SELECT session_key,agentsmd_version FROM sessions')} - groups = {} + groups, seen = {}, set() + coverage = defaultdict(set) gaps = Counter() - for c in con.execute('SELECT * FROM cue_candidates'): + # Source ordering, not import order or the first successful delivery, defines the judgment. + for c in con.execute('SELECT c.* FROM cue_candidates c JOIN cue_moments m ON m.id=c.moment_id ORDER BY m.at,m.id,c.id'): m = moments[c['moment_id']] + above = (c['score'] is not None and c['outcome'] not in ('below_threshold','not_scored') + and type(m['epoch']) is int and m['holdout'] in (0, 1)) + unit = (m['harness'],m['session'],c['entry'],m['epoch']) + first = above and unit not in seen + if above: + seen.add(unit) if not c['target']: gaps[c['target_status']] += 1 continue @@ -316,40 +372,55 @@ def report(con, machine=None, check_source=True, source=None): versions.get(f"{m['harness']}:{m['session']}", 'unknown')) if key not in groups: groups[key] = {'entry': key[0], 'harness': key[1], 'machine': key[2], - 'agentsmd_version': key[3], 'outcomes': Counter(), 'arms': {}} + 'agentsmd_version': key[3], 'outcomes': Counter(), + 'arms': {}, 'repeats': {}, 'excluded': {}} g = groups[key] g['outcomes'][c['outcome']] += 1 - arm = 'holdout' if c['outcome'] == 'holdout' and m['holdout'] == 1 else ( - 'served' if c['outcome'] in SERVED and m['holdout'] == 0 else None) - if arm is None: + if not above: continue - a = g['arms'].setdefault(arm, {'opportunities': 0, 'sessions': set(), 'read': 0, - 'full': 0, 'partial': 0, 'already_read': 0, 'already_full': 0, 'already_partial': 0, - 'pending_window': 0, 'native_matched_reads': 0}) - a['opportunities'] += 1 - a['sessions'].add(m['session']) - bs = boundaries[(m['harness'],m['session'])] + arm = 'holdout' if m['holdout'] == 1 else 'served' + # An inconsistent holdout row cannot turn a serving-enabled session into holdout. + if arm == 'served' and c['outcome'] == 'holdout': + continue + coverage[arm].add((m['harness'],m['session'])) + if not first: + section = 'repeats' + elif arm == 'served' and c['outcome'] not in SERVED: + section, arm = 'excluded', 'hidden' if c['outcome'] == 'hidden' else 'not_served' + elif arm == 'holdout' and c['outcome'] != 'holdout': + section, arm = 'excluded', 'not_served' + else: + section = 'arms' + session = (m['harness'],m['session']) + bs, flags = boundaries[session] index = bisect_right(bs, m['at']) - end = min(m['at'] + WINDOW_SECONDS, bs[index] if index < len(bs) else math.inf) + boundary = bs[index] if index < len(bs) else math.inf + cap = m['at'] + WINDOW_SECONDS + end = min(cap, boundary) watermark = state.get('watermark') or 0 - a['pending_window'] += end > watermark + metrics = {k: 0 for k in ARM_COUNTS if k != 'sessions'} + metrics['opportunities'] = 1 + metrics['pending_window'] = int(end > watermark) + metrics['boundary_without_cue_metadata'] = int(boundary <= cap and flags.get(boundary, False)) end = min(end, math.nextafter(watermark, math.inf)) rs = reads[(m['harness'],m['session'],m['epoch'],c['target'])] times = [r['at'] for r in rs] earlier = rs[:bisect_right(times,m['at'])] if earlier: - a['already_read'] += 1 - a['already_full' if any(r['kind'] in FULL for r in earlier) else 'already_partial'] += 1 + metrics['already_read'] = 1 + metrics['already_full' if any(r['kind'] in FULL for r in earlier) else 'already_partial'] = 1 after = rs[bisect_right(times,m['at']):bisect_left(times,end)] if after: - a['read'] += 1 - a['full' if any(r['kind'] in FULL for r in after) else 'partial'] += 1 - a['native_matched_reads'] += sum(r['native_event_id'] is not None for r in after) + metrics['read'] = 1 + metrics['full' if any(r['kind'] in FULL for r in after) else 'partial'] = 1 + metrics['native_matched_reads'] = sum(r['native_event_id'] is not None for r in after) + for container in (g,result['summary']): + _add(container[section].setdefault(arm,_bucket()),session,metrics) for g in groups.values(): - for a in g['arms'].values(): - a['sessions'] = len(a['sessions']) - a['read_rate'] = a['read'] / a['opportunities'] + _finish_buckets(g) g['outcomes'] = dict(g['outcomes']) + _finish_buckets(result['summary']) + result['sessions_by_arm'] = {arm: len(sessions) for arm,sessions in coverage.items()} result.update(groups=sorted(groups.values(),key=lambda g:(g['entry'],g['harness'],g['agentsmd_version'])), gaps=dict(gaps), read_sources=dict(provenance), cue_statuses=dict(Counter(m['cue_status'] or 'unknown' for m in moments.values()))) @@ -364,22 +435,31 @@ def _stored_health(con): def render(result): - lines = [f"Cue procedure reads: {result['status']}", result['interpretation'], result['window']] + lines = [f"Cue procedure reads: {result['status']}", result['interpretation'], result['window'], result['interval_note']] + lines.append('Observed session assignment coverage: ' + json.dumps(result['sessions_by_arm'],sort_keys=True)) + def buckets(container, prefix=''): + for section in ('arms', 'repeats', 'excluded'): + for arm,a in container[section].items(): + interval = a['interval95'] + ci = (f"{interval[0]:.1%} to {interval[1]:.1%}" if interval is not None else 'unavailable (<2 sessions)') + label = 'first judgments' if section == 'arms' else section + lines.append(f"{prefix}{label} {arm}: {a['read']}/{a['opportunities']} ({a['read_rate']:.1%}) read, {a['sessions']} sessions, 95% interval {ci}; " + f"full {a['full']}, partial {a['partial']}, earlier {a['already_read']}, pending {a['pending_window']}, " + f"boundaries needing missing metadata {a['boundary_without_cue_metadata']}") + buckets(result['summary']) for g in result['groups']: lines.append(f"{g['entry']} | {g['harness']} | {g['machine']} | AgentsMD {g['agentsmd_version']}") - for arm,a in g['arms'].items(): - lines.append(f" {arm}: {a['read']}/{a['opportunities']} read, {a['sessions']} sessions; " - f"full {a['full']}, partial {a['partial']}, earlier {a['already_read']}, " - f"pending windows {a['pending_window']}") + buckets(g, ' ') if result['gaps']: lines.append('Target gaps: ' + json.dumps(result['gaps'],sort_keys=True)) return '\n'.join(lines) EXPORT_FIELDS = frozenset({'status', 'window_seconds', 'holdout_unit', 'groups', - 'gaps', 'read_sources', 'cue_statuses'}) + 'gaps', 'read_sources', 'cue_statuses', 'summary', 'comparison', 'interval_method', + 'sessions_by_arm'}) ARM_COUNTS = frozenset({'opportunities', 'sessions', 'read', 'full', 'partial', - 'already_read', 'already_full', 'already_partial', 'pending_window', 'native_matched_reads'}) + 'already_read', 'already_full', 'already_partial', 'pending_window', 'native_matched_reads', 'boundary_without_cue_metadata'}) def export(con, machine=None): @@ -403,28 +483,41 @@ def counts(mapping, allowed=None): and (allowed is None or k in allowed) and type(v) is int and 0 <= v <= 2**53) + def validate_buckets(container): + require(isinstance(container, dict)) + for section, allowed in (('arms', {'served','holdout'}), ('repeats', {'served','holdout'}), + ('excluded', {'hidden','not_served'})): + require(isinstance(container.get(section), dict) and set(container[section]) <= allowed) + for arm in container[section].values(): + require(isinstance(arm, dict) and set(arm) == ARM_COUNTS | {'read_rate','interval95'}) + counts({k: arm[k] for k in ARM_COUNTS}, ARM_COUNTS) + require(_number(arm['read_rate']) and 0 <= arm['read_rate'] <= 1) + interval = arm['interval95'] + require(interval is None or (isinstance(interval, list) and len(interval) == 2 + and all(_number(v) and 0 <= v <= 1 for v in interval) and interval[0] <= interval[1])) + require(arm['read'] == arm['full'] + arm['partial'] <= arm['opportunities']) + require(arm['already_read'] == arm['already_full'] + arm['already_partial'] <= arm['opportunities']) + require(isinstance(value, dict) and set(value) == EXPORT_FIELDS) require(isinstance(value['status'], str) and value['status'] in {'ok','missing_database','schema_drift','no_cue', 'unreadable_database','not_imported'}) require(value['window_seconds'] == WINDOW_SECONDS and value['holdout_unit'] == 'session') + require(value['comparison'] == COMPARISON and value['interval_method'] == INTERVAL_METHOD) + require(isinstance(value['summary'], dict) and set(value['summary']) == {'arms','repeats','excluded'}) + counts(value['sessions_by_arm'], {'served','holdout'}) counts(value['gaps'], {'version_unavailable','no_target','ambiguous_target'}) counts(value['read_sources'], {'native_and_contextmd','contextmd_only'}) counts(value['cue_statuses']) require(isinstance(value['groups'], list)) for g in value['groups']: require(isinstance(g, dict) and set(g) == {'entry','harness','machine', - 'agentsmd_version','outcomes','arms'}) + 'agentsmd_version','outcomes','arms','repeats','excluded'}) require(isinstance(g['entry'], str) and ENTRY_ID.fullmatch(g['entry'])) require(isinstance(g['harness'], str) and g['harness'] in {'codex','claude','grok','opencode'}) require(isinstance(g['machine'], str) and re.fullmatch(r'[a-z0-9][a-z0-9-]{0,62}', g['machine'])) require(isinstance(g['agentsmd_version'], str) and re.fullmatch(r'(?:[0-9]+\.[0-9]+\.[0-9]+|unknown)', g['agentsmd_version'])) counts(g['outcomes']) - require(isinstance(g['arms'], dict) and set(g['arms']) <= {'served','holdout'}) - for arm in g['arms'].values(): - require(isinstance(arm, dict) and set(arm) == ARM_COUNTS | {'read_rate'}) - counts({k: arm[k] for k in ARM_COUNTS}, ARM_COUNTS) - require(_number(arm['read_rate']) and 0 <= arm['read_rate'] <= 1) - require(arm['read'] == arm['full'] + arm['partial'] <= arm['opportunities']) - require(arm['already_read'] == arm['already_full'] + arm['already_partial'] <= arm['opportunities']) + validate_buckets(g) + validate_buckets(value['summary']) return value diff --git a/docs/contracts.md b/docs/contracts.md index 2d88b93..093cdfe 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -37,7 +37,7 @@ are committed. | `router_jobs`, `router_invocations`, `router_readings` | router ids | Model Router ledger rows, copied read-only. | | `agentsmd_versions` | AGENTS.md SHA-256 | Release map from the local AgentsMD tags. | | `cue_import` | one snapshot | ContextMD import status, time and source watermark; unavailable evidence is a gap. | -| `cue_moments`, `cue_candidates` | source moment id, candidate id | Cue decision metadata, scores, all candidate outcomes and delivery stages; version-matched targets, never prompt, window or served text. | +| `cue_moments`, `cue_candidates` | source moment id, candidate id | Moment metadata including user-message boundaries with unknown epoch/arm, Cue scores, all candidate outcomes and delivery stages; version-matched targets, never prompt, window or served text. | | `cue_reads` | harness, session, epoch, path, kind, moment, time | Settled successful ContextMD reads with optional exact native event linkage; no duplicate native events are inserted. | | `import_errors` | none | Quarantined records; prior valid data is kept. | diff --git a/tests/test_cue.py b/tests/test_cue.py index f2a7ee0..42e47bc 100644 --- a/tests/test_cue.py +++ b/tests/test_cue.py @@ -86,7 +86,8 @@ def test_holdout_is_session_wide_and_counts_sessions(self): result = self.refresh() self.assertEqual(result['holdout_unit'],'session') arm = self.arm(result,'holdout') - self.assertEqual((arm['opportunities'],arm['sessions'],arm['partial']),(2,1,2)) + self.assertEqual((arm['opportunities'],arm['sessions'],arm['partial']),(1,1,1)) + self.assertEqual(result['groups'][0]['repeats']['holdout']['opportunities'],1) self.assertIn('causal', result['interpretation']) def test_pending_failed_and_duplicate_reads_not_counted(self): @@ -237,3 +238,115 @@ def test_oversized_json_rate_is_a_validation_error(self): # Exercise JSON's real integer decoding before the fleet trust boundary. with self.assertRaisesRegex(ValueError, 'Cue aggregate'): fleet.validate(json.loads(json.dumps(exported))) + + def test_headline_compares_first_judgments_not_repeated_deliveries(self): + for id,at in ((4,120),(5,130)): + self.moment(id,at) + self.candidate(id,id,'deduped' if id==4 else 'injected') + for id,at in ((6,100),(7,120),(8,130)): + self.moment(id,at,session='h',holdout=1) + self.candidate(id,id,'holdout') + self.moment(9,110,session='h',holdout=1) + self.read(9,110,session='h') + result=self.refresh() + for arm in ('served','holdout'): + self.assertEqual(self.arm(result,arm)['opportunities'],1) + self.assertEqual(self.arm(result,arm)['read'],1) + self.assertEqual(result['groups'][0]['repeats'][arm]['opportunities'],2) + self.assertEqual(result['summary']['arms']['holdout']['sessions'],1) + + def test_hidden_first_judgment_cannot_be_promoted_by_later_injection(self): + self.native.execute("UPDATE candidates SET outcome='hidden' WHERE id=1") + self.moment(4,120) + self.candidate(4,4,'injected') + result=self.refresh() + self.assertNotIn('served',result['groups'][0]['arms']) + self.assertEqual(result['groups'][0]['excluded']['hidden']['opportunities'],1) + self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],1) + + def test_first_judgment_resets_per_epoch_and_ignores_below_threshold(self): + self.moment(4,90) + self.candidate(4,4,'below_threshold') + self.moment(5,120,epoch=1) + self.candidate(5,5,'injected') + self.moment(6,130,epoch=1) + self.candidate(6,6,'deduped') + result=self.refresh() + self.assertEqual(self.arm(result)['opportunities'],2) + self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],1) + + def test_user_message_boundary_survives_missing_cue_metadata(self): + for arm,assignment,outcome in (('served',0,'injected'),('holdout',1,'holdout')): + self.native.execute('UPDATE moments SET holdout=? WHERE id=1',(assignment,)) + self.native.execute('UPDATE candidates SET outcome=? WHERE id=1',(outcome,)) + for epoch,holdout in ((None,0),(0,None),(None,None),('unknown',None)): + with self.subTest(arm=arm,epoch=epoch,holdout=holdout): + self.native.execute('DELETE FROM moments WHERE id=4') + self.moment(4,105,epoch=epoch,holdout=holdout,kind='user_message') + result=self.refresh() + self.assertEqual(self.arm(result,arm)['read'],0) + self.assertEqual(self.arm(result,arm)['boundary_without_cue_metadata'],1) + + def test_summary_sessions_are_distinct_across_entries(self): + # Same session reads two different entry targets; the headline has one session. + other=self.raw.replace(b'id: research',b'id: other') + (self.library/'other.md').write_bytes(other) + self.candidate(4,1,entry='other',version=hashlib.sha256(other).hexdigest()[:12]) + result=self.refresh() + self.assertEqual(result['summary']['arms']['served']['opportunities'],2) + self.assertEqual(result['summary']['arms']['served']['sessions'],1) + self.assertEqual(result['sessions_by_arm']['served'],1) + + def test_cluster_interval_uses_sessions_and_requires_two(self): + self.assertIsNone(self.arm(self.refresh())['interval95']) + self.moment(4,100,session='no-read') + self.candidate(4,4) + result=self.refresh() + self.assertEqual(self.arm(result)['sessions'],2) + # Resampling two one-opportunity sessions can yield all failure or all success. + self.assertEqual(self.arm(result)['interval95'],[0.0,1.0]) + self.assertIn('session_cluster',result['interval_method']) + + def test_first_unavailable_target_is_not_replaced_by_later_delivery(self): + self.native.execute("UPDATE candidates SET version='unavailable' WHERE id=1") + self.moment(4,120) + self.candidate(4,4) + result=self.refresh() + self.assertEqual(result['groups'][0]['arms'],{}) + self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],1) + self.assertEqual(result['gaps'],{'version_unavailable':1}) + + def test_first_undelivered_judgment_is_not_replaced_by_later_delivery(self): + self.native.execute("UPDATE candidates SET outcome='cut_budget' WHERE id=1") + self.moment(4,120) + self.candidate(4,4) + result=self.refresh() + self.assertEqual(result['groups'][0]['arms'],{}) + self.assertEqual(result['groups'][0]['excluded']['not_served']['opportunities'],1) + self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],1) + + def test_later_missing_metadata_boundary_does_not_count_when_not_needed(self): + self.moment(4,105,kind='user_message') + self.moment(5,106,epoch=None,holdout=None,kind='user_message') + arm=self.arm(self.refresh()) + self.assertEqual(arm['read'],0) + self.assertEqual(arm['boundary_without_cue_metadata'],0) + + def test_holdout_gets_a_new_first_judgment_after_compaction(self): + self.native.execute("UPDATE moments SET holdout=1 WHERE id=1") + self.native.execute("UPDATE candidates SET outcome='holdout' WHERE id=1") + self.moment(4,120,epoch=1,holdout=1,kind='compaction') + self.candidate(4,4,'holdout') + self.moment(5,130,epoch=1,holdout=1) + self.candidate(5,5,'holdout') + result=self.refresh() + self.assertEqual(self.arm(result,'holdout')['opportunities'],2) + self.assertEqual(result['groups'][0]['repeats']['holdout']['opportunities'],1) + + def test_assignment_session_count_includes_excluded_first_deliveries(self): + self.moment(4,100,session='hidden') + self.candidate(4,4,'hidden') + result=self.refresh() + self.assertEqual(result['sessions_by_arm']['served'],2) + self.assertEqual(self.arm(result)['sessions'],1) + self.assertEqual(result['summary']['excluded']['hidden']['sessions'],1) From d06c150b4a2f93c4bcebb148289c6a298268b956 Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 15:26:13 +0200 Subject: [PATCH 5/7] fix(cue): resolve first carried units at eventual delivery --- CHANGELOG.md | 1 + README.md | 23 ++++++++++---- agent_observer/cue.py | 74 ++++++++++++++++++++++++++++++++++++------- tests/test_cue.py | 60 +++++++++++++++++++++++++++++++++++ 4 files changed, 140 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f0df71..8654d72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Fixed +- Resolve Grok carry units at their eventual print, start read windows at delivery, separate unprinted carries and expose headline rates by harness. - Compare first above-threshold Cue judgments per session, entry and epoch; separate repeats and hidden deliveries, retain all user-message window boundaries and report session-cluster intervals. ### Added diff --git a/README.md b/README.md index 196a7ae..f10bf24 100644 --- a/README.md +++ b/README.md @@ -106,19 +106,26 @@ The headline compares the **first above-threshold judgment per harness session, entry and context epoch** in each arm. It does not count repeated injections in the serving arm against every repeated score in holdout. The first judgment is chosen in source order, before filtering delivery outcomes or target availability; -a later delivery or entry version cannot replace it. `below_threshold` and -`not_scored` rows do not consume the first judgment. +a later entry version cannot replace it. A first `carried` judgment uses its +eventual delivery in the same epoch, even after intervening `deduped` or +`carry_expired` rows. That print belongs to the original unit. First `hidden`, +`cut_budget` or missing-target judgments still consume the slot. `below_threshold` +and `not_scored` rows do not consume the first judgment. Only delivered first judgments enter the served denominator. `hidden` first judgments have their own excluded group, and other undelivered first judgments -are excluded as `not_served`. Every later above-threshold judgment is reported -under repeats, including serving-enabled `deduped` rows, never in headline rates. +are excluded as `not_served`. Carries that never print have their own excluded +`carry_unprinted` group. Later above-threshold judgments, except a resolved +carry's print, are reported under repeats, including serving-enabled `deduped` +rows, never in headline rates. `carry_prints` counts resolved carries. Repeat arm labels describe the session's assignment, not proof of a fresh delivery. All source outcome counts remain visible. The summary gives distinct harness-session counts per arm across all entries, both observed assignment coverage (`sessions_by_arm`) and sessions contributing -to each first-judgment denominator. Excluded first deliveries can make these differ; +to each first-judgment denominator. The `by_harness` summary reports both arms +for Claude, Codex and Grok separately so differences in harness mix stay visible. +Excluded first deliveries can make observed and contributing session counts differ; groups split by entry, harness, machine and observed AgentsMD version (`unknown` when unidentifiable). Each group and summary separates full/partial target reads, reads already present in that epoch and pending observation windows. Reads must @@ -126,8 +133,10 @@ be settled and successful in ContextMD; pending calls are excluded. Native reads link by exact session, call id and path without inserting duplicate events. ContextMD-only and native-matched read counts remain visible. -The observation window begins strictly after the judgment and ends before the -next user message, known context epoch change or 30 minutes, whichever is first. +The observation window begins strictly after actual delivery, at the print moment +for a resolved carry. Holdout and excluded diagnostics use the first judgment. +It ends before the next user message, known context epoch change or 30 minutes, +whichever is first. Every user message ends it, including rows with missing epoch or holdout metadata; `boundary_without_cue_metadata` counts windows that needed such a boundary. Missing epochs do not create invented epoch changes. The window is clipped to the imported diff --git a/agent_observer/cue.py b/agent_observer/cue.py index c4f8d4c..aa7ecf9 100644 --- a/agent_observer/cue.py +++ b/agent_observer/cue.py @@ -317,16 +317,40 @@ def _finish_buckets(container): bucket['read_rate'] = bucket['read'] / bucket['opportunities'] +def _carry_prints(con): + """Pair a unit's first carried judgment with its first eventual delivery. + + A queue row is not a delivery. Intermediate deduplication, expiration and + further scoring rows do not replace the unit; the print may be much later. + Other first outcomes and unavailable first targets still consume the slot. + """ + first, pending, resolved, printed = set(), {}, {}, set() + for c in con.execute('''SELECT c.*,m.harness,m.session,m.epoch,m.holdout + FROM cue_candidates c JOIN cue_moments m ON m.id=c.moment_id + WHERE c.score IS NOT NULL AND c.outcome NOT IN ('below_threshold','not_scored') + AND typeof(m.epoch)='integer' AND m.holdout IN (0,1) + ORDER BY m.at,m.id,c.id'''): + unit = (c['harness'],c['session'],c['entry'],c['epoch']) + if unit not in first: + first.add(unit) + if c['outcome'] == 'carried' and c['target'] and c['holdout'] == 0: + pending[unit] = c['id'] + elif unit in pending and c['holdout'] == 0 and c['outcome'] in SERVED | {'hidden'}: + resolved[pending.pop(unit)] = c + printed.add(c['id']) + return resolved, printed + + def report(con, machine=None, check_source=True, source=None): from .fleet import machine_name state = health(con, source) if check_source else _stored_health(con) result = {'status': state['status'], 'import': state, 'window_seconds': WINDOW_SECONDS, 'holdout_unit': 'session', 'comparison': COMPARISON, 'interval_method': INTERVAL_METHOD, - 'interpretation': 'Compare first above-threshold judgments per harness session, entry and context epoch. Hidden and other undelivered first judgments are excluded; repeats are separate. Holdout withholds all Cue entries for a session. Read evidence proves neither following a procedure nor a per-procedure causal effect. Small holdout session samples limit precision.', - 'window': 'After the judgment, before every next user message (even with missing Cue metadata) or known context epoch change, capped at 30 minutes and the imported watermark. Earlier reads are separate; incomplete windows are pending.', + 'interpretation': 'Compare first above-threshold units per harness session, entry and context epoch. A first carried judgment uses its eventual print in that epoch; a carry never printed is excluded separately. Hidden and other undelivered first judgments are excluded; repeats are separate. Holdout withholds all Cue entries for a session. Read evidence proves neither following a procedure nor a per-procedure causal effect. Per-harness headlines expose arm mix differences; small holdout samples limit precision.', + 'window': 'After actual delivery (the print moment for a resolved carry), or the first judgment for holdout/excluded diagnostics, before every next user message (even with missing Cue metadata) or known context epoch change, capped at 30 minutes and the imported watermark. Earlier reads are separate; incomplete windows are pending.', 'interval_note': '95% percentile intervals resample whole observed harness sessions 2000 times with a fixed seed. Fewer than two sessions gives no interval; small samples and all-equal outcomes can give unstable or degenerate intervals. These are descriptive, not causal.', 'summary': {'arms': {}, 'repeats': {}, 'excluded': {}}, - 'sessions_by_arm': {}, 'groups': [], 'gaps': {}} + 'sessions_by_arm': {}, 'by_harness': {}, 'groups': [], 'gaps': {}} if state['status'] != 'ok': return result moments = {r['id']: dict(r) for r in con.execute('SELECT * FROM cue_moments')} @@ -354,6 +378,7 @@ def report(con, machine=None, check_source=True, source=None): provenance['native_and_contextmd' if r['native_event_id'] else 'contextmd_only'] += 1 versions = {r[0]: r[1] or 'unknown' for r in con.execute('SELECT session_key,agentsmd_version FROM sessions')} groups, seen = {}, set() + carry_prints, printed = _carry_prints(con) coverage = defaultdict(set) gaps = Counter() # Source ordering, not import order or the first successful delivery, defines the judgment. @@ -383,14 +408,26 @@ def report(con, machine=None, check_source=True, source=None): if arm == 'served' and c['outcome'] == 'holdout': continue coverage[arm].add((m['harness'],m['session'])) + harness_summary = result['by_harness'].setdefault(m['harness'], + {'arms': {}, 'repeats': {}, 'excluded': {}, 'sessions_by_arm': {}}) + # The print belongs to its original first unit, not to a second opportunity. + if c['id'] in printed: + continue + observation = carry_prints.get(c['id'], c) + if not observation['target']: + continue # The print's source row records its target gap exactly once. + outcome = observation['outcome'] if not first: section = 'repeats' - elif arm == 'served' and c['outcome'] not in SERVED: - section, arm = 'excluded', 'hidden' if c['outcome'] == 'hidden' else 'not_served' + elif arm == 'served' and outcome not in SERVED: + section = 'excluded' + arm = ('hidden' if outcome == 'hidden' else 'carry_unprinted' + if outcome in ('carried','carry_expired') else 'not_served') elif arm == 'holdout' and c['outcome'] != 'holdout': section, arm = 'excluded', 'not_served' else: section = 'arms' + m = moments[observation['moment_id']] session = (m['harness'],m['session']) bs, flags = boundaries[session] index = bisect_right(bs, m['at']) @@ -400,10 +437,11 @@ def report(con, machine=None, check_source=True, source=None): watermark = state.get('watermark') or 0 metrics = {k: 0 for k in ARM_COUNTS if k != 'sessions'} metrics['opportunities'] = 1 + metrics['carry_prints'] = int(c['id'] in carry_prints) metrics['pending_window'] = int(end > watermark) metrics['boundary_without_cue_metadata'] = int(boundary <= cap and flags.get(boundary, False)) end = min(end, math.nextafter(watermark, math.inf)) - rs = reads[(m['harness'],m['session'],m['epoch'],c['target'])] + rs = reads[(m['harness'],m['session'],m['epoch'],observation['target'])] times = [r['at'] for r in rs] earlier = rs[:bisect_right(times,m['at'])] if earlier: @@ -414,13 +452,18 @@ def report(con, machine=None, check_source=True, source=None): metrics['read'] = 1 metrics['full' if any(r['kind'] in FULL for r in after) else 'partial'] = 1 metrics['native_matched_reads'] = sum(r['native_event_id'] is not None for r in after) - for container in (g,result['summary']): + for container in (g,result['summary'],harness_summary): _add(container[section].setdefault(arm,_bucket()),session,metrics) for g in groups.values(): _finish_buckets(g) g['outcomes'] = dict(g['outcomes']) _finish_buckets(result['summary']) result['sessions_by_arm'] = {arm: len(sessions) for arm,sessions in coverage.items()} + for harness,container in result['by_harness'].items(): + _finish_buckets(container) + container['sessions_by_arm'] = { + arm: sum(h == harness for h,s in sessions) for arm,sessions in coverage.items() + if any(h == harness for h,s in sessions)} result.update(groups=sorted(groups.values(),key=lambda g:(g['entry'],g['harness'],g['agentsmd_version'])), gaps=dict(gaps), read_sources=dict(provenance), cue_statuses=dict(Counter(m['cue_status'] or 'unknown' for m in moments.values()))) @@ -445,8 +488,11 @@ def buckets(container, prefix=''): label = 'first judgments' if section == 'arms' else section lines.append(f"{prefix}{label} {arm}: {a['read']}/{a['opportunities']} ({a['read_rate']:.1%}) read, {a['sessions']} sessions, 95% interval {ci}; " f"full {a['full']}, partial {a['partial']}, earlier {a['already_read']}, pending {a['pending_window']}, " - f"boundaries needing missing metadata {a['boundary_without_cue_metadata']}") + f"carry prints {a['carry_prints']}, boundaries needing missing metadata {a['boundary_without_cue_metadata']}") buckets(result['summary']) + for harness,container in sorted(result['by_harness'].items()): + lines.append(f"Harness {harness}; observed sessions: {json.dumps(container['sessions_by_arm'],sort_keys=True)}") + buckets(container, ' ') for g in result['groups']: lines.append(f"{g['entry']} | {g['harness']} | {g['machine']} | AgentsMD {g['agentsmd_version']}") buckets(g, ' ') @@ -457,9 +503,9 @@ def buckets(container, prefix=''): EXPORT_FIELDS = frozenset({'status', 'window_seconds', 'holdout_unit', 'groups', 'gaps', 'read_sources', 'cue_statuses', 'summary', 'comparison', 'interval_method', - 'sessions_by_arm'}) + 'sessions_by_arm', 'by_harness'}) ARM_COUNTS = frozenset({'opportunities', 'sessions', 'read', 'full', 'partial', - 'already_read', 'already_full', 'already_partial', 'pending_window', 'native_matched_reads', 'boundary_without_cue_metadata'}) + 'already_read', 'already_full', 'already_partial', 'pending_window', 'native_matched_reads', 'boundary_without_cue_metadata', 'carry_prints'}) def export(con, machine=None): @@ -486,7 +532,7 @@ def counts(mapping, allowed=None): def validate_buckets(container): require(isinstance(container, dict)) for section, allowed in (('arms', {'served','holdout'}), ('repeats', {'served','holdout'}), - ('excluded', {'hidden','not_served'})): + ('excluded', {'hidden','not_served','carry_unprinted'})): require(isinstance(container.get(section), dict) and set(container[section]) <= allowed) for arm in container[section].values(): require(isinstance(arm, dict) and set(arm) == ARM_COUNTS | {'read_rate','interval95'}) @@ -505,6 +551,12 @@ def validate_buckets(container): require(value['comparison'] == COMPARISON and value['interval_method'] == INTERVAL_METHOD) require(isinstance(value['summary'], dict) and set(value['summary']) == {'arms','repeats','excluded'}) counts(value['sessions_by_arm'], {'served','holdout'}) + require(isinstance(value['by_harness'], dict) + and set(value['by_harness']) <= {'codex','claude','grok','opencode'}) + for container in value['by_harness'].values(): + require(isinstance(container, dict) and set(container) == {'arms','repeats','excluded','sessions_by_arm'}) + counts(container['sessions_by_arm'], {'served','holdout'}) + validate_buckets(container) counts(value['gaps'], {'version_unavailable','no_target','ambiguous_target'}) counts(value['read_sources'], {'native_and_contextmd','contextmd_only'}) counts(value['cue_statuses']) diff --git a/tests/test_cue.py b/tests/test_cue.py index 42e47bc..f3a4366 100644 --- a/tests/test_cue.py +++ b/tests/test_cue.py @@ -350,3 +350,63 @@ def test_assignment_session_count_includes_excluded_first_deliveries(self): self.assertEqual(result['sessions_by_arm']['served'],2) self.assertEqual(self.arm(result)['sessions'],1) self.assertEqual(result['summary']['excluded']['hidden']['sessions'],1) + + def _carry_fixture(self, intermediates=(), printed=True, print_epoch=0): + self.native.execute("UPDATE moments SET harness='grok'") + self.native.execute("UPDATE reads SET harness='grok'") + self.native.execute("UPDATE candidates SET outcome='carried' WHERE id=1") + self.native.execute('DELETE FROM reads') + for i,outcome in enumerate(intermediates,4): + self.moment(i,120+i) + self.candidate(i,i,outcome) + if printed: + self.moment(20,3000,epoch=print_epoch,kind='pre_tool') + self.candidate(20,20,'injected') + self.moment(21,3010,epoch=print_epoch,kind='tool_result') + self.read(21,3010,epoch=print_epoch) + self.moment(22,5000,epoch=print_epoch) + self.native.execute("UPDATE moments SET harness='grok'") + self.native.execute("UPDATE reads SET harness='grok'") + + def test_carry_print_at_second_row_supplies_first_delivery_window(self): + self._carry_fixture() + result=self.refresh() + self.assertIn('served',result['groups'][0]['arms']) + arm=self.arm(result) + # The print is beyond the first score's 30-minute cap; only its own window sees this read. + self.assertEqual((arm['opportunities'],arm['read'],arm['already_read']),(1,1,0)) + self.assertEqual(arm['carry_prints'],1) + self.assertEqual(result['groups'][0]['repeats'],{}) + self.assertEqual(result['by_harness']['grok']['arms']['served']['read'],1) + + def test_carry_print_after_deduped_and_expired_rows_is_still_first_delivery(self): + self._carry_fixture(('deduped','carry_expired','deduped','carried')) + result=self.refresh() + self.assertIn('served',result['groups'][0]['arms']) + self.assertEqual((self.arm(result)['opportunities'],self.arm(result)['read']),(1,1)) + self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],4) + self.assertEqual(self.arm(result)['carry_prints'],1) + + def test_carry_never_printed_is_its_own_excluded_group(self): + self._carry_fixture(('deduped','carry_expired'),printed=False) + result=self.refresh() + self.assertNotIn('served',result['groups'][0]['arms']) + self.assertIn('carry_unprinted',result['summary']['excluded']) + self.assertEqual(result['summary']['excluded']['carry_unprinted']['opportunities'],1) + self.assertEqual(result['summary']['excluded']['carry_unprinted']['carry_prints'],0) + self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],2) + + def test_carry_print_in_another_epoch_does_not_resolve_the_original(self): + self._carry_fixture(print_epoch=1) + result=self.refresh() + self.assertEqual(self.arm(result)['opportunities'],1) + self.assertEqual(self.arm(result)['carry_prints'],0) + self.assertEqual(result['summary']['excluded']['carry_unprinted']['opportunities'],1) + + def test_missing_target_on_first_carry_still_takes_the_slot(self): + self._carry_fixture() + self.native.execute("UPDATE candidates SET version='unavailable' WHERE id=1") + result=self.refresh() + self.assertEqual(result['groups'][0]['arms'],{}) + self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],1) + self.assertEqual(result['gaps'],{'version_unavailable':1}) From adda27b5ce62287ae46d73029b5455e6a2f6d1bd Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 15:57:49 +0200 Subject: [PATCH 6/7] fix(cue): isolate optional source gaps and import failures --- CHANGELOG.md | 6 ++-- README.md | 13 ++++++-- agent_observer/cli.py | 5 ++-- agent_observer/cue.py | 41 ++++++++++++++++++++----- tests/test_cue.py | 70 +++++++++++++++++++++++++++++++++++++++++++ tests/test_health.py | 21 +++++++++++++ 6 files changed, 140 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8654d72..2ffe4d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,10 +2,10 @@ ## [Unreleased] -### Fixed +### Changed -- Resolve Grok carry units at their eventual print, start read windows at delivery, separate unprinted carries and expose headline rates by harness. -- Compare first above-threshold Cue judgments per session, entry and epoch; separate repeats and hidden deliveries, retain all user-message window boundaries and report session-cluster intervals. +- `sync` now reads ContextMD Cue evidence by default alongside the T3 import; a failed Cue import records a gap without discarding other sync writes. +- `health` shows optional Cue gaps without failing; Cue status causes exit 4 only for `schema_drift` or `unreadable_database`. ### Added diff --git a/README.md b/README.md index f10bf24..0908f07 100644 --- a/README.md +++ b/README.md @@ -159,7 +159,12 @@ owners, with optional extra directories supplied by `cue import --library DIR`. Changed or missing entry versions, ambiguous targets and entries without targets are explicit gaps, not unread procedures. Missing databases, incompatible schemas and machines without successful Cue moments show gaps in both `health` and the -report. `cue` exits 4 for unavailable evidence; `health` also exits 4 for a Cue gap. +report. `cue` exits 4 for unavailable evidence. Optional Cue gaps (`missing_database`, +`no_cue`, `not_imported`) do not fail `health`; Cue causes exit 4 only for +`schema_drift` or `unreadable_database`. A malformed or unreadable Library file +leaves only its own candidates unresolved. `sync` imports Cue by default with T3; +a failed Cue import rolls back only Cue changes, records a gap and finishes the +remaining sync work before returning exit 1 for schema/read failures. No prompt, window, entry body or served text is imported. `export` carries only the imported snapshot's aggregates, without read paths, session ids or individual moments; it does not require the original database to remain available. @@ -655,8 +660,10 @@ the process, directory and fix. Missing, failed, malformed or timed-out process scans are reported as unavailable with a reason; each `ps` or `lsof` invocation has a five-second timeout, and other health checks continue. -Health exits 0 when it finds no invisible sessions or unavailable scans, 4 when -either is reported, and 2 for invalid command-line arguments. An unread ledger +Health exits 0 when it finds no invisible sessions, unavailable scans or Cue +schema/read failures; it exits 4 when any is reported, and 2 for invalid +command-line arguments. Optional missing or unsynced Cue evidence stays visible +without failing health. An unread ledger is reported as unknown and does not by itself change the exit status. ## Ledger schema 3 diff --git a/agent_observer/cli.py b/agent_observer/cli.py index 4b0d6ef..62f0470 100644 --- a/agent_observer/cli.py +++ b/agent_observer/cli.py @@ -755,7 +755,7 @@ def _health_check(ns) -> int: for harness, roots in result["bare_roots"].items(): for line in _bare_lines(harness, roots): print(f"- {harness} {line.strip()}") - return 4 if result["findings"] or result["unavailable"] or result["cue"]["status"] != "ok" else 0 + return 4 if result["findings"] or result["unavailable"] or result["cue"]["status"] in {"schema_drift", "unreadable_database"} else 0 def _capabilities(ns) -> int: @@ -1075,7 +1075,8 @@ def _sync(con, ns) -> int: if cue_result is not None: payload["cue"] = cue_result _emit(payload, ns.as_json) - return 1 if any(r.get("failed") for r in results) else 0 + return 1 if (any(r.get("failed") for r in results) or + cue_result and cue_result["status"] in {"schema_drift", "unreadable_database"}) else 0 def _when(value): diff --git a/agent_observer/cue.py b/agent_observer/cue.py index aa7ecf9..6a6daf4 100644 --- a/agent_observer/cue.py +++ b/agent_observer/cue.py @@ -116,8 +116,11 @@ def _targets(roots): found = {} for root in roots: try: - paths = Path(root).glob('*.md') - for file in paths: + paths = list(Path(root).glob('*.md')) + except OSError: + continue + for file in paths: + try: if file.name.lower() == 'readme.md': continue raw = file.read_bytes() @@ -152,8 +155,9 @@ def _targets(roots): found[key] = (None, 'ambiguous_target') else: found[key] = value - except (OSError, ValueError, UnicodeError): - continue + except (OSError, ValueError, UnicodeError): + # Only candidates for this file remain unresolved, not its siblings. + continue return found @@ -177,14 +181,35 @@ def _native_reads(con): return found +def _record_gap(con, status): + con.execute('DELETE FROM cue_import') + con.execute('INSERT INTO cue_import VALUES(?,?,NULL)', (status, time.time())) + return {'status': status} + + def import_source(con, source=None, libraries=()): - """Import one consistent source snapshot. A failed refresh preserves prior data.""" + """Isolate Cue failure from other pending imports, placement and identity writes.""" _ensure(con) + con.execute('SAVEPOINT cue_attempt') + try: + return _import_snapshot(con, source, libraries) + except (sqlite3.Error, TypeError) as exc: + con.execute('ROLLBACK TO cue_attempt') + # Exception messages can contain source data; retain only a safe gap category. + return _record_gap(con, 'schema_drift' if isinstance(exc, TypeError) + else 'unreadable_database') + except BaseException: + con.execute('ROLLBACK TO cue_attempt') + raise + finally: + con.execute('RELEASE cue_attempt') + + +def _import_snapshot(con, source, libraries): + """Import one consistent source snapshot. A failed refresh preserves prior data.""" status = source_status(source) if status != 'ok': - con.execute('DELETE FROM cue_import') - con.execute('INSERT INTO cue_import VALUES(?,?,NULL)', (status, time.time())) - return {'status': status} + return _record_gap(con, status) with closing(_open(source)) as native: native.execute('BEGIN') moments = native.execute('SELECT id,received_at,harness,session,epoch,holdout,kind,' diff --git a/tests/test_cue.py b/tests/test_cue.py index f3a4366..132b316 100644 --- a/tests/test_cue.py +++ b/tests/test_cue.py @@ -4,6 +4,7 @@ import io import json from pathlib import Path +from types import SimpleNamespace import sqlite3 from unittest.mock import patch @@ -410,3 +411,72 @@ def test_missing_target_on_first_carry_still_takes_the_slot(self): self.assertEqual(result['groups'][0]['arms'],{}) self.assertEqual(result['groups'][0]['repeats']['served']['opportunities'],1) self.assertEqual(result['gaps'],{'version_unavailable':1}) + + def test_bad_library_file_does_not_hide_healthy_targets(self): + bad=self.library/'bad.md' + for raw in (b'---\nid: bad\ntarget: "unterminated\n---\n',b'\xff'): + with self.subTest(raw=raw): + bad.write_bytes(raw) + version=hashlib.sha256(raw).hexdigest()[:12] + self.native.execute("DELETE FROM candidates WHERE id=4") + self.candidate(4,1,entry='bad',version=version) + # Filesystem enumeration order must not decide which entries survive. + with patch.object(Path,'glob',return_value=iter([bad,self.library/'research.md'])): + result=self.refresh() + self.assertEqual(self.arm(result)['opportunities'],1) + self.assertEqual(result['gaps'],{'version_unavailable':1}) + + def test_unreadable_library_file_does_not_hide_healthy_targets(self): + bad=self.library/'unreadable.md' + read=Path.read_bytes + def read_bytes(path): + if path==bad:raise PermissionError('fixture') + return read(path) + self.candidate(4,1,entry='unreadable',version='unavailable') + with patch.object(Path,'glob',return_value=iter([bad,self.library/'research.md'])), \ + patch.object(Path,'read_bytes',read_bytes): + result=self.refresh() + self.assertEqual(self.arm(result)['opportunities'],1) + self.assertEqual(result['gaps'],{'version_unavailable':1}) + + def _sync_after_cue_failure(self,expected): + def placement(con): + con.execute("INSERT INTO schema_meta VALUES('placement-proof','kept')") + return 1 + refresh=cli._identity.refresh_session_versions + def refresh_versions(con): + refresh(con) + con.execute("INSERT INTO schema_meta VALUES('version-proof','kept')") + ns=SimpleNamespace(harness=['t3'],source=None,root=None,full=False, + agentsmd_repo=str(self.library/'absent'),as_json=True) + adapter=SimpleNamespace(sync=lambda *a,**kw: {'harness':'t3'}) + output=io.StringIO() + with patch.object(cli._adapters,'available',return_value={'t3':adapter}), \ + patch.object(cli.placement,'record_shared_checkouts',side_effect=placement), \ + patch.object(cli._identity,'refresh_session_versions',side_effect=refresh_versions), \ + patch.object(cue,'source_path',return_value=self.source), \ + contextlib.redirect_stdout(output): + code=cli._sync(self.con,ns) + self.assertEqual(code,1) + self.assertEqual(json.loads(output.getvalue())['cue']['status'],expected) + # A new connection proves the non-Cue writes were committed, not just retained in memory. + with contextlib.closing(sqlite3.connect(self.db_path)) as other: + self.assertEqual(other.execute("SELECT count(*) FROM schema_meta WHERE key IN ('placement-proof','version-proof') AND value='kept'").fetchone()[0],2) + self.assertEqual(other.execute('SELECT count(*) FROM cue_candidates').fetchone()[0],1) + self.assertEqual(other.execute('SELECT status FROM cue_import').fetchone()[0],expected) + self.assertNotIn('PRIVATE',output.getvalue()) + + def test_sync_sqlite_failure_rolls_back_only_cue_and_records_gap(self): + self.refresh() + self.con.commit() + self.con.execute("CREATE TRIGGER deny_cue BEFORE INSERT ON cue_candidates BEGIN SELECT RAISE(ABORT,'fixture Cue failure'); END") + self._sync_after_cue_failure('unreadable_database') + + def test_sync_nontext_owners_records_gap_and_preserves_other_writes(self): + self.refresh() + self.con.commit() + self.native.execute('ALTER TABLE moments RENAME COLUMN owners_json TO old_owners') + self.native.execute('ALTER TABLE moments ADD COLUMN owners_json') + self.native.execute('UPDATE moments SET owners_json=17') + self.native.commit() + self._sync_after_cue_failure('schema_drift') diff --git a/tests/test_health.py b/tests/test_health.py index 5a9ff5b..aa68474 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -154,6 +154,27 @@ def test_cli_shows_gap_and_does_not_return_success(self): self.assertIn("lsof", output.getvalue()) +class CueHealthExitTest(unittest.TestCase): + def test_each_cue_status_remains_visible_with_appropriate_exit(self): + for status,expected in (('ok',0),('missing_database',0),('no_cue',0), + ('not_imported',0),('schema_drift',4),('unreadable_database',4)): + for as_json in (False,True): + with self.subTest(status=status,json=as_json): + result={'checked': {'claude_sessions':0,'processes':0}, + 'findings':[],'unavailable':[],'identity_conflicts':[], + 'bare_roots':{},'cue':{'status':status}} + output=io.StringIO() + with patch.object(health,'check',return_value=result), \ + contextlib.redirect_stdout(output): + code=cli.main(['--db','/nonexistent/cue-test.db','health']+ + (['--json'] if as_json else [])) + self.assertEqual(code,expected) + if as_json: + self.assertEqual(json.loads(output.getvalue())['cue']['status'],status) + else: + self.assertIn('Cue evidence: '+status,output.getvalue()) + + class ClaudeVisibilityTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() From 5089a01ee3c49c60f8d517e1de0f96baf01f31cf Mon Sep 17 00:00:00 2001 From: Luke Maj Date: Mon, 19 Oct 2026 16:08:18 +0200 Subject: [PATCH 7/7] fix(cue): contain target expansion and unexpected import errors --- CHANGELOG.md | 8 ++++---- agent_observer/cue.py | 4 ++-- tests/test_cue.py | 20 ++++++++++++++++++++ 3 files changed, 26 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ffe4d1..5fe9ae5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,15 +2,15 @@ ## [Unreleased] +### Added + +- Import Cue decisions and settled ContextMD reads read-only; report procedure reads by entry, harness, machine and AgentsMD version against session-wide holdout, with source and target gaps. + ### Changed - `sync` now reads ContextMD Cue evidence by default alongside the T3 import; a failed Cue import records a gap without discarding other sync writes. - `health` shows optional Cue gaps without failing; Cue status causes exit 4 only for `schema_drift` or `unreadable_database`. -### Added - -- Import Cue decisions and settled ContextMD reads read-only; report procedure reads by entry, harness, machine and AgentsMD version against session-wide holdout, with source and target gaps. - ### Fixed - Parse sectioned AgentsMD direction identity, Codex native durations and Read calls, and Grok MCP timing and outcomes. Retain read success state and count only successful procedure reads; replay older sources to repair stored evidence (#128). diff --git a/agent_observer/cue.py b/agent_observer/cue.py index 6a6daf4..5450cdc 100644 --- a/agent_observer/cue.py +++ b/agent_observer/cue.py @@ -155,7 +155,7 @@ def _targets(roots): found[key] = (None, 'ambiguous_target') else: found[key] = value - except (OSError, ValueError, UnicodeError): + except (OSError, ValueError, UnicodeError, RuntimeError): # Only candidates for this file remain unresolved, not its siblings. continue return found @@ -193,7 +193,7 @@ def import_source(con, source=None, libraries=()): con.execute('SAVEPOINT cue_attempt') try: return _import_snapshot(con, source, libraries) - except (sqlite3.Error, TypeError) as exc: + except Exception as exc: con.execute('ROLLBACK TO cue_attempt') # Exception messages can contain source data; retain only a safe gap category. return _record_gap(con, 'schema_drift' if isinstance(exc, TypeError) diff --git a/tests/test_cue.py b/tests/test_cue.py index 132b316..6baf75f 100644 --- a/tests/test_cue.py +++ b/tests/test_cue.py @@ -480,3 +480,23 @@ def test_sync_nontext_owners_records_gap_and_preserves_other_writes(self): self.native.execute('UPDATE moments SET owners_json=17') self.native.commit() self._sync_after_cue_failure('schema_drift') + + def test_unknown_target_user_is_a_file_gap_without_hiding_siblings(self): + raw=b'---\nid: bad-home\ntarget: ~agent_observer_missing_user_129/x.md\n---\n' + bad=self.library/'bad-home.md' + bad.write_bytes(raw) + self.candidate(4,1,entry='bad-home',version=hashlib.sha256(raw).hexdigest()[:12]) + with patch.object(Path,'glob',return_value=iter([bad,self.library/'research.md'])): + result=self.refresh() + self.assertEqual(self.arm(result)['opportunities'],1) + self.assertEqual(result['gaps'],{'version_unavailable':1}) + + def test_sync_deep_owners_json_records_gap_and_preserves_other_writes(self): + self.refresh() + self.con.commit() + owners='['*100000+'0'+']'*100000 + with self.assertRaises(RecursionError): + json.loads(owners) + self.native.execute('UPDATE moments SET owners_json=?',(owners,)) + self.native.commit() + self._sync_after_cue_failure('unreadable_database')