perf: replace global _lifecycle_lock with per-conversation locks to eliminate cross-conversation wedges

[neubig]

Actual Behavior

uv run pytest tests/agent_server/stress/test_concurrent_conversations.py reproduces the concurrency regression.

ConversationService guards all conversation lifecycle operations with a single global asyncio.Lock (_lifecycle_lock, conversation_service.py:644). It is acquired in 9 places across the service, and two of those perform slow/blocking work under the lock:

Line	Method	Blocking work under the lock
~1064	_get_or_load_event_service_locked	asyncio.to_thread(_prepare_persisted_runtime) — disk I/O + runtime prep
~1697	delete_conversation	await event_service.close() — can hang indefinitely (see #4546)

Because every conversation operation (create, open, search, delete, event-stream subscribe) acquires the same lock, a stuck or slow operation on conversation A blocks every other conversation — the entire server wedges while /health and metadata routes keep answering, making it look healthy when it isn't.

Why PRs #4513 and #4548 are mitigations, not fixes

• PR test(agent-server): lifecycle lock deadlocks on thread-pool exhaustion #4513 (cache fast-path) only helps the read path for conversations already in the _event_services cache. New conversation creation (line ~1332) and delete/close (line ~1697) still take the global lock and can still hang.
• PR fix(sdk): bound AsyncExecutor.close() so it cannot hang forever #4548 (bounded close()) replaces "hang forever" with "block all conversations for 30s". That's a shorter wedge, but still a wedge — and serial 30s stalls accumulate under load.

Both fix symptoms (one read-path deadlock, one unbounded hang) without addressing the structural issue: a single coarse-grained lock for the entire conversation lifecycle.

Expected Behavior

A stuck or slow operation on one conversation must not block operations on any other conversation. The conversation-lifecycle lock should be per-conversation, not global.

Proposed Solution

Replace the single _lifecycle_lock with per-conversation locks, plus a tiny lock for catalog (dict) mutation only:

# Before (current):
_lifecycle_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)

# After (proposed):
_conversation_locks: dict[UUID, asyncio.Lock] = field(default_factory=dict, init=False)
_catalog_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)  # protects only dict mutation

def _get_conversation_lock(self, conversation_id: UUID) -> asyncio.Lock:
    with self._catalog_lock:  # brief, only protects the dict itself
        if conversation_id not in self._conversation_locks:
            self._conversation_locks[conversation_id] = asyncio.Lock()
        return self._conversation_locks[conversation_id]

Each method acquires only its conversation's lock:

• A stuck close() on conversation A blocks only conversation A
• Conversation B's create/search/open proceeds unimpeded
• The catalog dict (_event_services, _conversation_records) gets its own tiny lock for mutation only

The SDK already has the right pattern: ResourceLockManager (resource_lock_manager.py) does per-resource FIFO locks for tools. The same approach should be applied at the conversation-lifecycle layer.

Acceptance Criteria

• ConversationService uses per-conversation locks instead of a single global _lifecycle_lock
• A slow close() / _prepare_persisted_runtime on conversation A does not block get_event_service / create_conversation / delete_conversation for conversation B
• The catalog dict mutations are still race-free (protected by a separate brief lock)
• A stress test with concurrent create + search + close traffic shows no cross-conversation blocking (p99 < 500ms under 10 concurrent workers)
• All existing conversation_service and event_service tests pass
• PRs test(agent-server): lifecycle lock deadlocks on thread-pool exhaustion #4513 and fix(sdk): bound AsyncExecutor.close() so it cannot hang forever #4548 become redundant (their tests still pass, but the underlying scenario can no longer occur)

Mitigations (in-flight)

• test(agent-server): lifecycle lock deadlocks on thread-pool exhaustion #4513 — cache fast-path (read-path only; does not help create/close)
• fix(sdk): bound AsyncExecutor.close() so it cannot hang forever #4548 — bounded AsyncExecutor.close() timeout (shortens the wedge but doesn't eliminate cross-conversation blocking)

Once this issue is implemented, those PRs can likely be closed as superseded.

This issue body was updated by an AI agent (OpenHands) on behalf of the user to satisfy the repository readiness check.

[github-actions]

✅ ready-for-dev label applied

This issue meets the readiness criteria and is ready for contribution.
A contributor can now open a PR that links this issue (Fixes #<number>).

This is an automated check — no AI was used to generate this comment.

[all-hands-bot]

#4569 shares the same root cause as #4514 (global _lifecycle_lock in ConversationService causing cross-conversation blocking) but proposes a broader structural fix (per-conversation locks) that would supersede #4514's narrower cache-fast-path approach. Both filed by the same author. #4569 explicitly references #4514's fix (PR #4513) as a mitigation, not a fix. Discussion about the locking strategy should be consolidated.

Found 3 possible duplicate issues:

1. #4514 — Lifecycle lock deadlock: thread-pool exhaustion blocks all event loading
2. #4546 — [Bug]: AsyncExecutor.close() can block forever, wedging conversation shutdown
3. #4555 — [Bug]: Agent Server event endpoints hang indefinitely when LocalConversation cleanup blocks

These may not be exact duplicates, but the scope appears to overlap enough that keeping discussion in one place may be more useful.

This comment was created by an AI assistant (OpenHands) on behalf of the repository maintainer.