diff --git a/CHANGELOG.md b/CHANGELOG.md index 9023306..5fe9ae5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +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`. + ### 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..0908f07 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,85 @@ 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 --source /path/to/contextmd.sqlite +agent-observer cue report --source /path/to/contextmd.sqlite --json +``` + +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 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`. 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. 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 +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 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 +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`. +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. 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. + ## Evaluating AgentsMD releases `compare --by agentsmd` groups sessions from every project by the AgentsMD @@ -581,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 a8a3388..62f0470 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"] in {"schema_drift", "unreadable_database"} 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,8 +1072,11 @@ 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 + 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 new file mode 100644 index 0000000..5450cdc --- /dev/null +++ b/agent_observer/cue.py @@ -0,0 +1,600 @@ +"""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 random +import sqlite3 +import time + +from . import privacy + +WINDOW_SECONDS = 1800 +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}$') +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): + # 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): + """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 = list(Path(root).glob('*.md')) + except OSError: + continue + for file in paths: + try: + 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, RuntimeError): + # Only candidates for this file remain unresolved, not its siblings. + 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 _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=()): + """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 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) + 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': + 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,' + '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'])): + continue + con.execute('INSERT INTO cue_moments VALUES(?,?,?,?,?,?,?,?,?)', + (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 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']), + (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 _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 _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 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': {}, 'by_harness': {}, 'groups': [], 'gaps': {}} + if state['status'] != 'ok': + return result + moments = {r['id']: dict(r) for r in con.execute('SELECT * FROM cue_moments')} + 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'])): + 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, 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. + 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 + 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': {}, 'repeats': {}, 'excluded': {}} + g = groups[key] + g['outcomes'][c['outcome']] += 1 + if not above: + continue + 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'])) + 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 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']) + 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 + 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'],observation['target'])] + times = [r['at'] for r in rs] + earlier = rs[:bisect_right(times,m['at'])] + if earlier: + 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: + 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'],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()))) + 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'], 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"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, ' ') + 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', 'summary', 'comparison', 'interval_method', + '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', 'carry_prints'}) + + +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, check_source=False) + 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) + + def validate_buckets(container): + require(isinstance(container, dict)) + for section, allowed in (('arms', {'served','holdout'}), ('repeats', {'served','holdout'}), + ('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'}) + 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'}) + 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']) + require(isinstance(value['groups'], list)) + for g in value['groups']: + require(isinstance(g, dict) and set(g) == {'entry','harness','machine', + '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']) + validate_buckets(g) + validate_buckets(value['summary']) + return value diff --git a/agent_observer/fleet.py b/agent_observer/fleet.py index e11585e..af9537e 100644 --- a/agent_observer/fleet.py +++ b/agent_observer/fleet.py @@ -132,7 +132,10 @@ 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 + 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, @@ -446,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 @@ -458,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/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..093cdfe 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 | 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. | 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..6baf75f --- /dev/null +++ b/tests/test_cue.py @@ -0,0 +1,502 @@ +"""Cue's real SQLite boundary and aggregate report contract.""" +import contextlib +import hashlib +import io +import json +from pathlib import Path +from types import SimpleNamespace +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']),(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): + 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(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' + 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) + + 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))) + + 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) + + 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}) + + 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') + + 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') diff --git a/tests/test_health.py b/tests/test_health.py index 40fb25c..aa68474 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"], []) @@ -153,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()