diff --git a/docs/architecture/lease-broker-protocol.md b/docs/architecture/lease-broker-protocol.md
index 707e3e91..a3218dfe 100644
--- a/docs/architecture/lease-broker-protocol.md
+++ b/docs/architecture/lease-broker-protocol.md
@@ -15,6 +15,9 @@ Each connection carries exactly one UTF-8 JSON object followed by one newline, c
 - `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.
diff --git a/docs/compaction-refresh/probes/p6_constrained_recovery.py b/docs/compaction-refresh/probes/p6_constrained_recovery.py
index 101f9c95..3b4a2224 100644
--- a/docs/compaction-refresh/probes/p6_constrained_recovery.py
+++ b/docs/compaction-refresh/probes/p6_constrained_recovery.py
@@ -1,13 +1,13 @@
 #!/usr/bin/env python3
 """P6 constrained-recovery probe; BUILT ONLY and Mos-gated.
 
-DO NOT self-fire.  Under Mos authorization only:
+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 each start a fresh shipped daemon on a private
-Unix socket and drive the shipped recovery *command* out of process.  This
-probe never resets broker state, replaces the observer seam, or reimplements
-promotion: it exercises `recover-context.py` and the daemon's real socket.
+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
@@ -31,8 +31,12 @@ 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():
@@ -76,35 +80,72 @@ def wait_ready(process: subprocess.Popen[str], socket_path: Path) -> None:
     raise TimeoutError("shipped daemon did not create private probe socket")
 
 
-def recovery_command(arguments: list[str], environment: dict[str, str]) -> dict[str, object]:
+def run_json(command: list[str], environment: dict[str, str], input_value: object | None = None) -> dict[str, object]:
     completed = subprocess.run(
-        [sys.executable, "-I", "-S", "-B", str(RECOVERY_COMMAND), *arguments],
+        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"recovery command omitted framed result: {completed.stderr!r}")
+        raise AssertionError(f"command omitted framed result: {completed.stderr!r}")
     reply = json.loads(completed.stdout)
     if not isinstance(reply, dict):
-        raise AssertionError("recovery command result is not an object")
+        raise AssertionError("command result is not an object")
     return reply
 
 
-def run_once(index: int) -> None:
-    # Parity guard: this is an out-of-process driver of the shipped recovery
-    # entrypoint, not a shadow receipt/promotion implementation.
-    source = RECOVERY_COMMAND.read_text(encoding="utf-8")
-    if '"action": "begin_recovery"' not in source or '"action": "complete_recovery"' not in source:
+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")
+    if "--recovery-command" not in CLAUDE_SETTINGS.read_text(encoding="utf-8"):
+        raise AssertionError("P6 parity guard: Claude recovery mapping is missing")
+    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"
-    observer_path = root / "test-observer.json"
     construction_path = root / "construction.json"
     content = b"P6 constrained recovery fixture\n"
     construction = {
@@ -120,7 +161,7 @@ def run_once(index: int) -> None:
     os.chmod(construction_path, 0o600)
     process = subprocess.Popen(
         [sys.executable, "-I", "-S", "-B", str(DAEMON), "--socket", str(socket_path),
-         "--state", str(state_path), "--test-observer-file", str(observer_path)],
+         "--state", str(state_path), "--observer-socket", str(observer_socket)],
         stdin=subprocess.DEVNULL,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
@@ -135,7 +176,7 @@ def run_once(index: int) -> None:
         built = fragments.build_payload_from_wire(construction)
         normal = request(socket_path, {
             "action": "begin_verification", "session_id": session_id, "runtime_generation": 1,
-            "runtime": "pi", "construction": construction,
+            "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},
         })
@@ -146,13 +187,15 @@ def run_once(index: int) -> None:
         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": "pi",
+            "MOSAIC_LEASE_RUNTIME": runtime,
         }
-        recovery = recovery_command([
-            "begin", "--construction", str(construction_path), "--compaction-epoch", str(index + 10),
-            "--request-epoch", str(index + 110),
+        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")
@@ -161,32 +204,26 @@ def run_once(index: int) -> None:
         if challenge == normal_challenge:
             raise AssertionError("recovery reused a normal-path challenge")
 
-        # C4: a normal-path receipt is not an admissible recovery receipt.
-        observer_path.write_text(json.dumps({
-            "session_id": session_id, "runtime_generation": 1, "latest_assistant_message": normal_receipt,
-        }), encoding="utf-8")
-        os.chmod(observer_path, 0o600)
-        refused = recovery_command(["complete"], environment)
+        # 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}")
 
-        # Begin a new recovery cycle, then prove shipped evidence -> consume ->
-        # promote ordering and post-consumption non-repromotion.
-        recovery = recovery_command([
-            "begin", "--construction", str(construction_path), "--compaction-epoch", str(index + 20),
-            "--request-epoch", str(index + 120),
+        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}")
-        observer_path.write_text(json.dumps({
-            "session_id": session_id, "runtime_generation": 1, "latest_assistant_message": receipt,
-        }), encoding="utf-8")
-        os.chmod(observer_path, 0o600)
-        promoted = recovery_command(["complete"], environment)
+        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 = recovery_command(["complete"], environment)
+        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:
@@ -206,8 +243,8 @@ def main() -> None:
     arguments = parser.parse_args()
     if arguments.runs != 3:
         raise SystemExit("P6 requires exactly three isolated runs")
-    for index in range(arguments.runs):
-        run_once(index)
+    for index, runtime in enumerate(("pi", "claude", "pi")):
+        run_once(index, runtime)
     print("P6 constrained recovery probe PASS: 3 isolated shipped recovery-command runs")
 
 
diff --git a/docs/guides/lease-broker-operations.md b/docs/guides/lease-broker-operations.md
index b0fac791..ee7caf57 100644
--- a/docs/guides/lease-broker-operations.md
+++ b/docs/guides/lease-broker-operations.md
@@ -31,7 +31,9 @@ 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.
 
-`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. It uses only the trusted `ReceiptObserver` seam for the exact latest assistant entry, commits evidence, consumes the broker-held challenge, and promotes VERIFIED last. 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.
+`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.
 
diff --git a/docs/scratchpads/833-constrained-recovery-command.md b/docs/scratchpads/833-constrained-recovery-command.md
index b4ea57af..2161066e 100644
--- a/docs/scratchpads/833-constrained-recovery-command.md
+++ b/docs/scratchpads/833-constrained-recovery-command.md
@@ -1,9 +1,9 @@
 # #833 constrained recovery command — build scratchpad
 
-- **Objective:** Deliver WI-6 only: the single ungated constrained recovery command, durable `mosaic-context-refresh` wrapper skill, AC-1 observable partial-delivery and C4 replay evidence, and an unfired P6 probe.
+- **Objective:** Deliver WI-6 plus Mos-ruled B1/B2 repair only: authenticated narrow recovery invocation in Claude/Pi, production receipt-observer transport, AC-1/C4 preservation, and an unfired repaired P6 probe.
 - **Authority:** STEP-0 SHA-256 verified 4/4 against the supplied BUILD-BRIEF, SPEC-v5, ratification, and red-team records.
 - **Base:** exact `07553ead337a70a9241f826d27571650262b289c`; new branch `feat/833-constrained-recovery-command`; merge-base assertion passed before any commit.
-- **Constraints:** No rebase/pull during build; no live install/symlink or live broker/socket/tmux/systemd mutation; no self-review, PR, merge, push, or P6 fire. `docs/TASKS.md` is orchestrator-owned and will not be modified.
+- **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.
@@ -11,5 +11,5 @@
   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:** RED test committed at `3b5513bd6efad0c06b599fc759a66cff4286db04`; expected `UNKNOWN_ACTION` is logged in `/home/hermes/agent-work/reviews/833-wi6-red.log`. GREEN: `pnpm --filter @mosaicstack/mosaic run test:framework-shell` passed (deadline 2, normative 5, receipt 5, recovery 4, runtime tools 25, launch guard 12, inventory 14/14); `git diff --check`; Python `py_compile`; and a tmp-only execution of the real #824 `syncClaudeSkills` bridge projected the source skill (`WI-6 #824 tmp projection PASS`). P6 was built but not fired. 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.
