Verify Packet

Source content

"""Fail-closed verification for the EM-0033 candidate research packet."""

from __future__ import annotations

import argparse

import copy

import hashlib

import json

import re

import subprocess

from datetime import datetime

from pathlib import Path

from typing import Any

from build_packet import (

REQUIRED_LINEAGE_DIMENSIONS,

build_derivations,

build_packet,

canonical_bytes,

)

PACKET_ROOT = Path(__file__).resolve().parent

REPO_ROOT = PACKET_ROOT.parents[2]

SOURCE_RECORDS = PACKET_ROOT / "source-records.json"

CANDIDATE_PACKET = PACKET_ROOT / "candidate-packet.json"

DEFAULT_REVIEW_RECEIPT = PACKET_ROOT / "independent-review-receipt.json"

REPOSITORY_URL = "https://github.com/yoheinakajima/epistemedia"

REVIEW_FORMAT = "epistemedia-independent-research-review-v1"

EXPECTED_PACKET_ID = (

"em:research-packet:sha256:a73dc29f0a0c3f05a112b7c392d115cfcc38a1136a683325bf74bebf0c6b2e40"

)

EXPECTED_SOURCE_RECORDS = {

"bytes": 53857,

"sha256": "0ac4f87236882d32aa6018e50600a79642b87acdf05e27c291514e770fda081c",

}

EXPECTED_CANDIDATE_PACKET = {

"bytes": 65077,

"sha256": "6253a8ece3bb3b5bfb393ec04ad4e0f14b88f567ddea2923300538eadb7c8cde",

}

EXPECTED_SOURCE_IDS = {

"source-argyle-1970",

"source-argyle-1971",

"source-birmingham-events-2020",

"source-hampshire-pcc-2022",

"source-hegstrom-1979",

"source-lapakko-1997",

"source-lapakko-2007",

"source-mehrabian-author-qualification",

"source-mehrabian-ferris-1967",

"source-mehrabian-wiener-1967",

"source-silent-messages-1971",

"source-silent-messages-1981",

}

EXPECTED_CLAIM_IDS = {

"claim-1981-edition",

"claim-book-boundary",

"claim-hegstrom-rebuttal",

"claim-p1-tone-dominance",

"claim-p2-face-tone",

"claim-popular-rule",

"claim-propagation",

"claim-related-context",

"claim-replication-search",

"claim-seven-origin",

"claim-three-coefficient-proposal",

}

EXPECTED_DERIVATION_IDS = {

"derive-implied-vocal-verbal-ratio",

"derive-p2-allocation-with-seven-reserved",

"derive-p2-facial-vocal-ratio",

"derive-proposed-facial-vocal-ratio",

"derive-proposed-sum",

"derive-ratio-difference",

}

EXPECTED_LINEAGE_EDGE_IDS = {

"edge-p1-p2-author-social",

"edge-p1-p2-grant",

"edge-p1-p2-material",

"edge-p1-p2-method",

"edge-p1-p2-participant",

"edge-p1-p2-scale",

"edge-p1-p2-speaker",

"edge-p1-p2-stimulus",

"edge-p1-p2-proposal-derivation",

"edge-p2-cites-p1",

"edge-silent-messages-editions",

}

EXPECTED_PROPAGATION_IDS = {

"propagation-birmingham-events",

"propagation-hampshire-pcc",

"propagation-lapakko-web-sample",

}

def fail(message: str) -> None:

raise SystemExit(message)

def require(condition: bool, message: str) -> None:

if not condition:

fail(message)

def load(path: Path) -> dict[str, Any]:

try:

value = json.loads(path.read_text(encoding="utf-8"))

except (OSError, json.JSONDecodeError) as exc:

fail(f"cannot parse {path}: {exc}")

require(isinstance(value, dict), f"{path}: root must be an object")

return value

def sha256(payload: bytes) -> str:

return hashlib.sha256(payload).hexdigest()

def identity(path: Path) -> dict[str, Any]:

payload = path.read_bytes()

return {"bytes": len(payload), "sha256": sha256(payload)}

def git_bytes(*args: str, check: bool = True) -> bytes:

