fix(mosaic): register canonical framework skills

docs/SITEMAP.md

@@ -3,7 +3,8 @@
 ## Compaction refresh lease broker
 
 - [Internal broker protocol](architecture/lease-broker-protocol.md) — kernel identity, ancestry and generation invariants, framed requests, responses, and persisted cycle bindings.
-- [Broker operations](guides/lease-broker-operations.md) — protected paths, startup, fail-closed recovery posture, distinct-principal deployment, and residual risk.
+- [Broker operations](guides/lease-broker-operations.md) — protected paths, startup, constrained recovery, fail-closed posture, distinct-principal deployment, and residual risk.
+- [Constrained recovery skill](../packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md) — source-resident thin wrapper, receipt scope, C4 replay boundary, and T-C middle-drop disclosure.
 - [Lease-broker security notes](architecture/lease-broker-security.md) — identity, whole-class authorization, threat boundaries, and coordinator review requirements.
 - [Whole mutator-class gate](architecture/mutator-class-gate.md) — default-deny policy, revoke-first/promote-last state machine, TTL, runtime adapters, and T-B/T-C assurance boundary.
 - [Compaction revocation lifecycle](architecture/compaction-revocation.md) — Claude/Pi observer matrix, same-PID generation rollover, failure fencing, and the named bounded residual stale window.

docs/architecture/lease-broker-protocol.md

@@ -13,12 +13,17 @@ Each connection carries exactly one UTF-8 JSON object followed by one newline, c
 - `mint_token`: authenticated identity plus `binding` containing exactly `compaction_epoch`, `request_epoch`, `h_source`, `h_payload`, and `schema_version`.
 - `consume_token`: authenticated identity plus `token`.
 - `begin_verification`: authenticated identity, runtime (`claude` or `pi`), cycle `binding`, and a TTL no greater than 300 seconds. The broker revokes existing authority first, enters `PENDING_VERIFICATION`, and returns a single-use promotion token.
+- `begin_recovery`: the constrained recovery entrypoint. It rejects caller-provided receipt/challenge fields and delegates to the same `begin_verification` transition, but reports `PENDING_DELIVERY` and marks the volatile cycle as recovery-owned.
+- `complete_recovery`: authenticated identity only. It rejects caller-provided receipt/challenge fields, obtains the current recovery challenge only from broker state, and delegates to the same trusted-observer → evidence → consume → promote sequence. An observation failure revokes recovery authority; retry starts a fresh challenge.
+
+The daemon owns a second protected production observer socket (mode `0600`) unless a private `--test-observer-file` fixture is selected. That transport accepts only the exact `record_runtime_observation` schema after kernel `SO_PEERCRED` plus the existing anchor/ancestry authentication; it validates the pending runtime/generation before storing one finalized assistant entry for the in-process `RuntimeReceiptObserver`. It is **not** a broker request action. Claude sends its latest assistant entry from the Stop-hook transport; Pi sends only finalized `message_end` assistant content. The public broker socket continues to reject request-supplied `latest_assistant_message` in begin, observe, and complete paths.
+
 - `promote_lease`: authenticated identity plus the exact pending promotion token. The broker commits token consumption before making `VERIFIED` visible.
 - `revoke_lease`: authenticated observer signal; deletes pending tokens and makes the session `UNVERIFIED` immediately. WI-3 Claude/Pi hooks send this existing action; `runtime` and bounded `reason` fields are diagnostic input only and never identity authority.
 - `authorize_tool`: authenticated identity, runtime, and exact runtime-reported tool name. The broker returns an explicit allow/deny decision from the whole-class policy and current lease.
 
 A higher generation for the same anchor atomically replaces the stored incarnation and deletes all prior tokens and lease authority for that session. A lower generation is stale. Runtime descendants resolve the current generation from an owner-only, locked generation file created by the register-before-exec launcher; reload/new/resume/fork observers advance and `fsync` it before broker revocation. This supports generation replacement even when PID/starttime do not change. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. At most 256 pending tokens may be persisted; another mint fails with `TOKEN_CAPACITY` before mutation. Successful consumption deletes the token, while a replay still fails with `TOKEN_REPLAY`. Live v1 token records retain the existing `consumed: false` schema.
 
-VERIFIED leases are volatile and monotonic-time bounded: broker restart, generation change, explicit observer revocation, or expiry returns the session to `UNVERIFIED`. `begin_verification` always revokes before minting a new prerequisite. `promote_lease` is valid only from the matching pending cycle; persistence failure rolls token and lease state back, while post-rename durability uncertainty terminates the broker. The WI-1 token is the atomic promotion prerequisite substrate. A later receipt implementation must satisfy that prerequisite but cannot replace the mechanical mutator gate as safety authority.
+VERIFIED leases are volatile and monotonic-time bounded: broker restart, generation change, explicit observer revocation, or expiry returns the session to `UNVERIFIED`. `begin_verification` always revokes before minting a new prerequisite. `begin_recovery` reuses that exact transition and mints a new challenge, so a normal-path receipt/challenge cannot be replayed through recovery. `promote_lease` is valid only from the matching pending cycle; persistence failure rolls token and lease state back, while post-rename durability uncertainty terminates the broker. The WI-1 token is the atomic promotion prerequisite substrate. Receipt evidence is a T-A delivery/liveness prerequisite only; it cannot replace the mechanical mutator gate as safety authority. A tail-preserving middle drop is not receipt-detectable and remains the disclosed T-C/server-side residual.
 
 State replacement serializes and enforces the 4 MiB maximum before opening a temporary file, then uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Every broker mutation snapshots the prior v1 state. A commit failure before rename restores that snapshot and leaves durable state unchanged. A failure after rename makes durability uncertain, so the store is poisoned without rolling memory back and the daemon terminates rather than serving with divergent state. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Persisted tokens must be unconsumed, match their session's current generation, and remain within the 256-token cap. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens.

