From 96548b121e7afac58610ff095b60a4fd9739021b Mon Sep 17 00:00:00 2001 From: SemTiOne Date: Sat, 4 Jul 2026 22:48:51 +0700 Subject: [PATCH 1/2] fix(escalation): wait full quorum timeout instead of resolving on first vote resolve() woke on the first quorum vote's event and ran the quorum check exactly once, falling through to default_action before later votes could arrive -- an M-of-N quorum could never be satisfied once M > 1. wait_for_quorum() now loops: it re-checks quorum after every vote and re-arms the event (inside the same lock approve()/deny() hold before calling set()) so a vote landing in that window is never lost. approve()/deny() previously only fired the wakeup event on the vote that first set req.decision; later votes were still recorded but never woke a blocked resolve(). Left as-is, a partial fix would still land on the correct decision but only after blocking for the entire timeout, defeating the point of the fix. Both now fire on every accepted vote. Fixes #3186 Signed-off-by: SemTiOne --- .../src/agent_os/integrations/escalation.py | 139 ++++++++++++++---- .../agent-os/tests/test_escalation.py | 58 ++++++++ 2 files changed, 171 insertions(+), 26 deletions(-) diff --git a/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py b/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py index 6a5e83627..77cc4b3fa 100644 --- a/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py +++ b/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py @@ -34,6 +34,7 @@ import abc import logging import threading +import time import uuid from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone @@ -173,11 +174,15 @@ def approve(self, request_id: str, approver: str = "") -> bool: if any(a == approver for a, _, _ in req.votes): return False req.votes.append((approver, "ALLOW", datetime.now(timezone.utc))) - if req.decision != EscalationDecision.PENDING: - return True - req.decision = EscalationDecision.ALLOW - req.resolved_by = approver - req.resolved_at = datetime.now(timezone.utc) + if req.decision == EscalationDecision.PENDING: + req.decision = EscalationDecision.ALLOW + req.resolved_by = approver + req.resolved_at = datetime.now(timezone.utc) + # Fire on every accepted vote, not just the one that first sets + # req.decision, wait_for_quorum() (see below) needs a wakeup + # for each subsequent vote too, to re-check quorum promptly + # instead of blocking for the remainder of the timeout even + # after enough votes have already arrived. event = self._events.get(request_id) if event: event.set() @@ -193,11 +198,10 @@ def deny(self, request_id: str, approver: str = "") -> bool: if any(a == approver for a, _, _ in req.votes): return False req.votes.append((approver, "DENY", datetime.now(timezone.utc))) - if req.decision != EscalationDecision.PENDING: - return True - req.decision = EscalationDecision.DENY - req.resolved_by = approver - req.resolved_at = datetime.now(timezone.utc) + if req.decision == EscalationDecision.PENDING: + req.decision = EscalationDecision.DENY + req.resolved_by = approver + req.resolved_at = datetime.now(timezone.utc) event = self._events.get(request_id) if event: event.set() @@ -226,6 +230,57 @@ def wait_for_decision( req = self._requests.get(request_id) return req.decision if req else EscalationDecision.PENDING + def wait_for_quorum( + self, + request_id: str, + timeout: float | None, + is_satisfied: Callable[[EscalationRequest], bool], + ) -> EscalationRequest | None: + """Block until ``is_satisfied(request)`` is true, or until + ``timeout`` seconds have elapsed *in total* since this call started. + + Unlike :meth:`wait_for_decision`, which returns as soon as the + underlying event fires once, this re-checks ``is_satisfied`` after + every vote and keeps waiting for the remaining time budget if it + still isn't met. The per-request ``threading.Event`` is fired by + ``approve``/``deny`` on every accepted vote; re-arming it here with + ``clear()`` -- always while holding ``self._lock``, the same lock + ``approve``/``deny`` hold while mutating the request and before + they call ``set()``, means no vote recorded between our check + and the next ``wait()`` can be missed. + + Returns: + The ``EscalationRequest`` (whatever its latest state is when + ``is_satisfied`` becomes true or time runs out), or ``None`` + if the request is unknown. + """ + event = self._events.get(request_id) + if event is None: + return self._requests.get(request_id) + + deadline = None if timeout is None else time.monotonic() + timeout + while True: + with self._lock: + req = self._requests.get(request_id) + if req is None or is_satisfied(req): + return req + # Not satisfied yet, re-arm so the next approve()/deny() + # wakes us again. Safe: still holding self._lock here, and + # approve()/deny() only call set() after releasing it, so + # any vote racing with this clear() is strictly ordered + # before or after this critical section, never lost. + event.clear() + + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + with self._lock: + return self._requests.get(request_id) + else: + remaining = None + + event.wait(timeout=remaining) + class WebhookApprovalBackend(ApprovalBackend): """Approval backend that sends webhook notifications for escalations. @@ -409,6 +464,23 @@ def escalate( self._on_escalate(request) return request + def _quorum_outcome(self, req: EscalationRequest) -> EscalationDecision | None: + """Evaluate ``req.votes`` against ``self.quorum``. + + Returns ``ALLOW``/``DENY`` once enough votes are in to decide, + or ``None`` if quorum has not been satisfied yet. Single source + of truth for the vote tally. Used both as the wait predicate + and for the final decision, so the two can never disagree. + """ + assert self.quorum is not None + approvals = sum(1 for _, v, _ in req.votes if v == "ALLOW") + denials = sum(1 for _, v, _ in req.votes if v == "DENY") + if denials >= self.quorum.required_denials: + return EscalationDecision.DENY + if approvals >= self.quorum.required_approvals: + return EscalationDecision.ALLOW + return None + def resolve(self, request_id: str) -> EscalationDecision: """Check or wait for a resolution. @@ -416,33 +488,48 @@ def resolve(self, request_id: str) -> EscalationDecision: For other backends, this polls once and returns the current state. When quorum is configured, the decision is evaluated against - quorum thresholds instead of accepting a single vote. + quorum thresholds instead of accepting a single vote. With + ``InMemoryApprovalQueue``, this waits across all votes that + arrive within ``timeout_seconds``, not just the first one, + so an M-of-N quorum gets its full window to collect M votes. + + Note: + Quorum with a non-``InMemoryApprovalQueue`` backend (e.g. + :class:`WebhookApprovalBackend`) still only polls once; it + cannot block on new votes the way the in-memory queue can. + Configure quorum with an in-memory backend, or poll + ``resolve()`` again yourself until enough votes are in. Returns: The final decision. If the timeout expires, applies the ``default_action`` and returns that. """ if isinstance(self.backend, InMemoryApprovalQueue): - decision = self.backend.wait_for_decision( - request_id, timeout=self.timeout_seconds - ) + if self.quorum: + req = self.backend.wait_for_quorum( + request_id, + timeout=self.timeout_seconds, + is_satisfied=lambda r: self._quorum_outcome(r) is not None, + ) + decision = req.decision if req else EscalationDecision.PENDING + else: + req = None + decision = self.backend.wait_for_decision( + request_id, timeout=self.timeout_seconds + ) else: req = self.backend.get_decision(request_id) decision = req.decision if req else EscalationDecision.PENDING - # Quorum evaluation + # Quorum evaluation (also catches the non-blocking backend case + # above, and re-derives the decision from votes either way so a + # raw req.decision set by a single approve()/deny() call never + # leaks through without being checked against quorum). if self.quorum and decision != EscalationDecision.PENDING: - req = self.backend.get_decision(request_id) - if req: - approvals = sum(1 for _, v, _ in req.votes if v == "ALLOW") - denials = sum(1 for _, v, _ in req.votes if v == "DENY") - - if denials >= self.quorum.required_denials: - return EscalationDecision.DENY - if approvals >= self.quorum.required_approvals: - return EscalationDecision.ALLOW - # Not enough votes yet — treat as pending/timeout - decision = EscalationDecision.PENDING + decision = ( + (self._quorum_outcome(req) if req else None) + or EscalationDecision.PENDING + ) if decision == EscalationDecision.PENDING: # Timeout — apply default diff --git a/agent-governance-python/agent-os/tests/test_escalation.py b/agent-governance-python/agent-os/tests/test_escalation.py index dfe62b176..e7114f970 100644 --- a/agent-governance-python/agent-os/tests/test_escalation.py +++ b/agent-governance-python/agent-os/tests/test_escalation.py @@ -350,3 +350,61 @@ def test_empty_approver_rejected_on_deny(self): assert queue.deny(req.request_id, approver=" ") is False retrieved = queue.get_decision(req.request_id) assert len(retrieved.votes) == 0 + + +class TestQuorumWaitsFullTimeout: + """Regression coverage for #3186. + + ``resolve()`` woke as soon as the *first* quorum vote's event fired + and ran the quorum check exactly once, falling through to + ``default_action`` instead of waiting out the rest of + ``timeout_seconds`` for additional votes. (Line numbers in the + original issue report -- escalation.py:176, :424-431 -- refer to + the file as it stood before this fix; see ``resolve()`` and the new + ``wait_for_quorum()`` below for the current locations.) + + ``TestQuorumResolution`` above (landed with #3126's vote-tracking fix) + doesn't exercise this: its quorum-met case uses a single-vote quorum + (``required_approvals=1``) and its timeout case never sends a second + vote at all, so neither ever exercises a second, later vote arriving + within the timeout window. This class adds that missing case. + """ + + def test_resolve_waits_for_a_late_second_vote_instead_of_defaulting_early(self): + queue = InMemoryApprovalQueue() + handler = EscalationHandler( + backend=queue, + timeout_seconds=2, + default_action=DefaultTimeoutAction.DENY, + quorum=QuorumConfig(required_approvals=2, required_denials=1), + ) + request = handler.escalate("agent-1", "deploy", "needs review") + + def cast_votes(): + time.sleep(0.1) + queue.approve(request.request_id, approver="reviewer-1") + time.sleep(0.3) + queue.approve(request.request_id, approver="reviewer-2") + + t = threading.Thread(target=cast_votes) + t.start() + started = time.monotonic() + decision = handler.resolve(request.request_id) + elapsed = time.monotonic() - started + t.join() + # Bug #3186: resolve() woke on reviewer-1's vote at t=0.1s, found + # quorum (2) unmet with only 1 vote, and fell straight through to + # default_action (DENY) instead of waiting out the remaining + # ~1.9s of its 2s timeout for reviewer-2's vote at t=0.4s. + assert decision == EscalationDecision.ALLOW + # A decision-only assertion isn't enough: approve()/deny() only + # fire the wakeup event on the vote that first sets req.decision. + # A resolve()/wait_for_quorum() fix with no matching fix there + # would still land on ALLOW, but only by blocking for the entire + # timeout_seconds and re-checking votes at the deadline, exactly + # the unresponsiveness #3186 exists to eliminate. Asserting on + # elapsed time is what actually catches that regression. + assert elapsed < 1.0, ( + f"resolved in {elapsed:.3f}s -- expected a prompt wakeup near " + f"reviewer-2's vote at ~0.4s, not a wait out to the 2s timeout" + ) From b48874f983a118206e0df0838298d4b34f085ec9 Mon Sep 17 00:00:00 2001 From: SemTiOne Date: Wed, 8 Jul 2026 14:20:18 +0700 Subject: [PATCH 2/2] style(escalation): reword wakeup to wake-up Signed-off-by: SemTiOne --- .../agent-os/src/agent_os/integrations/escalation.py | 2 +- agent-governance-python/agent-os/tests/test_escalation.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py b/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py index 77cc4b3fa..23ca97de6 100644 --- a/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py +++ b/agent-governance-python/agent-os/src/agent_os/integrations/escalation.py @@ -179,7 +179,7 @@ def approve(self, request_id: str, approver: str = "") -> bool: req.resolved_by = approver req.resolved_at = datetime.now(timezone.utc) # Fire on every accepted vote, not just the one that first sets - # req.decision, wait_for_quorum() (see below) needs a wakeup + # req.decision, wait_for_quorum() (see below) needs a wake-up # for each subsequent vote too, to re-check quorum promptly # instead of blocking for the remainder of the timeout even # after enough votes have already arrived. diff --git a/agent-governance-python/agent-os/tests/test_escalation.py b/agent-governance-python/agent-os/tests/test_escalation.py index e7114f970..1864f56c8 100644 --- a/agent-governance-python/agent-os/tests/test_escalation.py +++ b/agent-governance-python/agent-os/tests/test_escalation.py @@ -398,13 +398,13 @@ def cast_votes(): # ~1.9s of its 2s timeout for reviewer-2's vote at t=0.4s. assert decision == EscalationDecision.ALLOW # A decision-only assertion isn't enough: approve()/deny() only - # fire the wakeup event on the vote that first sets req.decision. + # fire the wake-up event on the vote that first sets req.decision. # A resolve()/wait_for_quorum() fix with no matching fix there # would still land on ALLOW, but only by blocking for the entire # timeout_seconds and re-checking votes at the deadline, exactly # the unresponsiveness #3186 exists to eliminate. Asserting on # elapsed time is what actually catches that regression. assert elapsed < 1.0, ( - f"resolved in {elapsed:.3f}s -- expected a prompt wakeup near " + f"resolved in {elapsed:.3f}s -- expected a prompt wake-up near " f"reviewer-2's vote at ~0.4s, not a wait out to the 2s timeout" )