From 366bd8c83ddc19161c52c7e1425e0f1f343a4acc Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 17 Aug 2026 13:29:54 +0000 Subject: [PATCH 01/12] test(agent-server): lifecycle lock deadlocks on thread-pool exhaustion --- ...st_event_service_thread_pool_exhaustion.py | 136 ++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 tests/agent_server/test_event_service_thread_pool_exhaustion.py diff --git a/tests/agent_server/test_event_service_thread_pool_exhaustion.py b/tests/agent_server/test_event_service_thread_pool_exhaustion.py new file mode 100644 index 0000000000..0c871c3eb4 --- /dev/null +++ b/tests/agent_server/test_event_service_thread_pool_exhaustion.py @@ -0,0 +1,136 @@ +"""Test: thread-pool exhaustion must not deadlock the lifecycle lock. + +``_get_or_load_event_service`` acquires ``lifecycle_lock`` and then calls +``asyncio.to_thread(_prepare_persisted_runtime)`` inside the lock. If the +default thread pool is exhausted, the ``to_thread`` call queues indefinitely +while still holding the lock. Every subsequent ``get_event_service`` call — +including the WebSocket event-stream path and the REST ``/events/search`` +endpoint — blocks waiting for the lock, making the entire agent-server appear +wedged even though simple endpoints (``/ready``, ``/api/settings``) still +respond. + +This test asserts the **correct** behavior: + +* Loading an already-cached conversation (no ``to_thread`` needed) must + succeed immediately even when the thread pool is full, because the + lifecycle lock must not be held while waiting for a thread. + +The test currently **fails** on unpatched code, demonstrating the bug. +""" + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from uuid import UUID + +import pytest + +from openhands.agent_server.conversation_service import ConversationService +from openhands.agent_server.models import StartConversationRequest +from openhands.sdk import LLM, Agent +from openhands.sdk.security.confirmation_policy import NeverConfirm +from openhands.sdk.workspace import LocalWorkspace + + +async def _create_persisted_conversation( + conversations_dir: Path, workspace_dir: Path +) -> UUID: + """Create a conversation on disk so it can be loaded later.""" + workspace_dir.mkdir(parents=True, exist_ok=True) + request = StartConversationRequest( + agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), + workspace=LocalWorkspace(working_dir=str(workspace_dir)), + confirmation_policy=NeverConfirm(), + ) + async with ConversationService(conversations_dir=conversations_dir) as service: + info, _ = await service.start_conversation(request) + return info.id + + +@pytest.mark.asyncio +async def test_thread_pool_exhaustion_does_not_block_cached_conversation( + tmp_path, +): + """Loading an already-cached conversation must not deadlock when the + thread pool is exhausted. + + The lifecycle lock is held while ``asyncio.to_thread`` waits for a free + thread. If the pool is full, the lock is held indefinitely, blocking + every subsequent ``get_event_service`` call — even for conversations that + are already in the cache and need no thread work at all. + + This is the root cause of the "events don't load" production incident. + """ + conversations_dir = tmp_path / "conversations" + workspace_dir = tmp_path / "workspace" + + # 1. Create two persisted conversations on disk. + conv_a = await _create_persisted_conversation( + conversations_dir, workspace_dir / "a" + ) + conv_b = await _create_persisted_conversation( + conversations_dir, workspace_dir / "b" + ) + + # 2. Replace the event-loop's default executor with a 1-worker pool. + loop = asyncio.get_running_loop() + tiny_pool = ThreadPoolExecutor(max_workers=1) + loop.set_default_executor(tiny_pool) + + service = ConversationService(conversations_dir=conversations_dir) + await service.__aenter__() + block_event: threading.Event | None = None + block_task: asyncio.Task[None] | None = None + try: + # Pre-load both conversations into the cache. + es_a = await service.get_event_service(conv_a) + assert es_a is not None + es_b = await service.get_event_service(conv_b) + assert es_b is not None + + # 3. Saturate the 1-thread pool with a blocking call that never + # returns on its own. + block_event = threading.Event() + + async def _block_pool() -> None: + await loop.run_in_executor(None, block_event.wait) + + block_task = asyncio.create_task(_block_pool()) + await asyncio.sleep(0.3) + + # 4. Evict conv_a from the cache so the next load needs + # asyncio.to_thread(_prepare_persisted_runtime) — this call will + # hang waiting for a thread. + if service._event_services is not None: + service._event_services.pop(conv_a, None) + + # Kick off the stuck load in the background. Give it time to + # acquire the lifecycle lock and enter asyncio.to_thread before we + # try the cached load below. + stuck_task = asyncio.create_task(service.get_event_service(conv_a)) + await asyncio.sleep(0.3) + + # 5. Loading conv_b — which IS in the cache and needs no thread work — + # must succeed immediately. The lifecycle lock must not be held + # while waiting for a thread. + es_b2 = await asyncio.wait_for(service.get_event_service(conv_b), timeout=3.0) + assert es_b2 is not None + assert es_b2 is es_b + + # Clean up the stuck task. + block_event.set() + try: + await asyncio.wait_for(stuck_task, timeout=10.0) + except (TimeoutError, asyncio.CancelledError): + stuck_task.cancel() + finally: + if block_event is not None: + block_event.set() + if block_task is not None: + try: + await asyncio.wait_for(block_task, timeout=5.0) + except (TimeoutError, asyncio.CancelledError): + block_task.cancel() + await service.__aexit__(None, None, None) + tiny_pool.shutdown(wait=False) From 5ff593b7f67080c3d9a39725febef3b26d1636a2 Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 17 Aug 2026 13:32:50 +0000 Subject: [PATCH 02/12] fix(agent-server): check event-service cache before acquiring lifecycle lock --- .../openhands/agent_server/conversation_service.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index f641d7e3dd..f6786e225c 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -1029,6 +1029,20 @@ def _prepare_persisted_runtime(self, stored: StoredConversation) -> None: async def _get_or_load_event_service( self, conversation_id: UUID ) -> EventService | None: + # Fast path: check the cache *without* acquiring the lifecycle lock. + # This ensures that already-loaded conversations can be retrieved even + # when the lock is held by a slow ``asyncio.to_thread`` call for a + # different conversation. The dict lookup is safe because + # ``_event_services`` is only mutated under the lock (insert / delete); + # a concurrent reader may see a stale snapshot, but the worst case is a + # cache miss that falls through to the locked path below. + event_services = self._event_services + if event_services is not None: + cached = event_services.get(conversation_id) + if cached is not None and cached.is_open(): + cached.touch() + return cached + async with self._lifecycle_lock: return await self._get_or_load_event_service_locked(conversation_id) From f269e7a4f2a959cef29fbaf5709fac0db24a9cd5 Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 17 Aug 2026 13:38:07 +0000 Subject: [PATCH 03/12] ci: re-trigger PR description check From 3391fe50c08c978e8d9dde47a52fc0cf89b43558 Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 17 Aug 2026 13:40:56 +0000 Subject: [PATCH 04/12] ci: re-trigger after issue label From a2b81124bb4e50d111804558ac3dbbe031c31d91 Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 17 Aug 2026 13:45:03 +0000 Subject: [PATCH 05/12] ci: re-trigger after label fix From 5c0ff27386daabd05830cd70cca7c805072978e3 Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 17 Aug 2026 13:48:09 +0000 Subject: [PATCH 06/12] ci: re-trigger after issue readiness From e3890ce1e2052d68def743239c2096dbbe213a90 Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 17 Aug 2026 20:59:05 +0000 Subject: [PATCH 07/12] ci: re-trigger flaky release-note label sync Co-authored-by: openhands From b25bdc2c932b3d54aff804a0338f34c80ed840b3 Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 21 Aug 2026 11:37:38 +0000 Subject: [PATCH 08/12] chore: add live repro script for PR evidence --- .pr/repro-async-executor-close-hang.py | 289 +++++++++++++++++++++++++ 1 file changed, 289 insertions(+) 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..f87a21f7de --- /dev/null +++ b/.pr/repro-async-executor-close-hang.py @@ -0,0 +1,289 @@ +#!/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.request +import urllib.error +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()) From 0608752f118322b887efc664b7b78638dbb46028 Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 21 Aug 2026 13:04:18 +0000 Subject: [PATCH 09/12] fix: remove unrelated repro script causing lint failures The repro script (.pr/repro-async-executor-close-hang.py) was accidentally committed from a separate debugging session (references PR #4548/issue #4546, not this PR's issue #4514). It fails pre-commit (import ordering, ARG001 unused arg) and has hardcoded developer paths and credential references. Deleting it resolves both the CI lint failures and the 3 review threads. Co-authored-by: openhands --- .pr/repro-async-executor-close-hang.py | 289 ------------------------- 1 file changed, 289 deletions(-) delete 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 deleted file mode 100644 index f87a21f7de..0000000000 --- a/.pr/repro-async-executor-close-hang.py +++ /dev/null @@ -1,289 +0,0 @@ -#!/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.request -import urllib.error -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()) From 61069b2dec7a840bebfeecab42bac114cd1a130c Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 21 Aug 2026 13:06:40 +0000 Subject: [PATCH 10/12] ci: re-trigger unresolved-review-threads check after resolving all threads Co-authored-by: openhands From 37099a14e5ae6de48f12737e154873ac5fe9e4f7 Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 21 Aug 2026 13:38:36 +0000 Subject: [PATCH 11/12] fix: address bot review suggestions (comment wording + stuck_task cleanup in finally) - Reword fast-path comment per bot suggestion: replace stale-snapshot language with accurate description of the fast-path miss scenario. - Declare stuck_task before try block, init to None, cancel in finally to prevent pending task warning if TimeoutError jumps to finally. Co-authored-by: openhands --- .../openhands/agent_server/conversation_service.py | 4 ++-- .../test_event_service_thread_pool_exhaustion.py | 6 ++++++ 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index f6786e225c..6327258fa5 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -1034,8 +1034,8 @@ async def _get_or_load_event_service( # when the lock is held by a slow ``asyncio.to_thread`` call for a # different conversation. The dict lookup is safe because # ``_event_services`` is only mutated under the lock (insert / delete); - # a concurrent reader may see a stale snapshot, but the worst case is a - # cache miss that falls through to the locked path below. + # a fast-path miss falls through to the locked slow path, which + # re-checks the cache and handles any state changes correctly. event_services = self._event_services if event_services is not None: cached = event_services.get(conversation_id) diff --git a/tests/agent_server/test_event_service_thread_pool_exhaustion.py b/tests/agent_server/test_event_service_thread_pool_exhaustion.py index 0c871c3eb4..8a927337b0 100644 --- a/tests/agent_server/test_event_service_thread_pool_exhaustion.py +++ b/tests/agent_server/test_event_service_thread_pool_exhaustion.py @@ -82,6 +82,7 @@ async def test_thread_pool_exhaustion_does_not_block_cached_conversation( await service.__aenter__() block_event: threading.Event | None = None block_task: asyncio.Task[None] | None = None + stuck_task: asyncio.Task[None] | None = None try: # Pre-load both conversations into the cache. es_a = await service.get_event_service(conv_a) @@ -132,5 +133,10 @@ async def _block_pool() -> None: await asyncio.wait_for(block_task, timeout=5.0) except (TimeoutError, asyncio.CancelledError): block_task.cancel() + if stuck_task is not None: + try: + await asyncio.wait_for(stuck_task, timeout=5.0) + except (TimeoutError, asyncio.CancelledError): + stuck_task.cancel() await service.__aexit__(None, None, None) tiny_pool.shutdown(wait=False) From d1b25f5837b0157ca34fe8ed32546cc75821900a Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 21 Aug 2026 13:43:18 +0000 Subject: [PATCH 12/12] fix: correct stuck_task type annotation for pyright Use Task[EventService | None] instead of Task[None] to match the return type of get_event_service. Add assert for type narrowing in the happy-path cleanup block. Co-authored-by: openhands --- .../agent_server/test_event_service_thread_pool_exhaustion.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/agent_server/test_event_service_thread_pool_exhaustion.py b/tests/agent_server/test_event_service_thread_pool_exhaustion.py index 8a927337b0..c0febbfb85 100644 --- a/tests/agent_server/test_event_service_thread_pool_exhaustion.py +++ b/tests/agent_server/test_event_service_thread_pool_exhaustion.py @@ -27,6 +27,7 @@ import pytest from openhands.agent_server.conversation_service import ConversationService +from openhands.agent_server.event_service import EventService from openhands.agent_server.models import StartConversationRequest from openhands.sdk import LLM, Agent from openhands.sdk.security.confirmation_policy import NeverConfirm @@ -82,7 +83,7 @@ async def test_thread_pool_exhaustion_does_not_block_cached_conversation( await service.__aenter__() block_event: threading.Event | None = None block_task: asyncio.Task[None] | None = None - stuck_task: asyncio.Task[None] | None = None + stuck_task: asyncio.Task[EventService | None] | None = None try: # Pre-load both conversations into the cache. es_a = await service.get_event_service(conv_a) @@ -121,6 +122,7 @@ async def _block_pool() -> None: # Clean up the stuck task. block_event.set() + assert stuck_task is not None try: await asyncio.wait_for(stuck_task, timeout=10.0) except (TimeoutError, asyncio.CancelledError):