← back to Exo
Runner error handling (#2093)
14aab356880b84ab2033a14f5e687a591f49742d · 2026-05-15 13:40:59 +0100 · Andrei Cravtov
# Runner error handling
## Motivation
Runner failures were mostly surfaced as plain shutdown messages, which
made root cause hard to spot from API errors or runner status.
This adds a MVP path for preserving runner crash context and attaching
known stderr diagnostics to failure reports.
## Changes
- Added `RunnerTerminationError` for Python exceptions raised inside
runner bootstrap
- Changed runner bootstrap to send `Event | RunnerTerminationError` over
the private runner channel
- Moved public `RunnerFailed` emission back into supervisor
- Added stderr-only `RunnerDiagnosticCollector`
- + Added known diagnostics for Metal GPU timeout, ring socket receive
errno, and ring transport abort
- Added diagnostics to `RunnerFailed` and `ErrorChunk`
- Tweaked async process termination to join briefly before
terminate/kill
- Updated tests/fixtures for new failure payload shape
- Added Ruff VS Code formatter settings
## Why It Works
Runner child now reports raw-ish failure context to supervisor instead
of publishing failed status directly.
Supervisor still owns process lifecycle, exit code/signal handling,
in-flight task error chunks, and final runner status. Stderr diagnostics
stay best effort and only known root-cause variants are surfaced.
## Test Plan
### Manual Testing
Hardware: remote runner logs from e16/e11/e4/e2
What you did:
- inspected live runner stderr logs
- used observed Metal GPU timeout and ring socket errors as initial
diagnostic targets
### Automated Testing
- `nix flake check`
- supervisor test covers error chunk + failed status emission
- plan lifecycle test updated for failed runner diagnostics
- type/lint checks cover new runner channel union
---------
Co-authored-by: Evan Quiney <evanev7@gmail.com>
Files touched
M .gitignoreM .vscode/extensions.jsonM .vscode/settings.jsonM pyproject.tomlM src/exo/shared/types/chunks.pyM src/exo/shared/types/worker/runners.pyM src/exo/utils/async_process.pyM src/exo/utils/tests/test_async_process.pyM src/exo/worker/runner/bootstrap.pyA src/exo/worker/runner/diagnostics.pyM src/exo/worker/runner/supervisor.pyM src/exo/worker/tests/unittests/test_plan/test_runner_lifecycle.pyM src/exo/worker/tests/unittests/test_runner/test_runner_supervisor.py
Diff
commit 14aab356880b84ab2033a14f5e687a591f49742d
Author: Andrei Cravtov <the.andrei.cravtov@gmail.com>
Date: Fri May 15 13:40:59 2026 +0100
Runner error handling (#2093)
# Runner error handling
## Motivation
Runner failures were mostly surfaced as plain shutdown messages, which
made root cause hard to spot from API errors or runner status.
This adds a MVP path for preserving runner crash context and attaching
known stderr diagnostics to failure reports.
## Changes
- Added `RunnerTerminationError` for Python exceptions raised inside
runner bootstrap
- Changed runner bootstrap to send `Event | RunnerTerminationError` over
the private runner channel
- Moved public `RunnerFailed` emission back into supervisor
- Added stderr-only `RunnerDiagnosticCollector`
- + Added known diagnostics for Metal GPU timeout, ring socket receive
errno, and ring transport abort
- Added diagnostics to `RunnerFailed` and `ErrorChunk`
- Tweaked async process termination to join briefly before
terminate/kill
- Updated tests/fixtures for new failure payload shape
- Added Ruff VS Code formatter settings
## Why It Works
Runner child now reports raw-ish failure context to supervisor instead
of publishing failed status directly.
Supervisor still owns process lifecycle, exit code/signal handling,
in-flight task error chunks, and final runner status. Stderr diagnostics
stay best effort and only known root-cause variants are surfaced.
## Test Plan
### Manual Testing
Hardware: remote runner logs from e16/e11/e4/e2
What you did:
- inspected live runner stderr logs
- used observed Metal GPU timeout and ring socket errors as initial
diagnostic targets
### Automated Testing
- `nix flake check`
- supervisor test covers error chunk + failed status emission
- plan lifecycle test updated for failed runner diagnostics
- type/lint checks cover new runner channel union
---------
Co-authored-by: Evan Quiney <evanev7@gmail.com>
---
.gitignore | 2 +-
.vscode/extensions.json | 3 +-
.vscode/settings.json | 3 +
pyproject.toml | 2 +-
src/exo/shared/types/chunks.py | 5 +
src/exo/shared/types/worker/runners.py | 2 +
src/exo/utils/async_process.py | 14 +-
src/exo/utils/tests/test_async_process.py | 102 +--------------
src/exo/worker/runner/bootstrap.py | 48 +++++--
src/exo/worker/runner/diagnostics.py | 144 +++++++++++++++++++++
src/exo/worker/runner/supervisor.py | 71 +++++++---
.../unittests/test_plan/test_runner_lifecycle.py | 2 +-
.../test_runner/test_runner_supervisor.py | 3 +-
13 files changed, 258 insertions(+), 143 deletions(-)
diff --git a/.gitignore b/.gitignore
index a73d27af..fa09fb01 100644
--- a/.gitignore
+++ b/.gitignore
@@ -18,7 +18,6 @@ digest.txt
app/EXO/build/
dist/
-
# rust
target/
**/*.rs.bk
@@ -41,3 +40,4 @@ tmp/models
/build/exo
/.claude/skills
/.claude
+/.codex
diff --git a/.vscode/extensions.json b/.vscode/extensions.json
index b2e0978e..7b7d7b21 100644
--- a/.vscode/extensions.json
+++ b/.vscode/extensions.json
@@ -2,7 +2,8 @@
"recommendations": [
"detachhead.basedpyright",
"ms-python.python",
- "jnoortheen.nix-ide"
+ "jnoortheen.nix-ide",
+ "charliermarsh.ruff",
],
"unwantedRecommendations": [
"ms-python.vscode-pylance",
diff --git a/.vscode/settings.json b/.vscode/settings.json
index 19dfab4d..37c85975 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -26,6 +26,9 @@
"editor.defaultFormatter": "jnoortheen.nix-ide"
},
+ "[python]": {
+ "editor.defaultFormatter": "charliermarsh.ruff",
+ },
"python.defaultInterpreterPath": "${workspaceFolder}/.venv/bin/python",
"basedpyright.analysis.configFilePath": "${workspaceFolder}/pyproject.toml",
"basedpyright.importStrategy": "fromEnvironment",
diff --git a/pyproject.toml b/pyproject.toml
index 4fef8eea..570aba3b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -254,5 +254,5 @@ pythonpath = "."
asyncio_mode = "auto"
markers = ["slow: marks tests as slow (deselected by default)"]
env = ["EXO_TESTS=1"]
-addopts = "-m 'not slow' --ignore=tests"
+addopts = "-m 'not slow' --ignore=tests --ignore=tmp"
filterwarnings = ["ignore:builtin type Swig:DeprecationWarning"]
diff --git a/src/exo/shared/types/chunks.py b/src/exo/shared/types/chunks.py
index 2e51c076..82425d9f 100644
--- a/src/exo/shared/types/chunks.py
+++ b/src/exo/shared/types/chunks.py
@@ -11,6 +11,7 @@ from exo.api.types import (
)
from exo.shared.models.model_cards import ModelId
from exo.utils.pydantic_ext import TaggedModel
+from exo.worker.runner.diagnostics import KnownRunnerDiagnostic
from .common import CommandId
@@ -34,6 +35,10 @@ class ErrorChunk(BaseChunk):
error_message: str
finish_reason: Literal["error"] = "error"
+ # NOTE: this is a bad place to put this, creates semantic overlap/confusion;
+ # at some point someone put this somewhere else, thanks :)
+ diagnostics: list[KnownRunnerDiagnostic] = []
+
class ToolCallChunk(BaseChunk):
tool_calls: list[ToolCallItem]
diff --git a/src/exo/shared/types/worker/runners.py b/src/exo/shared/types/worker/runners.py
index c906b6ec..c9ff3b5c 100644
--- a/src/exo/shared/types/worker/runners.py
+++ b/src/exo/shared/types/worker/runners.py
@@ -6,6 +6,7 @@ from exo.shared.models.model_cards import ModelId
from exo.shared.types.common import Id, NodeId
from exo.shared.types.worker.shards import ShardMetadata
from exo.utils.pydantic_ext import FrozenModel, TaggedModel
+from exo.worker.runner.diagnostics import KnownRunnerDiagnostic
class RunnerId(Id):
@@ -64,6 +65,7 @@ class RunnerShutdown(BaseRunnerStatus):
class RunnerFailed(BaseRunnerStatus):
error_message: str | None = None
+ diagnostics: list[KnownRunnerDiagnostic]
RunnerStatus = (
diff --git a/src/exo/utils/async_process.py b/src/exo/utils/async_process.py
index 3866037d..f7633374 100644
--- a/src/exo/utils/async_process.py
+++ b/src/exo/utils/async_process.py
@@ -19,6 +19,7 @@ from anyio import (
create_task_group,
move_on_after,
sleep,
+ to_thread,
wait_readable,
)
from anyio.abc import TaskStatus
@@ -29,10 +30,11 @@ from exo.utils.channels import Receiver, Sender, channel
_STDOUT_FD = 1
_STDERR_FD = 2
_READ_CHUNK_SIZE = 64 * 1024
-_TERMINATE_GRACE_SECONDS = 10.0
+_JOIN_GRACE_SECONDS = 3.0
+_TERMINATE_GRACE_SECONDS = 5.0
_TERMINATE_RETRY_GRACE_SECONDS = 2.0
_TERMINATE_ATTEMPTS = 10
-_KILL_GRACE_SECONDS = 5.0
+_KILL_GRACE_SECONDS = 2.0
@final
@@ -204,14 +206,12 @@ class AsyncProcess:
async def _terminate_if_still_alive(self) -> None:
process = self._process
- if process is None:
- return
-
- if self.exitcode is not None:
+ if process is None or self.exitcode is not None:
return
with contextlib.suppress(ValueError):
- if not process.is_alive():
+ await to_thread.run_sync(process.join, _JOIN_GRACE_SECONDS)
+ if self.exitcode is not None or not process.is_alive():
return
logger.warning("Child process didn't shut down successfully, terminating")
diff --git a/src/exo/utils/tests/test_async_process.py b/src/exo/utils/tests/test_async_process.py
index 0e275cfc..07e567c1 100644
--- a/src/exo/utils/tests/test_async_process.py
+++ b/src/exo/utils/tests/test_async_process.py
@@ -2,17 +2,12 @@ import contextlib
import os
import signal
import sys
-import time
from collections.abc import AsyncIterator, Callable
-from types import FrameType
-import mlx.core as mx
import pytest
from _pytest.capture import CaptureFixture
from anyio import EndOfStream, create_task_group, fail_after
-from pytest import MonkeyPatch
-import exo.utils.async_process as async_process
from exo.utils.async_process import (
AsyncProcess,
)
@@ -66,43 +61,6 @@ def _close_stdio_and_exit() -> None:
os._exit(0)
-def _exit_on_sigterm(exitcode: int) -> None:
- def handle_sigterm(_signum: int, _frame: FrameType | None) -> None:
- os._exit(exitcode)
-
- signal.signal(signal.SIGTERM, handle_sigterm)
- os.write(1, b"sigterm-ready\n")
- while True:
- time.sleep(0.1)
-
-
-def _exit_after_repeated_sigterm(required_count: int, exitcode: int) -> None:
- sigterm_count = 0
-
- def handle_sigterm(_signum: int, _frame: FrameType | None) -> None:
- nonlocal sigterm_count
- sigterm_count += 1
- if sigterm_count >= required_count:
- os._exit(exitcode)
-
- signal.signal(signal.SIGTERM, handle_sigterm)
- os.write(1, b"sigterm-ready\n")
- while True:
- time.sleep(0.1)
-
-
-def _ignore_sigterm_forever() -> None:
- signal.signal(signal.SIGTERM, signal.SIG_IGN)
- os.write(1, b"sigterm-ready\n")
- while True:
- time.sleep(0.1)
-
-
-def _sleep_forever() -> None:
- while True:
- time.sleep(0.1)
-
-
def _send_over_mp_channel(send: MpSender[str]) -> None:
send.send("hello from child")
send.close()
@@ -112,6 +70,8 @@ def _mlx_force_oom(size: int = 40_000) -> None:
"""
Force an Out-Of-Memory (OOM) error in MLX by performing large tensor operations.
"""
+ import mlx.core as mx
+
print("CHILD: start")
mx.set_default_device(mx.gpu)
@@ -274,19 +234,6 @@ async def test_process_run_is_one_shot() -> None:
await process.run()
-@pytest.mark.anyio
-async def test_process_started_with_task_group_start_can_stop_immediately() -> None:
- process = AsyncProcess(_sleep_forever)
-
- async with create_task_group() as task_group:
- await task_group.start(process.run)
- assert process.is_alive()
- with fail_after(2):
- await process.stop()
-
- assert not process.is_alive()
-
-
@pytest.mark.anyio
async def test_stdout_receiver_yields_bytes_chunks() -> None:
process = AsyncProcess(_write_large_output)
@@ -440,51 +387,6 @@ async def test_repeated_crashing_children_do_not_grow_parent_fd_table() -> None:
assert after <= before + 2
-@pytest.mark.anyio
-async def test_stop_allows_child_to_exit_after_sigterm() -> None:
- process = AsyncProcess(_exit_on_sigterm, args=(43,))
-
- async with _started_process(process):
- assert await process.stdout.receive() == b"sigterm-ready\n"
-
- with fail_after(2):
- await process.stop()
-
- assert process.exitcode == 43
-
-
-@pytest.mark.anyio
-async def test_stop_retries_sigterm_before_sigkill(monkeypatch: MonkeyPatch) -> None:
- monkeypatch.setattr(async_process, "_TERMINATE_GRACE_SECONDS", 0.01)
- monkeypatch.setattr(async_process, "_TERMINATE_RETRY_GRACE_SECONDS", 0.01)
- process = AsyncProcess(_exit_after_repeated_sigterm, args=(3, 44))
-
- async with _started_process(process):
- assert await process.stdout.receive() == b"sigterm-ready\n"
-
- with fail_after(2):
- await process.stop()
-
- assert process.exitcode == 44
-
-
-@pytest.mark.anyio
-async def test_stop_escalates_to_sigkill_when_child_ignores_sigterm(
- monkeypatch: MonkeyPatch,
-) -> None:
- monkeypatch.setattr(async_process, "_TERMINATE_GRACE_SECONDS", 0.1)
- monkeypatch.setattr(async_process, "_TERMINATE_RETRY_GRACE_SECONDS", 0.01)
- process = AsyncProcess(_ignore_sigterm_forever)
-
- async with _started_process(process):
- assert await process.stdout.receive() == b"sigterm-ready\n"
-
- with fail_after(3):
- await process.stop()
-
- assert process.exitcode == -signal.SIGKILL
-
-
@pytest.mark.anyio
async def test_process_can_use_mp_channel_with_global_spawn_context() -> None:
send, recv = mp_channel[str]()
diff --git a/src/exo/worker/runner/bootstrap.py b/src/exo/worker/runner/bootstrap.py
index e9041180..f981667c 100644
--- a/src/exo/worker/runner/bootstrap.py
+++ b/src/exo/worker/runner/bootstrap.py
@@ -1,21 +1,45 @@
import os
import resource
+import traceback
+from dataclasses import dataclass
+from typing import Self, cast
import loguru
-from exo.shared.types.events import Event, RunnerStatusUpdated
+from exo.shared.types.events import Event
from exo.shared.types.tasks import Task, TaskId
from exo.shared.types.worker.instances import BoundInstance
-from exo.shared.types.worker.runners import RunnerFailed
from exo.utils.channels import ClosedResourceError, MpReceiver, MpSender
from exo.worker.engines.base import Builder
logger: "loguru.Logger" = loguru.logger
+@dataclass(frozen=True)
+class RunnerTerminationError:
+ exception_type: str
+ exception_message: str
+ exception_repr: str
+ traceback: str
+
+ @classmethod
+ def from_exception(cls, e: Exception) -> Self:
+ return cls(
+ exception_type=type(e).__qualname__,
+ exception_message=str(e),
+ exception_repr=repr(e),
+ traceback="".join(
+ traceback.TracebackException.from_exception(e).format(chain=True)
+ ),
+ )
+
+ def __str__(self) -> str:
+ return f"{self.exception_type}: {self.exception_message}\n{self.traceback}"
+
+
def entrypoint(
bound_instance: BoundInstance,
- event_sender: MpSender[Event],
+ event_sender: MpSender[Event | RunnerTerminationError],
task_receiver: MpReceiver[Task],
cancel_receiver: MpReceiver[TaskId],
_logger: "loguru.Logger",
@@ -36,15 +60,16 @@ def entrypoint(
# Import main after setting global logger - this lets us just import logger from this module
try:
+ event_sender_downcast: MpSender[Event] = cast(MpSender[Event], event_sender)
+
from exo.worker.runner.runner import Runner
builder: Builder
-
if bound_instance.is_image_model:
from exo.worker.engines.image.builder import MfluxBuilder
builder = MfluxBuilder(
- event_sender, cancel_receiver, bound_instance.bound_shard
+ event_sender_downcast, cancel_receiver, bound_instance.bound_shard
)
else:
from exo.worker.engines.mlx.patches import apply_mlx_patches
@@ -56,25 +81,20 @@ def entrypoint(
# evil sharing of the event sender
builder = MlxBuilder(
model_id=bound_instance.bound_shard.model_card.model_id,
- event_sender=event_sender,
+ event_sender=event_sender_downcast,
cancel_receiver=cancel_receiver,
)
- runner = Runner(bound_instance, builder, event_sender, task_receiver)
+ runner = Runner(bound_instance, builder, event_sender_downcast, task_receiver)
runner.main()
-
except ClosedResourceError:
logger.warning("Runner communication closed unexpectedly")
except Exception as e:
logger.opt(exception=e).warning(
f"Runner {bound_instance.bound_runner_id} crashed with critical exception {e}"
)
- event_sender.send(
- RunnerStatusUpdated(
- runner_id=bound_instance.bound_runner_id,
- runner_status=RunnerFailed(error_message=str(e)),
- )
- )
+ event_sender.send(RunnerTerminationError.from_exception(e))
+ raise SystemExit(1) from e
finally:
try:
event_sender.close()
diff --git a/src/exo/worker/runner/diagnostics.py b/src/exo/worker/runner/diagnostics.py
new file mode 100644
index 00000000..e106c66a
--- /dev/null
+++ b/src/exo/worker/runner/diagnostics.py
@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import errno
+import os
+import re
+from collections import deque
+from typing import final
+
+from exo.utils.pydantic_ext import TaggedModel
+
+_EVIDENCE_LINES = 4
+
+_METAL_GPU_TIMEOUT_RE = re.compile(
+ r"^\s*(?:libc\+\+abi:.*std::runtime_error:\s*)?"
+ r"(?P<message>\[METAL\].*GPU\s+Timeout.*)\s*$",
+ re.IGNORECASE,
+)
+_RING_SOCKET_ERRNO_RE = re.compile(
+ r"^\s*\[ring\]\s+Receiving\s+from\s+socket\s+\d+\s+failed\s+with\s+errno\s+"
+ r"(?P<error_number>\d+)\s*$",
+ re.IGNORECASE,
+)
+_RING_TRANSPORT_ABORT_RE = re.compile(
+ r"^\s*\[ring\]\s+Too\s+many\s+send/recv\s+errors\.\s+Aborting\.\.\.\s*$",
+ re.IGNORECASE,
+)
+
+
+class BaseRunnerDiagnostic(TaggedModel):
+ message: str
+ evidence: tuple[str, ...] = ()
+
+
+class RunnerMetalGpuTimeout(BaseRunnerDiagnostic):
+ pass
+
+
+class RunnerRingTransportError(BaseRunnerDiagnostic):
+ pass
+
+
+class RunnerRingSocketReceivingError(BaseRunnerDiagnostic):
+ error_number: int
+ error_name: str
+ error_description: str
+
+
+class RunnerUnknown(BaseRunnerDiagnostic):
+ pass
+
+
+KnownRunnerDiagnostic = (
+ RunnerMetalGpuTimeout | RunnerRingTransportError | RunnerRingSocketReceivingError
+)
+
+RunnerDiagnostic = KnownRunnerDiagnostic | RunnerUnknown
+
+
+@final
+class RunnerDiagnosticCollector:
+ def __init__(self) -> None:
+ self._stderr_tail: deque[str] = deque(maxlen=_EVIDENCE_LINES)
+ self._diagnostics: list[RunnerDiagnostic] = []
+
+ def record_line(self, line: str) -> None:
+ if not line or line.isspace():
+ return
+
+ self._stderr_tail.append(line)
+ evidence = tuple(self._stderr_tail)
+ diagnostic = self._classify_line(line, evidence) or RunnerUnknown(
+ message="Unclassified runner stderr line",
+ evidence=evidence,
+ )
+
+ # TODO: Eventually this will become a stateful parser with a more advanced architecture,
+ # right now the statefulness is restricted to bespoke handling of specific errors
+ #
+ # `RunnerRingSocketReceivingError` usually happens a few times before `RunnerRingTransportError`
+ # therefore we deduplicate and only keep last `RunnerRingSocketReceivingError`
+ if len(self._diagnostics) > 0 and (
+ isinstance(self._diagnostics[-1], RunnerRingSocketReceivingError)
+ and isinstance(diagnostic, RunnerRingSocketReceivingError)
+ ):
+ self._diagnostics[-1] = diagnostic
+ return
+
+ self._diagnostics.append(diagnostic)
+
+ def diagnostics(self) -> tuple[RunnerDiagnostic, ...]:
+ return tuple(self._diagnostics)
+
+ def _classify_line(
+ self, line: str, evidence: tuple[str, ...]
+ ) -> KnownRunnerDiagnostic | None:
+ if metal_error := _parse_metal_gpu_timeout(line, evidence):
+ return metal_error
+
+ if socket_error := _parse_ring_socket_error(line, evidence):
+ return socket_error
+
+ if _RING_TRANSPORT_ABORT_RE.match(line):
+ return RunnerRingTransportError(
+ message="Ring transport aborted after too many send/recv errors",
+ evidence=evidence,
+ )
+
+ return None
+
+
+def _parse_metal_gpu_timeout(
+ line: str, evidence: tuple[str, ...]
+) -> RunnerMetalGpuTimeout | None:
+ match = _METAL_GPU_TIMEOUT_RE.match(line)
+ if match is None:
+ return None
+
+ return RunnerMetalGpuTimeout(
+ message=f"Metal GPU timeout: {match.group('message')}",
+ evidence=evidence,
+ )
+
+
+def _parse_ring_socket_error(
+ line: str, evidence: tuple[str, ...]
+) -> RunnerRingSocketReceivingError | None:
+ match = _RING_SOCKET_ERRNO_RE.match(line)
+ if match is None:
+ return None
+
+ error_number = int(match.group("error_number"))
+ error_name = errno.errorcode.get(error_number, "UNKNOWN_ERRNO")
+ error_description = os.strerror(error_number)
+
+ return RunnerRingSocketReceivingError(
+ error_number=error_number,
+ error_name=error_name,
+ error_description=error_description,
+ evidence=evidence,
+ message=(
+ f"Ring socket receive failed: errno {error_number} "
+ f"{error_name} ({error_description})"
+ ),
+ )
diff --git a/src/exo/worker/runner/supervisor.py b/src/exo/worker/runner/supervisor.py
index 7e4ee9ae..96112624 100644
--- a/src/exo/worker/runner/supervisor.py
+++ b/src/exo/worker/runner/supervisor.py
@@ -48,7 +48,11 @@ from exo.utils.async_process import AsyncProcess
from exo.utils.channels import MpReceiver, MpSender, Receiver, Sender, mp_channel
from exo.utils.fs import ensure_parent_directory_exists
from exo.utils.task_group import TaskGroup
-from exo.worker.runner.bootstrap import entrypoint
+from exo.worker.runner.bootstrap import RunnerTerminationError, entrypoint
+from exo.worker.runner.diagnostics import (
+ RunnerDiagnosticCollector,
+ RunnerUnknown,
+)
PREFILL_TIMEOUT_SECONDS = 60
DECODE_TIMEOUT_SECONDS = 5
@@ -60,6 +64,9 @@ class RunnerStdioHandler:
_stderr_rx: Receiver[bytes]
_stdout_log: AsyncFile[str]
_stderr_log: AsyncFile[str]
+ diagnostics: RunnerDiagnosticCollector = field(
+ default_factory=RunnerDiagnosticCollector
+ )
_tg: TaskGroup = field(default_factory=TaskGroup, init=False)
@@ -98,12 +105,14 @@ class RunnerStdioHandler:
self._stdout_rx,
self._stdout_log,
lambda line: logger.info(f"Runner stdout: {line}"), # pyright: ignore[reportUnknownLambdaType]
+ lambda _: None, # pyright: ignore[reportUnknownLambdaType]
)
tg.start_soon( # pyright: ignore[reportUnknownArgumentType]
self._handle_runner_output,
self._stderr_rx,
self._stderr_log,
lambda line: logger.warning(f"Runner stderr: {line}"), # pyright: ignore[reportUnknownLambdaType]
+ self.diagnostics.record_line,
)
finally:
with CancelScope(shield=True):
@@ -115,12 +124,12 @@ class RunnerStdioHandler:
rx: Receiver[bytes],
logfile: AsyncFile[str],
log_line: Callable[[str], None],
+ record_diagnostic_line: Callable[[str], None],
):
- # TODO: right now it logs them as warnings, but in the future they should be split
- # into being logged AND a seperate task which tries to best-effort figure out cause
- # of error and package into error enum, which then is used by rest of app to act on it;
- # inferring what the error is would be done by pattern-matching in the text for things
- # e.g. certain VLLM error codes and so on
+ # The diagnostic collector is deliberately line-level for now. It records
+ # bounded stderr context and known failure anchors; the supervisor
+ # correlates those hints with the runner exit status before surfacing an
+ # error.
# not using TextReceiveStream because it doesn't do final=True handling on errors
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
@@ -134,7 +143,7 @@ class RunnerStdioHandler:
# Send to logger & error recovery task
log_line(line)
- # TODO: error recovery task
+ record_diagnostic_line(line)
async def handle_text(text: str):
nonlocal pending_line
@@ -176,7 +185,7 @@ class RunnerSupervisor:
runner_process: AsyncProcess
_runner_stdio_handler: RunnerStdioHandler
initialize_timeout: float
- _ev_recv: MpReceiver[Event]
+ _ev_recv: MpReceiver[Event | RunnerTerminationError]
_task_sender: MpSender[Task]
_event_sender: Sender[Event]
_cancel_sender: MpSender[TaskId]
@@ -198,7 +207,7 @@ class RunnerSupervisor:
event_sender: Sender[Event],
initialize_timeout: float = 400,
) -> Self:
- ev_send, ev_recv = mp_channel[Event]()
+ ev_send, ev_recv = mp_channel[Event | RunnerTerminationError]()
task_sender, task_recv = mp_channel[Task]()
cancel_sender, cancel_recv = mp_channel[TaskId]()
@@ -311,6 +320,10 @@ class RunnerSupervisor:
try:
with self._ev_recv as events:
async for event in events:
+ if isinstance(event, RunnerTerminationError):
+ # try to get exception if possible
+ await self._check_runner(event)
+ break
if isinstance(event, RunnerStatusUpdated):
self.status = event.runner_status
if isinstance(event, TaskAcknowledged):
@@ -334,8 +347,9 @@ class RunnerSupervisor:
self.in_progress.pop(event.task_id, None)
self.completed.add(event.task_id)
await self._event_sender.send(event)
- except (ClosedResourceError, BrokenResourceError) as e:
- await self._check_runner(e)
+ except (ClosedResourceError, BrokenResourceError):
+ # this is the happy path shutdown - we don't need to spam log with it
+ await self._check_runner()
finally:
for tid in self.pending:
self.pending[tid].set()
@@ -347,7 +361,9 @@ class RunnerSupervisor:
if not self.runner_process.is_alive():
await self._check_runner(RuntimeError("Runner found to be dead"))
- async def _check_runner(self, e: Exception) -> None:
+ async def _check_runner(
+ self, e: RunnerTerminationError | Exception | None = None
+ ) -> None:
if not self._cancel_watch_runner.cancel_called:
self._cancel_watch_runner.cancel()
logger.info("Checking runner's status")
@@ -357,20 +373,38 @@ class RunnerSupervisor:
await self.runner_process.stop()
rc = self.runner_process.exitcode
logger.info(f"Runner exited with exit code {rc}")
+
+ # If exit code is 0 then the transient errors were recoverable, meaning we don't need runner diagnostics
if rc == 0:
return
if isinstance(rc, int) and rc < 0:
sig = -rc
try:
- cause = f"signal={sig} ({signal.strsignal(sig)})"
+ if (description := signal.strsignal(sig)) is not None:
+ cause = f"signal={sig} ({description})"
+ else:
+ cause = f"signal={sig}"
except Exception:
cause = f"signal={sig}"
else:
- cause = f"exitcode={rc}"
-
- logger.opt(exception=e).error(f"Runner terminated with {cause}")
+ cause: str = f"exitcode={rc}"
+
+ if e is not None:
+ # Record how runner shut down, try exception, resort to RunnerTerminationError fallback
+ if isinstance(e, Exception):
+ logger.opt(exception=e).error(f"Runner terminated with {cause}")
+ else:
+ cause = f"{cause}\nRunner error: {e}"
+ logger.error(f"Runner terminated with {cause}")
+ else:
+ logger.error(f"Runner terminated with {cause}")
+ diagnostics = [
+ d
+ for d in self._runner_stdio_handler.diagnostics.diagnostics()
+ if not isinstance(d, RunnerUnknown)
+ ]
for task in self.in_progress.values():
if isinstance(task, (TextGeneration, ImageGeneration, ImageEdits)):
with anyio.CancelScope(shield=True):
@@ -379,6 +413,7 @@ class RunnerSupervisor:
command_id=task.command_id,
chunk=ErrorChunk(
model=self.shard_metadata.model_card.model_id,
+ diagnostics=diagnostics,
error_message=(
"Runner shutdown before completing command "
f"({cause})"
@@ -388,7 +423,9 @@ class RunnerSupervisor:
)
try:
- self.status = RunnerFailed(error_message=f"Terminated ({cause})")
+ self.status = RunnerFailed(
+ error_message=f"Terminated ({cause})", diagnostics=diagnostics
+ )
with anyio.CancelScope(shield=True):
await self._event_sender.send(
RunnerStatusUpdated(
diff --git a/src/exo/worker/tests/unittests/test_plan/test_runner_lifecycle.py b/src/exo/worker/tests/unittests/test_plan/test_runner_lifecycle.py
index bfde8a1d..743a4329 100644
--- a/src/exo/worker/tests/unittests/test_plan/test_runner_lifecycle.py
+++ b/src/exo/worker/tests/unittests/test_plan/test_runner_lifecycle.py
@@ -86,7 +86,7 @@ def test_plan_kills_runner_when_sibling_failed():
instances = {INSTANCE_1_ID: instance}
all_runners = {
RUNNER_1_ID: RunnerReady(),
- RUNNER_2_ID: RunnerFailed(error_message="boom"),
+ RUNNER_2_ID: RunnerFailed(error_message="boom", diagnostics=[]),
}
result = plan_mod.plan(
diff --git a/src/exo/worker/tests/unittests/test_runner/test_runner_supervisor.py b/src/exo/worker/tests/unittests/test_runner/test_runner_supervisor.py
index 2845ea01..87cb9c74 100644
--- a/src/exo/worker/tests/unittests/test_runner/test_runner_supervisor.py
+++ b/src/exo/worker/tests/unittests/test_runner/test_runner_supervisor.py
@@ -17,6 +17,7 @@ from exo.shared.types.worker.instances import BoundInstance, InstanceId
from exo.shared.types.worker.runners import RunnerFailed, RunnerId
from exo.utils.async_process import AsyncProcess
from exo.utils.channels import channel, mp_channel
+from exo.worker.runner.bootstrap import RunnerTerminationError
from exo.worker.runner.supervisor import RunnerStdioHandler, RunnerSupervisor
from exo.worker.tests.unittests.conftest import get_bound_mlx_ring_instance
@@ -39,7 +40,7 @@ async def test_check_runner_emits_error_chunk_for_inflight_text_generation() ->
event_sender, event_receiver = channel[Event]()
task_sender, _ = mp_channel[Task]()
cancel_sender, _ = mp_channel[TaskId]()
- _, ev_recv = mp_channel[Event]()
+ _, ev_recv = mp_channel[Event | RunnerTerminationError]()
bound_instance: BoundInstance = get_bound_mlx_ring_instance(
instance_id=InstanceId("instance-a"),
← 88d46d46 fix: omit null delta fields in streaming chat completions (i
·
back to Exo
·
Add node backends to model cards (#2071) bc6661e6 →