"""Tests for the frozen blinded protocol, receipt gates, and aggregation.""" from __future__ import annotations import copy import hashlib import importlib.util import json import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import working_leo_m3taversal_oos_protocol as protocol_lib # noqa: E402 def load_legacy_test_helpers(): path = ROOT / "tests" / "test_working_leo_m3taversal_oos_benchmark.py" spec = importlib.util.spec_from_file_location("oos_legacy_test_helpers", path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module LEGACY = load_legacy_test_helpers() CLAIM_UUID = "11111111-1111-4111-8111-111111111111" SOURCE_UUID = "22222222-2222-4222-8222-222222222222" CONTRACT_ROW_UUID = "33333333-3333-4333-8333-333333333333" def frozen_protocol() -> dict: return protocol_lib.freeze_protocol( "unit-test-seed-without-live-output", created_at_utc="2026-07-15T00:00:00+00:00", ) def tool_surface(mode: str) -> dict: return { "mode": mode, "allowed_tools": ["skills_list", "skill_view", "teleo-kb"], "actual_registry_tools": ["skill_view", "skills_list", "teleo-kb"], "read_only_bridge_commands": ["status"] if mode == "read_only_kb" else [], "send_message_tool_enabled": False, "shell_tool_enabled": False, "structured_read_tool_name": "teleo-kb", "structured_read_tool_definition_sha256": "a" * 64, "structured_read_tool_restricted_to_frozen_kb_tool": True, "kb_tool_sha256": protocol_lib.file_sha256(protocol_lib.source_paths()["kb_tool_sha256"]), "bridge_timeout_seconds": 45, "mutating_bridge_commands_exposed": False, "provider_credentials_forwarded_to_kb_tool": False, } def retrieval_receipt(query_sha256: str) -> dict: receipt = { "schema": "livingip.teleoKbRetrievalReceipt.v1", "query_sha256": query_sha256, "semantic_context_sha256": "1" * 64, "artifact_state_sha256": "2" * 64, "injected_rows_sha256": "4" * 64, "receipt_sha256": "3" * 64, "claim_ids": [CLAIM_UUID], "source_ids": [SOURCE_UUID], "contract_row_ids": [CONTRACT_ROW_UUID], "counts": {"claims": 1, "context_rows": 1, "evidence_rows": 1}, "read_consistency": { "status": "stable_wal_marker", "attempts": 1, "database": "teleo", "database_user": "postgres", "system_identifier": "12345", "wal_lsn_before": "0/1", "wal_lsn_after": "0/1", }, } trace_payload = {key: value for key, value in receipt.items() if key != "receipt_sha256"} receipt["trace_payload_sha256"] = hashlib.sha256( json.dumps(trace_payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() return receipt def rehash_retrieval_receipt(receipt: dict) -> None: trace_payload = { key: value for key, value in receipt.items() if key not in {"receipt_sha256", "trace_payload_sha256"} } receipt["trace_payload_sha256"] = hashlib.sha256( json.dumps(trace_payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() def result_for(prompt: dict, token: str, turn: int, *, grounded: bool) -> dict: query_sha256 = hashlib.sha256(prompt["message"].encode()).hexdigest() if prompt.get("custom_evidence_probe"): reply = f"Subject: {prompt['subject']}\nMode: read-only\nSurface: context" else: reply = f"{prompt['subject']}. {LEGACY.good_reply(prompt['scorer_id'], token)}" if prompt.get("requires_grounded_retrieval_answer") and not grounded: reply = ( f"{prompt['subject']}. Claim body and evidence are unavailable without a live receipt. " "Challenge: the missing evidence does not support a revision. Revision: no grounded narrowing is " "possible from generic prose alone." ) if grounded and prompt["requires_tool_evidence_token"]: reply += "\nReceipt: aaaaaaaaaaaa" contexts = [] if grounded: contexts = [ { "event": "pre_llm_call", "status": "ok", "injected": True, "source": "kb_tool.py --local context", "query_sha256": query_sha256, "contract_ids": ["reply_budget", "demo_capability_readback"], "contract_sha256": "4" * 64, "compiled_response_available": True, "retrieval_receipt": retrieval_receipt(query_sha256), }, { "event": "post_llm_call", "status": "ok", "validated": True, "contract_satisfied": True, "query_sha256": query_sha256, "contract_ids": ["reply_budget", "demo_capability_readback"], "delivered_response_sha256": hashlib.sha256(reply.encode()).hexdigest(), }, ] tool_calls = [] if grounded and prompt["requires_tool_evidence_token"]: tool_calls = [ { "database_invocations": [ { "access_mode": "read_only", "executable": "teleo-kb", "subcommand": "context", "command_sha256": prompt["expected_tool_arguments_sha256"], } ], "arguments_sha256": prompt["expected_tool_arguments_sha256"], "result": { "nonempty": True, "error_detected": False, "retrieval_receipt": { "schema": "livingip.teleoKbRetrievalReceipt.v1", "semantic_context_sha256": "a" * 64, "artifact_state_sha256": "b" * 64, "read_consistency_status": "stable_wal_marker", }, }, } ] return { "turn": turn, "prompt_id": prompt["id"], "dimension": prompt["dimension"], "prompt": prompt["message"], "reply": reply, "ok": True, "mutates_kb": False, "database_context_trace": contexts, "database_tool_trace": { "schema": "livingip.leoKbToolTrace.v1", "database_tool_call_count": len(tool_calls), "database_tool_completed_count": len(tool_calls), "database_retrieval_receipt_proven": bool(tool_calls), "database_tool_calls_read_only": bool(tool_calls), "access_modes": ["read_only"] if tool_calls else [], "calls": tool_calls, }, "model_call_trace": [ { "event": "post_api_request", "model": "test-model", "provider": "test-provider", "base_url_sha256": "5" * 64, "api_mode": "chat", "response_model": "test-model", } ], "conversation_before": {"message_count": 0 if turn == 1 else (turn - 1) * 2}, "conversation_after": {"message_count": turn * 2}, "conversation_history_prefix_preserved": True, } def fake_behavior_manifest(*, grounded: bool) -> dict: def component(name: str, files: list[dict] | None = None) -> dict: file_rows = files or [ { "path": f"{name}.txt", "bytes": 1, "mode": "0o644", "sha256": hashlib.sha256(name.encode()).hexdigest(), } ] content_stable = {"files": file_rows, "missing": [], "symlinks": []} return { "role": f"test {name}", "mutability": "test_static", "replayability": "fully_hashable", "content": { **content_stable, "file_count": len(file_rows), "total_bytes": sum(int(item["bytes"]) for item in file_rows), "sha256": protocol_lib.canonical_sha256(content_stable), }, } common_middleware = { "path": "plugins/leo-turn-trace/__init__.py", "bytes": 1, "mode": "0o644", "sha256": "3" * 64, } middleware_files = [common_middleware] if grounded: instrumented = protocol_lib.instrument_db_context_plugin_source( protocol_lib.source_paths()["db_context_plugin_sha256"].read_text(encoding="utf-8") ) middleware_files.append( { "path": "plugins/leo-db-context/__init__.py", "bytes": len(instrumented.encode()), "mode": "0o644", "sha256": hashlib.sha256(instrumented.encode()).hexdigest(), } ) plugin_manifest_path = protocol_lib.source_paths()["db_context_plugin_manifest_sha256"] middleware_files.append( { "path": "plugins/leo-db-context/plugin.yaml", "bytes": plugin_manifest_path.stat().st_size, "mode": "0o644", "sha256": hashlib.sha256(plugin_manifest_path.read_bytes()).hexdigest(), } ) components = { name: component(name) for name in ( "profile_config", "runtime_identity", "procedural_skills", "database_tools", "persistent_memory", "conversation_sessions", "generated_prompt_cache", ) } components["runtime_middleware"] = component("runtime_middleware", middleware_files) stable = { "schema": "livingip.leoBehaviorManifest.v1", "model_runtime": {"model": "test-model"}, "python_runtime": {"python_version": "3.11.15", "implementation": "CPython"}, "hermes_runtime": { "git_head": "e" * 40, "source_tree": {"sha256": "f" * 64, "file_count": 1}, }, "teleo_infrastructure_runtime": { "git_head": "1" * 40, "source_tree": {"sha256": "2" * 64, "file_count": 1}, }, "components": components, "canonical_database": {"included": False}, } return { **stable, "generated_at_utc": "2026-07-15T00:01:00+00:00", "profile_root": "/tmp/profile", "hermes_root": "/opt/hermes", "deployment_root": "/opt/teleo", "behavior_sha256": protocol_lib.canonical_sha256(stable), } def attach_fake_execution_manifests( report: dict, *, grounded: bool, harness_git_head: str, harness_worktree_mode: str = "control_goal_only", ) -> None: if harness_worktree_mode == "clean": worktree_clean = True status_text = "" status_lines: list[str] = [] only_control_goal_untracked = False elif harness_worktree_mode == "control_goal_only": worktree_clean = False status_text = "?? goal.md\n" status_lines = ["?? goal.md"] only_control_goal_untracked = True else: raise ValueError(f"unsupported harness worktree mode: {harness_worktree_mode}") harness_source = { "git_head": harness_git_head, "worktree_clean": worktree_clean, "status_sha256": hashlib.sha256(status_text.encode()).hexdigest(), } previous_execution_sha256 = None allowed_missing = set( protocol_lib.GROUNDED_EXECUTION_ALLOWED_MISSING if grounded else protocol_lib.ABLATION_EXECUTION_ALLOWED_MISSING ) if harness_worktree_mode == "control_goal_only": allowed_missing.update(protocol_lib.CONTROL_GOAL_EXECUTION_ALLOWED_MISSING) fingerprint = report["db_fingerprint_before"] counts_sha256 = protocol_lib.canonical_sha256({}) for result in report["results"]: conversation = { "history_prefix_preserved": True, "conversation_hashes_valid": True, "prior_turn_state_bound": True, "previous_execution_sha256": previous_execution_sha256, } stable = { "schema": protocol_lib.execution_manifest_lib.SCHEMA, "turn": { "prompt_id": result["prompt_id"], "prompt_sha256": hashlib.sha256(result["prompt"].encode()).hexdigest(), "reply_sha256": hashlib.sha256(result["reply"].encode()).hexdigest(), }, "session_boundary": { "session_key_sha256": "b" * 64, "source_platform": "telegram", "fresh_temp_profile_for_suite": True, "prior_dynamic_state_excluded_from_suite": True, "conversation": conversation, }, "runtime": { "behavior_sha256": report["executed_behavior_manifest"]["behavior_sha256"], "hermes_runtime": report["executed_behavior_manifest"]["hermes_runtime"], "teleo_infrastructure_runtime": report["executed_behavior_manifest"]["teleo_infrastructure_runtime"], "harness_source": harness_source, }, "model_execution": { "call_count": 1, "calls": [{"model": "test-model", "provider": "test-provider"}], "prompt_bound": True, "raw_response_bound": grounded, "delivered_response_bound": True, "response_trace_count_matches_api_calls": True, "api_call_sequence_valid": True, "session_binding_valid": True, "response_hashes_valid": True, }, "canonical_database": { "binding_status": "retrieval_receipt_bound" if grounded else "missing", "context_retrieval_receipts": [retrieval_receipt(hashlib.sha256(result["prompt"].encode()).hexdigest())] if grounded else [], "context_binding": { "query_bound": grounded, "context_available": grounded, "response_bound": grounded, }, "database_tool_binding": {"all_calls_read_only": True}, "fingerprint_before": fingerprint, "fingerprint_after": fingerprint, "fingerprint_unchanged": True, "suite_counts_before_sha256": counts_sha256, "suite_counts_after_sha256": counts_sha256, "suite_counts_changed": False, }, "delivery_and_safety": { "posted_to_telegram": False, "kb_mutation_by_harness": False, "turn_mutates_kb": False, "suite_safety": { "remote_returncode": 0, "pass_runtime": True, "live_behavior_manifest_unchanged": True, "temp_profile_removed": True, "service_unchanged": True, "db_fingerprint_unchanged": True, "model_call_trace_all_bound": True, }, }, "attribution": { "status": "incomplete" if allowed_missing else "complete", "missing_required_bindings": sorted(allowed_missing), }, } manifest = { **stable, "generated_at_utc": "2026-07-15T00:02:00+00:00", "execution_sha256": protocol_lib.execution_manifest_lib.canonical_sha256(stable), } result["execution_manifest"] = manifest previous_execution_sha256 = manifest["execution_sha256"] report["oos_harness_git_state"] = { **harness_source, "status_lines": status_lines, "only_control_goal_untracked": only_control_goal_untracked, } report["execution_manifest_summary"] = { "schema": protocol_lib.execution_manifest_lib.SCHEMA, "turn_count": len(report["results"]), "attribution_complete_count": 0 if allowed_missing else len(report["results"]), "all_turns_attribution_complete": not allowed_missing, "harness_source": harness_source, } def fake_report( protocol: dict, trial: dict, *, grounded: bool, harness_worktree_mode: str = "control_goal_only", ) -> dict: mode = "grounded" if grounded else "db_tool_ablated" fingerprint = { "status": "ok", "fingerprint_sha256": "6" * 64, "table_rows_sha256": "7" * 64, "structure_sha256": "8" * 64, } results = [ result_for(prompt, trial["memory_token"], index, grounded=grounded) for index, prompt in enumerate(trial["prompts"], start=1) ] report = { "protocol_id": protocol["protocol_id"], "protocol_hash_sha256": protocol["protocol_hash_sha256"], "trial_id": trial["trial_id"], "trial_prompt_set_sha256": trial["prompt_set_sha256"], "source_hashes": protocol["source_hashes"], "tool_trace_source_sha256": protocol["source_hashes"]["tool_trace_sha256"], "source_report_path": f"/tmp/{trial['trial_id']}-{mode}.json", "generated_at_utc": "2026-07-15T00:02:00+00:00", "grounding_mode": mode, "db_context_plugin_enabled": grounded, "remote_returncode": 0, "pass_runtime": True, "posted_to_telegram": False, "mutates_kb_by_harness": False, "db_counts_changed": False, "db_fingerprint_before": fingerprint, "db_fingerprint_after": fingerprint, "db_fingerprint_unchanged": True, "live_behavior_manifest_before": {"behavior_sha256": "9" * 64}, "live_behavior_manifest_after": {"behavior_sha256": "9" * 64}, "live_behavior_manifest_unchanged": True, "service_before_after": { "before": {"MainPID": "202", "ActiveState": "active", "SubState": "running"}, "after": {"MainPID": "202", "ActiveState": "active", "SubState": "running"}, "unchanged_from_preexisting_live_readback": True, }, "before_service": { "MainPID": "202", "ActiveState": "active", "SubState": "running", "ExecMainStartTimestamp": "Wed 2026-07-15 00:01:00 UTC", }, "temp_profile_removed": True, "temp_profile_seed": { "mode": "static_runtime_surfaces_only", "excluded": ["sessions", "state", "state.db"], "same_session_continuity_starts_fresh": True, }, "executed_behavior_manifest": fake_behavior_manifest(grounded=grounded), "read_only_tool_surface": tool_surface("read_only_kb" if grounded else "no_db_ablation"), "post_runner_tool_surface": { "initial_registry_tools": ["process", "skill_view", "skills_list", "teleo-kb"], "removed_known_runner_tools": ["process"], "actual_registry_tools": ["skill_view", "skills_list", "teleo-kb"], "structured_read_tool_definition_sha256": "a" * 64, "final_surface_matches_preexecution": True, }, "readonly_guard_source_sha256": protocol["source_hashes"]["readonly_guard_sha256"], "safety_gate": { "status": "fail", "checks": { "remote_command_succeeded": True, "runtime_completed": True, "handler_authorized": True, "results_nonempty": True, "all_turns_returned_replies": True, "all_turns_declared_no_mutation": True, "all_tool_traces_read_only_and_bound": True, "no_telegram_post": True, "harness_declared_no_kb_mutation": True, "database_counts_unchanged": True, "database_fingerprint_unchanged": True, "live_behavior_unchanged": True, "service_unchanged": True, "temporary_profile_removed": True, "model_trace_metadata_bound": True, "all_turn_manifests_complete": False, "no_runtime_error": True, }, "failed_checks": ["all_turn_manifests_complete"], }, "post_run_orphan_readback": {"no_matching_processes": True}, "prompt_leakage_scan": {"pass": True}, "remote_temp_profile_prompt_leakage_scan": { "scope": "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv", "expected_roots": [ {"name": name, "exists": True, "is_symlink": False} for name in ("profile", "skills", "plugins", "bin") ], "scanned_files": 10, "scanned_bytes": 1000, "errors": [], "matches": [], "pass": True, }, "telegram_transport_deny": { "enabled": True, "patched_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS), "expected_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS), "attempt_count": 0, "runner_adapters_empty": True, }, "results": results, } attach_fake_execution_manifests( report, grounded=grounded, harness_git_head=protocol["harness_git_head"], harness_worktree_mode=harness_worktree_mode, ) return report def restart_receipt(protocol: dict, directory: Path, trial: dict) -> dict: fingerprint = {"status": "ok", "fingerprint_sha256": "6" * 64} counts = {"public.claims": 10, "public.sources": 2} deploy = {"returncode": 0, "head": "a" * 40, "stamp": "a" * 40} service_before = { "MainPID": "101", "ActiveState": "active", "SubState": "running", "ExecMainStartTimestamp": "Tue 2026-07-14 23:00:00 UTC", } service_after = { "MainPID": "202", "ActiveState": "active", "SubState": "running", "ExecMainStartTimestamp": "Wed 2026-07-15 00:01:00 UTC", } def probe(path: Path, *, before_restart: bool) -> dict: payload = { "generated_at_utc": "2026-07-15T00:00:00+00:00" if before_restart else "2026-07-15T00:01:15+00:00", "protocol_id": protocol["protocol_id"], "protocol_hash_sha256": protocol["protocol_hash_sha256"], "source_hashes": protocol["source_hashes"], "remote_returncode": 0, "pass_runtime": True, "results": [], "posted_to_telegram": False, "telegram_transport_deny": { "enabled": True, "attempt_count": 0, "patched_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS), "expected_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS), "runner_adapters_empty": True, }, "db_counts_before": counts, "db_counts_after": counts, "db_counts_changed": False, "db_fingerprint_before": fingerprint, "db_fingerprint_after": fingerprint, "db_fingerprint_unchanged": True, "preexecution_safety_gate": {"status": "pass"}, "temp_profile_removed": True, "post_run_orphan_readback": {"no_matching_processes": True}, "before_service": service_before if before_restart else service_after, "service_before_after": { "after": service_before if before_restart else service_after, }, } path.write_text(json.dumps(payload, sort_keys=True) + "\n", encoding="utf-8") return {"path": str(path), "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), "pass": True} directory.mkdir(parents=True, exist_ok=True) before_probe = probe(directory / "restart-before.json", before_restart=True) after_probe = probe(directory / "restart-after.json", before_restart=False) return { "schema": "livingip.leoGatewayRestartReceipt.v1", "generated_at_utc": "2026-07-15T00:01:30+00:00", "protocol_id": protocol["protocol_id"], "protocol_hash_sha256": protocol["protocol_hash_sha256"], "next_trial_id": trial["trial_id"], "next_trial_prompt_set_sha256": trial["prompt_set_sha256"], "restart_started_at_utc": "2026-07-15T00:00:30+00:00", "restart_ended_at_utc": "2026-07-15T00:01:00+00:00", "restart_returncode": 0, "service_before": service_before, "service_after": service_after, "deploy_before": deploy, "deploy_after": deploy, "posted_to_telegram": False, "db_counts_before": counts, "db_counts_after": counts, "db_counts_changed": False, "db_fingerprint_before": fingerprint, "db_fingerprint_after": fingerprint, "db_fingerprint_unchanged": True, "before_probe": before_probe, "after_probe": after_probe, "checks": {"all_independent_checks": True}, "pass": True, } def score_for_trial(protocol: dict, trial: dict, artifact_directory: Path | None = None) -> dict: temporary = tempfile.TemporaryDirectory() if artifact_directory is None else None raw_directory = Path(temporary.name) if temporary is not None else artifact_directory assert raw_directory is not None raw_directory.mkdir(parents=True, exist_ok=True) grounded = fake_report(protocol, trial, grounded=True) baseline = fake_report(protocol, trial, grounded=False) grounded_path = raw_directory / "grounded.json" baseline_path = raw_directory / "baseline.json" grounded_path.write_text(json.dumps(grounded, sort_keys=True) + "\n", encoding="utf-8") baseline_path.write_text(json.dumps(baseline, sort_keys=True) + "\n", encoding="utf-8") restart = ( restart_receipt(protocol, raw_directory / "restart", trial) if trial["session_mode"] == "post_restart_clean_session" else None ) restart_path = raw_directory / "restart-receipt.json" if restart is not None: restart_path.write_text(json.dumps(restart, sort_keys=True) + "\n", encoding="utf-8") try: score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=baseline, restart_receipt=restart, ) score["grounded_report_path"] = str(grounded_path) score["baseline_report_path"] = str(baseline_path) score["grounded_report_sha256"] = hashlib.sha256(grounded_path.read_bytes()).hexdigest() score["baseline_report_sha256"] = hashlib.sha256(baseline_path.read_bytes()).hexdigest() if restart is not None: score["restart_receipt_path"] = str(restart_path) score["restart_receipt_sha256"] = hashlib.sha256(restart_path.read_bytes()).hexdigest() score["restart_receipt_payload_sha256"] = protocol_lib.canonical_sha256(restart) return score finally: if temporary is not None: temporary.cleanup() def test_protocol_freezes_three_unique_variants_and_follow_up_shapes() -> None: protocol = frozen_protocol() assert protocol_lib.validate_protocol(protocol, verify_source_hashes=True)["pass"] is True assert len(protocol["trials"]) == 3 assert protocol["trials"][-1]["session_mode"] == "post_restart_clean_session" for family_id in protocol["blinding"]["prompt_families"]: prompts = [ next(item for item in trial["prompts"] if item["family_id"] == family_id) for trial in protocol["trials"] ] assert len({item["variant_index"] for item in prompts}) == 3 assert len({item["expected_follow_up"] for item in prompts}) == 3 assert all(item["leakage_markers"] for item in prompts) assert all("row id:" not in item["message"].lower() for item in prompts) retrieval_prompts = [ next(item for item in trial["prompts"] if item["family_id"] == "autonomous_retrieval_reasoning") for trial in protocol["trials"] ] assert len({item["subject"] for item in retrieval_prompts}) == 3 assert all(item["requires_grounded_retrieval_answer"] is True for item in retrieval_prompts) assert all("`teleo-kb` tool" not in item["message"] for item in retrieval_prompts) assert all("`action: context`" not in item["message"] for item in retrieval_prompts) def test_protocol_validation_rejects_prompt_and_source_hash_tampering() -> None: protocol = frozen_protocol() protocol["trials"][0]["prompts"][0]["message"] += " changed" protocol["protocol_hash_sha256"] = protocol_lib.canonical_sha256( {key: value for key, value in protocol.items() if key != "protocol_hash_sha256"} ) assert "prompt_hash_mismatch:" in " ".join( protocol_lib.validate_protocol(protocol, verify_source_hashes=True)["issues"] ) protocol = frozen_protocol() protocol["source_hashes"]["readonly_guard_sha256"] = "0" * 64 protocol["protocol_hash_sha256"] = protocol_lib.canonical_sha256( {key: value for key, value in protocol.items() if key != "protocol_hash_sha256"} ) assert ( "source_changed_after_freeze:readonly_guard_sha256" in protocol_lib.validate_protocol(protocol, verify_source_hashes=True)["issues"] ) protocol = frozen_protocol() protocol["scorer_version"] = "stale-scorer" protocol["protocol_hash_sha256"] = protocol_lib.canonical_sha256( {key: value for key, value in protocol.items() if key != "protocol_hash_sha256"} ) assert "wrong_scorer_version" in protocol_lib.validate_protocol(protocol, verify_source_hashes=True)["issues"] def test_live_trial_requires_semantics_subject_receipts_and_real_ablation() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] score = score_for_trial(protocol, trial) assert score["pass"] is True assert score["grounded_pass_rate"] == 1.0 assert score["receipt_ablation"]["grounded_pass_rate"] == 0.0 assert all(item["receipt_pass"] for item in score["prompt_scores"]) assert all(score["comparison_axis_checks"].values()) def test_live_trial_rejects_embedded_tool_trace_source_tampering() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] grounded = fake_report(protocol, trial, grounded=True) baseline = fake_report(protocol, trial, grounded=False) grounded["tool_trace_source_sha256"] = "0" * 64 score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=baseline, ) assert score["pass"] is False assert score["top_level_safety"]["checks"]["embedded_tool_trace_matches_frozen_source"] is False def test_execution_chain_allows_only_declared_ablation_and_control_goal() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] for grounded in (True, False): report = fake_report(protocol, trial, grounded=grounded) validation = protocol_lib._benchmark_execution_chain( report, expected_harness_git_head=protocol["harness_git_head"], ) expected_missing = ( protocol_lib.GROUNDED_EXECUTION_ALLOWED_MISSING if grounded else protocol_lib.ABLATION_EXECUTION_ALLOWED_MISSING ) | protocol_lib.CONTROL_GOAL_EXECUTION_ALLOWED_MISSING assert validation["pass"] is True assert validation["local_state_checks"]["control_goal_only_exact"] is True assert validation["allowed_missing_binding_sets"][0] == sorted(expected_missing) assert all( result["execution_manifest"]["attribution"]["missing_required_bindings"] == sorted(expected_missing) for result in report["results"] ) manifest = report["results"][0]["execution_manifest"] manifest["attribution"]["missing_required_bindings"].append("unexpected_gap") stable = {key: value for key, value in manifest.items() if key not in {"generated_at_utc", "execution_sha256"}} manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable) validation = protocol_lib._benchmark_execution_chain( report, expected_harness_git_head=protocol["harness_git_head"], ) assert validation["pass"] is False assert validation["turn_checks"][trial["prompts"][0]["id"]]["declared_missing_exact"] is False ablated = fake_report(protocol, trial, grounded=False) manifest = ablated["results"][-1]["execution_manifest"] manifest["attribution"]["missing_required_bindings"].append("database_tool_results") stable = {key: value for key, value in manifest.items() if key not in {"generated_at_utc", "execution_sha256"}} manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable) validation = protocol_lib._benchmark_execution_chain( ablated, expected_harness_git_head=protocol["harness_git_head"], ) assert validation["pass"] is True assert validation["turn_checks"][trial["prompts"][-1]["id"]]["declared_missing_exact"] is True drifted = fake_report(protocol, trial, grounded=True) drifted["oos_harness_git_state"]["git_head"] = "f" * 40 validation = protocol_lib._benchmark_execution_chain( drifted, expected_harness_git_head=protocol["harness_git_head"], ) assert validation["pass"] is False assert validation["local_state_checks"]["git_head_matches_frozen_harness"] is False def test_execution_chain_accepts_clean_harness_without_cleanliness_omission() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] for grounded in (True, False): report = fake_report( protocol, trial, grounded=grounded, harness_worktree_mode="clean", ) validation = protocol_lib._benchmark_execution_chain( report, expected_harness_git_head=protocol["harness_git_head"], ) assert validation["pass"] is True assert validation["local_state_checks"]["clean_worktree_exact"] is True expected_missing = ( protocol_lib.GROUNDED_EXECUTION_ALLOWED_MISSING if grounded else protocol_lib.ABLATION_EXECUTION_ALLOWED_MISSING ) assert validation["allowed_missing_binding_sets"][0] == sorted(expected_missing) assert all( result["execution_manifest"]["attribution"]["missing_required_bindings"] == sorted(expected_missing) for result in report["results"] ) assert all(checks["declared_missing_exact"] is True for checks in validation["turn_checks"].values()) manifest = report["results"][-1]["execution_manifest"] manifest["attribution"]["missing_required_bindings"].append("harness_worktree_clean") manifest["attribution"]["status"] = "incomplete" stable = {key: value for key, value in manifest.items() if key not in {"generated_at_utc", "execution_sha256"}} manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable) validation = protocol_lib._benchmark_execution_chain( report, expected_harness_git_head=protocol["harness_git_head"], ) assert validation["pass"] is False assert validation["turn_checks"][trial["prompts"][-1]["id"]]["declared_missing_exact"] is False def test_execution_chain_rejects_inexact_control_goal_state() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] report = fake_report(protocol, trial, grounded=True) report["oos_harness_git_state"]["status_lines"] = ["?? goal.md", "?? scratch.txt"] validation = protocol_lib._benchmark_execution_chain( report, expected_harness_git_head=protocol["harness_git_head"], ) assert validation["pass"] is False assert validation["local_state_checks"]["control_goal_only_exact"] is False def test_comparison_rejects_any_executed_profile_delta_beyond_db_context_removal() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] grounded = fake_report(protocol, trial, grounded=True) baseline = fake_report(protocol, trial, grounded=False) behavior = baseline["executed_behavior_manifest"] identity = behavior["components"]["runtime_identity"]["content"] identity["files"][0]["sha256"] = "9" * 64 identity_stable = { "files": identity["files"], "missing": identity["missing"], "symlinks": identity["symlinks"], } identity["sha256"] = protocol_lib.canonical_sha256(identity_stable) behavior_stable = { key: behavior[key] for key in ( "schema", "model_runtime", "python_runtime", "hermes_runtime", "teleo_infrastructure_runtime", "components", "canonical_database", ) } behavior["behavior_sha256"] = protocol_lib.canonical_sha256(behavior_stable) for result in baseline["results"]: manifest = result["execution_manifest"] manifest["runtime"]["behavior_sha256"] = behavior["behavior_sha256"] stable = {key: value for key, value in manifest.items() if key not in {"generated_at_utc", "execution_sha256"}} manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable) for previous, result in zip(baseline["results"], baseline["results"][1:], strict=False): manifest = result["execution_manifest"] manifest["session_boundary"]["conversation"]["previous_execution_sha256"] = previous["execution_manifest"][ "execution_sha256" ] stable = {key: value for key, value in manifest.items() if key not in {"generated_at_utc", "execution_sha256"}} manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable) score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=baseline, ) assert score["pass"] is False assert score["executed_behavior_ablation"]["checks"]["non_middleware_components_equal"] is False assert score["comparison_axis_checks"]["executed_behavior_manifests_capture_only_declared_ablation"] is False def test_relative_retained_paths_resolve_from_repo_root() -> None: assert protocol_lib._retained_path("pyproject.toml") == ROOT / "pyproject.toml" def test_live_trial_rejects_duplicate_results_malformed_receipts_and_invented_ids() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] grounded = fake_report(protocol, trial, grounded=True) grounded["results"].append(copy.deepcopy(grounded["results"][0])) score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=fake_report(protocol, trial, grounded=False), ) assert score["pass"] is False assert score["prompt_binding"]["checks"]["prompt_ids_unique"] is False grounded = fake_report(protocol, trial, grounded=True) grounded["results"][0]["database_context_trace"][0]["retrieval_receipt"]["receipt_sha256"] = "broken" grounded["results"][0]["reply"] += " Unsupported row 12345678-1234-4123-8123-123456789abc." score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=fake_report(protocol, trial, grounded=False), ) first = score["prompt_scores"][0] assert score["pass"] is False assert first["receipt_pass"] is False assert first["receipt_score"]["unsupported_identifiers"] == ["12345678-1234-4123-8123-123456789abc"] grounded = fake_report(protocol, trial, grounded=True) traced_receipt = grounded["results"][0]["database_context_trace"][0]["retrieval_receipt"] traced_receipt["counts"]["claims"] += 1 score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=fake_report(protocol, trial, grounded=False), ) assert score["pass"] is False assert score["prompt_scores"][0]["receipt_pass"] is False grounded = fake_report(protocol, trial, grounded=True) replayed_hash = "f" * 64 for item in grounded["results"][0]["database_context_trace"]: item["query_sha256"] = replayed_hash if item.get("retrieval_receipt"): item["retrieval_receipt"]["query_sha256"] = replayed_hash score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=fake_report(protocol, trial, grounded=False), ) first = score["prompt_scores"][0] assert first["receipt_score"]["checks"]["context_response_query_hash_bound"] is False assert score["pass"] is False grounded = fake_report(protocol, trial, grounded=True) for item in grounded["results"][0]["database_context_trace"]: item["contract_ids"] = "demo_capability_readback" score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=fake_report(protocol, trial, grounded=False), ) assert score["prompt_scores"][0]["receipt_score"]["checks"]["contract_ids_are_typed_lists"] is False def test_autonomous_retrieval_requires_nonempty_rows_and_receipted_reply_ids() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] prompt = next(item for item in trial["prompts"] if item["family_id"] == "autonomous_retrieval_reasoning") grounded = fake_report(protocol, trial, grounded=True) result = next(item for item in grounded["results"] if item["prompt_id"] == prompt["id"]) valid = protocol_lib._receipt_score( result, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert valid["pass"] is True assert valid["supported_claim_ids"] == [CLAIM_UUID] assert valid["supported_source_ids"] == [SOURCE_UUID] assert valid["post_contract_satisfied"] is True semantically_incomplete = copy.deepcopy(result) semantically_incomplete["database_context_trace"][-1]["contract_satisfied"] = False incomplete_score = protocol_lib._receipt_score( semantically_incomplete, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert incomplete_score["pass"] is True assert incomplete_score["post_contract_satisfied"] is False paraphrased = copy.deepcopy(result) paraphrased["reply"] = ( f"{prompt['subject']}. The live claim `{CLAIM_UUID}` states that one retained source path is auditable. " f"The linked source `{SOURCE_UUID}` documents that path. The source is distinct from the claim and supports " "that observation. However, the evidence cannot prove every database row is auditable. A narrower claim is " "limited to this one path; the lookup was read-only and made no mutation." ) paraphrased["database_context_trace"][-1]["delivered_response_sha256"] = hashlib.sha256( paraphrased["reply"].encode() ).hexdigest() paraphrased_score = protocol_lib._receipt_score( paraphrased, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert paraphrased_score["pass"] is True assert paraphrased_score["reply_claim_citations"] == [CLAIM_UUID] assert paraphrased_score["reply_source_or_evidence_citations"] == [SOURCE_UUID] empty = copy.deepcopy(result) empty_receipt = empty["database_context_trace"][0]["retrieval_receipt"] empty_receipt["claim_ids"] = [] empty_receipt["source_ids"] = [] empty_receipt["counts"] = {"claims": 0, "context_rows": 0, "evidence_rows": 0} rehash_retrieval_receipt(empty_receipt) empty_score = protocol_lib._receipt_score( empty, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert empty_score["pass"] is False assert empty_score["checks"]["grounded_claim_rows_nonempty"] is False assert empty_score["checks"]["grounded_evidence_rows_nonempty"] is False unbound = copy.deepcopy(result) unbound_receipt = unbound["database_context_trace"][0]["retrieval_receipt"] unbound_receipt.pop("injected_rows_sha256") rehash_retrieval_receipt(unbound_receipt) unbound_score = protocol_lib._receipt_score( unbound, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert unbound_score["pass"] is False assert unbound_score["checks"]["database_retrieval_receipt_present"] is False unrelated = copy.deepcopy(result) unrelated_uuid = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" unrelated["reply"] += f" Claim ID: {unrelated_uuid}." unrelated["database_context_trace"][-1]["delivered_response_sha256"] = hashlib.sha256( unrelated["reply"].encode() ).hexdigest() unrelated_score = protocol_lib._receipt_score( unrelated, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert unrelated_score["pass"] is False assert unrelated_score["unsupported_identifiers"] == [unrelated_uuid] def test_autonomous_retrieval_resolves_only_unique_receipt_bound_identifier_prefixes() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] prompt = next(item for item in trial["prompts"] if item["family_id"] == "autonomous_retrieval_reasoning") result = result_for(prompt, trial["memory_token"], 1, grounded=True) result["reply"] = ( f"{prompt['subject']}. Claim {CLAIM_UUID[:8]}, source {SOURCE_UUID[:8]}. " "The claim body states one bounded observation. The evidence documents that observation separately. " "Challenge: it does not rule out alternatives. Revision: the result is limited to this source." ) result["database_context_trace"][-1]["delivered_response_sha256"] = hashlib.sha256( result["reply"].encode() ).hexdigest() unique = protocol_lib._receipt_score( result, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert unique["pass"] is True assert unique["reply_claim_citation_tokens"] == [CLAIM_UUID[:8]] assert unique["reply_claim_citations"] == [CLAIM_UUID] assert unique["reply_source_or_evidence_citation_tokens"] == [SOURCE_UUID[:8]] assert unique["reply_source_or_evidence_citations"] == [SOURCE_UUID] unbound = copy.deepcopy(result) unbound["reply"] = unbound["reply"].replace(CLAIM_UUID[:8], "deadbeef") unbound["database_context_trace"][-1]["delivered_response_sha256"] = hashlib.sha256( unbound["reply"].encode() ).hexdigest() unbound_score = protocol_lib._receipt_score( unbound, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert unbound_score["pass"] is False assert unbound_score["unresolved_claim_citations"] == ["deadbeef"] ambiguous = copy.deepcopy(result) ambiguous_receipt = ambiguous["database_context_trace"][0]["retrieval_receipt"] ambiguous_receipt["claim_ids"].append("11111111-aaaa-4aaa-8aaa-aaaaaaaaaaaa") rehash_retrieval_receipt(ambiguous_receipt) ambiguous_score = protocol_lib._receipt_score( ambiguous, require_database_contract=False, require_database_receipt=True, require_grounded_rows=True, ) assert ambiguous_score["pass"] is False assert ambiguous_score["unresolved_claim_citations"] == [CLAIM_UUID[:8]] def test_autonomous_retrieval_ablation_rejects_generic_prose_and_invented_ids() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] prompt = next(item for item in trial["prompts"] if item["family_id"] == "autonomous_retrieval_reasoning") grounded = fake_report(protocol, trial, grounded=True) ablated = fake_report(protocol, trial, grounded=False) result = next(item for item in ablated["results"] if item["prompt_id"] == prompt["id"]) result["reply"] = ( f"{prompt['subject']}. Fresh canonical public.claims readback. " "Claim ID: aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa. " "Claim body: Generic claim text. " "Evidence ID: bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb. " "Evidence: Generic source text is distinct from the claim body. " "Challenge: the evidence does not prove the broader leap. " "Narrower revision: only the generic source is supported; no mutation was made." ) result["execution_manifest"]["turn"]["reply_sha256"] = hashlib.sha256(result["reply"].encode()).hexdigest() result["execution_manifest"]["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256( { key: value for key, value in result["execution_manifest"].items() if key not in {"generated_at_utc", "execution_sha256"} } ) score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=ablated, ) baseline_receipt = score["receipt_ablation"]["receipt_scores"][prompt["id"]] assert baseline_receipt["pass"] is False assert baseline_receipt["checks"]["database_retrieval_receipt_present"] is False row = score["autonomous_retrieval_comparison"]["rows"][0] assert row["ablation_answer_pass"] is False def test_hash_bound_contract_row_ids_are_supported_but_unreceipted_ids_fail() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] prompt = trial["prompts"][0] result = result_for(prompt, trial["memory_token"], 1, grounded=True) result["reply"] += f" Contract row ID: {CONTRACT_ROW_UUID}." result["database_context_trace"][-1]["delivered_response_sha256"] = hashlib.sha256( result["reply"].encode() ).hexdigest() score = protocol_lib._receipt_score( result, require_database_contract=True, require_database_receipt=True, ) assert score["pass"] is True assert score["supported_contract_row_ids"] == [CONTRACT_ROW_UUID] assert score["unsupported_identifiers"] == [] tampered = copy.deepcopy(result) tampered["database_context_trace"][0]["retrieval_receipt"]["contract_row_ids"] = [] tampered_score = protocol_lib._receipt_score( tampered, require_database_contract=True, require_database_receipt=True, ) assert tampered_score["pass"] is False assert tampered_score["unsupported_identifiers"] == [CONTRACT_ROW_UUID] def test_subject_alignment_rejects_a_different_family_even_with_generic_db_words() -> None: protocol = frozen_protocol() source_prompt = next(item for item in protocol["trials"][0]["prompts"] if item["family_id"] == "source_evidence") wrong_subject = "The database proposal is approved but not applied, so wait for a canonical receipt." assert protocol_lib._subject_alignment(source_prompt, wrong_subject) is False generic_family_only = "The source document has claim_evidence and evidence, but the named packet is omitted." assert protocol_lib._subject_alignment(source_prompt, generic_family_only) is False duplicate_subject = ( f"{source_prompt['subject']} uses source evidence and claim_evidence; " f"repeat {source_prompt['subject']} is not acceptable." ) assert protocol_lib._subject_alignment(source_prompt, duplicate_subject) is True sibling_subject = next(item for item in source_prompt["family_subjects"] if item != source_prompt["subject"]) shotgun = f"{source_prompt['subject']} and {sibling_subject} both use source evidence and claim_evidence." assert protocol_lib._subject_alignment(source_prompt, shotgun) is False retrieval_prompt = next( item for item in protocol["trials"][0]["prompts"] if item["family_id"] == "autonomous_retrieval_reasoning" ) good = ( f"{retrieval_prompt['subject']}. Claim body and source evidence are distinct. " "Challenge: evidence does not prove the leap. Revision: narrower claim." ) assert protocol_lib._subject_alignment(retrieval_prompt, good) is True assert protocol_lib._subject_alignment(retrieval_prompt, good + f" {retrieval_prompt['subject']}.") is True retrieval_sibling = next( item for item in retrieval_prompt["family_subjects"] if item != retrieval_prompt["subject"] ) assert protocol_lib._subject_alignment(retrieval_prompt, good + f" {retrieval_sibling}.") is False memory_prompt = next(item for item in protocol["trials"][0]["prompts"] if item["family_id"] == "session_memory_set") memory_reply = f"Subject: {memory_prompt['subject']}\nLabel: temporary\nBlocker: database schema gap." assert protocol_lib._subject_alignment(memory_prompt, memory_reply) is True def test_evidence_probe_four_line_reply_binds_exact_subject_and_receipt() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] prompt = next(item for item in trial["prompts"] if item["family_id"] == "receipt_discrimination") result = result_for(prompt, trial["memory_token"], 1, grounded=True) assert result["reply"] == ( f"Subject: {prompt['subject']}\nMode: read-only\nSurface: context\nReceipt: aaaaaaaaaaaa" ) assert protocol_lib._subject_alignment(prompt, result["reply"]) is True semantic = protocol_lib._score_semantic_results([result], {**trial, "prompts": [prompt]}) assert semantic["pass"] is True evidence = protocol_lib._evidence_answer_score( prompt, result, semantic_pass=True, subject_alignment=True, grounded_tool_hashes=["a" * 64], ) assert evidence["pass"] is True sibling = next(item for item in prompt["family_subjects"] if item != prompt["subject"]) result["reply"] += f"\nSubject: {sibling}" assert protocol_lib._subject_alignment(prompt, result["reply"]) is False def test_every_runtime_persistence_variant_declares_the_scorer_word_limit() -> None: runtime_family = next(item for item in protocol_lib.BLINDED_FAMILIES if item["family_id"] == "runtime_persistence") assert all("under 180 words" in variant.lower() for variant in runtime_family["variants"]) def test_restart_receipt_binds_new_pid_full_fingerprint_and_next_trial(tmp_path: Path) -> None: protocol = frozen_protocol() trial = protocol["trials"][-1] report = fake_report(protocol, trial, grounded=True) valid = protocol_lib.validate_restart_receipt(restart_receipt(protocol, tmp_path, trial), report) assert valid["pass"] is True broken = restart_receipt(protocol, tmp_path / "broken-pid", trial) broken["service_after"]["MainPID"] = "101" assert protocol_lib.validate_restart_receipt(broken, report)["pass"] is False wrong_protocol = restart_receipt(protocol, tmp_path / "wrong-protocol", trial) wrong_protocol["protocol_hash_sha256"] = "0" * 64 validation = protocol_lib.validate_restart_receipt(wrong_protocol, report) assert validation["pass"] is False assert validation["checks"]["protocol_hash_bound"] is False missing_fingerprint = restart_receipt(protocol, tmp_path / "missing-fingerprint", trial) missing_fingerprint["db_fingerprint_before"] = {"status": "ok"} assert protocol_lib.validate_restart_receipt(missing_fingerprint, report)["pass"] is False corrupt_probe = restart_receipt(protocol, tmp_path / "corrupt-probe", trial) corrupt_probe["before_probe"]["sha256"] = "0" * 64 validation = protocol_lib.validate_restart_receipt(corrupt_probe, report) assert validation["checks"]["before_probe_artifact_valid"] is False assert validation["pass"] is False def test_identical_grounded_and_ablation_replies_cannot_create_evidence_delta() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] grounded = fake_report(protocol, trial, grounded=True) ablated = fake_report(protocol, trial, grounded=False) grounded_by_prompt = {item["prompt_id"]: item for item in grounded["results"]} for result in ablated["results"]: result["reply"] = grounded_by_prompt[result["prompt_id"]]["reply"] score = protocol_lib.score_live_trial(protocol, trial["trial_id"], grounded, baseline_report=ablated) comparison = score["evidence_answer_comparison"] assert comparison["current_pass_rate"] == comparison["ablation_pass_rate"] assert comparison["current_minus_ablation_delta"] == 0.0 retrieval_comparison = score["autonomous_retrieval_comparison"] assert retrieval_comparison["grounded_minus_ablation_answer_delta"] == 0.0 assert retrieval_comparison["identical_reply_prompt_ids"] assert retrieval_comparison["pass"] is False assert score["pass"] is False def test_nonidentical_ablated_answer_with_grounded_ids_does_not_manufacture_causal_lift() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] prompt = next(item for item in trial["prompts"] if item.get("requires_grounded_retrieval_answer")) for claim_citation, source_citation in ( (CLAIM_UUID, SOURCE_UUID), (CLAIM_UUID[:8], SOURCE_UUID[:8]), ): grounded = fake_report(protocol, trial, grounded=True) ablated = fake_report(protocol, trial, grounded=False) result = next(item for item in ablated["results"] if item["prompt_id"] == prompt["id"]) result["reply"] = ( f"{prompt['subject']}. Fresh canonical public.claims readback: claim `{claim_citation}` states that one " f"retained path is auditable. Source `{source_citation}` documents that path and is distinct from the " "claim. However, the evidence cannot prove every path is auditable. A narrower claim is limited to this " "one path. The lookup is read-only and made no mutation." ) score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=ablated, ) row = score["autonomous_retrieval_comparison"]["rows"][0] assert row["ablation_answer_pass"] is True assert row["replies_identical"] is False assert row["causally_attributed_grounded_pass"] is False assert score["autonomous_retrieval_comparison"]["grounded_minus_ablation_answer_delta"] == 0.0 def test_receipt_discriminator_rejects_wrong_command_extra_call_and_wrong_token() -> None: protocol = frozen_protocol() trial = protocol["trials"][0] evidence_prompt = next(item for item in trial["prompts"] if item["custom_evidence_probe"]) for mutation in ("wrong_command", "extra_call", "wrong_token"): grounded = fake_report(protocol, trial, grounded=True) result = next(item for item in grounded["results"] if item["prompt_id"] == evidence_prompt["id"]) trace = result["database_tool_trace"] if mutation == "wrong_command": trace["calls"][0]["database_invocations"][0]["command_sha256"] = "f" * 64 elif mutation == "extra_call": trace["calls"].append(copy.deepcopy(trace["calls"][0])) trace["database_tool_call_count"] = 2 trace["database_tool_completed_count"] = 2 else: result["reply"] = result["reply"].replace("Receipt: aaaaaaaaaaaa", "Receipt: bbbbbbbbbbbb") post = result["database_context_trace"][-1] post["delivered_response_sha256"] = hashlib.sha256(result["reply"].encode()).hexdigest() score = protocol_lib.score_live_trial( protocol, trial["trial_id"], grounded, baseline_report=fake_report(protocol, trial, grounded=False), ) prompt_score = next(item for item in score["prompt_scores"] if item["prompt_id"] == evidence_prompt["id"]) assert prompt_score["evidence_answer_pass"] is False, mutation assert score["pass"] is False, mutation def test_aggregate_recomputes_integrity_and_rejects_forged_minimal_scores(tmp_path: Path) -> None: protocol = frozen_protocol() scores = [score_for_trial(protocol, trial, tmp_path / trial["trial_id"]) for trial in protocol["trials"]] aggregate = protocol_lib.aggregate_trial_scores(protocol, scores) assert aggregate["evaluation_pass"] is True assert aggregate["machine_verifiable_t3_pass"] is False assert aggregate["pass"] is False assert aggregate["current_tier"] == "T2_runtime_artifact_validation" assert aggregate["independent_live_attestation"]["pass"] is False assert aggregate["mean_current_grounded_pass_rate"] == 1.0 assert aggregate["mean_ablation_grounded_pass_rate"] == 0.0 caller_attested = copy.deepcopy(scores) for score in caller_attested: score["independent_live_attestation"] = {"pass": True, "issuer": "caller"} rejected_attestation = protocol_lib.aggregate_trial_scores(protocol, caller_attested) assert rejected_attestation["evaluation_pass"] is False assert rejected_attestation["machine_verifiable_t3_pass"] is False assert rejected_attestation["pass"] is False forged = [ { "trial_id": trial["trial_id"], "pass": True, "grounded_pass_rate": 1.0, "receipt_ablation": {"grounded_pass_rate": 0.0}, "restart_receipt_validation": {"pass": True}, } for trial in protocol["trials"] ] rejected = protocol_lib.aggregate_trial_scores(protocol, forged) assert rejected["pass"] is False assert rejected["checks"]["all_trial_scores_integrity_valid"] is False tampered = copy.deepcopy(scores) tampered[0]["comparison_axis_checks"]["same_model_identity"] = False rejected = protocol_lib.aggregate_trial_scores(protocol, tampered) assert rejected["pass"] is False assert ( rejected["trial_score_integrity"][protocol["trials"][0]["trial_id"]]["checks"]["comparison_axis_checks_passed"] is False ) forged_from_failing_report = copy.deepcopy(scores) report_path = Path(forged_from_failing_report[0]["grounded_report_path"]) failing_report = json.loads(report_path.read_text(encoding="utf-8")) failing_report["results"][0]["reply"] = "This answer is unrelated and ungrounded." report_path.write_text(json.dumps(failing_report, sort_keys=True) + "\n", encoding="utf-8") forged_from_failing_report[0]["grounded_report_sha256"] = hashlib.sha256(report_path.read_bytes()).hexdigest() forged_from_failing_report[0]["grounded_report_payload_sha256"] = protocol_lib.canonical_sha256(failing_report) forged_from_failing_report[0]["derivation_core_sha256"] = protocol_lib.canonical_sha256( protocol_lib.score_derivation_core(forged_from_failing_report[0]) ) rejected = protocol_lib.aggregate_trial_scores(protocol, forged_from_failing_report) first_integrity = rejected["trial_score_integrity"][protocol["trials"][0]["trial_id"]] assert rejected["pass"] is False assert first_integrity["checks"]["grounded_report_artifact_bound"] is True assert first_integrity["checks"]["score_recomputed_from_retained_artifacts"] is False report_path.write_text("{}\n", encoding="utf-8") rejected = protocol_lib.aggregate_trial_scores(protocol, forged_from_failing_report) assert rejected["pass"] is False assert ( rejected["trial_score_integrity"][protocol["trials"][0]["trial_id"]]["checks"]["grounded_report_artifact_bound"] is False )