feat(#832): enforce broker-minted receipt challenges

docs/scratchpads/832-receipt-challenge-protocol.md

@@ -10,4 +10,4 @@
   3. Implement the broker-minted receipt challenge and exact receipt observation/consume/promote path.
   4. Run unit, framework-shell, compile, lint, and type checks; push after the required queue guard; report to `mosaic-100`.
 - **Risks:** The standalone harness must drive the real daemon without a divergent fixture. If that is impossible, stop and flag Mos.
-- **Evidence:** Pending red run, green checks, and push.
+- **Evidence:** RED recorded in `/home/hermes/agent-work/reviews/832-wi5-red-receipt-challenge.log`: receipt module absent before implementation. GREEN: receipt T26/T29 unittest (2 tests), normative-fragments unittest (5 tests), state-store regression (10 tests), `py_compile`, Mosaic package lint, and Mosaic package typecheck. Coverage tooling is unavailable (`python3 -m coverage`: module not installed); no probe was fired. Push pending.

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

@@ -20,6 +20,13 @@ import time
 from pathlib import Path
 from typing import Final
 
+# This script is loaded both as an executable and through isolated stdlib tests.
+# Keep its co-located receipt implementation importable in both modes.
+_MODULE_DIRECTORY = str(Path(__file__).resolve().parent)
+if _MODULE_DIRECTORY not in sys.path:
+    sys.path.insert(0, _MODULE_DIRECTORY)
+from receipt_challenge import is_verbatim_receipt, latest_assistant_digest, receipt_for
+
 MAX_FRAME: Final = 64 * 1024
 MAX_STATE: Final = 4 * 1024 * 1024
 MAX_PENDING_TOKENS: Final = 256
@@ -90,6 +97,14 @@ def valid_binding(binding: object) -> bool:
     )
 
 
+def valid_receipt_evidence(evidence: object) -> bool:
+    return (
+        isinstance(evidence, dict)
+        and set(evidence) == {"h_latest_assistant"}
+        and is_hex_256(evidence["h_latest_assistant"])
+    )
+
+
 def validate_state(value: object) -> dict[str, object]:
     if not isinstance(value, dict) or set(value) != {"version", "sessions", "tokens"}:
         raise BrokerFailure("STATE_INTEGRITY")
@@ -124,7 +139,11 @@ def validate_state(value: object) -> dict[str, object]:
     for token_value, token in tokens.items():
         if not is_hex_256(token_value) or not isinstance(token, dict):
             raise BrokerFailure("STATE_INTEGRITY")
-        if set(token) != {"session_id", "runtime_generation", "binding", "consumed"}:
+        token_fields = set(token)
+        if token_fields not in (
+            {"session_id", "runtime_generation", "binding", "consumed"},
+            {"session_id", "runtime_generation", "binding", "consumed", "evidence"},
+        ):
             raise BrokerFailure("STATE_INTEGRITY")
         session_id = token["session_id"]
         generation = token["runtime_generation"]
@@ -137,6 +156,9 @@ def validate_state(value: object) -> dict[str, object]:
             raise BrokerFailure("STATE_INTEGRITY")
         if not valid_binding(token["binding"]) or token["consumed"] is not False:
             raise BrokerFailure("STATE_INTEGRITY")
+        evidence = token.get("evidence")
+        if evidence is not None and not valid_receipt_evidence(evidence):
+            raise BrokerFailure("STATE_INTEGRITY")
     return value
 
 
@@ -348,6 +370,9 @@ class Broker:
             "runtime_generation": generation,
             "binding": copy.deepcopy(binding),
             "consumed": False,
+            # Receipt evidence is durably committed before this challenge may
+            # be consumed and promotion made externally visible.
+            "evidence": None,
         }
         return token
 
@@ -443,47 +468,88 @@ class Broker:
                 raise BrokerFailure("INVALID_LEASE_TTL")
             # Revoke-first is a broker operation, not advisory adapter order.
             self.revoke_session_authority(session_id)
-            token = self.mint_token(session_id, request["runtime_generation"], binding)
+            cycle_binding = copy.deepcopy(binding)
+            cycle_binding["runtime_generation"] = request["runtime_generation"]
+            challenge = self.mint_token(session_id, request["runtime_generation"], binding)
             self.leases[session_id] = {
                 "state": LEASE_PENDING,
                 "runtime": runtime,
                 "runtime_generation": request["runtime_generation"],
                 "binding": copy.deepcopy(binding),
-                "promotion_token": token,
+                "receipt_challenge": challenge,
                 "ttl_seconds": ttl_seconds,
             }
+            # The broker constructs both the bound challenge and the delivery
+            # text. The model's role is an exact copy, never hashing its output.
             return {
                 "ok": True,
                 "state": LEASE_PENDING,
-                "promotion_token": token,
+                "receipt_challenge": challenge,
+                "receipt": receipt_for(challenge, cycle_binding),
+                "binding": cycle_binding,
             }
-        if action == "promote_lease":
+        if action == "observe_receipt":
             session_id, _ = self.authenticate(peer_pid, request)
