From 962c38fe9ae0748799fa34ae745c249858e5f077 Mon Sep 17 00:00:00 2001 From: Aaron Abu Usama <50079365+AaronAbuUsama@users.noreply.github.com> Date: Wed, 19 Aug 2026 17:00:15 +0000 Subject: [PATCH 1/2] fix(sdk): bound AsyncExecutor.close() so it cannot hang forever close() passed no exception to the portal context manager, so anyio took its graceful path -- portal.stop(cancel_remaining=False) -- and waited for in-flight tasks to finish on their own. It then joined the portal thread with no timeout. Either half can block the caller indefinitely. That matters because LocalConversation.close() releases tool executors in an unbounded loop, so one stuck portal task wedges conversation shutdown and every later operation that needs the conversation lock. Cancel remaining tasks on shutdown, and bound the wait for the portal thread. The portal thread is a daemon, so abandoning it with a warning is safe when it is stuck on work that ignores cancellation. Closes #4546 Co-authored-by: openhands --- .../openhands/sdk/utils/async_executor.py | 116 +++++++++++++++++- tests/sdk/utils/test_async_executor.py | 71 +++++++++++ 2 files changed, 183 insertions(+), 4 deletions(-) create mode 100644 tests/sdk/utils/test_async_executor.py diff --git a/openhands-sdk/openhands/sdk/utils/async_executor.py b/openhands-sdk/openhands/sdk/utils/async_executor.py index 611f511734..29087a17e1 100644 --- a/openhands-sdk/openhands/sdk/utils/async_executor.py +++ b/openhands-sdk/openhands/sdk/utils/async_executor.py @@ -13,6 +13,19 @@ logger = get_logger(__name__) +# Upper bound on how long close() waits for the portal thread to wind down. +# The portal thread is a daemon thread, so abandoning it is safe; blocking the +# caller forever is not. +# +# The default is intentionally short for lifecycle teardown: a successful +# portal stop + join normally completes in milliseconds once remaining tasks +# are cancelled, so 10s is already a generous margin for a healthy shutdown. +# It is deliberately *not* the 30s inherited from BrowserToolExecutor cleanup +# — that value predates the lifecycle-lock path and is too long for teardown +# that may hold a conversation lock while waiting. See PR #4548 and the +# discussion of #4598 in its description. +DEFAULT_CLOSE_TIMEOUT = 10.0 + class AsyncExecutor: """ @@ -101,17 +114,112 @@ async def _execute(): return portal.call(_execute) - def close(self): + def close(self, timeout: float | None = DEFAULT_CLOSE_TIMEOUT): + """Shut down the portal, without ever blocking the caller forever. + + Semantics + --------- + This is **bounded, best-effort shutdown, not guaranteed cleanup.** + + - It first cancels any remaining portal tasks (``portal.stop(True)``) + and then waits up to ``timeout`` seconds for the portal thread to + exit. + - If the portal thread is stuck on work that does *not* honour + cancellation — for example a task awaiting inside a worker thread, + which anyio cannot interrupt until the thread returns (see #4598) — + the wait expires and the helper thread is **abandoned**. The portal + thread is a daemon, so the process can still exit; but the thread + and any resources it holds (subprocess handles, sockets, file + descriptors) may remain alive until process exit. This is a + deliberate trade-off: blocking the caller forever is worse. + - The shutdown path is non-raising. Failures inside the portal + teardown are logged (with traceback) and swallowed so that + ``close()`` is always safe to call from a destructor or + ``__exit__``. + - **Idempotent.** Calling ``close()`` on an already-closed executor + is a no-op. + + Args: + timeout: seconds to wait for the portal thread to exit. ``None`` + waits indefinitely (the previous, pre-#4548 behaviour) and + is kept only for compatibility — **do not use it on any + production path**, since it reintroduces the unbounded hang + this fix exists to prevent. See PR #4548 / issue #4598. + """ with self._lock: portal_cm = self._portal_cm + portal = self._portal self._portal_cm = None self._portal = None - if portal_cm is not None: + if portal_cm is None: + return + + # Stamp the owner into the thread name so an abandoned thread is + # identifiable in py-spy / py-dump traces (the portal thread itself + # does not carry the executor identity). + owner = type(self).__qualname__ + thread_name = f"{owner}-close" + + def _shutdown() -> None: + try: + # Cancel whatever is still running. Without this, anyio's + # graceful path (portal.stop(cancel_remaining=False)) waits + # for in-flight tasks that may never complete on their own. + if portal is not None: + portal.call(portal.stop, True) + except RuntimeError: + pass # portal already stopped + except Exception: + # Teardown must stay non-raising: close() is called from + # __del__/__exit__/atexit. Log with traceback so the failure + # remains diagnosable instead of being reduced to str(exc). + logger.warning( + "Error stopping BlockingPortal during AsyncExecutor.close " + "(owner=%s); teardown continues.", + owner, + exc_info=True, + ) try: portal_cm.__exit__(None, None, None) - except Exception as e: - logger.warning(f"Error closing BlockingPortal: {e}") + except Exception: + logger.warning( + "Error closing BlockingPortal context manager during " + "AsyncExecutor.close (owner=%s); teardown continues.", + owner, + exc_info=True, + ) + + # Run the shutdown on a helper thread so we can bound the wait: the + # portal thread can be stuck on work that does not honour cancellation + # (for example an await blocked inside a worker thread), and anyio + # joins it with no timeout. + try: + waiter = threading.Thread( + target=_shutdown, name=thread_name, daemon=True + ) + waiter.start() + except RuntimeError: + # Interpreter is shutting down and will not start new threads. + # The portal thread is a daemon; let the process reap it. + return + + waiter.join(timeout) + if waiter.is_alive(): + # Abandonment is observable: name the owner, the timeout, and + # that cancellation was already attempted, so the operator can + # correlate with py-spy / the wedged resource. Per #4598 the + # thread may genuinely be un-interruptible from Python. + logger.warning( + "AsyncExecutor (owner=%s): BlockingPortal did not shut down " + "within %.1fs; abandoning its helper thread. Cancellation was " + "attempted but the portal thread appears to be stuck on work " + "that does not honour cancellation. Its resources (subprocess " + "handles, sockets, fds) may remain alive until process exit. " + "See issue #4598.", + owner, + float(timeout) if timeout is not None else -1.0, + ) def __enter__(self): return self diff --git a/tests/sdk/utils/test_async_executor.py b/tests/sdk/utils/test_async_executor.py new file mode 100644 index 0000000000..c5c72dc930 --- /dev/null +++ b/tests/sdk/utils/test_async_executor.py @@ -0,0 +1,71 @@ +"""Tests for AsyncExecutor shutdown behaviour.""" + +import threading +import time + +import anyio +from anyio.to_thread import run_sync + +from openhands.sdk.utils.async_executor import AsyncExecutor + + +def _close_in_background(executor: AsyncExecutor, **kwargs) -> threading.Event: + """Call close() off-thread so a hang fails the test instead of freezing it.""" + done = threading.Event() + + def run() -> None: + executor.close(**kwargs) + done.set() + + threading.Thread(target=run, daemon=True).start() + return done + + +def test_close_returns_with_task_still_running(): + """close() must not wait for in-flight tasks to finish on their own.""" + executor = AsyncExecutor() + executor.portal.start_task_soon(anyio.sleep_forever) + time.sleep(0.1) + + done = _close_in_background(executor) + + assert done.wait(timeout=10), "close() blocked on a task that never finishes" + + +def test_close_gives_up_on_uncancellable_task(): + """A task that ignores cancellation must not block close() past the timeout.""" + + async def blocked_in_worker_thread() -> None: + # anyio cannot deliver cancellation until the worker thread returns. + await run_sync(lambda: time.sleep(30)) + + executor = AsyncExecutor() + executor.portal.start_task_soon(blocked_in_worker_thread) + time.sleep(0.1) + + done = _close_in_background(executor, timeout=1.0) + + assert done.wait(timeout=15), "close() blocked on an uncancellable task" + + +def test_close_is_idempotent(): + executor = AsyncExecutor() + _ = executor.portal + + executor.close() + executor.close() + + +def test_close_without_started_portal(): + """close() on a lazily-created executor that never started a portal.""" + AsyncExecutor().close() + + +def test_run_async_still_works_before_close(): + executor = AsyncExecutor() + + async def add(a: int, b: int) -> int: + return a + b + + assert executor.run_async(add, 1, 2) == 3 + executor.close() From 102e862821e08525c2a76464371e1a8209ffd553 Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 21 Aug 2026 11:37:31 +0000 Subject: [PATCH 2/2] chore: add live repro script for PR evidence --- .pr/repro-async-executor-close-hang.py | 295 ++++++++++++++++++ .../openhands/sdk/utils/async_executor.py | 4 +- 2 files changed, 296 insertions(+), 3 deletions(-) create mode 100644 .pr/repro-async-executor-close-hang.py diff --git a/.pr/repro-async-executor-close-hang.py b/.pr/repro-async-executor-close-hang.py new file mode 100644 index 0000000000..54fc7f9b1a --- /dev/null +++ b/.pr/repro-async-executor-close-hang.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +""" +repro-async-executor-close-hang.py — Live evidence for PR #4548 / issue #4546. + +Reproduces the AsyncExecutor.close() hang that wedges the agent-server's +conversation lifecycle lock, and verifies the fix bounds it. + +Two phases: + + Phase 1 (deterministic, the smoking gun): + Directly exercise AsyncExecutor.close() with a task that never completes. + Without the fix: close() blocks forever (we abort after PROBE_DEADLINE). + With the fix: close() returns within DEFAULT_CLOSE_TIMEOUT (30s), and + near-instantly for a cancellable task (anyio.sleep_forever). + + Phase 2 (HTTP, live backend): + Hammer the running agent-server with concurrent conversation create + + search + delete traffic. With the fix in place, all requests succeed + quickly and the backend stays responsive. (Phase 1 is the deterministic + proof that without the fix, close() hangs; Phase 2 confirms the live + backend does not stall once the fix is applied.) + +USAGE + python3 repro-async-executor-close-hang.py [--http URL] [--no-unit] + + --http URL Also run the HTTP concurrent-load phase against URL + (default: http://localhost:8000) + --no-unit Skip the deterministic unit phase + +EXIT CODE + 0 all phases passed (fix is working) + 1 a phase failed/stalled (bug reproduced) +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import threading +import time +import urllib.error +import urllib.request +from concurrent.futures import ThreadPoolExecutor, as_completed + + +# Make the SDK importable from the local checkout. +_SDK = "/home/gneubig/work/software-agent-sdk/openhands-sdk" +if _SDK not in sys.path: + sys.path.insert(0, _SDK) + +PROBE_DEADLINE = 15.0 # seconds we wait before declaring close() a hang + + +def _banner(title: str) -> None: + print(f"\n{'═' * 70}\n {title}\n{'═' * 70}") + + +def _status(ok: bool, msg: str) -> int: + tag = "✓ PASS" if ok else "✗ FAIL" + print(f" {tag} — {msg}") + return 0 if ok else 1 + + +# ── Phase 1: deterministic AsyncExecutor.close() repro ─────────────────── + + +def _close_in_background(executor, **kwargs) -> threading.Event: + """Call close() off-thread so a hang doesn't freeze the script.""" + done = threading.Event() + + def run(): + try: + executor.close(**kwargs) + except Exception as e: + print(f" (close() raised: {e})") + finally: + done.set() + + threading.Thread(target=run, daemon=True).start() + return done + + +def phase1_unit() -> int: + """Deterministic reproduction of the AsyncExecutor.close() hang.""" + import anyio # noqa: F401 (prove it's importable) + + from openhands.sdk.utils.async_executor import AsyncExecutor + + rc = 0 + _banner("Phase 1 — AsyncExecutor.close() with a never-finishing task") + + # Case A: a cancellable task (anyio.sleep_forever). With the fix, + # cancellation is delivered and close() returns almost instantly. + # Without the fix, close() waits forever for the task to finish on its own. + print("\n Case A: cancellable task (anyio.sleep_forever)") + executor = AsyncExecutor() + executor.portal.start_task_soon(anyio.sleep_forever) + time.sleep(0.2) # let the task start + + t0 = time.monotonic() + done = _close_in_background(executor) + finished = done.wait(timeout=PROBE_DEADLINE) + elapsed = time.monotonic() - t0 + + if finished: + rc |= _status(True, f"close() returned in {elapsed:.2f}s (fix working)") + else: + rc |= _status(False, f"close() hung > {PROBE_DEADLINE:.0f}s (BUG reproduced)") + # best-effort: leave the daemon thread to die with the process + + # Case B: a task blocked in a worker thread (uncancellable). With the fix, + # close() waits up to DEFAULT_CLOSE_TIMEOUT then abandons the daemon thread. + # Without the fix, close() hangs forever. + print("\n Case B: uncancellable task (blocked in worker thread)") + from anyio.to_thread import run_sync + + async def blocked_in_worker_thread(): + await run_sync(lambda: time.sleep(60)) + + executor2 = AsyncExecutor() + executor2.portal.start_task_soon(blocked_in_worker_thread) + time.sleep(0.2) + + t0 = time.monotonic() + done2 = _close_in_background(executor2, timeout=2.0) + finished2 = done2.wait(timeout=PROBE_DEADLINE) + elapsed2 = time.monotonic() - t0 + + if finished2: + rc |= _status( + True, + f"close() returned in {elapsed2:.2f}s with timeout=2.0 (fix working)", + ) + else: + rc |= _status( + False, f"close() hung > {PROBE_DEADLINE:.0f}s even with timeout (BUG)" + ) + + # Case C: idempotent close (no hang on a clean executor). + print("\n Case C: idempotent close on a clean executor") + executor3 = AsyncExecutor() + _ = executor3.portal + t0 = time.monotonic() + executor3.close() + executor3.close() + elapsed3 = time.monotonic() - t0 + rc |= _status(True, f"double close() returned in {elapsed3:.2f}s") + + return rc + + +# ── Phase 2: HTTP concurrent load against the live backend ──────────────── + + +def _http( + method: str, url: str, key: str, body: dict | None = None +) -> tuple[int, float]: + """Fire one HTTP request; return (status_code, elapsed_seconds).""" + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request( + url, + data=data, + method=method, + headers={ + "X-Session-API-Key": key, + "Content-Type": "application/json", + }, + ) + t0 = time.monotonic() + try: + with urllib.request.urlopen(req, timeout=20) as resp: + resp.read() + return resp.status, time.monotonic() - t0 + except urllib.error.HTTPError as e: + return e.code, time.monotonic() - t0 + except Exception: + return 0, time.monotonic() - t0 + + +def phase2_http(base_url: str, key: str, concurrency: int = 10, rounds: int = 3) -> int: + """Hammer the backend with concurrent READ traffic. + + Uses search/health/alive — which exercise the lifecycle-lock read-path + (the fast-path fixed by PR #4513) — without depending on LLM auth, so the + result isolates the deadlock-fix behaviour from LLM availability. + """ + _banner( + f"Phase 2 — HTTP concurrent read load ({concurrency} workers × {rounds} rounds)" + ) + + # Preflight + code, _ = _http("GET", f"{base_url}/health", key) + if code != 200: + print(f" ✗ FAIL — backend not healthy (health={code})") + return 1 + print(f" preflight /health = {code} ✓") + + latencies: list[float] = [] + failures = 0 + total = 0 + + def one_cycle(_i: int) -> bool: + nonlocal failures, total + ok = True + # 1. search (exercises the lifecycle-lock read-path fast-path) + code, t = _http("GET", f"{base_url}/api/conversations/search?limit=5", key) + latencies.append(t) + total += 1 + if code != 200: + failures += 1 + ok = False + # 2. /alive (liveness, no lock) + code, t = _http("GET", f"{base_url}/alive", key) + latencies.append(t) + total += 1 + if code != 200: + failures += 1 + ok = False + # 3. /server_info (metadata, no lock) + code, t = _http("GET", f"{base_url}/server_info", key) + latencies.append(t) + total += 1 + if code != 200: + failures += 1 + ok = False + return ok + + for r in range(rounds): + t0 = time.monotonic() + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futs = [ + pool.submit(one_cycle, r * concurrency + i) for i in range(concurrency) + ] + results = [f.result() for f in as_completed(futs)] + round_elapsed = time.monotonic() - t0 + ok = sum(results) + print( + f" round {r + 1}/{rounds}: {ok}/{concurrency} cycles ok " + f"in {round_elapsed:.2f}s" + ) + + if not latencies: + print(" ✗ FAIL — no requests completed") + return 1 + + latencies.sort() + p50 = latencies[len(latencies) // 2] + p99 = latencies[int(len(latencies) * 0.99)] + print( + f"\n {total} requests: {failures} failures, " + f"p50={p50 * 1000:.0f}ms, p99={p99 * 1000:.0f}ms" + ) + + if failures > 0: + return _status(False, f"{failures} requests failed/stalled (backend unhealthy)") + return _status(True, "all requests succeeded, backend stayed responsive") + + +# ── Main ───────────────────────────────────────────────────────────────── + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument( + "--http", metavar="URL", default=None, help="run the HTTP phase against URL" + ) + ap.add_argument("--no-unit", action="store_true", help="skip the unit phase") + ap.add_argument( + "--api-key", + default=os.environ.get( + "WATCHDOG_API_KEY", + open("/home/gneubig/.openhands/agent-canvas/api-key.txt").read().strip(), + ), + ) + args = ap.parse_args() + + rc = 0 + if not args.no_unit: + rc |= phase1_unit() + + if args.http: + rc |= phase2_http(args.http, args.api_key) + + _banner("RESULT") + if rc == 0: + print(" ✓ All phases passed — fix is working, no stall.") + else: + print(" ✗ A phase failed/stalled — bug reproduced.") + return rc + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/openhands-sdk/openhands/sdk/utils/async_executor.py b/openhands-sdk/openhands/sdk/utils/async_executor.py index 29087a17e1..39fe81cd4e 100644 --- a/openhands-sdk/openhands/sdk/utils/async_executor.py +++ b/openhands-sdk/openhands/sdk/utils/async_executor.py @@ -195,9 +195,7 @@ def _shutdown() -> None: # (for example an await blocked inside a worker thread), and anyio # joins it with no timeout. try: - waiter = threading.Thread( - target=_shutdown, name=thread_name, daemon=True - ) + waiter = threading.Thread(target=_shutdown, name=thread_name, daemon=True) waiter.start() except RuntimeError: # Interpreter is shutting down and will not start new threads.