Keep recovery independent of its phase writer interface

Exercise direct recovering and terminal OSError injection as well as atomic receipt faults. Preserve the real restored estate and complete watches while evidence loss remains failed and unverified.

Change-source: pulse-maintainer

.github/scripts/demo-runtime-transaction.py

@@ -417,11 +417,14 @@ class Transaction:
 
     def recovery_save(self, status):
         """Receipt storage must not gate the already-owned restoration."""
+        self.receipt["status"] = status
         try:
             self.save(status)
-        except Failure as error:
-            if str(error) != "receipt-observation":
+        except (OSError, Failure) as error:
+            if isinstance(error, Failure) and str(error) != "receipt-observation":
                 raise
+            if status not in self.receipt["observation_failures"]:
+                self.receipt["observation_failures"].append(status)
 
     def finish(self, status):
         """Best-effort failed evidence is not a verified terminal outcome."""
@@ -437,9 +440,11 @@ class Transaction:
             unverified()
         try:
             self.save(self.receipt["status"] if self.receipt["observation_failures"] else status)
-        except Failure as error:
-            if str(error) != "receipt-observation":
+        except (OSError, Failure) as error:
+            if isinstance(error, Failure) and str(error) != "receipt-observation":
                 raise
+            if self.receipt["status"] not in self.receipt["observation_failures"]:
+                self.receipt["observation_failures"].append(self.receipt["status"])
             unverified()
             # One bounded attempt to retain the failure, not to obtain a pass.
             self.recovery_save("observation_failed")

.github/scripts/tests/test_demo_runtime_transaction.py

@@ -318,6 +318,35 @@ class TransactionTest(unittest.TestCase):
         self.assertEqual(tx.receipt["forward"]["elapsed_seconds"], 50)
         self.assertEqual(tx.receipt, json.loads((self.attempt / "receipt.json").read_text()))
 
+    def test_recovery_survives_direct_phase_and_terminal_writer_oserror(self):
+        for phase in ("recovering", "rolled_back"):
+            with self.subTest(phase=phase):
+                req = request("update"); req["version"] = "v6.4.6"
+                tx = engine.Transaction(self.host, req, self.attempt)
+                self.host.transaction = tx; self.host.fail_at = self.host.clock + 55
+                self.host.failed = False
+                original = tx.save
+
+                def write(status):
+                    if status == phase:
+                        raise OSError("private writer error")
+                    original(status)
+
+                with patch.object(tx, "save", side_effect=write):
+                    code = tx.run()
+                self.assertEqual((code, tx.receipt["status"], tx.receipt["rollback"]),
+                                 (2, "observation_failed", "unverified"))
+                self.assertEqual(engine.estate_hash(self.paths), self.before)
+                self.assertEqual(tx.receipt["recovery"]["elapsed_seconds"], 300)
+                self.assertEqual(tx.receipt["failure"], "new-service-crash")
+                self.assertEqual(tx.receipt, json.loads((self.attempt / "receipt.json").read_text()))
+                # Keep each independent snapshot/intent; never reuse its path.
+                self.attempt = self.root / "attempts" / ("b" * 64)
+                if phase == "recovering":
+                    # The prior unverified estate correctly blocks a new
+                    # request; use an independent fixture for the next fault.
+                    self.doCleanups(); self.setUp()
+
     def test_recovery_terminal_receipt_failure_retains_full_restoration(self):
         tx = self.run_fault("rolled_back")
         self.assertEqual(tx.receipt["observed_outcome"], "rolled_back")