1409 lines
62 KiB
Python
1409 lines
62 KiB
Python
"""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", "terminal"],
|
|
"actual_registry_tools": ["skill_view", "skills_list", "terminal"],
|
|
"read_only_bridge_commands": ["status"] if mode == "read_only_kb" else [],
|
|
"send_message_tool_enabled": False,
|
|
"terminal_restricted_to_exact_wrapper": True,
|
|
"mutating_bridge_commands_exposed": False,
|
|
"provider_credentials_forwarded_to_terminal": 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_command_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"),
|
|
"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 context" not in item["message"] for item in retrieval_prompts)
|
|
assert all("--limit" 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_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
|
|
)
|