From 824c42f7e3f436eef05272276c9ced958c37a945 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 24 Jul 2026 22:36:25 -0400 Subject: [PATCH] fix(drain): fail closed on missing or unproven acknowledgement evidence (#661) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 1cbbde00895dce1545290a660bc31c4ac9210b3e: 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) Claude-Session: https://claude.ai/code/session_01VEaP3TohHLFWkp3Z2mmuZw --- drain_proof.py | 137 +++++++++++++++++++++--- tests/test_drain_proof.py | 213 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 334 insertions(+), 16 deletions(-) diff --git a/drain_proof.py b/drain_proof.py index d81c26e..5eec7b6 100644 --- a/drain_proof.py +++ b/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 [] - all_acked = bool(ack_values) and all( - str(v).strip().lower() in {"ack", "acked", "acknowledged"} - for v in ack_values - ) - no_sessions_to_ack = isinstance(acks, Mapping) and len(ack_values) == 0 + # + # Whether acknowledgements are *required* is answered by the impact report, + # never inferred from the shape of the drain state. Previously an absent + # ``acks`` key collapsed to ``{}`` and was read as "no session needed to + # acknowledge", so a proof minted clean — and the restart gate allowed — + # while the report still showed other live sessions. Absence of evidence is + # 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: - if timeout_policy and not all_acked: - detail = "explicit ack timeout policy applied" - elif no_sessions_to_ack: - detail = "no other live sessions required to acknowledge" - else: - detail = "all affected sessions acknowledged" + + # Present-but-unacknowledged entries always fail closed, even when the + # report claims nobody was live: the drain state naming an outstanding + # session contradicts that claim, and the safe reading of a contradiction + # is that an acknowledgement is still owed. + 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 = "outstanding acks with no timeout policy (fail closed)" + 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 = ( + 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 diff --git a/tests/test_drain_proof.py b/tests/test_drain_proof.py index eda7d54..410af33 100644 --- a/tests/test_drain_proof.py +++ b/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()