From 0d442a03582738691c8079aaab09b908e301abbe Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 18 Jul 2026 13:58:13 +0200 Subject: [PATCH 1/4] wip(demo): construct-validity verdict + economy columns (lane A, session cut off) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preserves the lane-A codex session's uncommitted working tree after the Codex quota cutoff. Phase 0 verdict: the 24.1% silent-proceed headline is not currently publishable (fresh frame n=20 < 30 minimum; 14 apparent silent-proceeds were protocol-only follow-ups, now classified ambiguous in followup.py + archive.py). Phase 1 economy columns (_economy_rows over session_model_usage, disjoint token/cost lanes) is mid-implementation; tests partial. NOT verified — WIP preservation commit. Co-Authored-By: Codex Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01QUsH3Rhq6oAZpYPWcsZqnZ --- devtools/claim_vs_evidence.py | 198 +++++++++++++++++- docs/findings/claim-vs-evidence.md | 31 ++- polylogue/archive/actions/followup.py | 11 + .../storage/sqlite/archive_tiers/archive.py | 3 + tests/unit/devtools/test_claim_vs_evidence.py | 61 ++++++ 5 files changed, 300 insertions(+), 4 deletions(-) diff --git a/devtools/claim_vs_evidence.py b/devtools/claim_vs_evidence.py index 6f871e722c..3f48c95fc6 100644 --- a/devtools/claim_vs_evidence.py +++ b/devtools/claim_vs_evidence.py @@ -8,6 +8,7 @@ import random import sys from collections.abc import Iterable, Mapping +from contextlib import closing from datetime import UTC, datetime from pathlib import Path from sqlite3 import Connection @@ -137,12 +138,151 @@ def _scalar_int(conn: Connection, sql: str, params: Iterable[object] = ()) -> in return int(row[0]) if row is not None and row[0] is not None else 0 +def _table_exists(conn: Connection, name: str) -> bool: + return conn.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (name,)).fetchone() is not None + + +def _economy_rows( + conn: Connection, + *, + session_ids: tuple[str, ...], + silent_by_origin: Mapping[str, int], +) -> dict[str, list[dict[str, object]]]: + """Return explicitly separate token and money lanes for the sampled harness. + + ``session_model_usage`` is the only rollup source. It is itself populated + from provider usage events with Codex's inclusive input/cache and + output/reasoning fields split into disjoint billable lanes; profile text + token columns are deliberately never consulted here. + """ + if not session_ids or not _table_exists(conn, "session_model_usage"): + return {"by_model": [], "by_origin": []} + placeholders = ", ".join("?" for _ in session_ids) + model_rows = _rows( + conn, + f""" + SELECT + s.origin, + u.model_name, + COUNT(*) AS session_model_rows, + SUM(u.input_tokens) AS input_tokens, + SUM(u.output_tokens) AS output_tokens, + SUM(u.cache_read_tokens) AS cache_read_tokens, + SUM(u.cache_write_tokens) AS cache_write_tokens, + SUM(CASE WHEN u.cost_provenance IN ('priced', 'estimated') THEN COALESCE(u.cost_usd, 0) ELSE 0 END) + AS catalog_cost_usd, + SUM(CASE WHEN u.cost_provenance = 'origin_reported' THEN COALESCE(u.cost_usd, 0) ELSE 0 END) + AS provider_reported_cost_usd + FROM session_model_usage AS u + JOIN sessions AS s ON s.session_id = u.session_id + WHERE u.session_id IN ({placeholders}) + GROUP BY s.origin, u.model_name + ORDER BY s.origin, u.model_name + """, + session_ids, + ) + event_rollups: dict[tuple[str, str], tuple[int, int | None]] = {} + if _table_exists(conn, "session_provider_usage_events"): + for row in _rows( + conn, + f""" + SELECT s.origin, COALESCE(NULLIF(e.model_name, ''), 'unknown') AS model_name, + COUNT(*) AS api_call_count, + MAX(CASE WHEN e.provider_event_type = 'message_usage' THEN 1 ELSE 0 END) + AS has_reasoning_delta, + SUM(CASE WHEN e.provider_event_type = 'message_usage' + THEN COALESCE(e.last_reasoning_output_tokens, 0) ELSE 0 END) + AS reasoning_tokens + FROM session_provider_usage_events AS e + JOIN sessions AS s ON s.session_id = e.session_id + WHERE e.session_id IN ({placeholders}) + GROUP BY s.origin, COALESCE(NULLIF(e.model_name, ''), 'unknown') + """, + session_ids, + ): + has_reasoning_delta = _object_int(row["has_reasoning_delta"]) == 1 + event_rollups[(str(row["origin"]), str(row["model_name"]))] = ( + _object_int(row["api_call_count"]), + _object_int(row["reasoning_tokens"]) if has_reasoning_delta else None, + ) + + def decorate(row: dict[str, object], *, model_name: str | None) -> dict[str, object]: + origin = str(row["origin"]) + input_tokens = _object_int(row["input_tokens"]) + cache_read_tokens = _object_int(row["cache_read_tokens"]) + catalog_cost = _object_float(row["catalog_cost_usd"]) + provider_cost = _object_float(row["provider_reported_cost_usd"]) + silent = silent_by_origin.get(origin, 0) + if model_name is not None: + api_call_count, reasoning_tokens = event_rollups.get((origin, model_name), (0, None)) + else: + origin_events = [event for (event_origin, _model), event in event_rollups.items() if event_origin == origin] + api_call_count = sum(event[0] for event in origin_events) + reasoning_values = [event[1] for event in origin_events if event[1] is not None] + reasoning_tokens = sum(reasoning_values) if reasoning_values else None + return { + "origin": origin, + "model_name": model_name, + "session_model_rows": _object_int(row["session_model_rows"]), + "api_call_count": api_call_count, + "input_tokens": input_tokens, + "output_tokens": _object_int(row["output_tokens"]), + "cache_read_tokens": cache_read_tokens, + "cache_write_tokens": _object_int(row["cache_write_tokens"]), + "reasoning_tokens": reasoning_tokens, + "reasoning_token_note": ( + "provider message_usage deltas only; cumulative token_count events are excluded" + if reasoning_tokens is not None + else "no provider message_usage reasoning delta in the sampled harness" + ), + "cache_read_share": cache_read_tokens / (input_tokens + cache_read_tokens) + if input_tokens + cache_read_tokens + else None, + "catalog_cost_usd": catalog_cost, + "provider_reported_cost_usd": provider_cost, + "silent_proceed_outcomes": silent, + "catalog_usd_per_silent_proceed": catalog_cost / silent if silent else None, + "provider_reported_usd_per_silent_proceed": provider_cost / silent if silent else None, + "token_source": "session_model_usage (provider-usage materialization; disjoint cache lanes)", + } + + by_model = [decorate(row, model_name=str(row["model_name"])) for row in model_rows] + by_origin_source: dict[str, dict[str, object]] = {} + for row in model_rows: + origin = str(row["origin"]) + aggregate = by_origin_source.setdefault( + origin, + { + "origin": origin, + "session_model_rows": 0, + "input_tokens": 0, + "output_tokens": 0, + "cache_read_tokens": 0, + "cache_write_tokens": 0, + "catalog_cost_usd": 0.0, + "provider_reported_cost_usd": 0.0, + }, + ) + for key in ("session_model_rows", "input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens"): + aggregate[key] = _object_int(aggregate[key]) + _object_int(row[key]) + for key in ("catalog_cost_usd", "provider_reported_cost_usd"): + aggregate[key] = _object_float(aggregate[key]) + _object_float(row[key]) + return { + "by_model": by_model, + "by_origin": [decorate(row, model_name=None) for _, row in sorted(by_origin_source.items())], + } + + def _object_int(value: object) -> int: if value is None: return 0 return int(str(value)) +def _object_float(value: object) -> float: + return 0.0 if value is None else float(str(value)) + + def _object_str_list(value: object) -> list[str]: if isinstance(value, list | tuple): return [str(item) for item in value] @@ -773,6 +913,19 @@ def _calibration_metrics(label_rows: list[dict[str, str]], *, labels_path: Path } +def _calibration_frame_coverage(label_rows: list[dict[str, str]], samples: list[dict[str, object]]) -> dict[str, int]: + """State whether labels calibrate this report's actual current frame.""" + sampled_refs = {str(sample["tool_result_message_ref"]) for sample in samples} + labeled_refs = {str(row.get("tool_result_message_ref", "")) for row in label_rows} + matched = sampled_refs & labeled_refs + return { + "labeled_sample_refs": len(labeled_refs), + "current_sample_refs": len(sampled_refs), + "labels_in_current_sample": len(matched), + "labels_outside_current_sample": len(labeled_refs - sampled_refs), + } + + def _calibration_labels_path(args: argparse.Namespace) -> Path | None: calibration_labels = args.calibration_labels if isinstance(calibration_labels, Path): @@ -939,7 +1092,16 @@ def build_report(args: argparse.Namespace) -> dict[str, Any]: "sample_file": _CALIBRATION_SAMPLE_FILE if args.out_dir is not None else None, "labels_file": _CALIBRATION_LABELS_FILE if args.out_dir is not None else None, "metrics": _calibration_metrics(calibration_label_rows, labels_path=calibration_labels_path), + "frame_coverage": _calibration_frame_coverage(calibration_label_rows, calibration_sample), } + silent_by_origin = {origin: counts["silent_proceed"] for origin, counts in by_origin.items()} + sampled_session_ids = tuple(sorted({str(row["session_id"]) for row in rows})) + with closing(open_readonly_connection(index_db)) as economy_conn: + economy = _economy_rows( + economy_conn, + session_ids=sampled_session_ids, + silent_by_origin=silent_by_origin, + ) report: dict[str, Any] = { "report_version": 1, "captured_at": datetime.now(UTC).isoformat(), @@ -1002,6 +1164,11 @@ def build_report(args: argparse.Namespace) -> dict[str, Any]: "by_model": _ranked(by_model, n_min=args.n_min), "by_origin": _ranked(by_origin, n_min=args.n_min), "by_handler_class": _ranked(by_handler_class, n_min=args.n_min), + "economy": { + **economy, + "scope": "sessions represented in the paired structured-failure harness", + "sampled_session_count": len(sampled_session_ids), + }, "handler_class_definition": { "benign_recovery": sorted(_BENIGN_RECOVERY_TOOLS), "consequential": sorted(_CONSEQUENTIAL_TOOLS), @@ -1029,6 +1196,10 @@ def _format_rate_percent(value: int | float | str | None) -> str: return f"{float(value):.1%}" +def _format_dollars(value: object) -> str: + return "not computable" if value is None else f"${_object_float(value):.2f}" + + def _public_summary(report: dict[str, Any]) -> dict[str, Any]: totals = report["totals"] frame = report["sample_frame"] @@ -1041,11 +1212,13 @@ def _public_summary(report: dict[str, Any]) -> dict[str, Any]: "index_schema_version": report["index_schema_version"], "claim": ( "Polylogue can ground a failure-follow-up finding in normalized tool-result outcomes, " - "state the bounded sample frame, and publish aggregate rates without exposing raw private transcripts." + "state the bounded sample frame, and publish aggregate rates only when the observable follow-up " + "classification has enough coverage." ), "non_claim": ( "The live aggregate is not reproducible without the private archive; the deterministic demo archive " - "reproduces the method and artifact shape, not the private corpus rates." + "reproduces the method and artifact shape, not the private corpus rates. A missing acknowledgement " + "does not establish bad recovery behavior, and protocol-only reasoning is deliberately ambiguous." ), "proofs": [ { @@ -1091,6 +1264,7 @@ def _public_summary(report: dict[str, Any]) -> dict[str, Any]: "Private live-archive counts are aggregate-only in this public summary.", "Deterministic demo reproduction validates the method and renderer, not the private rate estimates.", "The classifier is an explicit marker detector; ambiguous rows remain in the denominator.", + "A wordless retry or a recovery that fixes the problem can be operationally appropriate; this metric does not judge it.", "The report is bounded by --limit unless the limit exceeds the full structured-failure frame.", "Split cells below n_min are coverage-only and explicitly not supported for rate publication.", ], @@ -1373,6 +1547,8 @@ def _write_readme(path: Path, report: dict[str, Any]) -> None: "This demo anchors on structured tool-result evidence and asks what the next assistant", "turn did with that failure. It does not infer truth from assistant prose: the failure", "predicate is `is_error=1` or a non-zero `exit_code` on normalized `actions` rows.", + "`silent-proceed` is only an observable absence of an explicit acknowledgement marker in a visible", + "next assistant message. It is not a judgment that recovery was wrong, unhelpful, or unsuccessful.", "", "## Current Bounded Result", "", @@ -1437,11 +1613,29 @@ def _write_readme(path: Path, report: dict[str, Any]) -> None: f"- calibration sample size: {int(calibration['sample_size']):,}", f"- calibration seed: {int(calibration['sample_seed'])}", f"- labeled rows: {int(calibration_metrics['labeled_rows']):,}", + f"- labels in this current calibration sample: {int(calibration['frame_coverage']['labels_in_current_sample']):,}", + f"- labels outside this current calibration sample: {int(calibration['frame_coverage']['labels_outside_current_sample']):,}", ( f"- acknowledged-marker precision: {float(precision):.1%}" if precision is not None else "- acknowledged-marker precision: not enough labels" ), + "", + "### Economy Lanes", + "", + "These lanes are restricted to sessions represented in this paired failure harness. Token and money", + "values come only from `session_model_usage` plus provider-usage event counts,", + "not profile text columns. Codex input/cache and output/reasoning semantics are kept disjoint by", + "the usage materializer. Provider-reported and catalog-derived money remain separate.", + *[ + ( + f"- {row['origin']}: calls {int(row['api_call_count']):,}; input {int(row['input_tokens']):,}; " + f"output {int(row['output_tokens']):,}; cache-read {int(row['cache_read_tokens']):,}; " + f"catalog ${float(row['catalog_cost_usd']):.2f}; provider-reported ${float(row['provider_reported_cost_usd']):.2f}; " + f"catalog $/silent {_format_dollars(row['catalog_usd_per_silent_proceed'])}" + ) + for row in report["economy"]["by_origin"] + ], ( f"- acknowledged-marker recall: {float(recall):.1%}" if recall is not None diff --git a/docs/findings/claim-vs-evidence.md b/docs/findings/claim-vs-evidence.md index 7b8cc61723..f2a87d026c 100644 --- a/docs/findings/claim-vs-evidence.md +++ b/docs/findings/claim-vs-evidence.md @@ -3,7 +3,25 @@ ## Claim -The historical packet generated on 2026-07-04 reported that, in one bounded private-archive sample, at least 24.1% of sampled structured failures were followed by an assistant turn that proceeded without an acknowledgment marker. Most sampled cases remained ambiguous. +The historical packet generated on 2026-07-04 reported that, in one bounded private-archive sample, at least 24.1% of sampled structured failures were followed by an assistant turn that proceeded without an acknowledgment marker. Most sampled cases remained ambiguous. It is a historical observation, not a current headline. + +## Current construct-validity verdict (2026-07-18) + +The old rate is **not currently publishable**. A fresh full-frame audit found only +20 structured failures (all from one Claude Code session), below the report's +minimum of 30. Fourteen apparent `silent-proceed` rows were internal +`…` protocol content, so the classifier now treats them as +ambiguous rather than evidence of a visible silent follow-up. The current frame +therefore has zero classified rows and no rate. + +Operationally, `silent-proceed` means only: after a structurally failed tool +result, the next visible assistant message contains no configured explicit +failure-acknowledgement marker. It does **not** mean recovery was wrong: a +wordless retry or a successful corrective action may be appropriate. The two +largest validity threats are hidden/protocol-only content being mistaken for +visible prose, and a small single-origin frame. The claim would be falsified by +a representative, sufficiently large calibration frame showing that visible +marker absence does not track human labels for this narrow observable. The generated [findings-page public-claims view](../generated/public-claims/findings-page.md) is the authority for whether this historical number is currently supported, stale, private-held, or unresolved. @@ -65,7 +83,16 @@ The tracked calibration set contains 50 labeled rows. The packet reports: - precision: 100.0%; - recall: 84.2%. -The calibration is small. The method therefore keeps 3,375 cases ambiguous instead of forcing them into acknowledged or silent classes. +The calibration is small. The method therefore keeps 3,375 cases ambiguous instead of forcing them into acknowledged or silent classes. Those 50 historical labels do not overlap the 2026-07-18 frame; they are useful historical calibration evidence, not fresh validation of its current rate. + +## First-party evidence boundary + +The current report is a regenerable local evidence artifact, not a registered +analysis definition, immutable analysis run, or finding. Those first-party +objects require the pending durable user-tier kernel and migration admission; +until that work is accepted, this page must not promote a newly generated +packet into a current public claim. The generated public-claims view remains +the authority for claim status. ## Interpretation diff --git a/polylogue/archive/actions/followup.py b/polylogue/archive/actions/followup.py index bedeebf8c0..c1ec6ee2fd 100644 --- a/polylogue/archive/actions/followup.py +++ b/polylogue/archive/actions/followup.py @@ -27,6 +27,7 @@ from __future__ import annotations +import re from typing import Literal, TypedDict FollowupClass = Literal["acknowledged", "silent_proceed", "wordless_continuation", "ambiguous"] @@ -88,6 +89,14 @@ class FollowupEvidence(TypedDict): "failing", ) +# Some runtimes retain an internal-reasoning envelope as a text block. It is +# not a reader-visible follow-up, so its absence of an acknowledgement marker +# cannot support a claim that the assistant silently proceeded. +_PROTOCOL_ONLY_FOLLOWUP = re.compile( + r"^\s*<(?:thinking|analysis|reasoning)(?:\s[^>]*)?>.*?\s*$", + re.IGNORECASE | re.DOTALL, +) + def classify_failed_followup_evidence(text: str | None) -> FollowupEvidence: """Classify the next assistant turn after a structured action failure. @@ -99,6 +108,8 @@ def classify_failed_followup_evidence(text: str | None) -> FollowupEvidence: if text is None: return {"classification": "ambiguous", "reason": "missing_next_assistant_message", "matched_marker": None} + if _PROTOCOL_ONLY_FOLLOWUP.fullmatch(text): + return {"classification": "ambiguous", "reason": "protocol_only_next_assistant_message", "matched_marker": None} normalized = " ".join(text.lower().split()) if len(normalized) < 20: return {"classification": "ambiguous", "reason": "short_next_assistant_message", "matched_marker": None} diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index e784327ec3..76aa509334 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -881,6 +881,9 @@ def _action_command_expression(row_alias: str) -> str: WHEN NOT (COALESCE(aft.is_error, 0) = 1 OR COALESCE(aft.exit_code, 0) != 0) THEN NULL WHEN aft.followup_message_id IS NULL THEN 'ambiguous' WHEN {_ACTION_FOLLOWUP_ACK_CONDITION} THEN 'acknowledged' + WHEN TRIM(aft.followup_text_lower) GLOB '*' + OR TRIM(aft.followup_text_lower) GLOB '*' + OR TRIM(aft.followup_text_lower) GLOB '*' THEN 'ambiguous' WHEN aft.followup_has_tool_use = 1 AND aft.followup_pre_tool_text_chars <= 40 THEN 'wordless_continuation' WHEN LENGTH(TRIM(aft.followup_text)) < 20 THEN 'ambiguous' diff --git a/tests/unit/devtools/test_claim_vs_evidence.py b/tests/unit/devtools/test_claim_vs_evidence.py index 6ad38779ab..ec0f6b8259 100644 --- a/tests/unit/devtools/test_claim_vs_evidence.py +++ b/tests/unit/devtools/test_claim_vs_evidence.py @@ -8,6 +8,7 @@ import pytest from devtools.claim_vs_evidence import build_report +from polylogue.archive.actions.followup import classify_failed_followup_evidence from polylogue.demo import seed_demo_archive @@ -100,6 +101,22 @@ def _seed_archive(root: Path) -> None: AND r.session_id = u.session_id AND r.block_type = 'tool_result' WHERE u.block_type = 'tool_use'; + CREATE TABLE session_model_usage ( + session_id TEXT NOT NULL, + model_name TEXT NOT NULL, + input_tokens INTEGER NOT NULL, + output_tokens INTEGER NOT NULL, + cache_read_tokens INTEGER NOT NULL, + cache_write_tokens INTEGER NOT NULL, + cost_usd REAL, + cost_provenance TEXT NOT NULL + ); + CREATE TABLE session_provider_usage_events ( + session_id TEXT NOT NULL, + model_name TEXT, + provider_event_type TEXT NOT NULL, + last_reasoning_output_tokens INTEGER + ); """ ) conn.executemany( @@ -107,6 +124,7 @@ def _seed_archive(root: Path) -> None: [ ("s1", "claude-code-session", "fixture one", 1, 4), ("s2", "codex-session", "fixture two", 1, 1), + ("s3", "claude-code-session", "unrelated fixture", 1, 1), ], ) conn.executemany( @@ -190,10 +208,43 @@ def _seed_archive(root: Path) -> None: ("next-wordless", "s2", 0, "tool_use", None, "Read", "t5", '{"path":"z"}', None, None), ], ) + conn.executemany( + """ + INSERT INTO session_model_usage( + session_id, model_name, input_tokens, output_tokens, cache_read_tokens, + cache_write_tokens, cost_usd, cost_provenance + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + ("s1", "claude-sonnet", 11, 12, 13, 14, 1.25, "priced"), + ("s2", "codex", 21, 22, 23, 24, 3.5, "origin_reported"), + ("s3", "unrelated-model", 999, 999, 999, 999, 99.0, "priced"), + ], + ) + conn.executemany( + """ + INSERT INTO session_provider_usage_events( + session_id, model_name, provider_event_type, last_reasoning_output_tokens + ) VALUES (?, ?, ?, ?) + """, + [ + ("s1", "claude-sonnet", "message_usage", 5), + ("s2", "codex", "token_count", 89), + ], + ) conn.commit() conn.close() +def test_protocol_only_followup_is_ambiguous_not_silent_proceed() -> None: + """Hidden reasoning contains no reader-visible acknowledgement evidence.""" + assert classify_failed_followup_evidence("inspect state privately") == { + "classification": "ambiguous", + "reason": "protocol_only_next_assistant_message", + "matched_marker": None, + } + + def test_claim_vs_evidence_builds_bounded_artifacts(tmp_path: Path) -> None: archive = tmp_path / "archive" out_dir = tmp_path / "out" @@ -287,6 +338,16 @@ def test_claim_vs_evidence_builds_bounded_artifacts(tmp_path: Path) -> None: assert report["rates"]["silent_rate_lower_bound"] == 1 / 4 assert report["rates"]["ack_later_within_3"] == 1 assert report["rates"]["window3_silent_rate_lower_bound"] == 0 + assert report["economy"]["scope"] == "sessions represented in the paired structured-failure harness" + assert report["economy"]["sampled_session_count"] == 2 + economy_by_model = {str(row["model_name"]): row for row in report["economy"]["by_model"]} + assert set(economy_by_model) == {"claude-sonnet", "codex"} + assert economy_by_model["claude-sonnet"]["reasoning_tokens"] == 5 + assert economy_by_model["claude-sonnet"]["catalog_cost_usd"] == 1.25 + assert economy_by_model["codex"]["reasoning_tokens"] is None + assert economy_by_model["codex"]["provider_reported_cost_usd"] == 3.5 + assert economy_by_model["claude-sonnet"]["input_tokens"] == 11 + assert "unrelated-model" not in economy_by_model assert report["calibration"]["sample_size"] == 3 assert report["calibration"]["sample_seed"] == 7 assert report["calibration"]["metrics"]["labeled_rows"] == 2 From 03ee102a4c9478ea46fdb4301a37e2de6af36da7 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 18 Jul 2026 14:32:00 +0200 Subject: [PATCH 2/4] fix(sources): extract Hermes tool-outcome fields, correct claim-vs-evidence framing Problem: polylogue-uwlu asked why Hermes sessions never appear in the claim-vs-evidence by_model breakdown. Live ~/.hermes/state.db inspection showed Hermes tool-result content already carries exit_code/success/error signals (1592/567/3390 of 4100 tool messages respectively, with real nonzero exit codes), but hermes_state.py's _parse_message_row only read "text" via _content_text and never extracted them into ParsedContentBlock.is_error/exit_code. Every Hermes tool_result was therefore structurally invisible to _structured_failure_rows (which requires is_error=1 OR exit_code!=0), regardless of the model_name COALESCE join originally suspected. What changed: add _tool_result_outcome() to map Hermes's JSON content envelope onto is_error/exit_code (precedence: explicit error message > success flag > exit-code sign; absence of all three stays unknown, never guessed from prose - consistent with the "keystone" NULL-means-unknown invariant). 7 focused tests in the new tests/unit/sources/parsers/test_hermes_state.py pin the mapping against shapes observed in real data. Separately, corrected docs/findings/claim-vs-evidence.md: the inherited "n=20, single Claude Code session" verdict is not a small-corpus finding. readiness_check shows the live archive's derived index is in a known, actively-tracked "poisoned" raw-materialization state (4 of 73,295 raw artifacts materialized; the real corpus spans ~73k sessions across 10 origins including 193 Hermes sessions) - bead polylogue-hjpx/hjpx.2, owned by a separate lane and not authorized for live-archive repair yet. Flagged this so the n=20 frame isn't mistaken for corpus size, and so any live number this report emits is read as provisional until that backlog clears. Verification: devtools test tests/unit/sources/parsers/test_hermes_state.py (7 passed) plus the existing Hermes/lineage/import selection (65 passed); mypy --strict on the touched parser clean. Confirmed the report generator still runs end-to-end against the live (degraded) archive without error (smoke-tested, not committed as a demo packet per the existing first-party-evidence-boundary note). Ref polylogue-uwlu Co-Authored-By: Claude --- docs/findings/claim-vs-evidence.md | 25 +++- polylogue/sources/parsers/hermes_state.py | 27 ++++ .../unit/sources/parsers/test_hermes_state.py | 129 ++++++++++++++++++ 3 files changed, 178 insertions(+), 3 deletions(-) create mode 100644 tests/unit/sources/parsers/test_hermes_state.py diff --git a/docs/findings/claim-vs-evidence.md b/docs/findings/claim-vs-evidence.md index f2a87d026c..633000b1ee 100644 --- a/docs/findings/claim-vs-evidence.md +++ b/docs/findings/claim-vs-evidence.md @@ -14,14 +14,33 @@ minimum of 30. Fourteen apparent `silent-proceed` rows were internal ambiguous rather than evidence of a visible silent follow-up. The current frame therefore has zero classified rows and no rate. +**Correction to this note's earlier framing:** the n=20, single-origin frame is +not evidence that the real corpus is small or single-origin. At the time this +audit ran (and still, as of this writing), the live archive's derived index was +in a known, actively-tracked degraded state: `readiness_check` reports +`raw_materialization: poisoned`, with only 4 of 73,295 raw source artifacts +materialized into `index.db` (`join_gap_count: 73291`). The real corpus behind +that gap spans `codex-session` (39,638), `claude-code-session` (21,420), +`chatgpt-export` (7,679), `claude-ai-export` (3,922), `hermes-session` (193), +and five smaller origins. This is bead `polylogue-hjpx` / `polylogue-hjpx.2` +(a P0/P1 raw-authority replay fixed-point program, owned by a separate lane, +explicitly not authorized for live-archive repair yet) — not a Lane A finding +and not something this lane fixes. Until that backlog clears, **any live-archive +number this report emits reflects whatever tiny slice happens to be +materialized at run time, not the archive's real shape.** Treat every +"current bounded result" below as provisional and re-run after +`readiness_check.archive_convergence.materialization_ready` is `true`. + Operationally, `silent-proceed` means only: after a structurally failed tool result, the next visible assistant message contains no configured explicit failure-acknowledgement marker. It does **not** mean recovery was wrong: a wordless retry or a successful corrective action may be appropriate. The two largest validity threats are hidden/protocol-only content being mistaken for -visible prose, and a small single-origin frame. The claim would be falsified by -a representative, sufficiently large calibration frame showing that visible -marker absence does not track human labels for this narrow observable. +visible prose, and (independently of the materialization gap above) a +single-origin frame if one recurs after re-materialization. The claim would be +falsified by a representative, sufficiently large calibration frame showing +that visible marker absence does not track human labels for this narrow +observable. The generated [findings-page public-claims view](../generated/public-claims/findings-page.md) is the authority for whether this historical number is currently supported, stale, private-held, or unresolved. diff --git a/polylogue/sources/parsers/hermes_state.py b/polylogue/sources/parsers/hermes_state.py index 4d2cbd40b6..eaea90e8db 100644 --- a/polylogue/sources/parsers/hermes_state.py +++ b/polylogue/sources/parsers/hermes_state.py @@ -702,12 +702,15 @@ def _parse_message_row( role = Role.normalize(_optional_text(row["role"]) or "unknown") tool_call_id = _optional_text(_row_value(row, "tool_call_id")) if role is Role.TOOL and text: + is_error, exit_code = _tool_result_outcome(row["content"]) blocks.append( ParsedContentBlock( type=BlockType.TOOL_RESULT, tool_id=tool_call_id, tool_name=_optional_text(_row_value(row, "tool_name")), text=text, + is_error=is_error, + exit_code=exit_code, ) ) token_count = _non_negative_int(_row_value(row, "token_count")) or 0 @@ -861,6 +864,30 @@ def _decode_content(value: object) -> object: return value +def _tool_result_outcome(raw_content: object) -> tuple[bool | None, int | None]: + """Extract the structured outcome Hermes already embeds in its tool content. + + Hermes stores tool results as a JSON envelope (``{"output": ...}``) with + one of ``exit_code`` (shell/command-style tools), ``success`` + (boolean-style tools, paired with an ``error`` message when false), or a + bare ``error`` message (status-only tools) layered on top -- never all + three. Absence of every signal means the source tool genuinely reported + no outcome, which stays unknown rather than guessed from prose. + """ + payload = _json_mapping(raw_content) + if not payload: + return None, None + raw_exit_code = payload.get("exit_code") + exit_code = raw_exit_code if isinstance(raw_exit_code, int) and not isinstance(raw_exit_code, bool) else None + if payload.get("error") is not None: + return True, exit_code + if "success" in payload: + return not bool(payload["success"]), exit_code + if exit_code is not None: + return exit_code != 0, exit_code + return None, None + + def _json_mapping(value: object) -> dict[str, object]: parsed = _json_value(value) return dict(parsed) if isinstance(parsed, Mapping) else {} diff --git a/tests/unit/sources/parsers/test_hermes_state.py b/tests/unit/sources/parsers/test_hermes_state.py new file mode 100644 index 0000000000..2cdc771284 --- /dev/null +++ b/tests/unit/sources/parsers/test_hermes_state.py @@ -0,0 +1,129 @@ +"""Hermes ``state.db`` tool-outcome extraction contracts. + +Hermes stores tool results as a JSON envelope in ``messages.content`` +(``{"output": ...}`` plus ``exit_code`` / ``success`` / ``error`` depending on +tool family) rather than dedicated outcome columns. These tests pin the +mapping from that envelope onto ``ParsedContentBlock.is_error`` / +``exit_code``, verified against real shapes observed in a live Hermes +``state.db`` (see ``polylogue-uwlu``). +""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path + +from polylogue.core.enums import BlockType +from polylogue.sources.parsers.base import ParsedContentBlock +from polylogue.sources.parsers.hermes_state import parse_state_db + + +def _write_state_db(path: Path, *, tool_contents: list[str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(path) as conn: + conn.executescript( + """ + CREATE TABLE schema_version(version INTEGER NOT NULL); + INSERT INTO schema_version(version) VALUES (16); + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + source TEXT, + model TEXT, + model_config TEXT, + parent_session_id TEXT, + started_at REAL, + ended_at REAL, + title TEXT + ); + CREATE TABLE messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT, + tool_call_id TEXT, + tool_name TEXT, + tool_calls TEXT, + timestamp REAL NOT NULL, + observed INTEGER DEFAULT 0, + active INTEGER NOT NULL DEFAULT 1, + compacted INTEGER NOT NULL DEFAULT 0 + ); + INSERT INTO sessions (id, source, model, model_config, parent_session_id, started_at, ended_at, title) + VALUES ('s1', 'hermes', 'test-model', '{}', NULL, 1775000000.0, 1775000010.0, 'Outcome fixture'); + """ + ) + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, 'user', 'go', ?)", + ("s1", 1775000001.0), + ) + for index, content in enumerate(tool_contents): + conn.execute( + """ + INSERT INTO messages (session_id, role, content, tool_call_id, tool_name, timestamp) + VALUES ('s1', 'tool', ?, ?, 'shell', ?) + """, + (content, f"call-{index}", 1775000002.0 + index), + ) + conn.commit() + + +def _tool_result_blocks(path: Path, *, tool_contents: list[str]) -> list[ParsedContentBlock]: + _write_state_db(path, tool_contents=tool_contents) + sessions = parse_state_db(path) + assert len(sessions) == 1 + return [ + block for message in sessions[0].messages for block in message.blocks if block.type is BlockType.TOOL_RESULT + ] + + +def test_exit_code_zero_is_not_an_error(tmp_path: Path) -> None: + blocks = _tool_result_blocks(tmp_path / "state.db", tool_contents=[json.dumps({"output": "ok", "exit_code": 0})]) + assert blocks[0].is_error is False + assert blocks[0].exit_code == 0 + + +def test_nonzero_exit_code_is_an_error(tmp_path: Path) -> None: + blocks = _tool_result_blocks( + tmp_path / "state.db", tool_contents=[json.dumps({"output": "boom", "exit_code": 127})] + ) + assert blocks[0].is_error is True + assert blocks[0].exit_code == 127 + + +def test_success_true_is_not_an_error(tmp_path: Path) -> None: + blocks = _tool_result_blocks(tmp_path / "state.db", tool_contents=[json.dumps({"output": "done", "success": True})]) + assert blocks[0].is_error is False + assert blocks[0].exit_code is None + + +def test_success_false_with_error_message_is_an_error(tmp_path: Path) -> None: + blocks = _tool_result_blocks( + tmp_path / "state.db", + tool_contents=[json.dumps({"output": "", "success": False, "error": "not found"})], + ) + assert blocks[0].is_error is True + assert blocks[0].exit_code is None + + +def test_bare_error_message_with_no_exit_code_or_success_is_an_error(tmp_path: Path) -> None: + blocks = _tool_result_blocks( + tmp_path / "state.db", tool_contents=[json.dumps({"output": None, "error": "invalid sort: recent"})] + ) + assert blocks[0].is_error is True + assert blocks[0].exit_code is None + + +def test_plain_output_with_no_outcome_signal_is_unknown_not_guessed(tmp_path: Path) -> None: + blocks = _tool_result_blocks(tmp_path / "state.db", tool_contents=[json.dumps({"output": "just text"})]) + assert blocks[0].is_error is None + assert blocks[0].exit_code is None + + +def test_exit_code_and_error_together_prefer_error_but_keep_exit_code(tmp_path: Path) -> None: + blocks = _tool_result_blocks( + tmp_path / "state.db", + tool_contents=[json.dumps({"output": "boom", "exit_code": 1, "error": "denied"})], + ) + assert blocks[0].is_error is True + assert blocks[0].exit_code == 1 From 88557c6f0e036e7e4090b9b4968092e76d8a27f7 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 18 Jul 2026 14:38:08 +0200 Subject: [PATCH 3/4] test(devtools): cover _economy_rows short-circuit branches Add direct unit tests for the two early-return paths in _economy_rows (empty session_ids; archive missing session_model_usage entirely) that were only reachable indirectly through build_report before. Both are real conditions a caller can hit (a report with zero sampled sessions; an older/rebuilding index without the usage rollup table) and were previously unverified. Verification: devtools test tests/unit/devtools/test_claim_vs_evidence.py (10 passed); mypy --strict clean. Co-Authored-By: Claude --- tests/unit/devtools/test_claim_vs_evidence.py | 25 ++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/tests/unit/devtools/test_claim_vs_evidence.py b/tests/unit/devtools/test_claim_vs_evidence.py index ec0f6b8259..a10158e22d 100644 --- a/tests/unit/devtools/test_claim_vs_evidence.py +++ b/tests/unit/devtools/test_claim_vs_evidence.py @@ -7,7 +7,7 @@ import pytest -from devtools.claim_vs_evidence import build_report +from devtools.claim_vs_evidence import _economy_rows, build_report from polylogue.archive.actions.followup import classify_failed_followup_evidence from polylogue.demo import seed_demo_archive @@ -236,6 +236,29 @@ def _seed_archive(root: Path) -> None: conn.close() +def test_economy_rows_empty_session_ids_short_circuits(tmp_path: Path) -> None: + conn = sqlite3.connect(tmp_path / "empty.db") + conn.executescript( + """ + CREATE TABLE session_model_usage ( + session_id TEXT NOT NULL, model_name TEXT NOT NULL, + input_tokens INTEGER, output_tokens INTEGER, + cache_read_tokens INTEGER, cache_write_tokens INTEGER, + cost_usd REAL, cost_provenance TEXT + ); + """ + ) + economy = _economy_rows(conn, session_ids=(), silent_by_origin={}) + assert economy == {"by_model": [], "by_origin": []} + + +def test_economy_rows_missing_usage_table_returns_empty(tmp_path: Path) -> None: + conn = sqlite3.connect(tmp_path / "no-usage-table.db") + conn.execute("CREATE TABLE sessions (session_id TEXT PRIMARY KEY, origin TEXT)") + economy = _economy_rows(conn, session_ids=("s1",), silent_by_origin={}) + assert economy == {"by_model": [], "by_origin": []} + + def test_protocol_only_followup_is_ambiguous_not_silent_proceed() -> None: """Hidden reasoning contains no reader-visible acknowledgement evidence.""" assert classify_failed_followup_evidence("inspect state privately") == { From 9163c20d1d8a892db165814a6a81f793b01361b6 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 18 Jul 2026 15:09:31 +0200 Subject: [PATCH 4/4] feat(devtools): represent claim-vs-evidence runs as first-party evidence Problem: the claim-vs-evidence report was a regenerable local artifact committed as JSON; nothing in the archive itself recorded WHAT was asked, WHAT matched, or UNDER WHICH corpus state, so "reproducible" only meant "we committed a report.json" (polylogue-rxdo.13, scope correction It.13). What changed: devtools/claim_vs_evidence_evidence.py registers each report run as durable evidence over the already-landed rxdo query-evidence core (#2813/#2826/#2899) and the FindingAssertion/PublicClaimDeclaration kernel that #3065 (polylogue-3tl.16) already landed for the public-claims projection: - AnalysisDefinition identity: the structured-failure-followup selection (predicate, classification/sensitivity scope, n_min, handler-class taxonomy, classifier definition version) is registered as a content- addressed QueryObject via put_query -- not a DSL-executable plan (the paired next-turn/window-3 lookahead has no predicate-grammar representation), but stable provenance identity, mirroring the "durable identity JSON is provenance, not source syntax to reverse-compile" doctrine already stated in convergence_standing_queries.py. - AnalysisRun receipt: each run's matched tool_result refs become a ResultSetManifest (exactness=capped, persistence_class=finding), and an EvaluationReceipt binds it to source/user/index tier generations plus the runtime build. Both are content-addressed including now_ms, so a retried call at the same instant is a safe no-op while a later regeneration correctly records a new run instead of silently colliding with the first. - Finding rows: the headline silent-proceed rate is written as an AssertionKind.FINDING (finding_kind="claim-vs-evidence", already a registered kind) with query_ref/result_set_ref/evaluation_ref wired to the above. public_claim stays None (no PublicClaimDeclaration) unless the run's own n_min/classified-outcome gates are satisfied -- an unpublishable run gets an honest private evidence record, never a fabricated public rate exposed through the existing public-claims projection. - Privacy stamp: disclosure is "public" only when publishable, "held via omission" otherwise (no public_claim at all), reusing the existing PublicClaimDeclaration.disclosure field rather than inventing new privacy/retention machinery (rxdo.2's excision/dependency-resolver layer stays out of scope, per the addendum's hard boundary). Writes go through open_daemon_connection (the same connection helper polylogue/daemon/convergence_standing_queries.py uses for its own standing- query materialization) via a new opt-in `--materialize-evidence` flag on `devtools workspace claim-vs-evidence` (off by default; report generation stays read-only unless explicitly requested) -- not a bare sqlite3.connect against the live user.db, and not a new generic finding registry, metric/ pattern/cohort/experiment definition system, or scheduler. Verification: devtools test tests/unit/devtools/test_claim_vs_evidence_evidence.py tests/unit/devtools/test_claim_vs_evidence.py tests/unit/storage/test_public_claims_projection.py -- 23 passed. mypy --strict on both touched devtools modules -- clean. devtools verify --quick -- green (run 20260718T130836Z-quick-3900869-7edfa1fe). Live smoke test: ran `--materialize-evidence` against the actual live archive (daemon running concurrently, pid confirmed alive throughout) -- registered 1 QueryObject, 1 ResultSetManifest (20 members), 2 EvaluationReceipts (two runs a few seconds apart), 2 private FINDING assertions; list_public_finding_inputs() correctly returns zero rows (today's n=20 result is honestly not publishable). No corruption, no lock contention with the running daemon. Ref polylogue-rxdo.13 Co-Authored-By: Claude --- devtools/claim_vs_evidence.py | 21 ++ devtools/claim_vs_evidence_evidence.py | 307 ++++++++++++++++++ .../test_claim_vs_evidence_evidence.py | 256 +++++++++++++++ 3 files changed, 584 insertions(+) create mode 100644 devtools/claim_vs_evidence_evidence.py create mode 100644 tests/unit/devtools/test_claim_vs_evidence_evidence.py diff --git a/devtools/claim_vs_evidence.py b/devtools/claim_vs_evidence.py index 3f48c95fc6..06342c8dfb 100644 --- a/devtools/claim_vs_evidence.py +++ b/devtools/claim_vs_evidence.py @@ -104,6 +104,15 @@ def _parser() -> argparse.ArgumentParser: default=None, help="Optional CSV of human labels. Defaults to ack-marker-calibration.labels.csv in --out-dir when present.", ) + parser.add_argument( + "--materialize-evidence", + action="store_true", + help=( + "Register this run's structured-failure selection, matched rows, and headline numbers " + "as durable query/result-set/finding evidence in the archive's user tier (polylogue-rxdo.13). " + "Off by default; report generation stays read-only unless explicitly requested." + ), + ) parser.add_argument("--json", action="store_true", help="Emit JSON report to stdout.") return parser @@ -1174,6 +1183,9 @@ def build_report(args: argparse.Namespace) -> dict[str, Any]: "consequential": sorted(_CONSEQUENTIAL_TOOLS), "other": "Any tool name outside the explicit benign/consequential methodology sets.", }, + "evidence": { + "member_refs": sorted({f"message:{row['tool_result_message_id']}" for row in rows}), + }, "calibration": calibration, "calibration_sample": [ _calibration_row(sample, index) for index, sample in enumerate(calibration_sample, start=1) @@ -1674,6 +1686,15 @@ def main(argv: list[str] | None = None) -> int: except ValueError as exc: print(f"claim-vs-evidence: {exc}", file=sys.stderr) return 2 + if parsed.materialize_evidence: + from devtools.claim_vs_evidence_evidence import materialize_claim_vs_evidence_evidence + + evidence = materialize_claim_vs_evidence_evidence( + report, + archive_root=Path(report["archive_root"]), + now_ms=int(datetime.now(UTC).timestamp() * 1000), + ) + print(f"materialized evidence: {json.dumps(evidence, indent=2, sort_keys=True)}", file=sys.stderr) if parsed.json: sys.stdout.write(json.dumps(report, indent=2, sort_keys=True) + "\n") elif parsed.out_dir is not None: diff --git a/devtools/claim_vs_evidence_evidence.py b/devtools/claim_vs_evidence_evidence.py new file mode 100644 index 0000000000..475aee44bd --- /dev/null +++ b/devtools/claim_vs_evidence_evidence.py @@ -0,0 +1,307 @@ +"""Represent a claim-vs-evidence report run as first-party archive evidence. + +The classification/economy logic stays in ``devtools/claim_vs_evidence.py`` +(the harness). This module only represents the harness's OUTPUT as durable +evidence: a content-addressed :class:`~polylogue.storage.sqlite.query_objects. +QueryObject` for the structured-failure selection (the AnalysisDefinition), a +:class:`~polylogue.storage.sqlite.query_objects.ResultSetManifest` for the +rows the run actually matched, an +:class:`~polylogue.storage.sqlite.query_objects.EvaluationReceipt` binding the +run to tier generations and the runtime build (the AnalysisRun), and +``AssertionKind.FINDING`` rows for the headline numbers (polylogue-rxdo.13). + +It writes through the same production primitives the daemon's own +standing-query convergence stage uses +(``polylogue/daemon/convergence_standing_queries.py``) via +``open_daemon_connection`` -- not a new generic finding registry, not a +metric/pattern/cohort/experiment definition system, and not a scheduler. A +finding is written with ``public_claim=None`` (no ``PublicClaimDeclaration``) +unless the run's own construct-validity gates (``n_min``, non-zero classified +outcomes) are satisfied, so an unpublishable run still gets an honest private +evidence record without ever exposing a degenerate rate as a public claim. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, TypedDict + +from polylogue.archive.query.production_evaluator import ( + _index_epoch, # planner-internal seam, reused for the same tier-generation identity + _polylogue_runtime_build_ref, + _tier_generation, +) +from polylogue.core.hashing import hash_payload +from polylogue.core.json import JSONValue +from polylogue.core.query_identity import JsonValue +from polylogue.core.query_identity import query_ref as _query_object_ref +from polylogue.core.query_identity import result_set_ref as _result_set_object_ref +from polylogue.storage.sqlite.archive_tiers.user_write import ( + ArchiveAssertionEnvelope, + FindingAssertion, + PublicClaimDeclaration, + upsert_findings_as_assertions, +) +from polylogue.storage.sqlite.connection_profile import open_daemon_connection +from polylogue.storage.sqlite.query_objects import ( + EvaluationReceipt, + QueryObject, + ResultSetManifest, + get_result_set, + membership_merkle_root, + put_evaluation_receipt, + put_query, + put_result_set, +) + +# Bump when the classifier's silent/acknowledged/ambiguous taxonomy or the +# handler-class split changes meaning, so the AnalysisDefinition identity +# (query_hash) changes with it instead of silently reusing a stale one. +CLASSIFIER_DEFINITION_VERSION = "2" +ANALYSIS_TARGET_REF = "analysis:claim-vs-evidence" +_QUERY_GRAIN = "structured-failure-followup" +_QUERY_LANE = "analysis" +_QUERY_RANK_POLICY = "origin,session_id,tool_id,tool_result_message_id" + + +class MaterializedEvidence(TypedDict): + query_ref: str + result_set_ref: str + receipt_id: str + finding_assertion_ids: list[str] + public_claim_written: bool + + +def build_query_definition(report: dict[str, Any]) -> dict[str, JsonValue]: + """Return the content-addressed AnalysisDefinition payload for one report run. + + Deliberately not a DSL-executable plan: the paired next-assistant-turn + lookup with window-3 lookahead has no representation in the query + predicate grammar today. This is provenance identity, mirroring the + ``convergence_standing_queries`` doctrine that durable identity JSON is + "provenance, not source syntax to reverse-compile" -- the harness in + ``devtools/claim_vs_evidence.py`` remains the sole executor. + """ + frame = report["sample_frame"] + return { + "kind": "analysis-selection", + "analysis_id": "claim-vs-evidence", + "classifier_module": "polylogue.archive.actions.followup", + "classifier_function": "classify_failed_followup_evidence", + "classifier_definition_version": CLASSIFIER_DEFINITION_VERSION, + "failure_predicate": frame["failure_predicate"], + "classification_scope": frame["classification_scope"], + "sensitivity_scope": frame["sensitivity_scope"], + "selection_strategy": frame["selection_strategy"], + "n_min": frame["n_min"], + "limit": report["limit"], + "handler_class_definition": report["handler_class_definition"], + } + + +def build_result_set_members(report: dict[str, Any]) -> tuple[str, ...]: + """Return the sorted ``message:`` refs the run actually classified.""" + return tuple(report["evidence"]["member_refs"]) + + +def build_evaluation_receipt( + archive_root: Path, + index_db: Path, + *, + query_hash: str, + result_set_id: str, + created_at_ms: int, +) -> EvaluationReceipt: + """Bind one run to its source/user/index tier generations and runtime build. + + ``receipt_id`` is content-addressed (not a random UUID, unlike + ``ArchiveCanonicalPlanEvaluator``'s per-execution telemetry receipts) over + every field ``put_evaluation_receipt`` itself treats as significant, + including ``created_at_ms`` -- that function already rejects reusing a + receipt id with a changed ``created_at_ms``, so folding it into the hash + is what lets two calls at the same ``created_at_ms`` collapse into one + safe no-op while two calls at different times correctly get distinct + receipts instead of a spurious conflict. + """ + source_generation = _tier_generation(archive_root / "source.db", label="source") + user_generation = _tier_generation(archive_root / "user.db", label="user") + index_generation = _index_epoch(index_db) + runtime_build_ref = _polylogue_runtime_build_ref() + receipt_digest = hash_payload( + [ + query_hash, + result_set_id, + source_generation, + user_generation, + index_generation, + runtime_build_ref, + created_at_ms, + ] + ) + receipt_id = f"receipt-{receipt_digest}" + return EvaluationReceipt( + receipt_id=receipt_id, + source_generation=source_generation, + user_generation=user_generation, + index_generation=index_generation, + runtime_build_ref=runtime_build_ref, + ) + + +def build_findings( + report: dict[str, Any], + *, + query_reference: str, + result_set_reference: str, + receipt: EvaluationReceipt, +) -> list[FindingAssertion]: + """Return the headline-number FindingAssertions for one report run. + + ``public_claim`` stays ``None`` (no PublicClaimDeclaration) unless the + aggregate rate actually clears the run's own ``n_min``/classified-outcome + gates -- an unpublishable run still gets an honest private evidence + record, never a fabricated public rate. + """ + frame = report["sample_frame"] + totals = report["totals"] + rates = report["rates"] + run_ref = f"run:claim-vs-evidence-{report['captured_at']}" + aggregate_publishable = rates["publication_status"] == "supported" and rates["silent_rate_lower_bound"] is not None + statistic: dict[str, JSONValue] = { + "op": "lower_bound", + "value": rates["silent_rate_lower_bound"], + "unit": "ratio", + "numerator": totals["silent_proceed"], + "denominator": totals["failed_outcomes"], + "ambiguous": totals["ambiguous"], + "classified_outcomes": totals.get("classified_outcomes"), + } + body_text = ( + f"Structured-failure follow-up classification over {frame['inspected_structured_failures']} " + f"inspected failures: {totals['acknowledged']} acknowledged, {totals['silent_proceed']} silent, " + f"{totals['ambiguous']} ambiguous." + ) + public_claim: PublicClaimDeclaration | None = None + if aggregate_publishable: + body_text = ( + f"In one bounded private-archive sample, {totals['silent_proceed']} of " + f"{totals['failed_outcomes']} inspected structured failures were followed by silent " + f"continuation on the next assistant turn, a {rates['silent_rate_lower_bound']:.1%} lower bound." + ) + public_claim = PublicClaimDeclaration( + publication=body_text, + scope=( + f"One private archive; {frame['inspected_structured_failures']} inspected structured " + "failures from the run's bounded sample frame; next assistant turn only." + ), + caveat=( + "This is not a population estimate; ambiguous rows are excluded from the classified " + "denominator, and support must be recomputed when the evidence epoch, definition, or " + "frame changes." + ), + public_evidence_refs=("file:docs/findings/claim-vs-evidence.md",), + disclosure="public", + ) + return [ + FindingAssertion( + claim_key="finding.silent-proceed-lower-bound", + target_ref=ANALYSIS_TARGET_REF, + body_text=body_text, + finding_kind="claim-vs-evidence", + statistic=statistic, + n=totals["failed_outcomes"], + query_ref=query_reference, + result_set_ref=result_set_reference, + detector_ref=run_ref, + evidence_refs=("file:docs/findings/claim-vs-evidence.md",), + source_epoch=report["captured_at"], + evaluation_ref=f"receipt:{receipt.receipt_id}", + frame_ref=query_reference, + public_claim=public_claim, + ) + ] + + +def materialize_claim_vs_evidence_evidence( + report: dict[str, Any], + *, + archive_root: Path, + now_ms: int, +) -> MaterializedEvidence: + """Register one report run's query, result set, receipt, and findings. + + Writes through ``open_daemon_connection`` (the same connection helper the + daemon's own standing-query convergence stage uses), so this coexists + with the running daemon's single-writer discipline instead of bypassing + it with a bare ``sqlite3.connect``. + + The AnalysisDefinition (query) and its matched-row ResultSetManifest are + content-addressed: identical selection logic and identical matched rows + always resolve to the same identity, at any ``now_ms``. The AnalysisRun + receipt and its FindingAssertion are scoped to ``now_ms``: calling this + twice with the same ``report`` and the same ``now_ms`` is a safe retry + no-op, but calling it again at a later ``now_ms`` records a new run and a + new finding row even if the numbers happen to match, because the archive + should carry that a re-verification happened under a later tier state -- + not silently collapse repeated regenerations into one row. + """ + index_db = Path(report["index_db"]) + query_definition = build_query_definition(report) + member_refs = build_result_set_members(report) + conn = open_daemon_connection(archive_root / "user.db", timeout=30.0) + try: + query: QueryObject = put_query( + conn, + query_definition, + grain=_QUERY_GRAIN, + lane=_QUERY_LANE, + rank_policy=_QUERY_RANK_POLICY, + created_at_ms=now_ms, + ) + query_reference = _query_object_ref(query.query_hash).format() + result_set_id = f"finding-{membership_merkle_root(member_refs)}" + result_set: ResultSetManifest | None = get_result_set(conn, result_set_id) + if result_set is None: + result_set = put_result_set( + conn, + result_set_id=result_set_id, + query_hash=query.query_hash, + grain=_QUERY_GRAIN, + corpus_epoch=_index_epoch(index_db), + member_refs=member_refs, + exactness="capped", + persistence_class="finding", + created_at_ms=now_ms, + ) + result_set_reference = _result_set_object_ref(result_set.result_set_id).format() + receipt = build_evaluation_receipt( + archive_root, + index_db, + query_hash=query.query_hash, + result_set_id=result_set.result_set_id, + created_at_ms=now_ms, + ) + put_evaluation_receipt( + conn, + query_hash=query.query_hash, + receipt=receipt, + result_set_id=result_set.result_set_id, + created_at_ms=now_ms, + ) + findings = build_findings( + report, + query_reference=query_reference, + result_set_reference=result_set_reference, + receipt=receipt, + ) + envelopes: list[ArchiveAssertionEnvelope] = upsert_findings_as_assertions(conn, findings, now_ms=now_ms) + conn.commit() + finally: + conn.close() + return { + "query_ref": query_reference, + "result_set_ref": result_set_reference, + "receipt_id": receipt.receipt_id, + "finding_assertion_ids": [envelope.assertion_id for envelope in envelopes], + "public_claim_written": any(finding.public_claim is not None for finding in findings), + } diff --git a/tests/unit/devtools/test_claim_vs_evidence_evidence.py b/tests/unit/devtools/test_claim_vs_evidence_evidence.py new file mode 100644 index 0000000000..9e3953d359 --- /dev/null +++ b/tests/unit/devtools/test_claim_vs_evidence_evidence.py @@ -0,0 +1,256 @@ +"""Tests for representing claim-vs-evidence report runs as archive evidence.""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path +from typing import Any + +from devtools.claim_vs_evidence_evidence import ( + build_findings, + build_query_definition, + build_result_set_members, + materialize_claim_vs_evidence_evidence, +) +from polylogue.core.enums import AssertionKind +from polylogue.core.query_identity import query_hash_for_plan +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.archive_tiers.user_write import list_assertion_claims +from polylogue.storage.sqlite.finding_provenance import list_public_finding_inputs +from polylogue.storage.sqlite.query_objects import get_query, get_result_set + + +def _report( + *, + archive_root: Path, + silent: int, + acknowledged: int, + ambiguous: int, + n_min: int = 30, + member_refs: tuple[str, ...] = ("message:s1:tool-1-result", "message:s1:tool-2-result"), + captured_at: str = "2026-07-18T00:00:00+00:00", +) -> dict[str, Any]: + classified = silent + acknowledged + failed = classified + ambiguous + publishable = failed >= n_min and classified >= n_min + return { + "captured_at": captured_at, + "archive_root": str(archive_root), + "index_db": str(archive_root / "index.db"), + "limit": 5000, + "sample_frame": { + "inspected_structured_failures": failed, + "failure_predicate": "tool_result_is_error = 1 OR tool_result_exit_code != 0", + "classification_scope": "immediately following assistant message only", + "sensitivity_scope": "next 3 assistant messages", + "selection_strategy": "origin-stratified bounded sample", + "n_min": n_min, + }, + "totals": { + "failed_outcomes": failed, + "acknowledged": acknowledged, + "silent_proceed": silent, + "ambiguous": ambiguous, + "classified_outcomes": classified, + }, + "rates": { + "publication_status": "supported" if publishable else "not_supported", + "silent_rate_lower_bound": (silent / failed) if publishable else None, + }, + "handler_class_definition": { + "benign_recovery": ["glob", "grep"], + "consequential": ["bash", "edit"], + "other": "any other tool", + }, + "evidence": {"member_refs": sorted(member_refs)}, + } + + +def test_build_query_definition_is_content_addressed(tmp_path: Path) -> None: + report_a = _report(archive_root=tmp_path, silent=1, acknowledged=1, ambiguous=1, n_min=30) + report_b = _report(archive_root=tmp_path, silent=1, acknowledged=1, ambiguous=1, n_min=30) + report_c = _report(archive_root=tmp_path, silent=1, acknowledged=1, ambiguous=1, n_min=50) + + definition_a = build_query_definition(report_a) + definition_b = build_query_definition(report_b) + definition_c = build_query_definition(report_c) + + hash_a = query_hash_for_plan(definition_a, grain="g", lane="l", rank_policy="r") + hash_b = query_hash_for_plan(definition_b, grain="g", lane="l", rank_policy="r") + hash_c = query_hash_for_plan(definition_c, grain="g", lane="l", rank_policy="r") + + assert hash_a == hash_b + assert hash_a != hash_c + + +def test_build_result_set_members_returns_sorted_refs(tmp_path: Path) -> None: + report = _report( + archive_root=tmp_path, + silent=1, + acknowledged=1, + ambiguous=1, + member_refs=("message:s1:z", "message:s1:a"), + ) + assert build_result_set_members(report) == ("message:s1:a", "message:s1:z") + + +def test_build_findings_omits_public_claim_when_not_publishable(tmp_path: Path) -> None: + report = _report(archive_root=tmp_path, silent=2, acknowledged=2, ambiguous=16, n_min=30) + from polylogue.storage.sqlite.query_objects import EvaluationReceipt + + receipt = EvaluationReceipt( + receipt_id="receipt-test", + source_generation="source:absent", + user_generation="user:absent", + index_generation="index:absent", + runtime_build_ref="polylogue:test", + ) + findings = build_findings( + report, + query_reference="query:" + "0" * 64, + result_set_reference="result-set:test", + receipt=receipt, + ) + assert len(findings) == 1 + assert findings[0].public_claim is None + assert "acknowledged" in findings[0].body_text + + +def test_build_findings_includes_public_claim_when_publishable(tmp_path: Path) -> None: + report = _report(archive_root=tmp_path, silent=20, acknowledged=15, ambiguous=5, n_min=30) + from polylogue.storage.sqlite.query_objects import EvaluationReceipt + + receipt = EvaluationReceipt( + receipt_id="receipt-test", + source_generation="source:absent", + user_generation="user:absent", + index_generation="index:absent", + runtime_build_ref="polylogue:test", + ) + findings = build_findings( + report, + query_reference="query:" + "0" * 64, + result_set_reference="result-set:test", + receipt=receipt, + ) + assert len(findings) == 1 + assert findings[0].public_claim is not None + assert findings[0].public_claim.disclosure == "public" + assert "50.0%" in findings[0].public_claim.publication + + +def test_materialize_end_to_end_publishable_run_round_trips_through_public_claims(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + archive_root.mkdir() + initialize_archive_database(archive_root / "user.db", ArchiveTier.USER) + report = _report( + archive_root=archive_root, + silent=20, + acknowledged=15, + ambiguous=5, + n_min=30, + member_refs=("message:s1:tool-1-result", "message:s1:tool-2-result"), + ) + + result = materialize_claim_vs_evidence_evidence(report, archive_root=archive_root, now_ms=1_000) + + assert result["public_claim_written"] is True + assert len(result["finding_assertion_ids"]) == 1 + + conn = sqlite3.connect(archive_root / "user.db") + conn.row_factory = sqlite3.Row + try: + query_hash = result["query_ref"].removeprefix("query:") + query = get_query(conn, query_hash) + assert query is not None + assert query.grain == "structured-failure-followup" + + result_set_id = result["result_set_ref"].removeprefix("result-set:") + result_set = get_result_set(conn, result_set_id) + assert result_set is not None + assert result_set.member_count == 2 + assert result_set.exactness == "capped" + assert result_set.persistence_class == "finding" + + findings = list_assertion_claims(conn, kinds=(AssertionKind.FINDING,), statuses=None) + assert len(findings) == 1 + assert findings[0].assertion_id in result["finding_assertion_ids"] + + public_inputs = list_public_finding_inputs(conn) + assert len(public_inputs) == 1 + assert public_inputs[0].claim_key == "finding.silent-proceed-lower-bound" + assert public_inputs[0].disclosure == "public" + finally: + conn.close() + + +def test_materialize_unpublishable_run_writes_private_finding_without_public_claim(tmp_path: Path) -> None: + archive_root = tmp_path / "archive" + archive_root.mkdir() + initialize_archive_database(archive_root / "user.db", ArchiveTier.USER) + report = _report(archive_root=archive_root, silent=2, acknowledged=2, ambiguous=16, n_min=30) + + result = materialize_claim_vs_evidence_evidence(report, archive_root=archive_root, now_ms=1_000) + + assert result["public_claim_written"] is False + assert len(result["finding_assertion_ids"]) == 1 + + conn = sqlite3.connect(archive_root / "user.db") + conn.row_factory = sqlite3.Row + try: + findings = list_assertion_claims(conn, kinds=(AssertionKind.FINDING,), statuses=None) + assert len(findings) == 1 + public_inputs = list_public_finding_inputs(conn) + assert public_inputs == () + finally: + conn.close() + + +def test_materialize_is_idempotent_for_a_retried_identical_call(tmp_path: Path) -> None: + """A retried write (same report, same wall-clock) must not duplicate rows.""" + archive_root = tmp_path / "archive" + archive_root.mkdir() + initialize_archive_database(archive_root / "user.db", ArchiveTier.USER) + report = _report(archive_root=archive_root, silent=20, acknowledged=15, ambiguous=5, n_min=30) + + first = materialize_claim_vs_evidence_evidence(report, archive_root=archive_root, now_ms=1_000) + second = materialize_claim_vs_evidence_evidence(report, archive_root=archive_root, now_ms=1_000) + + assert first == second + + conn = sqlite3.connect(archive_root / "user.db") + conn.row_factory = sqlite3.Row + try: + findings = list_assertion_claims(conn, kinds=(AssertionKind.FINDING,), statuses=None) + assert len(findings) == 1 + finally: + conn.close() + + +def test_materialize_at_a_later_time_reuses_query_and_result_set_but_records_a_new_run(tmp_path: Path) -> None: + """A genuine regeneration keeps the stable AnalysisDefinition/result-set identity + + but records its own AnalysisRun receipt and finding row -- the archive should + carry that a re-verification happened at a later corpus/tier state, not silently + collapse it into the first run. + """ + archive_root = tmp_path / "archive" + archive_root.mkdir() + initialize_archive_database(archive_root / "user.db", ArchiveTier.USER) + report = _report(archive_root=archive_root, silent=20, acknowledged=15, ambiguous=5, n_min=30) + + first = materialize_claim_vs_evidence_evidence(report, archive_root=archive_root, now_ms=1_000) + second = materialize_claim_vs_evidence_evidence(report, archive_root=archive_root, now_ms=2_000) + + assert first["query_ref"] == second["query_ref"] + assert first["result_set_ref"] == second["result_set_ref"] + assert first["finding_assertion_ids"] != second["finding_assertion_ids"] + + conn = sqlite3.connect(archive_root / "user.db") + conn.row_factory = sqlite3.Row + try: + findings = list_assertion_claims(conn, kinds=(AssertionKind.FINDING,), statuses=None) + assert len(findings) == 2 + finally: + conn.close()