diff --git a/devtools/claim_vs_evidence.py b/devtools/claim_vs_evidence.py
index 6f871e722c..06342c8dfb 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
@@ -103,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
@@ -137,12 +147,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 +922,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 +1101,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,11 +1173,19 @@ 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),
"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)
@@ -1029,6 +1208,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 +1224,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 +1276,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 +1559,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 +1625,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
@@ -1480,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/docs/findings/claim-vs-evidence.md b/docs/findings/claim-vs-evidence.md
index 7b8cc61723..633000b1ee 100644
--- a/docs/findings/claim-vs-evidence.md
+++ b/docs/findings/claim-vs-evidence.md
@@ -3,7 +3,44 @@
## 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.
+
+**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 (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.
@@ -65,7 +102,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/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/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..a10158e22d 100644
--- a/tests/unit/devtools/test_claim_vs_evidence.py
+++ b/tests/unit/devtools/test_claim_vs_evidence.py
@@ -7,7 +7,8 @@
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
@@ -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,66 @@ 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_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") == {
+ "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 +361,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
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()
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