result = subprocess.run(

["git", *args],

cwd=REPO_ROOT,

check=False,

capture_output=True,

)

if check and result.returncode != 0:

fail(f"git {' '.join(args)} failed: {result.stderr.decode(errors='replace').strip()}")

return result.stdout

def git_text(*args: str, check: bool = True) -> str:

return git_bytes(*args, check=check).decode().strip()

def require_exact_fields(value: Any, fields: set[str], context: str) -> None:

require(isinstance(value, dict), f"{context}: must be an object")

actual = set(value)

require(actual == fields, f"{context}: fields differ: {sorted(actual ^ fields)}")

def require_string(value: Any, context: str) -> str:

require(isinstance(value, str) and value.strip(), f"{context}: must be a string")

return value

def require_sha256(value: Any, context: str) -> str:

require_string(value, context)

require(bool(re.fullmatch(r"[0-9a-f]{64}", value)), f"{context}: invalid SHA-256")

return value

def require_commit(value: Any, context: str) -> str:

require_string(value, context)

require(bool(re.fullmatch(r"[0-9a-f]{40}", value)), f"{context}: invalid Git identity")

return value

def require_timestamp(value: Any, context: str) -> datetime:

require_string(value, context)

try:

return datetime.fromisoformat(value.replace("Z", "+00:00"))

except ValueError as exc:

fail(f"{context}: invalid timestamp: {exc}")

def collect_ids(packet: dict[str, Any]) -> dict[str, set[str]]:

content = packet["content"]

sources = content["source_records"]

return {

"source_ids": {item["source_id"] for item in sources},

"capture_ids": {

capture["capture_id"] for source in sources for capture in source["captures"]

},

"span_ids": {

span["span_id"] for source in sources for span in source.get("spans", [])

},

"claim_ids": {item["claim_id"] for item in content["claims"]},

"derivation_ids": {item["derivation_id"] for item in content["derivations"]},

"lineage_edge_ids": {item["edge_id"] for item in content["lineage_edges"]},

"propagation_ids": {item["object_id"] for item in content["propagation_ledger"]},

"follow_up_source_ids": {item["source_id"] for item in content["follow_up_ledger"]},

}

def close(actual: float, expected: float, tolerance: float = 1e-12) -> bool:

return abs(actual - expected) <= tolerance

def independently_recompute_derivations(packet: dict[str, Any]) -> None:

content = packet["content"]

observed = {item["derivation_id"]: item for item in content["derivations"]}

expected = {

item["derivation_id"]: item for item in build_derivations(content["derivation_inputs"])

}

require(set(observed) == set(expected), "derivation ID drift")

for derivation_id, expected_item in expected.items():

observed_result = observed[derivation_id]["result"]

expected_result = expected_item["result"]

if isinstance(expected_result, dict):

require(

set(observed_result) == set(expected_result),

f"result fields drift: {derivation_id}",

)

for key, value in expected_result.items():

require(close(observed_result[key], value), f"result drift: {derivation_id}.{key}")

else:

require(close(observed_result, expected_result), f"result drift: {derivation_id}")

def verify_packet() -> tuple[dict[str, Any], dict[str, Any]]:

packet = load(CANDIDATE_PACKET)

require(packet == build_packet(), "candidate packet deterministic rebuild drift")

require(packet["packet_id"] == EXPECTED_PACKET_ID, "candidate packet ID drift")

require(identity(SOURCE_RECORDS) == EXPECTED_SOURCE_RECORDS, "source-record identity drift")

require(identity(CANDIDATE_PACKET) == EXPECTED_CANDIDATE_PACKET, "candidate identity drift")

ids = collect_ids(packet)

require(ids["source_ids"] == EXPECTED_SOURCE_IDS, "source set drift")

require(ids["claim_ids"] == EXPECTED_CLAIM_IDS, "claim set drift")

require(ids["derivation_ids"] == EXPECTED_DERIVATION_IDS, "derivation set drift")

require(ids["lineage_edge_ids"] == EXPECTED_LINEAGE_EDGE_IDS, "lineage edge set drift")

require(ids["propagation_ids"] == EXPECTED_PROPAGATION_IDS, "propagation set drift")

content = packet["content"]

counts = content["counts"]

