Recover EP purge after partial process crash

docs/development/ENGINEERING_PLATFORM_ROADMAP.md

@@ -9,6 +9,16 @@ may extend productive work; reasoning text cannot, and the hard maximum never
 resets. The Operations Console projects both limits. See
 [the owning design](../engineering/PROGRESS_AWARE_PROVIDER_DEADLINES_V2.md).
 
+## EP Server system-instance process-crash qualification — 2026-09-29
+
+The bounded process-crash subset of the existing OI-2 system-instance
+lifecycle is tracked in [EP Server system/multi-instance lifecycle](../engineering/EP_SERVER_SYSTEM_MULTI_INSTANCE_V1.md#abrupt-process-stop-qualification-revision-26)
+and canonical LANE_2 register `pcvantol/forge#142` revision 26. The public
+2.3.104 wheel has a reproduced partial-`PURGE` crash gap. Source and locally
+built candidate qualification are in progress; this note creates no new node,
+does not change broader OI-2 status and claims no published fix or installer
+acceptance.
+
 ## Release and operational installation lifecycle V1 — active coordinated plan
 
 Repository evidence recorded on 2026-09-09 reads `origin/main` as

docs/engineering/EP_SERVER_SYSTEM_MULTI_INSTANCE_V1.md

@@ -292,3 +292,49 @@ tamper rejection, explicit permanent purge, sibling-instance non-interference,
 lost-response recovery for preserve/restore/purge, receipt replay and existing
 destructive `remove` compatibility. Qualification uses isolated product
 fixtures only; no production Server/CENTRAL/credential mutation is authorized.
+
+### Abrupt process-stop qualification (revision 26)
+
+The historical 2.3.104 installed-wheel qualification above covered controlled
+interruption and lost responses. The separate LANE_2 revision-26 qualification
+found a real SIGKILL gap in `PURGE`: the legacy `REMOVE` receipt is published
+before recursive deletion. If the process dies after deleting `instance.json`
+but before the instance root and lifecycle tombstone are removed, the normal
+same-operation `purge` retry cannot reopen the descriptor. The exact published
+2.3.104 wheel reproduces this failure under Python 3.14; it remains historical
+`GAP_PROVEN` evidence and is not relabeled.
+
+The bounded recovery correction retains the prepared root's device/inode and
+minimal topology outside the mutable instance tree. After a terminal exact
+`REMOVE` receipt, recovery admits only that same root, rejects a replacement
+root or changed surviving descriptor, checks the remaining tree for unsafe
+entries, and finishes the original deletion before publishing the purge
+tombstone and lifecycle receipt. The existing lifecycle request and receipt
+identities remain unchanged. The reproducible installed-wheel SIGKILL harness
+is `tools/qualification/system_instance_purge_sigkill.py`; it reports durable
+`PURGE_READY`, physical partial deletion, exact mutator signal exit, surviving
+descendants, same-operation recovery, terminal replay and sibling byte identity.
+
+`tools/qualification/system_instance_lifecycle_sigkill.py` exercises the
+product-owned process-stop matrix through the installed provisioner using two
+disposable instances and a filesystem-only service adapter. Its 2.3.104
+published-wheel runs observed `PREPARED`, `VERIFIED`, receipt-before-`COMPLETE`
+and `COMPLETE` for PRESERVE and RESTORE. They observed `PREPARED`, `PURGE_READY`,
+terminal REMOVE receipt, tombstone-before-receipt, lifecycle receipt-before-
+`COMPLETE` and `COMPLETE` for PURGE. Each of these cells resumed the same
+operation to the expected terminal status with identical replay and unchanged
+sibling bytes. The separate partial physical deletion cell remains
+`GAP_PROVEN` on published 2.3.104 and passed only with the local correction
+candidate. A missed short phase is recorded as `NOT_HIT`, never promoted to a
+hit based on timing. The reusable harness also checks a held live lifecycle
+lock, changed request, foreign operation/instance status, OS-released lock
+reacquisition through normal retry, provider demotion, inactive restore,
+terminal replay and restore rejection after purge.
+
+Source tests and a locally built candidate-wheel SIGKILL run have passed this
+specific partial-deletion cell. The new protected source, any required normal
+release and requalification against its published bytes are still required
+before this process-crash subset may be called qualified. The remaining
+conflict, foreign-request and published-fixed-wheel checks must complete
+before the entire revision-26 subset can be called qualified. No physical
+reboot, power-loss, production service or installer acceptance is inferred.

src/engineering_platform/system_instance_lifecycle.py

@@ -477,40 +477,78 @@ def prepare_purge(
                 "PURGE_READY",
                 mutable_instance_data_digest=tree_digest,
                 descriptor_digest=request["descriptor_digest"],
+                # REMOVE publishes its receipt before deleting this tree.  A
+                # process kill can erase instance.json before rmtree finishes.
+                # Retain only the topology and root identity needed to resume
+                # that exact deletion without adopting a replacement root.
+                display_label=instance.display_label,
+                service_account=instance.service_account,
+                root_device=instance.root.stat(follow_symlinks=False).st_dev,
+                root_inode=instance.root.stat(follow_symlinks=False).st_ino,
             )
 
     def resume_purge_target(
         self,
-        instance: system_installation_topology.SystemInstanceTopology,
-        descriptor: Mapping[str, object],
+        instance_id: str,
         operation_id: str,
+        remove_receipt: Mapping[str, object],
     ) -> None:
-        root = self._operation_root(instance.instance_id, operation_id)
+        root = self._operation_root(instance_id, operation_id)
         state = _read_json(root / "state.json")
         if (
             state.get("contract") != CONTRACT
             or state.get("operation") != "PURGE"
             or state.get("phase") != "PURGE_READY"
-            or state.get("instance_id") != instance.instance_id
-            or state.get("descriptor_digest") != _json_digest(dict(descriptor))
+            or state.get("instance_id") != instance_id
+            or state.get("request") != {
+                "instance_id": instance_id,
+                "descriptor_digest": state.get("descriptor_digest"),
+            }
+            or state.get("request_digest") != _json_digest(state["request"])
+            or remove_receipt.get("operation") != "REMOVE"
+            or remove_receipt.get("operation_id") != operation_id
+            or remove_receipt.get("instance_id") != instance_id
+            or not isinstance(remove_receipt.get("evidence"), Mapping)
+            or remove_receipt["evidence"].get("removed_descriptor_digest") != state.get("descriptor_digest")
         ):
             raise SystemInstanceLifecycleError("EP purge recovery evidence is invalid")
-        with _Lock(instance.lifecycle_lock):
+        try:
+            instance = system_installation_topology.system_instance_topology(
+                self.product,
+                instance_id=instance_id,
+                display_label=state["display_label"],
+                service_account=state["service_account"],
+            )
+        except (KeyError, TypeError, ValueError) as error:
+            raise SystemInstanceLifecycleError("EP purge recovery topology is invalid") from error
+        expected = self.product.system_root / "instances" / instance.instance_id
+        if (
+            instance.root != expected
+            or instance.root.is_symlink()
+            or not instance.root.is_relative_to(self.product.system_root / "instances")
+        ):
+            raise SystemInstanceLifecycleError("EP purge recovery target is invalid")
+
+        def verify_original_root() -> None:
+            details = instance.root.stat(follow_symlinks=False)
+            if (details.st_dev, details.st_ino) != (state.get("root_device"), state.get("root_inode")):
+                raise SystemInstanceLifecycleError("EP purge recovery target was replaced")
+
+        # Check before opening the instance-owned lock, which may recreate a
+        # lock directory already erased by the interrupted deletion.
+        verify_original_root()
+        with _Lock(root / "purge-recovery.lock"), _Lock(instance.lifecycle_lock):
+            verify_original_root()
             self.controller.remove(instance)
             if self.controller.loaded(instance) or self.service_interpreter(instance) is not None:
                 raise SystemInstanceLifecycleError("EP purge recovery service was not fully removed")
-            if _tree_digest(
-                instance.root,
-                ignored_roots=(instance.operations_root / operation_id,),
-            ) != state.get("mutable_instance_data_digest"):
-                raise SystemInstanceLifecycleError("EP purge recovery target changed after preparation")
-            expected = self.product.system_root / "instances" / instance.instance_id
-            if (
-                instance.root != expected
-                or instance.root.is_symlink()
-                or not instance.root.is_relative_to(self.product.system_root / "instances")
-            ):
-                raise SystemInstanceLifecycleError("EP purge recovery target is invalid")
+            # The exact root survives until rmtree's final step.  Its contents
+            # may be a strict subset after SIGKILL, so the original whole-tree
+            # digest cannot be required here.  Reject unsafe remaining entries.
+            _tree_digest(instance.root, ignored_roots=(instance.operations_root / operation_id,))
+            if instance.descriptor.exists():
+                if _json_digest(_read_json(instance.descriptor)) != state.get("descriptor_digest"):
+                    raise SystemInstanceLifecycleError("EP purge recovery descriptor changed")
             shutil.rmtree(instance.root)
 
     def record_purge(

src/engineering_platform/system_instance_provisioner.py

@@ -665,8 +665,7 @@ def purge(self, instance_id: str, operation_id: str, *, confirm_instance_id: str
         else:
             remove_receipt = prior_remove
             if instance_root.exists() or instance_root.is_symlink():
-                instance, descriptor = self._instance_from_descriptor(instance_id)
-                lifecycle.resume_purge_target(instance, descriptor, operation_id)
+                lifecycle.resume_purge_target(instance_id, operation_id, remove_receipt)
         receipt = lifecycle.record_purge(instance_id, operation_id, remove_receipt)
         return {
             "contract": system_instance_lifecycle.CONTRACT,

tests/engineering/test_system_instance_provisioner.py

@@ -898,6 +898,66 @@ def test_purge_recovers_if_legacy_remove_receipt_precedes_rmtree(self) -> None:
             purged,
         )
 
+    def test_purge_recovers_when_process_dies_during_partial_tree_deletion(self) -> None:
+        self._create("ep-production-0001", "EP production", 8765)
+        self._create("ep-development-0001", "EP development", 8766)
+        instance = self._instance("ep-production-0001", "EP production")
+        sibling = self._instance("ep-development-0001", "EP development")
+        sibling_before = sibling.descriptor.read_bytes()
+        (instance.root / "data-sentinel").write_text("erase me", encoding="utf-8")
+
+        def partial_delete(_root: Path) -> None:
+            instance.descriptor.unlink()
+            (instance.root / "data-sentinel").unlink()
+            raise OSError("simulated abrupt stop after partial deletion")
+
+        with patch.object(provisioner_module.shutil, "rmtree", side_effect=partial_delete):
+            with self.assertRaisesRegex(OSError, "partial deletion"):
+                self.provisioner.purge(
+                    instance.instance_id, "purge-partial-delete",
+                    confirm_instance_id=instance.instance_id,
+                )
+        self.assertTrue(instance.root.exists())
+        self.assertFalse(instance.descriptor.exists())
+        self.assertIsNotNone(self.provisioner._terminal_receipt(
+            instance.instance_id, "purge-partial-delete", "REMOVE",
+        ))
+
+        receipt = self.provisioner.purge(
+            instance.instance_id, "purge-partial-delete",
+            confirm_instance_id=instance.instance_id,
+        )
+        self.assertEqual(receipt["receipt"]["evidence"]["lifecycle_state"], "PURGED")
+        self.assertFalse(instance.root.exists())
+        self.assertEqual(sibling.descriptor.read_bytes(), sibling_before)
+        self.assertEqual(self.provisioner.purge(
+            instance.instance_id, "purge-partial-delete",
+            confirm_instance_id=instance.instance_id,
+        ), receipt)
+
+    def test_purge_recovery_rejects_replaced_instance_root(self) -> None:
+        self._create("ep-production-0001", "EP production", 8765)
+        instance = self._instance("ep-production-0001", "EP production")
+        with patch.object(provisioner_module.shutil, "rmtree", side_effect=OSError("stop before deletion")):
+            with self.assertRaisesRegex(OSError, "stop before deletion"):
+                self.provisioner.purge(
+                    instance.instance_id, "purge-replaced-root",
+                    confirm_instance_id=instance.instance_id,
+                )
+        original = instance.root.with_name("original-instance-root")
+        instance.root.rename(original)
+        instance.root.mkdir()
+        with self.assertRaisesRegex(
+            provisioner_module.system_instance_lifecycle.SystemInstanceLifecycleError,
+            "replaced",
+        ):
+            self.provisioner.purge(
+                instance.instance_id, "purge-replaced-root",
+                confirm_instance_id=instance.instance_id,
+            )
+        self.assertTrue(instance.root.exists())
+        self.assertTrue(original.exists())
+
     def test_purge_recovers_after_remove_completed_before_lifecycle_projection(self) -> None:
         self._create("ep-production-0001", "EP production", 8765)
         instance = self._instance("ep-production-0001", "EP production")