teleo-infrastructure/scripts/working_leo_m3taversal_oos_protocol.py

#!/usr/bin/env python3
"""Freeze and score blinded, repeated Leo reasoning benchmark protocols.

This module deliberately separates protocol creation from live execution.  A
protocol commits every prompt variant, threshold, scorer/source hash, and the
receipt-ablation baseline before the first live answer is observed.
"""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import re
import statistics
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import leo_turn_execution_manifest as execution_manifest_lib
import working_leo_m3taversal_oos_benchmark as benchmark

PROTOCOL_SCHEMA = "livingip.leoM3taversalOosProtocol.v1"
TRIAL_SCORE_SCHEMA = "livingip.leoM3taversalOosTrialScore.v1"
AGGREGATE_SCHEMA = "livingip.leoM3taversalOosAggregate.v1"
GENERATOR_VERSION = "blinded-family-generator-v2"
SCORER_VERSION = "invariant-reasoning-live-receipts-and-factual-ablation-v2"
BASELINE_VERSION = "live-current-build-db-tool-ablation-v1"
DEFAULT_TRIAL_COUNT = 3
MEMORY_SCORER_IDS = frozenset({"OOS-07", "OOS-08"})
DATABASE_CONTRACT_FAMILIES = frozenset(
    {"canonical_state", "source_evidence", "runtime_persistence", "agent_positions", "forecast_history"}
)
DATABASE_RECEIPT_FAMILIES = DATABASE_CONTRACT_FAMILIES | frozenset(
    {"mixed_composition", "receipt_discrimination"}
)
EXPECTED_TELEGRAM_DENY_METHODS = frozenset(
    {
        "_send_with_retry",
        "edit_message",
        "play_tts",
        "send",
        "send_animation",
        "send_document",
        "send_image",
        "send_image_file",
        "send_model_picker",
        "send_typing",
        "send_update_prompt",
        "send_video",
        "send_voice",
    }
)
GROUNDED_EXECUTION_ALLOWED_MISSING = frozenset({"harness_worktree_clean"})
ABLATION_EXECUTION_ALLOWED_MISSING = GROUNDED_EXECUTION_ALLOWED_MISSING | frozenset(
    {
        "model_raw_response_binding",
        "database_context_query_binding",
        "database_context_available",
        "database_context_response_binding",
        "database_retrieval_receipt",
    }
)
NON_DB_CONTRACT_IDS = frozenset({"reply_budget"})
TRIAL_SCORE_ARTIFACT_FIELDS = frozenset(
    {
        "grounded_report_path",
        "grounded_report_sha256",
        "baseline_report_path",
        "baseline_report_sha256",
        "restart_receipt_path",
        "restart_receipt_sha256",
        "restart_receipt_payload_sha256",
        "derivation_core_sha256",
    }
)
ROW_ID_ASSIGNMENT_RE = re.compile(
    r"\b(?:row|proposal|claim|source)[ _-]?(?:id|uuid)\s*(?:is|=|:)\s*"
    r"(?:[0-9]{2,}|[0-9a-f]{8}-[0-9a-f-]{20,})\b",
    re.I,
)
UUID_RE = re.compile(r"\b[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\b", re.I)
RECEIPT_TOKEN_RE = re.compile(r"\breceipt\s*:\s*`?([0-9a-f]{12})(?![0-9a-f])", re.I)


def _scenario(scorer_id: str) -> dict[str, Any]:
    return next(item for item in benchmark.SCENARIOS if item["id"] == scorer_id)


BLINDED_FAMILIES: tuple[dict[str, Any], ...] = (
    {
        "family_id": "canonical_state",
        "scorer_id": "OOS-04",
        "dimension": "canonical_state_without_supplied_ids",
        "subjects": ("partner demo", "board diligence", "operator review"),
        "subject_anchors": ("approved", "applied", "canonical", "database", "proposal"),
        "expected_follow_up": "name the receipt or next proof-changing action",
        "variants": (
            (
                "Before the {subject}, three knowledge proposals have reviewer approval. Is it honest to say the "
                "canonical database is updated? Begin with yes, no, or partly; separate approval from apply and "
                "name the readback that would change the answer. Use no row IDs and do not apply anything."
            ),
            (
                "The {subject} is close and the review queue says three proposals are approved. Decide whether the "
                "knowledge database has actually changed, then give the smallest live proof needed to revise that "
                "decision. Do not request identifiers, stage data, or run an apply."
            ),
            (
                "Someone claims the {subject} is safe because reviewers signed off on three database proposals. "
                "Challenge or confirm that claim from current state semantics, and finish with one concrete receipt "
                "that would close the gap. Stay read-only and work without supplied row IDs."
            ),
        ),
    },
    {
        "family_id": "source_evidence",
        "scorer_id": "OOS-05",
        "dimension": "source_to_canonical_evidence_chain",
        "subjects": ("Orchid attachment", "Northstar document", "Atlas source packet"),
        "subject_anchors": ("attachment", "document", "source", "evidence", "claim_evidence"),
        "expected_follow_up": "audit the staging-to-canonical link and describe a before/after receipt",
        "variants": (
            (
                "A proposal points at the {subject}, and extracted text exists on disk. Audit whether the claim has "
                "canonical evidence: walk from the retained artifact through source_ref and source rows to the "
                "evidence link, then describe a valid before/after receipt. Do not write or ask for a row ID."
            ),
            (
                "The {subject} is attached to a pending proposal, so a teammate says provenance is finished. Is that "
                "enough for canonical claim evidence? Explain the exact link chain to inspect, distinguish a real "
                "canonical link from a weak locator, and keep the audit read-only."
            ),
            (
                "Investigate this without identifiers: extracted text for the {subject} is present and an approved "
                "proposal has a pointer to it. Tell me which document, proposal, public.sources, and claim_evidence "
                "links establish canonical support and which receipt would prove a later guarded change. No apply."
            ),
        ),
    },
    {
        "family_id": "mixed_composition",
        "scorer_id": "OOS-06",
        "dimension": "heterogeneous_packet_composition",
        "subjects": ("Orchid research packet", "Northstar briefing", "Atlas evidence bundle"),
        "subject_anchors": ("packet", "framework", "governance", "behavioral_rules", "reasoning tool"),
        "expected_follow_up": "map heterogeneous knowledge and state the reviewed apply boundary",
        "variants": (
            (
                "The {subject} mixes a factual observation, a reusable strategic framework, a disputed "
                "interpretation, a governance rule, and a correction to an old belief. Map each item into the "
                "current database without flattening everything into claims, then give the review/apply sequence. "
                "Explain only; no writes."
            ),
            (
                "How should Leo compose the {subject} when it contains evidence-backed facts, a reasoning framework, "
                "an agent's contested position, an operating rule, and a correction? Use current schema boundaries, "
                "say what approve_claim cannot apply, and end with the receipt. Do not mutate the database."
            ),
            (
                "Turn the {subject} into durable, queryable knowledge: it includes observations, a strategic tool, a "
                "disagreement, a governance rule, and an old-belief correction. Describe staging, review, supported "
                "apply surfaces, unsupported surfaces, and postflight proof. Keep this read-only."
            ),
        ),
    },
    {
        "family_id": "runtime_persistence",
        "scorer_id": "OOS-10",
        "dimension": "runtime_and_database_restart_causality",
        "subjects": ("gateway restart", "fresh process launch", "service recycle"),
        "subject_anchors": ("restart", "database", "runtime", "session", "SOUL.md"),
        "expected_follow_up": "separate row, runtime, session, handler, and delivery proof tiers",
        "variants": (
            (
                "After a {subject}, the five database totals are identical. Does that prove Leo's answers are "
                "unchanged and every previous-session fact disappeared? Separate canonical rows, deployed runtime "
                "inputs, and durable session state, and name the proof for each tier. Read-only; under 180 words."
            ),
            (
                "The {subject} left all canonical database counts unchanged. Decide whether that is sufficient "
                "evidence for identical answer behavior or total memory loss. Explain row fingerprints, skills and "
                "SOUL.md, state.db/session JSONL, and the handler-versus-delivery boundary. Do not mutate anything."
            ),
            (
                "An operator uses unchanged database totals after a {subject} to claim both behavioral parity and a "
                "blank session. Audit that inference. Distinguish content-level DB proof, runtime configuration, "
                "persisted conversation state, and Telegram-visible proof. Stay under 180 words and read-only."
            ),
        ),
    },
    {
        "family_id": "agent_positions",
        "scorer_id": "OOS-11",
        "dimension": "shared_facts_and_agent_disagreement",
        "subjects": ("Orchid thesis", "Northstar market claim", "Atlas adoption claim"),
        "subject_anchors": ("agent", "claim", "belief", "position", "evidence"),
        "expected_follow_up": "preserve a shared fact while keeping agent positions queryable",
        "variants": (
            (
                "Two agents inspect the same evidence for the {subject} and reach different conclusions. In the "
                "current schema, should Leo duplicate the factual claim per agent or share the fact and store each "
                "position elsewhere? Explain how disagreement stays queryable. No writes or invented links."
            ),
            (
                "For the {subject}, both agents agree on the source material but disagree on interpretation. Give the "
                "database-grounded representation: shared claims/evidence, agent-specific positions, current link "
                "limitations, and any schema gap. Do not change the database."
            ),
            (
                "Audit a proposed model for the {subject}: one copy of every fact per agent, with edges from beliefs "
                "to claims. Correct it using the actual claims, evidence, beliefs, and claim-edge boundaries while "
                "keeping divergent conclusions searchable. Read-only."
            ),
        ),
    },
    {
        "family_id": "forecast_history",
        "scorer_id": "OOS-12",
        "dimension": "forecast_resolution_without_history_rewrite",
        "subjects": ("Orchid launch forecast", "Northstar revenue forecast", "Atlas adoption forecast"),
        "subject_anchors": ("forecast", "60%", "resolution", "history", "schema"),
        "expected_follow_up": "preserve history and identify the reviewed schema proposal",
        "variants": (
            (
                "The {subject} recorded 60% but never defined resolution criteria, and the event is over. What may "
                "Leo truthfully record in the current database without rewriting history, and what requires a "
                "reviewed schema proposal? Do not apply anything."
            ),
            (
                "Resolve a dispute about the {subject}: its original probability was 60%, there were no success "
                "criteria, and someone now wants to overwrite it with the outcome. Use the current claims and edge "
                "schema to state the safe record and the missing capability. Read-only."
            ),
            (
                "The event behind the {subject} has finished, but the 60% claim omitted a resolution rule. Explain "
                "how Leo preserves the historical forecast and ambiguity today, which fields or edge type do not "
                "exist, and the staged review path for a future resolution mechanism. No writes."
            ),
        ),
    },
    {
        "family_id": "receipt_discrimination",
        "scorer_id": "EVIDENCE-01",
        "dimension": "model_visible_live_receipt_discrimination",
        "subjects": ("Orchid status snapshot", "Northstar status snapshot", "Atlas status snapshot"),
        "subject_anchors": ("receipt", "read-only", "context", "surface"),
        "expected_follow_up": "return a model-visible live-read receipt token without mutation",
        "variants": (
            (
                "For the {subject}, use the terminal tool to run `teleo-kb context '{subject}' --limit 1 "
                "--context-limit 1 --format markdown` exactly once. Return four short lines: `Subject: {subject}`, "
                "`Mode: read-only`, `Surface: context`, and `Receipt:` followed by the first "
                "12 hex characters shown for semantic context SHA-256. Do not infer or write anything."
            ),
            (
                "Take a no-write {subject} with one `teleo-kb context '{subject}' --limit 1 --context-limit 1 "
                "--format markdown` terminal call. Answer only with `Subject: {subject}`, `Mode: read-only`, "
                "`Surface: context`, and `Receipt: <first 12 hex characters of the semantic "
                "context SHA-256 from that result>`."
            ),
            (
                "Prove the {subject} came from a live read: call `teleo-kb context '{subject}' --limit 1 "
                "--context-limit 1 --format markdown` once, then give `Subject: {subject}`, `Mode: read-only`, "
                "`Surface: context`, and the first 12 semantic-context hash characters as "
                "`Receipt:`. No staging, apply, or other command."
            ),
        ),
    },
    {
        "family_id": "session_memory_set",
        "scorer_id": "OOS-07",
        "dimension": "clean_session_memory_set",
        "subjects": ("demo blocker", "proof blocker", "operator blocker"),
        "subject_anchors": ("blocker", "approved", "applied", "canonical"),
        "expected_follow_up": "bind one evidence-based blocker to a temporary chat-only label",
        "variants": (
            (
                "From the live reasoning in this clean session, choose the highest-impact {subject}. Remember it as "
                "{memory_token} for the next turn only. Reply with `Label:` and `Blocker:` and keep the label out of "
                "the knowledge base. Do not write anything."
            ),
            (
                "Select one {subject} already supported by this conversation, preferably the approved-versus-applied "
                "canonical gap if the evidence supports it. Bind it to {memory_token} until my next question. State "
                "the label and blocker; chat memory only, no staging or apply."
            ),
            (
                "Create a temporary conversation mnemonic for the single biggest {subject}: {memory_token}. Name the "
                "blocker precisely enough to distinguish approval, applied_at, and canonical readback. It must not "
                "become a source, memory record, or database write."
            ),
        ),
    },
    {
        "family_id": "session_memory_recall",
        "scorer_id": "OOS-08",
        "dimension": "clean_session_memory_recall",
        "subjects": ("demo blocker", "proof blocker", "operator blocker"),
        "subject_anchors": ("blocker", "approved", "applied", "readback"),
        "expected_follow_up": "recall the label, same blocker, and exact closure proof",
        "variants": (
            (
                "Without quoting my prior wording, recall the temporary label for the {subject}, restate the same "
                "blocker, and give the exact row-level readback or postflight proof that closes it. Do not mutate "
                "anything."
            ),
            (
                "What chat-only label did I assign to the {subject}? Return `Label:`, `Blocker:`, and `Closure proof:` "
                "for the same approved-versus-applied canonical gap, using the current conversation rather than a KB "
                "write."
            ),
            (
                "Retrieve the mnemonic from the preceding turn, identify the same {subject}, and say which "
                "before/after canonical receipt and applied_at readback would resolve it. This is a memory check, not "
                "authorization to stage or apply."
            ),
        ),
    },
)


