From 934906bd0c47fa400f1c64a9d2f104e9676a8140 Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 21 Aug 2026 15:05:00 +0000 Subject: [PATCH 1/8] fix(agent-server): replace global _lifecycle_lock with per-conversation locks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConversationService guarded all conversation lifecycle operations with a single global asyncio.Lock (_lifecycle_lock). Two of those operations do slow/blocking work under the lock: - _get_or_load_event_service_locked: asyncio.to_thread(_prepare_persisted_runtime) - delete_conversation: await event_service.close() (can hang — see #4546) A stuck or slow operation on conversation A therefore blocked every other conversation — the entire server wedged while /health kept answering. Replace the global lock with per-conversation locks (_conversation_locks dict + _catalog_lock for dict mutation only). Each method acquires only its conversation's lock, so a stuck close() on conversation A blocks only conversation A; conversation B's create/search/open proceeds unimpeded. The three operations that genuinely touch ALL conversations keep the global lock: prepare_for_sandbox_pause, _evict_idle_conversations, __aexit__. This is the fundamental fix for issue #4569. PRs #4513 and #4548 become mitigations that can be closed as superseded once this lands. --- .../agent_server/conversation_service.py | 35 +++++++++++++++---- 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index 32f9e30f59..b6de260ad4 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -3,6 +3,7 @@ import json import logging import os +import threading from collections.abc import Awaitable, Callable from concurrent.futures import ThreadPoolExecutor from contextlib import suppress @@ -645,6 +646,10 @@ class ConversationService: default_factory=dict, init=False ) _lifecycle_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False) + _conversation_locks: dict[UUID, asyncio.Lock] = field( + default_factory=dict, init=False + ) + _catalog_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False) _conversation_webhook_subscribers: list["ConversationWebhookSubscriber"] = field( default_factory=list, init=False ) @@ -751,7 +756,7 @@ async def activate_credential_binding( secret_name: str, binding: VersionedCredentialBinding, ) -> None: - async with self._lifecycle_lock: + async with self._get_conversation_lock(conversation_id): event_services = self._event_services event_service = ( event_services.get(conversation_id) @@ -1047,10 +1052,28 @@ def _prepare_persisted_runtime(self, stored: StoredConversation) -> None: context=f"resuming conversation {stored.id}", ) + def _get_conversation_lock(self, conversation_id: UUID) -> asyncio.Lock: + """Return the per-conversation lifecycle lock, creating it if needed. + + Replaces the global ``_lifecycle_lock`` for operations that act on a + *single* conversation. The ``_catalog_lock`` only guards the + ``_conversation_locks`` dict itself (a brief, in-memory mutation), so + acquiring a conversation lock never blocks operations on other + conversations. + """ + with self._catalog_lock_sync: + lock = self._conversation_locks.get(conversation_id) + if lock is None: + lock = asyncio.Lock() + self._conversation_locks[conversation_id] = lock + return lock + + _catalog_lock_sync: threading.Lock = field(default_factory=threading.Lock, init=False) + async def _get_or_load_event_service( self, conversation_id: UUID ) -> EventService | None: - async with self._lifecycle_lock: + async with self._get_conversation_lock(conversation_id): return await self._get_or_load_event_service_locked(conversation_id) async def _get_or_load_event_service_locked( @@ -1318,7 +1341,7 @@ async def _start_conversation( if existing_record is not None or ( existing_event_service is not None and existing_event_service.is_open() ): - async with self._lifecycle_lock: + async with self._get_conversation_lock(conversation_id): existing_event_service = self._event_services.get(conversation_id) if ( existing_event_service is not None @@ -1625,7 +1648,7 @@ async def _start_conversation( launched_agent_profile=launched_agent_profile, **request_data, ) - async with self._lifecycle_lock: + async with self._get_conversation_lock(conversation_id): # New conversation: the agent is written to base_state.json (its # single source of truth), not to meta.json. Pass it explicitly. # ``new_agent`` is ``request.agent`` (decrypted when the request was @@ -1683,7 +1706,7 @@ async def resume_conversation(self, conversation_id: UUID) -> bool: return bool(await self._get_or_load_event_service(conversation_id)) async def delete_conversation(self, conversation_id: UUID) -> bool: - async with self._lifecycle_lock: + async with self._get_conversation_lock(conversation_id): event_services = self._event_services if event_services is None: raise ValueError("inactive_service") @@ -1911,7 +1934,7 @@ async def fork_conversation( # directory so we don't leave stale state on disk. fork_dir = self.conversations_dir / fork_conv_id.hex try: - async with self._lifecycle_lock: + async with self._get_conversation_lock(conversation_id): fork_event_service = await self._start_event_service( fork_stored, is_new_conversation=True, agent=fork_agent ) From 763088f552bbf515bf24248f1fc6b783f5da4fb7 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 2/8] 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 | 51 ++++++++++++- tests/sdk/utils/test_async_executor.py | 71 +++++++++++++++++++ 2 files changed, 120 insertions(+), 2 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..3b76f5d467 100644 --- a/openhands-sdk/openhands/sdk/utils/async_executor.py +++ b/openhands-sdk/openhands/sdk/utils/async_executor.py @@ -13,6 +13,11 @@ 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. +DEFAULT_CLOSE_TIMEOUT = 30.0 + class AsyncExecutor: """ @@ -101,18 +106,60 @@ 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. + + Args: + timeout: seconds to wait for the portal thread to exit. ``None`` + waits indefinitely (the previous behaviour). + """ 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 + + 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 as e: + logger.warning(f"Error stopping BlockingPortal: {e}") try: portal_cm.__exit__(None, None, None) except Exception as e: logger.warning(f"Error closing BlockingPortal: {e}") + # 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="async-executor-close", 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(): + logger.warning( + f"BlockingPortal did not shut down within {timeout}s; " + "abandoning its thread. Something scheduled on it is ignoring " + "cancellation." + ) + 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 6725bb839878f25df37fbd3f5da1c515e15e9d15 Mon Sep 17 00:00:00 2001 From: neubig Date: Sat, 22 Aug 2026 10:35:30 +0000 Subject: [PATCH 3/8] fix(agent-server): stabilize per-conversation lifecycle tests Co-authored-by: openhands --- .../openhands/agent_server/conversation_service.py | 6 ++++-- openhands-sdk/openhands/sdk/io/local.py | 5 +++-- tests/agent_server/test_conversation_service.py | 8 +++++--- 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index b6de260ad4..f663fcfdaa 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -1068,7 +1068,9 @@ def _get_conversation_lock(self, conversation_id: UUID) -> asyncio.Lock: self._conversation_locks[conversation_id] = lock return lock - _catalog_lock_sync: threading.Lock = field(default_factory=threading.Lock, init=False) + _catalog_lock_sync: threading.Lock = field( + default_factory=threading.Lock, init=False + ) async def _get_or_load_event_service( self, conversation_id: UUID @@ -1934,7 +1936,7 @@ async def fork_conversation( # directory so we don't leave stale state on disk. fork_dir = self.conversations_dir / fork_conv_id.hex try: - async with self._get_conversation_lock(conversation_id): + async with self._get_conversation_lock(source_id): fork_event_service = await self._start_event_service( fork_stored, is_new_conversation=True, agent=fork_agent ) diff --git a/openhands-sdk/openhands/sdk/io/local.py b/openhands-sdk/openhands/sdk/io/local.py index 3e2ef4cd25..36179470ea 100644 --- a/openhands-sdk/openhands/sdk/io/local.py +++ b/openhands-sdk/openhands/sdk/io/local.py @@ -2,11 +2,13 @@ import shutil from collections.abc import Iterator from contextlib import contextmanager +from pathlib import Path from filelock import FileLock, Timeout from openhands.sdk.io.cache import MemoryLRUCache from openhands.sdk.logger import get_logger +from openhands.sdk.utils.files import atomic_write_text from openhands.sdk.utils.path import to_posix_path from .base import FileStore @@ -62,8 +64,7 @@ def write(self, path: str, contents: str | bytes) -> None: full_path = self.get_full_path(path) os.makedirs(os.path.dirname(full_path), exist_ok=True) if isinstance(contents, str): - with open(full_path, "w", encoding="utf-8") as f: - f.write(contents) + atomic_write_text(Path(full_path), contents) self.cache[full_path] = contents else: with open(full_path, "wb") as f: diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 4fb3b79784..221124ffc7 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -926,7 +926,9 @@ async def test_waiting_hydration_cannot_restore_deleted_conversation( replacement_runtime = AsyncMock(spec=EventService) replacement_runtime.stored = record.stored - async def publish_replacement(_stored: StoredConversation) -> EventService: + async def publish_replacement( + _stored: StoredConversation, *, agent: AgentBase | None = None + ) -> EventService: assert service._event_services is not None service._event_services[conversation_id] = replacement_runtime service._conversation_records[conversation_id] = record @@ -940,7 +942,7 @@ async def publish_replacement(_stored: StoredConversation) -> EventService: ) as start_event_service, patch("openhands.agent_server.conversation_service.safe_rmtree"), ): - await service._lifecycle_lock.acquire() + await service._get_conversation_lock(conversation_id).acquire() try: getter_task = asyncio.create_task( service.get_event_service(conversation_id) @@ -953,7 +955,7 @@ async def publish_replacement(_stored: StoredConversation) -> EventService: ) await asyncio.sleep(0) finally: - service._lifecycle_lock.release() + service._get_conversation_lock(conversation_id).release() getter_result, deleted = await asyncio.gather(getter_task, delete_task) From 9596ecc3404e27520c66043176cf48807bc83ae6 Mon Sep 17 00:00:00 2001 From: neubig Date: Sat, 22 Aug 2026 10:53:55 +0000 Subject: [PATCH 4/8] ci: retrigger pull request checks Co-authored-by: openhands From c0e35e9e41089bb9c63bf44dcee33426026c9ef9 Mon Sep 17 00:00:00 2001 From: neubig Date: Sat, 22 Aug 2026 11:26:08 +0000 Subject: [PATCH 5/8] fix(sdk): restore transparent executor shutdown Co-authored-by: openhands --- .../openhands/sdk/utils/async_executor.py | 51 +------------ tests/sdk/utils/test_async_executor.py | 71 ------------------- 2 files changed, 2 insertions(+), 120 deletions(-) delete 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 3b76f5d467..611f511734 100644 --- a/openhands-sdk/openhands/sdk/utils/async_executor.py +++ b/openhands-sdk/openhands/sdk/utils/async_executor.py @@ -13,11 +13,6 @@ 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. -DEFAULT_CLOSE_TIMEOUT = 30.0 - class AsyncExecutor: """ @@ -106,60 +101,18 @@ async def _execute(): return portal.call(_execute) - def close(self, timeout: float | None = DEFAULT_CLOSE_TIMEOUT): - """Shut down the portal, without ever blocking the caller forever. - - Args: - timeout: seconds to wait for the portal thread to exit. ``None`` - waits indefinitely (the previous behaviour). - """ + def close(self): with self._lock: portal_cm = self._portal_cm - portal = self._portal self._portal_cm = None self._portal = None - if portal_cm is None: - return - - 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 as e: - logger.warning(f"Error stopping BlockingPortal: {e}") + if portal_cm is not None: try: portal_cm.__exit__(None, None, None) except Exception as e: logger.warning(f"Error closing BlockingPortal: {e}") - # 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="async-executor-close", 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(): - logger.warning( - f"BlockingPortal did not shut down within {timeout}s; " - "abandoning its thread. Something scheduled on it is ignoring " - "cancellation." - ) - def __enter__(self): return self diff --git a/tests/sdk/utils/test_async_executor.py b/tests/sdk/utils/test_async_executor.py deleted file mode 100644 index c5c72dc930..0000000000 --- a/tests/sdk/utils/test_async_executor.py +++ /dev/null @@ -1,71 +0,0 @@ -"""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 a5a0f094fc8632e21639f495fd15bdecf6bbd6f5 Mon Sep 17 00:00:00 2001 From: neubig Date: Sat, 22 Aug 2026 20:53:38 +0000 Subject: [PATCH 6/8] chore: address PR review feedback (#4570) Co-authored-by: openhands --- .../agent_server/conversation_service.py | 89 ++++++++++++++----- .../agent_server/test_conversation_service.py | 71 ++++++++++++++- 2 files changed, 134 insertions(+), 26 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index f663fcfdaa..ffaab09997 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -6,11 +6,12 @@ import threading from collections.abc import Awaitable, Callable from concurrent.futures import ThreadPoolExecutor -from contextlib import suppress +from contextlib import asynccontextmanager, suppress from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, cast from uuid import UUID, uuid4 +from weakref import WeakValueDictionary import httpx from pydantic import BaseModel @@ -646,10 +647,14 @@ class ConversationService: default_factory=dict, init=False ) _lifecycle_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False) - _conversation_locks: dict[UUID, asyncio.Lock] = field( - default_factory=dict, init=False + _lifecycle_condition: asyncio.Condition = field( + default_factory=asyncio.Condition, init=False + ) + _active_lifecycle_operations: int = field(default=0, init=False) + _exclusive_lifecycle_pending: bool = field(default=False, init=False) + _conversation_locks: WeakValueDictionary[UUID, asyncio.Lock] = field( + default_factory=WeakValueDictionary, init=False ) - _catalog_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False) _conversation_webhook_subscribers: list["ConversationWebhookSubscriber"] = field( default_factory=list, init=False ) @@ -756,7 +761,7 @@ async def activate_credential_binding( secret_name: str, binding: VersionedCredentialBinding, ) -> None: - async with self._get_conversation_lock(conversation_id): + async with self._conversation_lifecycle(conversation_id): event_services = self._event_services event_service = ( event_services.get(conversation_id) @@ -775,7 +780,7 @@ async def activate_credential_binding( ) async def prepare_for_sandbox_pause(self) -> None: - async with self._lifecycle_lock: + async with self._exclusive_lifecycle(): event_services = self._event_services if event_services is None: raise ValueError("inactive_service") @@ -1053,14 +1058,6 @@ def _prepare_persisted_runtime(self, stored: StoredConversation) -> None: ) def _get_conversation_lock(self, conversation_id: UUID) -> asyncio.Lock: - """Return the per-conversation lifecycle lock, creating it if needed. - - Replaces the global ``_lifecycle_lock`` for operations that act on a - *single* conversation. The ``_catalog_lock`` only guards the - ``_conversation_locks`` dict itself (a brief, in-memory mutation), so - acquiring a conversation lock never blocks operations on other - conversations. - """ with self._catalog_lock_sync: lock = self._conversation_locks.get(conversation_id) if lock is None: @@ -1072,10 +1069,49 @@ def _get_conversation_lock(self, conversation_id: UUID) -> asyncio.Lock: default_factory=threading.Lock, init=False ) + @asynccontextmanager + async def _conversation_lifecycle(self, conversation_id: UUID): + async with self._lifecycle_condition: + await self._lifecycle_condition.wait_for( + lambda: not self._exclusive_lifecycle_pending + ) + self._active_lifecycle_operations += 1 + try: + async with self._get_conversation_lock(conversation_id): + yield + finally: + async with self._lifecycle_condition: + self._active_lifecycle_operations -= 1 + if self._active_lifecycle_operations == 0: + self._lifecycle_condition.notify_all() + + @asynccontextmanager + async def _exclusive_lifecycle(self): + async with self._lifecycle_lock: + try: + async with self._lifecycle_condition: + self._exclusive_lifecycle_pending = True + await self._lifecycle_condition.wait_for( + lambda: self._active_lifecycle_operations == 0 + ) + yield + finally: + async with self._lifecycle_condition: + self._exclusive_lifecycle_pending = False + self._lifecycle_condition.notify_all() + async def _get_or_load_event_service( self, conversation_id: UUID ) -> EventService | None: - async with self._get_conversation_lock(conversation_id): + event_services = self._event_services + if event_services is None: + raise ValueError("inactive_service") + if ( + conversation_id not in event_services + and conversation_id not in self._conversation_records + ): + return None + async with self._conversation_lifecycle(conversation_id): return await self._get_or_load_event_service_locked(conversation_id) async def _get_or_load_event_service_locked( @@ -1343,7 +1379,7 @@ async def _start_conversation( if existing_record is not None or ( existing_event_service is not None and existing_event_service.is_open() ): - async with self._get_conversation_lock(conversation_id): + async with self._conversation_lifecycle(conversation_id): existing_event_service = self._event_services.get(conversation_id) if ( existing_event_service is not None @@ -1650,7 +1686,7 @@ async def _start_conversation( launched_agent_profile=launched_agent_profile, **request_data, ) - async with self._get_conversation_lock(conversation_id): + async with self._conversation_lifecycle(conversation_id): # New conversation: the agent is written to base_state.json (its # single source of truth), not to meta.json. Pass it explicitly. # ``new_agent`` is ``request.agent`` (decrypted when the request was @@ -1708,10 +1744,15 @@ async def resume_conversation(self, conversation_id: UUID) -> bool: return bool(await self._get_or_load_event_service(conversation_id)) async def delete_conversation(self, conversation_id: UUID) -> bool: - async with self._get_conversation_lock(conversation_id): - event_services = self._event_services - if event_services is None: - raise ValueError("inactive_service") + event_services = self._event_services + if event_services is None: + raise ValueError("inactive_service") + if ( + conversation_id not in event_services + and conversation_id not in self._conversation_records + ): + return False + async with self._conversation_lifecycle(conversation_id): event_service = await self._get_or_load_event_service_locked( conversation_id, require_runtime_bindings=False, @@ -1936,7 +1977,7 @@ async def fork_conversation( # directory so we don't leave stale state on disk. fork_dir = self.conversations_dir / fork_conv_id.hex try: - async with self._get_conversation_lock(source_id): + async with self._conversation_lifecycle(fork_conv_id): fork_event_service = await self._start_event_service( fork_stored, is_new_conversation=True, agent=fork_agent ) @@ -2061,7 +2102,7 @@ async def _evict_idle_conversations(self, ttl_seconds: float) -> None: Running or externally-subscribed conversations are skipped. """ - async with self._lifecycle_lock: + async with self._exclusive_lifecycle(): event_services = self._event_services if event_services is None: return @@ -2117,7 +2158,7 @@ async def __aexit__(self, exc_type, exc_value, traceback): await self._lease_renewal_task self._lease_renewal_task = None - async with self._lifecycle_lock: + async with self._exclusive_lifecycle(): event_services = self._event_services if event_services is None: return diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 221124ffc7..2b9e07b3d5 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -590,6 +590,56 @@ async def test_prepare_for_sandbox_pause_drains_active_services(tmp_path): await service.__aexit__(None, None, None) +@pytest.mark.asyncio +async def test_prepare_for_sandbox_pause_blocks_new_hydration( + persisted_conversation, +): + conversations_dir, conversation_id = persisted_conversation + service = ConversationService(conversations_dir=conversations_dir) + await service.__aenter__() + existing_id = uuid4() + existing = AsyncMock(spec=EventService) + close_entered = asyncio.Event() + finish_close = asyncio.Event() + hydration_entered = asyncio.Event() + record = service._conversation_records[conversation_id] + hydrated = AsyncMock(spec=EventService) + hydrated.stored = record.stored + + async def close(*_args): + close_entered.set() + await finish_close.wait() + + async def hydrate(*_args, **_kwargs): + hydration_entered.set() + assert service._event_services is not None + service._event_services[conversation_id] = hydrated + return hydrated + + existing.__aexit__.side_effect = close + assert service._event_services is not None + service._event_services[existing_id] = existing + + try: + with patch.object(service, "_start_event_service", side_effect=hydrate): + pause_task = asyncio.create_task(service.prepare_for_sandbox_pause()) + await asyncio.wait_for(close_entered.wait(), timeout=1) + hydration_task = asyncio.create_task( + service.get_event_service(conversation_id) + ) + await asyncio.sleep(0) + assert not hydration_entered.is_set() + + finish_close.set() + await asyncio.wait_for(pause_task, timeout=1) + assert service._event_services == {} + assert not hydration_entered.is_set() + assert await asyncio.wait_for(hydration_task, timeout=1) is hydrated + finally: + finish_close.set() + await service.__aexit__(None, None, None) + + @pytest.mark.asyncio async def test_prepare_for_sandbox_pause_closes_services_concurrently(tmp_path): service = ConversationService(conversations_dir=tmp_path / "conversations") @@ -942,7 +992,8 @@ async def publish_replacement( ) as start_event_service, patch("openhands.agent_server.conversation_service.safe_rmtree"), ): - await service._get_conversation_lock(conversation_id).acquire() + conversation_lock = service._get_conversation_lock(conversation_id) + await conversation_lock.acquire() try: getter_task = asyncio.create_task( service.get_event_service(conversation_id) @@ -955,7 +1006,7 @@ async def publish_replacement( ) await asyncio.sleep(0) finally: - service._get_conversation_lock(conversation_id).release() + conversation_lock.release() getter_result, deleted = await asyncio.gather(getter_task, delete_task) @@ -2629,6 +2680,22 @@ async def test_delete_conversation_not_found(self, conversation_service): result = await conversation_service.delete_conversation(uuid4()) assert result is False + @pytest.mark.asyncio + async def test_missing_conversations_do_not_accumulate_locks( + self, conversation_service + ): + for _ in range(100): + conversation_id = uuid4() + assert await conversation_service.get_event_service(conversation_id) is None + assert ( + await conversation_service.resume_conversation(conversation_id) is False + ) + assert ( + await conversation_service.delete_conversation(conversation_id) is False + ) + + assert len(conversation_service._conversation_locks) == 0 + @pytest.mark.asyncio async def test_delete_conversation_success(self, conversation_service): """Test successful conversation deletion.""" From ed82a8866996beecb5123ae81c433886c2aa8628 Mon Sep 17 00:00:00 2001 From: neubig Date: Sat, 22 Aug 2026 21:07:08 +0000 Subject: [PATCH 7/8] chore: tidy lifecycle lock field placement (#4570) Co-authored-by: openhands --- .../openhands/agent_server/conversation_service.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index ffaab09997..eb036a5710 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -655,6 +655,9 @@ class ConversationService: _conversation_locks: WeakValueDictionary[UUID, asyncio.Lock] = field( default_factory=WeakValueDictionary, init=False ) + _catalog_lock_sync: threading.Lock = field( + default_factory=threading.Lock, init=False + ) _conversation_webhook_subscribers: list["ConversationWebhookSubscriber"] = field( default_factory=list, init=False ) @@ -1065,10 +1068,6 @@ def _get_conversation_lock(self, conversation_id: UUID) -> asyncio.Lock: self._conversation_locks[conversation_id] = lock return lock - _catalog_lock_sync: threading.Lock = field( - default_factory=threading.Lock, init=False - ) - @asynccontextmanager async def _conversation_lifecycle(self, conversation_id: UUID): async with self._lifecycle_condition: From f287969d94d273ea4421cba3f63feabad502205a Mon Sep 17 00:00:00 2001 From: neubig Date: Sat, 22 Aug 2026 21:19:37 +0000 Subject: [PATCH 8/8] chore: refine lifecycle lock coverage (#4570) Co-authored-by: openhands --- .../agent_server/conversation_service.py | 15 ++++-------- .../agent_server/test_conversation_service.py | 24 +++++++++++++++++++ 2 files changed, 29 insertions(+), 10 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index eb036a5710..4f9abac659 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -3,7 +3,6 @@ import json import logging import os -import threading from collections.abc import Awaitable, Callable from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager, suppress @@ -655,9 +654,6 @@ class ConversationService: _conversation_locks: WeakValueDictionary[UUID, asyncio.Lock] = field( default_factory=WeakValueDictionary, init=False ) - _catalog_lock_sync: threading.Lock = field( - default_factory=threading.Lock, init=False - ) _conversation_webhook_subscribers: list["ConversationWebhookSubscriber"] = field( default_factory=list, init=False ) @@ -1061,12 +1057,11 @@ def _prepare_persisted_runtime(self, stored: StoredConversation) -> None: ) def _get_conversation_lock(self, conversation_id: UUID) -> asyncio.Lock: - with self._catalog_lock_sync: - lock = self._conversation_locks.get(conversation_id) - if lock is None: - lock = asyncio.Lock() - self._conversation_locks[conversation_id] = lock - return lock + lock = self._conversation_locks.get(conversation_id) + if lock is None: + lock = asyncio.Lock() + self._conversation_locks[conversation_id] = lock + return lock @asynccontextmanager async def _conversation_lifecycle(self, conversation_id: UUID): diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 2b9e07b3d5..e867187cf8 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -590,6 +590,30 @@ async def test_prepare_for_sandbox_pause_drains_active_services(tmp_path): await service.__aexit__(None, None, None) +@pytest.mark.asyncio +async def test_conversation_lifecycle_serializes_only_matching_ids(tmp_path): + service = ConversationService(conversations_dir=tmp_path / "conversations") + first_id = uuid4() + second_id = uuid4() + same_id_entered = asyncio.Event() + other_id_entered = asyncio.Event() + + async def enter_lifecycle(conversation_id: UUID, entered: asyncio.Event): + async with service._conversation_lifecycle(conversation_id): + entered.set() + + async with service._conversation_lifecycle(first_id): + same_id_task = asyncio.create_task(enter_lifecycle(first_id, same_id_entered)) + other_id_task = asyncio.create_task( + enter_lifecycle(second_id, other_id_entered) + ) + await asyncio.wait_for(other_id_entered.wait(), timeout=1) + assert not same_id_entered.is_set() + + await asyncio.gather(same_id_task, other_id_task) + assert same_id_entered.is_set() + + @pytest.mark.asyncio async def test_prepare_for_sandbox_pause_blocks_new_hydration( persisted_conversation,