expected_counts = {

"source_records": 12,

"quote_minimal_spans": 40,

"claims": 11,

"derivations": 6,

"lineage_groups": 6,

"lineage_edges": 11,

"propagation_objects": 3,

"follow_up_objects": 5,

"participant_data_roots": 5,

}

require(counts == expected_counts, "packet count drift")

require(

{item["dimension"] for item in content["lineage_edges"]}

== REQUIRED_LINEAGE_DIMENSIONS,

"lineage dimension drift",

)

require(

content["recommendation"] == load(SOURCE_RECORDS)["recommendation"],

"recommendation projection drift",

)

require(content["recommendation"]["author"] == "GO", "author recommendation drift")

require(

content["recommendation"]["independent_review"] == "pending",

"candidate packet self-approved",

)

require(

"93 percent of all communication is nonverbal"

in content["recommendation"]["prohibited_interpretations"],

"general-rule prohibition missing",

)

require(

all(item["scientific_rule_evidence_credit"] == 0 for item in content["propagation_ledger"]),

"propagation evidence-credit drift",

)

require(not list(PACKET_ROOT.glob("*.pdf")), "restricted PDF committed beside packet")

require(not list(PACKET_ROOT.glob("*.html")), "raw HTML committed beside packet")

require(not list(PACKET_ROOT.glob("*.xml")), "raw XML committed beside packet")

independently_recompute_derivations(packet)

return packet, {

"packet_id": packet["packet_id"],

"source_records": identity(SOURCE_RECORDS),

"candidate_packet": identity(CANDIDATE_PACKET),

"ids": ids,

"counts": counts,

"author_recommendation": content["recommendation"]["author"],

}

def require_exact_coverage(actual: Any, expected: set[str], context: str) -> None:

require(isinstance(actual, list), f"{context}: must be an array")

require(actual == sorted(expected), f"{context}: incomplete or unsorted coverage")

def validate_commands(commands: Any) -> None:

require(isinstance(commands, list) and commands, "receipt.commands: must be non-empty")

rendered = []

for index, command in enumerate(commands):

context = f"receipt.commands[{index}]"

require_exact_fields(

command,

{

"argv",

"cwd",

"started_at",

"completed_at",

"exit_code",

"stdout_sha256",

"stderr_sha256",

},

context,

)

argv = command["argv"]

require(

isinstance(argv, list)

and argv

and all(isinstance(item, str) and item for item in argv),

f"{context}.argv: invalid",

)

require_string(command["cwd"], f"{context}.cwd")

started = require_timestamp(command["started_at"], f"{context}.started_at")

completed = require_timestamp(command["completed_at"], f"{context}.completed_at")

require(completed >= started, f"{context}: completion precedes start")

require(command["exit_code"] == 0, f"{context}.exit_code: must be zero")

require_sha256(command["stdout_sha256"], f"{context}.stdout_sha256")

require_sha256(command["stderr_sha256"], f"{context}.stderr_sha256")

rendered.append(" ".join(argv))

required = {

"packet build": lambda value: "build_packet.py" in value and "--check" in value,

"packet verification": lambda value: "verify_packet.py" in value,

"repository check": lambda value: "make check" in value,

}

for label, predicate in required.items():

require(any(predicate(value) for value in rendered), f"receipt.commands: missing {label}")

def validate_review_results(

results: Any,

summary: dict[str, Any],

packet: dict[str, Any],

) -> None:

require_exact_fields(

results,

{"sources", "spans", "derivations", "lineage_edges"},

"review_results",

)

source_records = results["sources"]

require(isinstance(source_records, list), "review_results.sources: must be an array")

require(

len(source_records) == len(summary["ids"]["source_ids"]),

"source result cardinality drift",

)

expected_capture_map = {

source["source_id"]: {capture["capture_id"] for capture in source["captures"]}

for source in packet["content"]["source_records"]

}

source_ids = set()

reviewed_capture_ids = set()

for index, record in enumerate(source_records):

context = f"review_results.sources[{index}]"

require_exact_fields(

record,

{

"source_id",

"capture_ids",

"identity_checked",

"access_and_gap_checked",

"license_checked",

"status",

},

context,

)