docs/compaction-refresh/probes/p6_constrained_recovery.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""P6 constrained-recovery probe; BUILT ONLY and Mos-gated.
+
+DO NOT self-fire. Under Mos authorization only:
+  python3 -I -S -B docs/compaction-refresh/probes/p6_constrained_recovery.py
+
+The default three isolated runs launch the shipped daemon plus its production
+observer transport on private sockets. The driver invokes the shipped recovery
+command and adapter gate identity; it never resets broker state, mocks promote,
+or taps a live model-output stream.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+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"
+GATE = TOOLS / "mutator-gate.py"
+RECOVERY_COMMAND = TOOLS / "recover-context.py"
+OBSERVER_CLIENT = TOOLS / "receipt-observer-client.py"
+FRAGMENTS = TOOLS / "normative_fragments.py"
+CLAUDE_SETTINGS = REPOSITORY / "packages/mosaic/framework/runtime/claude/settings.json"
+PI_EXTENSION = REPOSITORY / "packages/mosaic/framework/runtime/pi/mosaic-extension.ts"
+
+
+def load_shipped_fragments():
+    spec = importlib.util.spec_from_file_location("p6_shipped_fragments", FRAGMENTS)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("shipped normative construction unavailable")
+    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}")
+    reply = json.loads(response[:-1])
+    if not isinstance(reply, dict):
+        raise AssertionError("broker reply 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"shipped daemon exited before READY: {output}")
+        time.sleep(0.02)
+    raise TimeoutError("shipped daemon did not create private probe socket")
+
+
+def run_json(command: list[str], environment: dict[str, str], input_value: object | None = None) -> dict[str, object]:
+    completed = subprocess.run(
+        command,
+        input=None if input_value is None else json.dumps(input_value),
+        text=True,
+        capture_output=True,
+        env=environment,
+        check=False,
+    )
+    if not completed.stdout.endswith("\n"):
+        raise AssertionError(f"command omitted framed result: {completed.stderr!r}")
+    reply = json.loads(completed.stdout)
+    if not isinstance(reply, dict):
+        raise AssertionError("command result is not an object")
+    return reply
+
+
+def gate_recovery(runtime: str, phase: str, environment: dict[str, str]) -> None:
+    command = [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", runtime]
+    if runtime == "claude":
+        command.extend(["--recovery-command", str(RECOVERY_COMMAND)])
+        recovery_invocation = (
+            f"python3 {RECOVERY_COMMAND} begin --construction /tmp/p6.json "
+            "--compaction-epoch 1 --request-epoch 1"
+            if phase == "begin"
+            else f"python3 {RECOVERY_COMMAND} complete"
+        )
+        value = {"tool_name": "Bash", "tool_input": {"command": recovery_invocation}}
+    else:
+        value = {"tool_name": "mosaic_context_recover"}
+    completed = subprocess.run(command, input=json.dumps(value), text=True, capture_output=True, env=environment, check=False)
+    if completed.returncode != 0:
+        raise AssertionError(f"{runtime} recovery invocation remained gated: {completed.stderr!r}")
+
+
+def record_production_observation(runtime: str, message: str, root: Path, environment: dict[str, str]) -> None:
+    command = [sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", runtime]
+    if runtime == "claude":
+        transcript = root / "claude-transcript.jsonl"
+        transcript.write_text(json.dumps({"message": {"role": "assistant", "content": message}}) + "\n", encoding="utf-8")
+        payload = {"transcript_path": str(transcript)}
+        command.append("--latest-entry")
+    else:
+        payload = {"latest_assistant_message": message}
+    completed = subprocess.run(command, input=json.dumps(payload), text=True, capture_output=True, env=environment, check=False)
+    if completed.returncode != 0:
+        raise AssertionError(f"{runtime} production observer transport refused: {completed.stderr!r}")
+
+
+def run_once(index: int, runtime: str) -> None:
+    # Parity guard: drive the shipped command and the repaired adapter/observer
+    # bytes, not a shadow receipt or promotion implementation.
+    recovery_source = RECOVERY_COMMAND.read_text(encoding="utf-8")
+    if '"action": "begin_recovery"' not in recovery_source or '"action": "complete_recovery"' not in recovery_source:
+        raise AssertionError("P6 parity guard: recovery command no longer drives shipped broker entrypoints")
+    gate_source = GATE.read_text(encoding="utf-8")
+    if "--recovery-command" not in CLAUDE_SETTINGS.read_text(encoding="utf-8"):
+        raise AssertionError("P6 parity guard: Claude recovery mapping is missing")
+    if "_SHELL_ACTIVE" not in gate_source or "argv[1] != str(recovery_command)" not in gate_source:
+        raise AssertionError("P6 parity guard: Claude mapping is not literal-only")
+    if "const RECOVERY_TOOL = 'mosaic_context_recover'" not in PI_EXTENSION.read_text(encoding="utf-8"):
+        raise AssertionError("P6 parity guard: Pi recovery tool mapping is missing")
+
+    fragments = load_shipped_fragments()
+    root = Path(tempfile.mkdtemp(prefix=f"mosaic-p6-recovery-{index}-"))
+    os.chmod(root, 0o700)
+    socket_path = root / "broker.sock"
+    observer_socket = root / "observer.sock"
+    state_path = root / "state.json"
+    construction_path = root / "construction.json"
+    content = b"P6 constrained recovery fixture\n"
+    construction = {
+        "manifest_version": 1,
+        "generator_version": "p6-constrained-recovery",
+        "fragments": [{
+            "source_id": "authority/p6",
+            "content_base64": base64.b64encode(content).decode("ascii"),
+            "expected_sha256": hashlib.sha256(content).hexdigest(),
+        }],
+    }
+    construction_path.write_text(json.dumps(construction), encoding="utf-8")
+    os.chmod(construction_path, 0o600)
+    process = subprocess.Popen(
+        [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(socket_path),
+         "--state", str(state_path), "--observer-socket", str(observer_socket)],
+        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})
+        session_id = registered.get("session_id")
+        if registered.get("ok") is not True or not isinstance(session_id, str):
+            raise AssertionError(f"broker anchor registration failed: {registered!r}")
+        built = fragments.build_payload_from_wire(construction)
+        normal = request(socket_path, {
+            "action": "begin_verification", "session_id": session_id, "runtime_generation": 1,
+            "runtime": runtime, "construction": construction,
+            "binding": {"compaction_epoch": index, "request_epoch": index + 100,
+                        "h_source": built.h_source, "h_payload": built.h_payload, "schema_version": 1},
+        })
+        normal_challenge = normal.get("receipt_challenge")
+        normal_receipt = normal.get("receipt")
+        if not isinstance(normal_challenge, str) or not isinstance(normal_receipt, str):
+            raise AssertionError("normal path did not mint a receipt challenge")
+        environment = {
+            **os.environ,
+            "MOSAIC_LEASE_BROKER_SOCKET": str(socket_path),
+            "MOSAIC_RECEIPT_OBSERVER_SOCKET": str(observer_socket),
+            "MOSAIC_LEASE_SESSION_ID": session_id,
+            "MOSAIC_RUNTIME_GENERATION": "1",
+            "MOSAIC_LEASE_RUNTIME": runtime,
+        }
+        gate_recovery(runtime, "begin", environment)
+        recovery = run_json([
+            sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "begin", "--construction", str(construction_path),
+            "--compaction-epoch", str(index + 10), "--request-epoch", str(index + 110),
+        ], environment)
+        challenge = recovery.get("receipt_challenge")
+        receipt = recovery.get("receipt")
+        if recovery.get("state") != "PENDING_DELIVERY" or not isinstance(challenge, str) or not isinstance(receipt, str):
+            raise AssertionError(f"recovery command did not drive pending delivery: {recovery!r}")
+        if challenge == normal_challenge:
+            raise AssertionError("recovery reused a normal-path challenge")
+
+        # C4: production observer content is still exact-current-cycle only.
+        record_production_observation(runtime, normal_receipt, root, environment)
+        refused = run_json([sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "complete"], environment)
+        if refused.get("ok") is not False or refused.get("code") != "RECEIPT_MISMATCH":
+            raise AssertionError(f"normal-path receipt replay was not refused: {refused!r}")
+
+        gate_recovery(runtime, "begin", environment)
+        recovery = run_json([
+            sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "begin", "--construction", str(construction_path),
+            "--compaction-epoch", str(index + 20), "--request-epoch", str(index + 120),
+        ], environment)
+        receipt = recovery.get("receipt")
+        if recovery.get("state") != "PENDING_DELIVERY" or not isinstance(receipt, str):
+            raise AssertionError(f"fresh recovery retry did not pend: {recovery!r}")
+        record_production_observation(runtime, receipt, root, environment)
+        gate_recovery(runtime, "complete", environment)
+        promoted = run_json([sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "complete"], environment)
+        if promoted.get("ok") is not True or promoted.get("state") != "VERIFIED":
+            raise AssertionError(f"recovery consume-before-promote failed: {promoted!r}")
+        replay = run_json([sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), "complete"], environment)
+        if replay.get("ok") is not False or replay.get("code") != "INVALID_LEASE_TRANSITION":
+            raise AssertionError(f"consumed recovery challenge re-promoted: {replay!r}")
+    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("P6 requires exactly three isolated runs")
+    for index, runtime in enumerate(("pi", "claude", "pi")):
+        run_once(index, runtime)
+    print("P6 constrained recovery probe PASS: 3 isolated shipped recovery-command runs")
+
+
+if __name__ == "__main__":
+    main()

docs/guides/lease-broker-operations.md

@@ -31,7 +31,13 @@ The same check runs in the Mosaic package test suite and therefore in root CI. A
 
 Clients must complete the request boundary before waiting for a reply. After sending the single JSON object and its terminating newline, the client **MUST half-close the socket's write side** (`shutdown(SHUT_WR)` in POSIX clients; `socket.end()` in Node) and only then await the response. Merely calling `write()` and waiting is invalid: the broker waits for EOF to enforce the exact-one-frame contract and fails closed at its one-second deadline. Do not replace `end()` with `write()` in client helpers. A delayed second frame remains malformed and is rejected.
 
-There is no automated recovery workflow yet. `mosaic_context_recover` is reserved as the only unverified mutator class, but its fixed payload/receipt implementation lands in a later WI. After a runtime exits, its `generation-<session>.state` file may be removed only after verifying that no process for that broker-minted session remains; stale files carry no lease authority but should be retained during incident analysis. After a broker crash, preserve the protected state file and restart only after verifying that no broker owns the socket. Restart intentionally clears all volatile VERIFIED leases. A leftover socket requires an operator to verify the owning service is stopped and remove that exact socket deliberately. Corrupt, oversized, symlinked, or non-regular state fails closed; do not overwrite it. Preserve it for incident review and establish new state only through an explicit operational decision, which invalidates prior sessions and tokens.
+`mosaic_context_recover` is the only unverified mutator class. Its durable `mosaic-context-refresh` skill is a thin wrapper over `tools/lease-broker/recover-context.py`: `begin` has the broker rebuild the validated `B_payload`/`H_payload`, revoke first, and mint a new `PENDING_DELIVERY` receipt challenge; `complete` accepts neither receipt text nor a challenge argument. Claude maps only the exact direct recovery executable/validated arguments to this exempt tool identity; ordinary `Bash` remains gated. Pi exposes only the `mosaic_context_recover` custom tool; ordinary `bash` and all other tools remain gated. A normal-path receipt cannot be replayed through recovery because each retry begins a distinct recovery cycle and recovery completion cannot receive caller-presented evidence.
+
+Production daemon startup creates a separate private observer socket unless a test-only `--test-observer-file` fixture is selected. Claude's Stop hook sends its exact latest assistant entry and Pi's `message_end` handler sends only finalized assistant content to that authenticated transport; the broker public socket never accepts message text. This is byte-build and private out-of-process harness wiring only: do not activate it against a live daemon, live socket, systemd service, tmux session, or model-output stream outside the controlled integration procedure.
+
+Receipt honesty is load-bearing: absent, malformed, prefix-truncated, and observable adapter-mutated terminal receipts do not promote. A tail-only case is non-promoting only where the concrete terminal payload is malformed or observably incomplete. A tail-preserving middle drop is **not receipt-detectable**; it is the disclosed T-C injection-contract residual deferred to WI-7 server-side evidence. The receipt remains a T-A delivery/liveness prerequisite, never a safety, obedience, or residency proof. The framework skill is source-resident and bridge-projected on install/upgrade; do not hand-create a live runtime symlink.
+
+After a runtime exits, its `generation-<session>.state` file may be removed only after verifying that no process for that broker-minted session remains; stale files carry no lease authority but should be retained during incident analysis. After a broker crash, preserve the protected state file and restart only after verifying that no broker owns the socket. Restart intentionally clears all volatile VERIFIED leases. A leftover socket requires an operator to verify the owning service is stopped and remove that exact socket deliberately. Corrupt, oversized, symlinked, or non-regular state fails closed; do not overwrite it. Preserve it for incident review and establish new state only through an explicit operational decision, which invalidates prior sessions and tokens.
 
 ## Security posture
 

docs/scratchpads/833-constrained-recovery-command.md

@@ -0,0 +1,15 @@
+# #833 constrained recovery command — build scratchpad
+
+- **Objective:** Deliver WI-6 plus Mos-ruled B1/B2 and R2 Claude-only literal-argv repair: no shell-active recovery mapping bypass, unchanged Pi gate/B2 observer, AC-1/C4 preservation, 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/model-stream activation; 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).
+- **Evidence:** Original RED test committed at `3b5513bd6efad0c06b599fc759a66cff4286db04`; expected `UNKNOWN_ACTION` is logged in `/home/hermes/agent-work/reviews/833-wi6-red.log`. B1/B2 repair RED is `f4beedc3e7ac2e142dcbdeefb0e5ee40c20d9b86` in `/home/hermes/agent-work/reviews/833-wi6-repair-red.log`. R2 adversarial RED is committed at `65e2bd71cf360b6f45c86eec0b06ae24494832d8` in `/home/hermes/agent-work/reviews/833-wi6-repair-R2-red.log` before the literal-only gate source: the private real-gate/real-daemon battery covers every argv position (executable, path, phase, each flag, each value) for command substitution, backticks, parameter/arithmetic expansion, brace/tilde, process substitution, glob, redirects, control operations, embedded newline, and quotes. P6 remains rebuilt and unfired. The ordinary package Vitest/lint/typecheck/root-format commands cannot resolve their executables in this intentionally dependency-free fresh worktree (`node_modules` absent); no install/symlink workaround was used.
+- **Budget:** No explicit task token cap was supplied; scope is fixed to WI-6 and no unrelated behavior will be added.

