fix(release): close admission when maintenance confirmation fails

backend/tests/test_delivery.py

@@ -972,3 +972,60 @@ def test_journal_profile_never_substitutes_rotation_settings_for_a_dedicated_har
     observed[field] = value
     with pytest.raises(policy.DeliveryError):
         journal.validate_storage(observed, path)
+
+
+@pytest.mark.parametrize("failure", ["publication", "receipt", "readback"])
+def test_maintenance_does_not_leave_admission_open_when_final_confirmation_fails(
+    monkeypatch, failure
+):
+    # Pure controller error flow, no Root, manager or filesystem operations.
+    # Publication atomicity, actual stop and disk failure remain VM-only cases.
+    gate = []
+    maintenance = object.__new__(lifecycle.Maintenance)
+    maintenance.selected = {}
+    maintenance.payload = Path("/unused")
+    monkeypatch.setattr(maintenance, "change", lambda *_: None)
+
+    def publish(_, state):
+        gate.append(state)
+        if state == "open" and failure == "publication":
+            raise OSError("publication readback unavailable")
+
+    def save(_):
+        if failure == "receipt":
+            raise OSError("maintenance receipt unavailable")
+
+    monkeypatch.setattr(lifecycle.availability, "write", publish)
+    monkeypatch.setattr(maintenance, "save", save)
+    monkeypatch.setattr(
+        lifecycle.operator_units,
+        "run",
+        lambda *_: {
+            "service_preconditions_met": True,
+            "ready_for_admission": failure != "readback",
+        },
+    )
+    with pytest.raises((OSError, policy.DeliveryError)):
+        maintenance.start(resume=True)
+    assert gate == ["draining", "open", "draining"]
+
+
+def test_failed_gate_closure_requires_known_api_stop_without_forging_owner_cleanup(monkeypatch):
+    # Controller flow only; actual cgroup/stop/full-volume evidence is VM-only.
+    maintenance = object.__new__(lifecycle.Maintenance)
+    maintenance.selected = {}
+    calls = []
+
+    def unavailable(*_):
+        raise OSError("publication unavailable")
+
+    monkeypatch.setattr(lifecycle.availability, "write", unavailable)
+    monkeypatch.setattr(
+        maintenance, "unit", lambda name: {"ActiveState": "inactive", "MainPID": "0"}
+    )
+    monkeypatch.setattr(lifecycle.service_install, "systemctl", lambda *args: calls.append(args))
+    monkeypatch.setattr(
+        maintenance, "empty_cgroup", lambda name: calls.append(("prove_empty", name))
+    )
+    maintenance.close_failed_admission()
+    assert calls == [("--no-block", "stop", "kf-api.service"), ("prove_empty", "api")]

scripts/release/lifecycle.py

@@ -446,15 +446,37 @@ class Maintenance:
         ready = operator_units.run("node-health", self.payload)
         if ready.get("service_preconditions_met") is not True:
             raise DeliveryError("service preconditions unavailable; admission remains closed")
-        availability.write(self.selected, "open")
         try:
+            availability.write(self.selected, "open")
             ready = operator_units.run("node-health", self.payload)
             if ready.get("ready_for_admission") is not True:
                 raise DeliveryError("admission readiness readback unconfirmed")
+            self.save("accepting")
         except (OSError, ValueError, subprocess.SubprocessError):
-            availability.write(self.selected, "draining")
+            self.close_failed_admission()
             raise
-        self.save("accepting")
+
+    def close_failed_admission(self) -> None:
+        try:
+            availability.write(self.selected, "draining")
+            return
+        except (OSError, ValueError, subprocess.SubprocessError):
+            pass
+        # A full/read-only availability filesystem must not leave a successful
+        # earlier publication accepting mutations. Stop only the verified API
+        # target without depending on another receipt write. Owners retain work
+        # and capacity; this is not a factory cleanup or terminal-state proof.
+        self.unit("api")
+        service_install.systemctl("--no-block", "stop", "kf-api.service")
+        deadline = time.monotonic() + 50
+        while True:
+            values = self.unit("api")
+            if values["ActiveState"] == "inactive" and values["MainPID"] == "0":
+                self.empty_cgroup("api")
+                return
+            if time.monotonic() >= deadline or values["ActiveState"] == "failed":
+                raise DeliveryError("admission closure/API stop unconfirmed; operator required")
+            time.sleep(0.5)
 
 
 def run(action: str) -> dict:
@@ -473,9 +495,17 @@ def run(action: str) -> dict:
         install.exclusive_file(lockpath, b"", 0o600)
     preflight.protected(lockpath)
     configuration.private(lockpath)
+    if lockpath.stat().st_size != 0:
+        raise DeliveryError("unrecorded maintenance lock contents")
     fd = os.open(lockpath, os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC)
     try:
         fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        # Refuse unknown history before publishing availability, including
+        # resume where every known service could already be running.
+        if LAST.exists() or LAST.is_symlink():
+            preflight.protected(LAST)
+            configuration.private(LAST)
+            validate_record(read_json(LAST, 32768))
         maintenance = Maintenance(selected, paths, payload, action)
         try:
             if action in {"drain", "stop", "restart"}: