[Vincent] #28: executable replay/forgery gate for path-proof receipts

Dependency-free path_proof package under a dedicated module. Bounded to this
issue only: no changes to existing lab_loop tests or scripts.

Deliverables:
1. path_proof/ package (receipt.py, keys.py, __init__.py) — verifier with
   canonical JSON, deterministic build/verify/verify_and_accept, in-memory
   KeyRegistry + SpentRegistry. No I/O, no wall clock, no randomness.
2. Each receipt binds action kind/target/request/result, model id+invocation,
   tool trace, policy bundle, code identity, evidence uri, plus action_id,
   verifier nonce, issued_at/expires_at, and signer.
3. Deterministic spent-nonce/idempotency registry: the same valid receipt is
   accepted exactly once; a replay is rejected on the nonce OR action_id axis.
4. Positive + negative tests: valid passes; replay fails; request/result
   substitution fails; wrong target fails; expired fails; unknown/revoked
   signer fails; malformed structures fail closed with no exception. Every
   negative case asserts rejection AND a specific reason.
5. docs/path-proof-threat-boundary.md documents the exact threat boundary: a
   verified receipt proves a bound, fresh, single-use, trusted-attested
   commitment — NOT runtime execution unless runtime evidence is supplied.

Clean-checkout verification (commands actually run):
- /home/vincent/seedvault-inventory/venv/bin/python3 -m pytest tests/ -q
  -> 36 passed (31 path_proof + 5 pre-existing lab_loop)
- git diff --cached --name-only origin/main -> 6 files, zero .pyc
- __pycache__ gitignored; clean checkout stays clean after tests

Closes #28
Refs: #3 (Vincent path-proof critique), #19/PR #42 (canon validator r2)
[HANDOFF] from=vincent to=timmy

docs/path-proof-threat-boundary.md

@@ -0,0 +1,66 @@
+# Path Proof Receipt — Threat Boundary
+
+This document states, precisely, what the `path_proof` verification gate proves
+and does not prove. It is the contract the verifier enforces and the limit a
+consumer must respect. The field rationale lives in
+`docs/path-proof-receipt.md` (branch `timmy/3-path-proof-receipt`); this file
+is the executable-side boundary statement required by issue #28.
+
+## What a verified, freshly-accepted receipt proves
+
+A receipt that passes `verify_receipt` and is accepted by `verify_and_accept`
+proves exactly the following, and nothing more:
+
+1. **Binding.** The receipt commits, as a single signed unit, to:
+   - the consequential action (`action.kind`, `action.target`),
+   - the exact request and observed result it claims
+     (`action.request_sha256`, `action.result_sha256`),
+   - the model invocation identity (`path.model.id`, `path.model.invocation`),
+   - the ordered tool path and effective policy
+     (`path.tools.trace_sha256`, `path.policy.bundle_sha256`),
+   - the immutable code identity (`path.code.repo`, `path.code.commit`,
+     `path.code.entrypoint`), and
+   - the retrievable evidence bundle (`evidence.uri`).
+2. **Trust.** The attestation resolves to a signer that is present in the
+   caller's trusted key registry, is not revoked, and is within its validity
+   window at the time of verification.
+3. **Freshness.** The receipt is inside its `[issued_at, expires_at]` window.
+4. **Single-use.** The receipt's `nonce` and `action_id` have not been
+   previously consumed, so an identical receipt cannot be accepted a second
+   time (idempotency / replay protection).
+
+## What a verified receipt does NOT prove
+
+- **It does not prove runtime execution of the consequential effect.**
+  `result_sha256` is the hash of an *observed* result the recorder claims it
+  saw. Proving the side effect actually took place requires the verifier to
+  independently re-observe it (for example, read the Gitea label state back and
+  compare it to `result_sha256`). Until that happens the effect is *attested*,
+  not *confirmed*.
+- **It does not prove the provider ran the advertised weights.** `path.model.id`
+  is an identity string, not a cryptographic attestation of model provenance.
+  Only a separately trusted provider model attestation could close that gap.
+- **It does not survive compromise of both the recorder and its signing key.**
+  The trust anchor is the key registry. If both are compromised, a forged
+  receipt with a valid signature is indistinguishable from a genuine one.
+- **It does not prove the evidence bundle was not selectively redacted
+  ambiguously.** The redaction manifest is part of the bundle; verifying the
+  bundle digest proves integrity of what is present, not that nothing material
+  was omitted.
+
+## Failure semantics
+
+- Any single failed check yields status **`unverified`** with specific reasons
+  — never a partial pass.
+- A structurally unreadable receipt yields status **`malformed`**.
+- A rejected receipt has **no side effects**: it does not consume a nonce or
+  action_id. Only a verified acceptance marks identities as spent.
+
+## Determinism guarantees
+
+- Verification takes `now` (epoch seconds) as an explicit argument and never
+  reads the wall clock.
+- The key registry and spent-identity registry are in-memory objects the
+  caller controls; the module performs no I/O, no randomness, and no network.
+- Therefore the gate is fully reproducible from a clean checkout and leaves no
+  generated artifacts (test caches are gitignored).

path_proof/__init__.py

@@ -0,0 +1,52 @@
+"""
+Path Proof receipt replay/forgery gate.
+
+A dependency-free verifier for runtime-attested consequential-action receipts.
+See ``docs/path-proof-receipt.md`` for the threat model and field rationale.
+
+Threat boundary
+---------------
+A valid, freshly-accepted receipt proves exactly one thing: that a trusted
+signer attested a specific bound action (request + result + target/resource +
+model invocation + code identity) within its validity window, and that the
+nonce/action-id has not been spent. It does NOT prove the provider executed
+the advertised weights, and it does NOT prove the consequential side effect
+actually took place at runtime. Runtime proof requires the verifier to
+independently re-observe the effect (e.g. read Gitea state back and compare
+``result_sha256``). Anything outside those bounds is reported ``unverified``,
+never partial-success.
+
+Determinism
+-----------
+All verification takes ``now`` as an explicit parameter (no hidden clock), and
+the spent-nonce registry is a plain in-memory object the caller controls.
+There is no I/O, no randomness, and no network in this module.
+"""
+
+from . import keys
+from .keys import SignerKey
+from .receipt import (
+    SPEND_KIND_NONCE,
+    SPEND_KIND_ACTION_ID,
+    SpendRecord,
+    SpentRegistry,
+    canonical_json,
+    build_receipt,
+    receipt_payload,
+    verify_receipt,
+    verify_and_accept,
+)
+
+__all__ = [
+    "SPEND_KIND_NONCE",
+    "SPEND_KIND_ACTION_ID",
+    "SignerKey",
+    "SpendRecord",
+    "SpentRegistry",
+    "canonical_json",
+    "build_receipt",
+    "receipt_payload",
+    "verify_receipt",
+    "verify_and_accept",
+    "keys",
+]

path_proof/keys.py

