From f4beedc3e7ac2e142dcbdeefb0e5ee40c20d9b86 Mon Sep 17 00:00:00 2001 From: wjarvis mos-comms Date: Sun, 19 Jul 2026 19:49:04 -0500 Subject: [PATCH] test(mosaic): define recovery runtime boundary contracts --- packages/mosaic/package.json | 2 +- .../lease-broker/recovery_runtime_unittest.py | 213 ++++++++++++++++++ 2 files changed, 214 insertions(+), 1 deletion(-) create mode 100644 packages/mosaic/src/lease-broker/recovery_runtime_unittest.py diff --git a/packages/mosaic/package.json b/packages/mosaic/package.json index 64220d39..9805efb2 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/lease-broker/receipt_challenge_unittest.py && python3 src/lease-broker/context_recovery_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/lease-broker/context_recovery_unittest.py && python3 src/lease-broker/recovery_runtime_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/recovery_runtime_unittest.py b/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py new file mode 100644 index 00000000..2c9f62a0 --- /dev/null +++ b/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""RED-first B1/B2 runtime-boundary contracts for constrained recovery. + +Every runtime process in these tests is a fresh child against a private daemon +and Unix sockets. They do not activate a live Mosaic daemon or model stream. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import socket +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path + + +TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker" +FRAMEWORK = Path(__file__).parents[2] / "framework" +DAEMON = TOOLS / "daemon.py" +GATE = TOOLS / "mutator-gate.py" +RECOVERY = TOOLS / "recover-context.py" +OBSERVER_CLIENT = TOOLS / "receipt-observer-client.py" +CLAUDE_SETTINGS = FRAMEWORK / "runtime/claude/settings.json" +PI_EXTENSION = FRAMEWORK / "runtime/pi/mosaic-extension.ts" + + +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 response: {bytes(response)!r}") + reply = json.loads(response[:-1]) + if not isinstance(reply, dict): + raise AssertionError("broker response is not an object") + return reply + + +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"daemon exited before READY: {output}") + time.sleep(0.02) + raise TimeoutError("daemon did not create private broker socket") + + +class RuntimeBoundaryFixture(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name) + os.chmod(self.root, 0o700) + self.socket = self.root / "broker.sock" + self.observer_socket = self.root / "observer.sock" + self.state = self.root / "state.json" + self.children: list[subprocess.Popen[str]] = [] + + def tearDown(self) -> None: + for child in self.children: + if child.poll() is None: + child.terminate() + try: + child.wait(timeout=3.0) + except subprocess.TimeoutExpired: + child.kill() + child.wait() + self.temporary.cleanup() + + def start_daemon(self, *, production_observer: bool) -> None: + arguments = [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(self.socket), "--state", str(self.state)] + if production_observer: + arguments.extend(["--observer-socket", str(self.observer_socket)]) + process = subprocess.Popen( + arguments, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + self.children.append(process) + wait_ready(process, self.socket) + + def register(self) -> str: + reply = request(self.socket, {"action": "register_anchor", "runtime_generation": 1}) + session_id = reply.get("session_id") + self.assertTrue(reply.get("ok")) + self.assertIsInstance(session_id, str) + return session_id + + def environment(self, session_id: str) -> dict[str, str]: + return { + **os.environ, + "MOSAIC_LEASE_BROKER_SOCKET": str(self.socket), + "MOSAIC_RECEIPT_OBSERVER_SOCKET": str(self.observer_socket), + "MOSAIC_LEASE_SESSION_ID": session_id, + "MOSAIC_RUNTIME_GENERATION": "1", + "MOSAIC_LEASE_RUNTIME": "pi", + } + + def construction(self) -> Path: + content = b"WI-6 B2 production observer fixture\n" + path = self.root / "construction.json" + path.write_text(json.dumps({ + "manifest_version": 1, + "generator_version": "wi6-repair-runtime-boundary", + "fragments": [{ + "source_id": "authority/wi6-repair", + "content_base64": base64.b64encode(content).decode("ascii"), + "expected_sha256": hashlib.sha256(content).hexdigest(), + }], + }), encoding="utf-8") + os.chmod(path, 0o600) + return path + + +class RecoveryRuntimeBoundaryTest(RuntimeBoundaryFixture): + def test_b1_claude_exact_recovery_command_is_invocable_unverified_but_bash_is_not(self) -> None: + self.start_daemon(production_observer=False) + session_id = self.register() + environment = self.environment(session_id) + recovery_command = f"python3 {RECOVERY} complete" + mapped = subprocess.run( + [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)], + input=json.dumps({"tool_name": "Bash", "tool_input": {"command": recovery_command}}), + text=True, + capture_output=True, + env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"}, + check=False, + ) + self.assertEqual(mapped.returncode, 0, mapped.stderr) + ordinary_bash = subprocess.run( + [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)], + input=json.dumps({"tool_name": "Bash", "tool_input": {"command": "echo not-recovery"}}), + text=True, + capture_output=True, + env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"}, + check=False, + ) + self.assertEqual(ordinary_bash.returncode, 2) + + def test_b1_pi_registers_only_the_broker_exempt_recovery_tool(self) -> None: + extension = PI_EXTENSION.read_text(encoding="utf-8") + self.assertIn("const RECOVERY_TOOL = 'mosaic_context_recover'", extension) + self.assertIn("name: RECOVERY_TOOL", extension) + self.assertIn("checkPiMutatorGate(RECOVERY_TOOL)", extension) + self.assertNotIn("toolName === 'bash' ? RECOVERY_TOOL", extension) + + def test_b2_production_observer_promotes_over_private_transport_and_rejects_broker_supplied_message(self) -> None: + self.start_daemon(production_observer=True) + session_id = self.register() + environment = self.environment(session_id) + begin = subprocess.run( + [sys.executable, "-I", "-S", "-B", str(RECOVERY), "begin", "--construction", str(self.construction()), "--compaction-epoch", "1", "--request-epoch", "1"], + text=True, + capture_output=True, + env=environment, + check=False, + ) + self.assertEqual(begin.returncode, 0, begin.stderr) + cycle = json.loads(begin.stdout) + receipt = cycle.get("receipt") + self.assertIsInstance(receipt, str) + + # S1: production transport is not a broker request field. The public + # broker endpoint keeps rejecting caller-supplied assistant evidence. + rejected = request(self.socket, { + "action": "observe_receipt", + "session_id": session_id, + "runtime_generation": 1, + "receipt_challenge": cycle["receipt_challenge"], + "latest_assistant_message": receipt, + }) + self.assertEqual(rejected, {"ok": False, "code": "INVALID_RECEIPT"}) + + recorded = subprocess.run( + [sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", "pi"], + input=json.dumps({"latest_assistant_message": receipt}), + text=True, + capture_output=True, + env=environment, + check=False, + ) + self.assertEqual(recorded.returncode, 0, recorded.stderr) + complete = subprocess.run( + [sys.executable, "-I", "-S", "-B", str(RECOVERY), "complete"], + text=True, + capture_output=True, + env=environment, + check=False, + ) + self.assertEqual(complete.returncode, 0, complete.stderr) + self.assertEqual(json.loads(complete.stdout).get("state"), "VERIFIED") + + +if __name__ == "__main__": + unittest.main()