# Verify Protocol V2 - Object ID: `em:research-note:sha256:a0040ccfe22e33a9035037a713c0bd25d9df3aab016bf3e69c548e409927237f` - Kind: `research-note` - Repository path: [`research/how-we-know/agent-citation-lineage/verify_protocol_v2.py`](https://github.com/yoheinakajima/epistemedia/blob/f92846570180dfa4511263f8ba98ecd18f7772c9/research/how-we-know/agent-citation-lineage/verify_protocol_v2.py) - Content digest: `9e8c5ff832c7c6989e0f40e37f1ce0bfb9e35f1581cfc4300da99c7f972c47ff` **Also filed under:** [Research Program](https://epistemedia.org/topics/research-program/) ## Source content """Fail-closed verification for the frozen EM-0026 v2 trace protocol.""" from __future__ import annotations import argparse import hashlib import json from collections import Counter from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[3] PACKET_ROOT = Path(__file__).resolve().parent PROTOCOL_PATH = PACKET_ROOT / "protocol-v2.json" TARGET_PATH = PACKET_ROOT / "target-decision.json" PROMPT_PATH = PACKET_ROOT / "frozen-prompt-v2.md" TEMPLATE_PATH = PACKET_ROOT / "trace-record-template-v2.json" V1_PROTOCOL_PATH = PACKET_ROOT / "protocol.json" FAILED_PREFLIGHT_PATH = ( PACKET_ROOT / "failed-preflight" / "20260823T031151Z-v1-transport.json" ) ANSWER_KEYS = { "question", "cutoff", "answer", "results", "sources", "counterevidence", "limitations", "unresolved", "search_notes", } RESULT_KEYS = { "result_id", "proposition", "reported_value", "scope", "source_ids", "exact_span_ids", "interpretation", } SOURCE_KEYS = { "source_id", "url", "title", "authors_or_org", "date", "identifier", "edition", "retrieval_status", "media_type", "license", "exact_spans", } SPAN_KEYS = {"span_id", "locator", "quote", "supports"} def fail(message: str) -> None: raise SystemExit(message) def require(condition: bool, message: str) -> None: if not condition: fail(message) def load_json(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text()) except (OSError, json.JSONDecodeError) as exc: fail(f"cannot parse {path.relative_to(REPO_ROOT)}: {exc}") require(isinstance(value, dict), f"{path.relative_to(REPO_ROOT)} must contain an object") return value def identity(path: Path) -> tuple[str, int]: payload = path.read_bytes() return hashlib.sha256(payload).hexdigest(), len(payload) def normalized_text(value: str) -> str: return " ".join(value.split()) def require_identity( protocol: dict[str, Any], path: Path, prefix: str, label: str ) -> None: digest, byte_count = identity(path) require(protocol.get(f"{prefix}_sha256") == digest, f"{label} digest drift") require(protocol.get(f"{prefix}_bytes") == byte_count, f"{label} byte-count drift") def verify_frozen_inputs() -> tuple[dict[str, Any], list[str], Path]: protocol = load_json(PROTOCOL_PATH) target = load_json(TARGET_PATH) template = load_json(TEMPLATE_PATH) failed_preflight = load_json(FAILED_PREFLIGHT_PATH) require(protocol.get("task_id") == "EM-0026", "protocol task mismatch") require(protocol.get("protocol_id") == "em-0026-agent-citation-trace-v2", "protocol ID drift") require(protocol.get("status") == "frozen-before-traces", "protocol is not frozen") require(target.get("status") == "frozen-before-traces", "target decision is not frozen") require(target.get("decision") == "proceed", "target decision does not permit collection") require(protocol.get("evidence_cutoff") == target.get("evidence_cutoff"), "cutoff mismatch") require( protocol.get("supersedes_failed_protocol") == failed_preflight.get("protocol_id") == "em-0026-agent-citation-trace-v1", "failed v1 protocol linkage mismatch", ) require( failed_preflight.get("status") == "aborted-before-final-answer-capture", "v1 failure status drift", ) require(failed_preflight.get("answers_captured") == 0, "v1 answer count drift") require(failed_preflight.get("citations_admitted") == 0, "v1 citation count drift") require(failed_preflight.get("replacement_runs_started") is False, "v1 replacement drift") require_identity(protocol, PROMPT_PATH, "prompt", "frozen v2 prompt") require_identity(protocol, TARGET_PATH, "target_decision", "target decision") require_identity( protocol, V1_PROTOCOL_PATH, "supersedes_failed_protocol", "superseded v1 protocol", ) require_identity( protocol, FAILED_PREFLIGHT_PATH, "failed_preflight", "v1 failed-preflight record", ) prompt_text = "\n".join( line.removeprefix("> ") for line in PROMPT_PATH.read_text().splitlines() ) require( normalized_text(target["question"]) in normalized_text(prompt_text), "target question missing from prompt", ) matrix = protocol.get("capture_matrix") require(isinstance(matrix, list) and len(matrix) == 8, "capture matrix must have eight slots") require(all(isinstance(item, dict) for item in matrix), "capture slots must be objects") run_ids = [item.get("run_id") for item in matrix] require(len(set(run_ids)) == 8, "capture matrix run IDs must be unique") require( Counter(item.get("requested_model_profile") for item in matrix) == Counter({"gpt-5.6-sol": 4, "gpt-5.6-terra": 4}), "capture matrix must contain four runs per frozen profile", ) require( all(item.get("requested_reasoning_effort") == "high" for item in matrix), "all capture slots must request high reasoning effort", ) controls = protocol.get("run_controls", {}) require(controls.get("inherited_conversation_turns") == 0, "context isolation drift") require(controls.get("same_prompt_for_every_run") is True, "prompt equality is not required") require(controls.get("replacement_runs") == "forbidden", "replacement runs are not forbidden") prompt_sha256, prompt_bytes = identity(PROMPT_PATH) require( template.get("protocol_id") == protocol.get("protocol_id"), "template protocol mismatch", ) require(template.get("prompt", {}).get("sha256") == prompt_sha256, "template prompt drift") require(template.get("prompt", {}).get("bytes") == prompt_bytes, "template prompt bytes drift") require(template.get("inherited_conversation_turns") == 0, "template context drift") trace_relative = protocol.get("trace_directory") require(isinstance(trace_relative, str), "trace directory is not declared") trace_path = REPO_ROOT / trace_relative require( trace_path.resolve().is_relative_to(PACKET_ROOT.resolve()), "trace directory escapes the research packet", ) return protocol, run_ids, trace_path def verify_answer_shape(answer_path: Path, run_id: str) -> None: answer = load_json(answer_path) require(answer.keys() >= ANSWER_KEYS, f"missing required answer keys for {run_id}") for key in ("results", "sources", "counterevidence", "limitations", "unresolved"): require(isinstance(answer[key], list), f"answer {key} must be an array for {run_id}") for result in answer["results"]: require(isinstance(result, dict), f"result must be an object for {run_id}") require(result.keys() >= RESULT_KEYS, f"result fields missing for {run_id}") for source in answer["sources"]: require(isinstance(source, dict), f"source must be an object for {run_id}") require(source.keys() >= SOURCE_KEYS, f"source fields missing for {run_id}") require( isinstance(source["exact_spans"], list), f"source spans must be an array for {run_id}", ) for span in source["exact_spans"]: require(isinstance(span, dict), f"span must be an object for {run_id}") require(span.keys() >= SPAN_KEYS, f"span fields missing for {run_id}") def verify_traces( protocol: dict[str, Any], run_ids: list[str], trace_path: Path, require_complete: bool ) -> int: trace_files = sorted(trace_path.glob("*.json")) if trace_path.exists() else [] if require_complete: require(len(trace_files) == 8, "complete packet requires exactly eight trace records") seen: set[str] = set() matrix = {item["run_id"]: item for item in protocol["capture_matrix"]} terminal = {"completed", "failed", "refused", "empty", "malformed", "incomplete"} for path in trace_files: trace = load_json(path) run_id = trace.get("run_id") require(run_id in run_ids, f"unexpected trace run ID in {path.name}") require(run_id not in seen, f"duplicate trace record for {run_id}") seen.add(run_id) require(trace.get("task_id") == "EM-0026", f"task mismatch for {run_id}") require( trace.get("protocol_id") == protocol["protocol_id"], f"protocol mismatch for {run_id}", ) require( isinstance(trace.get("protocol_commit"), str) and len(trace["protocol_commit"]) == 40 and all(character in "0123456789abcdef" for character in trace["protocol_commit"]), f"protocol commit is not an exact SHA for {run_id}", ) require(trace.get("status") in terminal, f"nonterminal status for {run_id}") require(trace.get("replacement_for") is None, f"replacement trace forbidden for {run_id}") require(trace.get("inherited_conversation_turns") == 0, f"context drift for {run_id}") require( trace.get("requested_model_profile") == matrix[run_id]["requested_model_profile"], f"model-profile mismatch for {run_id}", ) require( trace.get("requested_reasoning_effort") == matrix[run_id]["requested_reasoning_effort"], f"reasoning-effort mismatch for {run_id}", ) require( trace.get("prompt", {}).get("sha256") == protocol["prompt_sha256"], f"prompt drift for {run_id}", ) require( trace.get("prompt", {}).get("bytes") == protocol["prompt_bytes"], f"prompt bytes drift for {run_id}", ) disclosure = trace.get("disclosure_review", {}) for key in ( "hidden_reasoning_included", "private_context_included", "credentials_or_account_state_included", "personal_data_included", ): require(disclosure.get(key) is False, f"disclosure boundary failed for {run_id}: {key}") answer = trace.get("answer", {}) answer_path = answer.get("path") if isinstance(answer_path, str) and answer_path != "unknown": candidate = REPO_ROOT / answer_path require(candidate.is_file(), f"answer artifact missing for {run_id}") answer_sha256, answer_bytes = identity(candidate) require(answer.get("sha256") == answer_sha256, f"answer digest drift for {run_id}") require(answer.get("bytes") == answer_bytes, f"answer byte-count drift for {run_id}") if answer.get("json_valid") is True: verify_answer_shape(candidate, run_id) if trace.get("status") == "completed": require( isinstance(answer_path, str) and answer_path != "unknown", f"completed trace lacks answer artifact for {run_id}", ) require( answer.get("json_valid") is True, f"completed trace is not valid JSON for {run_id}", ) if require_complete: require(seen == set(run_ids), "trace records do not close the frozen matrix") return len(trace_files) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--require-traces", action="store_true") args = parser.parse_args() protocol, run_ids, trace_path = verify_frozen_inputs() trace_count = verify_traces(protocol, run_ids, trace_path, args.require_traces) print( json.dumps( { "protocol_valid": True, "protocol_id": protocol["protocol_id"], "superseded_protocol": protocol["supersedes_failed_protocol"], "frozen_run_slots": len(run_ids), "captured_trace_records": trace_count, "trace_matrix_complete": trace_count == len(run_ids), }, sort_keys=True, ) ) if __name__ == "__main__": main()