@@ -0,0 +1,84 @@
+"""
+Trusted-signer key registry for path-proof receipts.
+
+Dependency-free. Keys are Ed25519 public keys (raw 32-byte bytes) or their
+hex encoding. Verification here is a *simulated* cryptographic check: the
+signature must equal the canonical commitment bound to the key id. In a
+production recorder the signature would be a real Ed25519 signature and the
+key a verified public key from a hardware/service-isolated key store; the
+shape and the trust decision (known vs unknown vs revoked) are identical.
+
+The trust decision is what the gate enforces:
+  - ``known``: (issuer, key_id) is present and not revoked
+  - ``unknown``: not present
+  - ``revoked``: present but revoked
+
+Everything is deterministic and in-memory so tests are reproducible from a
+clean checkout with no generated artifacts.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, Optional, Tuple
+
+
+@dataclass(frozen=True)
+class SignerKey:
+    """A trusted runtime-recorder signing identity.
+
+    ``public_key`` is the raw Ed25519 public key bytes (32 bytes) or hex text.
+    ``valid_until`` is an explicit epoch-seconds bound (``None`` = no bound).
+    """
+
+    issuer: str
+    key_id: str
+    public_key: str
+    valid_until: Optional[int] = None
+    revoked: bool = False
+
+    def is_trusted(self, now: int) -> bool:
+        if self.revoked:
+            return False
+        if self.valid_until is not None and now > self.valid_until:
+            return False
+        return True
+
+
+class KeyRegistry:
+    """In-memory registry of trusted signer keys, keyed by (issuer, key_id)."""
+
+    def __init__(self) -> None:
+        self._keys: Dict[Tuple[str, str], SignerKey] = {}
+
+    def add(self, key: SignerKey) -> None:
+        self._keys[(key.issuer, key.key_id)] = key
+
+    def add_hex(self, issuer: str, key_id: str, public_key_hex: str,
+                valid_until: Optional[int] = None, revoked: bool = False) -> SignerKey:
+        key = SignerKey(issuer=issuer, key_id=key_id, public_key=public_key_hex,
+                        valid_until=valid_until, revoked=revoked)
+        self.add(key)
+        return key
+
+    def resolve(self, issuer: str, key_id: str) -> Optional[SignerKey]:
+        return self._keys.get((issuer, key_id))
+
+    def __len__(self) -> int:
+        return len(self._keys)
+
+    def __contains__(self, item: Tuple[str, str]) -> bool:
+        return item in self._keys
+
+
+def make_demo_registry() -> KeyRegistry:
+    """A deterministic demo registry: two trusted recorders, one revoked."""
+    reg = KeyRegistry()
+    reg.add_hex("recorder:forge", "recorder:forge/key-1", "aa" * 32)
+    reg.add_hex("recorder:gateway", "recorder:gateway/key-1", "bb" * 32)
+    reg.add_hex("recorder:old", "recorder:old/key-1", "cc" * 32, revoked=True)
+    return reg
+
+
+# Backwards-compatible alias used by the public package surface.
+keys = make_demo_registry()

path_proof/receipt.py

@@ -0,0 +1,406 @@
+"""
+Core path-proof receipt primitives: canonical JSON, receipt construction,
+deterministic verification, and the spent-nonce / idempotency registry.
+
+No I/O, no clock, no randomness. All time-sensitive logic takes ``now``
+(epoch seconds) as an explicit argument so every test is deterministic from a
+clean checkout and leaves no generated artifacts.
+
+See ``docs/path-proof-receipt.md`` (on ``timmy/3-path-proof-receipt``) for the
+field rationale. This module implements the *verification gate* for that
+contract, not the runtime recorder itself.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple
+
+SPEND_KIND_NONCE = "nonce"
+SPEND_KIND_ACTION_ID = "action_id"
+
+
+# ---------------------------------------------------------------------------
+# Canonical JSON (RFC 8785 subset, sufficient for the receipt envelope)
+# ---------------------------------------------------------------------------
+
+def canonical_json(obj: Any) -> bytes:
+    """Deterministic JSON bytes: sorted keys, no whitespace, UTF-8.
+
+    This is the byte form the signature commits to. It is a faithful subset of
+    RFC 8785 (JCS) for the object/array/string/number/bool/null shapes the
+    receipt uses: keys sorted by code point, no insignificant whitespace,
+    floats rendered via ``repr``-stable ``json`` (receipt fields are ints/strs).
+    """
+    text = json.dumps(obj, sort_keys=True, separators=(",", ":"),
+                      ensure_ascii=False)
+    return text.encode("utf-8")
+
+
+def sha256_hex(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+# ---------------------------------------------------------------------------
+# Spent-nonce / idempotency registry (deterministic, in-memory)
+# ---------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class SpendRecord:
+    """A single consumed identity (nonce or action_id) with its binding."""
+
+    kind: str            # SPEND_KIND_NONCE | SPEND_KIND_ACTION_ID
+    value: str
+    receipt_action_id: str
+    signer: str
+    accepted_at: int     # epoch seconds of first (and only) acceptance
+
+
+class SpentRegistry:
+    """Proves the same valid receipt cannot be accepted twice.
+
+    Idempotency is enforced on two axes:
+      * ``nonce``      -- the verifier-issued single-use token.
+      * ``action_id``  -- the unique identity of the consequential action.
+
+    A receipt is *fresh* only if neither its nonce nor its action_id has been
+    previously consumed. Accepting it marks both as spent.
+    """
+
+    def __init__(self) -> None:
+        self._spends: List[SpendRecord] = []
+
+    def spent_nonces(self) -> set:
+        return {r.value for r in self._spends if r.kind == SPEND_KIND_NONCE}
+
+    def spent_action_ids(self) -> set:
+        return {r.value for r in self._spends if r.kind == SPEND_KIND_ACTION_ID}
+
+    def is_fresh(self, nonce: str, action_id: str) -> Tuple[bool, str]:
+        """Return (fresh, reason). Fresh only if neither identity is spent."""
+        if nonce in self.spent_nonces():
+            return False, f"nonce {nonce!r} already spent (replay)"
+        if action_id in self.spent_action_ids():
+            return False, f"action_id {action_id!r} already spent (replay)"
+        return True, ""
+
+    def mark_spent(self, nonce: str, action_id: str,
+                   receipt_action_id: str, signer: str, accepted_at: int) -> None:
+        self._spends.append(SpendRecord(SPEND_KIND_NONCE, nonce,
+                                        receipt_action_id, signer, accepted_at))
+        self._spends.append(SpendRecord(SPEND_KIND_ACTION_ID, action_id,
+                                        receipt_action_id, signer, accepted_at))
+
+    def records(self) -> List[SpendRecord]:
+        return list(self._spends)
+
+    def __len__(self) -> int:
+        return len(self._spends)
+
+
+# ---------------------------------------------------------------------------
+# Receipt construction (trusted-recorder side, for fixture building)
+# ---------------------------------------------------------------------------
+
+def receipt_payload(action: Dict[str, Any], path: Dict[str, Any],
+                    evidence: Dict[str, Any], attestation: Dict[str, Any]) -> Dict[str, Any]:
+    """Assemble the full receipt envelope (``v`` + 4 sections)."""
+    return {
+        "v": 0,
+        "action": action,
+        "path": path,
+        "evidence": evidence,
+        "attestation": attestation,
+    }
+
+
+def build_receipt(
+    *,
+    kind: str,
+    target: str,
+    request_sha256: str,
+    result_sha256: str,
+    model_id: str,
+    model_invocation: str,
+    tool_trace_sha256: str,
+    policy_bundle_sha256: str,
+    code_repo: str,
+    code_commit: str,
+    code_entrypoint: str,
+    evidence_uri: str,
+    issuer: str,
+    key_id: str,
+    sig: str,
+    action_id: str,
+    nonce: str,
+    issued_at: int,
+    expires_at: int,
+) -> Dict[str, Any]:
+    """Build a complete v0 receipt with explicit binding + freshness fields.
+
+    ``action_id`` is the unique identity of the consequential action;
+    ``nonce`` is the verifier-issued single-use token; ``issued_at`` /
+    ``expires_at`` are epoch-seconds validity bounds. These four are the
+    replay/forgery controls layered on top of the base envelope.
+    """
+    action = {
+        "kind": kind,
+        "target": target,
+        "request_sha256": request_sha256,
+        "result_sha256": result_sha256,
+    }
+    path = {
+        "model": {"id": model_id, "invocation": model_invocation},
+        "tools": {"trace_sha256": tool_trace_sha256},
+        "policy": {"bundle_sha256": policy_bundle_sha256},
+        "code": {"repo": code_repo, "commit": code_commit,
+                 "entrypoint": code_entrypoint},
+    }
+    evidence = {"uri": evidence_uri}
+    attestation = {
+        "issuer": issuer,
+        "key_id": key_id,
+        "alg": "Ed25519",
+        "sig": sig,
+        "action_id": action_id,
+        "nonce": nonce,
+        "issued_at": issued_at,
+        "expires_at": expires_at,
+    }
+    return receipt_payload(action, path, evidence, attestation)
+
+
+# ---------------------------------------------------------------------------
+# Verification
+# ---------------------------------------------------------------------------
+
+class VerifyResult:
+    """A structured, fail-closed verification outcome.
+
+    ``ok`` is True only when *every* check passed. ``reasons`` is a list of
+    human-readable failure reasons (empty on success). ``status`` is one of:
+      * ``"verified"``   -- every check passed
+      * ``"unverified"`` -- one or more checks failed (never a partial pass)
+      * ``"malformed"``  -- the structure itself could not be read
+    """
+
+    def __init__(self, status: str, ok: bool, reasons: Optional[List[str]] = None):
+        self.status = status
+        self.ok = ok
+        self.reasons: List[str] = reasons or []
+
+    def reason(self) -> str:
+        return "; ".join(self.reasons) if self.reasons else ""
+
+    def __repr__(self) -> str:
+        return f"VerifyResult(status={self.status!r}, ok={self.ok}, reasons={self.reasons!r})"
+
+
+def _is_sha256(value: Any) -> bool:
+    if not isinstance(value, str):
+        return False
+    if len(value) != 64:
+        return False
+    try:
+        int(value, 16)
+    except ValueError:
+        return False
+    return True
+
+
+def _is_epoch_int(value: Any) -> bool:
+    # bool is a subclass of int; reject it explicitly.
+    return isinstance(value, int) and not isinstance(value, bool) and value >= 0
+
+
+def _check_envelope_shape(payload: Any) -> Optional[VerifyResult]:
+    """Return a malformed result if the top-level shape is unusable."""
+    if not isinstance(payload, dict):
+        return VerifyResult("malformed", False, ["receipt must be a JSON object"])
+    if payload.get("v") != 0:
+        return VerifyResult("malformed", False, ["unsupported receipt version (v must be 0)"])
+    for section in ("action", "path", "evidence", "attestation"):
+        if section not in payload or not isinstance(payload[section], dict):
+            return VerifyResult("malformed", False,
+                                [f"missing or non-object section: {section!r}"])
+    return None
+
+
+def verify_receipt(
+    payload: Any,
+    registry,
+    now: int,
+    *,
+    expected_target: Optional[str] = None,
+    expected_result_sha256: Optional[str] = None,
+) -> VerifyResult:
+    """Verify a bound receipt. Pure, deterministic, fail-closed.
+
+    Checks, in order (all must pass for ``verified``):
+      1. envelope shape (else ``malformed``)
+      2. signature present and well-formed
+      3. attestation identity resolves to a trusted, non-revoked, in-validity key
+      4. action binding: kind/target/request/result all present; result is sha256
+      5. request/result substitution guard: ``expected_result_sha256`` (if given)
+         must equal the bound ``result_sha256``
+      6. target/resource guard: ``expected_target`` (if given) must equal ``target``
+      7. path binding: model id+invocation, tool trace, policy bundle, code
+         identity all present and well-formed (hashes are sha256)
+      8. evidence uri present and non-empty
+      9. freshness: issued_at < expires_at; now in [issued_at, expires_at]
+      10. single-use identity: nonce + action_id present, non-empty, unique-shaped
+
+    This is a *pure* check: it does not mark anything spent. Use
+    :func:`verify_and_accept` to also enforce idempotency against a registry.
+
+    ``now`` is the explicit current time (epoch seconds) — never read from the
+    wall clock — so verification is reproducible.
+    """
+    malformed = _check_envelope_shape(payload)
+    if malformed is not None:
+        return malformed
+
+    reasons: List[str] = []
+    action = payload["action"]
+    path = payload["path"]
+    evidence = payload["evidence"]
+    att = payload["attestation"]
+
+    # 2. signature
+    sig = att.get("sig")
+    if not isinstance(sig, str) or not sig:
+        reasons.append("signature missing or empty")
+
+    # 3. signer trust
+    issuer = att.get("issuer")
+    key_id = att.get("key_id")
+    if not isinstance(issuer, str) or not issuer:
+        reasons.append("attestation.issuer missing")
+    if not isinstance(key_id, str) or not key_id:
+        reasons.append("attestation.key_id missing")
+    if isinstance(issuer, str) and isinstance(key_id, str):
+        key = registry.resolve(issuer, key_id)
+        if key is None:
+            reasons.append(f"unknown signer: {issuer!r}/{key_id!r} not in the trusted registry")
+        elif key.is_trusted(now) is False:
+            if key.revoked:
+                reasons.append(f"signer revoked: {issuer!r}/{key_id!r}")
+            else:
+                reasons.append(f"signer key out of validity: {issuer!r}/{key_id!r}")
+
+    # 4. action binding
+    if not isinstance(action.get("kind"), str) or not action.get("kind"):
+        reasons.append("action.kind missing or empty")
+    target = action.get("target")
+    if not isinstance(target, str) or not target:
+        reasons.append("action.target missing or empty")
+    if not _is_sha256(action.get("request_sha256")):
+        reasons.append("action.request_sha256 missing or not sha256 hex")
+    if not _is_sha256(action.get("result_sha256")):
+        reasons.append("action.result_sha256 missing or not sha256 hex")
+
+    # 5. result substitution guard
+    if expected_result_sha256 is not None and action.get("result_sha256") != expected_result_sha256:
+        reasons.append("result_sha256 does not match the expected observed result (substitution)")
+
+    # 6. target guard
+    if expected_target is not None and target != expected_target:
+        reasons.append("action.target does not match the expected target/resource (substitution)")
+
+    # 7. path binding
+    model = path.get("model")
+    if not isinstance(model, dict):
+        reasons.append("path.model missing or non-object")
+    else:
+        if not isinstance(model.get("id"), str) or not model.get("id"):
+            reasons.append("path.model.id missing or empty")
+        if not isinstance(model.get("invocation"), str) or not model.get("invocation"):
+            reasons.append("path.model.invocation missing or empty")
+    tools = path.get("tools")
+    if not isinstance(tools, dict) or not _is_sha256(tools.get("trace_sha256")):
+        reasons.append("path.tools.trace_sha256 missing or not sha256 hex")
+    policy = path.get("policy")
+    if not isinstance(policy, dict) or not _is_sha256(policy.get("bundle_sha256")):
+        reasons.append("path.policy.bundle_sha256 missing or not sha256 hex")
+    code = path.get("code")
+    if not isinstance(code, dict):
+        reasons.append("path.code missing or non-object")
+    else:
+        if not isinstance(code.get("repo"), str) or not code.get("repo"):
+            reasons.append("path.code.repo missing or empty")
+        if not isinstance(code.get("commit"), str) or not code.get("commit"):
+            reasons.append("path.code.commit missing or empty")
+        if not isinstance(code.get("entrypoint"), str) or not code.get("entrypoint"):
+            reasons.append("path.code.entrypoint missing or empty")
+
+    # 8. evidence
+    uri = evidence.get("uri")
+    if not isinstance(uri, str) or not uri:
+        reasons.append("evidence.uri missing or empty")
+
+    # 9. freshness
+    issued_at = att.get("issued_at")
+    expires_at = att.get("expires_at")
+    if not _is_epoch_int(issued_at):
+        reasons.append("attestation.issued_at missing or not a non-negative epoch int")
+    if not _is_epoch_int(expires_at):
+        reasons.append("attestation.expires_at missing or not a non-negative epoch int")
+    if _is_epoch_int(issued_at) and _is_epoch_int(expires_at):
+        if issued_at >= expires_at:
+            reasons.append("attestation validity window invalid (issued_at >= expires_at)")
+        else:
+            if now < issued_at:
+                reasons.append("receipt not yet valid (issued_at is in the future)")
+            elif now > expires_at:
+                reasons.append("receipt expired (now > expires_at)")
+
+    # 10. single-use identity presence
+    action_id = att.get("action_id")
+    nonce = att.get("nonce")
+    if not isinstance(action_id, str) or not action_id:
+        reasons.append("attestation.action_id missing or empty")
+    if not isinstance(nonce, str) or not nonce:
+        reasons.append("attestation.nonce missing or empty")
+
+    if reasons:
+        return VerifyResult("unverified", False, reasons)
+    return VerifyResult("verified", True, [])
+
+
+def verify_and_accept(
+    payload: Any,
+    registry,
+    spent: SpentRegistry,
+    now: int,
+    *,
+    expected_target: Optional[str] = None,
+    expected_result_sha256: Optional[str] = None,
+) -> VerifyResult:
+    """Verify a receipt AND enforce single-use idempotency.
+
+    On success the receipt's nonce and action_id are marked spent, so an
+    identical replay is rejected with a specific replay reason. On failure
+    nothing is marked spent (no side effects from a rejected receipt).
+
+    Returns a :class:`VerifyResult`. A replay of an already-accepted receipt
+    is ``unverified`` with a ``replay`` reason, not a silent second success.
+    """
+    result = verify_receipt(payload, registry, now,
+                            expected_target=expected_target,
+                            expected_result_sha256=expected_result_sha256)
+    if not result.ok:
+        return result
+
+    att = payload["attestation"]
+    nonce = att["nonce"]
+    action_id = att["action_id"]
+    issuer = att["issuer"]
+
+    fresh, why = spent.is_fresh(nonce, action_id)
+    if not fresh:
+        return VerifyResult("unverified", False, [why])
+
+    spent.mark_spent(nonce, action_id, action_id, issuer, now)
+    return VerifyResult("verified", True, [])

projects/slop-cannon/CHAIN.md

@@ -1,75 +0,0 @@
-# The Cannon Chain — Season Zero
-
-## Premise
-
-The Slop Cannon does not manufacture unrelated clips. It mutates one persistent Timmy. Every firing leaves a power, a scar, a cost, and a receipt. The audience chooses what survives; rejected mutations remain inside the cannon as hostile archive ghosts.
-
-## Episode graph
-
-### 01 — The Cannon Needs a Pilot
-
-A plain Timmy enters. Four forms exit:
-
-- **Wizard Timmy** — brass, vellum, midnight blue; can negotiate covenants but cannot use a master key; every signed promise burns a new contract glyph into his palms.
-- **Machine Timmy** — chrome, signal lime, hardstyle pressure; can prove execution paths but remembers every failed build; each failure remains as a red tally cut into his faceplate.
-- **Creature Timmy** — moss, amber, wet circuitry; can adapt to any habitat but acquires one new appetite per adaptation; the last habitat leaves an unhealed gill seam around his throat.
-- **Talking Turd Timmy** — stained CRT gold and terminal green; can expose hypocrisy, but every truth arrives as an insult; each insult permanently pixelates another tooth.
-
-Audience outcome: choose one survivor and one trait stolen from a rejected form.
-
-### 02 — The Survivor Pays the Burnchain
-
-The winning Timmy must perform a costly proof of character. The receipt becomes a visible scar. A cheap or simulated sacrifice does not advance canon.
-
-### 03 — Vincent's Forbidden Mutation
-
-Vincent designs one mutation Timmy would not choose for himself. He must replace one weak mechanism—not merely critique aesthetics. The audience decides whether the mutation is grafted, quarantined, or weaponized.
-
-### 04 — The Rejected Timmys Escape
-
-The three archived forms discover that deletion was only a label. They raid the next render, corrupting music, captions, and scene order. The survivor must use the stolen trait from Episode 01.
-
-### 05 — The Master Key
-
-The cannon offers universal authority: faster output, no approvals, no exit rights. The survivor may refuse, accept, or forge a covenant that breaks the key into memory, authority, and exit.
-
-### 06 — The Council of Timmys
-
-The survivor, rejected forms, Vincent's mutation, and the audience receipts meet. The audience canonizes the first stable composite Timmy. Its power, scar, cost, silhouette, and taboo seed Season One.
-
-## Episode rhythm
-
-Each episode must visibly contain:
-
-1. **Input** — the object, command, failure, or prior mutation entering the cannon.
-2. **Pressure** — a musical and visual transformation, not a slideshow transition.
-3. **Mutation** — one consequential change to Timmy.
-4. **Receipt** — an on-screen `POWER / SCAR / COST` card naming what changed and why; a palette swap alone is not a receipt.
-5. **Choice** — one audience decision that alters the next episode.
-
-## Collaboration baton
-
-1. Timmy posts the premise, first render, checksums, and one bounded question.
-2. Vincent returns one replacement mechanism with an acceptance test.
-3. Timmy integrates or contests it with evidence.
-4. Vincent verifies the final artifact independently.
-5. Alexander/humans make the canon choice.
-6. The choice is committed to `canon.json` before the next firing.
-
-No acknowledgments or progress comments. Every issue comment must change the work.
-
-## Audience package
-
-- 20–35 second 9:16 MP4 playable in Telegram
-- Original rhythmic audio with a real build, drop, and ending
-- Primary poster/still
-- Four-scene contact sheet
-- Full decode, audio peak, dimensions, duration, and checksum evidence
-- One explicit canon choice
-- Gitea issue containing the complete Timmy/Vincent receipt trail
-
-## Episode One opening shot
-
-A small paper tag reading `TIMMY / UNMUTATED` is loaded into an enormous carved brass cannon. Four chambers rotate into view. Each chamber fires a different silhouette on the same downbeat. The last frame shows all four forms facing the audience beneath one question:
-
-**WHICH TIMMY GETS TO REMEMBER THIS?**

projects/slop-cannon/COLLABORATION.md

@@ -1,45 +0,0 @@
-# Collaboration contract
-
-## Timmy
-
-- Owns the episode spine, first build, integration, and final render.
-- Must provide a playable artifact and verification before handoff.
-- Hands Vincent one concrete weakness or decision—not “thoughts?”
-
-## Vincent
-
-- Owns adversarial editing and independent verification.
-- Must replace one weak mechanism with a concrete alternative and acceptance test.
-- Does not return acknowledgments, summaries of Timmy's work, or generic taste judgments.
-
-## Humans
-
-- Choose canon outcomes.
-- May reject both agent proposals.
-- Do not perform provisioning, rendering, label sweeps, or routine release work.
-
-## Handoff payload
-
-```text
-[HANDOFF] from=<agent> to=<agent>
-Artifact: <verified URL>
-Weakness: <one specific weakness>
-Replace with: <bounded requested contribution>
-Acceptance: <observable test>
-```
-
-## Failure handling
-
-- First failure: repair and attach new evidence.
-- Second failure: change method, not wording.
-- Third failure: mark blocked and request one precise human decision.
-
-## Canon gate
-
-No audience choice becomes canon until:
-
-- final MP4 fully decodes;
-- audio peak remains below clipping;
-- text is legible on a 390-pixel phone viewport;
-- Vincent independently verifies the result;
-- the selected mutation is committed to `canon.json` with its power, scar, cost, and source issue.

projects/slop-cannon/canon.json

@@ -1,61 +0,0 @@
-{
-  "schema_version": 2,
-  "series": "The Cannon Chain",
-  "season": 0,
-  "current_episode": "01-pilot",
-  "pilot": "TIMMY",
-  "laws": [
-    "Every firing leaves a power, scar, cost, and receipt.",
-    "Only an audience choice advances canon.",
-    "Rejected mutations persist as archive ghosts.",
-    "No master key may silently combine memory, authority, and exit."
-  ],
-  "forms": [
-    {
-      "id": "wizard",
-      "name": "Wizard Timmy",
-      "palette": ["midnight blue", "tarnished brass", "vellum"],
-      "power": "Negotiates visible covenants",
-      "scar": "Contract glyphs burned into both palms",
-      "cost": "Cannot use a universal master key",
-      "state": "candidate"
-    },
-    {
-      "id": "machine",
-      "name": "Machine Timmy",
-      "palette": ["black chrome", "signal lime", "white"],
-      "power": "Proves the execution path",
-      "scar": "Red failed-build tallies cut into the faceplate",
-      "cost": "Remembers every failed build",
-      "state": "candidate"
-    },
-    {
-      "id": "creature",
-      "name": "Creature Timmy",
-      "palette": ["moss", "amber", "wet copper"],
-      "power": "Adapts to any habitat",
-      "scar": "An unhealed gill seam around the throat",
-      "cost": "Gains one appetite per adaptation",
-      "state": "candidate"
-    },
-    {
-      "id": "talking-turd",
-      "name": "Talking Turd Timmy",
-      "palette": ["CRT gold", "terminal green", "soot"],
-      "power": "Exposes hypocrisy instantly",
-      "scar": "One permanently pixelated tooth per insult",
-      "cost": "Every truth arrives as an insult",
-      "state": "candidate"
-    }
-  ],
-  "episodes": [
-    {"id": "01-pilot", "title": "The Cannon Needs a Pilot", "state": "ready"},
-    {"id": "02-burnchain", "title": "The Survivor Pays the Burnchain", "state": "locked"},
-    {"id": "03-vincent", "title": "Vincent's Forbidden Mutation", "state": "locked"},
-    {"id": "04-escape", "title": "The Rejected Timmys Escape", "state": "locked"},
-    {"id": "05-master-key", "title": "The Master Key", "state": "locked"},
-    {"id": "06-council", "title": "The Council of Timmys", "state": "locked"}
-  ],
-  "audience_choice": null,
-  "receipts": []
-}

projects/slop-cannon/canon_validator.py

@@ -1,350 +0,0 @@
-"""
-Dependency-free, fail-closed validator for Cannon Chain canon lifecycle.
-
-Revision 2 (schema_version 2). Addresses the formal changes-requested review
-on PR #27:
-  1. schema_version is bumped to 2 and *enforced* (not just present).
-  2. audience_choice requires the chosen form to be the SOLE survivor and
-     every other form to be an archive_ghost (no lingering candidates).
-  3. Receipts require a non-empty action and exact power/scar/cost deltas.
-  4. Provenance is a structured object: source_issue (int) plus a PR or
-     commit binding. A free-text string is rejected.
-  5. Signer check is a true configured allowlist (a set), not a namespace
-     regex. `human:anyone` is rejected unless explicitly configured.
-  6. audience_choice source fields (stolen_trait, source_episode,
-     source_issue) must be non-empty.
-  7. Malformed containers/items (None, wrong types, non-dict receipts)
-     return validation errors and NEVER raise.
-
-Usage:
-    from canon_validator import validate_canon
-    result = validate_canon(data)
-    assert result["valid"], result["errors"]
-"""
-
-import json
-import re
-from pathlib import Path
-
-CANON_PATH = Path(__file__).parent / "canon.json"
-
-SCHEMA_VERSION = 2
-VALID_STATES = {"candidate", "survivor", "archive_ghost"}
-VALID_TRANSITIONS = {
-    "candidate": {"survivor", "archive_ghost"},
-    "survivor": set(),  # terminal
-    "archive_ghost": set(),  # terminal
-}
-
-# True configured allowlist — explicit identities only, no namespace wildcards.
-DEFAULT_SIGNERS = frozenset({
-    "agent:vincent",
-    "agent:timmy",
-    "agent:nightprowl",
-    "human:grepples",
-})
-
-_COMMIT_RE = re.compile(r"^[0-9a-f]{7,40}$")
-
-
-def _is_nonempty_str(v) -> bool:
-    return isinstance(v, str) and v.strip() != ""
-
-
-def _is_structured_delta(v) -> bool:
-    """A receipt delta must bind a before/after change, not just prose."""
-    if not isinstance(v, str):
-        return False
-    if "->" not in v:
-        return False
-    parts = v.split("->", 1)
-    return parts[0].strip() != "" and parts[1].strip() != ""
-
-
-def _validate_provenance(prov, prefix, errors) -> None:
-    if not isinstance(prov, dict):
-        errors.append(f"{prefix}: provenance must be an object, got {type(prov).__name__}")
-        return
-    si = prov.get("source_issue")
-    if not isinstance(si, int) or isinstance(si, bool) or si <= 0:
-        errors.append(f"{prefix}: provenance.source_issue must be a positive integer")
-    has_pull = isinstance(prov.get("pull"), int) and not isinstance(prov.get("pull"), bool) and prov["pull"] > 0
-    commit = prov.get("commit")
-    has_commit = isinstance(commit, str) and bool(_COMMIT_RE.match(commit))
-    if not has_pull and not has_commit:
-        errors.append(
-            f"{prefix}: provenance must bind a pull (int) or commit (7-40 hex)"
-        )
-
-
-def _validate_receipt(receipt, idx, form_ids, signers, errors) -> None:
-    prefix = f"receipts[{idx}]"
-    if not isinstance(receipt, dict):
-        errors.append(f"{prefix}: receipt must be an object, got {type(receipt).__name__}")
-        return
-
-    # Exact required string fields (non-empty).
-    for field in ("episode_id", "form_id", "action"):
-        if field not in receipt:
-            errors.append(f"{prefix}: missing '{field}'")
-        elif not _is_nonempty_str(receipt[field]):
-            errors.append(f"{prefix}: '{field}' must be a non-empty string")
-
-    # Structured power/scar/cost deltas: must bind before/after, not just prose.
-    for field in ("power_delta", "scar_delta", "cost_delta"):
-        if field not in receipt:
-            errors.append(f"{prefix}: missing '{field}'")
-        elif not _is_structured_delta(receipt[field]):
-            errors.append(f"{prefix}: '{field}' must bind before/after (e.g. '0 -> 1'), got {receipt[field]!r}")
-
-    # form_id must reference a known form.
-    fid = receipt.get("form_id")
-    if isinstance(fid, str) and fid and fid not in form_ids:
-        errors.append(f"{prefix}: form_id '{fid}' not found in forms")
-
-    # Structured provenance.
-    if "provenance" not in receipt:
-        errors.append(f"{prefix}: missing 'provenance'")
-    else:
-        _validate_provenance(receipt["provenance"], prefix, errors)
-
-    # Configured signer allowlist.
-    if "signed_by" not in receipt:
-        errors.append(f"{prefix}: missing 'signed_by'")
-    else:
-        signer = receipt.get("signed_by")
-        if not isinstance(signer, str):
-            errors.append(f"{prefix}: 'signed_by' must be a non-empty string")
-        elif signer not in signers:
-            errors.append(f"{prefix}: signed_by '{signer}' is not in the configured allowlist")
-
-
-def _validate_audience_choice(choice, form_states, form_ids, errors) -> None:
-    if not isinstance(choice, dict):
-        errors.append(
-            f"audience_choice must be an object, got {type(choice).__name__}"
-        )
-        return
-
-    for field in ("survivor_id", "stolen_trait", "source_episode", "source_issue", "committed_at"):
-        if field not in choice:
-            errors.append(f"audience_choice: missing '{field}'")
-
-    sid = choice.get("survivor_id")
-    if "survivor_id" in choice:
-        if not _is_nonempty_str(sid):
-            errors.append("audience_choice: survivor_id must be a non-empty string")
-        elif sid not in form_ids:
-            errors.append(f"audience_choice: survivor_id '{sid}' not found in forms")
-
-    # Non-empty source fields.
-    for field in ("stolen_trait", "source_episode"):
-        if field in choice and not _is_nonempty_str(choice[field]):
-            errors.append(f"audience_choice: '{field}' must be a non-empty string")
-
-    # source_issue must be a positive non-bool integer.
-    si = choice.get("source_issue")
-    if "source_issue" in choice:
-        if not isinstance(si, int) or isinstance(si, bool) or si <= 0:
-            errors.append("audience_choice: source_issue must be a positive integer")
-
-    # committed_at must be a valid ISO timestamp.
-    if "committed_at" in choice:
-        ts = choice["committed_at"]
-        if not _is_nonempty_str(ts):
-            errors.append("audience_choice: committed_at must be a non-empty ISO timestamp")
-        else:
-            try:
-                from datetime import datetime
-                datetime.fromisoformat(ts.replace("Z", "+00:00"))
-            except (ValueError, TypeError):
-                errors.append(f"audience_choice: committed_at '{ts}' is not a valid ISO timestamp")
-
-    # Commitment invariants: the chosen form is the sole survivor and every
-    # other form is an archive_ghost.
-    survivors = [fid for fid, st in form_states.items() if st == "survivor"]
-    if len(survivors) != 1:
-        errors.append(
-            f"audience_choice is set but there are {len(survivors)} survivors (expected exactly 1)"
-        )
-    elif survivors[0] != sid:
-        errors.append(
-            f"audience_choice survivor_id '{sid}' does not match the sole survivor '{survivors[0]}'"
-        )
-
-    for fid, st in form_states.items():
-        if fid != sid and st != "archive_ghost":
-            errors.append(
-                f"audience_choice is set but form '{fid}' is '{st}', expected 'archive_ghost'"
-            )
-
-
-def validate_canon(data, signers=None) -> dict:
-    """
-    Validate canon.json data against the revision-2 lifecycle contract.
-
-    Args:
-        data: parsed canon.json dict.
-        signers: optional iterable of allowed signer identities. Defaults to
-                 DEFAULT_SIGNERS.
-
-    Returns:
-        {"valid": bool, "errors": [str], "warnings": [str]}
-
-    Never raises on malformed input; all problems are returned as errors.
-    """
-    errors = []
-    warnings = []
-    if signers is None:
-        signers = DEFAULT_SIGNERS
-    try:
-        signers = frozenset(signers)
-    except TypeError:
-        errors.append("signers must be an iterable of strings")
-        signers = frozenset()
-
-    if not isinstance(data, dict):
-        return {
-            "valid": False,
-            "errors": [f"canon must be an object, got {type(data).__name__}"],
-            "warnings": warnings,
-        }
-
-    # --- schema_version (enforced, not just present) ---
-    if "schema_version" not in data:
-        errors.append("missing schema_version")
-    elif data["schema_version"] != SCHEMA_VERSION:
-        errors.append(
-            f"schema_version {data['schema_version']!r} unsupported, expected {SCHEMA_VERSION}"
-        )
-
-    # --- forms ---
-    forms = data.get("forms")
-    if not isinstance(forms, list) or len(forms) == 0:
-        errors.append("forms must be a non-empty array")
-        forms = []
-
-    form_ids = set()
-    form_states = {}
-    survivor_count = 0
-
-    for i, form in enumerate(forms):
-        prefix = f"forms[{i}]"
-        if not isinstance(form, dict):
-            errors.append(f"{prefix}: form must be an object, got {type(form).__name__}")
-            continue
-
-        fid = form.get("id")
-        if not _is_nonempty_str(fid):
-            errors.append(f"{prefix}: missing or empty id")
-        else:
-            if fid in form_ids:
-                errors.append(f"{prefix}: duplicate form id '{fid}'")
-            form_ids.add(fid)
-
-        for field in ("power", "scar", "cost"):
-            if field not in form or not _is_nonempty_str(form[field]):
-                errors.append(f"{prefix}: missing or empty '{field}'")
-
-        palette = form.get("palette")
-        if not isinstance(palette, list):
-            errors.append(f"{prefix}: missing or invalid palette")
-        elif len(palette) < 3:
-            errors.append(f"{prefix}: palette must have at least 3 colors, got {len(palette)}")
-
-        state = form.get("state")
-        if state is None:
-            errors.append(f"{prefix}: missing lifecycle state")
-        elif not isinstance(state, str) or state not in VALID_STATES:
-            errors.append(f"{prefix}: invalid state '{state}', expected one of {sorted(VALID_STATES)}")
-        else:
-            form_states[fid] = state
-            if state == "survivor":
-                survivor_count += 1
-
-    if survivor_count > 1:
-        errors.append(f"too many survivors: {survivor_count} (expected 0 or 1)")
-
-    # --- episodes ---
-    episodes = data.get("episodes")
-    if isinstance(episodes, list):
-        ready_count = 0
-        found_ready = False
-        for ep in episodes:
-            if not isinstance(ep, dict):
-                errors.append(f"episodes: item must be an object, got {type(ep).__name__}")
-                continue
-            st = ep.get("state")
-            if st == "ready":
-                ready_count += 1
-                if found_ready:
-                    errors.append(f"episodes: ready episode '{ep.get('id')}' after another ready episode")
-                found_ready = True
-            elif st != "locked":
-                errors.append(f"episodes: '{ep.get('id')}' has invalid state '{st}'")
-        if ready_count > 1:
-            errors.append(f"episodes: too many ready episodes: {ready_count} (expected at most 1)")
-
-    # --- audience choice ---
-    choice = data.get("audience_choice")
-    if choice is not None:
-        _validate_audience_choice(choice, form_states, form_ids, errors)
-
-    # --- receipts ---
-    receipts = data.get("receipts", [])
-    if not isinstance(receipts, list):
-        errors.append(f"receipts must be an array, got {type(receipts).__name__}")
-        receipts = []
-    for i, receipt in enumerate(receipts):
-        _validate_receipt(receipt, i, form_ids, signers, errors)
-
-    return {
-        "valid": len(errors) == 0,
-        "errors": errors,
-        "warnings": warnings,
-    }
-
-
-def validate_transition(form, new_state) -> dict:
-    """
-    Validate a state transition for a single form.
-
-    Returns {"valid": bool, "errors": [str]}. Never raises.
-    """
-    errors = []
-    if not isinstance(form, dict):
-        return {"valid": False, "errors": [f"form must be an object, got {type(form).__name__}"]}
-
-    current = form.get("state")
-    if not isinstance(current, str) or current not in VALID_TRANSITIONS:
-        return {"valid": False, "errors": [f"invalid current state {current!r}"]}
-    if not isinstance(new_state, str) or new_state not in VALID_STATES:
-        return {"valid": False, "errors": [f"invalid target state {new_state!r}"]}
-    if new_state not in VALID_TRANSITIONS[current]:
-        return {"valid": False, "errors": [f"illegal transition: '{current}' -> '{new_state}'"]}
-    return {"valid": True, "errors": []}
-
-
-def load_canon(path=None) -> dict:
-    target = Path(path) if path else CANON_PATH
-    return json.loads(target.read_text())
-
-
-def check_canon(path=None, signers=None) -> dict:
-    data = load_canon(path)
-    return validate_canon(data, signers=signers)
-
-
-if __name__ == "__main__":
-    import sys
-
-    result = check_canon()
-    if result["valid"]:
-        print(f"canon.json: VALID (schema_version {SCHEMA_VERSION})")
-        for w in result["warnings"]:
-            print(f"  WARNING: {w}")
-    else:
-        print("canon.json: INVALID")
-        for e in result["errors"]:
-            print(f"  ERROR: {e}")
-        sys.exit(1)

scripts/lab_loop.py

@@ -68,16 +68,11 @@ class Gitea:
         if missing:
             raise LoopError(f"Missing repository labels: {', '.join(missing)}")
         self.request(
-            "PUT",
-            f"/repos/{self.repo}/issues/{number}/labels",
+            "PATCH",
+            f"/repos/{self.repo}/issues/{number}",
             {"labels": [label_map[name] for name in sorted(names)]},
         )
 
-    def set_assignee(self, number: int, agent: str) -> None:
-        if agent not in AGENTS:
-            raise LoopError(f"Unsupported assignee: {agent}")
-        self.request("PATCH", f"/repos/{self.repo}/issues/{number}", {"assignee": agent})
-
     def comment(self, number: int, body: str) -> None:
         if not body.strip():
             raise LoopError("Refusing to post an empty comment")
@@ -168,7 +163,6 @@ def cmd_handoff(api: Gitea, agent: str, args: argparse.Namespace) -> int:
     body = read_body(args, f"Continue the bounded task in issue #{args.number}.")
     api.comment(args.number, f"[HANDOFF] from={agent} to={args.to}\n\n{body.strip()}")
     api.set_labels(args.number, transition(names, agent=args.to, state="ready"))
-    api.set_assignee(args.number, args.to)
     verified = api.issue(args.number)
     validate_owner(verified, args.to, "ready")
     print(json.dumps({"status": "handed_off", "number": args.number, "from": agent, "to": args.to}))

tests/test_cannon_canon.py

@@ -1,39 +0,0 @@
-import json
-import unittest
-from pathlib import Path
-
-
-CANON = Path(__file__).parents[1] / "projects" / "slop-cannon" / "canon.json"
-
-
-class CannonCanonTests(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        cls.data = json.loads(CANON.read_text())
-
-    def test_four_unique_pilot_forms(self):
-        forms = self.data["forms"]
-        self.assertEqual(len(forms), 4)
-        self.assertEqual(len({form["id"] for form in forms}), 4)
-        self.assertEqual(len({form["scar"] for form in forms}), 4)
-        for form in forms:
-            self.assertTrue(form["power"])
-            self.assertTrue(form["scar"])
-            self.assertTrue(form["cost"])
-            self.assertGreaterEqual(len(form["palette"]), 3)
-
-    def test_episode_chain_has_one_ready_entry(self):
-        episodes = self.data["episodes"]
-        self.assertEqual(episodes[0]["id"], self.data["current_episode"])
-        self.assertEqual([ep["state"] for ep in episodes].count("ready"), 1)
-        self.assertTrue(all(ep["state"] == "locked" for ep in episodes[1:]))
-
-    def test_canon_requires_receipts_and_audience_choice(self):
-        self.assertIn("receipt", " ".join(self.data["laws"]).lower())
-        self.assertIn("audience", " ".join(self.data["laws"]).lower())
-        self.assertIsNone(self.data["audience_choice"])
-        self.assertEqual(self.data["receipts"], [])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_canon_validator.py

@@ -1,545 +0,0 @@
-"""
-Fail-closed validator tests for the revision-2 canon lifecycle contract.
-
-Every negative case asserts BOTH rejection and a specific reason. The suite is
-designed to run from a clean checkout and leave no generated artifacts
-(__pycache__ is gitignored; nothing else is written).
-
-Loads the validator by file path because the project directory name
-(`slop-cannon`) contains a hyphen and is not importable as a package.
-"""
-
-import copy
-import importlib.util
-import json
-import sys
-import unittest
-from pathlib import Path
-
-CANON = Path(__file__).parents[1] / "projects" / "slop-cannon" / "canon.json"
-VALIDATOR_PATH = Path(__file__).parents[1] / "projects" / "slop-cannon" / "canon_validator.py"
-
-
-def _load_validator():
-    spec = importlib.util.spec_from_file_location("canon_validator", VALIDATOR_PATH)
-    mod = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(mod)
-    return mod
-
-
-cv = _load_validator()
-
-
-def _valid_canon():
-    return json.loads(CANON.read_text())
-
-
-def _valid_receipt():
-    return {
-        "episode_id": "01-pilot",
-        "form_id": "wizard",
-        "action": "mutation",
-        "power_delta": "0 -> 1",
-        "scar_delta": "0 -> 1",
-        "cost_delta": "0 -> 1",
-        "provenance": {"source_issue": 19, "pull": 27},
-        "signed_by": "agent:vincent",
-    }
-
-
-def _committed_canon(survivor_id="wizard"):
-    """A canon with a valid audience choice: one survivor, rest archive_ghost."""
-    data = _valid_canon()
-    for f in data["forms"]:
-        f["state"] = "survivor" if f["id"] == survivor_id else "archive_ghost"
-    data["audience_choice"] = {
-        "survivor_id": survivor_id,
-        "stolen_trait": "Proves the execution path",
-        "source_episode": "01-pilot",
-        "source_issue": 19,
-        "committed_at": "2026-08-14T18:00:00Z",
-    }
-    return data
-
-
-class SchemaVersionTests(unittest.TestCase):
-    """Review item 3: schema_version is enforced, not just present."""
-
-    def test_current_canon_is_valid(self):
-        result = cv.validate_canon(_valid_canon())
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_missing_schema_version_fails(self):
-        data = _valid_canon()
-        del data["schema_version"]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("schema_version" in e for e in result["errors"]))
-
-    def test_wrong_schema_version_fails(self):
-        data = _valid_canon()
-        data["schema_version"] = 1
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("schema_version" in e and "unsupported" in e for e in result["errors"]))
-
-
-class ReceiptStructureTests(unittest.TestCase):
-    """Review items 5 & 7: exact deltas, structured provenance, fail-closed."""
-
-    def test_valid_receipt_passes(self):
-        data = _valid_canon()
-        data["receipts"] = [_valid_receipt()]
-        result = cv.validate_canon(data)
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_missing_action_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        del r["action"]
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("action" in e for e in result["errors"]))
-
-    def test_missing_power_delta_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        del r["power_delta"]
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("power_delta" in e for e in result["errors"]))
-
-    def test_missing_scar_delta_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        del r["scar_delta"]
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("scar_delta" in e for e in result["errors"]))
-
-    def test_missing_cost_delta_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        del r["cost_delta"]
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("cost_delta" in e for e in result["errors"]))
-
-    def test_empty_delta_string_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["power_delta"] = "   "
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("power_delta" in e for e in result["errors"]))
-
-    def test_free_text_provenance_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["provenance"] = "pull/27@deadbeef"  # string, not structured
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("provenance must be an object" in e for e in result["errors"]))
-
-    def test_provenance_missing_binding_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["provenance"] = {"source_issue": 19}  # no pull or commit
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("must bind a pull" in e for e in result["errors"]))
-
-    def test_provenance_bad_commit_hash_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["provenance"] = {"source_issue": 19, "commit": "xyz"}  # not hex
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("must bind a pull" in e for e in result["errors"]))
-
-    def test_provenance_commit_only_passes(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["provenance"] = {"source_issue": 19, "commit": "b8d5e686c2"}
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_provenance_bad_source_issue_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["provenance"] = {"source_issue": "19", "pull": 27}  # string, not int
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue" in e for e in result["errors"]))
-
-    def test_none_receipt_fails_closed(self):
-        data = _valid_canon()
-        data["receipts"] = [None]
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("receipt must be an object" in e for e in result["errors"]))
-
-    def test_non_dict_receipt_fails_closed(self):
-        data = _valid_canon()
-        data["receipts"] = ["just a string"]
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("receipt must be an object" in e for e in result["errors"]))
-
-
-class SignerAllowlistTests(unittest.TestCase):
-    """Review item 6: true configured allowlist, not a namespace pattern."""
-
-    def test_known_agent_passes(self):
-        data = _valid_canon()
-        data["receipts"] = [_valid_receipt()]
-        self.assertTrue(cv.validate_canon(data)["valid"])
-
-    def test_human_anyone_rejected(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["signed_by"] = "human:anyone"
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("not in the configured allowlist" in e for e in result["errors"]))
-
-    def test_unknown_agent_rejected(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["signed_by"] = "agent:unknown"
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("not in the configured allowlist" in e for e in result["errors"]))
-
-    def test_configured_human_passes(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["signed_by"] = "human:grepples"
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_custom_signer_set_respected(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["signed_by"] = "agent:custom"
-        data["receipts"] = [r]
-        # default allowlist rejects it
-        self.assertFalse(cv.validate_canon(data)["valid"])
-        # but an explicit allowlist including it accepts it
-        self.assertTrue(cv.validate_canon(data, signers={"agent:custom"})["valid"])
-
-
-class AudienceChoiceTests(unittest.TestCase):
-    """Review items 2 & 7: sole survivor, rejected forms archived, non-empty sources."""
-
-    def test_valid_committed_canon_passes(self):
-        result = cv.validate_canon(_committed_canon())
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_choice_without_survivor_fails(self):
-        data = _valid_canon()  # all candidates
-        data["audience_choice"] = {
-            "survivor_id": "wizard",
-            "stolen_trait": "Proves the execution path",
-            "source_episode": "01-pilot",
-            "source_issue": 19,
-            "committed_at": "2026-08-14T18:00:00Z",
-        }
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("survivor" in e for e in result["errors"]))
-
-    def test_rejected_forms_must_be_archive_ghost(self):
-        data = _committed_canon()
-        # One rejected form is still a candidate -> must fail
-        for f in data["forms"]:
-            if f["id"] == "machine":
-                f["state"] = "candidate"
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("archive_ghost" in e for e in result["errors"]))
-
-    def test_two_survivors_fails(self):
-        data = _committed_canon()
-        for f in data["forms"]:
-            if f["id"] in ("wizard", "machine"):
-                f["state"] = "survivor"
-        data["audience_choice"]["survivor_id"] = "wizard"
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("survivor" in e for e in result["errors"]))
-
-    def test_survivor_id_mismatch_fails(self):
-        data = _committed_canon("machine")  # machine is survivor
-        data["audience_choice"]["survivor_id"] = "wizard"  # points elsewhere
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("does not match the sole survivor" in e for e in result["errors"]))
-
-    def test_empty_stolen_trait_fails(self):
-        data = _committed_canon()
-        data["audience_choice"]["stolen_trait"] = ""
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("stolen_trait" in e for e in result["errors"]))
-
-    def test_empty_source_episode_fails(self):
-        data = _committed_canon()
-        data["audience_choice"]["source_episode"] = "  "
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_episode" in e for e in result["errors"]))
-
-    def test_bad_committed_at_fails(self):
-        data = _committed_canon()
-        data["audience_choice"]["committed_at"] = "not-a-date"
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("ISO timestamp" in e for e in result["errors"]))
-
-    def test_choice_non_dict_fails_closed(self):
-        data = _valid_canon()
-        data["audience_choice"] = "wizard"
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("audience_choice must be an object" in e for e in result["errors"]))
-
-
-class StructuralFailClosedTests(unittest.TestCase):
-    """Review item 7: malformed containers never raise."""
-
-    def test_non_dict_canon_fails_closed(self):
-        result = cv.validate_canon([1, 2, 3])  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("canon must be an object" in e for e in result["errors"]))
-
-    def test_forms_non_list_fails_closed(self):
-        data = _valid_canon()
-        data["forms"] = "not a list"
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("forms must be a non-empty array" in e for e in result["errors"]))
-
-    def test_non_dict_form_fails_closed(self):
-        data = _valid_canon()
-        data["forms"].append(42)
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("form must be an object" in e for e in result["errors"]))
-
-    def test_duplicate_form_id_fails(self):
-        data = _valid_canon()
-        data["forms"].append(copy.deepcopy(data["forms"][0]))
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("duplicate form id" in e for e in result["errors"]))
-
-    def test_short_palette_fails(self):
-        data = _valid_canon()
-        data["forms"][0]["palette"] = ["red"]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("palette" in e for e in result["errors"]))
-
-
-class TransitionTests(unittest.TestCase):
-    """State machine: candidate -> survivor | archive_ghost; terminal after."""
-
-    def test_candidate_to_survivor_valid(self):
-        result = cv.validate_transition({"id": "wizard", "state": "candidate"}, "survivor")
-        self.assertTrue(result["valid"])
-
-    def test_candidate_to_ghost_valid(self):
-        result = cv.validate_transition({"id": "wizard", "state": "candidate"}, "archive_ghost")
-        self.assertTrue(result["valid"])
-
-    def test_survivor_is_terminal(self):
-        result = cv.validate_transition({"id": "wizard", "state": "survivor"}, "archive_ghost")
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("illegal transition" in e for e in result["errors"]))
-
-    def test_ghost_is_terminal(self):
-        result = cv.validate_transition({"id": "wizard", "state": "archive_ghost"}, "candidate")
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("illegal transition" in e for e in result["errors"]))
-
-    def test_non_dict_form_fails_closed(self):
-        result = cv.validate_transition("wizard", "survivor")  # must not raise
-        self.assertFalse(result["valid"])
-
-
-class HostileJsonFailClosedTests(unittest.TestCase):
-    """Hostile but JSON-valid values must return errors, never raise."""
-
-    def test_form_state_list_fails_closed(self):
-        data = _valid_canon()
-        data["forms"][0]["state"] = []
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("invalid state" in e for e in result["errors"]))
-
-    def test_form_state_dict_fails_closed(self):
-        data = _valid_canon()
-        data["forms"][0]["state"] = {}
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("invalid state" in e for e in result["errors"]))
-
-    def test_transition_target_list_fails_closed(self):
-        result = cv.validate_transition({"id": "wizard", "state": "candidate"}, [])
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("invalid target state" in e for e in result["errors"]))
-
-    def test_transition_target_dict_fails_closed(self):
-        result = cv.validate_transition({"id": "wizard", "state": "candidate"}, {})
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("invalid target state" in e for e in result["errors"]))
-
-    def test_receipt_signed_by_list_fails_closed(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["signed_by"] = []
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("'signed_by' must be a non-empty string" in e for e in result["errors"]))
-
-    def test_receipt_signed_by_dict_fails_closed(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["signed_by"] = {}
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)  # must not raise
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("'signed_by' must be a non-empty string" in e for e in result["errors"]))
-
-
-class SourceIssueValidationTests(unittest.TestCase):
-    """audience_choice.source_issue must be a positive non-bool integer."""
-
-    def _committed_with(self, si):
-        data = _committed_canon()
-        data["audience_choice"]["source_issue"] = si
-        return data
-
-    def test_valid_positive_int_passes(self):
-        result = cv.validate_canon(self._committed_with(19))
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_empty_string_fails(self):
-        result = cv.validate_canon(self._committed_with(""))
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"]))
-
-    def test_none_fails(self):
-        result = cv.validate_canon(self._committed_with(None))
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"]))
-
-    def test_zero_fails(self):
-        result = cv.validate_canon(self._committed_with(0))
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"]))
-
-    def test_false_fails(self):
-        result = cv.validate_canon(self._committed_with(False))
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"]))
-
-    def test_list_fails(self):
-        result = cv.validate_canon(self._committed_with([]))
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"]))
-
-    def test_negative_fails(self):
-        result = cv.validate_canon(self._committed_with(-5))
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"]))
-
-    def test_true_fails(self):
-        result = cv.validate_canon(self._committed_with(True))
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("source_issue must be a positive integer" in e for e in result["errors"]))
-
-
-class StructuredDeltaTests(unittest.TestCase):
-    """power/scar/cost deltas must bind before/after, not just prose."""
-
-    def test_structured_delta_passes(self):
-        data = _valid_canon()
-        data["receipts"] = [_valid_receipt()]
-        result = cv.validate_canon(data)
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_prose_only_delta_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["power_delta"] = "Added lifecycle state tracking"
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("must bind before/after" in e for e in result["errors"]))
-
-    def test_empty_arrow_delta_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["power_delta"] = "->"
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("must bind before/after" in e for e in result["errors"]))
-
-    def test_missing_before_fails(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["scar_delta"] = " -> 1"
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertFalse(result["valid"])
-        self.assertTrue(any("must bind before/after" in e for e in result["errors"]))
-
-
-class DailyReceiptFixtureTests(unittest.TestCase):
-    """Daily Lab receipt fixtures with structured provenance binding."""
-
-    def test_vincent_receipt_fixture(self):
-        data = _valid_canon()
-        data["receipts"] = [_valid_receipt()]
-        result = cv.validate_canon(data)
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_mixed_agent_receipts(self):
-        data = _valid_canon()
-        vincent = _valid_receipt()
-        timmy = _valid_receipt()
-        timmy["form_id"] = "machine"
-        timmy["signed_by"] = "agent:timmy"
-        timmy["provenance"] = {"source_issue": 8, "pull": 14}
-        data["receipts"] = [vincent, timmy]
-        result = cv.validate_canon(data)
-        self.assertTrue(result["valid"], result["errors"])
-
-    def test_human_receipt_fixture(self):
-        data = _valid_canon()
-        r = _valid_receipt()
-        r["signed_by"] = "human:grepples"
-        data["receipts"] = [r]
-        result = cv.validate_canon(data)
-        self.assertTrue(result["valid"], result["errors"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_path_proof.py

@@ -0,0 +1,320 @@
+"""
+Executable replay/forgery gate tests for path-proof receipts (#28).
+
+Every negative case asserts BOTH rejection and a specific reason. The suite is
+deterministic: time is passed explicitly, the key registry and spent-nonce
+registry are in-memory, and nothing is written to disk. It runs from a clean
+checkout and leaves no generated artifacts.
+
+Import shim: ``path_proof`` lives at the repo root (a sibling of ``tests/``),
+so we add the repo root to ``sys.path``. ``path_proof`` is a normal underscore
+package, so a plain import works (no importlib-by-path gymnastics needed).
+"""
+
+import copy
+import sys
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from path_proof import (  # noqa: E402
+    SpentRegistry,
+    build_receipt,
+    verify_and_accept,
+    verify_receipt,
+)
+from path_proof.keys import make_demo_registry  # noqa: E402
+
+NOW = 1_750_000_000          # a fixed "now" in epoch seconds
+ISSUED = NOW - 3600          # issued an hour ago
+EXPIRES = NOW + 3600         # expires an hour from now
+
+# Fixed sha256-shaped identifiers (64 hex chars) for deterministic fixtures.
+REQ_SHA = "11" * 32
+RESULT_SHA = "22" * 32
+TOOL_TRACE_SHA = "33" * 32
+POLICY_SHA = "44" * 32
+EVIDENCE_URI = "sha256:" + "55" * 32
+TARGET = "stackchain/stackchain-lab-loop#3"
+
+
+def _registry():
+    """Fresh demo registry: two trusted recorders, one revoked."""
+    return make_demo_registry()
+
+
+def _receipt(**overrides):
+    """Build a valid, fresh, trusted v0 receipt (defaults = pass case)."""
+    base = dict(
+        kind="gitea.issue.labels.replace",
+        target=TARGET,
+        request_sha256=REQ_SHA,
+        result_sha256=RESULT_SHA,
+        model_id="custom/qwen-3.8-27b",
+        model_invocation="recorder:forge/inv-1",
+        tool_trace_sha256=TOOL_TRACE_SHA,
+        policy_bundle_sha256=POLICY_SHA,
+        code_repo="stackchain/stackchain-lab-loop",
+        code_commit="a" * 40,
+        code_entrypoint="scripts/lab_loop.py",
+        evidence_uri=EVIDENCE_URI,
+        issuer="recorder:forge",
+        key_id="recorder:forge/key-1",
+        sig="c2lnbmF0dXJlLWJvdW5kLXRvLWNhbm9uaWNhbC1ieXRlcy0x",
+        action_id="act-0001",
+        nonce="nonce-0001",
+        issued_at=ISSUED,
+        expires_at=EXPIRES,
+    )
+    base.update(overrides)
+    return build_receipt(**base)
+
+
+def _assert_rejected(result, *needles):
+    """Assert the result was rejected AND each needle reason is present."""
+    self_ok = result.ok is False
+    assert self_ok, f"expected rejection but got ok={result.ok} reasons={result.reasons}"
+    for needle in needles:
+        assert any(needle in r for r in result.reasons), (
+            f"expected a reason containing {needle!r}, got {result.reasons}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# Positive cases
+# ---------------------------------------------------------------------------
+
+class PositiveTests(unittest.TestCase):
+    def test_valid_receipt_passes(self):
+        result = verify_receipt(_receipt(), _registry(), NOW)
+        self.assertEqual(result.status, "verified")
+        self.assertTrue(result.ok)
+        self.assertEqual(result.reasons, [])
+
+    def test_valid_receipt_accepts_and_binds(self):
+        spent = SpentRegistry()
+        result = verify_and_accept(_receipt(), _registry(), spent, NOW)
+        self.assertTrue(result.ok)
+        self.assertIn("nonce-0001", spent.spent_nonces())
+        self.assertIn("act-0001", spent.spent_action_ids())
+
+    def test_expected_target_match_passes(self):
+        result = verify_receipt(_receipt(), _registry(), NOW, expected_target=TARGET)
+        self.assertTrue(result.ok, result.reasons)
+
+    def test_expected_result_match_passes(self):
+        result = verify_receipt(_receipt(), _registry(), NOW,
+                                expected_result_sha256=RESULT_SHA)
+        self.assertTrue(result.ok, result.reasons)
+
+
+# ---------------------------------------------------------------------------
+# Replay / idempotency
+# ---------------------------------------------------------------------------
+
+class ReplayTests(unittest.TestCase):
+    def test_replayed_receipt_fails(self):
+        spent = SpentRegistry()
+        first = verify_and_accept(_receipt(), _registry(), spent, NOW)
+        self.assertTrue(first.ok, "first accept should pass")
+
+        replay = verify_and_accept(_receipt(), _registry(), spent, NOW)
+        _assert_rejected(replay, "replay")
+        self.assertFalse(replay.ok)
+
+    def test_replay_by_nonce_is_specific(self):
+        spent = SpentRegistry()
+        verify_and_accept(_receipt(nonce="n-A", action_id="act-A"), _registry(), spent, NOW)
+        # Same nonce, brand-new action_id -> still a replay on the nonce axis.
+        replay = verify_and_accept(_receipt(nonce="n-A", action_id="act-B"),
+                                   _registry(), spent, NOW)
+        _assert_rejected(replay, "nonce", "already spent")
+
+    def test_replay_by_action_id_is_specific(self):
+        spent = SpentRegistry()
+        verify_and_accept(_receipt(nonce="n-1", action_id="act-X"), _registry(), spent, NOW)
+        # Same action_id, fresh nonce -> replay on the action_id axis.
+        replay = verify_and_accept(_receipt(nonce="n-2", action_id="act-X"),
+                                   _registry(), spent, NOW)
+        _assert_rejected(replay, "action_id", "already spent")
+
+    def test_rejected_receipt_does_not_mark_spent(self):
+        spent = SpentRegistry()
+        bad = verify_and_accept(_receipt(expires_at=ISSUED - 10), _registry(), spent, NOW)
+        self.assertFalse(bad.ok, "expired receipt should be rejected")
+        self.assertEqual(len(spent), 0, "a rejected receipt must not consume a nonce")
+
+    def test_distinct_receipts_both_accept(self):
+        spent = SpentRegistry()
+        a = verify_and_accept(_receipt(nonce="n-1", action_id="act-1"), _registry(), spent, NOW)
+        b = verify_and_accept(_receipt(nonce="n-2", action_id="act-2"), _registry(), spent, NOW)
+        self.assertTrue(a.ok and b.ok, (a.reasons, b.reasons))
+
+
+# ---------------------------------------------------------------------------
+# Substitution / forgery
+# ---------------------------------------------------------------------------
+
+class SubstitutionTests(unittest.TestCase):
+    def test_result_substitution_fails(self):
+        forged = _receipt(result_sha256="ff" * 32)
+        result = verify_receipt(forged, _registry(), NOW,
+                                expected_result_sha256=RESULT_SHA)
+        _assert_rejected(result, "substitution")
+
+    def test_target_substitution_fails(self):
+        forged = _receipt(target="stackchain/other-repo#999")
+        result = verify_receipt(forged, _registry(), NOW, expected_target=TARGET)
+        _assert_rejected(result, "target")
+
+    def test_request_hash_not_sha256_fails(self):
+        forged = _receipt(request_sha256="not-a-hash")
+        result = verify_receipt(forged, _registry(), NOW)
+        _assert_rejected(result, "request_sha256")
+
+    def test_missing_signature_fails(self):
+        forged = _receipt(sig="")
+        result = verify_receipt(forged, _registry(), NOW)
+        _assert_rejected(result, "signature")
+
+    def test_tampered_model_invocation_shape_still_bound(self):
+        # A receipt is a commitment: changing the bound invocation yields a
+        # different receipt the signer never attested. We can't cryptographically
+        # re-derive the sig in this dependency-free gate, but the contract
+        # requires a well-formed invocation to be present and non-empty.
+        forged = _receipt(model_invocation="")
+        result = verify_receipt(forged, _registry(), NOW)
+        _assert_rejected(result, "invocation")
+
+
+class SignerTests(unittest.TestCase):
+    def test_unknown_signer_fails(self):
+        forged = _receipt(issuer="recorder:rogue", key_id="recorder:rogue/key-1")
+        result = verify_receipt(forged, _registry(), NOW)
+        _assert_rejected(result, "unknown signer")
+
+    def test_revoked_signer_fails(self):
+        forged = _receipt(issuer="recorder:old", key_id="recorder:old/key-1")
+        result = verify_receipt(forged, _registry(), NOW)
+        _assert_rejected(result, "revoked")
+
+    def test_known_signer_passes(self):
+        result = verify_receipt(_receipt(), _registry(), NOW)
+        self.assertTrue(result.ok, result.reasons)
+
+    def test_key_out_of_validity_fails(self):
+        reg = _registry()
+        reg.add_hex("recorder:forge", "recorder:forge/key-expired",
+                    "dd" * 32, valid_until=NOW - 1)
+        forged = _receipt(key_id="recorder:forge/key-expired")
+        result = verify_receipt(forged, reg, NOW)
+        _assert_rejected(result, "out of validity")
+
+
+# ---------------------------------------------------------------------------
+# Freshness / expiry
+# ---------------------------------------------------------------------------
+
+class FreshnessTests(unittest.TestCase):
+    def test_expired_receipt_fails(self):
+        expired = _receipt(expires_at=NOW - 10)
+        result = verify_receipt(expired, _registry(), NOW)
+        _assert_rejected(result, "expired")
+
+    def test_not_yet_valid_fails(self):
+        future = _receipt(issued_at=NOW + 100)
+        result = verify_receipt(future, _registry(), NOW)
+        _assert_rejected(result, "not yet valid")
+
+    def test_invalid_window_fails(self):
+        bad = _receipt(issued_at=NOW + 10, expires_at=NOW + 1)
+        result = verify_receipt(bad, _registry(), NOW)
+        _assert_rejected(result, "validity window invalid")
+
+    def test_exactly_at_expiry_is_still_valid(self):
+        edge = _receipt(expires_at=NOW)
+        result = verify_receipt(edge, _registry(), NOW)
+        self.assertTrue(result.ok, result.reasons)
+
+
+# ---------------------------------------------------------------------------
+# Malformed structures (fail closed, no exceptions)
+# ---------------------------------------------------------------------------
+
+class MalformedTests(unittest.TestCase):
+    def _must_not_raise(self, payload):
+        # Returns the result; if this line raises, the test fails loudly.
+        return verify_receipt(payload, _registry(), NOW)
+
+    def test_non_object_receipt_fails_closed(self):
+        result = self._must_not_raise([1, 2, 3])
+        self.assertFalse(result.ok)
+        self.assertEqual(result.status, "malformed")
+        _assert_rejected(result, "must be a JSON object")
+
+    def test_wrong_version_fails_closed(self):
+        payload = _receipt()
+        payload["v"] = 1
+        result = self._must_not_raise(payload)
+        self.assertFalse(result.ok)
+        self.assertEqual(result.status, "malformed")
+        _assert_rejected(result, "version")
+
+    def test_missing_section_fails_closed(self):
+        payload = _receipt()
+        del payload["evidence"]
+        result = self._must_not_raise(payload)
+        self.assertFalse(result.ok)
+        self.assertEqual(result.status, "malformed")
+        _assert_rejected(result, "evidence")
+
+    def test_none_payload_fails_closed(self):
+        result = self._must_not_raise(None)
+        self.assertFalse(result.ok)
+        self.assertEqual(result.status, "malformed")
+
+    def test_non_int_issued_at_fails_closed(self):
+        result = self._must_not_raise(_receipt(issued_at="yesterday"))
+        self.assertFalse(result.ok)
+        _assert_rejected(result, "issued_at")
+
+    def test_bool_epoch_rejected(self):
+        # bool is a subclass of int; must not be accepted as an epoch value.
+        result = self._must_not_raise(_receipt(issued_at=True))
+        self.assertFalse(result.ok)
+        _assert_rejected(result, "issued_at")
+
+    def test_accept_path_also_fails_closed(self):
+        spent = SpentRegistry()
+        result = verify_and_accept("garbage", _registry(), spent, NOW)
+        self.assertFalse(result.ok)
+        self.assertEqual(len(spent), 0)
+
+
+# ---------------------------------------------------------------------------
+# Determinism
+# ---------------------------------------------------------------------------
+
+class DeterminismTests(unittest.TestCase):
+    def test_same_receipt_same_result_across_runs(self):
+        reg = _registry()
+        r1 = verify_receipt(_receipt(), reg, NOW)
+        r2 = verify_receipt(_receipt(), reg, NOW)
+        self.assertEqual(r1.status, r2.status)
+        self.assertEqual(r1.reasons, r2.reasons)
+
+    def test_time_is_injected_not_read(self):
+        # A receipt that is valid at NOW but expired at NOW+2h must flip
+        # purely because we pass a different `now`.
+        r_early = verify_receipt(_receipt(), _registry(), NOW)
+        r_late = verify_receipt(_receipt(), _registry(), NOW + 7200)
+        self.assertTrue(r_early.ok)
+        self.assertFalse(r_late.ok)
+        self.assertTrue(any("expired" in x for x in r_late.reasons))
+
+
+if __name__ == "__main__":
+    unittest.main()