packages/mosaic/framework/framework-manifest.txt

@@ -35,6 +35,8 @@ CONTRIBUTING.md
 defaults/**
 examples/**
 guides/**
+# Shipped framework subtree — canonical skills are upgrade-reconciled.
+skills/**
 install.sh
 install.ps1
 LICENSE
@@ -67,6 +69,9 @@ policy/**
 memory/**
 sources/**
 credentials/**
+# Operator-authored/customized skills live separately from canonical skills/ and
+# must remain structurally unprunable even as skills/** is framework-owned.
+skills-local/**
 # Secret-bearing operator file INSIDE the framework-owned tools/ subtree.
 # Listed explicitly so the deny-wins rule carves it out of tools/**.
 tools/_lib/credentials.json

packages/mosaic/framework/runtime/claude/settings.json

@@ -38,7 +38,7 @@
         "hooks": [
           {
             "type": "command",
-            "command": "python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude",
+            "command": "python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude --recovery-command ~/.config/mosaic/tools/lease-broker/recover-context.py",
             "timeout": 3
           }
         ]
@@ -79,6 +79,11 @@
     "Stop": [
       {
         "hooks": [
+          {
+            "type": "command",
+            "command": "python3 ~/.config/mosaic/tools/lease-broker/receipt-observer-client.py --runtime claude --latest-entry",
+            "timeout": 3
+          },
           {
             "type": "command",
             "command": "~/.config/mosaic/tools/qa/reflect-stop-hook.sh",

packages/mosaic/framework/runtime/pi/mosaic-extension.ts

@@ -31,6 +31,14 @@ import { registerLeaseLifecycleHooks, type LeaseLifecyclePiApi } from './lease-l
 const MOSAIC_HOME = process.env['MOSAIC_HOME'] ?? join(homedir(), '.config', 'mosaic');
 const MUTATOR_GATE = join(MOSAIC_HOME, 'tools', 'lease-broker', 'mutator-gate.py');
 const LEASE_REVOKER = join(MOSAIC_HOME, 'tools', 'lease-broker', 'revoke-lease.py');
+const RECOVERY_COMMAND = join(MOSAIC_HOME, 'tools', 'lease-broker', 'recover-context.py');
+const RECEIPT_OBSERVER_CLIENT = join(
+  MOSAIC_HOME,
+  'tools',
+  'lease-broker',
+  'receipt-observer-client.py',
+);
+const RECOVERY_TOOL = 'mosaic_context_recover';
 
 // ---------------------------------------------------------------------------
 // Helpers
@@ -135,6 +143,78 @@ function checkPiMutatorGate(toolName: string): { block: true; reason: string } |
   };
 }
 
+function checkPiRecoveryGate(): { block: true; reason: string } | undefined {
+  return checkPiMutatorGate(RECOVERY_TOOL);
+}
+
+function assistantMessageText(message: unknown): string | undefined {
+  if (typeof message !== 'object' || message === null) return undefined;
+  const value = message as { role?: unknown; content?: unknown };
+  if (value.role !== 'assistant') return undefined;
+  if (typeof value.content === 'string') return value.content;
+  if (!Array.isArray(value.content)) return undefined;
+  const text: string[] = [];
+  for (const part of value.content) {
+    if (typeof part !== 'object' || part === null) return undefined;
+    const typed = part as { type?: unknown; text?: unknown };
+    if (typed.type !== 'text' || typeof typed.text !== 'string') return undefined;
+    text.push(typed.text);
+  }
+  return text.join('');
+}
+
+function recordPiMessageEnd(message: unknown): void {
+  const latestAssistantMessage = assistantMessageText(message);
+  if (latestAssistantMessage === undefined) return;
+  // This sends finalized Pi message_end content only to the daemon-owned
+  // authenticated observer transport, never to the public broker request API.
+  spawnSync('python3', [RECEIPT_OBSERVER_CLIENT, '--runtime', 'pi'], {
+    input: `${JSON.stringify({ latest_assistant_message: latestAssistantMessage })}\n`,
+    encoding: 'utf8',
+    timeout: 2_000,
+    env: process.env,
+  });
+}
+
+function runPiRecoveryCommand(params: {
+  phase: 'begin' | 'complete';
+  construction?: string;
+  compactionEpoch?: number;
+  requestEpoch?: number;
+}): { content: Array<{ type: 'text'; text: string }> } {
+  const args = [RECOVERY_COMMAND, params.phase];
+  if (params.phase === 'begin') {
+    if (
+      typeof params.construction !== 'string' ||
+      !Number.isInteger(params.compactionEpoch) ||
+      !Number.isInteger(params.requestEpoch) ||
+      params.compactionEpoch < 0 ||
+      params.requestEpoch < 0
+    ) {
+      return {
+        content: [
+          { type: 'text', text: 'Recovery begin requires construction and non-negative epochs.' },
+        ],
+      };
+    }
+    args.push(
+      '--construction',
+      params.construction,
+      '--compaction-epoch',
+      String(params.compactionEpoch),
+      '--request-epoch',
+      String(params.requestEpoch),
+    );
+  }
+  const result = spawnSync('python3', args, {
+    encoding: 'utf8',
+    timeout: 3_000,
+    env: process.env,
+  });
+  const output = result.status === 0 ? String(result.stdout ?? '') : String(result.stderr ?? '');
+  return { content: [{ type: 'text', text: output || 'Constrained recovery refused.' }] };
+}
+
 // ---------------------------------------------------------------------------
 // Mission detection
 // ---------------------------------------------------------------------------
@@ -287,6 +367,32 @@ export default function register(pi: ExtensionAPI) {
   // class gate before execution. Broker/script failure blocks fail-closed.
   pi.on('tool_call', async (event) => checkPiMutatorGate(event.toolName));
 
+  // Pi records only a finalized assistant entry at message_end. It never uses
+  // after_provider_response, which occurs before stream consumption.
+  pi.on('message_end', async (event) => {
+    recordPiMessageEnd((event as unknown as { message?: unknown }).message);
+  });
+
+  // The recovery custom tool is the only Pi invocation that maps to the
+  // broker's exempt RECOVERY_TOOL identity. It is not a Bash exception.
+  pi.registerTool({
+    name: RECOVERY_TOOL,
+    label: 'Mosaic Context Recovery',
+    description:
+      'Run the constrained broker-backed context recovery flow. This is the sole ungated mutator.',
+    parameters: Type.Object({
+      phase: Type.Union([Type.Literal('begin'), Type.Literal('complete')]),
+      construction: Type.Optional(Type.String()),
+      compactionEpoch: Type.Optional(Type.Integer({ minimum: 0 })),
+      requestEpoch: Type.Optional(Type.Integer({ minimum: 0 })),
+    }),
+    async execute(_toolCallId, params) {
+      const blocked = checkPiRecoveryGate();
+      if (blocked !== undefined) return { content: [{ type: 'text', text: blocked.reason }] };
+      return runPiRecoveryCommand(params);
+    },
+  });
+
   // ── Session Start ─────────────────────────────────────────────────────
   pi.on('session_start', async (_event, ctx) => {
     sessionCwd = process.cwd();

packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md

@@ -0,0 +1,65 @@
+---
+name: mosaic-context-refresh
+description: Run the constrained Mosaic context-recovery flow after compaction or directive-loss. This is a thin wrapper over the broker-backed recovery command; it never treats a receipt as a safety or residency proof.
+---
+
+# mosaic-context-refresh
+
+Use this only after compaction, session resume, or confirmed directive drift. It invokes the
+**single ungated mutator**, `tools/lease-broker/recover-context.py`; every other consequential
+mutator remains behind the verified lease gate.
+
+## Wrapper procedure
+
+1. The runtime supplies the exact validated normative-fragment construction and the current
+   compaction/request epochs.
+   - **Claude:** invoke only this direct command shape (no shell composition):
+
+     ```bash
+     python3 /absolute/path/to/mosaic/tools/lease-broker/recover-context.py begin --construction /absolute/path/to/mosaic-context-refresh-construction.json --compaction-epoch 0 --request-epoch 0
+     ```
+
+     This is a literal argv template: replace the recover-context.py path and construction JSON path
+     with the literal absolute paths for your install, then replace each epoch with literal decimal
+     digits. Do not use variables, quoting, globs, redirects,
+     shell operators, substitutions, or line continuations. Claude's all-tools gate maps only this
+     fully literal recovery shape to `mosaic_context_recover`; ordinary `Bash` remains gated.
+
+   - **Pi:** call the registered `mosaic_context_recover` tool with `phase: "begin"`,
+     `construction`, `compactionEpoch`, and `requestEpoch`. It is the exact broker-exempt tool name;
+     Pi `bash` and every other tool remain gated.
+
+   Both forms delegate to the shipped WI-5 broker transition: revoke first, build the canonical
+   `B_payload`/`H_payload`, enter `PENDING_DELIVERY`, and mint a fresh one-time challenge. They print
+   the terminal receipt envelope to deliver exactly as returned.
+
+2. The current assistant message copies that one terminal receipt verbatim. It does not compute a
+   hash, add prose, quote a prior receipt, or present a caller-supplied receipt/challenge.
+3. The production trusted-observer transport records that finalized assistant entry before completion:
+   - **Claude** selects the latest assistant entry at its `Stop` hook.
+   - **Pi** records only finalized assistant content at `message_end` (never
+     `after_provider_response`).
+
+   Then invoke completion with the same adapter form: Claude runs
+   `python3 /absolute/path/to/mosaic/tools/lease-broker/recover-context.py complete`; Pi calls
+   `mosaic_context_recover` with `phase: "complete"`. Completion supplies no receipt or challenge
+   argument. The broker observes the exact latest assistant entry, commits evidence, consumes its own
+   fresh challenge, and promotes VERIFIED last. If observation is absent, malformed, stale, or
+   duplicated, recovery remains UNVERIFIED and a retry begins a new cycle.
+
+## Scope and honesty
+
+- A receipt from the normal verification path cannot be replayed through recovery: recovery mints a
+  distinct current challenge and does not accept caller-provided receipt text as evidence.
+- Observable absent, malformed, prefix-truncated, and adapter-mutated terminal receipts do not
+  promote. “Tail-only” is non-promoting only when the delivered terminal bytes are concretely
+  malformed or incomplete.
+- **Negative capability:** a tail-preserving middle drop is not represented as receipt-detectable.
+  It is a T-C injection-contract residual deferred to WI-7 server-side evidence; do not claim this
+  skill or receipt catches it.
+- The receipt is a T-A delivery/liveness prerequisite only. It never proves obedience, comprehension,
+  durable residency, or safety; the whole mutator-class gate and server-side branch protection retain
+  those roles.
+
+This source-resident skill is projected by the Mosaic skill bridge after framework install/upgrade.
+Do not create a live symlink manually.

packages/mosaic/framework/tools/lease-broker/daemon.py

@@ -11,6 +11,7 @@ from concurrent.futures import ThreadPoolExecutor
 import json
 import os
 import secrets
+import select
 import signal
 import socket
 import stat
@@ -28,7 +29,11 @@ if _MODULE_DIRECTORY not in sys.path:
     sys.path.insert(0, _MODULE_DIRECTORY)
 from normative_fragments import build_payload_from_wire
 from receipt_challenge import is_verbatim_receipt, latest_assistant_digest, receipt_for
-from receipt_observer import FileTestReceiptObserver, ReceiptObserver, UnavailableReceiptObserver
+from receipt_observer import (
+    FileTestReceiptObserver,
+    ReceiptObserver,
+    RuntimeReceiptObserver,
+)
 
 MAX_FRAME: Final = 64 * 1024
 MAX_STATE: Final = 4 * 1024 * 1024
@@ -302,7 +307,9 @@ class StateStore:
 class Broker:
     def __init__(self, store: StateStore, observer: ReceiptObserver | None = None) -> None:
         self.store = store
-        self.observer: ReceiptObserver = observer if observer is not None else UnavailableReceiptObserver()
+        # Production construction always has a transport-capable observer. Test
+        # fixtures may inject their controlled observer explicitly.
+        self.observer: ReceiptObserver = observer if observer is not None else RuntimeReceiptObserver()
         # Set only by begin_verification after its mandatory revoke-first fence.
         # It is preserved if later cycle admission is refused; all other broker
         # actions retain the normal snapshot rollback behavior.
@@ -390,6 +397,40 @@ class Broker:
             raise BrokerFailure("STATE_INTEGRITY")
         lease["state"] = LEASE_VERIFIED
 
+    def record_runtime_observation(
+        self, peer_pid: int, request: dict[str, object]
+    ) -> dict[str, object]:
+        """Record only an authenticated adapter's finalized assistant message.
+
+        This method is deliberately unreachable through ``Broker.handle`` and
+        its public broker socket. The production observer socket calls it after
+        SO_PEERCRED/ancestry authentication, preserving the S1 rule that a
+        broker request can never carry ``latest_assistant_message``.
+        """
+
+        required = {
+            "action", "session_id", "runtime_generation", "runtime", "latest_assistant_message"
+        }
+        if set(request) != required or request.get("action") != "record_runtime_observation":
+            raise BrokerFailure("INVALID_OBSERVATION")
+        runtime = request.get("runtime")
+        message = request.get("latest_assistant_message")
+        if runtime not in READ_ONLY_TOOLS or not isinstance(message, str) or len(message.encode("utf-8")) > MAX_FRAME:
+            raise BrokerFailure("INVALID_OBSERVATION")
+        session_id, _ = self.authenticate(peer_pid, request)
+        generation = request["runtime_generation"]
+        lease = self.leases.get(session_id)
+        if (
+            not isinstance(lease, dict)
+            or lease.get("state") != LEASE_PENDING
+            or lease.get("runtime") != runtime
+            or lease.get("runtime_generation") != generation
+            or not isinstance(self.observer, RuntimeReceiptObserver)
+        ):
+            raise BrokerFailure("OBSERVATION_UNAVAILABLE")
+        self.observer.record_latest_assistant_message(session_id, runtime, generation, message)
+        return {"ok": True}
+
     def handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
         if self.store.poisoned:
             raise StateCommitUncertain()
@@ -400,7 +441,7 @@ class Broker:
             response = self._handle(peer, request)
             if self.store.value != previous:
                 self.store.commit()
-            if request.get("action") == "promote_lease":
+            if request.get("action") in {"promote_lease", "complete_recovery"}:
                 session_id = request.get("session_id")
                 if not isinstance(session_id, str):
                     raise BrokerFailure("INVALID_IDENTITY")
@@ -469,6 +510,71 @@ class Broker:
                 raise BrokerFailure("TOKEN_REPLAY")
             del self.store.tokens()[token_value]
             return {"ok": True}
+        if action == "begin_recovery":
+            # Recovery is the one ungated mutator, but it is not a second
+            # receipt protocol. It delegates to this exact normal-path
+            # transition, then marks its volatile pending cycle so completion
+            # can obtain the broker-minted challenge internally. Caller-supplied
+            # receipt text is never an input to recovery.
+            if any(field in request for field in ("receipt", "latest_assistant_message", "receipt_challenge")):
+                raise BrokerFailure("INVALID_RECOVERY_REQUEST")
+            normal_request = dict(request)
+            normal_request["action"] = "begin_verification"
+            response = self._handle(peer, normal_request)
+            session_id = response.get("session_id")
+            if not isinstance(session_id, str):
+                # begin_verification deliberately does not return identity;
+                # recover it only after its authenticated shared transition.
+                candidate = request.get("session_id")
+                if not isinstance(candidate, str):
+                    raise BrokerFailure("INVALID_IDENTITY")
+                session_id = candidate
+            lease = self.leases.get(session_id)
+            if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING:
+                raise BrokerFailure("STATE_INTEGRITY")
+            lease["cycle_kind"] = "recovery"
+            response["state"] = "PENDING_DELIVERY"
+            return response
+        if action == "complete_recovery":
+            if any(field in request for field in ("receipt", "latest_assistant_message", "receipt_challenge")):
+                raise BrokerFailure("INVALID_RECOVERY_REQUEST")
+            session_id, _ = self.authenticate(peer_pid, request)
+            lease = self.leases.get(session_id)
+            challenge = lease.get("receipt_challenge") if isinstance(lease, dict) else None
+            if (
+                not isinstance(lease, dict)
+                or lease.get("cycle_kind") != "recovery"
+                or lease.get("state") != LEASE_PENDING
+                or not isinstance(challenge, str)
+            ):
+                raise BrokerFailure("INVALID_LEASE_TRANSITION")
+            try:
+                # Reuse the shipped observe -> evidence commit -> consume ->
+                # promote transition. The recovery caller supplies neither a
+                # normal-path receipt nor a challenge; the trusted observer
+                # and current broker cycle remain the sole evidence authority.
+                observed_request = {
+                    "action": "observe_receipt",
+                    "session_id": session_id,
+                    "runtime_generation": request["runtime_generation"],
+                    "receipt_challenge": challenge,
+                }
+                self._handle(peer, observed_request)
+                return self._handle(peer, {
+                    "action": "promote_lease",
+                    "session_id": session_id,
+                    "runtime_generation": request["runtime_generation"],
+                    "receipt_challenge": challenge,
+                })
+            except Exception:
+                # A malformed, absent, or stale observed receipt must leave
+                # no pending recovery capability or live lease. A retry mints
+                # a new challenge through the shared begin transition.
+                self.revoke_session_authority(session_id)
+                self._rejected_cycle_fence = (
+                    copy.deepcopy(self.store.value), copy.deepcopy(self.leases)
+                )
+                raise
         if action == "begin_verification":
             session_id, _ = self.authenticate(peer_pid, request)
             runtime = request.get("runtime")
@@ -718,14 +824,63 @@ def handle_connection(
             return
 
 
+def handle_runtime_observation_connection(
+    connection: socket.socket,
+    broker: Broker,
+    broker_lock: threading.Lock,
+) -> None:
+    """Serve the authenticated production observer transport, never the broker API."""
+
+    with connection:
+        read_deadline = time.monotonic() + READ_DEADLINE_SECONDS
+        try:
+            raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
+            peer = struct.unpack("3i", raw)
+            request = read_frame(connection, read_deadline)
+        except BrokerFailure as exc:
+            reply = {"ok": False, "code": exc.code}
+        except OSError:
+            return
+        else:
+            acquired = broker_lock.acquire(timeout=HANDLE_QUEUE_TIMEOUT_SECONDS)
+            if not acquired:
+                reply = {"ok": False, "code": "BROKER_BUSY"}
+            else:
+                try:
+                    try:
+                        reply = broker.record_runtime_observation(peer[0], request)
+                    except BrokerFailure as exc:
+                        reply = {"ok": False, "code": exc.code}
+                finally:
+                    broker_lock.release()
+        try:
+            connection.settimeout(SEND_TIMEOUT_SECONDS)
+            connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
+        except OSError:
+            return
+
+
 def serve(
     socket_path: Path,
     state_path: Path,
     observer: ReceiptObserver | None = None,
+    observer_socket_path: Path | None = None,
 ) -> None:
     secure_parent(socket_path)
     if socket_path.exists() or socket_path.is_symlink():
         raise BrokerFailure("SOCKET_ALREADY_EXISTS")
+    runtime_observer = observer is None
+    if runtime_observer:
+        observer = RuntimeReceiptObserver()
+        observer_socket_path = observer_socket_path or socket_path.with_name("receipt-observer.sock")
+        if observer_socket_path == socket_path:
+            raise BrokerFailure("OBSERVER_SOCKET_CONFLICT")
+        secure_parent(observer_socket_path)
+        if observer_socket_path.exists() or observer_socket_path.is_symlink():
+            raise BrokerFailure("OBSERVER_SOCKET_ALREADY_EXISTS")
+    elif observer_socket_path is not None:
+        raise BrokerFailure("TEST_OBSERVER_SOCKET_CONFLICT")
+
     store = StateStore(state_path)
     broker = Broker(store, observer)
     broker_lock = threading.Lock()
@@ -740,16 +895,28 @@ def serve(
     server.bind(str(socket_path))
     os.chmod(socket_path, 0o600)
     owned = (socket_path.stat().st_dev, socket_path.stat().st_ino)
+    observer_server: socket.socket | None = None
+    observer_owned: tuple[int, int] | None = None
+    if runtime_observer and observer_socket_path is not None:
+        observer_server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        observer_server.bind(str(observer_socket_path))
+        os.chmod(observer_socket_path, 0o600)
+        observer_owned = (observer_socket_path.stat().st_dev, observer_socket_path.stat().st_ino)
     stopping = False
 
     def stop(_signum: int, _frame: object) -> None:
         nonlocal stopping
         stopping = True
         server.close()
+        if observer_server is not None:
+            observer_server.close()
 
-    def process_connection(connection: socket.socket) -> None:
+    def process_connection(connection: socket.socket, is_observer: bool) -> None:
         try:
-            handle_connection(connection, broker, broker_lock)
+            if is_observer:
+                handle_runtime_observation_connection(connection, broker, broker_lock)
+            else:
+                handle_connection(connection, broker, broker_lock)
         except Exception as exc:
             with fatal_lock:
                 if not fatal_errors:
@@ -764,57 +931,73 @@ def serve(
     signal.signal(signal.SIGTERM, stop)
     signal.signal(signal.SIGINT, stop)
     server.listen(MAX_IN_FLIGHT_CONNECTIONS)
-    server.settimeout(0.1)
+    server.setblocking(False)
+    if observer_server is not None:
+        observer_server.listen(MAX_IN_FLIGHT_CONNECTIONS)
+        observer_server.setblocking(False)
     print("READY", flush=True)
     try:
         while not stopping:
             failure = fatal_error()
             if failure is not None:
                 raise failure
-            if not slots.acquire(timeout=0.1):
-                continue
+            listeners = [server, *([observer_server] if observer_server is not None else [])]
             try:
-                connection, _ = server.accept()
-            except socket.timeout:
-                slots.release()
-                continue
-            except OSError:
-                slots.release()
-                failure = fatal_error()
-                if failure is not None:
-                    raise failure
+                ready, _, _ = select.select(listeners, [], [], 0.1)
+            except (OSError, ValueError):
                 if stopping:
                     break
                 raise
-            try:
-                executor.submit(process_connection, connection)
-            except Exception:
-                slots.release()
-                connection.close()
-                raise
+            for listener in ready:
+                if not slots.acquire(blocking=False):
+                    continue
+                try:
+                    connection, _ = listener.accept()
+                except BlockingIOError:
+                    slots.release()
+                    continue
+                except OSError:
+                    slots.release()
+                    if stopping:
+                        break
+                    raise
+                try:
+                    executor.submit(process_connection, connection, listener is observer_server)
+                except Exception:
+                    slots.release()
+                    connection.close()
+                    raise
     finally:
         server.close()
+        if observer_server is not None:
+            observer_server.close()
         executor.shutdown(wait=True)
-        try:
-            current = socket_path.stat()
-            if (current.st_dev, current.st_ino) == owned:
-                socket_path.unlink()
-        except FileNotFoundError:
-            pass
+        for path, inode in ((socket_path, owned), (observer_socket_path, observer_owned)):
+            if path is None or inode is None:
+                continue
+            try:
+                current = path.stat()
+                if (current.st_dev, current.st_ino) == inode:
+                    path.unlink()
+            except FileNotFoundError:
+                pass
 
 
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--socket", required=True, type=Path)
     parser.add_argument("--state", required=True, type=Path)
+    parser.add_argument("--observer-socket", type=Path)
     parser.add_argument("--test-observer-file", type=Path)
     arguments = parser.parse_args()
+    if arguments.observer_socket is not None and arguments.test_observer_file is not None:
+        raise BrokerFailure("TEST_OBSERVER_SOCKET_CONFLICT")
     observer = (
         FileTestReceiptObserver(arguments.test_observer_file)
         if arguments.test_observer_file is not None
         else None
     )
-    serve(arguments.socket, arguments.state, observer)
+    serve(arguments.socket, arguments.state, observer, arguments.observer_socket)
 
 
 if __name__ == "__main__":

packages/mosaic/framework/tools/lease-broker/launch-runtime.py

@@ -97,6 +97,12 @@ def main(
     environment["MOSAIC_RUNTIME_GENERATION"] = str(generation)
     environment["MOSAIC_LEASE_GENERATION_FILE"] = str(generation_file)
     environment["MOSAIC_LEASE_RUNTIME"] = arguments.runtime
+    # Matches daemon.py's production default; deployments using a distinct
+    # observer socket may set this authenticated transport path explicitly.
+    environment.setdefault(
+        "MOSAIC_RECEIPT_OBSERVER_SOCKET",
+        str(socket_path.with_name("receipt-observer.sock")),
+    )
     try:
         execute(command[0], command, environment)
     except OSError:

packages/mosaic/framework/tools/lease-broker/mutator-gate.py

@@ -8,14 +8,23 @@ import json
 import os
 import socket
 import sys
+import re
 from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import BinaryIO, Final
 
+# Isolated (`python -I`) adapter invocations must still import co-located
+# framework modules; never depend on the caller's PYTHONPATH.
+_MODULE_DIRECTORY = str(Path(__file__).resolve().parent)
+if _MODULE_DIRECTORY not in sys.path:
+    sys.path.insert(0, _MODULE_DIRECTORY)
 from lease_generation import read_runtime_generation
 
 MAX_FRAME: Final = 64 * 1024
 BROKER_TIMEOUT_SECONDS: Final = 1.5
+RECOVERY_TOOL: Final = "mosaic_context_recover"
+_LITERAL_ABSOLUTE_PATH: Final = re.compile(r"/[A-Za-z0-9._/-]+\Z")
+_SHELL_ACTIVE: Final = frozenset("$`~*?[]{}<>;|&" + '"' + "'" + "\\" + "\n\r\t")
 
 
 def deny(code: str) -> int:
@@ -23,7 +32,7 @@ def deny(code: str) -> int:
     return 2
 
 
-def read_tool_name(stream: BinaryIO | None = None) -> str:
+def read_tool_request(stream: BinaryIO | None = None) -> dict[str, object]:
     source = sys.stdin.buffer if stream is None else stream
     raw = source.read(MAX_FRAME + 1)
     if len(raw) > MAX_FRAME:
@@ -34,7 +43,61 @@ def read_tool_name(stream: BinaryIO | None = None) -> str:
     tool_name = value.get("tool_name")
     if not isinstance(tool_name, str) or not tool_name or len(tool_name) > 256:
         raise ValueError("INVALID_GATE_INPUT")
-    return tool_name
+    return value
+
+
+def read_tool_name(stream: BinaryIO | None = None) -> str:
+    """Backward-compatible strict extraction for callers that need only the name."""
+
+    return str(read_tool_request(stream)["tool_name"])
+
+
+def recovery_invocation_name(request: dict[str, object], recovery_command: Path | None) -> str:
+    """Map only a byte-literal Claude recovery argv to ``RECOVERY_TOOL``.
+
+    Claude's Bash tool evaluates its raw command with a real shell. Therefore
+    the gate never attempts a second shell parser: any quote, expansion,
+    redirection, operator, glob, newline, or non-space whitespace is refused
+    before tokenizing. The remaining plain-space split is an exact argv proof,
+    not a best-effort interpretation of shell syntax.
+    """
+
+    tool_name = request["tool_name"]
+    if tool_name != "Bash" or recovery_command is None:
+        return str(tool_name)
+    tool_input = request.get("tool_input")
+    if not isinstance(tool_input, dict) or set(tool_input) != {"command"}:
+        return str(tool_name)
+    command = tool_input.get("command")
+    if (
+        not isinstance(command, str)
+        or not command
+        or any(character in _SHELL_ACTIVE for character in command)
+        or command.startswith(" ")
+        or command.endswith(" ")
+        or "  " in command
+    ):
+        return str(tool_name)
+    argv = command.split(" ")
+    if " ".join(argv) != command or len(argv) < 3 or argv[0] != "python3":
+        return str(tool_name)
+    if argv[1] != str(recovery_command):
+        return str(tool_name)
+    phase = argv[2]
+    if phase == "complete" and len(argv) == 3:
+        return RECOVERY_TOOL
+    if phase != "begin" or len(argv) != 9:
+        return str(tool_name)
+    if argv[3::2] != ["--construction", "--compaction-epoch", "--request-epoch"]:
+        return str(tool_name)
+    construction, compaction_epoch, request_epoch = argv[4::2]
+    if (
+        _LITERAL_ABSOLUTE_PATH.fullmatch(construction) is None
+        or not compaction_epoch.isdecimal()
+        or not request_epoch.isdecimal()
+    ):
+        return str(tool_name)
+    return RECOVERY_TOOL
 
 
 def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
@@ -70,11 +133,13 @@ def main(
 ) -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("--recovery-command", type=Path)
     arguments = parser.parse_args(argv)
     source_environment = os.environ if environ is None else environ
 
     try:
-        tool_name = read_tool_name(stream)
+        request_input = read_tool_request(stream)
+        tool_name = recovery_invocation_name(request_input, arguments.recovery_command)
         socket_value = source_environment["MOSAIC_LEASE_BROKER_SOCKET"]
         session_id = source_environment["MOSAIC_LEASE_SESSION_ID"]
         generation = resolve_generation(source_environment)

packages/mosaic/framework/tools/lease-broker/receipt-observer-client.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""Authenticated adapter-to-daemon transport for finalized assistant receipts.
+
+This is not a broker request client. It writes only to the daemon-owned observer
+socket, which authenticates SO_PEERCRED/ancestry before retaining a message for
+the broker's ReceiptObserver seam.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import socket
+import stat
+import sys
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Final
+
+
+MAX_FRAME: Final = 64 * 1024
+BROKER_TIMEOUT_SECONDS: Final = 1.5
+MAX_TRANSCRIPT_BYTES: Final = 4 * 1024 * 1024
+
+
+def read_json(stream: object) -> dict[str, object]:
+    raw = getattr(stream, "buffer", stream).read(MAX_FRAME + 1)
+    if not isinstance(raw, bytes) or len(raw) > MAX_FRAME:
+        raise ValueError("invalid observer input")
+    value = json.loads(raw)
+    if not isinstance(value, dict):
+        raise ValueError("invalid observer input")
+    return value
+
+
+def assistant_text(entry: object) -> str | None:
+    if not isinstance(entry, dict):
+        return None
+    message = entry.get("message", entry)
+    if not isinstance(message, dict) or message.get("role") != "assistant":
+        return None
+    content = message.get("content")
+    if isinstance(content, str):
+        return content
+    if not isinstance(content, list):
+        return None
+    parts: list[str] = []
+    for item in content:
+        if not isinstance(item, dict) or item.get("type") != "text" or not isinstance(item.get("text"), str):
+            return None
+        parts.append(item["text"])
+    return "".join(parts)
+
+
+def claude_latest_entry(value: dict[str, object]) -> str:
+    transcript_path = value.get("transcript_path")
+    if not isinstance(transcript_path, str) or not transcript_path:
+        raise ValueError("invalid Claude observer input")
+    path = Path(transcript_path)
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(path, flags)
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_TRANSCRIPT_BYTES:
+            raise ValueError("unsafe Claude transcript")
+        raw = os.read(descriptor, MAX_TRANSCRIPT_BYTES + 1)
+    finally:
+        os.close(descriptor)
+    if len(raw) > MAX_TRANSCRIPT_BYTES:
+        raise ValueError("oversized Claude transcript")
+    for line in reversed(raw.decode("utf-8").splitlines()):
+        try:
+            text = assistant_text(json.loads(line))
+        except json.JSONDecodeError as exc:
+            raise ValueError("invalid Claude transcript") from exc
+        if text is not None:
+            return text
+    raise ValueError("Claude transcript has no assistant entry")
+
+
+def pi_message_end(value: dict[str, object]) -> str:
+    if set(value) != {"latest_assistant_message"} or not isinstance(value["latest_assistant_message"], str):
+        raise ValueError("invalid Pi observer input")
+    return value["latest_assistant_message"]
+
+
+def observer_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
+    payload = (json.dumps(request, separators=(",", ":")) + "\n").encode()
+    if len(payload) > MAX_FRAME:
+        raise ValueError("observer request too large")
+    response = bytearray()
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(BROKER_TIMEOUT_SECONDS)
+        connection.connect(str(socket_path))
+        connection.sendall(payload)
+        connection.shutdown(socket.SHUT_WR)
+        while len(response) <= MAX_FRAME:
+            chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(response)))
+            if not chunk:
+                break
+            response.extend(chunk)
+    if len(response) > MAX_FRAME or not response.endswith(b"\n"):
+        raise ValueError("invalid observer reply")
+    value = json.loads(response)
+    if not isinstance(value, dict):
+        raise ValueError("invalid observer reply")
+    return value
+
+
+def main(argv: Sequence[str] | None = None, *, environ: Mapping[str, str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("--latest-entry", action="store_true")
+    arguments = parser.parse_args(argv)
+    source_environment = os.environ if environ is None else environ
+    try:
+        source = read_json(sys.stdin)
+        if arguments.runtime == "claude":
+            if not arguments.latest_entry:
+                raise ValueError("Claude observer requires --latest-entry")
+            message = claude_latest_entry(source)
+        else:
+            if arguments.latest_entry:
+                raise ValueError("Pi observer is message_end only")
+            message = pi_message_end(source)
+        if len(message.encode("utf-8")) > MAX_FRAME:
+            raise ValueError("assistant message too large")
+        reply = observer_request(Path(source_environment["MOSAIC_RECEIPT_OBSERVER_SOCKET"]), {
+            "action": "record_runtime_observation",
+            "session_id": source_environment["MOSAIC_LEASE_SESSION_ID"],
+            "runtime_generation": int(source_environment["MOSAIC_RUNTIME_GENERATION"]),
+            "runtime": arguments.runtime,
+            "latest_assistant_message": message,
+        })
+    except (KeyError, OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Mosaic receipt observer refused: {error}", file=sys.stderr)
+        return 2
+    return 0 if reply == {"ok": True} else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

packages/mosaic/framework/tools/lease-broker/receipt_observer.py

@@ -1,9 +1,11 @@
 #!/usr/bin/env python3
 """Trusted latest-assistant-message observer boundary for receipt promotion.
 
-Runtime adapters must implement ``observe_latest_assistant_message`` directly:
-Claude selects the exact latest assistant entry and Pi selects ``message_end``.
-The broker accepts no observed message through its request protocol.
+Production adapters deliver finalized assistant content over the daemon-owned
+observer socket after the daemon authenticates their peer against the broker's
+kernel-anchored session identity. The broker request protocol never accepts
+assistant-message content. Claude supplies its latest assistant entry; Pi
+supplies finalized assistant content at ``message_end``.
 """
 
 from __future__ import annotations
@@ -26,7 +28,7 @@ class ReceiptObserver(Protocol):
 
 
 class UnavailableReceiptObserver:
-    """Production-safe default until a runtime adapter injects an observer."""
+    """Fail-closed only for direct unit construction without daemon transport."""
 
     def observe_latest_assistant_message(
         self,
@@ -38,6 +40,31 @@ class UnavailableReceiptObserver:
         return None
 
 
+class RuntimeReceiptObserver:
+    """Daemon-owned production observer populated only by authenticated adapters."""
+
+    def __init__(self) -> None:
+        self._messages: dict[tuple[str, str, int], str] = {}
+
+    def record_latest_assistant_message(
+        self,
+        session_id: str,
+        runtime: str,
+        runtime_generation: int,
+        message: str,
+    ) -> None:
+        self._messages[(session_id, runtime, runtime_generation)] = message
+
+    def observe_latest_assistant_message(
+        self,
+        session_id: str,
+        runtime: str,
+        runtime_generation: int,
+        _binding: dict[str, object],
+    ) -> str | None:
+        return self._messages.get((session_id, runtime, runtime_generation))
+
+
 class TestReceiptObserver:
     """Deterministic controlled observer used only by byte-build tests."""
 
@@ -60,7 +87,7 @@ class TestReceiptObserver:
 
 
 class FileTestReceiptObserver:
-    """Private fixture-file observer for isolated out-of-process test drivers."""
+    """Private fixture-file observer for isolated out-of-process test drivers only."""
 
     def __init__(self, path: Path) -> None:
         self.path = path

packages/mosaic/framework/tools/lease-broker/recover-context.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""Constrained recovery command: the sole ungated Mosaic mutator.
+
+This is deliberately a thin client of the broker's recovery entrypoint.  It
+never accepts receipt text or a caller-provided challenge: the broker mints the
+fresh challenge, delivers its exact receipt envelope, and later asks the
+trusted ReceiptObserver seam to observe that same pending cycle.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import socket
+import sys
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Final
+
+# The out-of-process recovery command is intentionally runnable with `python
+# -I`; locate its shipped construction module without caller-controlled paths.
+_MODULE_DIRECTORY = str(Path(__file__).resolve().parent)
+if _MODULE_DIRECTORY not in sys.path:
+    sys.path.insert(0, _MODULE_DIRECTORY)
+from normative_fragments import build_payload_from_wire
+
+MAX_FRAME: Final = 64 * 1024
+BROKER_TIMEOUT_SECONDS: Final = 1.5
+
+
+def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
+    payload = (json.dumps(request, separators=(",", ":")) + "\n").encode()
+    if len(payload) > MAX_FRAME:
+        raise ValueError("recovery request too large")
+    response = bytearray()
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(BROKER_TIMEOUT_SECONDS)
+        connection.connect(str(socket_path))
+        connection.sendall(payload)
+        connection.shutdown(socket.SHUT_WR)
+        while len(response) <= MAX_FRAME:
+            chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(response)))
+            if not chunk:
+                break
+            response.extend(chunk)
+    if len(response) > MAX_FRAME or not response.endswith(b"\n"):
+        raise ValueError("invalid broker reply")
+    value = json.loads(response)
+    if not isinstance(value, dict):
+        raise ValueError("invalid broker reply")
+    return value
+
+
+def load_construction(path: Path) -> dict[str, object]:
+    raw = path.read_bytes()
+    if len(raw) > MAX_FRAME:
+        raise ValueError("construction exceeds broker frame limit")
+    value = json.loads(raw)
+    if not isinstance(value, dict):
+        raise ValueError("construction must be an object")
+    return value
+
+
+def identity(environ: Mapping[str, str]) -> tuple[Path, str, int, str]:
+    socket_path = Path(environ["MOSAIC_LEASE_BROKER_SOCKET"])
+    session_id = environ["MOSAIC_LEASE_SESSION_ID"]
+    generation = int(environ["MOSAIC_RUNTIME_GENERATION"])
+    runtime = environ["MOSAIC_LEASE_RUNTIME"]
+    if generation < 0 or runtime not in {"claude", "pi"}:
+        raise ValueError("invalid runtime identity")
+    return socket_path, session_id, generation, runtime
+
+
+def begin(
+    construction_path: Path,
+    compaction_epoch: int,
+    request_epoch: int,
+    environ: Mapping[str, str],
+) -> dict[str, object]:
+    if compaction_epoch < 0 or request_epoch < 0:
+        raise ValueError("epochs must be non-negative")
+    construction = load_construction(construction_path)
+    # Invoke the shared WI-5 construction before asking the broker to repeat
+    # its authoritative admission/build. No digest or receipt enters via CLI.
+    built = build_payload_from_wire(construction)
+    if (
+        built.injectionDecision != "ACCEPTED"
+        or not built.promotion
+        or not isinstance(built.h_source, str)
+        or not isinstance(built.h_payload, str)
+    ):
+        raise ValueError("payload construction refused")
+    socket_path, session_id, generation, runtime = identity(environ)
+    return broker_request(socket_path, {
+        "action": "begin_recovery",
+        "session_id": session_id,
+        "runtime_generation": generation,
+        "runtime": runtime,
+        "binding": {
+            "compaction_epoch": compaction_epoch,
+            "request_epoch": request_epoch,
+            "h_source": built.h_source,
+            "h_payload": built.h_payload,
+            "schema_version": 1,
+        },
+        "construction": construction,
+    })
+
+
+def complete(environ: Mapping[str, str]) -> dict[str, object]:
+    socket_path, session_id, generation, _runtime = identity(environ)
+    # No receipt or challenge argument exists: recovery completion can only use
+    # the broker's current recovery cycle and its trusted observer seam.
+    return broker_request(socket_path, {
+        "action": "complete_recovery",
+        "session_id": session_id,
+        "runtime_generation": generation,
+    })
+
+
+def main(argv: Sequence[str] | None = None, *, environ: Mapping[str, str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    subcommands = parser.add_subparsers(dest="phase", required=True)
+    begin_parser = subcommands.add_parser("begin", help="mint and deliver a fresh recovery receipt")
+    begin_parser.add_argument("--construction", required=True, type=Path)
+    begin_parser.add_argument("--compaction-epoch", required=True, type=int)
+    begin_parser.add_argument("--request-epoch", required=True, type=int)
+    subcommands.add_parser("complete", help="observe and promote only the current recovery receipt")
+    arguments = parser.parse_args(argv)
+    source_environment = os.environ if environ is None else environ
+    try:
+        reply = (
+            begin(
+                arguments.construction,
+                arguments.compaction_epoch,
+                arguments.request_epoch,
+                source_environment,
+            )
+            if arguments.phase == "begin"
+            else complete(source_environment)
+        )
+    except (KeyError, OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Mosaic constrained recovery refused: {error}", file=sys.stderr)
+        return 2
+    print(json.dumps(reply, separators=(",", ":")))
+    return 0 if reply.get("ok") is True else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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/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/lease-broker/recovery_b1_adversarial_unittest.py && python3 src/lease-broker/framework_skill_portability_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/commands/skill.spec.ts

@@ -8,6 +8,7 @@ import {
   mkdirSync,
   mkdtempSync,
   readlinkSync,
+  readFileSync,
   rmSync,
   symlinkSync,
   writeFileSync,
@@ -26,6 +27,9 @@ import {
 const LEGACY_SYNC_SCRIPT = fileURLToPath(
   new URL('../../framework/tools/_scripts/mosaic-sync-skills', import.meta.url),
 );
+const CONTEXT_REFRESH_SOURCE_SKILL = fileURLToPath(
+  new URL('../../framework/skills/mosaic-context-refresh/SKILL.md', import.meta.url),
+);
 
 describe('Claude skill bridge', () => {
   let root: string;
@@ -360,6 +364,25 @@ describe('Claude skill bridge', () => {
   });
 
   describe('syncClaudeSkills', () => {
+    it('projects the durable context-refresh skill through the #824 bridge in a tmp root', () => {
+      expect(existsSync(CONTEXT_REFRESH_SOURCE_SKILL)).toBe(true);
+      const source = readFileSync(CONTEXT_REFRESH_SOURCE_SKILL, 'utf8');
+      expect(source).toContain('recover-context.py');
+      expect(source).toContain('middle drop is not represented as receipt-detectable');
+      const skillDir = createSkill('mosaic-context-refresh');
+      writeFileSync(join(skillDir, 'SKILL.md'), source);
+
+      const result = syncClaudeSkills(paths);
+
+      expect(result).toEqual({
+        registered: ['mosaic-context-refresh'],
+        repaired: [],
+        unchanged: [],
+        conflicts: [],
+      });
+      expectCorrectLink('mosaic-context-refresh');
+    });
+
     it('generically creates every missing canonical link and repairs managed broken links', () => {
       createSkill('added-after-setup');
       createSkill('another-new-skill');

packages/mosaic/src/framework/manifest-parity.spec.ts

@@ -78,6 +78,7 @@ const PROBE_PATHS = [
   'install.sh',
   'framework-manifest.txt',
   'guides/E2E-DELIVERY.md',
+  'skills/mosaic-context-refresh/SKILL.md',
   'tools/git/pr-create.sh',
   'tools/_lib/manifest.sh',
   'defaults/SOUL.md',
@@ -97,6 +98,7 @@ const PROBE_PATHS = [
   'memory/note.md',
   'sources/skills/x.md',
   'credentials/c.json',
+  'skills-local/custom/SKILL.md',
   'tools/_lib/credentials.json',
   'fleet/roster.yaml',
   'fleet/roster.json',

packages/mosaic/src/framework/manifest.spec.ts

@@ -304,6 +304,11 @@ describe('manifest completeness against shipped framework tree', () => {
     expect(misclassified).toEqual([]);
   });
 
+  it('shipped canonical skills are framework-owned while local skills are operator-owned', () => {
+    expect(resolveOwnership(manifest, 'skills/mosaic-context-refresh/SKILL.md')).toBe('framework');
+    expect(resolveOwnership(manifest, 'skills-local/custom/SKILL.md')).toBe('operator');
+  });
+
   it('the operator-owned surface from #791 resolves to operator', () => {
     const operatorPaths = [
       'agents/coder0.conf',
@@ -313,6 +318,7 @@ describe('manifest completeness against shipped framework tree', () => {
       'SOUL.local.md',
       'USER.local.md',
       'STANDARDS.local.md',
+      'skills-local/custom/SKILL.md',
       'tools/_lib/credentials.json',
       'fleet/roster.yaml',
       'fleet/roster.json',

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()

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

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""RED-first framework-firewall and portability contracts for shipped skills."""
+
+from __future__ import annotations
+
+import importlib.util
+import re
+import unittest
+from pathlib import Path
+
+
+MOSAIC = Path(__file__).parents[2]
+SKILLS = MOSAIC / "framework/skills"
+REFRESH_SKILL = SKILLS / "mosaic-context-refresh/SKILL.md"
+GATE_PATH = MOSAIC / "framework/tools/lease-broker/mutator-gate.py"
+OPERATOR_HOME = re.compile(r"/home/[^/\s]+/")
+RECOVERY_PLACEHOLDER = "/absolute/path/to/mosaic/tools/lease-broker/recover-context.py"
+CONSTRUCTION_PLACEHOLDER = "/absolute/path/to/mosaic-context-refresh-construction.json"
+
+
+def load_gate():
+    spec = importlib.util.spec_from_file_location("framework_skill_portability_gate", GATE_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("unable to load mutator gate")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+GATE = load_gate()
+
+
+class FrameworkSkillPortabilityTest(unittest.TestCase):
+    def test_shipped_framework_skills_contain_no_operator_home_path(self) -> None:
+        offenders = [
+            str(path.relative_to(SKILLS))
+            for path in SKILLS.rglob("SKILL.md")
+            if OPERATOR_HOME.search(path.read_text(encoding="utf-8"))
+        ]
+        self.assertEqual(offenders, [])
+
+    def test_shipped_recovery_template_resolves_to_a_literal_install_path_and_admits(self) -> None:
+        source = REFRESH_SKILL.read_text(encoding="utf-8")
+        self.assertIn(RECOVERY_PLACEHOLDER, source)
+        self.assertIn(CONSTRUCTION_PLACEHOLDER, source)
+        resolved_recovery = "/opt/mosaic/tools/lease-broker/recover-context.py"
+        resolved_construction = "/opt/mosaic/recovery/construction.json"
+        rendered = source.replace(RECOVERY_PLACEHOLDER, resolved_recovery).replace(
+            CONSTRUCTION_PLACEHOLDER, resolved_construction
+        )
+        match = re.search(r"```bash\s*\n\s*(.*?)\n\s*```", rendered, flags=re.DOTALL)
+        self.assertIsNotNone(match)
+        command = match.group(1) if match is not None else ""
+        self.assertEqual(
+            GATE.recovery_invocation_name(
+                {"tool_name": "Bash", "tool_input": {"command": command}},
+                Path(resolved_recovery),
+            ),
+            GATE.RECOVERY_TOOL,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+"""RED-first adversarial Claude B1 gate contracts.
+
+This private harness drives the shipped gate and daemon out of process.  It
+never contacts a live broker/runtime and executes a shell payload only after a
+regression has already (incorrectly) received the recovery exemption.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+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"
+SKILL = FRAMEWORK / "skills/mosaic-context-refresh/SKILL.md"
+
+
+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 socket")
+
+
+class ClaudeRecoveryGateAdversarialTest(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.daemon = subprocess.Popen(
+            [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(self.socket), "--state", str(self.root / "state.json")],
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+        wait_ready(self.daemon, self.socket)
+        registered = request(self.socket, {"action": "register_anchor", "runtime_generation": 1})
+        self.session_id = registered["session_id"]
+        self.environment = {
+            **os.environ,
+            "MOSAIC_LEASE_BROKER_SOCKET": str(self.socket),
+            "MOSAIC_LEASE_SESSION_ID": self.session_id,
+            "MOSAIC_RUNTIME_GENERATION": "1",
+            "MOSAIC_LEASE_RUNTIME": "claude",
+        }
+        # This private path is only a gate-classifier identity; if a regression
+        # blesses a payload, bash invokes no installed/live recovery command.
+        self.recovery_path = self.root / "recover-context.py"
+        self.canonical = (
+            f"python3 {self.recovery_path} begin "
+            f"--construction {self.root / 'construction.json'} --compaction-epoch 0 --request-epoch 0"
+        )
+
+    def tearDown(self) -> None:
+        if self.daemon.poll() is None:
+            self.daemon.terminate()
+            try:
+                self.daemon.wait(timeout=3.0)
+            except subprocess.TimeoutExpired:
+                self.daemon.kill()
+                self.daemon.wait()
+        if self.daemon.stdout is not None:
+            self.daemon.stdout.close()
+        self.temporary.cleanup()
+
+    def gate(
+        self, command: str, recovery_command: Path | str | None = None
+    ) -> subprocess.CompletedProcess[str]:
+        configured_recovery = self.recovery_path if recovery_command is None else recovery_command
+        return subprocess.run(
+            [
+                sys.executable,
+                "-I",
+                "-S",
+                "-B",
+                str(GATE),
+                "--runtime",
+                "claude",
+                "--recovery-command",
+                str(configured_recovery),
+            ],
+            input=json.dumps({"tool_name": "Bash", "tool_input": {"command": command}}),
+            text=True,
+            capture_output=True,
+            env=self.environment,
+            check=False,
+        )
+
+    def command_for(self, position: str, value: str) -> str:
+        argv = [
+            "python3",
+            str(self.recovery_path),
+            "begin",
+            "--construction",
+            str(self.root / "construction.json"),
+            "--compaction-epoch",
+            "0",
+            "--request-epoch",
+            "0",
+        ]
+        positions = {
+            "executable": 0,
+            "path": 1,
+            "phase": 2,
+            "construction_flag": 3,
+            "construction": 4,
+            "compaction_epoch_flag": 5,
+            "compaction_epoch": 6,
+            "request_epoch_flag": 7,
+            "request_epoch": 8,
+        }
+        try:
+            argv[positions[position]] = value
+        except KeyError as exc:
+            raise AssertionError(f"unknown argv position {position}") from exc
+        return " ".join(argv)
+
+    def test_canonical_literal_and_shipped_skill_invocation_are_ungated(self) -> None:
+        self.assertEqual(self.gate(self.canonical).returncode, 0)
+        source = SKILL.read_text(encoding="utf-8")
+        recovery_placeholder = "/absolute/path/to/mosaic/tools/lease-broker/recover-context.py"
+        construction_placeholder = "/absolute/path/to/mosaic-context-refresh-construction.json"
+        self.assertIn(recovery_placeholder, source)
+        self.assertIn(construction_placeholder, source)
+        resolved_recovery = "/opt/mosaic/tools/lease-broker/recover-context.py"
+        rendered = source.replace(recovery_placeholder, resolved_recovery).replace(
+            construction_placeholder, "/opt/mosaic/recovery/construction.json"
+        )
+        match = re.search(r"```bash\s*\n\s*(.*?)\n\s*```", rendered, flags=re.DOTALL)
+        self.assertIsNotNone(match)
+        shipped = match.group(1) if match is not None else ""
+        self.assertEqual(self.gate(shipped, resolved_recovery).returncode, 0)
+
+    def test_every_shell_active_vector_in_every_argv_position_falls_through_to_bash_deny(self) -> None:
+        positions = (
+            "executable",
+            "path",
+            "phase",
+            "construction_flag",
+            "construction",
+            "compaction_epoch_flag",
+            "compaction_epoch",
+            "request_epoch_flag",
+            "request_epoch",
+        )
+        marker = self.root / "PWNED"
+        marker_vector = f"$(touch${{IFS}}{marker})"
+        vectors = {
+            "command-substitution": marker_vector,
+            "backtick": f"`touch${{IFS}}{marker}`",
+            "process-substitution": f"<(touch${{IFS}}{marker})",
+            "parameter-expansion": "${IFS}",
+            "home-expansion": "${HOME}",
+            "arithmetic-expansion": "$((1+1))",
+            "brace-expansion": "{a,b}",
+            "tilde-expansion": "~",
+            "glob": "*",
+            "redirection": ">",
+            "semicolon": f";touch${{IFS}}{marker}",
+            "and": f"&&touch${{IFS}}{marker}",
+            "pipe": f"|touch${{IFS}}{marker}",
+            "embedded-newline": "literal\nnext",
+            "quoting-trick": "'literal'",
+        }
+        for position in positions:
+            for kind, vector in vectors.items():
+                with self.subTest(position=position, kind=kind):
+                    marker.unlink(missing_ok=True)
+                    command = self.command_for(position, vector)
+                    gated = self.gate(command)
+                    if gated.returncode == 0 and kind in {"command-substitution", "backtick", "process-substitution", "semicolon", "and", "pipe"}:
+                        subprocess.run(["bash", "-c", command], cwd=self.root, env=self.environment, check=False)
+                    self.assertEqual(gated.returncode, 2, f"unexpected recovery exemption: {command!r}")
+                    self.assertFalse(marker.exists(), f"shell payload executed: {command!r}")
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -0,0 +1,282 @@
+#!/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()
+            if child.stdout is not None:
+                child.stdout.close()
+        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)
+        mapped_begin = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(GATE), "--runtime", "claude", "--recovery-command", str(RECOVERY)],
+            input=json.dumps({
+                "tool_name": "Bash",
+                "tool_input": {
+                    "command": (
+                        f"python3 {RECOVERY} begin --construction /tmp/construction.json "
+                        "--compaction-epoch 1 --request-epoch 1"
+                    )
+                },
+            }),
+            text=True,
+            capture_output=True,
+            env={**environment, "MOSAIC_LEASE_RUNTIME": "claude"},
+            check=False,
+        )
+        self.assertEqual(mapped_begin.returncode, 0, mapped_begin.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_begin = request(self.socket, {
+            "action": "begin_recovery",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "receipt": receipt,
+        })
+        self.assertEqual(rejected_begin, {"ok": False, "code": "INVALID_RECOVERY_REQUEST"})
+        rejected_observe = 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_observe, {"ok": False, "code": "INVALID_RECEIPT"})
+        rejected_complete = request(self.socket, {
+            "action": "complete_recovery",
+            "session_id": session_id,
+            "runtime_generation": 1,
+            "latest_assistant_message": receipt,
+        })
+        self.assertEqual(rejected_complete, {"ok": False, "code": "INVALID_RECOVERY_REQUEST"})
+
+        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")
+
+        # The independent Claude transport selects one latest assistant entry
+        # from its hook transcript; it is not a Pi/message_end fallback.
+        claude_environment = {**environment, "MOSAIC_LEASE_RUNTIME": "claude"}
+        claude_begin = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(RECOVERY), "begin", "--construction", str(self.construction()), "--compaction-epoch", "2", "--request-epoch", "2"],
+            text=True,
+            capture_output=True,
+            env=claude_environment,
+            check=False,
+        )
+        self.assertEqual(claude_begin.returncode, 0, claude_begin.stderr)
+        claude_receipt = json.loads(claude_begin.stdout)["receipt"]
+        transcript = self.root / "claude-transcript.jsonl"
+        transcript.write_text(
+            json.dumps({"message": {"role": "assistant", "content": claude_receipt}}) + "\n",
+            encoding="utf-8",
+        )
+        claude_recorded = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(OBSERVER_CLIENT), "--runtime", "claude", "--latest-entry"],
+            input=json.dumps({"transcript_path": str(transcript)}),
+            text=True,
+            capture_output=True,
+            env=claude_environment,
+            check=False,
+        )
+        self.assertEqual(claude_recorded.returncode, 0, claude_recorded.stderr)
+        claude_complete = subprocess.run(
+            [sys.executable, "-I", "-S", "-B", str(RECOVERY), "complete"],
+            text=True,
+            capture_output=True,
+            env=claude_environment,
+            check=False,
+        )
+        self.assertEqual(claude_complete.returncode, 0, claude_complete.stderr)
+        self.assertEqual(json.loads(claude_complete.stdout).get("state"), "VERIFIED")
+
+
+if __name__ == "__main__":
+    unittest.main()