Avoid async goal cleanup shutdown hangs

This commit is contained in:
Ahmed Ibrahim
2026-06-08 12:52:25 -07:00
parent dab7b5e9b6
commit f50b49dba9
+21 -16
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import threading
from collections.abc import Iterator
from typing import AsyncIterator, Callable, ParamSpec, TypeVar
@@ -53,7 +54,6 @@ class AsyncCodexClient:
def __init__(self, config: CodexConfig | None = None) -> None:
"""Create the wrapped sync client that owns the transport process."""
self._sync = CodexClient(config=config)
self._background_tasks: set[asyncio.Task[None]] = set()
async def __aenter__(self) -> "AsyncCodexClient":
"""Start the Codex process when entering an async context."""
@@ -91,9 +91,6 @@ class AsyncCodexClient:
async def close(self) -> None:
"""Close the wrapped sync client in a worker thread."""
await self._call_sync(self._sync.close)
if self._background_tasks:
await asyncio.gather(*tuple(self._background_tasks), return_exceptions=True)
self._background_tasks.clear()
async def initialize(self) -> InitializeResponse:
"""Initialize the Codex session."""
@@ -252,22 +249,30 @@ class AsyncCodexClient:
return await asyncio.shield(operation)
except asyncio.CancelledError:
async def cleanup_cancelled_start() -> None:
def cleanup_cancelled_start(
completed: asyncio.Task[tuple[_GoalOperationState, str]],
) -> None:
try:
state, _ = await operation
state, _ = completed.result()
except BaseException:
return
try:
await self.pause_goal(thread_id)
except Exception:
pass
finally:
state.finish()
self.unregister_goal_operation(state)
cleanup = asyncio.create_task(cleanup_cancelled_start())
self._background_tasks.add(cleanup)
cleanup.add_done_callback(self._background_tasks.discard)
def stop_cancelled_goal() -> None:
try:
self._sync.pause_goal(thread_id)
except Exception:
pass
finally:
state.finish()
self._sync.unregister_goal_operation(state)
threading.Thread(
target=stop_cancelled_goal,
name="codex-goal-start-cleanup",
daemon=True,
).start()
operation.add_done_callback(cleanup_cancelled_start)
raise
async def turn_start(