diff --git a/docs/scratchpads/833-constrained-recovery-command.md b/docs/scratchpads/833-constrained-recovery-command.md new file mode 100644 index 00000000..a73010b3 --- /dev/null +++ b/docs/scratchpads/833-constrained-recovery-command.md @@ -0,0 +1,14 @@ +# #833 constrained recovery command — build scratchpad + +- **Objective:** Deliver WI-6 only: the single ungated constrained recovery command, durable `mosaic-context-refresh` wrapper skill, AC-1 observable partial-delivery and C4 replay evidence, and an unfired P6 probe. +- **Authority:** STEP-0 SHA-256 verified 4/4 against the supplied BUILD-BRIEF, SPEC-v5, ratification, and red-team records. +- **Base:** exact `07553ead337a70a9241f826d27571650262b289c`; new branch `feat/833-constrained-recovery-command`; merge-base assertion passed before any commit. +- **Constraints:** No rebase/pull during build; no live install/symlink or live broker/socket/tmux/systemd mutation; no self-review, PR, merge, push, or P6 fire. `docs/TASKS.md` is orchestrator-owned and will not be modified. +- **Plan:** + 1. Add red-first unit tests against the recovery broker entrypoint for fresh recovery challenge, normal-receipt replay refusal, observable partial-delivery refusal, and the explicit middle-drop negative capability; commit the RED test and preserve its command output. + 2. Implement the recovery command as a thin driver over shared WI-5 broker transitions and the trusted observer seam; it never accepts caller receipt text. + 3. Add the source-resident skill under `packages/mosaic/framework/skills/`, plus a tmp-only #824 bridge projection test. + 4. Build an unfired, default-3-run P6 standalone real-socket driver; it is not added to package scripts and will not be run. + 5. Run targeted broker/mutator/receipt suites, lint, and format check; report head to `mosaic-100` and stop. +- **Risks:** The observer can only represent an exact latest message. Tail-preserving middle-drop is intentionally not claimed receipt-detectable (T-C residual deferred to WI-7 server evidence). +- **Budget:** No explicit task token cap was supplied; scope is fixed to WI-6 and no unrelated behavior will be added. diff --git a/packages/mosaic/src/lease-broker/context_recovery_unittest.py b/packages/mosaic/src/lease-broker/context_recovery_unittest.py new file mode 100644 index 00000000..dcd9e1fd --- /dev/null +++ b/packages/mosaic/src/lease-broker/context_recovery_unittest.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +"""RED-first WI-6 contracts against the shipped constrained recovery entrypoint.""" + +from __future__ import annotations + +import base64 +import hashlib +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" +OBSERVER_PATH = TOOLS / "receipt_observer.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_recovery_daemon", DAEMON_PATH) +FRAGMENTS = load_module("lease_broker_recovery_fragments", FRAGMENTS_PATH) +OBSERVER = load_module("lease_broker_recovery_observer", OBSERVER_PATH) + + +class RecoveryFixture(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + root = Path(self.temporary.name) + os.chmod(root, 0o700) + self.peer = (os.getpid(), os.getuid(), os.getgid()) + self.observer = OBSERVER.TestReceiptObserver() + self.broker = DAEMON.Broker(DAEMON.StateStore(root / "state.json"), observer=self.observer) + registered = self.broker.handle(self.peer, { + "action": "register_anchor", + "runtime_generation": 11, + }) + self.session_id = registered["session_id"] + + def tearDown(self) -> None: + self.temporary.cleanup() + + def construction_and_binding(self) -> tuple[dict[str, object], dict[str, object]]: + content = b"Mosaic recovery authority\n" + construction = { + "manifest_version": 1, + "generator_version": "wi6-recovery-test", + "fragments": [{ + "source_id": "authority/recovery", + "content_base64": base64.b64encode(content).decode("ascii"), + "expected_sha256": hashlib.sha256(content).hexdigest(), + }], + } + built = FRAGMENTS.build_payload_from_wire(construction) + self.assertEqual(built.injectionDecision, "ACCEPTED") + self.assertTrue(built.promotion) + return construction, { + "compaction_epoch": 17, + "request_epoch": 23, + "h_source": built.h_source, + "h_payload": built.h_payload, + "schema_version": 1, + } + + def begin_normal(self) -> dict[str, object]: + construction, binding = self.construction_and_binding() + return self.broker.handle(self.peer, { + "action": "begin_verification", + "session_id": self.session_id, + "runtime_generation": 11, + "runtime": "pi", + "binding": binding, + "construction": construction, + }) + + def begin_recovery(self) -> dict[str, object]: + construction, binding = self.construction_and_binding() + return self.broker.handle(self.peer, { + "action": "begin_recovery", + "session_id": self.session_id, + "runtime_generation": 11, + "runtime": "pi", + "binding": binding, + "construction": construction, + }) + + def complete_recovery(self) -> dict[str, object]: + return self.broker.handle(self.peer, { + "action": "complete_recovery", + "session_id": self.session_id, + "runtime_generation": 11, + }) + + def assert_recovery_refused_unverified(self, code: str) -> None: + with self.assertRaisesRegex(DAEMON.BrokerFailure, code): + self.complete_recovery() + self.assertEqual(self.broker.leases[self.session_id]["state"], DAEMON.LEASE_UNVERIFIED) + denied = self.broker.handle(self.peer, { + "action": "authorize_tool", + "session_id": self.session_id, + "runtime_generation": 11, + "runtime": "pi", + "tool_name": "bash", + }) + self.assertEqual(denied["decision"], "deny") + + +class ConstrainedRecoveryContractTest(RecoveryFixture): + def test_recovery_mints_a_fresh_challenge_distinct_from_normal_path(self) -> None: + normal = self.begin_normal() + recovery = self.begin_recovery() + + self.assertEqual(recovery["state"], "PENDING_DELIVERY") + self.assertNotEqual(normal["receipt_challenge"], recovery["receipt_challenge"]) + self.assertIn(recovery["receipt_challenge"], recovery["receipt"]) + + def test_c4_normal_path_receipt_cannot_be_replayed_through_recovery(self) -> None: + normal = self.begin_normal() + recovery = self.begin_recovery() + self.assertNotEqual(normal["receipt_challenge"], recovery["receipt_challenge"]) + + self.observer.record_latest_assistant_message(self.session_id, 11, normal["receipt"]) + self.assert_recovery_refused_unverified("RECEIPT_MISMATCH") + + def test_t27_observable_partial_delivery_variants_never_promote(self) -> None: + variants = { + "absent": None, + "malformed": "MOSAIC-RECEIPT{malformed}", + "prefix-truncated": None, + "observable-adapter-mutation": None, + # Tail-only is represented only by this concrete malformed/incomplete + # delivery. It is not a category-wide tail-only detection claim. + "tail-only-malformed": "H_payload=tail-only", + } + for name, observed in variants.items(): + with self.subTest(name=name): + recovery = self.begin_recovery() + if name == "prefix-truncated": + observed = recovery["receipt"][:-1] + elif name == "observable-adapter-mutation": + observed = recovery["receipt"].replace("H_payload=", "H_payload=0", 1) + if observed is not None: + self.observer.record_latest_assistant_message(self.session_id, 11, observed) + self.assert_recovery_refused_unverified( + "RECEIPT_OBSERVATION_UNAVAILABLE" if observed is None else "RECEIPT_MISMATCH" + ) + + def test_negative_capability_tail_preserving_middle_drop_is_not_receipt_detectable(self) -> None: + recovery = self.begin_recovery() + + # The observer seam receives the exact terminal message, not the delivered + # payload bytes. A middle drop that preserves this tail is therefore T-C + # and deliberately NOT represented as receipt-detectable; WI-7 server-side + # evidence owns that residual. This is not an assertion that it is caught. + self.observer.record_latest_assistant_message(self.session_id, 11, recovery["receipt"]) + promoted = self.complete_recovery() + self.assertEqual(promoted["state"], DAEMON.LEASE_VERIFIED) + + +if __name__ == "__main__": + unittest.main()