-            promotion_token = request.get("promotion_token")
+            challenge = request.get("receipt_challenge")
-            lease = self.leases.get(session_id)
+            message = request.get("latest_assistant_message")
-            if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING:
+            if not isinstance(challenge, str) or not isinstance(message, str):
-                raise BrokerFailure("INVALID_LEASE_TRANSITION")
+                raise BrokerFailure("INVALID_RECEIPT")
-            expected_token = lease.get("promotion_token")
+            token = self.store.tokens().get(challenge)
-            if (
-                not isinstance(promotion_token, str)
-                or not isinstance(expected_token, str)
-                or not secrets.compare_digest(promotion_token, expected_token)
-            ):
-                raise BrokerFailure("PROMOTION_TOKEN_MISMATCH")
-            token = self.store.tokens().get(promotion_token)
             if (
                 not isinstance(token, dict)
                 or token.get("session_id") != session_id
                 or token.get("runtime_generation") != request.get("runtime_generation")
-                or token.get("binding") != lease.get("binding")
                 or token.get("consumed") is not False
             ):
-                raise BrokerFailure("PROMOTION_TOKEN_INVALID")
+                raise BrokerFailure("RECEIPT_REPLAY")
-            del self.store.tokens()[promotion_token]
+            lease = self.leases.get(session_id)
+            if not isinstance(lease, dict) or lease.get("state") != LEASE_PENDING:
+                raise BrokerFailure("RECEIPT_REPLAY")
+            expected_challenge = lease.get("receipt_challenge")
+            binding = lease.get("binding")
+            if (
+                not isinstance(expected_challenge, str)
+                or not secrets.compare_digest(challenge, expected_challenge)
+                or not isinstance(binding, dict)
+                or token.get("binding") != binding
+            ):
+                raise BrokerFailure("RECEIPT_REPLAY")
+            receipt_binding = copy.deepcopy(binding)
+            receipt_binding["runtime_generation"] = request["runtime_generation"]
+            # 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):
+                raise BrokerFailure("RECEIPT_MISMATCH")
+            evidence = {"h_latest_assistant": latest_assistant_digest(message)}
+            token["evidence"] = evidence
             lease["state"] = LEASE_PENDING_PROMOTION
+            lease["evidence"] = copy.deepcopy(evidence)
+            return {"ok": True, "state": LEASE_PENDING_PROMOTION}
+        if action == "promote_lease":
+            session_id, _ = self.authenticate(peer_pid, request)
+            challenge = request.get("receipt_challenge")
+            if not isinstance(challenge, str):
+                raise BrokerFailure("INVALID_RECEIPT")
+            token = self.store.tokens().get(challenge)
+            if (
+                not isinstance(token, dict)
+                or token.get("session_id") != session_id
+                or token.get("runtime_generation") != request.get("runtime_generation")
+                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:
+                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):
+                raise BrokerFailure("RECEIPT_REPLAY")
+            if token.get("binding") != lease.get("binding") or not valid_receipt_evidence(token.get("evidence")):
+                raise BrokerFailure("PROMOTION_TOKEN_INVALID")
+            del self.store.tokens()[challenge]
             lease["expires_at"] = time.monotonic() + int(lease["ttl_seconds"])
-            # handle() commits token consumption before finish_promotion() makes
+            # handle() commits the evidence-backed consumption before
-            # VERIFIED externally visible: promote-last by construction.
+            # finish_promotion() makes VERIFIED externally visible: promote-last.
             return {"ok": True, "state": LEASE_PENDING_PROMOTION}
         if action == "revoke_lease":
             session_id, _ = self.authenticate(peer_pid, request)

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

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+"""Broker-side construction and verification for one-time receipt challenges."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import struct
+from typing import Final
+
+
+LATEST_ASSISTANT_DOMAIN_SEPARATOR: Final = b"MOSAIC/H_LATEST_ASSISTANT/v1\x00"
+
+
+def _length_frame(value: bytes) -> bytes:
+    return struct.pack(">Q", len(value)) + value
+
+
+def receipt_for(challenge: str, binding: dict[str, object]) -> str:
+    """Return the sole receipt text a model may copy for this broker cycle."""
+
+    return (
+        "MOSAIC-RECEIPT{"
+        f"challenge={challenge}; "
+        f"H_payload={binding['h_payload']}; "
+        f"gen={binding['runtime_generation']}; "
+        f"cep={binding['compaction_epoch']}"
+        "}"
+    )
+
+
+def is_verbatim_receipt(message: str, challenge: str, binding: dict[str, object]) -> bool:
+    """Require the exact one current-cycle receipt, not a transcript substring."""
+
+    expected = receipt_for(challenge, binding)
+    return hmac.compare_digest(message, expected)
+
+
+def latest_assistant_digest(message: str) -> str:
+    """Record the broker-computed digest of the exact observed assistant entry."""
+
+    encoded = message.encode("utf-8")
+    return hashlib.sha256(LATEST_ASSISTANT_DOMAIN_SEPARATOR + _length_frame(encoded)).hexdigest()