From d60e01dcb3b7aaeaa7e511fc8bbcb797db2eb9cf Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:12:22 -0700 Subject: [PATCH] refactor(approval): unify gateway post-hook/decision shape, compact human-wait eviction (222+158->194+152) --- tools/approval_gateway_wait.py | 104 ++++++++++++--------------------- tools/approval_human_wait.py | 32 +++++----- 2 files changed, 51 insertions(+), 85 deletions(-) diff --git a/tools/approval_gateway_wait.py b/tools/approval_gateway_wait.py index c40fcc8245..eb75982203 100644 --- a/tools/approval_gateway_wait.py +++ b/tools/approval_gateway_wait.py @@ -15,7 +15,6 @@ import logging import threading import time import uuid -from typing import Optional from tools.interrupt import is_interrupted @@ -31,27 +30,10 @@ class _ApprovalEntry: self.data = dict(data) self.data.setdefault("request_id", uuid.uuid4().hex) self.acknowledged = False - self.result: Optional[str] = None # "once"|"session"|"always"|"deny" + self.result: str | None = None # "once"|"session"|"always"|"deny" # Free-text reason from ``/deny `` so the agent can adapt # instead of only hearing "denied". - self.reason: Optional[str] = None - - -def _hook_payload(approval_data: dict, session_key: str, surface: str) -> dict: - primary_key = approval_data.get("pattern_key", "") - return { - "command": approval_data.get("command", ""), - "description": approval_data.get("description", ""), - "pattern_key": primary_key, - "pattern_keys": list(approval_data.get("pattern_keys", [primary_key])), - "session_key": session_key, - "surface": surface, - } - - -def _hook_outcome(resolved: bool, choice: Optional[str]) -> str: - """Unresolved (timeout) and a None choice both mean the user never answered.""" - return "timeout" if not resolved else (choice or "timeout") + self.reason: str | None = None def _poll_event(event: threading.Event, session_key: str, *, interrupt_log: str) -> str: @@ -61,14 +43,14 @@ def _poll_event(event: threading.Event, session_key: str, *, interrupt_log: str) so activity heartbeats reach the agent's inactivity tracker every ~10s — otherwise the gateway watchdog kills the agent while the user is still responding (mirrors ``_wait_for_process()`` cadence). The loop is recorded - as human-wait time so the concurrent batch deadline excludes it (#79719). + as human-wait time so the concurrent batch deadline excludes it. ``is_interrupted()`` deliberately does NOT distinguish a deliberate /stop from a gateway inactivity timeout — both resolve as 'deny' (not outcome='timeout'). The per-thread interrupt flag carries no stable - machine-checkable cause, so a fail-closed deny preserves #8697 semantics; - changing this needs a dedicated interrupt-cause channel, not string - matching (#85125). + machine-checkable cause, so a fail-closed deny preserves the historical + semantics; changing this needs a dedicated interrupt-cause channel, not + string matching. """ from tools.approval import _get_approval_timeout, human_wait_window @@ -94,8 +76,18 @@ def _poll_event(event: threading.Event, session_key: str, *, interrupt_log: str) touch_activity_if_due(activity_state, "waiting for user approval") -def _await_coalesced_leader(session_key: str, leader, approval_data: dict, - *, surface: str = "gateway"): +def _finish(payload: dict, resolved: bool, choice: str | None, reason, **extra) -> dict: + """Fire the post hook and build the decision dict. Unresolved (timeout) and + a None choice both mean the user never answered.""" + from tools.approval import _fire_approval_hook + _fire_approval_hook( + "post_approval_response", **payload, + choice="timeout" if not resolved else (choice or "timeout"), **extra, + ) + return {"resolved": resolved, "choice": choice, "reason": reason, **extra} + + +def _await_coalesced_leader(session_key: str, leader, payload: dict): """Wait on an already-pending identical approval instead of re-prompting. Adopts the leader's decision: ``session``/``always`` → approval (same dict @@ -107,10 +99,7 @@ def _await_coalesced_leader(session_key: str, leader, approval_data: dict, so observers see the follower's lifecycle without a duplicate prompt. """ from tools.approval import _fire_approval_hook - - payload = _hook_payload(approval_data, session_key, surface) _fire_approval_hook("pre_approval_request", **payload, coalesced=True) - state = _poll_event( leader.event, session_key, interrupt_log="Coalesced approval wait interrupted by user signal — " @@ -124,20 +113,10 @@ def _await_coalesced_leader(session_key: str, leader, approval_data: dict, else: choice = leader.result resolved = choice is not None - if choice == "once": # The post hook fires for the fresh prompt's own lifecycle, not here. return None - _fire_approval_hook( - "post_approval_response", **payload, - choice=_hook_outcome(resolved, choice), coalesced=True, - ) - return { - "resolved": resolved, - "choice": choice, - "reason": getattr(leader, "reason", None), - "coalesced": True, - } + return _finish(payload, resolved, choice, getattr(leader, "reason", None), coalesced=True) def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, @@ -158,23 +137,25 @@ def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, """ from tools import approval as _approval - payload = _hook_payload(approval_data, session_key, surface) - - leader = None + primary_key = approval_data.get("pattern_key", "") + payload = { + "command": approval_data.get("command", ""), + "description": approval_data.get("description", ""), + "pattern_key": primary_key, + "pattern_keys": list(approval_data.get("pattern_keys", [primary_key])), + "session_key": session_key, + "surface": surface, + } + keys = list(approval_data.get("pattern_keys") or []) with _approval._lock: - for existing in _approval._gateway_queues.get(session_key, []): - data = existing.data - if ( - data.get("command") == approval_data.get("command") - and list(data.get("pattern_keys") or []) - == list(approval_data.get("pattern_keys") or []) - ): - leader = existing - break - if leader is not None: - adopted = _await_coalesced_leader( - session_key, leader, approval_data, surface=surface + leader = next( + (e for e in _approval._gateway_queues.get(session_key, []) + if e.data.get("command") == approval_data.get("command") + and list(e.data.get("pattern_keys") or []) == keys), + None, ) + if leader is not None: + adopted = _await_coalesced_leader(session_key, leader, payload) if adopted is not None: return adopted @@ -192,16 +173,13 @@ def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, # Plugins hear about the request before the gateway does (real-time observers). _approval._fire_approval_hook("pre_approval_request", **payload) - # Bridges sync agent thread → async gateway. try: notify_cb(dict(entry.data)) except Exception as exc: logger.warning("Gateway approval notify failed: %s", exc) _drop_entry() - _approval._fire_approval_hook( - "post_approval_response", **payload, choice="notify_failed" - ) + _approval._fire_approval_hook("post_approval_response", **payload, choice="notify_failed") return {"resolved": False, "choice": None, "notify_failed": True} state = _poll_event( @@ -212,11 +190,5 @@ def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict, if state == "interrupted": entry.result = "deny" entry.event.set() - resolved = state != "timeout" _drop_entry() - - choice = entry.result - _approval._fire_approval_hook( - "post_approval_response", **payload, choice=_hook_outcome(resolved, choice) - ) - return {"resolved": resolved, "choice": choice, "reason": entry.reason} + return _finish(payload, state != "timeout", entry.result, entry.reason) diff --git a/tools/approval_human_wait.py b/tools/approval_human_wait.py index db1969e0f9..24d2ca53e2 100644 --- a/tools/approval_human_wait.py +++ b/tools/approval_human_wait.py @@ -6,7 +6,7 @@ deadline in agent/tool_executor.py excludes this time so a slow human answer never times a batch out — but ONLY this time. Measuring at the source (rather than residency in the authorization gate, which is arbitrary code) is what keeps a wedged pre_tool_call plugin or a dead approval client from growing the -exclusion 1:1 with wall clock and defeating the deadline entirely (#79719). +exclusion 1:1 with wall clock and defeating the deadline entirely. Keyed by session so one gateway session's pending approval cannot extend a different session's batch deadline. State is process-global like the rest of @@ -48,19 +48,16 @@ def human_wait_ceiling() -> float: drift. Never call while holding ``_human_wait_lock`` — it reads the config cache. ``_get_approval_timeout`` caps at ``agent.deadline.MAX_SAFE_TIMEOUT_S`` so the value is always safe for - ``Lock.acquire(timeout=...)`` / ``Thread.join(timeout=...)`` (#83220). + ``Lock.acquire(timeout=...)`` / ``Thread.join(timeout=...)``. """ from tools.approval import _get_approval_timeout - return float(_get_approval_timeout()) + HUMAN_WAIT_MARGIN_S def _clamped_window_seconds(started: float, now: float, ceiling: float) -> float: - """Seconds an open window contributes: elapsed, floored at 0, capped. - - Shared by the close-time accrual and the open-window read so the two - clamps stay identical by construction. - """ + """Seconds an open window contributes: elapsed, floored at 0, capped. Shared + by the close-time accrual and the open-window read so the two clamps stay + identical by construction.""" return min(max(0.0, now - started), ceiling) @@ -75,14 +72,12 @@ def _human_wait_state(session_key: str) -> _HumanWaitState: """ state = _human_wait_states.get(session_key) if state is None: - if len(_human_wait_states) >= _HUMAN_WAIT_MAX_SESSIONS: - for key in list(_human_wait_states): - if len(_human_wait_states) < _HUMAN_WAIT_MAX_SESSIONS: - break - if _human_wait_states[key].pending == 0: - del _human_wait_states[key] - state = _HumanWaitState() - _human_wait_states[session_key] = state + for key in list(_human_wait_states): + if len(_human_wait_states) < _HUMAN_WAIT_MAX_SESSIONS: + break + if _human_wait_states[key].pending == 0: + del _human_wait_states[key] + state = _human_wait_states[session_key] = _HumanWaitState() return state @@ -90,7 +85,6 @@ def _resolve_key(session_key: str | None) -> str: if session_key is not None: return session_key from tools.approval import get_current_session_key - return get_current_session_key() @@ -101,7 +95,7 @@ def human_wait_window(session_key: str | None = None): Wrap ONLY code that is genuinely parked waiting for a user's answer (the CLI approval prompt, the gateway approval poll loop). The concurrent tool batch deadline excludes this time; wrapping anything else re-creates the - #79719 hang where arbitrary wedged code pushes the deadline out forever. + hang where arbitrary wedged code pushes the deadline out forever. Overlapping windows for the same session coalesce (pending counter), so two serialized approval prompts don't double-count the same wall clock. @@ -141,7 +135,7 @@ def human_wait_seconds(session_key: str | None = None) -> float: baseline delta to zero (the safe direction: the deadline fires sooner). Deadline consumers snapshot a baseline at batch start and use the delta. Each window's contribution is clamped to :func:`human_wait_ceiling` - (belt-and-braces for #79719). + (belt-and-braces against the wedged-window hang). """ key = _resolve_key(session_key) now = time.monotonic()