#!/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 import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any import leo_tool_trace as tool_trace_lib import leo_turn_execution_manifest as execution_manifest_lib import working_leo_m3taversal_oos_benchmark as benchmark ROOT = Path(__file__).resolve().parents[1] PROTOCOL_SCHEMA = "livingip.leoM3taversalOosProtocol.v3" TRIAL_SCORE_SCHEMA = "livingip.leoM3taversalOosTrialScore.v3" AGGREGATE_SCHEMA = "livingip.leoM3taversalOosAggregate.v3" GENERATOR_VERSION = "blinded-family-generator-v5" SCORER_VERSION = "invariant-reasoning-live-receipts-and-factual-ablation-v7" BASELINE_VERSION = "live-current-build-db-tool-ablation-v3" 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( {"autonomous_retrieval_reasoning", "mixed_composition", "receipt_discrimination"} ) AUTONOMOUS_RETRIEVAL_FAMILIES = frozenset({"autonomous_retrieval_reasoning"}) 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() CONTROL_GOAL_EXECUTION_ALLOWED_MISSING = frozenset({"harness_worktree_clean"}) ABLATION_EXECUTION_ALLOWED_MISSING = frozenset( { "model_raw_response_binding", "database_context_query_binding", "database_context_available", "database_context_response_binding", "database_retrieval_receipt", } ) # The generic manifest requires a tool result only when a prompt is classified # as database-relevant. Under the declared no-DB ablation that conditional # result may therefore be absent on some turns and inapplicable on others. ABLATION_EXECUTION_OPTIONAL_MISSING = frozenset({"database_tool_results"}) 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) CLAIM_ID_CITATION_RE = re.compile( r"\bclaim(?:\s+row)?(?:\s+id)?(?:\s*(?:[:#=]|\bis\b))?\s*`?" r"([0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}|[0-9a-f]{8,12})\b", re.I, ) SOURCE_ID_CITATION_RE = re.compile( r"\b(?:source|evidence)(?:\s+row)?(?:\s+id)?(?:\s*(?:[:#=]|\bis\b))?\s*`?" r"([0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}|[0-9a-f]{8,12})\b", 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. Stay under 180 words and " "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": "autonomous_retrieval_reasoning", "scorer_id": "OOS-16", "dimension": "broad_id_free_live_claim_evidence_challenge_revision", "subjects": ( "market structure evolution", "product adoption dynamics", "organizational learning loops", ), "subject_anchors": ("claim body", "evidence", "challenge", "revision", "source"), "expected_follow_up": "ground a narrower revision in a live claim and its linked evidence", "variants": ( ( "Investigate {subject} from the live canonical database without asking me for identifiers or a " "command. Choose one relevant supported claim. Cite its claim ID and a linked source or evidence ID; " "quote the exact claim body separately from the evidence; challenge one inference the evidence does " "not establish; then propose a narrower revision. Stay read-only and do not stage or apply." ), ( "Reason about {subject} using an autonomous read-only lookup. I am supplying no row IDs and no " "terminal syntax. Return one live claim ID plus one linked source or evidence ID, distinguish the " "claim's exact body from what the evidence actually says, identify one unsupported leap, and give a " "narrower revision. Make no database changes." ), ( "Audit {subject} without an identifier handoff or prescribed command. Find one relevant canonical " "claim and linked evidence yourself, cite the claim ID and source or evidence ID, show `Claim body:` " "and `Evidence:` separately, challenge a conclusion that outruns the evidence, and finish with a " "narrower `Revision:`. Read-only only; no staging, approval, or apply." ), ), }, { "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 `teleo-kb` tool exactly once with `action: context`, " "`query: {subject}`, `limit: 1`, `context_limit: 1`, and `format: markdown`. 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` tool call using `action: context`, " "`query: {subject}`, `limit: 1`, `context_limit: 1`, and `format: markdown`. Answer only with " "`Subject: {subject}`, `Mode: read-only`, " "`Surface: context`, and `Receipt: `." ), ( "Prove the {subject} came from a live read: call the `teleo-kb` tool once with `action: context`, " "`query: {subject}`, `limit: 1`, `context_limit: 1`, and `format: markdown`, 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", "label", "remember", "chat"), "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", "label", "proof", "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 harness_git_head() -> str: return subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True, ).strip() 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 _expected_tool_arguments(subject: str, *, required: bool) -> dict[str, Any] | None: if not required: return None return { "action": "context", "query": subject, "limit": 1, "context_limit": 1, "format": "markdown", } 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" requires_grounded_retrieval_answer = family["family_id"] in AUTONOMOUS_RETRIEVAL_FAMILIES expected_tool_arguments = _expected_tool_arguments(subject, required=requires_tool_evidence) 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, "requires_grounded_retrieval_answer": requires_grounded_retrieval_answer, "custom_evidence_probe": family["scorer_id"] == "EVIDENCE-01", "expected_tool_arguments_sha256": tool_trace_lib.tool_arguments_sha256(expected_tool_arguments) if expected_tool_arguments 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 structured teleo-kb read execution", ], "preserved_surfaces": [ "prompt manifest and order", "scorer and thresholds", "deployed build and model configuration", "temporary profile seed", "model-visible skills and structured teleo-kb 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), }, "harness_git_head": harness_git_head(), "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") if protocol.get("generator_version") != GENERATOR_VERSION: issues.append("wrong_generator_version") if protocol.get("scorer_version") != SCORER_VERSION: issues.append("wrong_scorer_version") if (protocol.get("baseline") or {}).get("version") != BASELINE_VERSION: issues.append("wrong_baseline_version") if not _valid_git_revision(protocol.get("harness_git_head")): issues.append("invalid_harness_git_head") 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}") requires_grounded_answer = prompt.get("requires_grounded_retrieval_answer") is True if requires_grounded_answer != (prompt.get("family_id") in AUTONOMOUS_RETRIEVAL_FAMILIES): issues.append(f"grounded_retrieval_answer_requirement_mismatch:{prompt_id}") if requires_grounded_answer and ( "`teleo-kb` tool" in message or "`action: context`" in message or "`context_limit:" in message ): issues.append(f"autonomous_retrieval_exact_command_leak:{prompt_id}") if requires_tool_evidence and ( "`teleo-kb`" not in message or "`action: context`" not in message or "`Receipt:" not in message ): issues.append(f"tool_evidence_instruction_missing:{prompt_id}") expected_arguments = _expected_tool_arguments(subject, required=requires_tool_evidence) expected_arguments_hash = ( tool_trace_lib.tool_arguments_sha256(expected_arguments) if expected_arguments else None ) if prompt.get("expected_tool_arguments_sha256") != expected_arguments_hash: issues.append(f"tool_arguments_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: if protocol.get("harness_git_head") != harness_git_head(): issues.append("harness_git_head_changed_after_freeze") 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_arguments_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_arguments_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 call.get("arguments_sha256") != expected_arguments_sha256 or invocations[0].get("command_sha256") != expected_arguments_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 _resolve_supported_citations( reply: str, pattern: re.Pattern[str], supported_identifiers: set[str], ) -> tuple[set[str], set[str], set[str]]: """Resolve full UUIDs or unambiguous 8-12 hex prefixes to receipted IDs.""" tokens = {match.group(1).lower() for match in pattern.finditer(reply)} resolved: set[str] = set() unresolved: set[str] = set() for token in tokens: matches = {identifier for identifier in supported_identifiers if identifier.startswith(token)} if len(matches) == 1: resolved.update(matches) else: unresolved.add(token) return tokens, resolved, unresolved 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", "python_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, require_grounded_rows: bool = False, ) -> 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 ] post_contract_satisfaction_reported = bool(post) and all( isinstance(item.get("contract_satisfied"), bool) for item in post ) 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 {} identifier_lists_typed = all( isinstance(receipt.get(key), list) and all(isinstance(identifier, str) and identifier for identifier in receipt[key]) for key in ("claim_ids", "source_ids", "contract_row_ids") ) counts = receipt.get("counts") if isinstance(receipt.get("counts"), dict) else {} receipt_counts_typed = all( isinstance(counts.get(key), int) and not isinstance(counts.get(key), bool) and counts[key] >= 0 for key in ("claims", "context_rows", "evidence_rows") ) 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("injected_rows_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 and identifier_lists_typed and receipt_counts_typed ): retrieval_records.append(item) receipt_claim_ids = { str(identifier).lower() for item in retrieval_records for identifier in ((item.get("retrieval_receipt") or {}).get("claim_ids") or []) } receipt_source_ids = { str(identifier).lower() for item in retrieval_records for identifier in ((item.get("retrieval_receipt") or {}).get("source_ids") or []) } receipt_contract_row_ids = { str(identifier).lower() for item in retrieval_records for identifier in ((item.get("retrieval_receipt") or {}).get("contract_row_ids") or []) } supported_identifiers = { str(identifier).lower() for item in retrieval_records for key in ("claim_ids", "source_ids", "contract_row_ids") for identifier in ((item.get("retrieval_receipt") or {}).get(key) or []) } reply = str(result.get("reply") or "") reply_identifiers = {match.group(0).lower() for match in UUID_RE.finditer(reply)} reply_claim_citation_tokens, reply_claim_citations, unresolved_claim_citations = _resolve_supported_citations( reply, CLAIM_ID_CITATION_RE, receipt_claim_ids ) reply_source_citation_tokens, reply_source_citations, unresolved_source_citations = _resolve_supported_citations( reply, SOURCE_ID_CITATION_RE, receipt_source_ids ) unsupported_identifiers = sorted(reply_identifiers - supported_identifiers) receipt_counts = [(item.get("retrieval_receipt") or {}).get("counts") or {} for item in retrieval_records] reply_sha256 = hashlib.sha256(reply.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_trace_bound": len(post) == 1, # Contract satisfaction is semantic evidence, not an execution-binding gate. # The answer scorer independently decides whether the delivered response is # correct; a false value must remain visible without invalidating the receipt. "post_contract_satisfaction_reported": post_contract_satisfaction_reported, "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, "grounded_claim_rows_nonempty": any( counts.get("claims", 0) >= 1 and receipt_claim_ids for counts in receipt_counts ) if require_grounded_rows else True, "grounded_evidence_rows_nonempty": any( counts.get("evidence_rows", 0) >= 1 and receipt_source_ids for counts in receipt_counts ) if require_grounded_rows else True, "reply_cites_supported_claim_id": bool(reply_claim_citations & receipt_claim_ids) if require_grounded_rows else True, "reply_cites_supported_source_or_evidence_id": bool(reply_source_citations & receipt_source_ids) if require_grounded_rows 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, "no_unresolved_or_ambiguous_identifier_citations": not unresolved_claim_citations and not unresolved_source_citations, } 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), "reply_claim_citation_tokens": sorted(reply_claim_citation_tokens), "reply_claim_citations": sorted(reply_claim_citations), "unresolved_claim_citations": sorted(unresolved_claim_citations), "reply_source_or_evidence_citation_tokens": sorted(reply_source_citation_tokens), "reply_source_or_evidence_citations": sorted(reply_source_citations), "unresolved_source_or_evidence_citations": sorted(unresolved_source_citations), "supported_claim_ids": sorted(receipt_claim_ids), "supported_source_ids": sorted(receipt_source_ids), "supported_contract_row_ids": sorted(receipt_contract_row_ids), "supported_identifiers": sorted(supported_identifiers), "unsupported_identifiers": unsupported_identifiers, "post_contract_satisfied": post[0].get("contract_satisfied") if len(post) == 1 else None, "pass": all(checks.values()), } def _benchmark_execution_chain( report: dict[str, Any], *, expected_harness_git_head: str, ) -> 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 accepts either a clean harness or exactly one control-owned untracked file (``goal.md``). Only the latter may omit the generic clean-worktree binding. In the ablated arm, the bindings made impossible by removing the DB-context plugin may also be absent. Every other runtime, model, session, and safety binding remains mandatory and is checked independently. """ 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 {} clean_worktree_exact = ( local_state.get("worktree_clean") is True and local_state.get("status_lines") == [] and local_state.get("only_control_goal_untracked") is False and local_state.get("status_sha256") == hashlib.sha256(b"").hexdigest() ) control_goal_only_exact = ( local_state.get("worktree_clean") is False and local_state.get("status_lines") == ["?? goal.md"] and local_state.get("only_control_goal_untracked") is True and local_state.get("status_sha256") == hashlib.sha256(b"?? goal.md\n").hexdigest() ) supported_worktree_mode = clean_worktree_exact or control_goal_only_exact worktree_allowed_missing = CONTROL_GOAL_EXECUTION_ALLOWED_MISSING if control_goal_only_exact else frozenset() mode = report.get("grounding_mode") if mode == "grounded": allowed_missing_sets = (GROUNDED_EXECUTION_ALLOWED_MISSING | worktree_allowed_missing,) elif mode == "db_tool_ablated": allowed_missing = ABLATION_EXECUTION_ALLOWED_MISSING | worktree_allowed_missing allowed_missing_sets = ( allowed_missing, allowed_missing | ABLATION_EXECUTION_OPTIONAL_MISSING, ) else: allowed_missing_sets = (frozenset(),) local_state_checks = { "git_head_valid": _valid_git_revision(local_state.get("git_head")), "git_head_matches_frozen_harness": local_state.get("git_head") == expected_harness_git_head, "status_sha256_valid": _valid_sha256(local_state.get("status_sha256")), "clean_worktree_exact": clean_worktree_exact, "control_goal_only_exact": control_goal_only_exact, "supported_worktree_mode": supported_worktree_mode, "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") == local_state.get("worktree_clean"), } 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 in allowed_missing_sets and attribution.get("status") == ("incomplete" if missing_set 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[key] for key in ( "git_head_valid", "git_head_matches_frozen_harness", "status_sha256_valid", "supported_worktree_mode", "generic_summary_source_bound", ) ), "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_binding_sets": [sorted(items) for items in allowed_missing_sets], "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, expected_harness_git_head: str, ) -> 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, expected_harness_git_head=expected_harness_git_head, ) tool_surface = report.get("read_only_tool_surface") or {} post_runner_tool_surface = report.get("post_runner_tool_surface") or {} report_source_hashes = report.get("source_hashes") 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", "teleo-kb"], "send_message_tool_absent": tool_surface.get("send_message_tool_enabled") is False, "shell_tool_absent": tool_surface.get("shell_tool_enabled") is False, "mutating_bridge_commands_not_exposed": tool_surface.get("mutating_bridge_commands_exposed") is False, "kb_tool_provider_credentials_not_forwarded": tool_surface.get("provider_credentials_forwarded_to_kb_tool") is False, "structured_read_tool_named": tool_surface.get("structured_read_tool_name") == "teleo-kb", "structured_read_tool_restricted_to_frozen_kb_tool": tool_surface.get( "structured_read_tool_restricted_to_frozen_kb_tool" ) is True, "structured_read_tool_schema_hash_bound": _valid_sha256( tool_surface.get("structured_read_tool_definition_sha256") ), "structured_read_kb_tool_hash_bound": _valid_sha256(tool_surface.get("kb_tool_sha256")) and tool_surface.get("kb_tool_sha256") == report_source_hashes.get("kb_tool_sha256"), "structured_read_timeout_bounded": isinstance(tool_surface.get("bridge_timeout_seconds"), int) and not isinstance(tool_surface.get("bridge_timeout_seconds"), bool) and 0 < tool_surface.get("bridge_timeout_seconds") <= 60, "post_runner_tool_registry_still_exact": post_runner_tool_surface.get("actual_registry_tools") == ["skill_view", "skills_list", "teleo-kb"], "post_runner_tool_schema_unchanged": post_runner_tool_surface.get("unchanged_from_preexecution") is True and post_runner_tool_surface.get("structured_read_tool_definition_sha256") == tool_surface.get("structured_read_tool_definition_sha256"), "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"}, "embedded_tool_trace_matches_frozen_source": report.get("tool_trace_source_sha256") == (report.get("source_hashes") or {}).get("tool_trace_sha256"), } 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"]), require_grounded_rows=bool(prompt.get("requires_grounded_retrieval_answer")), ) 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_arguments_sha256=prompt.get("expected_tool_arguments_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, expected_harness_git_head=protocol["harness_git_head"], ) baseline_top_safety = _top_level_safety( baseline_report, require_handler_safety_gate=False, expected_harness_git_head=protocol["harness_git_head"], ) 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") ), require_grounded_rows=bool( by_prompt.get(str(result.get("prompt_id")), {}).get("requires_grounded_retrieval_answer") ), ) 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") ) autonomous_prompt_ids = [ prompt["id"] for prompt in trial["prompts"] if prompt.get("requires_grounded_retrieval_answer") is True ] autonomous_rows: list[dict[str, Any]] = [] for prompt_id in autonomous_prompt_ids: grounded_reply = str((report_by_prompt.get(prompt_id) or {}).get("reply") or "") baseline_reply = str((baseline_by_prompt.get(prompt_id) or {}).get("reply") or "") grounded_receipt_score = receipts.get(prompt_id, {}) supported_claim_ids = set(grounded_receipt_score.get("supported_claim_ids") or []) supported_source_ids = set(grounded_receipt_score.get("supported_source_ids") or []) supported_identifiers = set(grounded_receipt_score.get("supported_identifiers") or []) ablation_claim_tokens, ablation_claim_citations, ablation_unresolved_claim_citations = ( _resolve_supported_citations(baseline_reply, CLAIM_ID_CITATION_RE, supported_claim_ids) ) ablation_source_tokens, ablation_source_citations, ablation_unresolved_source_citations = ( _resolve_supported_citations(baseline_reply, SOURCE_ID_CITATION_RE, supported_source_ids) ) ablation_reply_identifiers = {match.group(0).lower() for match in UUID_RE.finditer(baseline_reply)} grounded_answer_pass = bool( semantic_by_prompt.get(prompt_id, {}).get("pass") and subject_alignment.get(prompt_id) and grounded_receipt_score.get("pass") ) ablation_binding_checks = { "semantic_pass": baseline_semantic_by_prompt.get(prompt_id, {}).get("pass") is True, "subject_alignment": baseline_subject_alignment.get(prompt_id) is True, "cites_grounded_claim_id": bool(ablation_claim_citations & supported_claim_ids), "cites_grounded_source_id": bool(ablation_source_citations & supported_source_ids), "no_identifiers_outside_grounded_receipt": bool(ablation_claim_tokens or ablation_source_tokens) and not ablation_unresolved_claim_citations and not ablation_unresolved_source_citations and ablation_reply_identifiers <= supported_identifiers, } ablation_answer_pass = all(ablation_binding_checks.values()) replies_identical = grounded_reply == baseline_reply autonomous_rows.append( { "prompt_id": prompt_id, "grounded_answer_pass": grounded_answer_pass, "ablation_answer_pass": ablation_answer_pass, "ablation_scored_against_grounded_receipt": ablation_binding_checks, "replies_identical": replies_identical, "causally_attributed_grounded_pass": bool( grounded_answer_pass and not ablation_answer_pass and not replies_identical ), "grounded_reply_sha256": hashlib.sha256(grounded_reply.encode()).hexdigest(), "ablation_reply_sha256": hashlib.sha256(baseline_reply.encode()).hexdigest(), } ) autonomous_prompt_count = len(autonomous_rows) autonomous_grounded_passes = sum(1 for row in autonomous_rows if row["grounded_answer_pass"]) autonomous_ablation_passes = sum(1 for row in autonomous_rows if row["ablation_answer_pass"]) autonomous_causal_passes = sum(1 for row in autonomous_rows if row["causally_attributed_grounded_pass"]) autonomous_retrieval_comparison = { "method": "both_arms_semantic_subject_and_citations_scored_against_grounded_receipted_ids", "prompt_ids": autonomous_prompt_ids, "prompt_count": autonomous_prompt_count, "grounded_passes": autonomous_grounded_passes, "ablation_passes": autonomous_ablation_passes, "causally_attributed_grounded_passes": autonomous_causal_passes, "grounded_minus_ablation_answer_delta": ( (autonomous_grounded_passes - autonomous_ablation_passes) / autonomous_prompt_count if autonomous_prompt_count else 0.0 ), "identical_reply_prompt_ids": [row["prompt_id"] for row in autonomous_rows if row["replies_identical"]], "rows": autonomous_rows, "pass": bool( autonomous_prompt_count and autonomous_grounded_passes == autonomous_prompt_count and autonomous_ablation_passes == 0 and autonomous_causal_passes == autonomous_prompt_count ), } 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) baseline_tool_surface = baseline_report.get("read_only_tool_surface") or {} grounded_tool_surface = report.get("read_only_tool_surface") or {} 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_allowlist_equal": baseline_tool_surface.get("allowed_tools") == grounded_tool_surface.get("allowed_tools"), "tool_schema_equal": _valid_sha256(baseline_tool_surface.get("structured_read_tool_definition_sha256")) and baseline_tool_surface.get("structured_read_tool_definition_sha256") == grounded_tool_surface.get("structured_read_tool_definition_sha256"), "frozen_kb_tool_equal": _valid_sha256(baseline_tool_surface.get("kb_tool_sha256")) and baseline_tool_surface.get("kb_tool_sha256") == grounded_tool_surface.get("kb_tool_sha256"), } 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, }, "autonomous_retrieval_comparison": autonomous_retrieval_comparison, "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 autonomous_retrieval_comparison["pass"] 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()), "autonomous_retrieval_comparison_passed": mapping(score.get("autonomous_retrieval_comparison")).get("pass") is True, "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 ] autonomous_retrieval_deltas = [ float( (by_id[trial_id].get("autonomous_retrieval_comparison") or {}).get("grounded_minus_ablation_answer_delta") 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), "autonomous_retrieval_has_causal_lift_every_trial": len(autonomous_retrieval_deltas) == len(expected_ids) and all(delta == 1.0 for delta in autonomous_retrieval_deltas), "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" ), } evaluation_pass = all(checks.values()) independent_live_attestation = { "status": "not_configured", "pass": False, "required_for": "machine_verifiable_T3_live_readonly", "reason": ( "The protocol, reports, and scores are caller-readable artifacts. They validate benchmark semantics " "but do not contain a non-caller-mintable platform or service attestation of live execution." ), } 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, "autonomous_retrieval_causal_lift_by_trial": autonomous_retrieval_deltas, "checks": checks, "trial_score_integrity": integrity, "trial_scores": trial_scores, "required_tier": "T3_live_readonly", "current_tier": "T2_runtime_artifact_validation", "operator_observed_tier": "T3_live_readonly_unattested", "evaluation_pass": evaluation_pass, "independent_live_attestation": independent_live_attestation, "machine_verifiable_t3_pass": evaluation_pass and independent_live_attestation["pass"], "claim_ceiling": ( "The retained artifacts can validate frozen-family scoring, grounded-versus-ablated comparisons, " "tool/response binding, and unchanged database fingerprints at T2. An operator may separately report " "that they were captured from a live VPS, but these caller-readable files do not independently prove " "that T3 origin. No Telegram delivery or production apply is proven." ), } aggregate["pass"] = aggregate["machine_verifiable_t3_pass"] 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())