source_ids.add(require_string(record["source_id"], f"{context}.source_id"))

require(

record["source_id"] in expected_capture_map,

f"{context}.source_id: unknown",

)

require_exact_coverage(

record["capture_ids"],

expected_capture_map[record["source_id"]],

f"{context}.capture_ids",

)

reviewed_capture_ids.update(record["capture_ids"])

for field in ("identity_checked", "access_and_gap_checked", "license_checked"):

require(record[field] is True, f"{context}.{field}: must be true")

require(

record["status"] in {"pass", "pass-with-declared-gap"},

f"{context}.status: not pass",

)

require(source_ids == summary["ids"]["source_ids"], "source result coverage drift")

require(

reviewed_capture_ids == summary["ids"]["capture_ids"],

"capture result coverage drift",

)

span_records = results["spans"]

require(isinstance(span_records, list), "review_results.spans: must be an array")

require(

len(span_records) == len(summary["ids"]["span_ids"]),

"span result cardinality drift",

)

require(

{item.get("span_id") for item in span_records} == summary["ids"]["span_ids"],

"span result coverage drift",

)

expected_span_digests = {

span["span_id"]: span["quote_sha256"]

for source in packet["content"]["source_records"]

for span in source["spans"]

}

for index, record in enumerate(span_records):

context = f"review_results.spans[{index}]"

require_exact_fields(

record,

{

"span_id",

"verification",

"expected_sha256",

"observed_sha256",

"match",

},

context,

)

require_string(record["verification"], f"{context}.verification")

expected_digest = expected_span_digests[record["span_id"]]

require(

record["expected_sha256"] == expected_digest,

f"{context}.expected_sha256: stale",

)

require(

record["observed_sha256"] == expected_digest,

f"{context}.observed_sha256: mismatch",

)

require(record["match"] is True, f"{context}.match: must be true")

derivation_records = results["derivations"]

require(isinstance(derivation_records, list), "review_results.derivations: must be an array")

require(

len(derivation_records) == len(summary["ids"]["derivation_ids"]),

"derivation result cardinality drift",

)

require(

{item.get("derivation_id") for item in derivation_records}

== summary["ids"]["derivation_ids"],

"derivation result coverage drift",

)

expected_derivation_digests = {

item["derivation_id"]: sha256(canonical_bytes(item["result"]))

for item in packet["content"]["derivations"]

}

for index, record in enumerate(derivation_records):

context = f"review_results.derivations[{index}]"

require_exact_fields(

record,

{

"derivation_id",

"expected_result_sha256",

"observed_result_sha256",

"reproduced",

},

context,

)

expected_digest = expected_derivation_digests[record["derivation_id"]]

require(

record["expected_result_sha256"] == expected_digest,

f"{context}.expected_result_sha256: stale",

)

require(

record["observed_result_sha256"] == expected_digest,

f"{context}.observed_result_sha256: mismatch",

)

require(record["reproduced"] is True, f"{context}.reproduced: must be true")

edge_records = results["lineage_edges"]

require(isinstance(edge_records, list), "review_results.lineage_edges: must be an array")

require(

len(edge_records) == len(summary["ids"]["lineage_edge_ids"]),

"lineage result cardinality drift",

)

require(

{item.get("edge_id") for item in edge_records}

== summary["ids"]["lineage_edge_ids"],

"lineage result coverage drift",

)

expected_edge_spans = {

item["edge_id"]: set(item["evidence_span_ids"])

for item in packet["content"]["lineage_edges"]

}

for index, record in enumerate(edge_records):

context = f"review_results.lineage_edges[{index}]"

require_exact_fields(

record,

{

"edge_id",

"evidence_span_ids",

"evidence_checked",

"independence_effect_checked",

"status",

},

context,

)

require_exact_coverage(

record["evidence_span_ids"],

expected_edge_spans[record["edge_id"]],

f"{context}.evidence_span_ids",

)

require(record["evidence_checked"] is True, f"{context}.evidence_checked: must be true")

require(

record["independence_effect_checked"] is True,

f"{context}.independence_effect_checked: must be true",

)

require(record["status"] == "pass", f"{context}.status: must be pass")

