diff --git a/backend/tests/test_runner.py b/backend/tests/test_runner.py
new file mode 100644
index 0000000..2c90e72
--- /dev/null
+++ b/backend/tests/test_runner.py
@@ -0,0 +1,384 @@
+"""Unit tests for services/agent/runner.py — run_agent's reachable
+orchestration state machine (spawn -> run -> complete/error/cancel) and its
+interplay with the task registry in services/agent/tasks.py.
+
+`_drive_provider` (the LLM/provider boundary) is faked throughout — its own
+internals are already covered in depth by test_drive_provider.py. What's
+under test here is everything *around* that call: the state_change/error/
+timeout/abort events run_agent publishes, whether it re-raises or swallows
+each exception class, and whether the task registry (register_task /
+is_agent_running / abort_agent / cleanup_session) is left in a consistent
+state afterwards — using real asyncio Tasks, not mocked ones, for the
+registry tests so the cancellation semantics are real.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+from sqlalchemy import select
+
+from db import async_session
+from models import Message
+from models import Session as SessionModel
+from services.agent import runner
+from services.agent.tasks import (
+    _running_tasks,
+    _session_task_locks,
+    _silent_aborts,
+    abort_agent,
+    is_agent_running,
+    register_task,
+)
+
+
+async def _create_session(session_id: str) -> None:
+    """run_agent's `_publish` helper persists Message rows FK'd to sessions.id
+    (SQLite FK enforcement is on — see db.py), so every test needs a real row."""
+    async with async_session() as db:
+        db.add(SessionModel(id=session_id))
+        await db.commit()
+
+
+async def _messages(session_id: str) -> list[Message]:
+    async with async_session() as db:
+        result = await db.execute(
+            select(Message).where(Message.session_id == session_id).order_by(Message.id)
+        )
+        return list(result.scalars().all())
+
+
+def _state_changes(msgs: list[Message]) -> list[str]:
+    return [
+        m.metadata_["state"]
+        for m in msgs
+        if m.metadata_.get("event_type") == "state_change"
+    ]
+
+
+def _event_types(msgs: list[Message]) -> list[str]:
+    return [m.metadata_.get("event_type") for m in msgs]
+
+
+@pytest.fixture(autouse=True)
+def _clear_task_registry():
+    """The task registry is module-level state that normally only empties via
+    cleanup_session in run_agent's `finally`. If a test fails mid-flight (or a
+    future test skips cleanup), entries would leak across tests — make the
+    isolation explicit rather than relying on distinct session IDs."""
+    yield
+    _running_tasks.clear()
+    _silent_aborts.clear()
+    _session_task_locks.clear()
+
+
+@pytest.fixture(autouse=True)
+def _stub_post_run_hooks(monkeypatch):
+    """run_agent's success path fans out into workspace scanning + post-stage
+    hooks (S3 sync, validation, metadata extraction) that talk to Modal/
+    volume/S3 — orthogonal to the state machine under test and covered by
+    their own modules. Stub them so every test stays fast and deterministic."""
+    monkeypatch.setattr(runner, "publish_artifacts", AsyncMock(return_value=None))
+    monkeypatch.setattr(runner, "post_stage_hook", AsyncMock(return_value=None))
+
+
+# ---------------------------------------------------------------------------
+# Success path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_agent_success_runs_running_then_done_and_returns_text(
+    monkeypatch,
+):
+    session_id = "sess-success"
+    await _create_session(session_id)
+
+    calls: list[dict] = []
+
+    async def fake_drive(**kwargs):
+        calls.append(kwargs)
+        return "final report text"
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    text = await runner.run_agent(
+        session_id=session_id,
+        experiment_id="exp-1",
+        stage="eda",
+        instructions="profile the dataset",
+    )
+
+    assert text == "final report text"
+    assert len(calls) == 1
+    assert calls[0]["agent_type"] == "eda"
+    assert calls[0]["stage"] == "eda"
+    assert calls[0]["session_id"] == session_id
+    assert calls[0]["depth"] == 0
+    assert calls[0]["agent_id"] == "root"
+
+    msgs = await _messages(session_id)
+    assert _state_changes(msgs) == ["eda_running", "eda_done"]
+
+    runner.publish_artifacts.assert_awaited_once_with(session_id, "exp-1", "eda")
+    runner.post_stage_hook.assert_awaited_once_with(session_id, "exp-1", "eda")
+
+
+# ---------------------------------------------------------------------------
+# TimeoutError path — a provider SDK stalling. Must be contained: no raise,
+# a dedicated agent_timeout event, and a "timed_out" terminal state. The
+# post-run hooks must NOT fire since the run never actually produced output.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_agent_timeout_error_is_contained(monkeypatch):
+    session_id = "sess-timeout"
+    await _create_session(session_id)
+
+    async def fake_drive(**kwargs):
+        raise TimeoutError("provider stalled")
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    text = await runner.run_agent(
+        session_id=session_id, experiment_id="exp-2", stage="eda", instructions=""
+    )
+
+    # collected_text never got a chance to accumulate anything.
+    assert text == ""
+
+    msgs = await _messages(session_id)
+    assert "agent_timeout" in _event_types(msgs)
+    assert _state_changes(msgs) == ["eda_running", "timed_out"]
+
+    runner.publish_artifacts.assert_not_awaited()
+    runner.post_stage_hook.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# Generic exception path — must mark the run failed AND re-raise (unlike
+# TimeoutError/CancelledError, which are contained).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_agent_exception_marks_failed_and_reraises(monkeypatch):
+    session_id = "sess-error"
+    await _create_session(session_id)
+
+    async def fake_drive(**kwargs):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    with pytest.raises(RuntimeError, match="boom"):
+        await runner.run_agent(
+            session_id=session_id, experiment_id="exp-3", stage="eda", instructions=""
+        )
+
+    msgs = await _messages(session_id)
+    error_msgs = [m for m in msgs if m.metadata_.get("event_type") == "agent_error"]
+    assert len(error_msgs) == 1
+    assert error_msgs[0].metadata_.get("error") == "boom"
+    assert _state_changes(msgs) == ["eda_running", "failed"]
+
+    runner.publish_artifacts.assert_not_awaited()
+    runner.post_stage_hook.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# CancelledError path — contained (never re-raised, unlike the generic
+# Exception branch), and gated by the module-level `_silent_aborts` set that
+# `abort_agent(session_id, silent=True)` populates for "quiet" follow-up
+# swaps (see routers/sessions.py send_message).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_agent_cancelled_publishes_aborted_when_not_silent(monkeypatch):
+    session_id = "sess-cancel-loud"
+    await _create_session(session_id)
+
+    async def fake_drive(**kwargs):
+        raise asyncio.CancelledError()
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    text = await runner.run_agent(
+        session_id=session_id, experiment_id="exp-4", stage="eda", instructions=""
+    )
+    assert text == ""
+
+    msgs = await _messages(session_id)
+    assert "agent_aborted" in _event_types(msgs)
+    assert _state_changes(msgs) == ["eda_running", "cancelled"]
+
+
+@pytest.mark.asyncio
+async def test_run_agent_cancelled_silent_flag_suppresses_events_and_is_consumed(
+    monkeypatch,
+):
+    session_id = "sess-cancel-silent"
+    await _create_session(session_id)
+    _silent_aborts.add(session_id)
+
+    async def fake_drive(**kwargs):
+        raise asyncio.CancelledError()
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    await runner.run_agent(
+        session_id=session_id, experiment_id="exp-5", stage="eda", instructions=""
+    )
+
+    # The flag is a one-shot: consumed (discarded) regardless of outcome.
+    assert session_id not in _silent_aborts
+
+    msgs = await _messages(session_id)
+    assert "agent_aborted" not in _event_types(msgs)
+    # Only the initial "*_running" transition fired — no terminal state_change
+    # when the abort is silent.
+    assert _state_changes(msgs) == ["eda_running"]
+
+
+# ---------------------------------------------------------------------------
+# Task-registry bookkeeping — spawn -> run -> complete/cancel using REAL
+# asyncio Tasks and the real register_task/abort_agent/is_agent_running from
+# services.agent.tasks, mirroring how routers/sessions.py drives run_agent.
+# A controllable gate stands in for the provider boundary so the task is
+# reliably still "running" when we assert on it mid-flight.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_task_registry_spawn_run_complete_cycle(monkeypatch):
+    session_id = "sess-registry-complete"
+    await _create_session(session_id)
+
+    gate = asyncio.Event()
+    reached_gate = asyncio.Event()
+
+    async def fake_drive(**kwargs):
+        reached_gate.set()
+        await gate.wait()
+        return "done"
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    task = asyncio.create_task(
+        runner.run_agent(
+            session_id=session_id, experiment_id="exp-6", stage="eda", instructions=""
+        )
+    )
+    await register_task(session_id, task)
+    # Let the task run its own DB-backed setup (_load_prev_context /
+    # _load_project_context) all the way to the faked provider boundary,
+    # rather than a bare `sleep(0)` — which would only guarantee ONE
+    # scheduler tick and could still leave the task mid-flight on its own
+    # DB call when we act on it next.
+    await reached_gate.wait()
+
+    assert is_agent_running(session_id) is True
+
+    gate.set()
+    result = await task
+
+    assert result == "done"
+    assert is_agent_running(session_id) is False
+    # cleanup_session (called from run_agent's `finally`, depth==0) popped it.
+    assert session_id not in _running_tasks
+
+    msgs = await _messages(session_id)
+    assert _state_changes(msgs) == ["eda_running", "eda_done"]
+
+
+@pytest.mark.asyncio
+async def test_task_registry_spawn_run_abort_cycle(monkeypatch):
+    session_id = "sess-registry-abort"
+    await _create_session(session_id)
+
+    gate = asyncio.Event()  # never set — simulates a stuck provider call
+    reached_gate = asyncio.Event()
+
+    async def fake_drive(**kwargs):
+        reached_gate.set()
+        await gate.wait()
+        return "unreachable"
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    task = asyncio.create_task(
+        runner.run_agent(
+            session_id=session_id, experiment_id="exp-7", stage="eda", instructions=""
+        )
+    )
+    await register_task(session_id, task)
+    # Wait until the task has cleared its own DB-backed setup and is truly
+    # parked on the provider call before cancelling it — cancelling while
+    # it's mid-flight on its own (unrelated) DB query is a real hazard for
+    # the shared SQLite test connection, not something this test is meant
+    # to exercise.
+    await reached_gate.wait()
+
+    assert is_agent_running(session_id) is True
+
+    cancelled = await abort_agent(session_id)
+
+    assert cancelled is True
+    assert is_agent_running(session_id) is False
+    assert session_id not in _running_tasks
+
+    msgs = await _messages(session_id)
+    assert "agent_aborted" in _event_types(msgs)
+    assert _state_changes(msgs) == ["eda_running", "cancelled"]
+
+
+@pytest.mark.asyncio
+async def test_task_registry_new_message_swaps_stale_task(monkeypatch):
+    """register_task must cancel a still-running previous task for the same
+    session rather than leaking it — this is the "followup message arrives
+    while the agent is still working" case routers/sessions.py relies on."""
+    session_id = "sess-registry-swap"
+    await _create_session(session_id)
+
+    stale_started = asyncio.Event()
+    stale_cancelled = False
+
+    async def stale_coro():
+        nonlocal stale_cancelled
+        stale_started.set()
+        try:
+            await asyncio.sleep(100)
+        except asyncio.CancelledError:
+            stale_cancelled = True
+            raise
+
+    stale_task = asyncio.create_task(stale_coro())
+    await register_task(session_id, stale_task)
+    await stale_started.wait()
+    assert is_agent_running(session_id) is True
+
+    async def fake_drive(**kwargs):
+        return "new run done"
+
+    monkeypatch.setattr(runner, "_drive_provider", fake_drive)
+
+    new_task = asyncio.create_task(
+        runner.run_agent(
+            session_id=session_id, experiment_id="exp-8", stage="eda", instructions=""
+        )
+    )
+    await register_task(session_id, new_task)
+
+    result = await new_task
+    assert result == "new run done"
+    # Await the stale task's cancellation explicitly rather than relying on
+    # the event loop having already delivered its CancelledError during one of
+    # new_task's suspension points — that ordering is not guaranteed.
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(stale_task, timeout=1.0)
+    assert stale_cancelled is True
+    assert is_agent_running(session_id) is False
diff --git a/backend/tests/test_sandbox.py b/backend/tests/test_sandbox.py
new file mode 100644
index 0000000..c765aad
--- /dev/null
+++ b/backend/tests/test_sandbox.py
@@ -0,0 +1,386 @@
+"""Unit tests for services/sandbox.py — run_code's Modal Sandbox lifecycle.
+
+`modal.Sandbox` itself is faked end-to-end (create/stdout/stderr/wait); every
+other collaborator (metrics dispatch, usage recording, span attributes) is
+exercised through the real `run_code` control flow so we're asserting on
+actual routing/attribute-setting logic, not a mock calling a mock.
+
+Three paths per the issue:
+  - success:   returncode == 0, usage recorded is_error=False, span clean.
+  - "timeout": Modal kills the sandbox and returns a non-zero returncode
+    (this is how a per-call timeout actually surfaces here — run_code never
+    raises on its own for this case, it just reports the failure).
+  - exception: sandbox creation itself blows up (e.g. Modal API error) —
+    run_code must tag the span with error=True and re-raise.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import modal
+import pytest
+
+
+class _FakeStream:
+    """Mimics Modal's async stdout/stderr stream."""
+
+    def __init__(self, chunks: list[str]):
+        self._chunks = list(chunks)
+
+    def __aiter__(self):
+        return self._gen()
+
+    async def _gen(self):
+        for c in self._chunks:
+            yield c
+            # Real Modal streams cede control to the event loop between
+            # chunks (network I/O). A bare synchronous generator would let
+            # the stdout loop race all the way to completion — and cancel
+            # the concurrently-scheduled stderr drain task — before that
+            # task ever gets a turn to run. Ceding control here after each
+            # yield keeps the interleaving realistic.
+            await asyncio.sleep(0)
+
+
+class _FakeSandbox:
+    def __init__(
+        self, stdout_chunks: list[str], stderr_chunks: list[str], returncode: int = 0
+    ):
+        self.stdout = _FakeStream(stdout_chunks)
+        self.stderr = _FakeStream(stderr_chunks)
+        self.returncode = returncode
+        self.wait = SimpleNamespace(aio=AsyncMock(return_value=None))
+
+
+class _FakeSpanCtx:
+    """Fake context manager standing in for `sandbox_span(...)`; records the
+    span object so tests can assert on set_attribute calls after the fact."""
+
+    def __init__(self):
+        self.span = MagicMock()
+        self.exc_type = None
+
+    def __enter__(self):
+        return self.span
+
+    def __exit__(self, exc_type, exc, tb):
+        self.exc_type = exc_type
+        return False  # never swallow
+
+
+@pytest.fixture
+def fake_span(monkeypatch):
+    """Patch services.sandbox.sandbox_span with a recorder and hand back the
+    holder dict so the test can inspect the span used for the call."""
+    import services.sandbox as sandbox_module
+
+    holder: dict[str, _FakeSpanCtx] = {}
+
+    def _fake_sandbox_span(**kwargs):
+        ctx = _FakeSpanCtx()
+        ctx.kwargs = kwargs
+        holder["ctx"] = ctx
+        return ctx
+
+    monkeypatch.setattr(sandbox_module, "sandbox_span", _fake_sandbox_span)
+    return holder
+
+
+@pytest.fixture
+def patched_sandbox_deps(monkeypatch):
+    """Stub every Modal/volume/usage collaborator around run_code except
+    modal.Sandbox itself, which each test configures directly."""
+    import services.sandbox as sandbox_module
+    import services.volume as volume_module
+
+    monkeypatch.setattr(sandbox_module, "_get_app", AsyncMock(return_value=object()))
+    monkeypatch.setattr(sandbox_module, "_get_image", MagicMock(return_value=object()))
+    monkeypatch.setattr(sandbox_module, "get_volume", MagicMock(return_value=object()))
+
+    # These are imported locally inside run_code (`from services.volume import
+    # ensure_session_workspace as _ensure_ws, ...`), so patch the source
+    # module's attributes — the lazy import re-resolves them at call time.
+    monkeypatch.setattr(
+        volume_module, "ensure_session_workspace", AsyncMock(return_value=None)
+    )
+    monkeypatch.setattr(
+        volume_module, "reload_volume_async", AsyncMock(return_value=True)
+    )
+
+    usage_mock = AsyncMock(return_value=None)
+    monkeypatch.setattr(sandbox_module, "record_sandbox_usage", usage_mock)
+
+    dispatch_mocks = {
+        "persist_and_publish": AsyncMock(return_value=None),
+        "persist_and_publish_log_event": AsyncMock(return_value=None),
+        "publish_chart_config": AsyncMock(return_value=None),
+    }
+    for name, mock in dispatch_mocks.items():
+        monkeypatch.setattr(sandbox_module, name, mock)
+
+    return SimpleNamespace(usage=usage_mock, dispatch=dispatch_mocks)
+
+
+def _patch_create(monkeypatch, fake_sb: _FakeSandbox | Exception):
+    """Patch modal.Sandbox.create.aio to return fake_sb, or raise it if it's
+    an exception instance."""
+    if isinstance(fake_sb, Exception):
+        create_aio = AsyncMock(side_effect=fake_sb)
+    else:
+        create_aio = AsyncMock(return_value=fake_sb)
+    monkeypatch.setattr(modal.Sandbox, "create", SimpleNamespace(aio=create_aio))
+    return create_aio
+
+
+# ---------------------------------------------------------------------------
+# Success path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_code_success_returns_output_and_records_clean_usage(
+    patched_sandbox_deps, fake_span, monkeypatch
+):
+    from services.sandbox import run_code
+
+    fake_sb = _FakeSandbox(
+        stdout_chunks=["hello\n", "world\n"], stderr_chunks=[], returncode=0
+    )
+    create_aio = _patch_create(monkeypatch, fake_sb)
+
+    result = await run_code(
+        code="print('hello')",
+        session_id="sess-1",
+        stage="eda",
+        gpu=None,
+        agent_type="eda",
+        agent_id="root",
+    )
+
+    assert result == {"stdout": "hello\nworld\n", "stderr": "", "returncode": 0}
+
+    # Sandbox was created exactly once, with the code baked into the script.
+    assert create_aio.await_count == 1
+    _, _, _, code_arg = create_aio.call_args.args[:4]
+    assert "print('hello')" in code_arg
+    assert create_aio.call_args.kwargs["workdir"] == "/data/sessions/sess-1"
+
+    # Usage recorded as a clean run.
+    patched_sandbox_deps.usage.assert_awaited_once()
+    usage_kwargs = patched_sandbox_deps.usage.call_args.kwargs
+    assert usage_kwargs["session_id"] == "sess-1"
+    assert usage_kwargs["is_error"] is False
+    assert usage_kwargs["seconds"] >= 0
+
+    # Span got the expected attributes and was never tagged as an error.
+    span = fake_span["ctx"].span
+    attr_calls = {c.args[0]: c.args[1] for c in span.set_attribute.call_args_list}
+    assert attr_calls["sandbox.code_chars"] == len("print('hello')")
+    assert attr_calls["sandbox.returncode"] == 0
+    assert "sandbox.elapsed_s" in attr_calls
+    assert "error" not in attr_calls
+    assert fake_span["ctx"].exc_type is None
+
+
+@pytest.mark.asyncio
+async def test_run_code_streams_stdout_and_dispatches_metrics(
+    patched_sandbox_deps, fake_span, monkeypatch
+):
+    """A stdout line matching the metrics JSON envelope must be parsed and
+    routed to persist_and_publish; plain text lines must not."""
+    from services.sandbox import run_code
+
+    fake_sb = _FakeSandbox(
+        stdout_chunks=[
+            "not json\n",
+            '{"step": 1, "metrics": {"acc": 0.9}}\n',
+        ],
+        stderr_chunks=[],
+        returncode=0,
+    )
+    _patch_create(monkeypatch, fake_sb)
+
+    published: list[dict] = []
+    from services.broadcaster import broadcaster
+
+    orig_publish = broadcaster.publish
+
+    async def _capture(session_id, event):
+        published.append(event)
+        return await orig_publish(session_id, event)
+
+    monkeypatch.setattr(broadcaster, "publish", _capture)
+
+    await run_code(code="print(1)", session_id="sess-2", stage="eda")
+
+    # stdout chunks were broadcast as code_output events, in order.
+    code_output_texts = [
+        e["data"]["text"] for e in published if e["type"] == "code_output"
+    ]
+    assert code_output_texts == [
+        "not json\n",
+        '{"step": 1, "metrics": {"acc": 0.9}}\n',
+    ]
+
+    # Only the JSON metrics line was dispatched to persist_and_publish.
+    metrics_mock = patched_sandbox_deps.dispatch["persist_and_publish"]
+    metrics_mock.assert_awaited_once()
+    call_args = metrics_mock.call_args.args
+    assert call_args[0] == "sess-2"
+    assert call_args[1] == "eda"
+    assert call_args[2] == [{"step": 1, "name": "acc", "value": 0.9, "run_tag": None}]
+
+    # No stage -> no dispatch at all (regression guard for the `if stage:` gate).
+    patched_sandbox_deps.dispatch["publish_chart_config"].assert_not_awaited()
+    patched_sandbox_deps.dispatch["persist_and_publish_log_event"].assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_run_code_no_stage_skips_metric_parsing(
+    patched_sandbox_deps, fake_span, monkeypatch
+):
+    """Without a stage, run_code must not even attempt to parse stdout for
+    metrics/log events/chart_config."""
+    from services.sandbox import run_code
+
+    fake_sb = _FakeSandbox(
+        stdout_chunks=['{"step": 1, "metrics": {"acc": 0.9}}\n'],
+        stderr_chunks=[],
+        returncode=0,
+    )
+    _patch_create(monkeypatch, fake_sb)
+
+    await run_code(code="print(1)", session_id="sess-3", stage=None)
+
+    patched_sandbox_deps.dispatch["persist_and_publish"].assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# "Timeout" path — Modal kills the container; sandbox surfaces a non-zero
+# returncode rather than raising. run_code must still return normally but
+# flag the run as an error for usage accounting and tracing.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_code_nonzero_returncode_marks_error_but_does_not_raise(
+    patched_sandbox_deps, fake_span, monkeypatch
+):
+    from services.sandbox import run_code
+
+    fake_sb = _FakeSandbox(
+        stdout_chunks=["partial output\n"],
+        stderr_chunks=["Killed\n"],
+        returncode=137,  # SIGKILL exit code — how Modal reports a timeout kill
+    )
+    _patch_create(monkeypatch, fake_sb)
+
+    result = await run_code(
+        code="while True: pass",
+        session_id="sess-4",
+        stage="train",
+        timeout=5,
+        agent_type="trainer",
+    )
+
+    assert result["returncode"] == 137
+    assert result["stderr"] == "Killed\n"
+
+    patched_sandbox_deps.usage.assert_awaited_once()
+    usage_kwargs = patched_sandbox_deps.usage.call_args.kwargs
+    assert usage_kwargs["is_error"] is True
+
+    span = fake_span["ctx"].span
+    attr_calls = {c.args[0]: c.args[1] for c in span.set_attribute.call_args_list}
+    assert attr_calls["error"] is True
+    assert attr_calls["sandbox.returncode"] == 137
+    assert attr_calls["sandbox.timeout_s"] == 5
+    assert fake_span["ctx"].exc_type is None  # no exception propagated
+
+
+# ---------------------------------------------------------------------------
+# Exception path — sandbox creation itself raises. run_code must tag the
+# span with error=True and re-raise (never swallow).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_code_create_failure_tags_span_and_reraises(
+    patched_sandbox_deps, fake_span, monkeypatch
+):
+    from services.sandbox import run_code
+
+    boom = RuntimeError("modal API unavailable")
+    _patch_create(monkeypatch, boom)
+
+    with pytest.raises(RuntimeError, match="modal API unavailable"):
+        await run_code(code="print(1)", session_id="sess-5", stage="eda")
+
+    span = fake_span["ctx"].span
+    attr_calls = {c.args[0]: c.args[1] for c in span.set_attribute.call_args_list}
+    assert attr_calls.get("error") is True
+    assert fake_span["ctx"].exc_type is RuntimeError
+
+    # We never got far enough to record usage or a returncode.
+    patched_sandbox_deps.usage.assert_not_awaited()
+    assert "sandbox.returncode" not in attr_calls
+
+
+@pytest.mark.asyncio
+async def test_run_code_stdout_iteration_failure_tags_span_and_reraises(
+    patched_sandbox_deps, fake_span, monkeypatch
+):
+    """A mid-stream failure (e.g. the sandbox connection dropping while
+    reading stdout) must also be treated as an exception path: tag the span
+    and propagate, not silently report a fake success."""
+    from services.sandbox import run_code
+
+    class _BoomStream:
+        def __aiter__(self):
+            return self._gen()
+
+        async def _gen(self):
+            if False:
+                yield  # pragma: no cover - makes this an async generator
+            raise ConnectionError("stream dropped")
+
+    fake_sb = _FakeSandbox(stdout_chunks=[], stderr_chunks=[], returncode=0)
+    fake_sb.stdout = _BoomStream()
+    _patch_create(monkeypatch, fake_sb)
+
+    with pytest.raises(ConnectionError, match="stream dropped"):
+        await run_code(code="print(1)", session_id="sess-6", stage="eda")
+
+    span = fake_span["ctx"].span
+    attr_calls = {c.args[0]: c.args[1] for c in span.set_attribute.call_args_list}
+    assert attr_calls.get("error") is True
+    assert fake_span["ctx"].exc_type is ConnectionError
+    patched_sandbox_deps.usage.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_run_code_passes_gpu_and_timeout_through(
+    patched_sandbox_deps, fake_span, monkeypatch
+):
+    """Regression guard: gpu/timeout/volumes must reach modal.Sandbox.create
+    unchanged — a common typo-class bug (e.g. swapping args) would silently
+    run every sandbox on CPU or with the wrong timeout."""
+    from services.sandbox import run_code
+
+    fake_sb = _FakeSandbox(stdout_chunks=[], stderr_chunks=[], returncode=0)
+    create_aio = _patch_create(monkeypatch, fake_sb)
+
+    await run_code(
+        code="print(1)",
+        session_id="sess-7",
+        stage="train",
+        gpu="A10G",
+        timeout=42,
+    )
+
+    kwargs = create_aio.call_args.kwargs
+    assert kwargs["gpu"] == "A10G"
+    assert kwargs["timeout"] == 42