diff --git a/docs/compaction-refresh/probes/p5_receipt_replay.py b/docs/compaction-refresh/probes/p5_receipt_replay.py new file mode 100644 index 00000000..50540b57 --- /dev/null +++ b/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() diff --git a/docs/scratchpads/832-receipt-challenge-protocol.md b/docs/scratchpads/832-receipt-challenge-protocol.md new file mode 100644 index 00000000..1cbe9a61 --- /dev/null +++ b/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. diff --git a/packages/mosaic/package.json b/packages/mosaic/package.json index 20bb7384..c7096afe 100644 --- a/packages/mosaic/package.json +++ b/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:*", diff --git a/packages/mosaic/src/lease-broker/receipt_challenge_unittest.py b/packages/mosaic/src/lease-broker/receipt_challenge_unittest.py new file mode 100644 index 00000000..f1b6430e --- /dev/null +++ b/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()