def validate_receipt_document(

receipt: dict[str, Any],

packet: dict[str, Any],

summary: dict[str, Any],

*,

check_git: bool,

) -> None:

require_exact_fields(

receipt,

{

"format",

"task_id",

"reviewer",

"repository",

"started_at",

"completed_at",

"git_state",

"bindings",

"coverage",

"review_results",

"commands",

"findings",

"limitations",

"recommendation",

"decision",

"complete",

},

"receipt",

)

require(receipt["format"] == REVIEW_FORMAT, "receipt.format: unsupported")

require(receipt["task_id"] == "EM-0033", "receipt.task_id: mismatch")

started = require_timestamp(receipt["started_at"], "receipt.started_at")

completed = require_timestamp(receipt["completed_at"], "receipt.completed_at")

require(completed >= started, "receipt: completion precedes start")

reviewer = receipt["reviewer"]

require_exact_fields(

reviewer,

{

"id",

"role",

"fresh_clone",

"reviewer_was_author",

"independent_public_retrieval",

"authoring_notes_used_as_evidence",

"notes",

},

"receipt.reviewer",

)

require(

reviewer["id"] == "codex-independent-em0033-reviewer",

"receipt.reviewer.id: wrong reviewer",

)

require(reviewer["role"] == "independent-reviewer", "receipt.reviewer.role: mismatch")

require(reviewer["fresh_clone"] is True, "receipt.reviewer.fresh_clone: must be true")

require(

reviewer["reviewer_was_author"] is False,

"receipt.reviewer.reviewer_was_author: must be false",

)

require(

reviewer["independent_public_retrieval"] is True,

"receipt.reviewer.independent_public_retrieval: must be true",

)

require(

reviewer["authoring_notes_used_as_evidence"] is False,

"receipt.reviewer.authoring_notes_used_as_evidence: must be false",

)

require_string(reviewer["notes"], "receipt.reviewer.notes")

bindings = receipt["bindings"]

require_exact_fields(bindings, {"packet_id", "source_records", "candidate_packet"}, "bindings")

require(bindings["packet_id"] == summary["packet_id"], "receipt packet ID drift")

require(bindings["source_records"] == summary["source_records"], "receipt source drift")

require(bindings["candidate_packet"] == summary["candidate_packet"], "receipt packet drift")

coverage = receipt["coverage"]

require_exact_fields(

coverage,

{

"source_ids",

"capture_ids",

"span_ids",

"claim_ids",

"derivation_ids",

"lineage_edge_ids",

"propagation_ids",

"follow_up_source_ids",

},

"coverage",

)

for key, expected in summary["ids"].items():

require_exact_coverage(coverage[key], expected, f"receipt.coverage.{key}")

validate_review_results(receipt["review_results"], summary, packet)

require(receipt["recommendation"] == summary["author_recommendation"], "review differs")

require(receipt["decision"] == "pass", "receipt.decision: must be pass")

require(receipt["complete"] is True, "receipt.complete: must be true")

limitations = receipt["limitations"]

require(

isinstance(limitations, list)

and limitations

and all(isinstance(item, str) and item.strip() for item in limitations),

"receipt.limitations: must be non-empty",

)

findings = receipt["findings"]

require(isinstance(findings, list), "receipt.findings: must be an array")

for index, finding in enumerate(findings):

require_exact_fields(finding, {"severity", "status", "text"}, f"findings[{index}]")

require(finding["status"] in {"resolved", "informational"}, "unresolved finding")

require(finding["severity"] in {"material", "minor", "informational"}, "bad severity")

require_string(finding["text"], f"findings[{index}].text")

validate_commands(receipt["commands"])

repository = receipt["repository"]

require_exact_fields(

repository,

{

"url",

"pull_request",

"branch",

"reviewed_base",

"reviewed_author_head",

"reviewed_author_tree",

"diff_sha256",

},

"repository",

)

require(repository["url"] == REPOSITORY_URL, "receipt repository URL mismatch")

require(repository["pull_request"] > 0, "receipt pull request invalid")

require_string(repository["branch"], "receipt.repository.branch")

base = require_commit(repository["reviewed_base"], "receipt.repository.reviewed_base")

