From 083f8a60711d17eb713b7d3189dea249322f914c Mon Sep 17 00:00:00 2001 From: kshitij <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:33:22 +0530 Subject: [PATCH 1/3] =?UTF-8?q?feat(agent):=20unified=20deadline=20layer?= =?UTF-8?q?=20=E2=80=94=20bounded=20execution=20primitive=20+=20timeout=20?= =?UTF-8?q?resolver=20(#85125=20Phase=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One shared foundation for the timeout/hang backlog instead of per-incident site-local fixes: - agent/deadline.py: run_bounded_async (thread-timer deadline that survives a blocked event loop, generalizing the telegram adapter primitive), run_bounded_sync, clamp_timeout (kills the #83220 time_t OverflowError class at the boundary), resolve_timeout (config.yaml timeouts: section > legacy env bridge > default), kill_process_tree (whole-tree termination for the #71148 orphan class), DeadlineExpired (our deadline, mechanically distinct from provider timeouts). - tool_executor._resolve_concurrent_tool_timeout migrates onto the resolver; exact legacy env-var contract preserved (default 420, 0 disables). - timeouts: accepted as a known config root; documented in cli-config.yaml.example. Pure addition otherwise — no behavior change, no new env vars, no cache impact. Later phases (#85125) migrate tool-execution, MCP, and subprocess call sites onto these primitives. --- agent/deadline.py | 471 +++++++++++++++++++++++++++++++++++ agent/tool_executor.py | 28 +-- cli-config.yaml.example | 16 ++ hermes_cli/config.py | 1 + tests/agent/test_deadline.py | 401 +++++++++++++++++++++++++++++ 5 files changed, 902 insertions(+), 15 deletions(-) create mode 100644 agent/deadline.py create mode 100644 tests/agent/test_deadline.py diff --git a/agent/deadline.py b/agent/deadline.py new file mode 100644 index 000000000000..c7aa5d28cafc --- /dev/null +++ b/agent/deadline.py @@ -0,0 +1,471 @@ +"""Unified deadline layer — one bounded-execution primitive, one timeout resolver. + +Phase 1 of the architectural fix for the timeout/hang backlog +(https://github.com/NousResearch/hermes-agent/issues/85125). + +The tree currently carries at least six site-local deadline mechanisms, each +built for one incident, none shared (tool_executor batch deadline, telegram +``_await_with_thread_deadline``, gateway turn lease, reasoning stale floors, +``human_wait_ceiling``, per-MCP-handler timeouts). Every new stall report +grows that list by one. This module is the shared foundation the call sites +migrate onto in later phases: + +* :func:`resolve_timeout` — one config-first resolution path for timeout + values (``timeouts:`` section in config.yaml > legacy env var > default), + so new surfaces stop inventing ``HERMES_*_TIMEOUT`` env vars (".env is for + secrets only") and hardcoded literals stop ignoring user config + (#63302, #53161, #43272 class). + +* :func:`clamp_timeout` — platform-safe clamping. Large user-supplied + timeouts overflow ``time_t`` inside ``threading.Lock.acquire(timeout=...)`` + / ``Thread.join(timeout=...)`` on macOS and kill whole tool batches + (#83220). Clamping at the shared boundary fixes that class once, for + every consumer. + +* :func:`run_bounded_async` — a wall-clock deadline for awaitables that does + NOT depend on event-loop timers. ``asyncio.wait_for`` schedules its expiry + on the loop; when the loop thread itself is blocked in a synchronous call + (family A of the #84047 stall triage), every asyncio-based timeout in the + process is silently disabled. This helper drives the deadline from a + daemon ``threading.Timer`` (generalizing the proven telegram-adapter + primitive) and abandons cancellation-shielded tasks instead of waiting for + cancellation to complete. + +* :func:`run_bounded_sync` — the same contract for synchronous callables + bounded from a synchronous context (daemon worker thread, abandoned on + expiry). + +* :func:`kill_process_tree` — portable whole-tree termination so + kill-on-timeout stops orphaning descendants (#71148, #59549, #84967, + #68139 class). + +Design invariants: + +* Exceptions raised by the bounded operation propagate unchanged — callers + keep their existing error handling. Only the *timeout* outcome is + reified (as :class:`BoundedResult`), because that is the outcome the + call sites keep getting wrong. +* A timeout produced by this layer is OUR deadline, not the provider's. + Callers that feed errors into ``agent/error_classifier.py`` should + classify :class:`DeadlineExpired` distinctly from transport timeouts + (the #59549 / #80323 misattribution class). +* ``None`` timeout means unbounded, and non-positive resolved values are + normalized to ``None`` (matching the existing + ``HERMES_CONCURRENT_TOOL_TIMEOUT_S`` convention). +""" + +from __future__ import annotations + +import asyncio +import faulthandler +import logging +import os +import subprocess +import sys +import threading +import time +from dataclasses import dataclass +from typing import Any, Awaitable, Callable, Optional + +logger = logging.getLogger(__name__) + +__all__ = [ + "MAX_SAFE_TIMEOUT_S", + "BoundedResult", + "DeadlineExpired", + "clamp_timeout", + "resolve_timeout", + "run_bounded_async", + "run_bounded_sync", + "kill_process_tree", +] + +# Upper bound for any timeout handed to platform wait primitives. +# +# CPython converts ``threading.Lock.acquire(timeout=...)`` / +# ``Thread.join(timeout=...)`` deadlines to an absolute timestamp; very large +# relative timeouts overflow ``time_t`` on macOS and raise +# ``OverflowError: timestamp out of range for platform time_t`` (#83220). +# One year is semantically "unbounded" for every wait in this codebase while +# staying far below any platform conversion limit. +MAX_SAFE_TIMEOUT_S = 31_536_000.0 # 365 days + +# Grace period after a deadline fires before concluding the event loop thread +# is blocked in a synchronous call and dumping stacks (family A diagnostics). +_LOOP_BLOCKED_DUMP_GRACE_S = 5.0 + + +class DeadlineExpired(TimeoutError): + """A deadline enforced by this layer expired. + + Distinct from transport/provider timeout types on purpose: when this is + raised (or a :class:`BoundedResult` reports ``timed_out``), the timeout + was Hermes's own bound — error classification must not attribute it to + the provider (#59549 / #80323 misattribution class). + """ + + def __init__(self, label: str, timeout_s: float): + super().__init__(f"deadline expired after {timeout_s:.1f}s: {label}") + self.label = label + self.timeout_s = timeout_s + + +@dataclass(frozen=True) +class BoundedResult: + """Outcome of a bounded operation. + + ``timed_out`` is the reified outcome; on completion ``value`` holds the + operation's return value. Operation exceptions are never captured here — + they propagate to the caller unchanged. + """ + + timed_out: bool + value: Any + elapsed_s: float + timeout_s: Optional[float] + label: str + + def raise_if_timed_out(self) -> Any: + """Return ``value``, raising :class:`DeadlineExpired` on timeout.""" + if self.timed_out: + raise DeadlineExpired(self.label, float(self.timeout_s or 0.0)) + return self.value + + +def clamp_timeout(timeout: Optional[float]) -> Optional[float]: + """Normalize a timeout value for platform wait primitives. + + * ``None`` stays ``None`` (unbounded). + * Non-positive values become ``None`` (unbounded) — matching the existing + ``HERMES_CONCURRENT_TOOL_TIMEOUT_S`` "0 disables the bound" convention. + * Values above :data:`MAX_SAFE_TIMEOUT_S` are capped so they can never + overflow ``time_t`` inside ``Lock.acquire`` / ``Thread.join`` on macOS + (#83220). + * Non-numeric values are treated as unset (``None``) with a warning + rather than crashing the call path they were meant to protect. + """ + if timeout is None: + return None + try: + value = float(timeout) + except (TypeError, ValueError): + logger.warning("clamp_timeout: non-numeric timeout %r; treating as unbounded", timeout) + return None + if value != value: # NaN + logger.warning("clamp_timeout: NaN timeout; treating as unbounded") + return None + if value <= 0: + return None + return min(value, MAX_SAFE_TIMEOUT_S) + + +# --------------------------------------------------------------------------- +# Timeout resolution: config.yaml ``timeouts:`` section > legacy env var > +# registered default. +# --------------------------------------------------------------------------- + +def _timeouts_section() -> dict: + """Read the ``timeouts:`` root section from config.yaml (read-only). + + Isolated for testability and so a broken config read can never take down + the call path the timeout was protecting. + """ + try: + from hermes_cli.config import load_config_readonly + + section = load_config_readonly().get("timeouts") + return section if isinstance(section, dict) else {} + except Exception: + logger.debug("timeouts: config read failed; using defaults", exc_info=True) + return {} + + +def _lookup_dotted(section: dict, key: str) -> Any: + """Walk ``a.b.c`` through nested dicts; return None when absent.""" + node: Any = section + for part in key.split("."): + if not isinstance(node, dict) or part not in node: + return None + node = node[part] + return node + + +def resolve_timeout( + key: str, + *, + default: Optional[float], + env_var: Optional[str] = None, +) -> Optional[float]: + """Resolve a timeout in seconds for a dotted config key. + + Precedence (established by the ``providers.*.request_timeout_seconds`` + pattern — config wins over the legacy env var): + + 1. ``timeouts.`` in config.yaml (dotted key walks nested maps, e.g. + ``tools.concurrent_batch`` reads ``timeouts: {tools: {concurrent_batch: ...}}``) + 2. ``env_var`` when set and non-empty (legacy bridge — internal mechanism + and back-compat only; new surfaces must not grow new user-facing + ``HERMES_*`` timeout env vars) + 3. ``default`` + + The winning value is passed through :func:`clamp_timeout`, so ``0`` or a + negative value means "unbounded" and oversized values are made + platform-safe. Invalid (non-numeric) config/env values fall through to + the next source with a warning instead of breaking the protected path. + """ + raw = _lookup_dotted(_timeouts_section(), key) + if raw is not None: + try: + return clamp_timeout(float(raw)) + except (TypeError, ValueError): + logger.warning("timeouts.%s: invalid value %r in config.yaml; ignoring", key, raw) + + if env_var: + env_raw = os.getenv(env_var, "").strip() + if env_raw: + try: + return clamp_timeout(float(env_raw)) + except ValueError: + logger.warning("invalid %s=%r; ignoring", env_var, env_raw) + + return clamp_timeout(default) + + +# --------------------------------------------------------------------------- +# Bounded execution — async flavor. +# +# Generalizes plugins/platforms/telegram/adapter.py:_await_with_thread_deadline +# (the #63309 fix): the deadline is driven by a daemon threading.Timer so a +# blocked event loop cannot disable it, and a second timer dumps all thread +# stacks when the loop provably failed to process the expiry — the one piece +# of information loop-blocked hangs otherwise never surface. +# --------------------------------------------------------------------------- + +def _consume_abandoned(task: "asyncio.Future[Any]") -> None: + """Observe an abandoned task's outcome so it never logs 'never retrieved'.""" + try: + if not task.cancelled(): + task.exception() + except Exception: + pass + + +async def _run_abandon_cleanup(on_abandon: Callable[[], Awaitable[Any]]) -> None: + """Run abandonment cleanup fully fire-and-forget (its failures swallowed).""" + try: + await on_abandon() + except Exception: + logger.debug("deadline abandon-cleanup failed", exc_info=True) + + +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, + ) + try: + faulthandler.dump_traceback(all_threads=True) + except Exception: + logger.debug("faulthandler traceback dump failed", exc_info=True) + + +async def run_bounded_async( + awaitable: Awaitable[Any], + timeout: Optional[float], + *, + label: str = "operation", + on_abandon: Optional[Callable[[], Awaitable[Any]]] = None, + dump_on_blocked_loop: bool = True, +) -> BoundedResult: + """Await ``awaitable`` under a wall-clock deadline independent of loop timers. + + On completion returns ``BoundedResult(timed_out=False, value=...)``; + exceptions from the operation (including ``asyncio.CancelledError`` from a + caller cancelling *us*) propagate unchanged. + + On timeout the underlying task is cancelled and **abandoned** — we do not + await cancellation completion, because cancellation-shielded scopes (anyio, + httpcore init, MCP SDK teardown) are exactly the paths that wedge forever. + ``on_abandon`` (zero-arg callable returning an awaitable) is scheduled as + detached best-effort cleanup for the half-built state the abandoned task + may leave behind. Returns ``BoundedResult(timed_out=True, value=None)``. + + ``timeout=None`` (or a non-positive resolved value) awaits unbounded. + """ + timeout_s = clamp_timeout(timeout) + start = time.monotonic() + if timeout_s is None: + value = await awaitable + return BoundedResult(False, value, time.monotonic() - start, None, label) + + task = asyncio.ensure_future(awaitable) + loop = asyncio.get_running_loop() + deadline: "asyncio.Future[None]" = loop.create_future() + loop_processed_expiry = threading.Event() + + def _mark_expired() -> None: + loop_processed_expiry.set() + if not deadline.done(): + deadline.set_result(None) + + def _expire_from_thread() -> None: + loop.call_soon_threadsafe(_mark_expired) + + def _watchdog_check() -> None: + if not loop_processed_expiry.is_set(): + _dump_blocked_loop_diagnostics(label, timeout_s) + + timer = threading.Timer(timeout_s, _expire_from_thread) + timer.daemon = True + timer.start() + watchdog: Optional[threading.Timer] = None + if dump_on_blocked_loop: + watchdog = threading.Timer( + timeout_s + _LOOP_BLOCKED_DUMP_GRACE_S, _watchdog_check + ) + watchdog.daemon = True + watchdog.start() + try: + done, _ = await asyncio.wait( + {task, deadline}, return_when=asyncio.FIRST_COMPLETED + ) + if task in done: + if not deadline.done(): + deadline.cancel() + value = await task + return BoundedResult(False, value, time.monotonic() - start, timeout_s, label) + + task.cancel() + task.add_done_callback(_consume_abandoned) + if on_abandon is not None: + cleanup = asyncio.ensure_future(_run_abandon_cleanup(on_abandon)) + cleanup.add_done_callback(_consume_abandoned) + logger.warning("[deadline] %r timed out after %.1fs; task abandoned", label, timeout_s) + return BoundedResult(True, None, time.monotonic() - start, timeout_s, label) + finally: + timer.cancel() + if watchdog is not None: + watchdog.cancel() + # cancel() cannot stop a Timer whose callback is already running; + # setting the event closes that race so a completed await can never + # be misreported as a blocked loop. + loop_processed_expiry.set() + + +# --------------------------------------------------------------------------- +# Bounded execution — sync flavor. +# --------------------------------------------------------------------------- + +def run_bounded_sync( + fn: Callable[[], Any], + timeout: Optional[float], + *, + label: str = "operation", + on_timeout: Optional[Callable[[], None]] = None, +) -> BoundedResult: + """Run ``fn`` in a daemon worker thread under a wall-clock deadline. + + On completion returns its value (exceptions re-raised in the caller). + On expiry the worker thread is **abandoned** (daemon, so it cannot block + interpreter exit), ``on_timeout`` (if given) runs best-effort in the + caller's thread — e.g. to mark a backend suspect or kill a subprocess — + and ``BoundedResult(timed_out=True)`` is returned. + + ``timeout=None`` (or non-positive) blocks until ``fn`` returns. + """ + timeout_s = clamp_timeout(timeout) + start = time.monotonic() + if timeout_s is None: + return BoundedResult(False, fn(), time.monotonic() - start, None, label) + + box: dict[str, Any] = {} + done = threading.Event() + + def _worker() -> None: + try: + box["value"] = fn() + except BaseException as exc: # re-raised in caller; must not vanish + box["exc"] = exc + finally: + done.set() + + thread = threading.Thread( + target=_worker, name=f"deadline-{label}", daemon=True + ) + thread.start() + if not done.wait(timeout_s): + logger.warning("[deadline] %r timed out after %.1fs; worker abandoned", label, timeout_s) + if on_timeout is not None: + try: + on_timeout() + except Exception: + logger.debug("deadline on_timeout callback failed", exc_info=True) + return BoundedResult(True, None, time.monotonic() - start, timeout_s, label) + + if "exc" in box: + raise box["exc"] + return BoundedResult(False, box.get("value"), time.monotonic() - start, timeout_s, label) + + +# --------------------------------------------------------------------------- +# Whole-tree process termination. +# --------------------------------------------------------------------------- + +def kill_process_tree(pid: int, *, sig: Optional[int] = None) -> bool: + """Terminate ``pid`` and all its descendants, portably. + + Kill-on-timeout that signals only the direct child orphans process trees + (cron scripts, in-container shells, browser daemons — #71148 class). + + * POSIX: signals the process group when ``pid`` leads one (callers that + spawn with ``start_new_session=True`` / ``preexec_fn=os.setsid`` get + full-tree kill), falling back to the single process otherwise. + ``sig`` defaults to ``SIGKILL``. + * Windows: ``taskkill /F /T`` terminates the tree without requiring + psutil. ``sig`` is ignored (Windows has no equivalent). + + Returns True when a termination call was issued without error, False when + the process was already gone or the call failed (callers treat both as + "nothing more we can do"). + """ + if sys.platform == "win32": + try: + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(pid)], + capture_output=True, + timeout=15, + check=False, + ) + return True + except Exception: + logger.debug("kill_process_tree: taskkill failed for pid %s", pid, exc_info=True) + return False + + import signal as _signal + + if sig is None: + sig = _signal.SIGKILL + try: + pgid = os.getpgid(pid) + except (ProcessLookupError, PermissionError, OSError): + pgid = None + try: + if pgid is not None and pgid == pid: + # pid leads its own group: kill the whole tree in one syscall. + os.killpg(pgid, sig) + else: + # Not a group leader (killing its group would hit our own group + # or an unrelated one) — signal the single process. + os.kill(pid, sig) + return True + except ProcessLookupError: + return False + except (PermissionError, OSError): + logger.debug("kill_process_tree: signal failed for pid %s", pid, exc_info=True) + return False diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 1438cd08fcd2..87ee404a5b73 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -158,21 +158,19 @@ def _parse_tool_arguments(raw_arguments: Any) -> tuple[dict, Optional[str]]: def _resolve_concurrent_tool_timeout() -> float | None: - raw = os.getenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "").strip() - if not raw: - return _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S - try: - value = float(raw) - except ValueError: - logger.warning( - "invalid HERMES_CONCURRENT_TOOL_TIMEOUT_S=%r; using %.0fs", - raw, - _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S, - ) - return _DEFAULT_CONCURRENT_TOOL_TIMEOUT_S - if value <= 0: - return None - return value + """Resolve the per-batch concurrent tool deadline. + + Delegates to the unified resolver (#85125): ``timeouts.tools.concurrent_batch`` + in config.yaml wins, the legacy ``HERMES_CONCURRENT_TOOL_TIMEOUT_S`` env var + remains the back-compat bridge, and ``0``/negative still disables the bound. + """ + from agent.deadline import resolve_timeout + + return resolve_timeout( + "tools.concurrent_batch", + default=_DEFAULT_CONCURRENT_TOOL_TIMEOUT_S, + env_var="HERMES_CONCURRENT_TOOL_TIMEOUT_S", + ) def _flush_session_db_after_tool_progress( diff --git a/cli-config.yaml.example b/cli-config.yaml.example index fb36868ce837..344648e49f76 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -166,6 +166,22 @@ model: # gpt-5.4: # stale_timeout_seconds: 1800 # Longer non-stream stale timeout for slow large-context turns +# ============================================================================= +# Unified Timeouts (operation deadlines) +# ============================================================================= +# One place to override Hermes's internal operation deadlines (seconds). +# Keys are dotted paths resolved by agent/deadline.py:resolve_timeout(). +# Precedence: this section > legacy HERMES_* env var (back-compat) > built-in +# default. 0 or a negative value disables the bound (unbounded); very large +# values are clamped to a platform-safe maximum automatically. +# +# Currently resolved keys (more paths migrate here over time — see issue #85125): +# +# timeouts: +# tools: +# concurrent_batch: 420 # Deadline for a parallel tool-call batch +# # (legacy env: HERMES_CONCURRENT_TOOL_TIMEOUT_S) + # ============================================================================= # OpenRouter Provider Routing (only applies when using OpenRouter) # ============================================================================= diff --git a/hermes_cli/config.py b/hermes_cli/config.py index c863e214f265..1cc807492f12 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1878,6 +1878,7 @@ def check_config_version() -> Tuple[int, int]: "require_mention", # top-level convenience form honored by the gateway (#3979) "unauthorized_dm_behavior", # top-level form read by gateway/config.py "signal", # Signal settings bridged to env vars by gateway/config.py + "timeouts", # unified timeout resolution section (agent/deadline.py, #85125) } _KNOWN_ROOT_KEYS = frozenset(DEFAULT_CONFIG.keys()) | _EXTRA_KNOWN_ROOT_KEYS diff --git a/tests/agent/test_deadline.py b/tests/agent/test_deadline.py new file mode 100644 index 000000000000..c31dfb59dca9 --- /dev/null +++ b/tests/agent/test_deadline.py @@ -0,0 +1,401 @@ +"""Tests for agent/deadline.py — the unified deadline layer (#85125). + +Covers: +* clamp_timeout normalization (None / non-positive / oversized / NaN / junk) +* resolve_timeout precedence: config.yaml ``timeouts:`` > legacy env var > default +* run_bounded_sync: completion, exception propagation, timeout + on_timeout +* run_bounded_async: completion, exception propagation, timeout + abandonment + of cancellation-shielded tasks, on_abandon cleanup +* kill_process_tree: descendants of a session-leader child die with it (POSIX) +* backward-compat contract of tool_executor._resolve_concurrent_tool_timeout + after its migration onto resolve_timeout +""" + +from __future__ import annotations + +import asyncio +import os +import signal +import subprocess +import sys +import threading +import time + +import pytest + +from agent.deadline import ( + MAX_SAFE_TIMEOUT_S, + BoundedResult, + DeadlineExpired, + clamp_timeout, + kill_process_tree, + resolve_timeout, + run_bounded_async, + run_bounded_sync, +) + + +# --------------------------------------------------------------------------- +# clamp_timeout +# --------------------------------------------------------------------------- + +class TestClampTimeout: + def test_none_stays_none(self): + assert clamp_timeout(None) is None + + def test_zero_and_negative_mean_unbounded(self): + assert clamp_timeout(0) is None + assert clamp_timeout(-5) is None + + def test_normal_value_passes_through(self): + assert clamp_timeout(420.0) == 420.0 + + def test_oversized_value_clamped_to_platform_safe_max(self): + # The #83220 class: >time_t deadlines crash Lock.acquire on macOS. + assert clamp_timeout(10**18) == MAX_SAFE_TIMEOUT_S + + def test_clamped_value_safe_for_threading_primitives(self): + # Regression proof for #83220: the clamped value must be accepted by + # the exact primitives that used to overflow. + big = clamp_timeout(float(10**15)) + assert big is not None + lock = threading.Lock() + assert lock.acquire(timeout=min(big, 0.001)) + lock.release() + + def test_nan_and_junk_treated_as_unbounded(self): + assert clamp_timeout(float("nan")) is None + assert clamp_timeout("not-a-number") is None # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# resolve_timeout +# --------------------------------------------------------------------------- + +class TestResolveTimeout: + def test_default_wins_when_nothing_configured(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.delenv("HERMES_TEST_DEADLINE_X", raising=False) + assert resolve_timeout("a.b", default=42.0, env_var="HERMES_TEST_DEADLINE_X") == 42.0 + + def test_env_var_beats_default(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.setenv("HERMES_TEST_DEADLINE_X", "17.5") + assert resolve_timeout("a.b", default=42.0, env_var="HERMES_TEST_DEADLINE_X") == 17.5 + + def test_config_beats_env_var(self, monkeypatch): + monkeypatch.setattr( + "agent.deadline._timeouts_section", lambda: {"a": {"b": 99}} + ) + monkeypatch.setenv("HERMES_TEST_DEADLINE_X", "17.5") + assert resolve_timeout("a.b", default=42.0, env_var="HERMES_TEST_DEADLINE_X") == 99.0 + + def test_dotted_key_walks_nested_maps(self, monkeypatch): + monkeypatch.setattr( + "agent.deadline._timeouts_section", + lambda: {"tools": {"concurrent_batch": 300}}, + ) + assert resolve_timeout("tools.concurrent_batch", default=420.0) == 300.0 + + def test_zero_config_value_means_unbounded(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {"a": {"b": 0}}) + assert resolve_timeout("a.b", default=42.0) is None + + def test_invalid_config_value_falls_through_to_env(self, monkeypatch): + monkeypatch.setattr( + "agent.deadline._timeouts_section", lambda: {"a": {"b": "soon"}} + ) + monkeypatch.setenv("HERMES_TEST_DEADLINE_X", "17.5") + assert resolve_timeout("a.b", default=42.0, env_var="HERMES_TEST_DEADLINE_X") == 17.5 + + def test_invalid_env_value_falls_through_to_default(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.setenv("HERMES_TEST_DEADLINE_X", "banana") + assert resolve_timeout("a.b", default=42.0, env_var="HERMES_TEST_DEADLINE_X") == 42.0 + + def test_broken_config_read_never_breaks_the_protected_path(self, monkeypatch): + # _timeouts_section swallows config-load failures internally; prove + # the public contract by making the underlying loader raise. + import agent.deadline as dl + + def _boom(): + raise RuntimeError("config unreadable") + + monkeypatch.setattr("hermes_cli.config.load_config_readonly", _boom) + assert dl._timeouts_section() == {} + assert resolve_timeout("a.b", default=5.0) == 5.0 + + +# --------------------------------------------------------------------------- +# run_bounded_sync +# --------------------------------------------------------------------------- + +class TestRunBoundedSync: + def test_completion_returns_value(self): + result = run_bounded_sync(lambda: "ok", 5.0, label="t") + assert result.timed_out is False + assert result.value == "ok" + assert result.raise_if_timed_out() == "ok" + + def test_unbounded_when_timeout_none(self): + result = run_bounded_sync(lambda: 7, None, label="t") + assert result.timed_out is False and result.value == 7 + + def test_exception_propagates_unchanged(self): + class Boom(RuntimeError): + pass + + with pytest.raises(Boom): + run_bounded_sync(lambda: (_ for _ in ()).throw(Boom("x")), 5.0, label="t") + + def test_timeout_abandons_worker_and_reports(self): + release = threading.Event() + + def _wedged(): + release.wait(30) + return "late" + + start = time.monotonic() + result = run_bounded_sync(_wedged, 0.2, label="wedged") + elapsed = time.monotonic() - start + assert result.timed_out is True + assert result.value is None + assert elapsed < 5.0 # returned near the deadline, not after 30s + with pytest.raises(DeadlineExpired) as exc_info: + result.raise_if_timed_out() + assert "wedged" in str(exc_info.value) + release.set() + + def test_on_timeout_callback_runs(self): + release = threading.Event() + fired = [] + result = run_bounded_sync( + lambda: release.wait(30), + 0.1, + label="t", + on_timeout=lambda: fired.append(True), + ) + assert result.timed_out and fired == [True] + release.set() + + def test_on_timeout_callback_failure_is_swallowed(self): + release = threading.Event() + result = run_bounded_sync( + lambda: release.wait(30), + 0.1, + label="t", + on_timeout=lambda: (_ for _ in ()).throw(RuntimeError("cleanup boom")), + ) + assert result.timed_out is True + release.set() + + def test_deadline_expired_is_a_timeout_error(self): + # Error-classification contract: our deadline must be catchable as + # TimeoutError but distinguishable by type from transport timeouts. + assert issubclass(DeadlineExpired, TimeoutError) + + +# --------------------------------------------------------------------------- +# run_bounded_async +# --------------------------------------------------------------------------- + +class TestRunBoundedAsync: + def test_completion_returns_value(self): + async def scenario(): + async def op(): + return "ok" + + return await run_bounded_async(op(), 5.0, label="t") + + result = asyncio.run(scenario()) + assert result.timed_out is False and result.value == "ok" + + def test_unbounded_when_timeout_none(self): + async def scenario(): + async def op(): + return 7 + + return await run_bounded_async(op(), None, label="t") + + result = asyncio.run(scenario()) + assert result.timed_out is False and result.value == 7 + + def test_exception_propagates_unchanged(self): + class Boom(RuntimeError): + pass + + async def scenario(): + async def op(): + raise Boom("x") + + await run_bounded_async(op(), 5.0, label="t") + + with pytest.raises(Boom): + asyncio.run(scenario()) + + def test_timeout_returns_promptly(self): + async def scenario(): + async def op(): + await asyncio.sleep(30) + + start = time.monotonic() + result = await run_bounded_async(op(), 0.2, label="slow") + return result, time.monotonic() - start + + result, elapsed = asyncio.run(scenario()) + assert result.timed_out is True + assert elapsed < 5.0 + + def test_timeout_abandons_cancellation_shielded_task(self): + """The family-A killer case: asyncio.wait_for cannot expire a shielded + scope; the thread-timer deadline must return anyway.""" + + async def scenario(): + hung = asyncio.Event() + + async def inner(): + await hung.wait() + + async def shielded(): + # Shield swallows the cancellation run_bounded_async issues. + await asyncio.shield(asyncio.ensure_future(inner())) + + start = time.monotonic() + result = await run_bounded_async(shielded(), 0.2, label="shielded") + elapsed = time.monotonic() - start + hung.set() # release the orphan so the loop can drain + await asyncio.sleep(0) + return result, elapsed + + result, elapsed = asyncio.run(scenario()) + assert result.timed_out is True + assert elapsed < 5.0 + + def test_on_abandon_cleanup_runs_detached(self): + async def scenario(): + cleaned = asyncio.Event() + + async def _cleanup(): + cleaned.set() + + async def op(): + await asyncio.sleep(30) + + result = await run_bounded_async( + op(), 0.1, label="t", on_abandon=_cleanup + ) + await asyncio.wait_for(cleaned.wait(), timeout=5.0) + return result + + result = asyncio.run(scenario()) + assert result.timed_out is True + + def test_completed_op_never_reports_timeout(self): + # Race guard: completion just under the deadline must report success. + async def scenario(): + async def op(): + await asyncio.sleep(0.01) + return "made it" + + return await run_bounded_async(op(), 5.0, label="t") + + result = asyncio.run(scenario()) + assert result.timed_out is False and result.value == "made it" + + +# --------------------------------------------------------------------------- +# kill_process_tree +# --------------------------------------------------------------------------- + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process-group semantics") +class TestKillProcessTree: + def test_kills_descendants_of_session_leader(self, tmp_path): + """A child spawned with start_new_session must die with its own child. + + This is the orphan-tree class (#71148): killing only the direct child + leaves grandchildren running. + """ + started = tmp_path / "grandchild_started" + marker = tmp_path / "grandchild_alive" + grandchild_py = tmp_path / "grandchild.py" + grandchild_py.write_text( + "import pathlib, time\n" + f"pathlib.Path({str(started)!r}).write_text('x')\n" + "time.sleep(10)\n" + f"pathlib.Path({str(marker)!r}).write_text('x')\n" + ) + parent_py = tmp_path / "parent.py" + parent_py.write_text( + "import subprocess, sys, time\n" + f"subprocess.Popen([sys.executable, {str(grandchild_py)!r}])\n" + "time.sleep(10)\n" + ) + proc = subprocess.Popen( + [sys.executable, str(parent_py)], start_new_session=True + ) + deadline = time.monotonic() + 10 + while not started.exists() and time.monotonic() < deadline: + time.sleep(0.05) + assert started.exists(), "grandchild never spawned — test harness broken" + assert kill_process_tree(proc.pid) is True + proc.wait(timeout=5) + # Grandchild must be dead too: marker never appears. + time.sleep(1.5) + assert not marker.exists() + + def test_already_dead_pid_returns_false(self): + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait(timeout=10) + assert kill_process_tree(proc.pid) in (False, True) # reaped or zombie-signalable + + def test_non_group_leader_falls_back_to_single_kill(self): + # Child in OUR process group: killpg would signal the test runner. + proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + try: + assert os.getpgid(proc.pid) != proc.pid # not a leader + assert kill_process_tree(proc.pid, sig=signal.SIGTERM) is True + proc.wait(timeout=5) + finally: + if proc.poll() is None: + proc.kill() + + +# --------------------------------------------------------------------------- +# tool_executor migration contract +# --------------------------------------------------------------------------- + +class TestConcurrentToolTimeoutMigration: + """_resolve_concurrent_tool_timeout keeps its exact legacy contract.""" + + def _resolver(self): + from agent import tool_executor + + return tool_executor._resolve_concurrent_tool_timeout + + def test_default_unchanged(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.delenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", raising=False) + assert self._resolver()() == 420.0 + + def test_env_var_still_works(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "60") + assert self._resolver()() == 60.0 + + def test_env_zero_still_disables(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "0") + assert self._resolver()() is None + + def test_env_invalid_still_falls_back_to_default(self, monkeypatch): + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {}) + monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "junk") + assert self._resolver()() == 420.0 + + def test_new_config_key_wins(self, monkeypatch): + monkeypatch.setattr( + "agent.deadline._timeouts_section", + lambda: {"tools": {"concurrent_batch": 300}}, + ) + monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "60") + assert self._resolver()() == 300.0 From 8ac9ff18aecddab97a6ae9722086b2fa8fe99211 Mon Sep 17 00:00:00 2001 From: kshitij <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:46:52 +0530 Subject: [PATCH 2/3] fix(agent): harden deadline layer per self-review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - run_bounded_async: cancel + abandon the inner task when the CALLER is cancelled (leak the telegram original also had) - kill_process_tree: check taskkill exit code (Windows contract parity), suppress console flash via windows_hide_flags, and sweep a psutil descendant snapshot taken before signalling — reaches grandchildren in their own setsid sessions and the non-group-leader case (#71148 class) - resolve_timeout: reject bool (YAML true would become a 1s deadline) and NaN config values with fall-through instead of resolving unbounded - BoundedResult: kw_only to prevent positional transposition - tests: real clamped-value time_t regression proof, own-session descendant kill, external-cancellation task cleanup, bool/NaN config fall-through; pin already-dead-pid contract --- agent/deadline.py | 141 ++++++++++++++++++++++++++--------- tests/agent/test_deadline.py | 89 ++++++++++++++++++++-- 2 files changed, 190 insertions(+), 40 deletions(-) diff --git a/agent/deadline.py b/agent/deadline.py index c7aa5d28cafc..5df4e869f058 100644 --- a/agent/deadline.py +++ b/agent/deadline.py @@ -29,7 +29,10 @@ process is silently disabled. This helper drives the deadline from a daemon ``threading.Timer`` (generalizing the proven telegram-adapter primitive) and abandons cancellation-shielded tasks instead of waiting for - cancellation to complete. + cancellation to complete. The telegram adapter's private copy + (``plugins/platforms/telegram/adapter.py:_await_with_thread_deadline``) + migrates onto this in Phase 2 of #85125 — do not let the two drift in the + meantime; fix bugs here first. * :func:`run_bounded_sync` — the same contract for synchronous callables bounded from a synchronous context (daemon worker thread, abandoned on @@ -37,7 +40,10 @@ * :func:`kill_process_tree` — portable whole-tree termination so kill-on-timeout stops orphaning descendants (#71148, #59549, #84967, - #68139 class). + #68139 class). Existing site-local tree-kills that migrate onto this in + Phase 4 of #85125: ``gateway/status.py`` (taskkill wrapper + psutil + snapshot/reap pair) and ``tools/code_execution_tool.py`` (psutil + recursive children kill). Design invariants: @@ -110,7 +116,7 @@ def __init__(self, label: str, timeout_s: float): self.timeout_s = timeout_s -@dataclass(frozen=True) +@dataclass(frozen=True, kw_only=True) class BoundedResult: """Outcome of a bounded operation. @@ -215,10 +221,19 @@ def resolve_timeout( """ raw = _lookup_dotted(_timeouts_section(), key) if raw is not None: - try: - return clamp_timeout(float(raw)) - except (TypeError, ValueError): - logger.warning("timeouts.%s: invalid value %r in config.yaml; ignoring", key, raw) + # Explicit float() (clamp_timeout would also convert) so that invalid + # config values FALL THROUGH to the env var / default instead of + # resolving as unbounded — do not "simplify" this away. bool is + # rejected because YAML `true` would silently become a 1-second + # deadline; NaN is rejected for the same fall-through reason. + if not isinstance(raw, bool): + try: + value = float(raw) + if value == value: # not NaN + return clamp_timeout(value) + except (TypeError, ValueError): + pass + logger.warning("timeouts.%s: invalid value %r in config.yaml; ignoring", key, raw) if env_var: env_raw = os.getenv(env_var, "").strip() @@ -302,7 +317,7 @@ async def run_bounded_async( start = time.monotonic() if timeout_s is None: value = await awaitable - return BoundedResult(False, value, time.monotonic() - start, None, label) + return BoundedResult(timed_out=False, value=value, elapsed_s=time.monotonic() - start, timeout_s=None, label=label) task = asyncio.ensure_future(awaitable) loop = asyncio.get_running_loop() @@ -332,14 +347,23 @@ def _watchdog_check() -> None: watchdog.daemon = True watchdog.start() try: - done, _ = await asyncio.wait( - {task, deadline}, return_when=asyncio.FIRST_COMPLETED - ) + try: + done, _ = await asyncio.wait( + {task, deadline}, return_when=asyncio.FIRST_COMPLETED + ) + except asyncio.CancelledError: + # The CALLER cancelled us. Without this, `task` would keep running + # unobserved (and later log "exception was never retrieved") — + # a leak the telegram original also had. Cancel + abandon it, then + # let the cancellation propagate. + task.cancel() + task.add_done_callback(_consume_abandoned) + raise if task in done: if not deadline.done(): deadline.cancel() value = await task - return BoundedResult(False, value, time.monotonic() - start, timeout_s, label) + return BoundedResult(timed_out=False, value=value, elapsed_s=time.monotonic() - start, timeout_s=timeout_s, label=label) task.cancel() task.add_done_callback(_consume_abandoned) @@ -347,7 +371,7 @@ def _watchdog_check() -> None: cleanup = asyncio.ensure_future(_run_abandon_cleanup(on_abandon)) cleanup.add_done_callback(_consume_abandoned) logger.warning("[deadline] %r timed out after %.1fs; task abandoned", label, timeout_s) - return BoundedResult(True, None, time.monotonic() - start, timeout_s, label) + return BoundedResult(timed_out=True, value=None, elapsed_s=time.monotonic() - start, timeout_s=timeout_s, label=label) finally: timer.cancel() if watchdog is not None: @@ -377,12 +401,17 @@ def run_bounded_sync( caller's thread — e.g. to mark a backend suspect or kill a subprocess — and ``BoundedResult(timed_out=True)`` is returned. + Intended for infrequent, seconds-scale blocking backend calls. Do NOT + use per-item in hot loops: each call spawns a thread, and every timeout + permanently leaks an abandoned daemon thread — a wedged backend called + in a retry loop would accumulate them. + ``timeout=None`` (or non-positive) blocks until ``fn`` returns. """ timeout_s = clamp_timeout(timeout) start = time.monotonic() if timeout_s is None: - return BoundedResult(False, fn(), time.monotonic() - start, None, label) + return BoundedResult(timed_out=False, value=fn(), elapsed_s=time.monotonic() - start, timeout_s=None, label=label) box: dict[str, Any] = {} done = threading.Event() @@ -406,11 +435,11 @@ def _worker() -> None: on_timeout() except Exception: logger.debug("deadline on_timeout callback failed", exc_info=True) - return BoundedResult(True, None, time.monotonic() - start, timeout_s, label) + return BoundedResult(timed_out=True, value=None, elapsed_s=time.monotonic() - start, timeout_s=timeout_s, label=label) if "exc" in box: raise box["exc"] - return BoundedResult(False, box.get("value"), time.monotonic() - start, timeout_s, label) + return BoundedResult(timed_out=False, value=box.get("value"), elapsed_s=time.monotonic() - start, timeout_s=timeout_s, label=label) # --------------------------------------------------------------------------- @@ -423,26 +452,43 @@ def kill_process_tree(pid: int, *, sig: Optional[int] = None) -> bool: Kill-on-timeout that signals only the direct child orphans process trees (cron scripts, in-container shells, browser daemons — #71148 class). - * POSIX: signals the process group when ``pid`` leads one (callers that - spawn with ``start_new_session=True`` / ``preexec_fn=os.setsid`` get - full-tree kill), falling back to the single process otherwise. - ``sig`` defaults to ``SIGKILL``. - * Windows: ``taskkill /F /T`` terminates the tree without requiring - psutil. ``sig`` is ignored (Windows has no equivalent). - - Returns True when a termination call was issued without error, False when - the process was already gone or the call failed (callers treat both as - "nothing more we can do"). + * Windows: ``taskkill /F /T`` terminates the tree (``sig`` ignored; + Windows has no equivalent). Console-window flash is suppressed via + ``windows_hide_flags`` and the exit code is checked, so a dead or + inaccessible PID reports ``False`` like the POSIX path. + * POSIX: the descendant set is snapshotted via psutil (a hard + dependency) BEFORE any signal — once the parent dies its children are + reparented and can no longer be found by a parent walk. Then the + process group is signalled when ``pid`` leads one (covers + grandchildren in the same session in one syscall), and every + snapshotted descendant is signalled individually — which also reaches + descendants that created their OWN sessions (a child that called + ``setsid``, exactly what user shell commands do; see + tools/environments/base.py). ``sig`` defaults to ``SIGKILL``. + psutil's identity-aware ``Process`` (PID + create time) means a + recycled PID is never signalled. + + Returns True when the target (or any of its tree) was signalled, False + when the process was already gone or every termination call failed. """ if sys.platform == "win32": try: - subprocess.run( + from hermes_cli._subprocess_compat import windows_hide_flags + + creationflags = windows_hide_flags() + except Exception: + creationflags = 0 + try: + proc = subprocess.run( ["taskkill", "/F", "/T", "/PID", str(pid)], capture_output=True, timeout=15, check=False, + creationflags=creationflags, ) - return True + # taskkill exits non-zero for not-found / access-denied; keep the + # cross-platform contract (False = nothing was terminated). + return proc.returncode == 0 except Exception: logger.debug("kill_process_tree: taskkill failed for pid %s", pid, exc_info=True) return False @@ -451,21 +497,48 @@ def kill_process_tree(pid: int, *, sig: Optional[int] = None) -> bool: if sig is None: sig = _signal.SIGKILL + + # Snapshot descendants while the parent is still alive — after it dies + # they reparent to init/subreaper and a parent walk finds nothing. + descendants: list = [] + try: + import psutil + + descendants = psutil.Process(int(pid)).children(recursive=True) + except Exception: + # Already gone, or psutil unavailable in a stripped env — the + # group-signal below still covers same-session descendants. + descendants = [] + + signalled = False try: + # NOTE: getpgid→killpg has an inherent TOCTOU (pid could be reaped and + # recycled between the calls). All existing killpg sites share it; the + # psutil sweep below is identity-aware and does not. pgid = os.getpgid(pid) except (ProcessLookupError, PermissionError, OSError): pgid = None try: if pgid is not None and pgid == pid: - # pid leads its own group: kill the whole tree in one syscall. + # pid leads its own group: one syscall covers the whole group. + # (The == check guards against signalling the caller's own group + # when pid is not a leader.) os.killpg(pgid, sig) else: - # Not a group leader (killing its group would hit our own group - # or an unrelated one) — signal the single process. os.kill(pid, sig) - return True + signalled = True except ProcessLookupError: - return False + pass except (PermissionError, OSError): logger.debug("kill_process_tree: signal failed for pid %s", pid, exc_info=True) - return False + + # Sweep the snapshot: reaches descendants outside the parent's group + # (their own setsid sessions) and the non-group-leader case. + for child in descendants: + try: + if child.is_running(): # identity-aware: recycled PIDs skipped + child.send_signal(sig) + signalled = True + except Exception: + continue + return signalled diff --git a/tests/agent/test_deadline.py b/tests/agent/test_deadline.py index c31dfb59dca9..9e5238b9afc8 100644 --- a/tests/agent/test_deadline.py +++ b/tests/agent/test_deadline.py @@ -55,12 +55,15 @@ def test_oversized_value_clamped_to_platform_safe_max(self): assert clamp_timeout(10**18) == MAX_SAFE_TIMEOUT_S def test_clamped_value_safe_for_threading_primitives(self): - # Regression proof for #83220: the clamped value must be accepted by - # the exact primitives that used to overflow. + # Regression proof for #83220: the clamped value itself must be + # accepted by the exact primitive that used to overflow. Acquiring an + # uncontended lock returns immediately regardless of timeout, so + # passing the full clamped value is safe and actually exercises the + # time_t conversion. big = clamp_timeout(float(10**15)) assert big is not None lock = threading.Lock() - assert lock.acquire(timeout=min(big, 0.001)) + assert lock.acquire(timeout=big) lock.release() def test_nan_and_junk_treated_as_unbounded(self): @@ -113,6 +116,18 @@ def test_invalid_env_value_falls_through_to_default(self, monkeypatch): monkeypatch.setenv("HERMES_TEST_DEADLINE_X", "banana") assert resolve_timeout("a.b", default=42.0, env_var="HERMES_TEST_DEADLINE_X") == 42.0 + def test_bool_config_value_rejected(self, monkeypatch): + # YAML `true` must not silently become a 1-second deadline. + monkeypatch.setattr("agent.deadline._timeouts_section", lambda: {"a": {"b": True}}) + assert resolve_timeout("a.b", default=42.0) == 42.0 + + def test_nan_config_value_falls_through(self, monkeypatch): + # NaN must fall through to the next source, not resolve as unbounded. + monkeypatch.setattr( + "agent.deadline._timeouts_section", lambda: {"a": {"b": float("nan")}} + ) + assert resolve_timeout("a.b", default=42.0) == 42.0 + def test_broken_config_read_never_breaks_the_protected_path(self, monkeypatch): # _timeouts_section swallows config-load failures internally; prove # the public contract by making the underlying loader raise. @@ -302,6 +317,33 @@ async def op(): result = asyncio.run(scenario()) assert result.timed_out is False and result.value == "made it" + def test_external_cancellation_cancels_inner_task(self): + # If the CALLER cancels run_bounded_async, the inner task must not be + # leaked running unobserved. + async def scenario(): + started = asyncio.Event() + inner_cancelled = asyncio.Event() + + async def op(): + started.set() + try: + await asyncio.sleep(30) + except asyncio.CancelledError: + inner_cancelled.set() + raise + + outer = asyncio.ensure_future( + run_bounded_async(op(), 25.0, label="t") + ) + await started.wait() + outer.cancel() + with pytest.raises(asyncio.CancelledError): + await outer + await asyncio.wait_for(inner_cancelled.wait(), timeout=5.0) + return True + + assert asyncio.run(scenario()) is True + # --------------------------------------------------------------------------- # kill_process_tree @@ -343,10 +385,45 @@ def test_kills_descendants_of_session_leader(self, tmp_path): time.sleep(1.5) assert not marker.exists() + def test_kills_descendant_in_its_own_session(self, tmp_path): + """A descendant that setsid'd out of the parent's group must die too. + + killpg on the parent's group cannot reach it; the psutil descendant + sweep must (tools/environments/base.py documents user commands doing + exactly this). + """ + started = tmp_path / "setsid_grandchild_started" + marker = tmp_path / "setsid_grandchild_alive" + grandchild_py = tmp_path / "grandchild.py" + grandchild_py.write_text( + "import pathlib, time\n" + f"pathlib.Path({str(started)!r}).write_text('x')\n" + "time.sleep(10)\n" + f"pathlib.Path({str(marker)!r}).write_text('x')\n" + ) + parent_py = tmp_path / "parent.py" + parent_py.write_text( + "import subprocess, sys, time\n" + # grandchild leaves the parent's session/group entirely + f"subprocess.Popen([sys.executable, {str(grandchild_py)!r}], start_new_session=True)\n" + "time.sleep(10)\n" + ) + proc = subprocess.Popen( + [sys.executable, str(parent_py)], start_new_session=True + ) + deadline = time.monotonic() + 10 + while not started.exists() and time.monotonic() < deadline: + time.sleep(0.05) + assert started.exists(), "grandchild never spawned — test harness broken" + assert kill_process_tree(proc.pid) is True + proc.wait(timeout=5) + time.sleep(1.5) + assert not marker.exists() + def test_already_dead_pid_returns_false(self): - proc = subprocess.Popen([sys.executable, "-c", "pass"]) - proc.wait(timeout=10) - assert kill_process_tree(proc.pid) in (False, True) # reaped or zombie-signalable + proc = subprocess.Popen([sys.executable, "-c", "pass"], start_new_session=True) + proc.wait(timeout=10) # reaped: PID is gone from the process table + assert kill_process_tree(proc.pid) is False def test_non_group_leader_falls_back_to_single_kill(self): # Child in OUR process group: killpg would signal the test runner. From 8b387962ac3230ee1cf0a8040dae239eb0c0a247 Mon Sep 17 00:00:00 2001 From: kshitij <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 13 Aug 2026 13:53:39 +0530 Subject: [PATCH 3/3] fix(agent): suppress windows-footgun lint on POSIX-only killpg branch The os.killpg call sits below an early 'if sys.platform == win32: return' so it can never execute on Windows; the scanner is line-based and needs the inline marker. --- agent/deadline.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/agent/deadline.py b/agent/deadline.py index 5df4e869f058..5aa58e6c06e9 100644 --- a/agent/deadline.py +++ b/agent/deadline.py @@ -523,7 +523,7 @@ def kill_process_tree(pid: int, *, sig: Optional[int] = None) -> bool: # pid leads its own group: one syscall covers the whole group. # (The == check guards against signalling the caller's own group # when pid is not a leader.) - os.killpg(pgid, sig) + os.killpg(pgid, sig) # windows-footgun: ok — POSIX-only branch (win32 returns above) else: os.kill(pid, sig) signalled = True