WI-6 (#833): constrained recovery command + mosaic-context-refresh skill wrapper (#846)

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)