test(task-plane): cover fail and recovery rollback

tests/task_plane/public_interface/test_task_plane_security_contract.py

@@ -252,6 +252,63 @@ class AtomicMutationContractTests(unittest.TestCase):
         self.assertEqual([], service.store.list_evidence(task["task_id"]))
         self.assertEqual("RUNNING", service.store.get_task(task["task_id"]).status)
 
+    def test_fail_rolls_back_claim_state_and_event(self):
+        service = _service()
+        task, claim = _claim_for_service(service, task_id="task-fail-atomic")
+        service.store.fault_injector = self._fault("fail.after_state_before_commit")
+        with self.assertRaises(RuntimeError):
+            service.store.fail_claim_with_event(
+                task["task_id"], claim["claim_id"], "FAILED_FINAL", "injected"
+            )
+        self.assertEqual("ACTIVE", service.store.get_claim(claim["claim_id"]).status)
+        self.assertEqual("CLAIMED", service.store.get_task(task["task_id"]).status)
+        self.assertNotIn(
+            "TASK_FAILED",
+            {
+                event["type"]
+                for event in service.store.list_events(task["task_id"])
+            },
+        )
+
+    def test_recovery_rolls_back_probe_state_and_event(self):
+        service = _service(
+            probe_attestation_verifier=lambda probe: probe.attestation_ref
+            == "trusted://probe/atomic"
+        )
+        task, claim = _claim_for_service(service, task_id="task-recover-atomic")
+        service.store.conn.execute(
+            "UPDATE claims SET lease_until=? WHERE claim_id=?",
+            (time.time() - 10, claim["claim_id"]),
+        )
+        probe = ProbeEvidence.create(
+            task_id=task["task_id"],
+            claim_id=claim["claim_id"],
+            worker_id=WORKER_ID,
+            worker_identity=WORKER_IDENTITY,
+            operation_digest=claim["operation_digest"],
+            outcome=ProbeOutcome.NOT_APPLIED,
+            observed_at=time.time(),
+            expires_at=time.time() + 60,
+            probe_executor_identity=ADMIN_IDENTITY,
+            attestation_ref="trusted://probe/atomic",
+        )
+        service.store.fault_injector = self._fault(
+            "recover.after_probe_before_state"
+        )
+        with self.assertRaises(RuntimeError):
+            service.store.recover_claim_with_probe_and_event(
+                task["task_id"], claim["claim_id"], probe
+            )
+        self.assertEqual(
+            0,
+            service.store.conn.execute(
+                "SELECT COUNT(*) FROM probe_evidence WHERE probe_id=?",
+                (probe.probe_id,),
+            ).fetchone()[0],
+        )
+        self.assertEqual("ACTIVE", service.store.get_claim(claim["claim_id"]).status)
+        self.assertEqual("CLAIMED", service.store.get_task(task["task_id"]).status)
+
 
 class WorkerOwnershipContractTests(unittest.TestCase):
     def test_identity_takeover_escalation_and_disabled_reregistration_rejected(self):