Swallow sandbox container races in the stream consumer (#552)
This commit is contained in:
@@ -42,10 +42,14 @@ class AgentCoordinator:
|
|||||||
self.runtimes: dict[str, AgentRuntime] = {}
|
self.runtimes: dict[str, AgentRuntime] = {}
|
||||||
self._lock = asyncio.Lock()
|
self._lock = asyncio.Lock()
|
||||||
self._snapshot_path: Path | None = None
|
self._snapshot_path: Path | None = None
|
||||||
|
self.is_shutting_down = False
|
||||||
|
|
||||||
def set_snapshot_path(self, path: Path) -> None:
|
def set_snapshot_path(self, path: Path) -> None:
|
||||||
self._snapshot_path = path
|
self._snapshot_path = path
|
||||||
|
|
||||||
|
def mark_shutting_down(self) -> None:
|
||||||
|
self.is_shutting_down = True
|
||||||
|
|
||||||
async def register(
|
async def register(
|
||||||
self,
|
self,
|
||||||
agent_id: str,
|
agent_id: str,
|
||||||
|
|||||||
+13
-3
@@ -11,6 +11,8 @@ from typing import TYPE_CHECKING, Any, cast
|
|||||||
|
|
||||||
from agents import RunConfig, Runner
|
from agents import RunConfig, Runner
|
||||||
from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError
|
from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError
|
||||||
|
from agents.sandbox.errors import ExecTransportError
|
||||||
|
from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore]
|
||||||
from openai import APIError
|
from openai import APIError
|
||||||
|
|
||||||
from strix.core.inputs import child_initial_input
|
from strix.core.inputs import child_initial_input
|
||||||
@@ -320,7 +322,7 @@ async def _run_noninteractive_until_lifecycle(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
async def _run_cycle( # noqa: PLR0912
|
async def _run_cycle( # noqa: PLR0912, PLR0915
|
||||||
agent: Any,
|
agent: Any,
|
||||||
coordinator: AgentCoordinator,
|
coordinator: AgentCoordinator,
|
||||||
agent_id: str,
|
agent_id: str,
|
||||||
@@ -356,6 +358,8 @@ async def _run_cycle( # noqa: PLR0912
|
|||||||
event_sink(agent_id, event)
|
event_sink(agent_id, event)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("stream event sink failed for %s", agent_id)
|
logger.exception("stream event sink failed for %s", agent_id)
|
||||||
|
if stream.run_loop_exception is not None:
|
||||||
|
raise stream.run_loop_exception
|
||||||
except RuntimeError as stream_exc:
|
except RuntimeError as stream_exc:
|
||||||
if "after shutdown" not in str(stream_exc):
|
if "after shutdown" not in str(stream_exc):
|
||||||
raise
|
raise
|
||||||
@@ -363,8 +367,14 @@ async def _run_cycle( # noqa: PLR0912
|
|||||||
"Ignoring LiteLLM end-of-stream shutdown race for %s",
|
"Ignoring LiteLLM end-of-stream shutdown race for %s",
|
||||||
agent_id,
|
agent_id,
|
||||||
)
|
)
|
||||||
if stream.run_loop_exception is not None:
|
except (ExecTransportError, docker_errors.NotFound):
|
||||||
raise stream.run_loop_exception
|
if not coordinator.is_shutting_down:
|
||||||
|
raise
|
||||||
|
logger.warning(
|
||||||
|
"Ignoring sandbox container error during teardown for %s",
|
||||||
|
agent_id,
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
finally:
|
finally:
|
||||||
await coordinator.detach_stream(agent_id, stream)
|
await coordinator.detach_stream(agent_id, stream)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
|
|||||||
@@ -1715,6 +1715,7 @@ class StrixTUIApp(App): # type: ignore[misc]
|
|||||||
self.exit()
|
self.exit()
|
||||||
|
|
||||||
def _fire_sandbox_cleanup(self) -> None:
|
def _fire_sandbox_cleanup(self) -> None:
|
||||||
|
self.coordinator.mark_shutting_down()
|
||||||
loop = self._scan_loop
|
loop = self._scan_loop
|
||||||
if loop is None or loop.is_closed():
|
if loop is None or loop.is_closed():
|
||||||
return
|
return
|
||||||
|
|||||||
Reference in New Issue
Block a user