fix(mosaic): harden literal Claude recovery gate mapping

docs/compaction-refresh/probes/p6_constrained_recovery.py

@@ -135,8 +135,11 @@ def run_once(index: int, runtime: str) -> None:
     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")
 

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

@@ -1,6 +1,6 @@
 # #833 constrained recovery command — build scratchpad
 
-- **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.
+- **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.
@@ -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:** 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.
+- **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/skills/mosaic-context-refresh/SKILL.md

@@ -16,14 +16,13 @@ mutator remains behind the verified lease gate.
    - **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"
+     python3 /home/hermes/.config/mosaic/tools/lease-broker/recover-context.py begin --construction /absolute/path/to/mosaic-context-refresh-construction.json --compaction-epoch 0 --request-epoch 0
      ```
 
-     Claude's all-tools gate maps this exact recovery executable and validated argument shape to
-     `mosaic_context_recover`; ordinary `Bash` remains gated.
+     This is a literal argv template: replace the construction path with a literal absolute path and
+     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;
@@ -41,7 +40,7 @@ mutator remains behind the verified lease gate.
      `after_provider_response`).
 
    Then invoke completion with the same adapter form: Claude runs
-   `python3 "$MOSAIC_HOME/tools/lease-broker/recover-context.py" complete`; Pi calls
+   `python3 /home/hermes/.config/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

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

@@ -8,7 +8,7 @@ import json
 import os
 import socket
 import sys
-import shlex
+import re
 from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import BinaryIO, Final
@@ -23,6 +23,8 @@ 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:
@@ -51,11 +53,13 @@ def read_tool_name(stream: BinaryIO | None = None) -> str:
 
 
 def recovery_invocation_name(request: dict[str, object], recovery_command: Path | None) -> str:
-    """Map only a direct Claude Bash invocation of the recovery executable.
+    """Map only a byte-literal Claude recovery argv to ``RECOVERY_TOOL``.
 
-    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.
+    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"]
@@ -65,27 +69,33 @@ def recovery_invocation_name(request: dict[str, object], recovery_command: Path
     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:
+    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)
-    try:
-        argv = shlex.split(command, posix=True)
-    except ValueError:
+    argv = command.split(" ")
+    if " ".join(argv) != command or len(argv) < 3 or argv[0] != "python3":
         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:
+    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)
-    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]):
+    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
 

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

@@ -102,7 +102,10 @@ class ClaudeRecoveryGateAdversarialTest(unittest.TestCase):
             self.daemon.stdout.close()
         self.temporary.cleanup()
 
-    def gate(self, command: str) -> subprocess.CompletedProcess[str]:
+    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,
@@ -113,7 +116,7 @@ class ClaudeRecoveryGateAdversarialTest(unittest.TestCase):
                 "--runtime",
                 "claude",
                 "--recovery-command",
-                str(self.recovery_path),
+                str(configured_recovery),
             ],
             input=json.dumps({"tool_name": "Bash", "tool_input": {"command": command}}),
             text=True,
@@ -153,9 +156,18 @@ class ClaudeRecoveryGateAdversarialTest(unittest.TestCase):
 
     def test_canonical_literal_and_shipped_skill_invocation_are_ungated(self) -> None:
         self.assertEqual(self.gate(self.canonical).returncode, 0)
-        self.assertIn(
-            "python3 /home/hermes/.config/mosaic/tools/lease-broker/recover-context.py begin",
-            SKILL.read_text(encoding="utf-8"),
+        shipped = (
+            "python3 /home/hermes/.config/mosaic/tools/lease-broker/recover-context.py begin "
+            "--construction /absolute/path/to/mosaic-context-refresh-construction.json "
+            "--compaction-epoch 0 --request-epoch 0"
+        )
+        self.assertIn(shipped, SKILL.read_text(encoding="utf-8"))
+        self.assertEqual(
+            self.gate(
+                shipped,
+                "/home/hermes/.config/mosaic/tools/lease-broker/recover-context.py",
+            ).returncode,
+            0,
         )
 
     def test_every_shell_active_vector_in_every_argv_position_falls_through_to_bash_deny(self) -> None: