fix(drain): fail closed on missing or unproven acknowledgement evidence (#661)

The acks_or_timeout check treated an absent `acks` key as proof that no
session needed to acknowledge: `drain_state.get("acks") or {}` collapsed
absent, None, and empty into the same value, and the resulting empty mapping
satisfied `no_sessions_to_ack`. The impact report's counts.sessions_live_other
was never consulted, so absence of evidence was read as evidence of absence.

Reproduced at head 1cbbde0089: with
sessions_live_other = 3 and the acknowledgement key absent, acks_or_timeout
passed with detail "no other live sessions required to acknowledge", the proof
minted clean, and gate_apply_restart returned verdict allow — a restart
authorized against three live sessions with zero acknowledgement evidence, and
the resulting artifact carried a valid signature.

Whether acknowledgement is required is now derived from the impact report,
never from the shape of the drain state:

- _live_session_count() reads counts.sessions_live_other and returns None for a
  missing, malformed, negative, or bool value, so an unreadable report fails
  closed instead of reading as "nobody was live".
- Absent, None, non-mapping, empty, partially-covering, and unparseable or
  stale acknowledgement data all fail closed while live sessions require
  acknowledgement.
- _is_acknowledged() no longer coerces with str(); only an explicit
  "ack"/"acked"/"acknowledged" string counts, so None, timestamps, and
  "pending"/"stale" markers are never read as an acknowledgement.
- Present-but-unacknowledged entries fail closed even when the report claims
  zero live sessions: that contradiction is not safe to resolve in favour of
  the restart.
- ack_timeout_policy_applied stays strict (`value is True`), so an absent, null,
  or non-boolean value cannot open the gate on its own.

The genuine no-other-live-sessions case still passes, now justified by the
report proving sessions_live_other == 0 rather than by the absence of data.

Adds AcknowledgementFailClosedTests: 14 cases / 26 subtests covering missing,
null, empty, malformed, stale, partial-coverage, and unproven-count inputs,
the valid-acknowledgement and zero-live-session paths, timeout-policy
strictness, and that a failed check blocks proof.clean, verification, and the
restart gate.

Restart-surface suite: 132 passed, 38 subtests (branch baseline 118 passed,
12 subtests; +14 new tests, no regressions).

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01VEaP3TohHLFWkp3Z2mmuZw

drain_proof.py

@@ -71,6 +71,11 @@ CHECK_HANDOFFS_OK = "handoffs_ok"
 CHECK_LEASES_HANDLED = "leases_handled"
 CHECK_ACKS_OR_TIMEOUT = "acks_or_timeout"
 
+# The only values accepted as an explicit acknowledgement from a live session.
+# Anything else — including a missing entry, a null, or a "pending"/"stale"
+# marker — leaves that session unacknowledged.
+_ACK_TOKENS = frozenset({"ack", "acked", "acknowledged"})
+
 REQUIRED_CHECKS: tuple[str, ...] = (
     CHECK_NO_INFLIGHT_MUTATIONS,
     CHECK_ASSIGNMENTS_STOPPED,
@@ -291,6 +296,38 @@ def _bool_input(value: Any) -> bool:
     return value is True
 
 
+def _live_session_count(impact_report: Mapping[str, Any]) -> int | None:
+    """Other-live-session count from the report, or ``None`` when unproven.
+
+    Only a real, non-negative integer counts. A missing ``counts`` block, a
+    malformed one, a non-integer, a bool (``True`` is an ``int`` in Python), or
+    a negative value all return ``None`` so the caller fails closed rather than
+    treating an unreadable report as "nobody was live".
+    """
+
+    counts = impact_report.get("counts")
+    if not isinstance(counts, Mapping):
+        return None
+    value = counts.get("sessions_live_other")
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        return None
+    return value
+
+
+def _is_acknowledged(value: Any) -> bool:
+    """True only for an explicit acknowledgement token.
+
+    Deliberately strict: the value must already be a string carrying one of the
+    recognised tokens. Non-strings are not coerced, so ``None``, timestamps,
+    objects, and states such as "pending" or "stale" are never read as an
+    acknowledgement.
+    """
+
+    if not isinstance(value, str):
+        return False
+    return value.strip().lower() in _ACK_TOKENS
+
+
 def _evaluate_checks(
     impact_report: Mapping[str, Any],
     drain_state: Mapping[str, Any],
@@ -406,24 +443,92 @@ def _evaluate_checks(
         )
 
     # 6. Acknowledgements received, or an explicit timeout policy was applied.
-    acks = drain_state.get("acks") or {}
+    #
-    ack_values = list(acks.values()) if isinstance(acks, Mapping) else []
+    # Whether acknowledgements are *required* is answered by the impact report,
-    all_acked = bool(ack_values) and all(
+    # never inferred from the shape of the drain state. Previously an absent
-        str(v).strip().lower() in {"ack", "acked", "acknowledged"}
+    # ``acks`` key collapsed to ``{}`` and was read as "no session needed to
-        for v in ack_values
+    # acknowledge", so a proof minted clean — and the restart gate allowed —
-    )
+    # while the report still showed other live sessions. Absence of evidence is
-    no_sessions_to_ack = isinstance(acks, Mapping) and len(ack_values) == 0
+    # not evidence of absence: missing, malformed, stale, or otherwise unproven
+    # acknowledgement data fails closed whenever live sessions require
+    # acknowledgement, and only explicitly verified acknowledgement evidence
+    # may permit the operation.
+    sessions_live_other = _live_session_count(impact_report)
+    live_count_known = sessions_live_other is not None
+    # An unknown or malformed count fails closed: it cannot prove nobody had to
+    # acknowledge, so acknowledgement stays required.
+    acks_required = (not live_count_known) or sessions_live_other > 0
+
+    raw_acks = drain_state.get("acks")
+    acks_present = isinstance(raw_acks, Mapping)
+    ack_values = list(raw_acks.values()) if acks_present else []
+    acked_count = sum(1 for value in ack_values if _is_acknowledged(value))
+    every_entry_acked = bool(ack_values) and acked_count == len(ack_values)
+    covers_live_sessions = live_count_known and acked_count >= sessions_live_other
+    # Strict by construction: only an explicit ``True`` counts, so an absent,
+    # null, or non-boolean ``ack_timeout_policy_applied`` can never open this
+    # gate on its own.
     timeout_policy = _bool_input(drain_state.get("ack_timeout_policy_applied"))
-    acks_ok = all_acked or no_sessions_to_ack or timeout_policy
+
-    if acks_ok:
+    # Present-but-unacknowledged entries always fail closed, even when the
-        if timeout_policy and not all_acked:
+    # report claims nobody was live: the drain state naming an outstanding
-            detail = "explicit ack timeout policy applied"
+    # session contradicts that claim, and the safe reading of a contradiction
-        elif no_sessions_to_ack:
+    # is that an acknowledgement is still owed.
-            detail = "no other live sessions required to acknowledge"
+    outstanding_entries = acks_present and bool(ack_values) and not every_entry_acked
+
+    if timeout_policy and (acks_required or outstanding_entries):
+        acks_ok = True
+        detail = "explicit ack timeout policy applied and recorded"
+    elif outstanding_entries:
+        acks_ok = False
+        detail = (
+            f"{len(ack_values) - acked_count} of {len(ack_values)} "
+            "acknowledgement entries are unparseable, stale, or not "
+            "acknowledged (fail closed)"
+        )
+    elif not acks_required:
+        acks_ok = True
+        detail = (
+            "impact report proves no other live sessions required to "
+            "acknowledge (sessions_live_other=0)"
+        )
+    elif every_entry_acked and covers_live_sessions:
+        acks_ok = True
+        detail = (
+            f"{acked_count} acknowledgement entr"
+            f"{'y' if acked_count == 1 else 'ies'} verified; covers "
+            f"sessions_live_other={sessions_live_other}"
+        )
     else:
-            detail = "all affected sessions acknowledged"
+        acks_ok = False
+        if not live_count_known:
+            detail = (
+                "impact report does not prove the live-session count; "
+                "acknowledgement required and unproven (fail closed)"
+            )
+        elif raw_acks is None:
+            detail = (
+                "acknowledgement evidence absent while "
+                f"sessions_live_other={sessions_live_other} require "
+                "acknowledgement (fail closed)"
+            )
+        elif not acks_present:
+            detail = (
+                "acknowledgement evidence malformed: expected a mapping, got "
+                f"{type(raw_acks).__name__} (fail closed)"
+            )
+        elif not ack_values:
+            detail = (
+                "acknowledgement mapping empty while "
+                f"sessions_live_other={sessions_live_other} require "
+                "acknowledgement (fail closed)"
+            )
         else:
-        detail = "outstanding acks with no timeout policy (fail closed)"
+            detail = (
+                f"acknowledgements cover only {acked_count} session(s) but the "
+                f"report shows sessions_live_other={sessions_live_other} "
+                "(fail closed)"
+            )
     checks.append(DrainCheck(CHECK_ACKS_OR_TIMEOUT, acks_ok, detail))
 
     return checks

tests/test_drain_proof.py

@@ -379,5 +379,218 @@ class SecretHygieneTests(unittest.TestCase):
         self.assertNotIn(SECRET.decode(), blob)
 
 
+def _drained_report_with_live_sessions(count: int) -> dict:
+    """Report with ``count`` other live sessions but nothing in flight.
+
+    Every other checklist item passes against this report, so a failure
+    isolates the acknowledgement check rather than tripping on mutations.
+    """
+
+    sessions = [
+        {
+            "session_id": "prgs-controller-1-req",
+            "role": "controller",
+            "profile": "prgs-controller",
+            "pid": _live_pid(),
+            "status": "active",
+            "last_heartbeat_at": NOW.isoformat(),
+        }
+    ]
+    for index in range(count):
+        sessions.append(
+            {
+                "session_id": f"prgs-author-{index}",
+                "role": "author",
+                "profile": "prgs-author",
+                "pid": _live_pid(),
+                "status": "active",
+                "last_heartbeat_at": NOW.isoformat(),
+            }
+        )
+    report = rc.evaluate_restart_impact(
+        {"sessions": sessions, "leases": [], "inventory_complete": True},
+        now=NOW,
+        requesting_session_id="prgs-controller-1-req",
+    )
+    return report.as_dict()
+
+
+class AcknowledgementFailClosedTests(unittest.TestCase):
+    """Acknowledgement evidence must fail closed unless explicitly verified.
+
+    Regression cover for the reviewed fail-open on PR #882: an absent ``acks``
+    key collapsed to ``{}`` and was read as "no other live sessions required to
+    acknowledge", so a proof minted clean and the restart gate allowed while the
+    impact report still showed other live sessions.
+    """
+
+    def _state(self, **overrides) -> dict:
+        state = _clean_drain_state()
+        state.pop("acks", None)
+        state["ack_timeout_policy_applied"] = False
+        state.update(overrides)
+        return state
+
+    def _acks_check(self, proof) -> dp.DrainCheck:
+        return next(c for c in proof.checks if c.name == dp.CHECK_ACKS_OR_TIMEOUT)
+
+    def _build(self, report: dict, state: dict):
+        return dp.build_drain_proof(
+            impact_report=report, drain_state=state, now=NOW, secret=SECRET
+        )
+
+    def assertAcksFailClosed(self, report: dict, state: dict) -> None:
+        proof = self._build(report, state)
+        self.assertFalse(self._acks_check(proof).passed)
+        self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
+        self.assertFalse(proof.clean)
+
+    # --- missing / null / empty / malformed ------------------------------
+
+    def test_missing_acks_key_with_live_sessions_fails_closed(self):
+        """The exact reviewed defect: absent key, three other live sessions."""
+        report = _drained_report_with_live_sessions(3)
+        self.assertEqual(report["counts"]["sessions_live_other"], 3)
+        state = self._state()
+        self.assertNotIn("acks", state)
+        proof = self._build(report, state)
+        check = self._acks_check(proof)
+        self.assertFalse(check.passed)
+        self.assertNotIn("no other live sessions", check.detail)
+        self.assertIn("fail closed", check.detail)
+        self.assertFalse(proof.clean)
+        self.assertEqual(proof.failed_checks, [dp.CHECK_ACKS_OR_TIMEOUT])
+
+    def test_none_acks_with_live_sessions_fails_closed(self):
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(2), self._state(acks=None)
+        )
+
+    def test_empty_acks_with_live_sessions_fails_closed(self):
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(1), self._state(acks={})
+        )
+
+    def test_malformed_acks_fail_closed(self):
+        for malformed in ([], "ack", 7, ("ack",), True):
+            with self.subTest(malformed=malformed):
+                self.assertAcksFailClosed(
+                    _drained_report_with_live_sessions(1),
+                    self._state(acks=malformed),
+                )
+
+    # --- stale / unproven values -----------------------------------------
+
+    def test_stale_or_unproven_ack_values_fail_closed(self):
+        for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    _drained_report_with_live_sessions(1),
+                    self._state(acks={"prgs-author-0": value}),
+                )
+
+    def test_partial_coverage_fails_closed(self):
+        """Fewer acknowledgements than the report's live-session count."""
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(3),
+            self._state(acks={"prgs-author-0": "ack"}),
+        )
+
+    def test_one_unacked_entry_among_many_fails_closed(self):
+        self.assertAcksFailClosed(
+            _drained_report_with_live_sessions(2),
+            self._state(acks={"prgs-author-0": "ack", "prgs-author-1": "pending"}),
+        )
+
+    def test_unproven_live_session_count_fails_closed(self):
+        """A missing or malformed count cannot prove nobody had to acknowledge."""
+        malformed_counts = (
+            None,
+            {},
+            {"sessions_live_other": None},
+            {"sessions_live_other": "3"},
+            {"sessions_live_other": -1},
+            {"sessions_live_other": True},
+        )
+        for counts in malformed_counts:
+            with self.subTest(counts=counts):
+                report = _drained_report_with_live_sessions(0)
+                if counts is None:
+                    report.pop("counts", None)
+                else:
+                    report["counts"] = counts
+                self.assertAcksFailClosed(report, self._state())
+
+    # --- valid evidence still passes -------------------------------------
+
+    def test_complete_valid_acks_pass(self):
+        report = _drained_report_with_live_sessions(2)
+        state = self._state(
+            acks={"prgs-author-0": "ack", "prgs-author-1": "acknowledged"}
+        )
+        proof = self._build(report, state)
+        self.assertTrue(self._acks_check(proof).passed)
+        self.assertTrue(proof.clean)
+        self.assertEqual(proof.failed_checks, [])
+
+    def test_no_other_live_sessions_still_passes(self):
+        """Intended behavior retained: zero live sessions needs no acks."""
+        report = _drained_report_with_live_sessions(0)
+        self.assertEqual(report["counts"]["sessions_live_other"], 0)
+        proof = self._build(report, self._state())
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertIn("sessions_live_other=0", check.detail)
+        self.assertTrue(proof.clean)
+
+    # --- timeout policy cannot become a second fail-open ------------------
+
+    def test_unproven_timeout_policy_cannot_open_the_gate(self):
+        for value in (None, "true", "yes", 1, "True", [], {}):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    _drained_report_with_live_sessions(2),
+                    self._state(ack_timeout_policy_applied=value),
+                )
+
+    def test_explicit_timeout_policy_permits(self):
+        proof = self._build(
+            _drained_report_with_live_sessions(2),
+            self._state(ack_timeout_policy_applied=True),
+        )
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertIn("timeout policy", check.detail)
+        self.assertTrue(proof.clean)
+
+    # --- the gate itself must deny ---------------------------------------
+
+    def test_failed_ack_check_denies_the_restart_gate(self):
+        report = _drained_report_with_live_sessions(3)
+        proof = self._build(report, self._state())
+        self.assertFalse(proof.clean)
+        decision = dp.gate_apply_restart(
+            proof=proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertFalse(decision.allow)
+        self.assertEqual(decision.verdict, dp.GATE_DENY)
+        self.assertIsNotNone(decision.incident)
+
+    def test_unclean_ack_proof_fails_verification(self):
+        report = _drained_report_with_live_sessions(3)
+        proof = self._build(report, self._state())
+        result = dp.verify_drain_proof(
+            proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertFalse(result.valid)
+        self.assertFalse(result.clean)
+
+
 if __name__ == "__main__":
     unittest.main()