+- **Evidence:** Original RED test committed at `3b5513bd6efad0c06b599fc759a66cff4286db04`; expected `UNKNOWN_ACTION` is logged in `/home/hermes/agent-work/reviews/833-wi6-red.log`. Repair RED is committed at `f4beedc3e7ac2e142dcbdeefb0e5ee40c20d9b86` and logged in `/home/hermes/agent-work/reviews/833-wi6-repair-red.log` before B1/B2 source. The repair's private out-of-process harness proves exact Claude recovery mapping while UNVERIFIED, non-recovery Bash denial, Pi's exact custom recovery-tool registration, production Pi `message_end` and Claude latest-entry observer promotion, and S1 triple rejection. P6 was rebuilt but remains 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.
diff --git a/packages/mosaic/framework/runtime/claude/settings.json b/packages/mosaic/framework/runtime/claude/settings.json
index a37fca28..eada96fe 100644
--- a/packages/mosaic/framework/runtime/claude/settings.json
+++ b/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",
diff --git a/packages/mosaic/framework/runtime/pi/mosaic-extension.ts b/packages/mosaic/framework/runtime/pi/mosaic-extension.ts
index cdf9b443..10756ffe 100644
--- a/packages/mosaic/framework/runtime/pi/mosaic-extension.ts
+++ b/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();
diff --git a/packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md b/packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md
index 6b669970..e50693d7 100644
--- a/packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md
+++ b/packages/mosaic/framework/skills/mosaic-context-refresh/SKILL.md
@@ -12,31 +12,40 @@ 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, then invokes:
+   compaction/request epochs.
+   - **Claude:** invoke only this direct command shape (no shell composition):
 