head = require_commit(

repository["reviewed_author_head"], "receipt.repository.reviewed_author_head"

)

tree = require_commit(

repository["reviewed_author_tree"], "receipt.repository.reviewed_author_tree"

)

require_sha256(repository["diff_sha256"], "receipt.repository.diff_sha256")

git_state = receipt["git_state"]

require_exact_fields(

git_state,

{

"fresh_clone",

"pre_review_clean",

"post_review_clean",

"unchanged_during_review",

"pre_review_head",

"post_review_head",

},

"git_state",

)

clean_fields = (

"fresh_clone",

"pre_review_clean",

"post_review_clean",

"unchanged_during_review",

)

for field in clean_fields:

require(git_state[field] is True, f"receipt.git_state.{field}: must be true")

require(git_state["pre_review_head"] == head, "pre-review head mismatch")

require(git_state["post_review_head"] == head, "post-review head mismatch")

if not check_git:

return

require(git_text("rev-parse", f"{head}^{{tree}}") == tree, "reviewed tree mismatch")

require(git_text("merge-base", base, head) == base, "reviewed base is not merge-base")

origin_main = git_text("rev-parse", "origin/main")

require(origin_main == base, "reviewed base is stale")

require(git_text("merge-base", origin_main, head) == origin_main, "main not in author head")

require(

subprocess.run(

["git", "merge-base", "--is-ancestor", head, "HEAD"], cwd=REPO_ROOT, check=False

).returncode

== 0,

"reviewed author head is not an ancestor of HEAD",

)

changed_after = set(filter(None, git_text("diff", "--name-only", head, "HEAD").splitlines()))

receipt_rel = str(DEFAULT_REVIEW_RECEIPT.relative_to(REPO_ROOT))

require(changed_after == {receipt_rel}, "candidate changed after independent review")

diff = git_bytes("diff", "--binary", "--full-index", "--no-ext-diff", base, head)

require(sha256(diff) == repository["diff_sha256"], "review diff digest mismatch")

require(not git_text("status", "--porcelain"), "current tracked or untracked state is dirty")

tracked = git_text("ls-files", "--error-unmatch", receipt_rel, check=False)

require(bool(tracked), "independent receipt must be tracked")

def valid_shape_fixture(packet: dict[str, Any], summary: dict[str, Any]) -> dict[str, Any]:

empty_digest = sha256(b"")

now = "2026-08-27T00:00:00Z"

return {

"format": REVIEW_FORMAT,

"task_id": "EM-0033",

"reviewer": {

"id": "codex-independent-em0033-reviewer",

"role": "independent-reviewer",

"fresh_clone": True,

"reviewer_was_author": False,

"independent_public_retrieval": True,

"authoring_notes_used_as_evidence": False,

"notes": "Shape-only adversarial fixture.",

},

"repository": {

"url": REPOSITORY_URL,

"pull_request": 1,

"branch": "fixture",

"reviewed_base": "0" * 40,

"reviewed_author_head": "1" * 40,

"reviewed_author_tree": "2" * 40,

"diff_sha256": empty_digest,

},

"started_at": now,

"completed_at": now,

"git_state": {

"fresh_clone": True,

"pre_review_clean": True,

"post_review_clean": True,

"unchanged_during_review": True,

"pre_review_head": "1" * 40,

"post_review_head": "1" * 40,

},

"bindings": {

"packet_id": summary["packet_id"],

"source_records": summary["source_records"],

"candidate_packet": summary["candidate_packet"],

},

"coverage": {key: sorted(value) for key, value in summary["ids"].items()},

"review_results": {

"sources": [

{

"source_id": source["source_id"],

"capture_ids": sorted(item["capture_id"] for item in source["captures"]),

"identity_checked": True,

"access_and_gap_checked": True,

"license_checked": True,

"status": "pass",

}

for source in packet["content"]["source_records"]

],

"spans": [

{

"span_id": span["span_id"],

"verification": "fixture",

"expected_sha256": span["quote_sha256"],

"observed_sha256": span["quote_sha256"],

"match": True,

}

for source in packet["content"]["source_records"]

for span in source["spans"]

],

"derivations": [

{

"derivation_id": item["derivation_id"],

"expected_result_sha256": sha256(canonical_bytes(item["result"])),

"observed_result_sha256": sha256(canonical_bytes(item["result"])),

"reproduced": True,

}

for item in packet["content"]["derivations"]

],

"lineage_edges": [

{

"edge_id": edge["edge_id"],

"evidence_span_ids": sorted(edge["evidence_span_ids"]),

"evidence_checked": True,

"independence_effect_checked": True,

"status": "pass",

}

for edge in packet["content"]["lineage_edges"]

],

},

"commands": [

{

"argv": ["python", "build_packet.py", "--check"],

"cwd": ".",

"started_at": now,

"completed_at": now,

"exit_code": 0,

"stdout_sha256": empty_digest,

"stderr_sha256": empty_digest,

},

{

"argv": ["python", "verify_packet.py"],

"cwd": ".",

"started_at": now,

"completed_at": now,

"exit_code": 0,

"stdout_sha256": empty_digest,

"stderr_sha256": empty_digest,

},

{

"argv": ["make", "check"],

"cwd": ".",

"started_at": now,

"completed_at": now,

"exit_code": 0,

"stdout_sha256": empty_digest,

"stderr_sha256": empty_digest,

},

],

"findings": [],

"limitations": ["Shape-only fixture; no empirical pass is inferred."],

"recommendation": summary["author_recommendation"],

"decision": "pass",

"complete": True,

}