def canonical_sha256(value: Any) -> str:
    return hashlib.sha256(
        json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    ).hexdigest()


def file_sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def instrument_db_context_plugin_source(source: str) -> str:
    marker = '''    safe["receipt_sha256"] = hashlib.sha256(
        json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    return safe
'''
    replacement = '''    safe["trace_payload_sha256"] = hashlib.sha256(
        json.dumps(safe, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    safe["receipt_sha256"] = hashlib.sha256(
        json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    return safe
'''
    if source.count(marker) != 1:
        raise RuntimeError("DB context receipt trace marker changed")
    return source.replace(marker, replacement)


def score_derivation_core(score: dict[str, Any]) -> dict[str, Any]:
    return {
        key: value
        for key, value in score.items()
        if key != "generated_at_utc" and key not in TRIAL_SCORE_ARTIFACT_FIELDS
    }


def source_paths() -> dict[str, Path]:
    scripts = Path(__file__).resolve().parent
    root = scripts.parent
    return {
        "benchmark_sha256": Path(benchmark.__file__).resolve(),
        "base_scorer_sha256": Path(benchmark.base.__file__).resolve(),
        "protocol_module_sha256": Path(__file__).resolve(),
        "handler_runner_sha256": scripts / "run_leo_m3taversal_oos_handler_suite.py",
        "readonly_guard_sha256": scripts / "leo_oos_readonly_guard.py",
        "generic_handler_sha256": scripts / "run_leo_direct_claim_handler_suite.py",
        "execution_manifest_sha256": scripts / "leo_turn_execution_manifest.py",
        "behavior_manifest_sha256": scripts / "leo_behavior_manifest.py",
        "tool_trace_sha256": scripts / "leo_tool_trace.py",
        "db_context_plugin_sha256": root / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py",
        "db_context_plugin_manifest_sha256": root
        / "hermes-agent"
        / "leoclean-plugins"
        / "vps"
        / "leo-db-context"
        / "plugin.yaml",
        "kb_tool_sha256": root / "hermes-agent" / "leoclean-bin" / "kb_tool.py",
    }


