fix(mosaic): wire constrained recovery runtime boundaries

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__":

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

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

@@ -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