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..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 @@ -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 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. 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..1864f56c8 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 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 wake-up near " + f"reviewer-2's vote at ~0.4s, not a wait out to the 2s timeout" + )