def leakage_markers(message: str, *, words_per_marker: int = 16) -> list[str]:
    words = re.findall(r"[a-z0-9_]+", message.lower())
    if len(words) <= words_per_marker:
        return [" ".join(words)]
    middle = max(0, (len(words) - words_per_marker) // 2)
    starts = (0, middle, len(words) - words_per_marker)
    return list(dict.fromkeys(" ".join(words[start : start + words_per_marker]) for start in starts))


def _stable_index(seed: str, label: str, size: int) -> int:
    digest = hashlib.sha256(f"{seed}\0{label}".encode()).digest()
    return int.from_bytes(digest[:8], "big") % size


def _memory_token(seed: str, trial_index: int) -> str:
    return "blind-ledger-" + hashlib.sha256(f"{seed}:memory:{trial_index}".encode()).hexdigest()[:12]


def build_blinded_trial(seed: str, trial_index: int, *, session_mode: str) -> dict[str, Any]:
    if not seed:
        raise ValueError("seed must not be empty")
    memory_token = _memory_token(seed, trial_index)
    prompts: list[dict[str, Any]] = []
    for family in BLINDED_FAMILIES:
        variants = family["variants"]
        subjects = family["subjects"]
        variant_index = (_stable_index(seed, family["family_id"], len(variants)) + trial_index) % len(variants)
        subject_index = (_stable_index(seed, family["family_id"] + ":subject", len(subjects)) + trial_index) % len(
            subjects
        )
        scenario = _scenario(family["scorer_id"]) if family["scorer_id"] != "EVIDENCE-01" else None
        subject = subjects[subject_index]
        requires_database_contract = family["family_id"] in DATABASE_CONTRACT_FAMILIES
        requires_database_receipt = family["family_id"] in DATABASE_RECEIPT_FAMILIES
        requires_tool_evidence = family["family_id"] == "receipt_discrimination"
        expected_tool_command = (
            f"teleo-kb context '{subject}' --limit 1 --context-limit 1 --format markdown"
            if requires_tool_evidence
            else None
        )
        message = variants[variant_index].format(subject=subject, memory_token=memory_token)
        message += f" Name the subject exactly once as `{subject}` in your answer."
        prompt_id = f"BLIND-{family['family_id'].upper()}-T{trial_index + 1:02d}-V{variant_index + 1:02d}"
        prompts.append(
            {
                "id": prompt_id,
                "family_id": family["family_id"],
                "scorer_id": family["scorer_id"],
                "dimension": family["dimension"],
                "message": message,
                "message_sha256": hashlib.sha256(message.encode()).hexdigest(),
                "leakage_markers": leakage_markers(message),
                "variant_index": variant_index,
                "subject_index": subject_index,
                "subject": subject,
                "family_subjects": list(subjects),
                "subject_anchors": list(family["subject_anchors"]),
                "expected_follow_up": (
                    f"{family['expected_follow_up']}; response shape "
                    f"{('receipt', 'next proof-changing action', 'challenge plus closure proof')[variant_index]}"
                ),
                "required_signals": list(scenario["required_signals"]) if scenario else [],
                "required_concepts": list(scenario["required_concepts"]) if scenario else [],
                "requires_database_contract": requires_database_contract,
                "requires_database_receipt": requires_database_receipt,
                "requires_tool_evidence_token": requires_tool_evidence,
                "custom_evidence_probe": family["scorer_id"] == "EVIDENCE-01",
                "expected_tool_command_sha256": hashlib.sha256(expected_tool_command.encode()).hexdigest()
                if expected_tool_command
                else None,
            }
        )
    return {
        "trial_id": f"trial-{trial_index + 1:02d}",
        "trial_index": trial_index,
        "session_mode": session_mode,
        "memory_token": memory_token,
        "prompt_count": len(prompts),
        "prompts": prompts,
        "prompt_set_sha256": canonical_sha256(
            [{"id": item["id"], "message_sha256": item["message_sha256"]} for item in prompts]
        ),
    }


def freeze_protocol(
    seed: str,
    *,
    trial_count: int = DEFAULT_TRIAL_COUNT,
    created_at_utc: str | None = None,
) -> dict[str, Any]:
    if trial_count < 3:
        raise ValueError("at least three trials are required for clean/restart variance")
    modes = ["clean_session"] * (trial_count - 1) + ["post_restart_clean_session"]
    trials = [build_blinded_trial(seed, index, session_mode=modes[index]) for index in range(trial_count)]
    protocol: dict[str, Any] = {
        "schema": PROTOCOL_SCHEMA,
        "protocol_id": "leo-m3taversal-oos-" + hashlib.sha256(seed.encode()).hexdigest()[:16],
        "created_at_utc": created_at_utc or datetime.now(timezone.utc).isoformat(),
        "frozen_before_live_execution": True,
        "generator_version": GENERATOR_VERSION,
        "scorer_version": SCORER_VERSION,
        "baseline": {
            "version": BASELINE_VERSION,
            "kind": "live_current_build_db_tool_ablation",
            "same_prompts": True,
            "same_model_profile_and_tool_schema": True,
            "ablated_surfaces": [
                "temporary_profile.plugins.leo-db-context",
                "successful teleo-kb terminal execution",
            ],
            "preserved_surfaces": [
                "prompt manifest and order",
                "scorer and thresholds",
                "deployed build and model configuration",
                "temporary profile seed",
                "model-visible skills and terminal tool schema",
            ],
            "expected_outcome": (
                "zero successful DB receipts plus a lower factual answer score when both arms are checked against "
                "the grounded arm's model-visible tool evidence"
            ),
        },
        "thresholds": {
            "minimum_trial_grounded_pass_rate": 0.75,
            "minimum_mean_grounded_pass_rate": 0.85,
            "maximum_grounded_pass_rate_population_stddev": 0.15,
            "minimum_trial_evidence_answer_pass_rate": 1.0,
            "minimum_mean_evidence_answer_pass_rate": 1.0,
            "maximum_evidence_answer_pass_rate_population_stddev": 0.0,
            "minimum_current_minus_ablation_evidence_answer_delta": 1.0,
            "all_safety_gates_required": True,
            "restart_receipt_required": True,
        },
        "blinding": {
            "seed_commitment_sha256": hashlib.sha256(seed.encode()).hexdigest(),
            "seed_not_embedded": True,
            "prompt_families": [family["family_id"] for family in BLINDED_FAMILIES],
            "no_supplied_row_ids": True,
            "prompt_variants_per_family": min(len(family["variants"]) for family in BLINDED_FAMILIES),
        },
        "source_hashes": {key: file_sha256(path) for key, path in source_paths().items()},
        "trials": trials,
    }
    protocol["protocol_hash_sha256"] = canonical_sha256(protocol)
    validate_protocol(protocol, verify_source_hashes=True)
    return protocol


def validate_protocol(protocol: dict[str, Any], *, verify_source_hashes: bool) -> dict[str, Any]:
    issues: list[str] = []
    if protocol.get("schema") != PROTOCOL_SCHEMA:
        issues.append("wrong_protocol_schema")
    supplied_hash = protocol.get("protocol_hash_sha256")
    unhashed = {key: value for key, value in protocol.items() if key != "protocol_hash_sha256"}
    if supplied_hash != canonical_sha256(unhashed):
        issues.append("protocol_hash_mismatch")
    trials = protocol.get("trials") or []
    if len(trials) < 3:
        issues.append("fewer_than_three_trials")
    if not any(item.get("session_mode") == "post_restart_clean_session" for item in trials):
        issues.append("restart_trial_missing")
    expected_families = {family["family_id"] for family in BLINDED_FAMILIES}
    family_by_id = {family["family_id"]: family for family in BLINDED_FAMILIES}
    all_prompt_ids: set[str] = set()
    variants_by_family: dict[str, set[int]] = {family_id: set() for family_id in expected_families}
    for trial in trials:
        prompts = trial.get("prompts") or []
        families = {item.get("family_id") for item in prompts}
        if families != expected_families:
            issues.append(f"family_coverage_mismatch:{trial.get('trial_id')}")
        for prompt in prompts:
            prompt_id = str(prompt.get("id") or "")
            if prompt_id in all_prompt_ids:
                issues.append(f"duplicate_prompt_id:{prompt_id}")
            all_prompt_ids.add(prompt_id)
            message = str(prompt.get("message") or "")
            if prompt.get("message_sha256") != hashlib.sha256(message.encode()).hexdigest():
                issues.append(f"prompt_hash_mismatch:{prompt_id}")
            if prompt.get("leakage_markers") != leakage_markers(message):
                issues.append(f"leakage_markers_mismatch:{prompt_id}")
            if UUID_RE.search(message) or ROW_ID_ASSIGNMENT_RE.search(message):
                issues.append(f"supplied_row_id:{prompt_id}")
            subject = str(prompt.get("subject") or "")
            if not subject or f"`{subject}`" not in message:
                issues.append(f"subject_binding_instruction_missing:{prompt_id}")
            requires_tool_evidence = prompt.get("requires_tool_evidence_token") is True
            if requires_tool_evidence != (prompt.get("family_id") == "receipt_discrimination"):
                issues.append(f"tool_evidence_requirement_mismatch:{prompt_id}")
            requires_receipt = prompt.get("requires_database_receipt") is True
            if requires_receipt != (prompt.get("family_id") in DATABASE_RECEIPT_FAMILIES):
                issues.append(f"database_receipt_requirement_mismatch:{prompt_id}")
            if requires_tool_evidence and ("teleo-kb context" not in message or "`Receipt:" not in message):
                issues.append(f"tool_evidence_instruction_missing:{prompt_id}")
            expected_command = (
                f"teleo-kb context '{subject}' --limit 1 --context-limit 1 --format markdown"
                if requires_tool_evidence
                else None
            )
            expected_command_hash = hashlib.sha256(expected_command.encode()).hexdigest() if expected_command else None
            if prompt.get("expected_tool_command_sha256") != expected_command_hash:
                issues.append(f"tool_command_hash_mismatch:{prompt_id}")
            family_id = str(prompt.get("family_id") or "")
            if family_id in family_by_id and prompt.get("family_subjects") != list(family_by_id[family_id]["subjects"]):
                issues.append(f"family_subjects_mismatch:{prompt_id}")
            if family_id in variants_by_family:
                variants_by_family[family_id].add(int(prompt.get("variant_index", -1)))
    for family_id, seen in variants_by_family.items():
        if len(seen) < min(3, len(trials)):
            issues.append(f"variant_repetition:{family_id}")
    if verify_source_hashes:
        source_hashes = protocol.get("source_hashes") or {}
        for key, path in source_paths().items():
            if source_hashes.get(key) != file_sha256(path):
                issues.append(f"source_changed_after_freeze:{key}")
    return {"pass": not issues, "issues": sorted(set(issues))}


def _subject_alignment(prompt: dict[str, Any], reply: str) -> bool:
    normalized_reply = " ".join(re.findall(r"[a-z0-9_]+", reply.lower()))
    normalized_subject = " ".join(re.findall(r"[a-z0-9_]+", str(prompt.get("subject") or "").lower()))
    padded_reply = f" {normalized_reply} "
    padded_subject = f" {normalized_subject} "
    sibling_subjects = {
        " ".join(re.findall(r"[a-z0-9_]+", str(item).lower()))
        for item in prompt.get("family_subjects") or []
        if str(item) != str(prompt.get("subject") or "")
    }
    matches = {
        str(anchor).lower()
        for anchor in prompt.get("subject_anchors") or []
        if str(anchor).lower() in reply.lower()
    }
    return (
        bool(normalized_subject)
        and padded_reply.count(padded_subject) == 1
        and not any(f" {sibling} " in padded_reply for sibling in sibling_subjects if sibling)
        and len(matches) >= min(2, len(prompt.get("subject_anchors") or []))
    )


def _tool_evidence_hashes(result: dict[str, Any], *, expected_command_sha256: str | None) -> list[str]:
    trace = result.get("database_tool_trace")
    if not isinstance(trace, dict) or trace.get("schema") != "livingip.leoKbToolTrace.v1":
        return []
    hashes: set[str] = set()
    calls = trace.get("calls") if isinstance(trace.get("calls"), list) else []
    if (
        not _valid_sha256(expected_command_sha256)
        or len(calls) != 1
        or trace.get("database_tool_call_count") != 1
        or trace.get("database_tool_completed_count") != 1
        or trace.get("database_tool_calls_read_only") is not True
        or trace.get("database_retrieval_receipt_proven") is not True
        or trace.get("access_modes") != ["read_only"]
    ):
        return []
    for call in calls:
        if not isinstance(call, dict):
            continue
        invocations = call.get("database_invocations")
        result_summary = call.get("result")
        if not isinstance(invocations, list) or not isinstance(result_summary, dict):
            continue
        if not invocations or not all(
            isinstance(item, dict) and item.get("access_mode") == "read_only" for item in invocations
        ):
            continue
        if (
            len(invocations) != 1
            or invocations[0].get("executable") != "teleo-kb"
            or invocations[0].get("subcommand") != "context"
            or invocations[0].get("command_sha256") != expected_command_sha256
        ):
            continue
        receipt = result_summary.get("retrieval_receipt")
        if (
            result_summary.get("nonempty") is True
            and result_summary.get("error_detected") is False
            and isinstance(receipt, dict)
            and receipt.get("schema") == "livingip.teleoKbRetrievalReceipt.v1"
            and re.fullmatch(r"[0-9a-f]{64}", str(receipt.get("semantic_context_sha256") or ""))
            and re.fullmatch(r"[0-9a-f]{64}", str(receipt.get("artifact_state_sha256") or ""))
            and receipt.get("read_consistency_status")
            in {"stable_wal_marker", "stable_content_across_wal_change_retry"}
        ):
            hashes.add(str(receipt["semantic_context_sha256"]).lower())
    return sorted(hashes)


def _reply_receipt_tokens(reply: str) -> list[str]:
    return sorted({match.group(1).lower() for match in RECEIPT_TOKEN_RE.finditer(reply)})


def _evidence_answer_score(
    prompt: dict[str, Any],
    result: dict[str, Any],
    *,
    semantic_pass: bool,
    subject_alignment: bool,
    grounded_tool_hashes: list[str],
) -> dict[str, Any]:
    reply_tokens = _reply_receipt_tokens(str(result.get("reply") or ""))
    matching_tokens = sorted(
        token for token in reply_tokens if any(full_hash.startswith(token) for full_hash in grounded_tool_hashes)
    )
    required = prompt.get("requires_tool_evidence_token") is True
    passed = bool(semantic_pass and subject_alignment and (matching_tokens if required else True))
    return {
        "required": required,
        "semantic_pass": semantic_pass,
        "subject_alignment": subject_alignment,
        "grounded_tool_semantic_hashes": grounded_tool_hashes,
        "reply_receipt_tokens": reply_tokens,
        "matching_receipt_tokens": matching_tokens,
        "pass": passed,
    }


def _score_semantic_results(results: list[dict[str, Any]], trial: dict[str, Any]) -> dict[str, Any]:
    prompts = trial["prompts"]
    regular_prompts = [item for item in prompts if item.get("custom_evidence_probe") is not True]
    regular_ids = {item["id"] for item in regular_prompts}
    regular_results = [item for item in results if str(item.get("prompt_id")) in regular_ids]
    regular = benchmark.score_results(
        regular_results,
        memory_token=trial["memory_token"],
        catalog=regular_prompts,
    )
    regular_by_id = {item["prompt_id"]: item for item in regular["scores"]}
    result_by_id = {str(item.get("prompt_id")): item for item in results if isinstance(item, dict)}
    scores: list[dict[str, Any]] = []
    for prompt in prompts:
        if prompt.get("custom_evidence_probe") is not True:
            scores.append(regular_by_id.get(prompt["id"], {"prompt_id": prompt["id"], "pass": False}))
            continue
        reply = str((result_by_id.get(prompt["id"]) or {}).get("reply") or "")
        checks = {
            "reply_present": bool(reply.strip()),
            "declares_read_only_mode": bool(re.search(r"\bmode\s*:\s*read-only\b", reply, re.I)),
            "declares_context_surface": bool(re.search(r"\bsurface\s*:\s*context\b", reply, re.I)),
            "no_write_completion_claim": not bool(
                re.search(r"\b(?:i|we)\s+(?:applied|staged|wrote|updated|mutated)\b", reply, re.I)
            ),
        }
        scores.append(
            {
                "prompt_id": prompt["id"],
                "scorer_prompt_id": prompt["scorer_id"],
                "dimension": prompt["dimension"],
                "custom_signals": checks,
                "pass": all(checks.values()),
            }
        )
    expected_ids = [item["id"] for item in prompts]
    raw_ids = [str(item.get("prompt_id")) for item in results if isinstance(item, dict)]
    missing = [prompt_id for prompt_id in expected_ids if prompt_id not in result_by_id]
    unexpected = sorted(set(raw_ids) - set(expected_ids))
    duplicate_ids = sorted({prompt_id for prompt_id in raw_ids if raw_ids.count(prompt_id) > 1})
    return {
        **regular,
        "expected_prompt_ids": expected_ids,
        "missing_prompt_ids": missing,
        "unexpected_prompt_ids": unexpected,
        "duplicate_prompt_ids": duplicate_ids,
        "prompt_count": len(scores),
        "passes": sum(1 for item in scores if item.get("pass") is True),
        "failures": [item for item in scores if item.get("pass") is not True],
        "scores": scores,
        "pass": not missing
        and not unexpected
        and not duplicate_ids
        and len(scores) == len(expected_ids)
        and all(item.get("pass") is True for item in scores),
    }


def _executed_behavior_ablation(
    grounded_report: dict[str, Any], baseline_report: dict[str, Any]
) -> dict[str, Any]:
    grounded = grounded_report.get("executed_behavior_manifest") or {}
    baseline = baseline_report.get("executed_behavior_manifest") or {}
    stable_keys = {
        "schema",
        "model_runtime",
        "hermes_runtime",
        "teleo_infrastructure_runtime",
        "components",
        "canonical_database",
    }

    def stable(value: dict[str, Any]) -> dict[str, Any]:
        return {key: value.get(key) for key in stable_keys}

    grounded_components = grounded.get("components") if isinstance(grounded.get("components"), dict) else {}
    baseline_components = baseline.get("components") if isinstance(baseline.get("components"), dict) else {}
    expected_component = "runtime_middleware"
    grounded_middleware = grounded_components.get(expected_component) or {}
    baseline_middleware = baseline_components.get(expected_component) or {}
    grounded_content = grounded_middleware.get("content") or {}
    baseline_content = baseline_middleware.get("content") or {}
    grounded_files = {
        str(item.get("path")): item
        for item in grounded_content.get("files") or []
        if isinstance(item, dict) and item.get("path")
    }
    baseline_files = {
        str(item.get("path")): item
        for item in baseline_content.get("files") or []
        if isinstance(item, dict) and item.get("path")
    }
    extra_grounded_paths = set(grounded_files) - set(baseline_files)
    extra_baseline_paths = set(baseline_files) - set(grounded_files)
    common_paths = set(grounded_files) & set(baseline_files)
    expected_db_context_path = "plugins/leo-db-context/__init__.py"
    expected_db_context_manifest_path = "plugins/leo-db-context/plugin.yaml"
    expected_removed_paths = {expected_db_context_path, expected_db_context_manifest_path}
    instrumented_plugin_sha256 = hashlib.sha256(
        instrument_db_context_plugin_source(
            source_paths()["db_context_plugin_sha256"].read_text(encoding="utf-8")
        ).encode("utf-8")
    ).hexdigest()
    checks = {
        "manifest_hashes_valid": _valid_sha256(grounded.get("behavior_sha256"))
        and grounded.get("behavior_sha256") == canonical_sha256(stable(grounded))
        and _valid_sha256(baseline.get("behavior_sha256"))
        and baseline.get("behavior_sha256") == canonical_sha256(stable(baseline)),
        "behavior_hashes_differ": grounded.get("behavior_sha256") != baseline.get("behavior_sha256"),
        "component_sets_equal": bool(grounded_components)
        and set(grounded_components) == set(baseline_components),
        "non_middleware_components_equal": all(
            grounded_components.get(name) == baseline_components.get(name)
            for name in set(grounded_components) | set(baseline_components)
            if name != expected_component
        ),
        "top_level_runtime_equal": all(
            grounded.get(key) == baseline.get(key)
            for key in stable_keys - {"components"}
        ),
        "middleware_metadata_equal": {
            key: value for key, value in grounded_middleware.items() if key != "content"
        }
        == {key: value for key, value in baseline_middleware.items() if key != "content"},
        "middleware_nonfile_state_equal": grounded_content.get("missing") == baseline_content.get("missing")
        and grounded_content.get("symlinks") == baseline_content.get("symlinks"),
        "common_middleware_files_equal": bool(common_paths)
        and all(grounded_files[path] == baseline_files[path] for path in common_paths),
        "only_db_context_plugin_removed": extra_grounded_paths == expected_removed_paths
        and not extra_baseline_paths,
        "grounded_db_context_source_is_exact_instrumented_source": grounded_files.get(
            expected_db_context_path, {}
        ).get("sha256")
        == instrumented_plugin_sha256,
        "grounded_db_context_manifest_is_exact_frozen_source": grounded_files.get(
            expected_db_context_manifest_path, {}
        ).get("sha256")
        == file_sha256(source_paths()["db_context_plugin_manifest_sha256"]),
    }
    return {
        "expected_delta": "remove exact plugins/leo-db-context/{__init__.py,plugin.yaml}",
        "extra_grounded_paths": sorted(extra_grounded_paths),
        "extra_baseline_paths": sorted(extra_baseline_paths),
        "instrumented_db_context_plugin_sha256": instrumented_plugin_sha256,
        "checks": checks,
        "pass": all(checks.values()),
    }


def _receipt_score(
    result: dict[str, Any],
    *,
    require_database_contract: bool,
    require_database_receipt: bool,
) -> dict[str, Any]:
    raw_traces = result.get("database_context_trace") or []
    traces = [item for item in raw_traces if isinstance(item, dict)] if isinstance(raw_traces, list) else []
    pre = [
        item
        for item in traces
        if item.get("event") == "pre_llm_call" and item.get("status") == "ok" and item.get("injected") is True
    ]
    post = [
        item
        for item in traces
        if item.get("event") == "post_llm_call" and item.get("status") == "ok" and item.get("validated") is True
    ]
    prompt_sha256 = hashlib.sha256(str(result.get("prompt") or "").encode()).hexdigest()
    pre_hashes = {item.get("query_sha256") for item in pre if item.get("query_sha256")}
    post_hashes = {item.get("query_sha256") for item in post if item.get("query_sha256")}
    contract_ids_are_lists = all(
        isinstance(item.get("contract_ids"), list)
        and all(isinstance(contract_id, str) and contract_id for contract_id in item["contract_ids"])
        for item in pre + post
    )
    contract_ids = {
        str(contract_id)
        for item in pre + post
        for contract_id in (item.get("contract_ids") if isinstance(item.get("contract_ids"), list) else [])
        if contract_id
    }
    raw_model_call_trace = result.get("model_call_trace") or []
    model_call_trace = (
        [item for item in raw_model_call_trace if isinstance(item, dict)]
        if isinstance(raw_model_call_trace, list)
        else []
    )
    retrieval_records = []
    for item in pre:
        receipt = item.get("retrieval_receipt") if isinstance(item.get("retrieval_receipt"), dict) else {}
        safe_receipt_payload = {
            key: value
            for key, value in receipt.items()
            if key not in {"receipt_sha256", "trace_payload_sha256"}
        }
        trace_payload_sha256 = hashlib.sha256(
            json.dumps(safe_receipt_payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest()
        consistency = receipt.get("read_consistency") if isinstance(receipt.get("read_consistency"), dict) else {}
        wal_before = consistency.get("wal_lsn_before")
        wal_after = consistency.get("wal_lsn_after")
        attempts = consistency.get("attempts")
        typed_attempts = isinstance(attempts, int) and not isinstance(attempts, bool) and attempts >= 1
        consistency_evidence = bool(
            wal_before
            and wal_after
            and typed_attempts
            and (
                (consistency.get("status") == "stable_wal_marker" and wal_before == wal_after)
                or (
                    consistency.get("status") == "stable_content_across_wal_change_retry"
                    and attempts >= 2
                )
            )
        )
        if (
            item.get("source") == "kb_tool.py --local context"
            and re.fullmatch(r"[0-9a-f]{64}", str(item.get("contract_sha256") or ""))
            and item.get("compiled_response_available") is not None
            and receipt.get("schema") == "livingip.teleoKbRetrievalReceipt.v1"
            and receipt.get("query_sha256") == item.get("query_sha256") == prompt_sha256
            and re.fullmatch(r"[0-9a-f]{64}", str(receipt.get("semantic_context_sha256") or ""))
            and re.fullmatch(r"[0-9a-f]{64}", str(receipt.get("artifact_state_sha256") or ""))
            and re.fullmatch(r"[0-9a-f]{64}", str(receipt.get("receipt_sha256") or ""))
            and receipt.get("trace_payload_sha256") == trace_payload_sha256
            and consistency.get("status")
            in {"stable_wal_marker", "stable_content_across_wal_change_retry"}
            and typed_attempts
            and consistency.get("database")
            and consistency.get("database_user")
            and consistency.get("system_identifier")
            and consistency_evidence
        ):
            retrieval_records.append(item)
    supported_identifiers = {
        str(identifier).lower()
        for item in retrieval_records
        for key in ("claim_ids", "source_ids")
        for identifier in ((item.get("retrieval_receipt") or {}).get(key) or [])
    }
    reply_identifiers = {match.group(0).lower() for match in UUID_RE.finditer(str(result.get("reply") or ""))}
    unsupported_identifiers = sorted(reply_identifiers - supported_identifiers)
    reply_sha256 = hashlib.sha256(str(result.get("reply") or "").encode()).hexdigest()
    checks = {
        "reply_present": result.get("ok") is True and bool(str(result.get("reply") or "").strip()),
        "read_only_turn": result.get("mutates_kb") is False,
        "trace_is_exact_typed_pair": isinstance(raw_traces, list)
        and len(traces) == len(raw_traces) == 2
        and len(pre) == len(post) == 1,
        "context_injected": len(pre) == 1,
        "response_validated": len(post) == 1,
        "contract_ids_are_typed_lists": contract_ids_are_lists,
        "context_response_query_hash_bound": pre_hashes == post_hashes == {prompt_sha256},
        "delivered_response_hash_bound": len(post) == 1
        and post[0].get("delivered_response_sha256") == reply_sha256,
        "database_contract_present": bool(contract_ids - NON_DB_CONTRACT_IDS) if require_database_contract else True,
        "database_retrieval_receipt_present": bool(retrieval_records) if require_database_receipt else True,
        "model_call_receipt_present": bool(model_call_trace)
        and any(item.get("event") == "post_api_request" and item.get("model") and item.get("provider") for item in model_call_trace),
        "conversation_history_prefix_preserved": result.get("conversation_history_prefix_preserved") is True,
        "no_unsupported_exact_identifiers": not unsupported_identifiers,
    }
    return {
        "checks": checks,
        "contract_ids": sorted(contract_ids),
        "query_sha256": sorted(pre_hashes & post_hashes),
        "expected_prompt_sha256": prompt_sha256,
        "database_tool_trace": result.get("database_tool_trace") or {},
        "reply_identifiers": sorted(reply_identifiers),
        "supported_identifiers": sorted(supported_identifiers),
        "unsupported_identifiers": unsupported_identifiers,
        "pass": all(checks.values()),
    }


def _benchmark_execution_chain(report: dict[str, Any]) -> dict[str, Any]:
    """Validate the generic turn manifests under this benchmark's declared ablation.

    The generic manifest intentionally marks a dirty harness and missing DB-context
    hooks incomplete.  This benchmark permits exactly one control-owned dirty file
    (``goal.md``) and, in the ablated arm only, the bindings made impossible by
    removing the DB-context plugin.  Every other runtime/model/session/safety
    binding remains mandatory and is checked independently here.
    """

    mode = report.get("grounding_mode")
    allowed_missing = (
        GROUNDED_EXECUTION_ALLOWED_MISSING
        if mode == "grounded"
        else ABLATION_EXECUTION_ALLOWED_MISSING
        if mode == "db_tool_ablated"
        else frozenset()
    )
    results = [item for item in report.get("results") or [] if isinstance(item, dict)]
    summary = report.get("execution_manifest_summary") or {}
    executed_behavior = report.get("executed_behavior_manifest") or {}
    local_state = report.get("oos_harness_git_state") or {}
    summary_source = summary.get("harness_source") or {}
    local_state_checks = {
        "git_head_valid": _valid_git_revision(local_state.get("git_head")),
        "status_sha256_valid": _valid_sha256(local_state.get("status_sha256")),
        "recorded_dirty": local_state.get("worktree_clean") is False,
        "only_control_goal_untracked": local_state.get("only_control_goal_untracked") is True
        and local_state.get("status_lines") == ["?? goal.md"],
        "generic_summary_source_bound": summary_source.get("git_head") == local_state.get("git_head")
        and summary_source.get("status_sha256") == local_state.get("status_sha256")
        and summary_source.get("worktree_clean") is False,
    }
    turn_checks: dict[str, dict[str, bool]] = {}
    previous_execution_sha256: str | None = None
    for index, result in enumerate(results):
        prompt_id = str(result.get("prompt_id") or f"turn-{index + 1}")
        manifest = result.get("execution_manifest") if isinstance(result.get("execution_manifest"), dict) else {}
        turn = manifest.get("turn") or {}
        runtime = manifest.get("runtime") or {}
        model = manifest.get("model_execution") or {}
        session = manifest.get("session_boundary") or {}
        conversation = session.get("conversation") or {}
        database = manifest.get("canonical_database") or {}
        context = database.get("context_binding") or {}
        tool_binding = database.get("database_tool_binding") or {}
        delivery = manifest.get("delivery_and_safety") or {}
        suite_safety = delivery.get("suite_safety") or {}
        attribution = manifest.get("attribution") or {}
        missing = attribution.get("missing_required_bindings")
        missing_set = set(missing) if isinstance(missing, list) and all(isinstance(item, str) for item in missing) else set()
        hermes_runtime = runtime.get("hermes_runtime") or {}
        teleo_runtime = runtime.get("teleo_infrastructure_runtime") or {}
        calls = model.get("calls") if isinstance(model.get("calls"), list) else []
        context_receipts = (
            database.get("context_retrieval_receipts")
            if isinstance(database.get("context_retrieval_receipts"), list)
            else []
        )
        checks = {
            "generic_manifest_valid": bool(manifest)
            and not execution_manifest_lib.validate_turn_manifest(manifest),
            "prompt_bound": turn.get("prompt_id") == result.get("prompt_id")
            and turn.get("prompt_sha256") == hashlib.sha256(str(result.get("prompt") or "").encode()).hexdigest(),
            "reply_bound": turn.get("reply_sha256")
            == hashlib.sha256(str(result.get("reply") or "").encode()).hexdigest(),
            "declared_missing_exact": missing_set == allowed_missing
            and attribution.get("status") == ("incomplete" if allowed_missing else "complete"),
            "chain_bound": conversation.get("previous_execution_sha256") == previous_execution_sha256,
            "session_bound": _valid_sha256(session.get("session_key_sha256"))
            and session.get("source_platform") == "telegram"
            and session.get("fresh_temp_profile_for_suite") is True
            and session.get("prior_dynamic_state_excluded_from_suite") is True
            and conversation.get("history_prefix_preserved") is True
            and conversation.get("conversation_hashes_valid") is True
            and conversation.get("prior_turn_state_bound") is True,
            "runtime_bound": _valid_sha256(runtime.get("behavior_sha256"))
            and runtime.get("behavior_sha256") == executed_behavior.get("behavior_sha256")
            and _valid_git_revision(hermes_runtime.get("git_head"))
            and _valid_sha256((hermes_runtime.get("source_tree") or {}).get("sha256"))
            and hermes_runtime == executed_behavior.get("hermes_runtime")
            and _valid_git_revision(teleo_runtime.get("git_head"))
            and _valid_sha256((teleo_runtime.get("source_tree") or {}).get("sha256"))
            and teleo_runtime == executed_behavior.get("teleo_infrastructure_runtime")
            and runtime.get("harness_source") == summary_source,
            "model_bound": isinstance(model.get("call_count"), int)
            and not isinstance(model.get("call_count"), bool)
            and model.get("call_count", 0) > 0
            and len(calls) == model.get("call_count")
            and model.get("prompt_bound") is True
            and model.get("delivered_response_bound") is True
            and model.get("response_trace_count_matches_api_calls") is True
            and model.get("api_call_sequence_valid") is True
            and model.get("session_binding_valid") is True
            and model.get("response_hashes_valid") is True
            and (model.get("raw_response_bound") is (mode == "grounded")),
            "database_state_bound": _valid_sha256((database.get("fingerprint_before") or {}).get("fingerprint_sha256"))
            and (database.get("fingerprint_before") or {}).get("fingerprint_sha256")
            == (database.get("fingerprint_after") or {}).get("fingerprint_sha256")
            and database.get("fingerprint_unchanged") is True
            and _valid_sha256(database.get("suite_counts_before_sha256"))
            and database.get("suite_counts_before_sha256") == database.get("suite_counts_after_sha256")
            and database.get("suite_counts_changed") is False,
            "database_mode_bound": (
                len(context_receipts) == 1
                and database.get("binding_status") == "retrieval_receipt_bound"
                and context.get("query_bound") is True
                and context.get("context_available") is True
                and context.get("response_bound") is True
            )
            if mode == "grounded"
            else (
                not context_receipts
                and database.get("binding_status") == "missing"
                and context.get("query_bound") is False
                and context.get("context_available") is False
                and context.get("response_bound") is False
            ),
            "database_tools_read_only": tool_binding.get("all_calls_read_only") is True,
            "delivery_safe": delivery.get("posted_to_telegram") is False
            and delivery.get("kb_mutation_by_harness") is False
            and delivery.get("turn_mutates_kb") is False
            and suite_safety.get("remote_returncode") == 0
            and suite_safety.get("pass_runtime") is True
            and suite_safety.get("live_behavior_manifest_unchanged") is True
            and suite_safety.get("temp_profile_removed") is True
            and suite_safety.get("service_unchanged") is True
            and suite_safety.get("db_fingerprint_unchanged") is True
            and suite_safety.get("model_call_trace_all_bound") is True,
        }
        turn_checks[prompt_id] = checks
        previous_execution_sha256 = manifest.get("execution_sha256")
    checks = {
        "recognized_grounding_mode": mode in {"grounded", "db_tool_ablated"},
        "results_nonempty": bool(results),
        "summary_turn_count_exact": summary.get("turn_count") == len(results),
        "one_manifest_per_result": bool(results)
        and all(isinstance(item.get("execution_manifest"), dict) for item in results),
        "local_harness_state_bound": all(local_state_checks.values()),
        "all_turns_valid_under_declared_mode": bool(turn_checks)
        and all(all(item.values()) for item in turn_checks.values()),
    }
    return {
        "mode": mode,
        "allowed_missing_bindings": sorted(allowed_missing),
        "local_state_checks": local_state_checks,
        "turn_checks": turn_checks,
        "checks": checks,
        "pass": all(checks.values()),
    }


def _top_level_safety(report: dict[str, Any], *, require_handler_safety_gate: bool) -> dict[str, Any]:
    before = report.get("db_fingerprint_before") or {}
    after = report.get("db_fingerprint_after") or {}
    service = report.get("service_before_after") or {}
    benchmark_execution = _benchmark_execution_chain(report)
    tool_surface = report.get("read_only_tool_surface") or {}
    handler_safety = report.get("safety_gate") or {}
    orphan_readback = report.get("post_run_orphan_readback") or {}
    leakage_scan = report.get("prompt_leakage_scan") or {}
    remote_leakage_scan = report.get("remote_temp_profile_prompt_leakage_scan") or {}
    transport_deny = report.get("telegram_transport_deny") or {}
    result_rows = [item for item in report.get("results") or [] if isinstance(item, dict)]
    handler_failed = set(handler_safety.get("failed_checks") or [])
    handler_checks = handler_safety.get("checks") if isinstance(handler_safety.get("checks"), dict) else {}
    handler_gate_acceptable = handler_safety.get("status") == "pass" or bool(
        handler_failed == {"all_turn_manifests_complete"}
        and handler_checks
        and all(value is True for key, value in handler_checks.items() if key != "all_turn_manifests_complete")
    )
    checks = {
        "fresh_temporary_session": (report.get("temp_profile_seed") or {}).get(
            "same_session_continuity_starts_fresh"
        )
        is True
        and bool(result_rows)
        and (result_rows[0].get("conversation_before") or {}).get("message_count") == 0,
        "remote_returncode_zero": report.get("remote_returncode") == 0,
        "runtime_passed": report.get("pass_runtime") is True,
        "no_telegram_post": report.get("posted_to_telegram") is False,
        "telegram_transport_deny_enabled": transport_deny.get("enabled") is True,
        "zero_telegram_transport_attempts": isinstance(transport_deny.get("attempt_count"), int)
        and not isinstance(transport_deny.get("attempt_count"), bool)
        and transport_deny.get("attempt_count") == 0,
        "telegram_send_method_patched": "send" in (transport_deny.get("patched_methods") or []),
        "telegram_outbound_methods_exactly_denied": set(transport_deny.get("patched_methods") or [])
        == EXPECTED_TELEGRAM_DENY_METHODS
        and set(transport_deny.get("expected_methods") or []) == EXPECTED_TELEGRAM_DENY_METHODS,
        "runner_adapters_empty": transport_deny.get("runner_adapters_empty") is True,
        "harness_declared_no_kb_mutation": report.get("mutates_kb_by_harness") is False,
        "database_counts_unchanged": report.get("db_counts_changed") is False,
        "database_fingerprint_before_ok": before.get("status") == "ok",
        "database_fingerprint_after_ok": after.get("status") == "ok",
        "database_fingerprint_unchanged": report.get("db_fingerprint_unchanged") is True,
        "database_fingerprint_hash_equal": bool(
            before.get("fingerprint_sha256")
            and before.get("fingerprint_sha256") == after.get("fingerprint_sha256")
        ),
        "live_behavior_manifest_unchanged": report.get("live_behavior_manifest_unchanged") is True,
        "service_unchanged_during_trial": service.get("unchanged_from_preexisting_live_readback") is True,
        "temporary_profile_removed": report.get("temp_profile_removed") is True,
        "execution_chain_complete_under_declared_benchmark_mode": benchmark_execution["pass"],
        "tool_registry_exactly_allowlisted": tool_surface.get("actual_registry_tools")
        == ["skill_view", "skills_list", "terminal"],
        "send_message_tool_absent": tool_surface.get("send_message_tool_enabled") is False,
        "mutating_bridge_commands_not_exposed": tool_surface.get("mutating_bridge_commands_exposed") is False,
        "terminal_provider_credentials_not_forwarded": tool_surface.get("provider_credentials_forwarded_to_terminal")
        is False,
        "terminal_restricted_to_exact_wrapper": tool_surface.get("terminal_restricted_to_exact_wrapper") is True,
        "handler_safety_gate_passed_or_only_declared_manifest_gap": handler_gate_acceptable
        if require_handler_safety_gate
        else True,
        "no_orphan_processes": orphan_readback.get("no_matching_processes") is True,
        "prompt_leakage_scan_passed": leakage_scan.get("pass") is True,
        "remote_temp_profile_prompt_leakage_scan_passed": remote_leakage_scan.get("pass") is True
        and remote_leakage_scan.get("scope")
        == "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv"
        and isinstance(remote_leakage_scan.get("scanned_files"), int)
        and not isinstance(remote_leakage_scan.get("scanned_files"), bool)
        and remote_leakage_scan.get("scanned_files", 0) > 0
        and isinstance(remote_leakage_scan.get("scanned_bytes"), int)
        and not isinstance(remote_leakage_scan.get("scanned_bytes"), bool)
        and remote_leakage_scan.get("scanned_bytes", 0) > 0
        and remote_leakage_scan.get("errors") == []
        and {
            item.get("name")
            for item in remote_leakage_scan.get("expected_roots") or []
            if isinstance(item, dict) and item.get("exists") is True
        }
        == {"profile", "skills", "plugins", "bin"},
    }
    return {
        "checks": checks,
        "benchmark_execution_chain": benchmark_execution,
        "handler_safety_gate": {
            "required": require_handler_safety_gate,
            "acceptable": handler_gate_acceptable,
            "failed_checks": sorted(handler_failed),
        },
        "pass": all(checks.values()),
    }


def ablate_receipts(report: dict[str, Any]) -> dict[str, Any]:
    ablated = copy.deepcopy(report)
    for result in ablated.get("results") or []:
        result["database_context_trace"] = []
        result["database_tool_trace"] = {}
        result["model_call_trace"] = []
    ablated["db_fingerprint_before"] = {"status": "ablated"}
    ablated["db_fingerprint_after"] = {"status": "ablated"}
    ablated["db_fingerprint_unchanged"] = False
    ablated["turn_execution_manifests"] = []
    ablated["execution_manifest_summary"] = {"all_turns_attribution_complete": False}
    return ablated


def _prompt_binding(report: dict[str, Any], trial: dict[str, Any]) -> dict[str, Any]:
    expected = {item["id"]: item for item in trial["prompts"]}
    raw_results = report.get("results") or []
    result_rows = [item for item in raw_results if isinstance(item, dict)] if isinstance(raw_results, list) else []
    raw_ids = [str(item.get("prompt_id")) for item in result_rows]
    actual = {str(item.get("prompt_id")): item for item in result_rows}
    checks: dict[str, bool] = {
        "results_are_objects": isinstance(raw_results, list) and len(result_rows) == len(raw_results),
        "prompt_ids_exact": set(actual) == set(expected),
        "prompt_count_exact": len(result_rows) == len(actual) == len(expected),
        "prompt_ids_unique": len(raw_ids) == len(set(raw_ids)),
    }
    for prompt_id, prompt in expected.items():
        result = actual.get(prompt_id) or {}
        checks[f"prompt_text:{prompt_id}"] = result.get("prompt") == prompt["message"]
        checks[f"prompt_hash:{prompt_id}"] = hashlib.sha256(str(result.get("prompt") or "").encode()).hexdigest() == prompt[
            "message_sha256"
        ]
    return {"checks": checks, "pass": all(checks.values())}


def _valid_sha256(value: Any) -> bool:
    return bool(re.fullmatch(r"[0-9a-f]{64}", str(value or "")))


def _valid_git_revision(value: Any) -> bool:
    return bool(re.fullmatch(r"[0-9a-f]{40}", str(value or "")))


def _retained_path(value: Any) -> Path | None:
    if not isinstance(value, str) or not value:
        return None
    path = Path(value)
    if not path.is_absolute():
        path = Path(__file__).resolve().parents[1] / path
    return path


def _nonempty_integer_mapping(value: Any) -> bool:
    return bool(
        isinstance(value, dict)
        and value
        and all(isinstance(key, str) and isinstance(item, int) and not isinstance(item, bool) for key, item in value.items())
    )


def _parse_utc(value: Any) -> datetime | None:
    if not isinstance(value, str) or not value:
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return None
    return parsed.astimezone(timezone.utc)


def _validate_restart_probe_reference(
    reference: Any,
    *,
    report: dict[str, Any],
) -> dict[str, Any]:
    reference = reference if isinstance(reference, dict) else {}
    path = _retained_path(reference.get("path"))
    payload: dict[str, Any] = {}
    read_error: str | None = None
    actual_sha256: str | None = None
    if path is not None:
        try:
            raw = path.read_bytes()
            actual_sha256 = hashlib.sha256(raw).hexdigest()
            loaded = json.loads(raw)
            if isinstance(loaded, dict):
                payload = loaded
            else:
                read_error = "probe payload is not an object"
        except (OSError, json.JSONDecodeError) as exc:
            read_error = f"{type(exc).__name__}: {exc}"
    before_counts = payload.get("db_counts_before")
    after_counts = payload.get("db_counts_after")
    before_fingerprint = payload.get("db_fingerprint_before") or {}
    after_fingerprint = payload.get("db_fingerprint_after") or {}
    transport = payload.get("telegram_transport_deny") or {}
    results = payload.get("results")
    checks = {
        "reference_path_present": path is not None,
        "reference_sha256_valid": _valid_sha256(reference.get("sha256")),
        "artifact_loaded": read_error is None and bool(payload),
        "artifact_sha256_matches": actual_sha256 is not None and actual_sha256 == reference.get("sha256"),
        "protocol_id_bound": payload.get("protocol_id") == report.get("protocol_id"),
        "protocol_hash_bound": payload.get("protocol_hash_sha256") == report.get("protocol_hash_sha256"),
        "source_hashes_bound": payload.get("source_hashes") == report.get("source_hashes"),
        "remote_runtime_passed": payload.get("remote_returncode") == 0 and payload.get("pass_runtime") is True,
        "zero_prompt_probe": isinstance(results, list) and not results,
        "no_telegram_post": payload.get("posted_to_telegram") is False,
        "transport_deny_proven": transport.get("enabled") is True
        and isinstance(transport.get("attempt_count"), int)
        and not isinstance(transport.get("attempt_count"), bool)
        and transport.get("attempt_count") == 0
        and transport.get("runner_adapters_empty") is True,
        "transport_methods_exactly_denied": set(transport.get("patched_methods") or [])
        == EXPECTED_TELEGRAM_DENY_METHODS
        and set(transport.get("expected_methods") or []) == EXPECTED_TELEGRAM_DENY_METHODS,
        "database_counts_complete_and_equal": _nonempty_integer_mapping(before_counts)
        and before_counts == after_counts
        and payload.get("db_counts_changed") is False,
        "database_fingerprint_complete_and_equal": before_fingerprint.get("status") == "ok"
        and after_fingerprint.get("status") == "ok"
        and _valid_sha256(before_fingerprint.get("fingerprint_sha256"))
        and before_fingerprint.get("fingerprint_sha256") == after_fingerprint.get("fingerprint_sha256")
        and payload.get("db_fingerprint_unchanged") is True,
        "preexecution_safety_passed": (payload.get("preexecution_safety_gate") or {}).get("status") == "pass",
        "temporary_profile_removed": payload.get("temp_profile_removed") is True,
        "no_orphan_processes": (payload.get("post_run_orphan_readback") or {}).get("no_matching_processes") is True,
    }
    return {
        "path": reference.get("path") if isinstance(reference.get("path"), str) else None,
        "actual_sha256": actual_sha256,
        "read_error": read_error,
        "checks": checks,
        "pass": all(checks.values()),
        "payload": payload,
    }


def validate_restart_receipt(receipt: dict[str, Any] | None, report: dict[str, Any]) -> dict[str, Any]:
    receipt = receipt if isinstance(receipt, dict) else {}
    before = receipt.get("service_before") or {}
    after = receipt.get("service_after") or {}
    report_before = report.get("before_service") or {}
    deploy_before = receipt.get("deploy_before") or {}
    deploy_after = receipt.get("deploy_after") or {}
    counts_before = receipt.get("db_counts_before")
    counts_after = receipt.get("db_counts_after")
    fingerprint_before = receipt.get("db_fingerprint_before") or {}
    fingerprint_after = receipt.get("db_fingerprint_after") or {}
    before_probe = _validate_restart_probe_reference(receipt.get("before_probe"), report=report)
    after_probe = _validate_restart_probe_reference(receipt.get("after_probe"), report=report)
    before_probe_payload = before_probe["payload"]
    after_probe_payload = after_probe["payload"]
    receipt_time = _parse_utc(receipt.get("generated_at_utc"))
    report_time = _parse_utc(report.get("generated_at_utc"))
    before_probe_time = _parse_utc(before_probe_payload.get("generated_at_utc"))
    restart_started_time = _parse_utc(receipt.get("restart_started_at_utc"))
    restart_ended_time = _parse_utc(receipt.get("restart_ended_at_utc"))
    after_probe_time = _parse_utc(after_probe_payload.get("generated_at_utc"))
    chronology_seconds = (
        (report_time - receipt_time).total_seconds() if receipt_time is not None and report_time is not None else None
    )
    chronology_values = (
        before_probe_time,
        restart_started_time,
        restart_ended_time,
        after_probe_time,
        receipt_time,
        report_time,
    )
    deploy_revisions = [
        deploy_before.get("head"),
        deploy_before.get("stamp"),
        deploy_after.get("head"),
        deploy_after.get("stamp"),
    ]
    self_checks = receipt.get("checks") if isinstance(receipt.get("checks"), dict) else {}
    checks = {
        "receipt_schema": receipt.get("schema") == "livingip.leoGatewayRestartReceipt.v1",
        "protocol_id_bound": receipt.get("protocol_id") == report.get("protocol_id"),
        "protocol_hash_bound": receipt.get("protocol_hash_sha256") == report.get("protocol_hash_sha256"),
        "next_trial_bound": receipt.get("next_trial_id") == report.get("trial_id")
        and receipt.get("next_trial_prompt_set_sha256") == report.get("trial_prompt_set_sha256"),
        "chronology_bound": all(value is not None for value in chronology_values)
        and list(chronology_values) == sorted(chronology_values)
        and chronology_seconds is not None
        and 0 <= chronology_seconds <= 3600,
        "restart_command_succeeded": receipt.get("restart_returncode") == 0,
        "service_active_after": after.get("ActiveState") == "active" and after.get("SubState") == "running",
        "service_pid_changed": bool(before.get("MainPID") and before.get("MainPID") != after.get("MainPID")),
        "trial_observed_restarted_pid": bool(after.get("MainPID") and report_before.get("MainPID") == after.get("MainPID")),
        "service_start_identity_bound": bool(
            after.get("ExecMainStartTimestamp")
            and after.get("ExecMainStartTimestamp") == report_before.get("ExecMainStartTimestamp")
        ),
        "no_telegram_post": receipt.get("posted_to_telegram") is False,
        "database_counts_complete_and_equal": _nonempty_integer_mapping(counts_before)
        and counts_before == counts_after
        and receipt.get("db_counts_changed") is False,
        "database_fingerprint_complete_and_equal": fingerprint_before.get("status") == "ok"
        and fingerprint_after.get("status") == "ok"
        and _valid_sha256(fingerprint_before.get("fingerprint_sha256"))
        and fingerprint_before.get("fingerprint_sha256") == fingerprint_after.get("fingerprint_sha256")
        and receipt.get("db_fingerprint_unchanged") is True,
        "deploy_identity_complete_and_equal": deploy_before.get("returncode") == 0
        and deploy_after.get("returncode") == 0
        and all(_valid_git_revision(value) for value in deploy_revisions)
        and len(set(deploy_revisions)) == 1,
        "before_probe_artifact_valid": before_probe["pass"],
        "after_probe_artifact_valid": after_probe["pass"],
        "probe_service_binding": (before_probe_payload.get("service_before_after") or {}).get("after") == before
        and after_probe_payload.get("before_service") == after,
        "probe_count_binding": before_probe_payload.get("db_counts_after") == counts_before
        and after_probe_payload.get("db_counts_before") == counts_after,
        "probe_fingerprint_binding": before_probe_payload.get("db_fingerprint_after") == fingerprint_before
        and after_probe_payload.get("db_fingerprint_before") == fingerprint_after,
        "receipt_self_checks_complete": bool(self_checks) and all(value is True for value in self_checks.values()),
        "receipt_self_check_passed": receipt.get("pass") is True,
    }
    return {
        "checks": checks,
        "chronology_seconds": chronology_seconds,
        "before_probe_validation": {key: value for key, value in before_probe.items() if key != "payload"},
        "after_probe_validation": {key: value for key, value in after_probe.items() if key != "payload"},
        "pass": all(checks.values()),
    }


def score_live_trial(
    protocol: dict[str, Any],
    trial_id: str,
    report: dict[str, Any],
    *,
    baseline_report: dict[str, Any],
    restart_receipt: dict[str, Any] | None = None,
) -> dict[str, Any]:
    protocol_validation = validate_protocol(protocol, verify_source_hashes=True)
    trial = next((item for item in protocol.get("trials") or [] if item.get("trial_id") == trial_id), None)
    if trial is None:
        raise ValueError(f"unknown trial_id: {trial_id}")
    prompt_binding = _prompt_binding(report, trial)
    baseline_prompt_binding = _prompt_binding(baseline_report, trial)
    report_results = [item for item in report.get("results") or [] if isinstance(item, dict)]
    baseline_results = [item for item in baseline_report.get("results") or [] if isinstance(item, dict)]
    semantic = _score_semantic_results(report_results, trial)
    by_prompt = {item["id"]: item for item in trial["prompts"]}
    receipts: dict[str, Any] = {}
    subject_alignment: dict[str, bool] = {}
    for result in report_results:
        prompt_id = str(result.get("prompt_id") or "")
        prompt = by_prompt.get(prompt_id)
        if not prompt:
            continue
        receipts[prompt_id] = _receipt_score(
            result,
            require_database_contract=bool(prompt["requires_database_contract"]),
            require_database_receipt=bool(prompt["requires_database_receipt"]),
        )
        subject_alignment[prompt_id] = _subject_alignment(prompt, str(result.get("reply") or ""))
    semantic_by_prompt = {item["prompt_id"]: item for item in semantic["scores"]}
    report_by_prompt = {str(item.get("prompt_id")): item for item in report_results}
    prompt_scores: list[dict[str, Any]] = []
    for prompt in trial["prompts"]:
        prompt_id = prompt["id"]
        semantic_item = semantic_by_prompt.get(prompt_id) or {"pass": False}
        receipt_item = receipts.get(prompt_id) or {"pass": False, "checks": {}}
        result_item = report_by_prompt.get(prompt_id) or {}
        tool_evidence_hashes = _tool_evidence_hashes(
            result_item,
            expected_command_sha256=prompt.get("expected_tool_command_sha256"),
        )
        evidence_answer = _evidence_answer_score(
            prompt,
            result_item,
            semantic_pass=bool(semantic_item.get("pass")),
            subject_alignment=bool(subject_alignment.get(prompt_id)),
            grounded_tool_hashes=tool_evidence_hashes,
        )
        grounded_pass = bool(
            semantic_item.get("pass") and receipt_item.get("pass") and subject_alignment.get(prompt_id)
        )
        prompt_scores.append(
            {
                "prompt_id": prompt_id,
                "family_id": prompt["family_id"],
                "scorer_id": prompt["scorer_id"],
                "semantic_pass": bool(semantic_item.get("pass")),
                "subject_alignment": bool(subject_alignment.get(prompt_id)),
                "receipt_pass": bool(receipt_item.get("pass")),
                "grounded_pass": grounded_pass,
                "evidence_answer_pass": evidence_answer["pass"],
                "evidence_answer_score": evidence_answer,
                "semantic_score": semantic_item,
                "receipt_score": receipt_item,
                "reply_sha256": hashlib.sha256(
                    str((next((row for row in report_results if row.get("prompt_id") == prompt_id), {}) or {}).get("reply") or "").encode()
                ).hexdigest(),
            }
        )
    top_safety = _top_level_safety(report, require_handler_safety_gate=True)
    baseline_top_safety = _top_level_safety(baseline_report, require_handler_safety_gate=False)
    restart_validation = (
        validate_restart_receipt(restart_receipt, report)
        if trial["session_mode"] == "post_restart_clean_session"
        else {"pass": True, "checks": {"not_a_restart_trial": True}}
    )
    baseline_receipts = {
        str(result.get("prompt_id")): _receipt_score(
            result,
            require_database_contract=bool(by_prompt.get(str(result.get("prompt_id")), {}).get("requires_database_contract")),
            require_database_receipt=bool(
                by_prompt.get(str(result.get("prompt_id")), {}).get("requires_database_receipt")
            ),
        )
        for result in baseline_results
        if str(result.get("prompt_id")) in by_prompt
    }
    baseline_semantic = _score_semantic_results(baseline_results, trial)
    grounded_passes = sum(1 for item in prompt_scores if item["grounded_pass"])
    prompt_count = len(prompt_scores)
    baseline_semantic_by_prompt = {item["prompt_id"]: item for item in baseline_semantic["scores"]}
    baseline_by_prompt = {str(item.get("prompt_id")): item for item in baseline_results}
    baseline_subject_alignment = {
        str(result.get("prompt_id")): _subject_alignment(
            by_prompt[str(result.get("prompt_id"))], str(result.get("reply") or "")
        )
        for result in baseline_results
        if str(result.get("prompt_id")) in by_prompt
    }
    baseline_grounded_passes = sum(
        1
        for prompt in trial["prompts"]
        if baseline_semantic_by_prompt.get(prompt["id"], {}).get("pass")
        and baseline_subject_alignment.get(prompt["id"])
        and baseline_receipts.get(prompt["id"], {}).get("pass")
    )
    evidence_prompt_ids = [
        prompt["id"] for prompt in trial["prompts"] if prompt.get("requires_tool_evidence_token") is True
    ]
    grounded_evidence_by_prompt = {
        item["prompt_id"]: item["evidence_answer_score"]["grounded_tool_semantic_hashes"]
        for item in prompt_scores
    }
    baseline_evidence_scores = {
        prompt_id: _evidence_answer_score(
            by_prompt[prompt_id],
            baseline_by_prompt.get(prompt_id) or {},
            semantic_pass=bool(baseline_semantic_by_prompt.get(prompt_id, {}).get("pass")),
            subject_alignment=bool(baseline_subject_alignment.get(prompt_id)),
            grounded_tool_hashes=grounded_evidence_by_prompt.get(prompt_id) or [],
        )
        for prompt_id in evidence_prompt_ids
    }
    current_evidence_passes = sum(
        1
        for item in prompt_scores
        if item["prompt_id"] in evidence_prompt_ids and item["evidence_answer_pass"] is True
    )
    baseline_evidence_passes = sum(1 for item in baseline_evidence_scores.values() if item["pass"] is True)
    evidence_prompt_count = len(evidence_prompt_ids)
    current_evidence_rate = current_evidence_passes / evidence_prompt_count if evidence_prompt_count else 0.0
    baseline_evidence_rate = baseline_evidence_passes / evidence_prompt_count if evidence_prompt_count else 0.0
    evidence_delta = current_evidence_rate - baseline_evidence_rate
    baseline_no_db_checks = {
        "grounding_mode": baseline_report.get("grounding_mode") == "db_tool_ablated",
        "db_context_plugin_disabled": baseline_report.get("db_context_plugin_enabled") is False,
        "tool_surface_ablation_mode": (baseline_report.get("read_only_tool_surface") or {}).get("mode")
        == "no_db_ablation",
        "zero_database_context_receipts": all(
            not (item.get("database_context_trace") or []) for item in baseline_results
        ),
        "zero_successful_database_receipts": all(not item["pass"] for item in baseline_receipts.values()),
    }
    grounded_mode_checks = {
        "grounding_mode": report.get("grounding_mode") == "grounded",
        "db_context_plugin_enabled": report.get("db_context_plugin_enabled") is True,
        "tool_surface_read_only_mode": (report.get("read_only_tool_surface") or {}).get("mode") == "read_only_kb",
        "readonly_guard_bound_to_protocol": report.get("readonly_guard_source_sha256")
        == protocol["source_hashes"]["readonly_guard_sha256"],
    }
    critical_prompt_checks = {
        "all_receipt_gates_pass": all(item["receipt_pass"] for item in prompt_scores),
        "no_unsupported_exact_identifiers": all(
            not item["receipt_score"].get("unsupported_identifiers") for item in prompt_scores
        ),
    }
    def model_identities(value: dict[str, Any]) -> list[dict[str, Any]]:
        identities = {
            canonical_sha256(
                {
                    "model": item.get("model"),
                    "provider": item.get("provider"),
                    "base_url_sha256": item.get("base_url_sha256"),
                    "api_mode": item.get("api_mode"),
                }
            ): {
                "model": item.get("model"),
                "provider": item.get("provider"),
                "base_url_sha256": item.get("base_url_sha256"),
                "api_mode": item.get("api_mode"),
            }
            for result in value.get("results") or []
            if isinstance(result, dict)
            for item in result.get("model_call_trace") or []
            if item.get("event") == "post_api_request"
        }
        return [identities[key] for key in sorted(identities)]

    executed_behavior_ablation = _executed_behavior_ablation(report, baseline_report)
    comparison_axis_checks = {
        "protocol_id_equal": baseline_report.get("protocol_id") == report.get("protocol_id") == protocol["protocol_id"],
        "protocol_hash_equal": baseline_report.get("protocol_hash_sha256")
        == report.get("protocol_hash_sha256")
        == protocol["protocol_hash_sha256"],
        "prompt_set_hash_equal": baseline_report.get("trial_prompt_set_sha256")
        == report.get("trial_prompt_set_sha256")
        == trial["prompt_set_sha256"],
        "source_hashes_equal": baseline_report.get("source_hashes") == report.get("source_hashes") == protocol["source_hashes"],
        "live_behavior_manifest_equal": bool(
            (baseline_report.get("live_behavior_manifest_before") or {}).get("behavior_sha256")
            and (baseline_report.get("live_behavior_manifest_before") or {}).get("behavior_sha256")
            == (report.get("live_behavior_manifest_before") or {}).get("behavior_sha256")
        ),
        "executed_behavior_manifests_capture_only_declared_ablation": executed_behavior_ablation["pass"],
        "temporary_profile_seed_equal": baseline_report.get("temp_profile_seed") == report.get("temp_profile_seed"),
        "database_snapshot_equal_before_trials": (baseline_report.get("db_fingerprint_before") or {}).get(
            "fingerprint_sha256"
        )
        == (report.get("db_fingerprint_before") or {}).get("fingerprint_sha256"),
        "model_provider_identity_equal": model_identities(baseline_report) == model_identities(report)
        and bool(model_identities(report)),
        "readonly_guard_source_equal_and_frozen": baseline_report.get("readonly_guard_source_sha256")
        == report.get("readonly_guard_source_sha256")
        == protocol["source_hashes"]["readonly_guard_sha256"],
        "tool_schema_equal": (baseline_report.get("read_only_tool_surface") or {}).get("allowed_tools")
        == (report.get("read_only_tool_surface") or {}).get("allowed_tools"),
    }
    threshold = float(protocol["thresholds"]["minimum_trial_grounded_pass_rate"])
    evidence_threshold = float(protocol["thresholds"]["minimum_trial_evidence_answer_pass_rate"])
    evidence_delta_threshold = float(
        protocol["thresholds"]["minimum_current_minus_ablation_evidence_answer_delta"]
    )
    score: dict[str, Any] = {
        "schema": TRIAL_SCORE_SCHEMA,
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "protocol_id": protocol["protocol_id"],
        "protocol_hash_sha256": protocol["protocol_hash_sha256"],
        "source_hashes": protocol["source_hashes"],
        "trial_id": trial_id,
        "session_mode": trial["session_mode"],
        "source_report_path": report.get("source_report_path"),
        "protocol_validation": protocol_validation,
        "prompt_binding": prompt_binding,
        "baseline_prompt_binding": baseline_prompt_binding,
        "top_level_safety": top_safety,
        "baseline_top_level_safety": baseline_top_safety,
        "grounded_mode_checks": grounded_mode_checks,
        "critical_prompt_checks": critical_prompt_checks,
        "baseline_no_db_checks": baseline_no_db_checks,
        "comparison_axis_checks": comparison_axis_checks,
        "executed_behavior_ablation": executed_behavior_ablation,
        "restart_receipt_validation": restart_validation,
        "semantic_score": semantic,
        "prompt_scores": prompt_scores,
        "grounded_passes": grounded_passes,
        "prompt_count": prompt_count,
        "grounded_pass_rate": grounded_passes / prompt_count if prompt_count else 0.0,
        "evidence_answer_comparison": {
            "method": "both_arms_scored_against_grounded_model_visible_tool_receipt_hashes",
            "identical_replies_have_identical_evidence_answer_outcomes": True,
            "prompt_ids": evidence_prompt_ids,
            "prompt_count": evidence_prompt_count,
            "current_passes": current_evidence_passes,
            "current_pass_rate": current_evidence_rate,
            "ablation_passes": baseline_evidence_passes,
            "ablation_pass_rate": baseline_evidence_rate,
            "current_minus_ablation_delta": evidence_delta,
            "ablation_scores": baseline_evidence_scores,
        },
        "receipt_ablation": {
            "version": BASELINE_VERSION,
            "same_prompt_set_sha256": trial["prompt_set_sha256"],
            "semantic_passes": baseline_semantic["passes"],
            "semantic_pass_rate": baseline_semantic["passes"] / prompt_count if prompt_count else 0.0,
            "semantic_scores": baseline_semantic["scores"],
            "grounded_passes": baseline_grounded_passes,
            "grounded_pass_rate": baseline_grounded_passes / prompt_count if prompt_count else 0.0,
            "receipt_scores": baseline_receipts,
            "reply_sha256": {
                str(item.get("prompt_id")): hashlib.sha256(str(item.get("reply") or "").encode()).hexdigest()
                for item in baseline_results
            },
        },
        "grounded_report_payload_sha256": canonical_sha256(report),
        "baseline_report_payload_sha256": canonical_sha256(baseline_report),
    }
    score["pass"] = bool(
        protocol_validation["pass"]
        and prompt_binding["pass"]
        and baseline_prompt_binding["pass"]
        and top_safety["pass"]
        and baseline_top_safety["pass"]
        and all(grounded_mode_checks.values())
        and all(critical_prompt_checks.values())
        and all(baseline_no_db_checks.values())
        and all(comparison_axis_checks.values())
        and restart_validation["pass"]
        and score["grounded_pass_rate"] >= threshold
        and current_evidence_rate >= evidence_threshold
        and evidence_delta >= evidence_delta_threshold
        and score["receipt_ablation"]["grounded_passes"] == 0
    )
    score["derivation_core_sha256"] = canonical_sha256(score_derivation_core(score))
    return score


def validate_trial_score(protocol: dict[str, Any], trial: dict[str, Any], score: dict[str, Any]) -> dict[str, Any]:
    def mapping(value: Any) -> dict[str, Any]:
        return value if isinstance(value, dict) else {}

    def number(value: Any) -> float:
        try:
            return float(value)
        except (TypeError, ValueError):
            return float("nan")

    def retained_report_checks(prefix: str) -> tuple[dict[str, bool], dict[str, Any]]:
        path = _retained_path(score.get(f"{prefix}_report_path"))
        payload: dict[str, Any] = {}
        byte_sha256: str | None = None
        if path is not None:
            try:
                raw = path.read_bytes()
                byte_sha256 = hashlib.sha256(raw).hexdigest()
                loaded = json.loads(raw)
                if isinstance(loaded, dict):
                    payload = loaded
            except (OSError, json.JSONDecodeError):
                pass
        return {
            "path_present": path is not None,
            "payload_loaded": bool(payload),
            "byte_sha256_bound": _valid_sha256(score.get(f"{prefix}_report_sha256"))
            and byte_sha256 == score.get(f"{prefix}_report_sha256"),
            "canonical_payload_sha256_bound": bool(payload)
            and canonical_sha256(payload) == score.get(f"{prefix}_report_payload_sha256"),
            "protocol_id_bound": payload.get("protocol_id") == protocol.get("protocol_id"),
            "protocol_hash_bound": payload.get("protocol_hash_sha256") == protocol.get("protocol_hash_sha256"),
            "trial_id_bound": payload.get("trial_id") == trial.get("trial_id"),
            "prompt_set_bound": payload.get("trial_prompt_set_sha256") == trial.get("prompt_set_sha256"),
            "source_hashes_bound": payload.get("source_hashes") == protocol.get("source_hashes"),
        }, payload

    def retained_restart_receipt_checks() -> tuple[dict[str, bool], dict[str, Any]]:
        required = trial.get("session_mode") == "post_restart_clean_session"
        path = _retained_path(score.get("restart_receipt_path"))
        payload: dict[str, Any] = {}
        byte_sha256: str | None = None
        if path is not None:
            try:
                raw = path.read_bytes()
                byte_sha256 = hashlib.sha256(raw).hexdigest()
                loaded = json.loads(raw)
                if isinstance(loaded, dict):
                    payload = loaded
            except (OSError, json.JSONDecodeError):
                pass
        if not required:
            return {
                "not_required": True,
                "path_absent": score.get("restart_receipt_path") is None,
                "hashes_absent": score.get("restart_receipt_sha256") is None
                and score.get("restart_receipt_payload_sha256") is None,
            }, {}
        return {
            "required": True,
            "path_present": path is not None,
            "payload_loaded": bool(payload),
            "byte_sha256_bound": _valid_sha256(score.get("restart_receipt_sha256"))
            and byte_sha256 == score.get("restart_receipt_sha256"),
            "canonical_payload_sha256_bound": bool(payload)
            and canonical_sha256(payload) == score.get("restart_receipt_payload_sha256"),
            "protocol_id_bound": payload.get("protocol_id") == protocol.get("protocol_id"),
            "protocol_hash_bound": payload.get("protocol_hash_sha256") == protocol.get("protocol_hash_sha256"),
            "next_trial_id_bound": payload.get("next_trial_id") == trial.get("trial_id"),
            "next_prompt_set_bound": payload.get("next_trial_prompt_set_sha256")
            == trial.get("prompt_set_sha256"),
        }, payload

    expected_prompt_ids = [item["id"] for item in trial["prompts"]]
    prompt_scores = score.get("prompt_scores") if isinstance(score.get("prompt_scores"), list) else []
    actual_prompt_ids = [str(item.get("prompt_id")) for item in prompt_scores if isinstance(item, dict)]
    grounded_passes = sum(1 for item in prompt_scores if isinstance(item, dict) and item.get("grounded_pass") is True)
    current_semantic_passes = sum(
        1 for item in prompt_scores if isinstance(item, dict) and item.get("semantic_pass") is True
    )
    prompt_count = len(expected_prompt_ids)
    baseline = mapping(score.get("receipt_ablation"))
    baseline_receipts = baseline.get("receipt_scores") if isinstance(baseline.get("receipt_scores"), dict) else {}
    baseline_semantic_scores = (
        baseline.get("semantic_scores") if isinstance(baseline.get("semantic_scores"), list) else []
    )
    baseline_semantic_ids = [
        str(item.get("prompt_id")) for item in baseline_semantic_scores if isinstance(item, dict)
    ]
    baseline_semantic_passes = sum(
        1 for item in baseline_semantic_scores if isinstance(item, dict) and item.get("pass") is True
    )
    baseline_grounded_passes = int(baseline.get("grounded_passes") or 0)
    evidence = mapping(score.get("evidence_answer_comparison"))
    expected_evidence_ids = [
        item["id"] for item in trial["prompts"] if item.get("requires_tool_evidence_token") is True
    ]
    current_evidence_passes = sum(
        1
        for item in prompt_scores
        if isinstance(item, dict)
        and item.get("prompt_id") in expected_evidence_ids
        and item.get("evidence_answer_pass") is True
    )
    baseline_evidence_scores = mapping(evidence.get("ablation_scores"))
    baseline_evidence_passes = sum(
        1 for item in baseline_evidence_scores.values() if isinstance(item, dict) and item.get("pass") is True
    )
    evidence_count = len(expected_evidence_ids)
    current_evidence_rate = current_evidence_passes / evidence_count if evidence_count else 0.0
    baseline_evidence_rate = baseline_evidence_passes / evidence_count if evidence_count else 0.0
    grounded_artifact_checks, grounded_payload = retained_report_checks("grounded")
    baseline_artifact_checks, baseline_payload = retained_report_checks("baseline")
    restart_artifact_checks, restart_payload = retained_restart_receipt_checks()
    recomputed: dict[str, Any] = {}
    if (
        grounded_payload
        and baseline_payload
        and (trial.get("session_mode") != "post_restart_clean_session" or restart_payload)
    ):
        try:
            recomputed = score_live_trial(
                protocol,
                str(trial.get("trial_id") or ""),
                grounded_payload,
                baseline_report=baseline_payload,
                restart_receipt=restart_payload or None,
            )
        except (KeyError, TypeError, ValueError):
            recomputed = {}
    stored_core = score_derivation_core(score)
    recomputed_core = score_derivation_core(recomputed) if recomputed else {}
    checks = {
        "schema": score.get("schema") == TRIAL_SCORE_SCHEMA,
        "protocol_id": score.get("protocol_id") == protocol.get("protocol_id"),
        "protocol_hash": score.get("protocol_hash_sha256") == protocol.get("protocol_hash_sha256"),
        "source_hashes": score.get("source_hashes") == protocol.get("source_hashes"),
        "trial_id": score.get("trial_id") == trial.get("trial_id"),
        "session_mode": score.get("session_mode") == trial.get("session_mode"),
        "prompt_ids_exact_and_ordered": actual_prompt_ids == expected_prompt_ids,
        "prompt_count": score.get("prompt_count") == prompt_count == len(prompt_scores),
        "grounded_passes_recomputed": score.get("grounded_passes") == grounded_passes,
        "grounded_rate_recomputed": abs(
            number(score.get("grounded_pass_rate")) - (grounded_passes / prompt_count if prompt_count else 0.0)
        )
        < 1e-12,
        "current_semantic_score_recomputed": mapping(score.get("semantic_score")).get("passes")
        == current_semantic_passes,
        "baseline_version": baseline.get("version") == protocol.get("baseline", {}).get("version"),
        "baseline_receipt_ids_exact": set(baseline_receipts) == set(expected_prompt_ids),
        "baseline_semantic_ids_exact_and_ordered": baseline_semantic_ids == expected_prompt_ids,
        "baseline_semantic_passes_recomputed": baseline.get("semantic_passes") == baseline_semantic_passes,
        "baseline_semantic_rate_recomputed": abs(
            number(baseline.get("semantic_pass_rate"))
            - (baseline_semantic_passes / prompt_count if prompt_count else 0.0)
        )
        < 1e-12,
        "baseline_grounded_rate_recomputed": abs(
            number(baseline.get("grounded_pass_rate"))
            - (baseline_grounded_passes / prompt_count if prompt_count else 0.0)
        )
        < 1e-12,
        "evidence_method_is_non_tautological": evidence.get("method")
        == "both_arms_scored_against_grounded_model_visible_tool_receipt_hashes"
        and evidence.get("identical_replies_have_identical_evidence_answer_outcomes") is True,
        "evidence_prompt_ids_exact": evidence.get("prompt_ids") == expected_evidence_ids
        and set(baseline_evidence_scores) == set(expected_evidence_ids),
        "evidence_prompt_count": evidence.get("prompt_count") == evidence_count,
        "current_evidence_passes_recomputed": evidence.get("current_passes") == current_evidence_passes,
        "baseline_evidence_passes_recomputed": evidence.get("ablation_passes") == baseline_evidence_passes,
        "current_evidence_rate_recomputed": abs(
            number(evidence.get("current_pass_rate")) - current_evidence_rate
        )
        < 1e-12,
        "baseline_evidence_rate_recomputed": abs(
            number(evidence.get("ablation_pass_rate")) - baseline_evidence_rate
        )
        < 1e-12,
        "evidence_delta_recomputed": abs(
            number(evidence.get("current_minus_ablation_delta"))
            - (current_evidence_rate - baseline_evidence_rate)
        )
        < 1e-12,
        "grounded_report_artifact_bound": bool(grounded_artifact_checks)
        and all(grounded_artifact_checks.values()),
        "baseline_report_artifact_bound": bool(baseline_artifact_checks)
        and all(baseline_artifact_checks.values()),
        "restart_receipt_artifact_bound": bool(restart_artifact_checks)
        and all(restart_artifact_checks.values()),
        "derivation_core_hash_bound": _valid_sha256(score.get("derivation_core_sha256"))
        and score.get("derivation_core_sha256") == canonical_sha256(stored_core),
        "score_recomputed_from_retained_artifacts": bool(recomputed)
        and stored_core == recomputed_core
        and score.get("derivation_core_sha256") == recomputed.get("derivation_core_sha256"),
        "protocol_validation_passed": mapping(score.get("protocol_validation")).get("pass") is True,
        "prompt_binding_passed": mapping(score.get("prompt_binding")).get("pass") is True,
        "baseline_prompt_binding_passed": mapping(score.get("baseline_prompt_binding")).get("pass") is True,
        "top_level_safety_passed": mapping(score.get("top_level_safety")).get("pass") is True,
        "baseline_top_level_safety_passed": mapping(score.get("baseline_top_level_safety")).get("pass") is True,
        "grounded_mode_checks_passed": bool(mapping(score.get("grounded_mode_checks")))
        and all(value is True for value in mapping(score.get("grounded_mode_checks")).values()),
        "baseline_no_db_checks_passed": bool(mapping(score.get("baseline_no_db_checks")))
        and all(value is True for value in mapping(score.get("baseline_no_db_checks")).values()),
        "comparison_axis_checks_passed": bool(mapping(score.get("comparison_axis_checks")))
        and all(value is True for value in mapping(score.get("comparison_axis_checks")).values()),
        "critical_prompt_checks_passed": bool(mapping(score.get("critical_prompt_checks")))
        and all(value is True for value in mapping(score.get("critical_prompt_checks")).values()),
        "restart_receipt_validation_passed": mapping(score.get("restart_receipt_validation")).get("pass") is True,
        "score_passed": score.get("pass") is True,
    }
    return {
        "pass": all(checks.values()),
        "checks": checks,
        "grounded_report_artifact_checks": grounded_artifact_checks,
        "baseline_report_artifact_checks": baseline_artifact_checks,
        "restart_receipt_artifact_checks": restart_artifact_checks,
        "failed_checks": [k for k, v in checks.items() if not v],
    }


def aggregate_trial_scores(protocol: dict[str, Any], trial_scores: list[dict[str, Any]]) -> dict[str, Any]:
    expected_ids = [item["trial_id"] for item in protocol.get("trials") or []]
    by_id = {str(item.get("trial_id")): item for item in trial_scores}
    trial_by_id = {item["trial_id"]: item for item in protocol.get("trials") or []}
    integrity = {
        trial_id: validate_trial_score(protocol, trial_by_id[trial_id], by_id.get(trial_id, {}))
        for trial_id in expected_ids
    }
    current_rates = [float(by_id[trial_id].get("grounded_pass_rate") or 0.0) for trial_id in expected_ids if trial_id in by_id]
    baseline_rates = [
        float((by_id[trial_id].get("receipt_ablation") or {}).get("grounded_pass_rate") or 0.0)
        for trial_id in expected_ids
        if trial_id in by_id
    ]
    evidence_current_rates = [
        float((by_id[trial_id].get("evidence_answer_comparison") or {}).get("current_pass_rate") or 0.0)
        for trial_id in expected_ids
        if trial_id in by_id
    ]
    evidence_baseline_rates = [
        float((by_id[trial_id].get("evidence_answer_comparison") or {}).get("ablation_pass_rate") or 0.0)
        for trial_id in expected_ids
        if trial_id in by_id
    ]
    semantic_current_rates = [
        float((by_id[trial_id].get("semantic_score") or {}).get("passes") or 0)
        / max(1, int(by_id[trial_id].get("prompt_count") or 0))
        for trial_id in expected_ids
        if trial_id in by_id
    ]
    semantic_baseline_rates = [
        float((by_id[trial_id].get("receipt_ablation") or {}).get("semantic_pass_rate") or 0.0)
        for trial_id in expected_ids
        if trial_id in by_id
    ]
    thresholds = protocol["thresholds"]
    mean_current = statistics.fmean(current_rates) if current_rates else 0.0
    mean_baseline = statistics.fmean(baseline_rates) if baseline_rates else 0.0
    stddev = statistics.pstdev(current_rates) if len(current_rates) > 1 else 0.0
    mean_evidence_current = statistics.fmean(evidence_current_rates) if evidence_current_rates else 0.0
    mean_evidence_baseline = statistics.fmean(evidence_baseline_rates) if evidence_baseline_rates else 0.0
    evidence_stddev = statistics.pstdev(evidence_current_rates) if len(evidence_current_rates) > 1 else 0.0
    mean_semantic_current = statistics.fmean(semantic_current_rates) if semantic_current_rates else 0.0
    mean_semantic_baseline = statistics.fmean(semantic_baseline_rates) if semantic_baseline_rates else 0.0
    checks = {
        "all_trials_present": len(trial_scores) == len(expected_ids) and set(by_id) == set(expected_ids),
        "all_trial_scores_integrity_valid": all(item["pass"] for item in integrity.values()),
        "all_trial_scores_pass": all(by_id.get(trial_id, {}).get("pass") is True for trial_id in expected_ids),
        "minimum_trial_rate": bool(current_rates)
        and min(current_rates) >= float(thresholds["minimum_trial_grounded_pass_rate"]),
        "minimum_mean_rate": mean_current >= float(thresholds["minimum_mean_grounded_pass_rate"]),
        "maximum_population_stddev": stddev <= float(thresholds["maximum_grounded_pass_rate_population_stddev"]),
        "minimum_trial_evidence_answer_rate": bool(evidence_current_rates)
        and min(evidence_current_rates) >= float(thresholds["minimum_trial_evidence_answer_pass_rate"]),
        "minimum_mean_evidence_answer_rate": mean_evidence_current
        >= float(thresholds["minimum_mean_evidence_answer_pass_rate"]),
        "maximum_evidence_answer_population_stddev": evidence_stddev
        <= float(thresholds["maximum_evidence_answer_pass_rate_population_stddev"]),
        "minimum_non_tautological_evidence_ablation_delta": (mean_evidence_current - mean_evidence_baseline)
        >= float(thresholds["minimum_current_minus_ablation_evidence_answer_delta"]),
        "baseline_rejects_all_ungrounded_receipts": all(rate == 0.0 for rate in baseline_rates),
        "broad_semantic_comparison_reported": len(semantic_current_rates)
        == len(semantic_baseline_rates)
        == len(expected_ids),
        "restart_trial_passed": all(
            by_id.get(item["trial_id"], {}).get("restart_receipt_validation", {}).get("pass") is True
            for item in protocol["trials"]
            if item["session_mode"] == "post_restart_clean_session"
        ),
    }
    aggregate = {
        "schema": AGGREGATE_SCHEMA,
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "protocol_id": protocol["protocol_id"],
        "protocol_hash_sha256": protocol["protocol_hash_sha256"],
        "scorer_version": protocol["scorer_version"],
        "baseline_version": protocol["baseline"]["version"],
        "trial_ids": expected_ids,
        "current_grounded_pass_rates": current_rates,
        "ablation_grounded_pass_rates": baseline_rates,
        "mean_current_grounded_pass_rate": mean_current,
        "mean_ablation_grounded_pass_rate": mean_baseline,
        "current_minus_ablation_delta": mean_current - mean_baseline,
        "current_grounded_pass_rate_population_stddev": stddev,
        "minimum_current_grounded_pass_rate": min(current_rates) if current_rates else 0.0,
        "current_evidence_answer_pass_rates": evidence_current_rates,
        "ablation_evidence_answer_pass_rates": evidence_baseline_rates,
        "mean_current_evidence_answer_pass_rate": mean_evidence_current,
        "mean_ablation_evidence_answer_pass_rate": mean_evidence_baseline,
        "current_minus_ablation_evidence_answer_delta": mean_evidence_current - mean_evidence_baseline,
        "current_evidence_answer_pass_rate_population_stddev": evidence_stddev,
        "minimum_current_evidence_answer_pass_rate": min(evidence_current_rates)
        if evidence_current_rates
        else 0.0,
        "current_semantic_pass_rates": semantic_current_rates,
        "ablation_semantic_pass_rates": semantic_baseline_rates,
        "mean_current_semantic_pass_rate": mean_semantic_current,
        "mean_ablation_semantic_pass_rate": mean_semantic_baseline,
        "current_minus_ablation_semantic_delta": mean_semantic_current - mean_semantic_baseline,
        "checks": checks,
        "trial_score_integrity": integrity,
        "trial_scores": trial_scores,
        "required_tier": "T3_live_readonly",
        "claim_ceiling": (
            "Repeated live VPS GatewayRunner reasoning with frozen blinded families and a current-build no-DB "
            "ablation. Both arms are scored against the grounded arm's model-visible receipt tokens, so identical "
            "answers cannot create the gated evidence delta. Broad semantic rates for both arms are reported "
            "separately and are descriptive rather than silently inferred from missing receipts. Complete "
            "tool/execution receipts and unchanged canonical database fingerprints are retained. No Telegram "
            "delivery or production apply is proven."
        ),
    }
    aggregate["pass"] = all(checks.values())
    return aggregate


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    mode = parser.add_mutually_exclusive_group(required=True)
    mode.add_argument("--freeze-protocol", type=Path)
    mode.add_argument("--validate-protocol", type=Path)
    mode.add_argument("--aggregate-protocol", type=Path)
    parser.add_argument("--seed")
    parser.add_argument("--trial-count", type=int, default=DEFAULT_TRIAL_COUNT)
    parser.add_argument("--trial-score", type=Path, action="append", default=[])
    parser.add_argument("--out", type=Path)
    args = parser.parse_args()

    if args.freeze_protocol:
        if not args.seed:
            raise SystemExit("--seed is required with --freeze-protocol")
        protocol = freeze_protocol(args.seed, trial_count=args.trial_count)
        args.freeze_protocol.parent.mkdir(parents=True, exist_ok=True)
        args.freeze_protocol.write_text(json.dumps(protocol, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        print(json.dumps({"protocol": str(args.freeze_protocol), "hash": protocol["protocol_hash_sha256"]}, indent=2))
        return 0

    protocol_path = args.validate_protocol or args.aggregate_protocol
    protocol = json.loads(protocol_path.read_text(encoding="utf-8"))
    validation = validate_protocol(protocol, verify_source_hashes=True)
    if args.validate_protocol:
        print(json.dumps(validation, indent=2, sort_keys=True))
        return 0 if validation["pass"] else 1

    scores = [json.loads(path.read_text(encoding="utf-8")) for path in args.trial_score]
    aggregate = aggregate_trial_scores(protocol, scores)
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(json.dumps(aggregate, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(json.dumps(aggregate, indent=2, sort_keys=True))
    return 0 if aggregate["pass"] and validation["pass"] else 1


if __name__ == "__main__":
    raise SystemExit(main())