test(#832): retain revoke fence on rejected receipt cycles

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

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import base64
+import copy
 import hashlib
 import importlib.util
 import os
@@ -151,6 +152,46 @@ class ReceiptObserverTest(BrokerFixture):
             "tool_name": "bash",
         })["decision"], "allow")
 
+    def test_rejected_begin_keeps_revoke_first_fence_for_all_construction_refusals(self) -> None:
+        construction, binding = self.construction()
+        refusal_cases = {
+            "INVALID_CONSTRUCTION": {"bad": "construction"},
+            "PAYLOAD_CONSTRUCTION_REFUSED": {
+                **construction,
+                "fragments": [{
+                    **construction["fragments"][0],
+                    "expected_sha256": "0" * 64,
+                }],
+            },
+            "PAYLOAD_BINDING_MISMATCH": None,
+        }
+        for expected_code, rejected_construction in refusal_cases.items():
+            with self.subTest(expected_code=expected_code):
+                verified = self.begin(binding, construction)
+                self.record(verified["receipt"])
+                self.observe(verified["receipt_challenge"])
+                self.assertEqual(self.promote(verified["receipt_challenge"])["state"], DAEMON.LEASE_VERIFIED)
+
+                rejected_binding = copy.deepcopy(binding)
+                if expected_code == "PAYLOAD_BINDING_MISMATCH":
+                    rejected_binding["h_payload"] = "f" * 64
+                    rejected_construction = construction
+                with self.assertRaisesRegex(DAEMON.BrokerFailure, expected_code):
+                    self.begin(rejected_binding, rejected_construction)
+
+                self.assertEqual(
+                    self.broker.leases[self.session_id]["state"], DAEMON.LEASE_UNVERIFIED
+                )
+                denied = self.broker.handle(self.peer, {
+                    "action": "authorize_tool",
+                    "session_id": self.session_id,
+                    "runtime_generation": 7,
+                    "runtime": "pi",
+                    "tool_name": "bash",
+                })
+                self.assertEqual(denied["decision"], "deny")
+                self.assertEqual(denied["state"], DAEMON.LEASE_UNVERIFIED)
+
     def test_t26_stale_epoch_receipt_cannot_promote_against_shipped_binding(self) -> None:
         construction, stale_binding = self.construction()
         stale = self.begin(stale_binding, construction)