def run_adversarial_self_test(packet: dict[str, Any], summary: dict[str, Any]) -> None:

fixture = valid_shape_fixture(packet, summary)

validate_receipt_document(fixture, packet, summary, check_git=False)

def swap_source_captures(value: dict[str, Any]) -> None:

first = value["review_results"]["sources"][0]

second = value["review_results"]["sources"][1]

first_captures = list(first["capture_ids"])

first["capture_ids"] = list(second["capture_ids"])

second["capture_ids"] = first_captures

mutations = [

lambda value: value["reviewer"].update({"reviewer_was_author": True}),

lambda value: value["reviewer"].update({"authoring_notes_used_as_evidence": True}),

lambda value: value["bindings"].update({"packet_id": "stale"}),

lambda value: value["coverage"]["span_ids"].pop(),

lambda value: value["review_results"]["spans"].pop(),

lambda value: value["review_results"]["spans"].append(

copy.deepcopy(value["review_results"]["spans"][0])

),

lambda value: value["review_results"]["spans"][0].update(

{"observed_sha256": "0" * 64}

),

swap_source_captures,

lambda value: value["review_results"]["lineage_edges"][0].update(

{"evidence_checked": False}

),

lambda value: value["review_results"]["lineage_edges"][0][

"evidence_span_ids"

].pop(),

lambda value: value.update({"limitations": []}),

lambda value: value.update({"complete": False}),

lambda value: value["commands"].pop(),

lambda value: value.update(

{"findings": [{"severity": "material", "status": "open", "text": "gap"}]}

),

]

for index, mutate in enumerate(mutations):

forged = copy.deepcopy(fixture)

mutate(forged)

try:

validate_receipt_document(forged, packet, summary, check_git=False)

except SystemExit:

continue

fail(f"adversarial receipt mutation {index} was accepted")

def main() -> None:

parser = argparse.ArgumentParser()

parser.add_argument("--review-receipt", type=Path, default=DEFAULT_REVIEW_RECEIPT)

parser.add_argument("--require-review", action="store_true")

parser.add_argument("--self-test", action="store_true")

args = parser.parse_args()

packet, summary = verify_packet()

if args.self_test:

run_adversarial_self_test(packet, summary)

summary["adversarial_receipt_tests"] = "passed"

review = None

if args.review_receipt.is_file() or args.require_review:

require(args.review_receipt.is_file(), "independent review receipt missing")

review = load(args.review_receipt)

validate_receipt_document(review, packet, summary, check_git=True)

summary["independent_review_complete"] = review is not None

if review is not None:

summary["reviewer"] = review["reviewer"]["id"]

summary["review_decision"] = review["decision"]

summary.pop("ids")

print(json.dumps(summary, indent=2, sort_keys=True))

if __name__ == "__main__":

main()