test(#832): define receipt challenge replay contracts

docs/compaction-refresh/probes/p5_receipt_replay.py

@@ -0,0 +1,209 @@
+#!/usr/bin/env python3
+"""P5 Gate0 replay probe; BUILT ONLY, execution is Mos-gated.
+
+Run only under fresh-executor authorization:
+  python3 -I -S -B docs/compaction-refresh/probes/p5_receipt_replay.py
+
+Each of the default three isolated runs launches the shipped lease-broker daemon
+in a distinct private temporary directory.  This driver never changes broker
+state directly and does not replace the promote gate: every transition is sent
+over the daemon's real Unix socket.  It proves the shipped order is
+PENDING_DELIVERY -> observe/evidence commit -> consume -> VERIFIED and that a
+consumed challenge cannot be replayed or reopen/renew its lease.
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+
+HERE = Path(__file__).resolve().parent
+REPOSITORY = HERE.parents[2]
+TOOLS = REPOSITORY / "packages/mosaic/framework/tools/lease-broker"
+DAEMON = TOOLS / "daemon.py"
+FRAGMENTS = TOOLS / "normative_fragments.py"
+
+
+def load_shipped_fragments():
+    if not FRAGMENTS.is_file():
+        raise RuntimeError(f"shipped normative construction missing: {FRAGMENTS}")
+    spec = importlib.util.spec_from_file_location("p5_shipped_normative_fragments", FRAGMENTS)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("unable to load shipped normative construction")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def request(socket_path: Path, value: dict[str, object]) -> dict[str, object]:
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(3.0)
+        connection.connect(str(socket_path))
+        connection.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode())
+        connection.shutdown(socket.SHUT_WR)
+        response = bytearray()
+        while True:
+            chunk = connection.recv(4096)
+            if not chunk:
+                break
+            response.extend(chunk)
+    if not response.endswith(b"\n") or response.count(b"\n") != 1:
+        raise AssertionError(f"unframed broker reply: {bytes(response)!r}")
+    parsed = json.loads(response[:-1])
+    if not isinstance(parsed, dict):
+        raise AssertionError(f"non-object broker reply: {parsed!r}")
+    return parsed
+
+
+def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        if socket_path.exists():
+            return
+        if process.poll() is not None:
+            output = process.stdout.read() if process.stdout is not None else ""
+            raise RuntimeError(f"shipped daemon exited before READY: {output}")
+        time.sleep(0.02)
+    raise TimeoutError("shipped daemon did not create private probe socket")
+
+
+def expect_refused(reply: dict[str, object], code: str) -> None:
+    if reply != {"ok": False, "code": code}:
+        raise AssertionError(f"expected refusal {code}, got {reply!r}")
+
+
+def run_once(index: int) -> str:
+    fragments = load_shipped_fragments()
+    root = Path(tempfile.mkdtemp(prefix=f"mosaic-p5-replay-{index}-"))
+    os.chmod(root, 0o700)
+    socket_path = root / "broker.sock"
+    state_path = root / "state.json"
+    process = subprocess.Popen(
+        [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(socket_path), "--state", str(state_path)],
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+    )
+    try:
+        wait_ready(process, socket_path)
+        registered = request(socket_path, {"action": "register_anchor", "runtime_generation": 1})
+        if registered.get("ok") is not True or not isinstance(registered.get("session_id"), str):
+            raise AssertionError(f"registration failed: {registered!r}")
+        session_id = registered["session_id"]
+        construction = fragments.build_payload(
+            manifest_version=1,
+            generator_version="p5-replay-probe",
+            fragments=[
+                fragments.NormativeFragment(
+                    "authority/probe",
+                    b"P5 shipped transition driver\n",
+                    "63537df1a6cb0d80195a96757ab11d629e5b5e1f23be167218b84cb195b1c1d6",
+                ),
+            ],
+        )
+        if construction.injectionDecision != "ACCEPTED" or not construction.promotion:
+            raise AssertionError("shipped normative construction refused P5 fixture")
+        binding = {
+            "compaction_epoch": index,
+            "request_epoch": index + 100,
+            "h_source": construction.h_source,
+            "h_payload": construction.h_payload,
+            "schema_version": 1,
+        }
+        pending = request(socket_path, {
+            "action": "begin_verification",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "runtime": "pi",
+            "binding": binding,
+        })
+        if pending.get("ok") is not True or pending.get("state") != "PENDING_VERIFICATION":
+            raise AssertionError(f"shipped pending-delivery transition failed: {pending!r}")
+        challenge = pending.get("receipt_challenge")
+        receipt = pending.get("receipt")
+        if not isinstance(challenge, str) or not isinstance(receipt, str):
+            raise AssertionError(f"shipped broker did not mint a receipt challenge: {pending!r}")
+
+        # Promotion before observation/evidence/consumption is forbidden.
+        expect_refused(request(socket_path, {
+            "action": "promote_lease",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        }), "INVALID_LEASE_TRANSITION")
+
+        observed = request(socket_path, {
+            "action": "observe_receipt",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+            "latest_assistant_message": receipt,
+        })
+        if observed.get("ok") is not True or observed.get("state") != "PENDING_PROMOTION":
+            raise AssertionError(f"shipped evidence transition failed: {observed!r}")
+        durable = json.loads(state_path.read_text(encoding="utf-8"))
+        evidence = durable["tokens"][challenge].get("evidence")
+        if not isinstance(evidence, dict) or not isinstance(evidence.get("h_latest_assistant"), str):
+            raise AssertionError("shipped receipt evidence was not committed before consume/promote")
+
+        promoted = request(socket_path, {
+            "action": "promote_lease",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        })
+        if promoted.get("ok") is not True or promoted.get("state") != "VERIFIED":
+            raise AssertionError(f"shipped consume-before-promote transition failed: {promoted!r}")
+
+        # T25/T28: the actual consumed challenge, re-presented through the
+        # shipped daemon, can neither be observed again nor re-promote/reopen.
+        expect_refused(request(socket_path, {
+            "action": "observe_receipt",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+            "latest_assistant_message": receipt,
+        }), "RECEIPT_REPLAY")
+        expect_refused(request(socket_path, {
+            "action": "promote_lease",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt_challenge": challenge,
+        }), "RECEIPT_REPLAY")
+        return challenge
+    finally:
+        if process.poll() is None:
+            process.terminate()
+            try:
+                process.wait(timeout=3.0)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+        shutil.rmtree(root, ignore_errors=True)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--runs", type=int, default=3)
+    arguments = parser.parse_args()
+    if arguments.runs != 3:
+        raise SystemExit("P5 requires exactly three isolated runs")
+    challenges = [run_once(index) for index in range(arguments.runs)]
+    if len(set(challenges)) != arguments.runs:
+        raise AssertionError("separate shipped cycles did not mint unique challenges")
+    print("P5 receipt replay probe PASS: 3 isolated shipped-daemon runs")
+
+
+if __name__ == "__main__":
+    main()

docs/scratchpads/832-receipt-challenge-protocol.md

@@ -0,0 +1,13 @@
+# #832 Receipt-challenge protocol — build scratchpad
+
+- **Objective:** Deliver WI-5 receipt-challenge protocol ACs T25, T26, T28, and T29 only.
+- **Authority:** BUILD-BRIEF, SPEC-v5, ratification, and red-team hashes verified in STEP-0.
+- **Base:** `e522b22fa4492861b0fcd4a956a8795c54eb9bfe` (`origin/main`).
+- **Constraints:** Byte-build only: no live broker/socket/systemd/tmux mutation. No PR, self-review, or probe fire. T27/T30 are out of scope.
+- **Plan:**
+  1. Add red-first deterministic T26/T29 in-build tests that call shipped normative construction and broker path.
+  2. Add an unexecuted, isolated P5 out-of-process replay harness that drives the shipped daemon and asserts consume-before-promote for T25/T28.
+  3. Implement the broker-minted receipt challenge and exact receipt observation/consume/promote path.
+  4. Run unit, framework-shell, compile, lint, and type checks; push after the required queue guard; report to `mosaic-100`.
+- **Risks:** The standalone harness must drive the real daemon without a divergent fixture. If that is impossible, stop and flag Mos.
+- **Evidence:** Pending red run, green checks, and push.

packages/mosaic/package.json

@@ -25,7 +25,7 @@
     "lint": "eslint src",
     "typecheck": "tsc --noEmit",
     "test": "vitest run --passWithNoTests && pnpm run test:framework-shell",
-    "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh"
+    "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/lease-broker/receipt_challenge_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh"
   },
   "dependencies": {
     "@mosaicstack/brain": "workspace:*",

packages/mosaic/src/lease-broker/receipt_challenge_unittest.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""RED-first T26/T29 contract tests for the shipped receipt-challenge path."""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+
+TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
+DAEMON_PATH = TOOLS / "daemon.py"
+FRAGMENTS_PATH = TOOLS / "normative_fragments.py"
+RECEIPT_PATH = TOOLS / "receipt_challenge.py"
+
+
+def load_module(name: str, path: Path):
+    assert path.is_file(), f"shipped module is missing: {path}"
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"unable to load {name}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+DAEMON = load_module("lease_broker_receipt_daemon", DAEMON_PATH)
+FRAGMENTS = load_module("lease_broker_normative_fragments", FRAGMENTS_PATH)
+RECEIPTS = load_module("lease_broker_receipt_challenge", RECEIPT_PATH)
+
+
+class ReceiptChallengeTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        root = Path(self.temporary.name)
+        os.chmod(root, 0o700)
+        self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
+        self.peer = (os.getpid(), os.getuid(), os.getgid())
+        registered = self.broker.handle(self.peer, {
+            "action": "register_anchor",
+            "runtime_generation": 7,
+        })
+        self.session_id = registered["session_id"]
+        self.assertIsInstance(self.session_id, str)
+
+    def tearDown(self) -> None:
+        self.temporary.cleanup()
+
+    def binding(self, compaction_epoch: int, request_epoch: int) -> dict[str, object]:
+        construction = FRAGMENTS.build_payload(
+            manifest_version=1,
+            generator_version="wi5-receipt-test",
+            fragments=[
+                FRAGMENTS.NormativeFragment(
+                    "authority/constitution",
+                    b"Constitution\n",
+                    "b1f7989bf4a3feee87aaf841d3c94385337f12cdbf356e0db33f68edfe535f16",
+                ),
+            ],
+        )
+        self.assertEqual(construction.injectionDecision, "ACCEPTED")
+        self.assertTrue(construction.promotion)
+        return {
+            "compaction_epoch": compaction_epoch,
+            "request_epoch": request_epoch,
+            "h_source": construction.h_source,
+            "h_payload": construction.h_payload,
+            "schema_version": 1,
+        }
+
+    def begin(self, binding: dict[str, object]) -> dict[str, object]:
+        response = self.broker.handle(self.peer, {
+            "action": "begin_verification",
+            "session_id": self.session_id,
+            "runtime_generation": 7,
+            "runtime": "pi",
+            "binding": binding,
+        })
+        self.assertEqual(response["state"], DAEMON.LEASE_PENDING)
+        self.assertIsInstance(response.get("receipt_challenge"), str)
+        self.assertIsInstance(response.get("receipt"), str)
+        return response
+
+    def observe(self, challenge: str, message: str) -> dict[str, object]:
+        return self.broker.handle(self.peer, {
+            "action": "observe_receipt",
+            "session_id": self.session_id,
+            "runtime_generation": 7,
+            "receipt_challenge": challenge,
+            "latest_assistant_message": message,
+        })
+
+    def test_t26_stale_epoch_receipt_cannot_promote_against_shipped_binding(self) -> None:
+        stale = self.begin(self.binding(compaction_epoch=3, request_epoch=8))
+        current = self.begin(self.binding(compaction_epoch=4, request_epoch=9))
+        self.assertNotEqual(stale["receipt_challenge"], current["receipt_challenge"])
+
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
+            self.observe(current["receipt_challenge"], stale["receipt"])
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
+            self.broker.handle(self.peer, {
+                "action": "promote_lease",
+                "session_id": self.session_id,
+                "runtime_generation": 7,
+                "receipt_challenge": current["receipt_challenge"],
+            })
+
+    def test_t29_altered_model_hash_cannot_promote_against_shipped_binding(self) -> None:
+        cycle = self.begin(self.binding(compaction_epoch=5, request_epoch=10))
+        expected = cycle["receipt"]
+        self.assertIsInstance(expected, str)
+        altered_hash = "f" * 64
+        self.assertNotEqual(altered_hash, cycle["binding"]["h_payload"])
+        altered = expected.replace(cycle["binding"]["h_payload"], altered_hash, 1)
+        self.assertNotEqual(altered, expected)
+
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
+            self.observe(cycle["receipt_challenge"], altered)
+        with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
+            self.broker.handle(self.peer, {
+                "action": "promote_lease",
+                "session_id": self.session_id,
+                "runtime_generation": 7,
+                "receipt_challenge": cycle["receipt_challenge"],
+            })
+
+
+if __name__ == "__main__":
+    unittest.main()