fix(#832): bind receipt promotion to trusted observation

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

@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import copy
 import errno
+import hmac
 from concurrent.futures import ThreadPoolExecutor
 import json
 import os
@@ -25,7 +26,9 @@ from typing import Final
 _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
 from receipt_challenge import is_verbatim_receipt, latest_assistant_digest, receipt_for
+from receipt_observer import FileTestReceiptObserver, ReceiptObserver, UnavailableReceiptObserver
 
 MAX_FRAME: Final = 64 * 1024
 MAX_STATE: Final = 4 * 1024 * 1024
@@ -297,8 +300,9 @@ class StateStore:
 
 
 class Broker:
-    def __init__(self, store: StateStore) -> None:
+    def __init__(self, store: StateStore, observer: ReceiptObserver | None = None) -> None:
         self.store = store
+        self.observer: ReceiptObserver = observer if observer is not None else UnavailableReceiptObserver()
         # VERIFIED authority is deliberately volatile: broker restart revokes all
         # leases while preserving WI-1 identity and pending-token integrity.
         self.leases: dict[str, dict[str, object]] = {}
@@ -455,6 +459,7 @@ class Broker:
             session_id, _ = self.authenticate(peer_pid, request)
             runtime = request.get("runtime")
             binding = request.get("binding")
+            construction = request.get("construction")
             ttl_seconds = request.get("ttl_seconds", MAX_LEASE_TTL_SECONDS)
             if runtime not in READ_ONLY_TOOLS:
                 raise BrokerFailure("INVALID_RUNTIME")
@@ -468,6 +473,22 @@ class Broker:
                 raise BrokerFailure("INVALID_LEASE_TTL")
             # Revoke-first is a broker operation, not advisory adapter order.
             self.revoke_session_authority(session_id)
+            try:
+                constructed = build_payload_from_wire(construction)
+            except ValueError as exc:
+                raise BrokerFailure("INVALID_CONSTRUCTION") from exc
+            if (
+                constructed.injectionDecision != "ACCEPTED"
+                or not constructed.promotion
+                or not isinstance(constructed.h_source, str)
+                or not isinstance(constructed.h_payload, str)
+            ):
+                raise BrokerFailure("PAYLOAD_CONSTRUCTION_REFUSED")
+            if (
+                not hmac.compare_digest(binding["h_source"], constructed.h_source)
+                or not hmac.compare_digest(binding["h_payload"], constructed.h_payload)
+            ):
+                raise BrokerFailure("PAYLOAD_BINDING_MISMATCH")
             cycle_binding = copy.deepcopy(binding)
             cycle_binding["runtime_generation"] = request["runtime_generation"]
             challenge = self.mint_token(session_id, request["runtime_generation"], binding)
@@ -491,8 +512,9 @@ class Broker:
         if action == "observe_receipt":
             session_id, _ = self.authenticate(peer_pid, request)
             challenge = request.get("receipt_challenge")
-            message = request.get("latest_assistant_message")
-            if not isinstance(challenge, str) or not isinstance(message, str):
+            if "latest_assistant_message" in request:
+                raise BrokerFailure("INVALID_RECEIPT")
+            if not isinstance(challenge, str):
                 raise BrokerFailure("INVALID_RECEIPT")
             token = self.store.tokens().get(challenge)
             if (
@@ -516,6 +538,11 @@ class Broker:
                 raise BrokerFailure("RECEIPT_REPLAY")
             receipt_binding = copy.deepcopy(binding)
             receipt_binding["runtime_generation"] = request["runtime_generation"]
+            message = self.observer.observe_latest_assistant_message(
+                session_id, str(lease["runtime"]), request["runtime_generation"], receipt_binding
+            )
+            if not isinstance(message, str):
+                raise BrokerFailure("RECEIPT_OBSERVATION_UNAVAILABLE")
             # This accepts one exact current-cycle assistant entry only. It is
             # deliberately not a transcript search and rejects quoted/extra text.
             if not is_verbatim_receipt(message, challenge, receipt_binding):
@@ -530,6 +557,9 @@ class Broker:
             challenge = request.get("receipt_challenge")
             if not isinstance(challenge, str):
                 raise BrokerFailure("INVALID_RECEIPT")
+            lease = self.leases.get(session_id)
+            if not isinstance(lease, dict) or lease.get("state") == LEASE_UNVERIFIED:
+                raise BrokerFailure("INVALID_LEASE_TRANSITION")
             token = self.store.tokens().get(challenge)
             if (
                 not isinstance(token, dict)
@@ -538,8 +568,7 @@ class Broker:
                 or token.get("consumed") is not False
             ):
                 raise BrokerFailure("RECEIPT_REPLAY")
-            lease = self.leases.get(session_id)
-            if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING_PROMOTION:
+            if lease.get("state") != LEASE_PENDING_PROMOTION:
                 raise BrokerFailure("INVALID_LEASE_TRANSITION")
             expected_challenge = lease.get("receipt_challenge")
             if not isinstance(expected_challenge, str) or not secrets.compare_digest(challenge, expected_challenge):
@@ -670,12 +699,16 @@ def handle_connection(
             return
 
 
-def serve(socket_path: Path, state_path: Path) -> None:
+def serve(
+    socket_path: Path,
+    state_path: Path,
+    observer: ReceiptObserver | None = None,
+) -> None:
     secure_parent(socket_path)
     if socket_path.exists() or socket_path.is_symlink():
         raise BrokerFailure("SOCKET_ALREADY_EXISTS")
     store = StateStore(state_path)
-    broker = Broker(store)
+    broker = Broker(store, observer)
     broker_lock = threading.Lock()
     slots = threading.BoundedSemaphore(MAX_IN_FLIGHT_CONNECTIONS)
     fatal_lock = threading.Lock()
@@ -755,8 +788,14 @@ 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("--test-observer-file", type=Path)
     arguments = parser.parse_args()
-    serve(arguments.socket, arguments.state)
+    observer = (
+        FileTestReceiptObserver(arguments.test_observer_file)
+        if arguments.test_observer_file is not None
+        else None
+    )
+    serve(arguments.socket, arguments.state, observer)
 
 
 if __name__ == "__main__":

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

@@ -9,6 +9,7 @@ as a payload input, preventing cryptographic self-reference.
 
 from __future__ import annotations
 
+import base64
 import hashlib
 import hmac
 import struct
@@ -164,6 +165,44 @@ def build_payload(
     )
 
 
+def build_payload_from_wire(value: object) -> ConstructionResult:
+    """Decode a bounded broker request then invoke the sole payload builder.
+
+    Wire inputs contain only source bytes and their claimed source digests. The
+    authoritative ``build_payload`` implementation remains the only code that
+    admits those bytes and derives ``h_source``/``h_payload``.
+    """
+
+    if not isinstance(value, dict) or set(value) != {
+        "manifest_version", "generator_version", "fragments"
+    }:
+        raise ValueError("invalid construction")
+    raw_fragments = value["fragments"]
+    if not isinstance(raw_fragments, list):
+        raise ValueError("invalid construction")
+    fragments: list[NormativeFragment] = []
+    for raw_fragment in raw_fragments:
+        if not isinstance(raw_fragment, dict) or set(raw_fragment) != {
+            "source_id", "content_base64", "expected_sha256"
+        }:
+            raise ValueError("invalid construction")
+        source_id = raw_fragment["source_id"]
+        encoded = raw_fragment["content_base64"]
+        expected_sha256 = raw_fragment["expected_sha256"]
+        if not isinstance(source_id, str) or not isinstance(encoded, str) or not isinstance(expected_sha256, str):
+            raise ValueError("invalid construction")
+        try:
+            content = base64.b64decode(encoded.encode("ascii"), validate=True)
+        except (UnicodeEncodeError, ValueError) as exc:
+            raise ValueError("invalid construction") from exc
+        fragments.append(NormativeFragment(source_id, content, expected_sha256))
+    return build_payload(
+        manifest_version=value["manifest_version"],
+        generator_version=value["generator_version"],
+        fragments=fragments,
+    )
+
+
 def build_for_claude(**kwargs: object) -> ConstructionResult:
     """Claude adapter entrypoint; delegates to the sole shared constructor."""
 

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

@@ -0,0 +1,106 @@
+#!/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.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+from pathlib import Path
+from typing import Protocol
+
+
+class ReceiptObserver(Protocol):
+    def observe_latest_assistant_message(
+        self,
+        session_id: str,
+        runtime: str,
+        runtime_generation: int,
+        binding: dict[str, object],
+    ) -> str | None: ...
+
+
+class UnavailableReceiptObserver:
+    """Production-safe default until a runtime adapter injects an observer."""
+
+    def observe_latest_assistant_message(
+        self,
+        _session_id: str,
+        _runtime: str,
+        _runtime_generation: int,
+        _binding: dict[str, object],
+    ) -> str | None:
+        return None
+
+
+class TestReceiptObserver:
+    """Deterministic controlled observer used only by byte-build tests."""
+
+    def __init__(self) -> None:
+        self._messages: dict[tuple[str, int], str] = {}
+
+    def record_latest_assistant_message(
+        self, session_id: str, runtime_generation: int, message: str
+    ) -> None:
+        self._messages[(session_id, 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_generation))
+
+
+class FileTestReceiptObserver:
+    """Private fixture-file observer for isolated out-of-process test drivers."""
+
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    def observe_latest_assistant_message(
+        self,
+        session_id: str,
+        _runtime: str,
+        runtime_generation: int,
+        _binding: dict[str, object],
+    ) -> str | None:
+        flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+        try:
+            descriptor = os.open(self.path, flags)
+        except OSError:
+            return None
+        try:
+            metadata = os.fstat(descriptor)
+            if (
+                not stat.S_ISREG(metadata.st_mode)
+                or stat.S_IMODE(metadata.st_mode) != 0o600
+                or metadata.st_uid != os.geteuid()
+                or metadata.st_size > 64 * 1024
+            ):
+                return None
+            raw = os.read(descriptor, 64 * 1024 + 1)
+        finally:
+            os.close(descriptor)
+        if len(raw) > 64 * 1024:
+            return None
+        try:
+            value = json.loads(raw)
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            return None
+        if (
+            not isinstance(value, dict)
+            or set(value) != {"session_id", "runtime_generation", "latest_assistant_message"}
+            or value["session_id"] != session_id
+            or value["runtime_generation"] != runtime_generation
+            or not isinstance(value["latest_assistant_message"], str)
+        ):
+            return None
+        return value["latest_assistant_message"]