feat(mosaic): add correlated lease promotion CLI

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

@@ -112,13 +112,20 @@ class PromotionHookFixture(unittest.TestCase):
         self.pending_file.write_text(challenge, encoding="utf-8")
         self.pending_file.chmod(0o600)
 
-    def run_complete(self, runner: mock.Mock) -> tuple[int, str]:
+    def run_complete(
+        self,
+        runner: mock.Mock,
+        now: float | None = None,
+    ) -> tuple[int, str]:
         stderr = io.StringIO()
-        result = self.complete.main(
-            environ=self.environment,
-            stderr=stderr,
-            run=runner,
-        )
+        options: dict[str, object] = {
+            "environ": self.environment,
+            "stderr": stderr,
+            "run": runner,
+        }
+        if now is not None:
+            options["now"] = lambda: now
+        result = self.complete.main(**options)
         return result, stderr.getvalue()
 
 
@@ -292,6 +299,62 @@ class PromotionCompleteTest(PromotionHookFixture):
         self.assertFalse(self.pending_file.exists())
         self.assertEqual(runner.call_args.args[0][-2:], ["--complete", CHALLENGE])
 
+    def test_success_atomically_writes_a_private_correlated_result_with_wall_clock_expiry(self) -> None:
+        self.write_pending()
+        runner = mock.Mock(
+            return_value=self.completed(
+                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
+            )
+        )
+
+        with mock.patch.object(self.complete.os, "replace", wraps=os.replace) as replace:
+            result, _stderr = self.run_complete(runner, now=12_345.0)
+
+        result_file = self.pending_dir / "last-result.json"
+        self.assertEqual(result, 0)
+        self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
+        self.assertEqual(
+            json.loads(result_file.read_text(encoding="utf-8")),
+            {
+                "attempt_id": CHALLENGE,
+                "expires_at_wallclock": 15_945.0,
+                "reason": None,
+                "session_id": SESSION_ID,
+                "ts": 12_345.0,
+                "verified": True,
+            },
+        )
+        temporary, destination = replace.call_args.args
+        self.assertRegex(temporary, r"^\.last-result\.json\.tmp-[0-9a-f]+$")
+        self.assertEqual(destination, "last-result.json")
+        self.assertFalse(any(path.name.startswith(".last-result.json.tmp-") for path in self.pending_dir.iterdir()))
+
+    def test_terminal_failure_writes_a_private_correlated_unverified_result(self) -> None:
+        self.write_pending()
+        runner = mock.Mock(
+            return_value=self.completed(
+                {"stage": "observe_receipt", "ok": False, "code": "RECEIPT_MISMATCH"}
+            )
+        )
+
+        result, _stderr = self.run_complete(runner, now=12_345.0)
+
+        result_file = self.pending_dir / "last-result.json"
+        self.assertEqual(result, 0)
+        self.assertFalse(self.pending_file.exists())
+        self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
+        self.assertEqual(
+            json.loads(result_file.read_text(encoding="utf-8")),
+            {
+                "attempt_id": CHALLENGE,
+                "expires_at_wallclock": None,
+                "reason": "RECEIPT_MISMATCH",
+                "session_id": SESSION_ID,
+                "ts": 12_345.0,
+                "verified": False,
+            },
+        )
+
     def test_each_terminal_failure_deletes_pending_file(self) -> None:
         terminal_codes = (
             "RECEIPT_REPLAY",