-   ```bash
-   python3 "$MOSAIC_HOME/tools/lease-broker/recover-context.py" begin \
-     --construction "$MOSAIC_CONTEXT_REFRESH_CONSTRUCTION" \
-     --compaction-epoch "$MOSAIC_COMPACTION_EPOCH" \
-     --request-epoch "$MOSAIC_REQUEST_EPOCH"
-   ```
+     ```bash
+     python3 "$MOSAIC_HOME/tools/lease-broker/recover-context.py" begin \
+       --construction "$MOSAIC_CONTEXT_REFRESH_CONSTRUCTION" \
+       --compaction-epoch "$MOSAIC_COMPACTION_EPOCH" \
+       --request-epoch "$MOSAIC_REQUEST_EPOCH"
+     ```
 
-   The command delegates 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. It prints
+     Claude's all-tools gate maps this exact recovery executable and validated argument 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. At `message_end`, the trusted runtime `ReceiptObserver` seam invokes:
+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`).
 
-   ```bash
-   python3 "$MOSAIC_HOME/tools/lease-broker/recover-context.py" 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.
+   Then invoke completion with the same adapter form: Claude runs
+   `python3 "$MOSAIC_HOME/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
 
diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py
index 4dcd5c66..5829a7ef 100644
--- a/packages/mosaic/framework/tools/lease-broker/daemon.py
+++ b/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()
@@ -783,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()
@@ -805,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:
@@ -829,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__":
diff --git a/packages/mosaic/framework/tools/lease-broker/launch-runtime.py b/packages/mosaic/framework/tools/lease-broker/launch-runtime.py
index 8019f474..f54aec46 100644
--- a/packages/mosaic/framework/tools/lease-broker/launch-runtime.py
+++ b/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:
diff --git a/packages/mosaic/framework/tools/lease-broker/mutator-gate.py b/packages/mosaic/framework/tools/lease-broker/mutator-gate.py
index 62a0a173..c0dca085 100644
--- a/packages/mosaic/framework/tools/lease-broker/mutator-gate.py
+++ b/packages/mosaic/framework/tools/lease-broker/mutator-gate.py
@@ -8,14 +8,21 @@ import json
 import os
 import socket
 import sys
+import shlex
 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"
 
 
 def deny(code: str) -> int:
@@ -23,7 +30,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 +41,53 @@ 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 direct Claude Bash invocation of the recovery executable.
+
+    The mapping is adapter-configured and remains broker-authenticated through
+    the ordinary authorization request. It never blesses Bash generally, shell
+    composition, a different executable, or unsupported recovery arguments.
+    """
+
+    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:
+        return str(tool_name)
+    try:
+        argv = shlex.split(command, posix=True)
+    except ValueError:
+        return str(tool_name)
+    if len(argv) < 3 or argv[0] != "python3":
+        return str(tool_name)
+    try:
+        if Path(argv[1]).resolve(strict=False) != recovery_command.resolve(strict=False):
+            return str(tool_name)
+    except OSError:
+        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)
+    expected_flags = {"--construction", "--compaction-epoch", "--request-epoch"}
+    supplied_flags = set(argv[3::2])
+    if supplied_flags != expected_flags or any(not value for value in argv[4::2]):
+        return str(tool_name)
+    return RECOVERY_TOOL
 
 
 def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
@@ -70,11 +123,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)
diff --git a/packages/mosaic/framework/tools/lease-broker/receipt-observer-client.py b/packages/mosaic/framework/tools/lease-broker/receipt-observer-client.py
new file mode 100644
index 00000000..02f22d3f
--- /dev/null
+++ b/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())
diff --git a/packages/mosaic/framework/tools/lease-broker/receipt_observer.py b/packages/mosaic/framework/tools/lease-broker/receipt_observer.py
index 49cac951..6ea70357 100644
--- a/packages/mosaic/framework/tools/lease-broker/receipt_observer.py
+++ b/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
diff --git a/packages/mosaic/framework/tools/lease-broker/recover-context.py b/packages/mosaic/framework/tools/lease-broker/recover-context.py
index 2bb58156..8a3e029c 100644
--- a/packages/mosaic/framework/tools/lease-broker/recover-context.py
+++ b/packages/mosaic/framework/tools/lease-broker/recover-context.py
@@ -18,6 +18,11 @@ 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
diff --git a/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py b/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py
index 2c9f62a0..684daece 100644
--- a/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py
+++ b/packages/mosaic/src/lease-broker/recovery_runtime_unittest.py
@@ -81,6 +81,8 @@ class RuntimeBoundaryFixture(unittest.TestCase):
                 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:
@@ -145,6 +147,23 @@ class RecoveryRuntimeBoundaryTest(RuntimeBoundaryFixture):
             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"}}),
@@ -180,14 +199,28 @@ class RecoveryRuntimeBoundaryTest(RuntimeBoundaryFixture):
 
         # S1: production transport is not a broker request field. The public
         # broker endpoint keeps rejecting caller-supplied assistant evidence.
-        rejected = request(self.socket, {
+        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, {"ok": False, "code": "INVALID_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"],
@@ -208,6 +241,42 @@ class RecoveryRuntimeBoundaryTest(RuntimeBoundaryFixture):
         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()