drain_proof.py

@@ -328,6 +328,160 @@ def _is_acknowledged(value: Any) -> bool:
     return value.strip().lower() in _ACK_TOKENS
 
 
+def _session_ids_from_affected_sessions(
+    raw: Any,
+) -> tuple[set[str] | None, str | None]:
+    """Live, non-requester session ids from the report's ``affected_sessions``.
+
+    The requester is excluded only on authoritative ``is_requester`` evidence;
+    an entry whose ``live`` or ``is_requester`` flag is absent or not a real
+    bool is malformed, never assumed. Returns ``(ids, None)`` or
+    ``(None, detail)``.
+    """
+
+    if not isinstance(raw, (list, tuple)):
+        return None, (
+            "impact report 'affected_sessions' is malformed: expected a list, "
+            f"got {type(raw).__name__} (fail closed)"
+        )
+    collected: list[str] = []
+    for index, entry in enumerate(raw):
+        if not isinstance(entry, Mapping):
+            return None, (
+                f"impact report 'affected_sessions[{index}]' is malformed: "
+                f"expected a mapping, got {type(entry).__name__} (fail closed)"
+            )
+        session_id = entry.get("session_id")
+        if not isinstance(session_id, str) or not session_id.strip():
+            return None, (
+                f"impact report 'affected_sessions[{index}]' has no usable "
+                "session_id (fail closed)"
+            )
+        live = entry.get("live")
+        is_requester = entry.get("is_requester")
+        if not isinstance(live, bool) or not isinstance(is_requester, bool):
+            return None, (
+                f"impact report 'affected_sessions[{index}]' "
+                f"({session_id.strip()}) does not prove live/is_requester with "
+                "explicit booleans (fail closed)"
+            )
+        if live and not is_requester:
+            collected.append(session_id.strip())
+    if len(set(collected)) != len(collected):
+        return None, (
+            "impact report 'affected_sessions' names the same live session more "
+            "than once; required acknowledgement identities are ambiguous "
+            "(fail closed)"
+        )
+    return set(collected), None
+
+
+def _session_ids_from_ack_state(raw: Any) -> tuple[set[str] | None, str | None]:
+    """Session ids keyed by the report's ``ack_state``.
+
+    ``ack_state`` is minted as ``{s.session_id: "pending" for s in
+    other_live_sessions}``, so its keys *are* the required set. Only the keys
+    are trusted here; the per-session value is the report's own placeholder and
+    is never read as an acknowledgement (acknowledgements come from the drain
+    state and must be explicit).
+    """
+
+    if not isinstance(raw, Mapping):
+        return None, (
+            "impact report 'ack_state' is malformed: expected a mapping, got "
+            f"{type(raw).__name__} (fail closed)"
+        )
+    keys: set[str] = set()
+    for key in raw:
+        if not isinstance(key, str) or not key.strip():
+            return None, (
+                "impact report 'ack_state' carries a non-string or empty "
+                "session id (fail closed)"
+            )
+        keys.add(key.strip())
+    if len(keys) != len(raw):
+        return None, (
+            "impact report 'ack_state' names the same live session more than "
+            "once; required acknowledgement identities are ambiguous "
+            "(fail closed)"
+        )
+    return keys, None
+
+
+def _required_ack_sessions(
+    impact_report: Mapping[str, Any],
+) -> tuple[frozenset[str] | None, str | None]:
+    """Identities of the live, non-requester sessions that owe an acknowledgement.
+
+    Derived only from authoritative impact-report evidence, never from the
+    caller-supplied drain state — the same principle checks 1 and 5 already
+    apply. Returns ``(ids, None)`` when the report proves the required set, or
+    ``(None, detail)`` when the evidence is missing, malformed, contradictory,
+    or cannot be reconciled; every one of those fails the checklist closed.
+
+    The report carries two independent views of the same set and both are
+    validated: ``affected_sessions`` filtered on ``live and not is_requester``,
+    and ``ack_state`` whose keys are exactly those sessions. When both are
+    present they must name the same set, and the result is reconciled against
+    ``counts.sessions_live_other`` — a report that disagrees with itself can
+    never authorise a restart.
+    """
+
+    from_sessions: set[str] | None = None
+    if "affected_sessions" in impact_report:
+        from_sessions, detail = _session_ids_from_affected_sessions(
+            impact_report.get("affected_sessions")
+        )
+        if detail is not None:
+            return None, detail
+
+    from_ack_state: set[str] | None = None
+    if "ack_state" in impact_report:
+        from_ack_state, detail = _session_ids_from_ack_state(
+            impact_report.get("ack_state")
+        )
+        if detail is not None:
+            return None, detail
+
+    if from_sessions is None and from_ack_state is None:
+        return None, (
+            "impact report carries no session-identity evidence (neither "
+            "ack_state nor affected_sessions); required acknowledgements "
+            "cannot be attributed to a session (fail closed)"
+        )
+
+    if (
+        from_sessions is not None
+        and from_ack_state is not None
+        and from_sessions != from_ack_state
+    ):
+        only_ack_state = sorted(from_ack_state - from_sessions)
+        only_sessions = sorted(from_sessions - from_ack_state)
+        return None, (
+            "impact report contradicts itself about which sessions must "
+            f"acknowledge: ack_state-only={only_ack_state}, "
+            f"affected_sessions-only={only_sessions} (fail closed)"
+        )
+
+    required = from_ack_state if from_ack_state is not None else from_sessions
+    if required is None:  # unreachable; guarded above, kept fail-closed
+        return None, "required acknowledgement identities unresolved (fail closed)"
+
+    live_count = _live_session_count(impact_report)
+    if live_count is None:
+        return None, (
+            "impact report does not prove the live-session count; required "
+            "acknowledgement identities cannot be reconciled (fail closed)"
+        )
+    if len(required) != live_count:
+        return None, (
+            f"impact report names {len(required)} live session(s) requiring "
+            f"acknowledgement but counts.sessions_live_other={live_count}; "
+            "identity evidence cannot be reconciled (fail closed)"
+        )
+    return frozenset(required), None
+
+
 def _evaluate_checks(
     impact_report: Mapping[str, Any],
     drain_state: Mapping[str, Any],
@@ -444,27 +598,55 @@ def _evaluate_checks(
 
     # 6. Acknowledgements received, or an explicit timeout policy was applied.
     #
-    # Whether acknowledgements are *required* is answered by the impact report,
+    # Whether acknowledgements are *required*, and *which sessions owe them*,
-    # never inferred from the shape of the drain state. Previously an absent
+    # are both answered by the impact report — never inferred from the shape or
-    # ``acks`` key collapsed to ``{}`` and was read as "no session needed to
+    # the cardinality of the drain state. Two fail-opens of the same class were
-    # acknowledge", so a proof minted clean — and the restart gate allowed —
+    # closed here in turn. First, an absent ``acks`` key collapsed to ``{}`` and
-    # while the report still showed other live sessions. Absence of evidence is
+    # was read as "no session needed to acknowledge", so a proof minted clean
-    # not evidence of absence: missing, malformed, stale, or otherwise unproven
+    # while the report still showed other live sessions. Then coverage was
-    # acknowledgement data fails closed whenever live sessions require
+    # decided by comparing an acknowledgement *count* against
-    # acknowledgement, and only explicitly verified acknowledgement evidence
+    # ``sessions_live_other``, so acknowledgements from the requesting session
-    # may permit the operation.
+    # or from ids that do not exist satisfied the obligations of the live
+    # sessions that never answered.
+    #
+    # Coverage is therefore bound to identity: every live, non-requester session
+    # the report names must itself carry an explicit acknowledgement. An
+    # acknowledgement from any other id — the requester, an unknown id, a
+    # fabricated one — is not evidence about a required session and never
+    # increases coverage.
     sessions_live_other = _live_session_count(impact_report)
-    live_count_known = sessions_live_other is not None
+    required_sessions, identity_detail = _required_ack_sessions(impact_report)
-    # An unknown or malformed count fails closed: it cannot prove nobody had to
+    # An unknown count or unresolvable identity evidence fails closed: neither
-    # acknowledge, so acknowledgement stays required.
+    # can prove nobody had to acknowledge, so acknowledgement stays required.
-    acks_required = (not live_count_known) or sessions_live_other > 0
+    acks_required = required_sessions is None or bool(required_sessions)
 
     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
+    # Identity-bound coverage. Only an entry keyed by a required session id and
+    # carrying an explicit token counts, so neither a requester acknowledgement
+    # nor an unknown id can stand in for a session that stayed silent.
+    if acks_present and required_sessions is not None:
+        acked_sessions = {
+            session_id
+            for session_id in required_sessions
+            if any(
+                isinstance(key, str)
+                and key.strip() == session_id
+                and _is_acknowledged(value)
+                for key, value in raw_acks.items()
+            )
+        }
+    else:
+        acked_sessions = set()
+    missing_sessions = (
+        sorted(required_sessions - acked_sessions)
+        if required_sessions is not None
+        else []
+    )
+    covers_live_sessions = required_sessions is not None and not missing_sessions
     # 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.
@@ -476,7 +658,13 @@ def _evaluate_checks(
     # 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):
+    if identity_detail is not None:
+        # Unresolvable identity evidence outranks every permitting path,
+        # including the timeout policy: when the report cannot say who owed an
+        # acknowledgement, nothing can show the obligation was discharged.
+        acks_ok = False
+        detail = identity_detail
+    elif timeout_policy and (acks_required or outstanding_entries):
         acks_ok = True
         detail = "explicit ack timeout policy applied and recorded"
     elif outstanding_entries:
@@ -492,21 +680,16 @@ def _evaluate_checks(
             "impact report proves no other live sessions required to "
             "acknowledge (sessions_live_other=0)"
         )
-    elif every_entry_acked and covers_live_sessions:
+    elif covers_live_sessions:
         acks_ok = True
         detail = (
-            f"{acked_count} acknowledgement entr"
+            f"{len(acked_sessions)} required live session(s) each acknowledged "
-            f"{'y' if acked_count == 1 else 'ies'} verified; covers "
+            f"by identity: {sorted(acked_sessions)}; covers "
             f"sessions_live_other={sessions_live_other}"
         )
     else:
         acks_ok = False
-        if not live_count_known:
+        if raw_acks is None:
-            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 "
@@ -525,9 +708,10 @@ def _evaluate_checks(
             )
         else:
             detail = (
-                f"acknowledgements cover only {acked_count} session(s) but the "
+                "no valid acknowledgement attributed to required live "
-                f"report shows sessions_live_other={sessions_live_other} "
+                f"session(s) {missing_sessions}; the report shows "
-                "(fail closed)"
+                f"sessions_live_other={sessions_live_other} and supplied "
+                "acknowledgements for other ids do not count (fail closed)"
             )
     checks.append(DrainCheck(CHECK_ACKS_OR_TIMEOUT, acks_ok, detail))
 

tests/test_drain_proof.py

@@ -592,5 +592,316 @@ class AcknowledgementFailClosedTests(unittest.TestCase):
         self.assertFalse(result.clean)
 
 
+def _identity_report(*, requester: str, others: tuple[str, ...]) -> dict:
+    """Report with explicitly named requester and other live sessions.
+
+    Unlike :func:`_drained_report_with_live_sessions`, the session ids are
+    chosen by the caller so a test can supply acknowledgements for the *wrong*
+    identities while keeping the count correct.
+    """
+
+    sessions = [
+        {
+            "session_id": requester,
+            "role": "controller",
+            "profile": "prgs-controller",
+            "pid": _live_pid(),
+            "status": "active",
+            "last_heartbeat_at": NOW.isoformat(),
+        }
+    ]
+    for session_id in others:
+        sessions.append(
+            {
+                "session_id": session_id,
+                "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=requester,
+    )
+    return report.as_dict()
+
+
+class AcknowledgementIdentityBindingTests(unittest.TestCase):
+    """Acknowledgement coverage must be bound to session identity, not counted.
+
+    Regression cover for the second reviewed fail-open on PR #882 (review 582,
+    blocker B1): coverage compared ``acked_count`` against
+    ``counts.sessions_live_other``, so acknowledgements supplied for the
+    requesting session and for ids that do not exist satisfied the obligations
+    of the live sessions that never answered. The required identities are
+    carried by the report itself — ``ack_state`` keys and ``affected_sessions``
+    filtered on ``live and not is_requester`` — and only an acknowledgement
+    keyed by one of those ids may count for it.
+    """
+
+    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) -> dp.DrainCheck:
+        """Failure must propagate through the check, the proof, and the gate."""
+
+        proof = self._build(report, state)
+        check = self._acks_check(proof)
+        self.assertFalse(check.passed)
+        self.assertFalse(proof.clean)
+        self.assertIn(dp.CHECK_ACKS_OR_TIMEOUT, proof.failed_checks)
+        decision = dp.gate_apply_restart(
+            proof=proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertEqual(decision.verdict, dp.GATE_DENY)
+        self.assertFalse(decision.allow)
+        return check
+
+    # --- the reviewer's exact reproduction --------------------------------
+
+    def test_requester_plus_unknown_id_cannot_satisfy_two_live_sessions(self):
+        """Review 582 B1 verbatim: requester + a nonexistent session.
+
+        ``sessions_live_other=2`` with ``ack_state`` naming ``other-0`` and
+        ``other-1``; the drain state supplies an acknowledgement from the
+        requesting session itself and from a session that does not exist. The
+        count matches, the identities do not.
+        """
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        self.assertEqual(report["counts"]["sessions_live_other"], 2)
+        self.assertEqual(
+            report["ack_state"], {"other-0": "pending", "other-1": "pending"}
+        )
+        state = self._state(acks={"req": "ack", "totally-bogus-session": "ack"})
+        check = self.assertAcksFailClosed(report, state)
+        self.assertIn("other-0", check.detail)
+        self.assertIn("other-1", check.detail)
+        self.assertIn("fail closed", check.detail)
+
+    # --- wrong / unknown / requester identities ---------------------------
+
+    def test_sufficient_count_of_wrong_ids_fails_closed(self):
+        """Right cardinality, wrong identities: two acks, neither required."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        state = self._state(acks={"ghost-a": "ack", "ghost-b": "ack"})
+        check = self.assertAcksFailClosed(report, state)
+        self.assertIn("do not count", check.detail)
+
+    def test_more_acks_than_required_still_fails_on_wrong_ids(self):
+        """Coverage cannot be bought with volume: five acks, none required."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        state = self._state(acks={f"ghost-{i}": "acknowledged" for i in range(5)})
+        self.assertAcksFailClosed(report, state)
+
+    def test_partial_identity_match_fails_closed(self):
+        """One required id acknowledged, the rest padded with unknown ids."""
+
+        report = _identity_report(
+            requester="req", others=("other-0", "other-1", "other-2")
+        )
+        state = self._state(
+            acks={"other-0": "ack", "ghost-1": "ack", "ghost-2": "ack"}
+        )
+        check = self.assertAcksFailClosed(report, state)
+        self.assertIn("other-1", check.detail)
+        self.assertIn("other-2", check.detail)
+
+    def test_requester_ack_never_satisfies_another_sessions_obligation(self):
+        """The requester is excluded from the required set and stays excluded."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        requester_rows = [s for s in report["affected_sessions"] if s["is_requester"]]
+        self.assertEqual([s["session_id"] for s in requester_rows], ["req"])
+        self.assertNotIn("req", report["ack_state"])
+        check = self.assertAcksFailClosed(report, self._state(acks={"req": "ack"}))
+        self.assertIn("other-0", check.detail)
+
+    def test_fabricated_ids_do_not_count_toward_coverage(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        for bogus in ("", " ", "other-0 extra", "OTHER-0", "other-01", "0"):
+            with self.subTest(bogus=bogus):
+                self.assertAcksFailClosed(report, self._state(acks={bogus: "ack"}))
+
+    # --- per-session state must be explicitly valid ------------------------
+
+    def test_unproven_per_session_states_fail_closed(self):
+        """A required id present but not explicitly acknowledged fails closed."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        for value in ("pending", "stale", "unknown", "", None, True, 1, NOW):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    report,
+                    self._state(acks={"other-0": "ack", "other-1": value}),
+                )
+
+    def test_report_ack_state_placeholder_is_never_read_as_an_ack(self):
+        """``ack_state`` values are the report's own placeholders, not evidence."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        report["ack_state"] = {"other-0": "ack"}
+        self.assertAcksFailClosed(report, self._state())
+
+    # --- missing / malformed / contradictory identity evidence -------------
+
+    def test_missing_identity_evidence_fails_closed(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        report.pop("ack_state", None)
+        report.pop("affected_sessions", None)
+        check = self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+        self.assertIn("no session-identity evidence", check.detail)
+
+    def test_malformed_ack_state_fails_closed(self):
+        for malformed in ([], "other-0", 7, None, ("other-0",)):
+            with self.subTest(malformed=malformed):
+                report = _identity_report(requester="req", others=("other-0",))
+                report["ack_state"] = malformed
+                self.assertAcksFailClosed(
+                    report, self._state(acks={"other-0": "ack"})
+                )
+
+    def test_non_string_ack_state_key_fails_closed(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        report["ack_state"] = {7: "pending"}
+        self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+
+    def test_malformed_affected_sessions_fails_closed(self):
+        for malformed in ("sessions", 7, {"session_id": "other-0"}, [None], [7]):
+            with self.subTest(malformed=malformed):
+                report = _identity_report(requester="req", others=("other-0",))
+                report.pop("ack_state", None)
+                report["affected_sessions"] = malformed
+                self.assertAcksFailClosed(
+                    report, self._state(acks={"other-0": "ack"})
+                )
+
+    def test_affected_sessions_without_explicit_booleans_fails_closed(self):
+        """``live``/``is_requester`` must be real booleans, never inferred."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        report.pop("ack_state", None)
+        for row in report["affected_sessions"]:
+            if row["session_id"] == "other-0":
+                row["is_requester"] = "false"
+        self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+
+    def test_affected_sessions_missing_live_flag_fails_closed(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        report.pop("ack_state", None)
+        for row in report["affected_sessions"]:
+            row.pop("live", None)
+        self.assertAcksFailClosed(report, self._state(acks={"other-0": "ack"}))
+
+    def test_contradictory_ack_state_and_affected_sessions_fails_closed(self):
+        """Both views present and disagreeing is unresolvable, not a tie-break."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        report["ack_state"] = {"other-0": "pending", "other-9": "pending"}
+        check = self.assertAcksFailClosed(
+            report, self._state(acks={"other-0": "ack", "other-9": "ack"})
+        )
+        self.assertIn("contradicts itself", check.detail)
+
+    def test_identity_count_mismatch_fails_closed(self):
+        """Identity evidence that cannot be reconciled with the count denies."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        report["counts"] = dict(report["counts"], sessions_live_other=1)
+        check = self.assertAcksFailClosed(
+            report, self._state(acks={"other-0": "ack", "other-1": "ack"})
+        )
+        self.assertIn("cannot be reconciled", check.detail)
+
+    def test_broken_identity_evidence_outranks_timeout_policy(self):
+        """The sanctioned timeout path cannot paper over an unreadable report."""
+
+        report = _identity_report(requester="req", others=("other-0",))
+        report["ack_state"] = "not-a-mapping"
+        self.assertAcksFailClosed(report, self._state(ack_timeout_policy_applied=True))
+
+    # --- legitimate success is preserved -----------------------------------
+
+    def test_every_required_session_acknowledged_passes(self):
+        report = _identity_report(
+            requester="req", others=("other-0", "other-1", "other-2")
+        )
+        state = self._state(
+            acks={
+                "other-0": "ack",
+                "other-1": "acked",
+                "other-2": "acknowledged",
+            }
+        )
+        proof = self._build(report, state)
+        check = self._acks_check(proof)
+        self.assertTrue(check.passed)
+        self.assertTrue(proof.clean)
+        self.assertEqual(proof.failed_checks, [])
+        self.assertIn("acknowledged by identity", check.detail)
+        decision = dp.gate_apply_restart(
+            proof=proof.as_dict(),
+            now=NOW,
+            secret=SECRET,
+            expected_impact_fingerprint=dp.impact_fingerprint(report),
+        )
+        self.assertEqual(decision.verdict, dp.GATE_ALLOW)
+        self.assertTrue(decision.allow)
+
+    def test_required_session_ack_tolerates_surrounding_whitespace(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        proof = self._build(report, self._state(acks={" other-0 ": " ACK "}))
+        self.assertTrue(self._acks_check(proof).passed)
+        self.assertTrue(proof.clean)
+
+    def test_no_other_live_sessions_still_passes_with_identity_evidence(self):
+        report = _identity_report(requester="req", others=())
+        self.assertEqual(report["counts"]["sessions_live_other"], 0)
+        self.assertEqual(report["ack_state"], {})
+        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)
+
+    def test_explicit_timeout_policy_retains_intended_behavior(self):
+        """Valid, correctly typed timeout evidence still permits the check."""
+
+        report = _identity_report(requester="req", others=("other-0", "other-1"))
+        proof = self._build(report, 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)
+
+    def test_timeout_policy_still_strictly_typed_under_identity_binding(self):
+        report = _identity_report(requester="req", others=("other-0",))
+        for value in (None, "true", "True", 1, [], {}):
+            with self.subTest(value=value):
+                self.assertAcksFailClosed(
+                    report, self._state(ack_timeout_policy_applied=value)
+                )
+
+
 if __name__ == "__main__":
     unittest.main()