diff --git a/agent/deadline.py b/agent/deadline.py index 33453a746c..96355b8934 100644 --- a/agent/deadline.py +++ b/agent/deadline.py @@ -29,14 +29,8 @@ from typing import Any, Awaitable, Callable, Optional, Protocol logger = logging.getLogger(__name__) __all__ = [ - "MAX_SAFE_TIMEOUT_S", - "BoundedResult", - "DeadlineExpired", - "clamp_timeout", - "resolve_timeout", - "run_bounded_async", - "run_bounded_sync", - "kill_process_tree", + "MAX_SAFE_TIMEOUT_S", "BoundedResult", "DeadlineExpired", "clamp_timeout", "resolve_timeout", + "run_bounded_async", "run_bounded_sync", "kill_process_tree", ] # One year: semantically "unbounded" yet far below any platform time_t limit (#83220). @@ -138,12 +132,7 @@ def _lookup_dotted(section: dict, key: str) -> Any: return node -def resolve_timeout( - key: str, - *, - default: Optional[float], - env_var: Optional[str] = None, -) -> Optional[float]: +def resolve_timeout(key: str, *, default: Optional[float], env_var: Optional[str] = None) -> Optional[float]: """Resolve a timeout (seconds): dotted ``timeouts.`` > ``env_var`` > ``default``; the winner goes through :func:`clamp_timeout`, invalid config/env values fall through with a warning.""" raw = _lookup_dotted(_timeouts_section(), key) @@ -201,14 +190,11 @@ async def _run_abandon_cleanup(on_abandon: Callable[[], Awaitable[Any]]) -> None def _dump_blocked_loop_diagnostics(label: str, timeout_s: float) -> None: logger.warning( - "[deadline] %r deadline (%.0fs) expired but the event loop has not " - "processed the expiry after a further %.0fs ā€” the loop thread appears " - "BLOCKED in a synchronous call, which is why no asyncio timeout can " - "fire. Dumping all thread stacks to stderr to identify the blocking " - "frame.", - label, - timeout_s, - _LOOP_BLOCKED_DUMP_GRACE_S, + "[deadline] %r deadline (%.0fs) expired but the event loop has not processed the expiry " + "after a further %.0fs ā€” the loop thread appears BLOCKED in a synchronous call, which is " + "why no asyncio timeout can fire. Dumping all thread stacks to stderr to identify the " + "blocking frame.", + label, timeout_s, _LOOP_BLOCKED_DUMP_GRACE_S, ) try: faulthandler.dump_traceback(all_threads=True) diff --git a/agent/interrupt_control.py b/agent/interrupt_control.py index 7fdce7db24..e6ad163f44 100644 --- a/agent/interrupt_control.py +++ b/agent/interrupt_control.py @@ -81,12 +81,8 @@ class InterruptControlMixin: """interrupt()/hard_interrupt()/clear_interrupt()/steer()/redirect() (see module docstring).""" def interrupt( - self, - message: Optional[str] = None, - *, - hard_cancel: bool = False, - tool_reason: Optional[str] = None, - require_generation: Optional[int] = None, + self, message: Optional[str] = None, *, hard_cancel: bool = False, + tool_reason: Optional[str] = None, require_generation: Optional[int] = None, ) -> bool: """Request the agent to interrupt its current tool-calling loop (call from another thread). @@ -106,8 +102,7 @@ class InterruptControlMixin: # Tool cancellation attribution stays separate from _interrupt_message, which may carry the user's # full next message. tool_interrupt_reason = ( - (tool_reason or "explicit stop requested") - if hard_cancel + (tool_reason or "explicit stop requested") if hard_cancel else ("user sent a new message" if message else "user interrupt") ) @@ -115,22 +110,21 @@ class InterruptControlMixin: self._interrupt_requested = True self._interrupt_message = message self._tool_interrupt_reason = tool_interrupt_reason - if hard_cancel: - _hard_event = getattr(self, "_hard_interrupt_requested", None) - if _hard_event is not None: - _hard_event.set() + _hard_event = getattr(self, "_hard_interrupt_requested", None) if hard_cancel else None + if _hard_event is not None: + _hard_event.set() + + def _fence(): # re-read each time: a finished commit may replace or clear the slot + return vars(self).get("_active_compression_commit_fence") if hard_cancel else None # A hard stop and redirect share one lock so /stop cannot race with an accepted correction and # accidentally turn itself into a retry. The blocking in-flight-commit wait runs BEFORE the atomic # claim edge (redirect lock still held); the destructive pending-commit cancel runs AFTER the claim # survives (#99758 P1). with _ic_lock(self, "_pending_redirect_lock"): - if hard_cancel: - _fence_cancel_before_commit( - vars(self).get("_active_compression_commit_fence"), - when_in_flight=True, - failure_log="Compression hard-cancel fence wait failed", - ) + _fence_cancel_before_commit( + _fence(), when_in_flight=True, failure_log="Compression hard-cancel fence wait failed" + ) if require_generation is None: # No claim to race: publish WITHOUT the liveness lock (bare AIAgent stand-ins in other # suites lack the liveness seam and would AttributeError). @@ -144,12 +138,9 @@ class InterruptControlMixin: return False self._turn_liveness_abort_claim = None _publish_interrupt_state() - if hard_cancel: - _fence_cancel_before_commit( - vars(self).get("_active_compression_commit_fence"), - when_in_flight=False, - failure_log="Compression hard-cancel fence admission failed", - ) + _fence_cancel_before_commit( + _fence(), when_in_flight=False, failure_log="Compression hard-cancel fence admission failed" + ) self._pending_redirect = None # Codex watches a private interrupt event rather than Hermes' per-thread flag. @@ -187,12 +178,7 @@ class InterruptControlMixin: print("\nāš” Interrupt requested" + (f": '{message[:40]}...'" if message and len(message) > 40 else f": '{message}'" if message else "")) return True - def hard_interrupt( - self, - message: Optional[str] = None, - *, - tool_reason: Optional[str] = None, - ) -> None: + def hard_interrupt(self, message: Optional[str] = None, *, tool_reason: Optional[str] = None) -> None: """Explicit stop preserving the ``interrupt()`` ABI (frontends feature-detect this and fall back to legacy ``interrupt()`` for third-party agents). Bypasses dynamic dispatch: legacy subclasses may override interrupt(message=None) without hard_cancel.""" @@ -205,8 +191,7 @@ class InterruptControlMixin: if preserve_redirect and not getattr(self, "_pending_redirect", None): return False self._interrupt_requested = False - self._interrupt_message = None - self._tool_interrupt_reason = None + self._interrupt_message = self._tool_interrupt_reason = None getattr(self, "_hard_interrupt_requested", threading.Event()).clear() if not preserve_redirect: self._pending_redirect = None