From cc8d9eb62d0bbee4919cf2e0ac4fc5e877c1b867 Mon Sep 17 00:00:00 2001
From: shfunc <hagforall@proton.me>
Date: Tue, 28 Jul 2026 16:32:46 +0200
Subject: [PATCH 1/4] fix(capabilities,environment): anchor a symlinked
 workspace path instead of nesting it

---
 hud/capabilities/base.py                |  6 +++++
 hud/capabilities/ssh.py                 | 12 +++++++--
 hud/environment/tests/test_workspace.py | 33 +++++++++++++++++++++++++
 hud/environment/workspace.py            |  8 ++++++
 4 files changed, 57 insertions(+), 2 deletions(-)

diff --git a/hud/capabilities/base.py b/hud/capabilities/base.py
index 02ad1b199..7c243fd62 100644
--- a/hud/capabilities/base.py
+++ b/hud/capabilities/base.py
@@ -79,6 +79,7 @@ def ssh(
         client_key_path: str | os.PathLike[str] | None = None,
         shell: str | None = None,
         cwd: str | None = None,
+        cwd_alias: str | None = None,
     ) -> Capability:
         """``ssh/2`` — SSH daemon with publickey auth.
 
@@ -90,6 +91,9 @@ def ssh(
         from ``sys.platform`` at construction time. Agents read this to
         format commands correctly. ``cwd`` is the absolute path sessions
         start in (the served workspace); clients anchor file paths to it.
+        ``cwd_alias`` is a second spelling of that same directory (a path
+        that reaches it through a symlink), which clients treat as already
+        anchored rather than as a workspace-relative address.
         """
         normalized = normalize_url(url, default_scheme="ssh", default_port=22)
         if shell is None:
@@ -101,6 +105,8 @@ def ssh(
             params["client_key_path"] = os.fspath(client_key_path)
         if cwd is not None:
             params["cwd"] = cwd
+        if cwd_alias is not None:
+            params["cwd_alias"] = cwd_alias
         return cls(name=name, protocol="ssh/2", url=normalized, params=params)
 
     @classmethod
diff --git a/hud/capabilities/ssh.py b/hud/capabilities/ssh.py
index d9a817cd6..c3372c6b9 100644
--- a/hud/capabilities/ssh.py
+++ b/hud/capabilities/ssh.py
@@ -56,10 +56,15 @@ def map_path(self, path: str) -> str:
         real filesystem; replicate the chroot: strip the cwd prefix if present,
         normalize the remainder against ``/`` (clamping ``..`` at the root,
         exactly as a chroot does), and re-anchor under the cwd. Idempotent.
+
+        ``cwd_alias`` is the same directory reached through a symlink, and is
+        stripped like the cwd: re-anchoring it instead would turn an already
+        correct absolute path into a nested one that does not exist.
         """
         cwd = str(self.capability.params.get("cwd", "")).rstrip("/")
         if not cwd:
             return path
+        alias = str(self.capability.params.get("cwd_alias", "")).rstrip("/")
         if self._is_windows:
             # The workspace publishes cwd via as_posix() (e.g. "C:/work") but
             # callers pass native paths ("C:\work\file.txt"); NTFS paths are
@@ -70,8 +75,11 @@ def map_path(self, path: str) -> str:
             elif len(path) >= 2 and path[1] == ":" and path[0].isalpha():
                 # Drive-absolute outside the workspace: anchor like the chroot.
                 path = path[2:]
-        elif path == cwd or path.startswith(cwd + "/"):
-            path = path[len(cwd) :]
+        else:
+            for prefix in (cwd, alias):
+                if prefix and (path == prefix or path.startswith(prefix + "/")):
+                    path = path[len(prefix) :]
+                    break
         normalized = posixpath.normpath("/" + path.lstrip("/"))
         return cwd if normalized == "/" else cwd + normalized
 
diff --git a/hud/environment/tests/test_workspace.py b/hud/environment/tests/test_workspace.py
index 66f6bc2eb..90c10d0bc 100644
--- a/hud/environment/tests/test_workspace.py
+++ b/hud/environment/tests/test_workspace.py
@@ -6,6 +6,7 @@
 import sys
 import tempfile
 from pathlib import Path
+from typing import Any, cast
 
 import asyncssh
 import pytest
@@ -268,3 +269,35 @@ def test_required_isolation_refuses_when_unavailable(monkeypatch, tmp_path) -> N
 
     with pytest.raises(RuntimeError, match="isolation was required"):
         ws.Workspace(tmp_path, require_isolation=True)
+
+
+@pytest.mark.asyncio
+async def test_a_symlinked_root_publishes_both_spellings(tmp_path: Path) -> None:
+    """A workspace addressed through a symlink (macOS /tmp -> /private/tmp) serves the
+    real path, so it must publish the caller's spelling too or clients re-anchor it."""
+    real = tmp_path / "real"
+    real.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(real, target_is_directory=True)
+
+    ws = Workspace(link)
+    await ws.start()
+    try:
+        cap = ws.capability()
+        assert cap.params["cwd"] == real.as_posix()
+        assert cap.params["cwd_alias"] == link.as_posix()
+
+        client = SSHClient(cap, cast("Any", None))
+        assert client.map_path(f"{link}/calc.py") == f"{real}/calc.py"
+    finally:
+        await ws.stop()
+
+
+@pytest.mark.asyncio
+async def test_a_plain_root_publishes_no_alias(tmp_path: Path) -> None:
+    ws = Workspace(tmp_path / "root")
+    await ws.start()
+    try:
+        assert "cwd_alias" not in ws.capability().params
+    finally:
+        await ws.stop()
diff --git a/hud/environment/workspace.py b/hud/environment/workspace.py
index 13ba752c0..ab3a2c26f 100644
--- a/hud/environment/workspace.py
+++ b/hud/environment/workspace.py
@@ -203,6 +203,13 @@ def __init__(
         # Only override the default; respect an explicit guest_path.
         if self._bwrap is None and guest_path == "/workspace":
             self._guest_path = self.root.as_posix()
+        # The caller's spelling of the same directory when it differs from the real
+        # path (macOS resolves /tmp to /private/tmp) and sessions run under the real
+        # one. A client that knows only the real path treats the other spelling as a
+        # workspace-relative address and re-anchors it somewhere that does not exist.
+        given = Path(root).absolute().as_posix()
+        real = self.root.as_posix()
+        self._cwd_alias = given if given != real and self._guest_path == real else None
         # ssh config
         self._ssh_host = host
         self._ssh_port = port
@@ -430,6 +437,7 @@ def capability(self, name: str = "shell") -> Capability:
             client_key=key_path.read_text() if key_path else None,
             client_key_path=key_path,
             cwd=self._guest_path,
+            cwd_alias=self._cwd_alias,
         )
 
     @property

From fedbd4e46e6c1e1fe9d7ad1fc3de6cc15be57dec Mon Sep 17 00:00:00 2001
From: shfunc <hagforall@proton.me>
Date: Tue, 28 Jul 2026 16:33:22 +0200
Subject: [PATCH 2/4] fix(agents,cli): report tool failures and trace output
 verbatim

---
 .../tests/test_provider_native_tools.py       | 58 ++++++++++++++++++-
 hud/agents/tools/ssh.py                       | 29 ++++++++--
 hud/cli/trace.py                              | 17 +++---
 3 files changed, 92 insertions(+), 12 deletions(-)

diff --git a/hud/agents/tests/test_provider_native_tools.py b/hud/agents/tests/test_provider_native_tools.py
index 3711b0832..0272df301 100644
--- a/hud/agents/tests/test_provider_native_tools.py
+++ b/hud/agents/tests/test_provider_native_tools.py
@@ -10,6 +10,7 @@
 import shlex
 from typing import Any, cast
 
+import asyncssh
 import mcp.types as mcp_types
 import pytest
 
@@ -47,7 +48,13 @@ async def run(
         self.commands.append(command)
         parts = shlex.split(command)
         if len(parts) == 3 and parts[:2] == ["cat", "--"]:
-            return _Completed(stdout=self._store.get(parts[2], b"").decode())
+            if parts[2] not in self._store:
+                return self._finish(
+                    command,
+                    check,
+                    _Completed(stderr=f"cat: {parts[2]}: No such file or directory", exit_status=1),
+                )
+            return _Completed(stdout=self._store[parts[2]].decode())
         if len(parts) == 3 and parts[:2] == ["cat", ">"]:
             assert input is not None
             self._store[parts[2]] = input.encode()
@@ -75,6 +82,21 @@ async def run(
             return _Completed(exit_status=0)
         return self._completed
 
+    @staticmethod
+    def _finish(command: str, check: bool, completed: _Completed) -> _Completed:
+        if check and completed.exit_status:
+            raise asyncssh.ProcessError(
+                env=None,
+                command=command,
+                subsystem=None,
+                exit_status=completed.exit_status,
+                exit_signal=None,
+                returncode=completed.exit_status,
+                stdout=completed.stdout,
+                stderr=completed.stderr,
+            )
+        return completed
+
 
 class _FakeSSH(SSHClient):
     """SSH client with an in-memory exec-channel filesystem."""
@@ -520,3 +542,37 @@ async def test_gemini_edit_creates_file_when_old_string_empty() -> None:
     await tool.execute({"file_path": "/n.txt", "old_string": "", "new_string": "fresh"})
 
     assert ssh.files["/n.txt"] == b"fresh"
+
+
+def test_map_path_leaves_a_symlinked_spelling_of_the_workspace_alone() -> None:
+    """A workspace made at /tmp/w is served as /private/tmp/w on macOS; re-anchoring
+    the caller's spelling instead of stripping it nests the path under itself."""
+    cap = Capability(
+        name="shell",
+        protocol="ssh/2",
+        url="ssh://localhost:22",
+        params={"cwd": "/private/tmp/w", "cwd_alias": "/tmp/w"},
+    )
+    ssh = SSHClient(cap, cast("Any", None))
+
+    assert ssh.map_path("/tmp/w/calc.py") == "/private/tmp/w/calc.py"
+    assert ssh.map_path("/private/tmp/w/calc.py") == "/private/tmp/w/calc.py"
+    assert ssh.map_path("/tmp/w") == "/private/tmp/w"
+    # Workspace-relative addressing still anchors, and an unrelated absolute
+    # path is still clamped into the workspace like a chroot.
+    assert ssh.map_path("/REPORT.md") == "/private/tmp/w/REPORT.md"
+    assert ssh.map_path("/tmp/elsewhere/f.txt") == "/private/tmp/w/tmp/elsewhere/f.txt"
+
+
+async def test_reading_a_missing_file_is_a_tool_error_not_a_raised_traceback() -> None:
+    """Reading before creating is the first thing an editor tool does; that failure
+    must come back as a tool result carrying the shell's message."""
+    ssh = _FakeSSH()
+    tool = ClaudeTextEditorTool(
+        spec=ClaudeTextEditorTool.default_spec("claude"), client=cast("SSHClient", ssh)
+    )
+
+    result = await tool.execute({"command": "view", "path": "/nope.txt"})
+
+    assert result.isError is True
+    assert "No such file or directory" in result_text(result)
diff --git a/hud/agents/tools/ssh.py b/hud/agents/tools/ssh.py
index 6649535df..b88707271 100644
--- a/hud/agents/tools/ssh.py
+++ b/hud/agents/tools/ssh.py
@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import asyncssh
 import mcp.types as mcp_types
 
 from hud.agents.tools.base import AgentTool
@@ -37,19 +38,39 @@ async def bash(self, command: str) -> MCPToolResult:
 
     async def file_read(self, path: str) -> MCPToolResult:
         """Read a text file through SSH exec."""
-        return tool_ok(await self.client.read_text(path))
+        try:
+            return tool_ok(await self.client.read_text(path))
+        except asyncssh.ProcessError as e:
+            return tool_err(_remote_error(e))
 
     async def file_write(self, path: str, content: str) -> MCPToolResult:
         """Write a text file through SSH exec."""
-        await self.client.write_text(path, content)
+        try:
+            await self.client.write_text(path, content)
+        except asyncssh.ProcessError as e:
+            return tool_err(_remote_error(e))
         return tool_ok(f"wrote {len(content)} bytes to {path}")
 
     async def file_list(self, path: str = "/") -> MCPToolResult:
         """List directory entries through SSH exec."""
-        names = await self.client.listdir(path)
+        try:
+            names = await self.client.listdir(path)
+        except asyncssh.ProcessError as e:
+            return tool_err(_remote_error(e))
         return tool_ok("\n".join(names) if names else "(empty)")
 
 
-from hud.agents.tools.base import tool_ok  # noqa: E402
+def _remote_error(exc: asyncssh.ProcessError) -> str:
+    """The line the remote command printed, not asyncssh's whole connection repr.
+
+    A failed file command is an ordinary tool outcome — reading a file that does
+    not exist yet is the first thing an editor tool does — so the agent gets what
+    the shell said, not a ~30-line traceback.
+    """
+    stderr = exc.stderr.decode("utf-8", "replace") if isinstance(exc.stderr, bytes) else exc.stderr
+    return stderr.strip() or f"exit {exc.exit_status}"
+
+
+from hud.agents.tools.base import tool_err, tool_ok  # noqa: E402
 
 __all__ = ["SSHTool"]
diff --git a/hud/cli/trace.py b/hud/cli/trace.py
index 7b95486c9..3d1a1a555 100644
--- a/hud/cli/trace.py
+++ b/hud/cli/trace.py
@@ -8,6 +8,7 @@
 
 import typer
 from rich.console import Console
+from rich.markup import escape
 from rich.panel import Panel
 from rich.rule import Rule
 from rich.text import Text
@@ -161,6 +162,7 @@ def _load_remote(trace_id: str) -> list[dict[str, Any]]:
 
 
 def _render_events(events: list[dict[str, Any]]) -> None:
+    # Payloads are escaped: rich reads `[len(s) // 2]` as a style tag and drops it
     turn = 0
     for ev in events:
         kind = ev.get("kind")
@@ -175,7 +177,7 @@ def _render_events(events: list[dict[str, Any]]) -> None:
 
             text = ev.get("text")
             if text:
-                console.print(text)
+                console.print(escape(str(text)))
 
             for tc in ev.get("tool_calls") or []:
                 name = tc.get("name") or tc.get("function", {}).get("name", "?")
@@ -184,29 +186,30 @@ def _render_events(events: list[dict[str, Any]]) -> None:
                     with contextlib.suppress(Exception):
                         args = json.loads(args)
                 console.print(
-                    f"  [green]→[/green] [bold]{name}[/bold]({_fmt_args(args)})",
+                    f"  [green]→[/green] [bold]{escape(str(name))}[/bold]"
+                    f"({escape(_fmt_args(args))})",
                     highlight=False,
                 )
 
             if ev.get("error"):
-                console.print(f"  [red]error: {ev['error']}[/red]")
+                console.print(f"  [red]error: {escape(str(ev['error']))}[/red]")
 
         elif kind in ("tool_call", "tool_result"):
             name = ev.get("tool_name") or ev.get("name") or "?"
             result = ev.get("result_text") or ev.get("result") or ""
             error = ev.get("error")
             if error:
-                console.print(f"  [red]✗ {name}: {error}[/red]")
+                console.print(f"  [red]✗ {escape(str(name))}: {escape(str(error))}[/red]")
             else:
-                console.print(f"  [dim]{name} →[/dim]")
+                console.print(f"  [dim]{escape(str(name))} →[/dim]")
                 for line in str(result).splitlines():
-                    console.print(f"    {line}")
+                    console.print(f"    {escape(line)}")
 
         elif kind == "environment":
             msg = ev.get("text") or ev.get("content") or ""
             if msg:
                 console.print(Rule("[yellow]env[/yellow]", style="yellow"))
-                console.print(msg)
+                console.print(escape(str(msg)))
 
 
 def _fmt_args(args: Any) -> str:

From 56a22c5c7a4c32bcc75d0bd1cf1ed5fc8792683f Mon Sep 17 00:00:00 2001
From: shfunc <hagforall@proton.me>
Date: Tue, 28 Jul 2026 16:33:53 +0200
Subject: [PATCH 3/4] fix(eval,train): make failed rollouts say why, and
 Daytona snapshots track their env

---
 docs/v6/reference/runtime.mdx          |   6 +-
 hud/eval/run.py                        |  19 ++-
 hud/eval/runtime.py                    | 173 ++++++++++++++++----
 hud/eval/tests/test_docker_provider.py | 214 ++++++++++++++++++++++++-
 hud/eval/tests/test_rollout.py         |  15 ++
 hud/train/client.py                    |  25 +++
 hud/train/tests/test_client.py         |  35 ++++
 7 files changed, 451 insertions(+), 36 deletions(-)

diff --git a/docs/v6/reference/runtime.mdx b/docs/v6/reference/runtime.mdx
index e64f2a21c..4e60103e2 100644
--- a/docs/v6/reference/runtime.mdx
+++ b/docs/v6/reference/runtime.mdx
@@ -152,10 +152,14 @@ DaytonaRuntime(snapshot_name=None, *, image=None, command=None, workdir="/app",
 ```
 
 - **`snapshot_name`** - Daytona snapshot to boot from (the durable handle).
-- **`image`** - Dockerfile/registry ref to build the snapshot once if it's missing. Resources (cpu/memory/gpu) live on the snapshot.
+- **`image`** - Dockerfile/registry ref to build the snapshot once if it's missing. Resources (cpu/memory/gpu) live on the snapshot. A snapshot built from `image` is content-addressed — its name carries a digest of the Dockerfile and everything it copies in (`my-env-9f2c1d0ab3e4`), so editing the env boots a rebuilt snapshot instead of silently reusing the one from before the edit. The digest is part of the stored name, so `DaytonaRuntime("my-env")` with no `image` will not find what `DaytonaRuntime("my-env", image=…)` built: keep passing `image` (it re-resolves to the same snapshot for free when nothing changed), or pass the full digested name.
 - **`workdir`** / **`port`** - guest working directory and in-sandbox serving port.
 - **`ssh_host`** / **`ssh_expires_minutes`** - SSH tunnel settings (Daytona exposes services over an SSH local-forward).
 
+`runtime_config.resources` (cpu/memory/gpu) is applied when the snapshot is built from `image`, and is part of the digest — same image at a different size is a different snapshot. Booting a snapshot you did not build (`snapshot_name` with no `image`) cannot resize it, since resources are fixed at build time.
+
+`await runtime.prune_snapshots()` deletes snapshots under this `snapshot_name` that this runtime has not booted — content-addressing costs one snapshot per edit, and nothing else reaps them. Every digest the instance booted is kept (a taskset with per-row `resources` uses several at once), so reaping across env edits means pruning from a fresh runtime.
+
 ### `HUDRuntime`
 
 ```python
diff --git a/hud/eval/run.py b/hud/eval/run.py
index 396a84feb..76075cb06 100644
--- a/hud/eval/run.py
+++ b/hud/eval/run.py
@@ -388,13 +388,14 @@ async def _drive() -> None:
             driver.add_done_callback(_consume_task_result)
             raise
         except Exception as exc:
+            detail = _detail(exc)
             if run is None:
-                logger.warning("rollout failed before launch (%s): %s", _phase, exc)
-                run = Run.failed(f"[{_phase}] {exc}")
+                logger.warning("rollout failed before launch (%s): %s", _phase, detail)
+                run = Run.failed(f"[{_phase}] {detail}")
             else:
-                logger.warning("rollout failed mid-run (%s): %s", _phase, exc)
+                logger.warning("rollout failed mid-run (%s): %s", _phase, detail)
                 run.trace.status = "error"
-                run.record(Step(source="system", error=f"[{_phase}] {exc}"))
+                run.record(Step(source="system", error=f"[{_phase}] {detail}"))
         assert run is not None  # the body bound it, or the handler synthesized it
         run.trace.trace_id = trace_id
         run.job_id = job_id
@@ -404,6 +405,16 @@ async def _drive() -> None:
     return run
 
 
+def _detail(exc: BaseException) -> str:
+    """The failure text plus any notes attached to it.
+
+    Providers attach what only they can see — a sandbox's env output on a failed
+    handshake — as notes. ``str(exc)`` drops them, so a rollout that died at
+    import would report ``closed connection during 'hello'`` and nothing else.
+    """
+    return "\n".join([str(exc), *getattr(exc, "__notes__", [])])
+
+
 def _consume_task_result(task: asyncio.Future[Any]) -> None:
     with contextlib.suppress(asyncio.CancelledError, Exception):
         task.result()
diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py
index bb674732a..a8c38a572 100644
--- a/hud/eval/runtime.py
+++ b/hud/eval/runtime.py
@@ -32,7 +32,9 @@
 
 import asyncio
 import contextlib
+import hashlib
 import logging
+import re
 import sys
 import uuid
 from collections import deque
@@ -651,6 +653,54 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                 await sb.terminate.aio()
 
 
+# Build noise that must not perturb the snapshot digest: these are uploaded with
+# the context (Daytona's archiver has no .dockerignore) but never define the env,
+# so hashing them would rebuild the snapshot every time the host venv is touched.
+_DIGEST_PATTERN = re.compile(r"[0-9a-f]{12}")
+
+_DIGEST_SKIP = frozenset(
+    {".git", ".venv", "venv", "__pycache__", ".pytest_cache", ".ruff_cache", ".mypy_cache"}
+)
+
+
+def _snapshot_digest(context_digest: str, resources: Any) -> str:
+    """Name suffix for a snapshot: its content plus the sizing baked into it."""
+    return hashlib.sha256(f"{context_digest}{resources!r}".encode()).hexdigest()[:12]
+
+
+def _image_digest(image: Any) -> str:
+    """Content hash of a Daytona ``Image``: its Dockerfile plus every byte it copies in.
+
+    A Daytona snapshot is identified by name alone, so the name has to carry the
+    content — otherwise an edited env resolves to the snapshot built before the
+    edit and every rollout after it silently measures the old code. ``_context_list``
+    is the same private attribute ``snapshot.create`` reads to decide what to upload.
+    """
+    digest = hashlib.sha256()
+    digest.update(image.dockerfile().encode())
+    for entry in image._context_list:
+        source = Path(entry.source_path)
+        is_dir = source.is_dir()
+        for file in sorted(source.rglob("*")) if is_dir else [source]:
+            # as_posix so the same tree digests identically on every host OS.
+            rel = file.relative_to(source).as_posix() if is_dir else file.name
+            if not file.is_file() or _DIGEST_SKIP & set(rel.split("/")):
+                continue
+            digest.update(rel.encode())
+            digest.update(file.read_bytes())
+    return digest.hexdigest()[:12]
+
+
+async def _daytona_serve_output(sandbox: Any, session: str, cmd_id: str) -> str:
+    """What the env process printed inside the sandbox, for a failed handshake."""
+    try:
+        logs = await sandbox.process.get_session_command_logs(session, cmd_id)
+    except Exception as e:  # the sandbox may already be gone
+        return f"env output unavailable: {e}"
+    output = (logs.stderr or logs.output or logs.stdout or "").strip()
+    return f"env output in sandbox {sandbox.id}:\n{output}" if output else "env printed nothing"
+
+
 class DaytonaRuntime:
     """The Daytona provider: each acquisition creates a fresh sandbox from a snapshot.
 
@@ -663,6 +713,12 @@ class DaytonaRuntime:
 
     Pass a snapshot name — ``DaytonaRuntime("hud-libero-env")`` — optionally with an
     ``image`` (Dockerfile/registry ref) to build that snapshot once if it is missing.
+    A built snapshot is content-addressed: its name carries a digest of the image
+    and everything the image copies in (``hud-libero-env-9f2c1d0ab3e4``), so editing
+    the env boots a rebuilt snapshot instead of silently reusing the stale one. That
+    digest is the name Daytona stores, so booting it later by name alone needs the
+    full ``hud-libero-env-9f2c1d0ab3e4`` — keep passing ``image`` instead, which
+    re-resolves to the same snapshot for free when nothing changed.
     Resources (cpu/memory/gpu) live on the snapshot, not here. *workdir* defaults to
     ``/app`` (the scaffolded ``Dockerfile.hud`` WORKDIR) since a Daytona session
     starts in ``~``, not the image's WORKDIR; override only for a non-standard layout.
@@ -695,12 +751,43 @@ def __init__(
         if runtime_config is not None:
             config = RuntimeConfig.model_validate(runtime_config)
         self.runtime_config = config
-        # Build the snapshot from *image* once if it's missing; lock so concurrent
-        # first acquisitions resolve exactly once.
         self._image = image
-        self._resolved = False
+        # The context is walked once; resources come from the row, so they fold
+        # into the name per acquisition.
+        self._context_digest: str | None = None
+        self._ensured: set[str] = set()
         self._snapshot_lock = asyncio.Lock()
 
+    async def prune_snapshots(self) -> list[str]:
+        """Delete snapshots under this *snapshot_name* that this runtime has not booted.
+
+        Content-addressed names cost one snapshot per edit and nothing reaps them.
+        Every digest this instance booted is kept — a taskset with per-row resources
+        uses several at once — so reaping across edits means pruning from a fresh
+        runtime, not from one that already booted the old digest.
+
+        Only names this runtime could have minted are touched: the prefix plus a
+        digest, never a hand-named snapshot that shares it.
+        """
+        from daytona import AsyncDaytona
+
+        if not self._ensured:
+            raise RuntimeError("prune_snapshots needs a resolved snapshot: run a rollout first")
+
+        prefix = f"{self.snapshot_name}-"
+        deleted: list[str] = []
+        async with AsyncDaytona() as daytona:
+            page = await daytona.snapshot.list()
+            for snapshot in page.items:
+                name = str(snapshot.name)
+                if name in self._ensured or not name.startswith(prefix):
+                    continue
+                if _DIGEST_PATTERN.fullmatch(name[len(prefix) :]):
+                    await daytona.snapshot.delete(snapshot)
+                    deleted.append(name)
+        logger.info("pruned %d Daytona snapshot(s), kept %s", len(deleted), sorted(self._ensured))
+        return deleted
+
     @asynccontextmanager
     async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
         import asyncssh
@@ -725,7 +812,13 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
             if config.resources is not None:
                 resource_kwargs: dict[str, Any] = {}
                 if config.resources.cpu is not None:
-                    resource_kwargs["cpu"] = config.resources.cpu
+                    # Daytona allocates whole cores; truncating resizes silently.
+                    if not config.resources.cpu.is_integer():
+                        raise ValueError(
+                            "DaytonaRuntime needs a whole number of CPUs, got "
+                            f"{config.resources.cpu}"
+                        )
+                    resource_kwargs["cpu"] = int(config.resources.cpu)
                 if config.resources.memory_mb is not None:
                     resource_kwargs["memory"] = max(
                         1,
@@ -748,31 +841,40 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                     kwargs["resources"] = daytona_resources
                 sandbox_params = CreateSandboxFromImageParams(**kwargs)
             else:
-                if daytona_resources is not None:
-                    raise ValueError(
-                        "DaytonaRuntime cannot override resources for snapshot_name; "
-                        "use runtime_config.image"
-                    )
                 if self.snapshot_name is None:
                     raise ValueError(
                         "DaytonaRuntime requires snapshot_name or runtime_config.image"
                     )
-                if not self._resolved:
+                if daytona_resources is not None and self._image is None:
+                    raise ValueError(
+                        "DaytonaRuntime cannot resize an already-built snapshot: resources "
+                        "are fixed when it is built, so pass image= to build one"
+                    )
+                snapshot_name = self.snapshot_name
+                if self._image is not None:
                     async with self._snapshot_lock:
-                        if not self._resolved:
-                            if self._image is not None:
-                                try:
-                                    await daytona.snapshot.get(self.snapshot_name)
-                                except DaytonaNotFoundError:
-                                    await daytona.snapshot.create(
-                                        CreateSnapshotParams(
-                                            name=self.snapshot_name,
-                                            image=self._image,
-                                        )
+                        if self._context_digest is None:
+                            self._context_digest = _image_digest(self._image)
+                        snapshot_name = (
+                            f"{self.snapshot_name}-"
+                            f"{_snapshot_digest(self._context_digest, daytona_resources)}"
+                        )
+                        if snapshot_name not in self._ensured:
+                            try:
+                                await daytona.snapshot.get(snapshot_name)
+                                logger.info("reusing Daytona snapshot %s", snapshot_name)
+                            except DaytonaNotFoundError:
+                                logger.info("building Daytona snapshot %s", snapshot_name)
+                                await daytona.snapshot.create(
+                                    CreateSnapshotParams(
+                                        name=snapshot_name,
+                                        image=self._image,
+                                        resources=daytona_resources,
                                     )
-                            self._resolved = True
+                                )
+                            self._ensured.add(snapshot_name)
                 sandbox_params = CreateSandboxFromSnapshotParams(
-                    snapshot=self.snapshot_name,
+                    snapshot=snapshot_name,
                     ephemeral=True,
                     auto_stop_interval=0,
                 )
@@ -793,7 +895,7 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                 session: str = "hud-serve"
                 await sandbox.process.create_session(session)
                 cmd = f"cd {self.workdir} && {self.command}" if self.workdir else self.command
-                await sandbox.process.execute_session_command(
+                started = await sandbox.process.execute_session_command(
                     session, SessionExecuteRequest(command=cmd, run_async=True)
                 )
                 ssh = await sandbox.create_ssh_access(expires_in_minutes=self.ssh_expires_minutes)
@@ -801,15 +903,26 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                     self.ssh_host, username=ssh.token, known_hosts=None
                 ) as conn:
                     listener = await conn.forward_local_port("127.0.0.1", 0, "127.0.0.1", self.port)
-                    yield Runtime(
-                        f"tcp://127.0.0.1:{listener.get_port()}",
-                        params={"provider": "daytona", "instance_id": sandbox.id},
-                        config=config if config.model_dump(exclude_none=True) else None,
-                    )
+                    try:
+                        yield Runtime(
+                            f"tcp://127.0.0.1:{listener.get_port()}",
+                            params={"provider": "daytona", "instance_id": sandbox.id},
+                            config=config if config.model_dump(exclude_none=True) else None,
+                        )
+                    except (EOFError, OSError) as exc:
+                        # Why it died only exists inside the sandbox.
+                        exc.add_note(await _daytona_serve_output(sandbox, session, started.cmd_id))
+                        raise
             finally:
-                # check-free teardown: never shadow the run's own error.
-                with contextlib.suppress(Exception):
+                try:
                     await daytona.delete(sandbox)
+                except Exception:
+                    # Swallowing this is how a billable sandbox outlives its process.
+                    logger.warning(
+                        "failed to delete Daytona sandbox %s; it may still be running",
+                        sandbox.id,
+                        exc_info=True,
+                    )
 
 
 @asynccontextmanager
diff --git a/hud/eval/tests/test_docker_provider.py b/hud/eval/tests/test_docker_provider.py
index 561c53796..e5f3a58c6 100644
--- a/hud/eval/tests/test_docker_provider.py
+++ b/hud/eval/tests/test_docker_provider.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import os
 import sys
 from dataclasses import dataclass
@@ -202,6 +203,7 @@ class _CreateSandboxFromSnapshotParams:
 class _CreateSnapshotParams:
     name: str
     image: object
+    resources: object | None = None
 
 
 @dataclass(frozen=True)
@@ -239,12 +241,20 @@ class _SessionExecuteRequest:
 class _FakeDaytonaProcess:
     def __init__(self, calls: dict[str, object]) -> None:
         self._calls = calls
+        self.logs = SimpleNamespace(
+            stderr="ImportError: no module named bugs", output="", stdout=""
+        )
 
     async def create_session(self, session: str) -> None:
         self._calls["session"] = session
 
-    async def execute_session_command(self, session: str, request: object) -> None:
+    async def execute_session_command(self, session: str, request: object) -> SimpleNamespace:
         self._calls["execute"] = (session, request)
+        return SimpleNamespace(cmd_id="cmd-1")
+
+    async def get_session_command_logs(self, session: str, cmd_id: str) -> SimpleNamespace:
+        self._calls["logs"] = (session, cmd_id)
+        return self.logs
 
 
 class _FakeDaytonaSandbox:
@@ -259,16 +269,42 @@ async def create_ssh_access(self, *, expires_in_minutes: int) -> SimpleNamespace
         return SimpleNamespace(token="ssh-token")
 
 
+class _FakeSnapshotApi:
+    """The registry a Daytona snapshot name resolves against: ``get`` 404s until
+    something with that exact name has been built."""
+
+    def __init__(self, calls: dict[str, object]) -> None:
+        self.built: list[str] = []
+        calls["snapshot_create"] = self.built
+
+    async def get(self, name: str) -> SimpleNamespace:
+        if name not in self.built:
+            raise RuntimeError(f"snapshot {name} not found")
+        return SimpleNamespace(name=name)
+
+    async def create(self, params: _CreateSnapshotParams) -> None:
+        self.built.append(params.name)
+
+    async def list(self) -> SimpleNamespace:
+        return SimpleNamespace(items=[SimpleNamespace(name=name) for name in self.built])
+
+    async def delete(self, snapshot: SimpleNamespace) -> None:
+        self.built.remove(snapshot.name)
+
+
 class _FakeDaytonaClient:
     def __init__(self, calls: dict[str, object]) -> None:
         self.calls = calls
         self.sandbox = _FakeDaytonaSandbox(calls)
+        self.snapshot = _FakeSnapshotApi(calls)
 
     async def create(self, params: object, **kwargs: object) -> _FakeDaytonaSandbox:
         self.calls["create"] = (params, kwargs["timeout"])
         return self.sandbox
 
     async def delete(self, sandbox: _FakeDaytonaSandbox) -> None:
+        if self.calls.get("delete_fails"):
+            raise RuntimeError("daytona API unreachable")
         self.calls["delete"] = sandbox.id
 
 
@@ -722,6 +758,182 @@ async def test_daytona_snapshot_sandboxes_disable_auto_stop(
     assert create_timeout == 120
 
 
+def _build_image(context: Path) -> SimpleNamespace:
+    """A stand-in for ``daytona.Image.from_dockerfile``: the Dockerfile text plus
+    the context entries the SDK would archive and upload."""
+    return SimpleNamespace(
+        dockerfile=lambda: "FROM python:3.11-slim\nCOPY . .\n",
+        _context_list=[SimpleNamespace(source_path=str(context), archive_path=".")],
+    )
+
+
+async def _boot_snapshot(context: Path, calls: dict[str, object]) -> str:
+    """Acquire once through a fresh provider; return the snapshot it booted."""
+    async with DaytonaRuntime("env", image=_build_image(context))(_row()):
+        pass
+    create_call = calls["create"]
+    assert isinstance(create_call, tuple)
+    snapshot = create_call[0].snapshot
+    assert isinstance(snapshot, str)
+    return snapshot
+
+
+async def test_daytona_rebuilds_the_snapshot_when_the_env_changes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # A name that doesn't track content resolves to whatever was built first.
+    calls = _install_fake_daytona(monkeypatch)
+    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
+
+    first = await _boot_snapshot(tmp_path, calls)
+    unchanged = await _boot_snapshot(tmp_path, calls)
+    (tmp_path / "env.py").write_text("REWARD = 0.0\n")
+    edited = await _boot_snapshot(tmp_path, calls)
+
+    assert unchanged == first
+    assert edited != first
+    assert first.startswith("env-")
+    # Built once per distinct content; the unchanged re-acquisition reused it.
+    assert calls["snapshot_create"] == [first, edited]
+
+
+async def test_daytona_snapshot_name_ignores_build_noise(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Daytona's archiver has no .dockerignore, so the host venv rides along in
+    # the context; digesting it would rebuild on every local `uv sync`.
+    calls = _install_fake_daytona(monkeypatch)
+    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
+    before = await _boot_snapshot(tmp_path, calls)
+
+    (tmp_path / ".venv" / "bin").mkdir(parents=True)
+    (tmp_path / ".venv" / "bin" / "python").write_text("#!/host/python\n")
+    (tmp_path / "__pycache__").mkdir()
+    (tmp_path / "__pycache__" / "env.cpython-311.pyc").write_bytes(b"\x00\x01")
+
+    assert await _boot_snapshot(tmp_path, calls) == before
+
+
+async def test_daytona_sends_cpu_as_a_whole_number(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Daytona's API rejects 2.0, and no equality assertion can catch it: 2.0 == 2.
+    calls = _install_fake_daytona(monkeypatch)
+    config = RuntimeConfig(image="img:tag", resources=RuntimeResources(cpu=2))
+
+    async with DaytonaRuntime(runtime_config=config)(_row()):
+        pass
+
+    create_call = calls["create"]
+    assert isinstance(create_call, tuple)
+    assert isinstance(create_call[0].resources.cpu, int)
+
+
+async def test_daytona_rejects_a_fractional_cpu_request(monkeypatch: pytest.MonkeyPatch) -> None:
+    _install_fake_daytona(monkeypatch)
+    config = RuntimeConfig(image="img:tag", resources=RuntimeResources(cpu=1.5))
+
+    with pytest.raises(ValueError, match="whole number of CPUs"):
+        async with DaytonaRuntime(runtime_config=config)(_row()):
+            pass
+
+
+async def test_daytona_names_a_sandbox_it_could_not_delete(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Teardown must report the leak without shadowing the run's own error.
+    calls = _install_fake_daytona(monkeypatch)
+    calls["delete_fails"] = True
+
+    with caplog.at_level(logging.WARNING, logger="hud.eval.runtime"):
+        async with DaytonaRuntime("snapshot")(_row()):
+            pass
+
+    assert "sandbox-1" in caplog.text
+
+
+async def test_daytona_sizes_each_row_from_its_own_runtime_config(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Resolving the name once would boot every row on the first row's cores.
+    calls = _install_fake_daytona(monkeypatch)
+    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
+    provider = DaytonaRuntime("env", image=_build_image(tmp_path))
+
+    booted = []
+    for cpu in (2, 4):
+        task = Task(
+            env="any-env",
+            id="t",
+            runtime_config=RuntimeConfig(resources=RuntimeResources(cpu=cpu)),
+        )
+        async with provider(task):
+            pass
+        create_call = calls["create"]
+        assert isinstance(create_call, tuple)
+        booted.append(create_call[0].snapshot)
+
+    assert booted[0] != booted[1]
+    assert calls["snapshot_create"] == booted
+
+
+async def test_daytona_prune_leaves_hand_named_snapshots_alone(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # "env-baseline" shares the prefix but is nothing this runtime minted.
+    calls = _install_fake_daytona(monkeypatch)
+    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
+    stale = await _boot_snapshot(tmp_path, calls)
+    created = calls["snapshot_create"]
+    assert isinstance(created, list)
+    created.append("env-baseline")
+
+    (tmp_path / "env.py").write_text("REWARD = 0.0\n")
+    provider = DaytonaRuntime("env", image=_build_image(tmp_path))
+    async with provider(_row()):
+        pass
+
+    assert await provider.prune_snapshots() == [stale]
+    assert "env-baseline" in created
+
+
+async def test_daytona_prune_keeps_only_the_snapshot_in_use(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    calls = _install_fake_daytona(monkeypatch)
+    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
+    stale = await _boot_snapshot(tmp_path, calls)
+    (tmp_path / "env.py").write_text("REWARD = 0.0\n")
+
+    provider = DaytonaRuntime("env", image=_build_image(tmp_path))
+    async with provider(_row()):
+        pass
+    current = await _boot_snapshot(tmp_path, calls)
+
+    assert await provider.prune_snapshots() == [stale]
+    assert calls["snapshot_create"] == [current]
+
+
+async def test_daytona_prune_refuses_before_a_snapshot_is_resolved(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Nothing to keep means every digest looks prunable, including a live one.
+    _install_fake_daytona(monkeypatch)
+    with pytest.raises(RuntimeError, match="resolved snapshot"):
+        await DaytonaRuntime("env").prune_snapshots()
+
+
+async def test_daytona_attaches_the_env_output_to_a_failed_handshake(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # The caller gets "closed connection during 'hello'" and nothing else.
+    _install_fake_daytona(monkeypatch)
+
+    with pytest.raises(EOFError) as excinfo:
+        async with DaytonaRuntime("snapshot")(_row()):
+            raise EOFError("env closed connection during 'hello'")
+
+    assert any("no module named bugs" in note for note in excinfo.value.__notes__)
+
+
 async def test_daytona_runtime_config_rejects_unsupported_fields(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
diff --git a/hud/eval/tests/test_rollout.py b/hud/eval/tests/test_rollout.py
index f1aafc410..972b38e94 100644
--- a/hud/eval/tests/test_rollout.py
+++ b/hud/eval/tests/test_rollout.py
@@ -425,6 +425,21 @@ async def broken_provider(task: TaskRow) -> AsyncIterator[Runtime]:
     assert run.runtime is None
 
 
+async def test_pre_launch_failure_keeps_the_notes_the_provider_attached() -> None:
+    # A provider attaches what only it can see — the env's output inside a remote
+    # sandbox — as a note, and str(exc) drops those on the way to the receipt.
+    @asynccontextmanager
+    async def broken_provider(task: TaskRow) -> AsyncIterator[Runtime]:
+        exc = EOFError("env closed connection during 'hello'")
+        exc.add_note("env output in sandbox ae964e25:\nModuleNotFoundError: No module named 'bugs'")
+        raise exc
+        yield  # pragma: no cover
+
+    run = await rollout(_add_task(1, 1), _FnAgent(_solve_add), runtime=broken_provider)
+
+    assert "No module named 'bugs'" in (run.trace.error or "")
+
+
 async def test_provider_is_called_with_the_task_row_being_placed(env_file: Path) -> None:
     placed: list[str] = []
 
diff --git a/hud/train/client.py b/hud/train/client.py
index e40353ea0..2c9b001fe 100644
--- a/hud/train/client.py
+++ b/hud/train/client.py
@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import logging
 from collections import Counter
 from typing import TYPE_CHECKING
 
@@ -44,6 +45,8 @@
         tuple["torch.Tensor", dict[str, float]],
     ]
 
+logger = logging.getLogger("hud.train")
+
 
 def _run_to_input(run: Run) -> TrainInput:
     """Turn a graded :class:`hud.Run` into a training input: inline payload when
@@ -106,6 +109,27 @@ def _check_groups(
         raise ValueError(f"incomplete GRPO groups: {incomplete}")
 
 
+def _check_reward_spread(
+    trajectories: Sequence[str | Run | TrajectoryPayload],
+    group_size: int | None,
+) -> None:
+    """Warn when no group has any reward spread: GRPO normalizes advantages within a
+    group, so identical rewards zero every one of them and the pass accumulates
+    nothing — which surfaces much later as a 409 from ``optim_step`` blaming the
+    wrong call. Only inline rewards are visible here; ``trace_id``s are skipped."""
+    rewards = [t.reward for t in trajectories if not isinstance(t, str)]
+    if len(rewards) != len(trajectories) or len(rewards) < 2:
+        return
+    size = group_size or len(rewards)
+    if all(len(set(rewards[start : start + size])) == 1 for start in range(0, len(rewards), size)):
+        logger.warning(
+            "every reward is identical within its GRPO group (%s), so all advantages are "
+            "zero and this pass accumulates no gradient — the next optim_step will fail. "
+            "Vary task difficulty until rollouts disagree.",
+            rewards[0],
+        )
+
+
 class TrainingClient(BaseTrainingClient):
     """Train an LLM model through the HUD training service.
 
@@ -138,6 +162,7 @@ async def forward_backward(
         """
         inputs = _to_inputs(trajectories)
         _check_groups(trajectories, group_size)
+        _check_reward_spread(trajectories, group_size)
         request = ForwardBackwardRequest(
             inputs=inputs,
             loss_fn=loss_fn,
diff --git a/hud/train/tests/test_client.py b/hud/train/tests/test_client.py
index 4f644baaa..6124bc7f0 100644
--- a/hud/train/tests/test_client.py
+++ b/hud/train/tests/test_client.py
@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import contextlib
+import logging
+
 import pytest
 
 from hud.eval.run import Run
@@ -22,3 +25,35 @@ async def test_training_rejects_incomplete_run_groups() -> None:
 
     with pytest.raises(ValueError, match="incomplete GRPO groups"):
         await TrainingClient("test-model").forward_backward(runs, group_size=2)
+
+
+async def test_training_warns_when_no_group_has_reward_spread(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # Identical rewards zero every GRPO advantage, so nothing accumulates and the
+    # next optim_step 409s with a message that blames the wrong call.
+    runs = [Run(None, "", {}) for _ in range(4)]
+    for index, run in enumerate(runs):
+        run.trace.trace_id = str(index)
+        run.group_id = "a" if index < 2 else "b"
+        run.grade.reward = 1.0
+
+    with caplog.at_level(logging.WARNING, logger="hud.train"), contextlib.suppress(Exception):
+        await TrainingClient("test-model").forward_backward(runs, group_size=2)
+
+    assert "no gradient" in caplog.text
+
+
+async def test_training_is_quiet_when_a_group_has_reward_spread(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    runs = [Run(None, "", {}) for _ in range(4)]
+    for index, run in enumerate(runs):
+        run.trace.trace_id = str(index)
+        run.group_id = "a" if index < 2 else "b"
+        run.grade.reward = float(index % 2)
+
+    with caplog.at_level(logging.WARNING, logger="hud.train"), contextlib.suppress(Exception):
+        await TrainingClient("test-model").forward_backward(runs, group_size=2)
+
+    assert "no gradient" not in caplog.text

From 8bf590d0c520f4e5e85fe6e5c7bb7398bc925864 Mon Sep 17 00:00:00 2001
From: Jaideep <67646710+jdchawla29@users.noreply.github.com>
Date: Wed, 29 Jul 2026 11:22:26 -0700
Subject: [PATCH 4/4] refactor: stable Daytona snapshot names, correct GRPO
 spread grouping

A snapshot built from image= is now checked against the build Daytona
recorded for it (Dockerfile text plus the SDK's own context hashes) and
rebuilt in place when they differ. Names stay durable handles: no digest
suffixes, no process-local registry, no prune_snapshots(), and a plain
registry-ref string keeps working as image=. Per-row resources build sized
variants under readable names (env-4cpu) since sizing is baked in at build.
Verified against the live API: dockerfile_content round-trips verbatim,
locally computed context hashes match the stored ones, and a deleted name
stays taken for ~10s, so the rebuild waits for it to free before creating.

The reward-spread warning forms groups by group_id the same way
_check_groups does; positional slicing misjudged interleaved batches. Its
tests stub the training service instead of suppressing a live request.

hud trace renders payloads as rich Text rather than escaping them into
markup strings, and the SSH tool error helper folds into the module's
existing imports.
---
 docs/v6/reference/runtime.mdx                 |   6 +-
 .../tests/test_provider_native_tools.py       |  34 ++-
 hud/agents/tools/ssh.py                       |  23 +-
 hud/capabilities/base.py                      |  17 +-
 hud/capabilities/ssh.py                       |  12 +-
 hud/cli/trace.py                              |  22 +-
 hud/environment/tests/test_workspace.py       |   4 +-
 hud/environment/workspace.py                  |   4 +-
 hud/eval/run.py                               |  16 +-
 hud/eval/runtime.py                           | 186 +++++++--------
 hud/eval/tests/test_docker_provider.py        | 219 +++++++++---------
 hud/train/client.py                           |  65 +++---
 hud/train/tests/test_client.py                |  65 ++++--
 13 files changed, 337 insertions(+), 336 deletions(-)

diff --git a/docs/v6/reference/runtime.mdx b/docs/v6/reference/runtime.mdx
index 4e60103e2..cb378f708 100644
--- a/docs/v6/reference/runtime.mdx
+++ b/docs/v6/reference/runtime.mdx
@@ -152,13 +152,11 @@ DaytonaRuntime(snapshot_name=None, *, image=None, command=None, workdir="/app",
 ```
 
 - **`snapshot_name`** - Daytona snapshot to boot from (the durable handle).
-- **`image`** - Dockerfile/registry ref to build the snapshot once if it's missing. Resources (cpu/memory/gpu) live on the snapshot. A snapshot built from `image` is content-addressed — its name carries a digest of the Dockerfile and everything it copies in (`my-env-9f2c1d0ab3e4`), so editing the env boots a rebuilt snapshot instead of silently reusing the one from before the edit. The digest is part of the stored name, so `DaytonaRuntime("my-env")` with no `image` will not find what `DaytonaRuntime("my-env", image=…)` built: keep passing `image` (it re-resolves to the same snapshot for free when nothing changed), or pass the full digested name.
+- **`image`** - Dockerfile/registry ref to build the snapshot if it's missing. Daytona records what a snapshot was built from, so when the image content changes (an edited Dockerfile or context file, a repointed registry ref) the snapshot is rebuilt in place under the same name instead of silently reusing the build from before the edit.
 - **`workdir`** / **`port`** - guest working directory and in-sandbox serving port.
 - **`ssh_host`** / **`ssh_expires_minutes`** - SSH tunnel settings (Daytona exposes services over an SSH local-forward).
 
-`runtime_config.resources` (cpu/memory/gpu) is applied when the snapshot is built from `image`, and is part of the digest — same image at a different size is a different snapshot. Booting a snapshot you did not build (`snapshot_name` with no `image`) cannot resize it, since resources are fixed at build time.
-
-`await runtime.prune_snapshots()` deletes snapshots under this `snapshot_name` that this runtime has not booted — content-addressing costs one snapshot per edit, and nothing else reaps them. Every digest the instance booted is kept (a taskset with per-row `resources` uses several at once), so reaping across env edits means pruning from a fresh runtime.
+Resources (cpu/memory/gpu) are fixed on the snapshot at build time. With `image`, a task's `runtime_config.resources` builds a sized variant under a suffixed name (`my-env-4cpu`); without `image`, an already-built snapshot cannot be resized.
 
 ### `HUDRuntime`
 
diff --git a/hud/agents/tests/test_provider_native_tools.py b/hud/agents/tests/test_provider_native_tools.py
index 0272df301..014990f79 100644
--- a/hud/agents/tests/test_provider_native_tools.py
+++ b/hud/agents/tests/test_provider_native_tools.py
@@ -49,10 +49,19 @@ async def run(
         parts = shlex.split(command)
         if len(parts) == 3 and parts[:2] == ["cat", "--"]:
             if parts[2] not in self._store:
-                return self._finish(
-                    command,
-                    check,
-                    _Completed(stderr=f"cat: {parts[2]}: No such file or directory", exit_status=1),
+                if check:
+                    raise asyncssh.ProcessError(
+                        env=None,
+                        command=command,
+                        subsystem=None,
+                        exit_status=1,
+                        exit_signal=None,
+                        returncode=1,
+                        stdout="",
+                        stderr=f"cat: {parts[2]}: No such file or directory",
+                    )
+                return _Completed(
+                    stderr=f"cat: {parts[2]}: No such file or directory", exit_status=1
                 )
             return _Completed(stdout=self._store[parts[2]].decode())
         if len(parts) == 3 and parts[:2] == ["cat", ">"]:
@@ -82,21 +91,6 @@ async def run(
             return _Completed(exit_status=0)
         return self._completed
 
-    @staticmethod
-    def _finish(command: str, check: bool, completed: _Completed) -> _Completed:
-        if check and completed.exit_status:
-            raise asyncssh.ProcessError(
-                env=None,
-                command=command,
-                subsystem=None,
-                exit_status=completed.exit_status,
-                exit_signal=None,
-                returncode=completed.exit_status,
-                stdout=completed.stdout,
-                stderr=completed.stderr,
-            )
-        return completed
-
 
 class _FakeSSH(SSHClient):
     """SSH client with an in-memory exec-channel filesystem."""
@@ -551,7 +545,7 @@ def test_map_path_leaves_a_symlinked_spelling_of_the_workspace_alone() -> None:
         name="shell",
         protocol="ssh/2",
         url="ssh://localhost:22",
-        params={"cwd": "/private/tmp/w", "cwd_alias": "/tmp/w"},
+        params={"cwd": "/private/tmp/w", "cwd_aliases": ["/tmp/w"]},
     )
     ssh = SSHClient(cap, cast("Any", None))
 
diff --git a/hud/agents/tools/ssh.py b/hud/agents/tools/ssh.py
index b88707271..a6c9ce9e7 100644
--- a/hud/agents/tools/ssh.py
+++ b/hud/agents/tools/ssh.py
@@ -10,11 +10,19 @@
 import asyncssh
 import mcp.types as mcp_types
 
-from hud.agents.tools.base import AgentTool
+from hud.agents.tools.base import AgentTool, tool_err, tool_ok
 from hud.capabilities import SSHClient
 from hud.types import MCPToolResult
 
 
+def _remote_error(exc: asyncssh.ProcessError) -> str:
+    """What the remote command printed to stderr — a failed file op is an
+    ordinary tool outcome, so the agent gets the shell's message, not the
+    exception's repr."""
+    stderr = exc.stderr.decode("utf-8", "replace") if isinstance(exc.stderr, bytes) else exc.stderr
+    return (stderr or "").strip() or f"exit {exc.exit_status}"
+
+
 class SSHTool(AgentTool[SSHClient]):
     """Capability base: tool driven by an ``SSHClient``."""
 
@@ -60,17 +68,4 @@ async def file_list(self, path: str = "/") -> MCPToolResult:
         return tool_ok("\n".join(names) if names else "(empty)")
 
 
-def _remote_error(exc: asyncssh.ProcessError) -> str:
-    """The line the remote command printed, not asyncssh's whole connection repr.
-
-    A failed file command is an ordinary tool outcome — reading a file that does
-    not exist yet is the first thing an editor tool does — so the agent gets what
-    the shell said, not a ~30-line traceback.
-    """
-    stderr = exc.stderr.decode("utf-8", "replace") if isinstance(exc.stderr, bytes) else exc.stderr
-    return stderr.strip() or f"exit {exc.exit_status}"
-
-
-from hud.agents.tools.base import tool_err, tool_ok  # noqa: E402
-
 __all__ = ["SSHTool"]
diff --git a/hud/capabilities/base.py b/hud/capabilities/base.py
index 7c243fd62..d8d9548b8 100644
--- a/hud/capabilities/base.py
+++ b/hud/capabilities/base.py
@@ -7,9 +7,12 @@
 import sys
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import Any, ClassVar, Self
+from typing import TYPE_CHECKING, Any, ClassVar, Self
 from urllib.parse import urlsplit
 
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
 #: Matches the scheme prefix of a URL (RFC 3986).
 SCHEME_RE: re.Pattern[str] = re.compile(r"^([a-zA-Z][a-zA-Z0-9+\-.]*):")
 
@@ -79,7 +82,7 @@ def ssh(
         client_key_path: str | os.PathLike[str] | None = None,
         shell: str | None = None,
         cwd: str | None = None,
-        cwd_alias: str | None = None,
+        cwd_aliases: Sequence[str] | None = None,
     ) -> Capability:
         """``ssh/2`` — SSH daemon with publickey auth.
 
@@ -91,9 +94,9 @@ def ssh(
         from ``sys.platform`` at construction time. Agents read this to
         format commands correctly. ``cwd`` is the absolute path sessions
         start in (the served workspace); clients anchor file paths to it.
-        ``cwd_alias`` is a second spelling of that same directory (a path
-        that reaches it through a symlink), which clients treat as already
-        anchored rather than as a workspace-relative address.
+        ``cwd_aliases`` are other names the same directory answers to (paths
+        that reach it through symlinks), which clients treat as already
+        anchored rather than as workspace-relative addresses.
         """
         normalized = normalize_url(url, default_scheme="ssh", default_port=22)
         if shell is None:
@@ -105,8 +108,8 @@ def ssh(
             params["client_key_path"] = os.fspath(client_key_path)
         if cwd is not None:
             params["cwd"] = cwd
-        if cwd_alias is not None:
-            params["cwd_alias"] = cwd_alias
+        if cwd_aliases:
+            params["cwd_aliases"] = list(cwd_aliases)
         return cls(name=name, protocol="ssh/2", url=normalized, params=params)
 
     @classmethod
diff --git a/hud/capabilities/ssh.py b/hud/capabilities/ssh.py
index c3372c6b9..a07fbed54 100644
--- a/hud/capabilities/ssh.py
+++ b/hud/capabilities/ssh.py
@@ -57,14 +57,16 @@ def map_path(self, path: str) -> str:
         normalize the remainder against ``/`` (clamping ``..`` at the root,
         exactly as a chroot does), and re-anchor under the cwd. Idempotent.
 
-        ``cwd_alias`` is the same directory reached through a symlink, and is
-        stripped like the cwd: re-anchoring it instead would turn an already
-        correct absolute path into a nested one that does not exist.
+        ``cwd_aliases`` are the same directory reached through symlinks, and are
+        stripped like the cwd: re-anchoring one would turn an already correct
+        absolute path into a nested one that does not exist.
         """
         cwd = str(self.capability.params.get("cwd", "")).rstrip("/")
         if not cwd:
             return path
-        alias = str(self.capability.params.get("cwd_alias", "")).rstrip("/")
+        aliases = [
+            str(alias).rstrip("/") for alias in self.capability.params.get("cwd_aliases") or []
+        ]
         if self._is_windows:
             # The workspace publishes cwd via as_posix() (e.g. "C:/work") but
             # callers pass native paths ("C:\work\file.txt"); NTFS paths are
@@ -76,7 +78,7 @@ def map_path(self, path: str) -> str:
                 # Drive-absolute outside the workspace: anchor like the chroot.
                 path = path[2:]
         else:
-            for prefix in (cwd, alias):
+            for prefix in (cwd, *aliases):
                 if prefix and (path == prefix or path.startswith(prefix + "/")):
                     path = path[len(prefix) :]
                     break
diff --git a/hud/cli/trace.py b/hud/cli/trace.py
index 3d1a1a555..077ea47d9 100644
--- a/hud/cli/trace.py
+++ b/hud/cli/trace.py
@@ -8,7 +8,6 @@
 
 import typer
 from rich.console import Console
-from rich.markup import escape
 from rich.panel import Panel
 from rich.rule import Rule
 from rich.text import Text
@@ -162,7 +161,8 @@ def _load_remote(trace_id: str) -> list[dict[str, Any]]:
 
 
 def _render_events(events: list[dict[str, Any]]) -> None:
-    # Payloads are escaped: rich reads `[len(s) // 2]` as a style tag and drops it
+    # Payloads render as Text, never as markup: rich would read a literal
+    # `[len(s) // 2]` in agent output as a style tag and drop it.
     turn = 0
     for ev in events:
         kind = ev.get("kind")
@@ -177,7 +177,7 @@ def _render_events(events: list[dict[str, Any]]) -> None:
 
             text = ev.get("text")
             if text:
-                console.print(escape(str(text)))
+                console.print(Text(str(text)))
 
             for tc in ev.get("tool_calls") or []:
                 name = tc.get("name") or tc.get("function", {}).get("name", "?")
@@ -186,30 +186,30 @@ def _render_events(events: list[dict[str, Any]]) -> None:
                     with contextlib.suppress(Exception):
                         args = json.loads(args)
                 console.print(
-                    f"  [green]→[/green] [bold]{escape(str(name))}[/bold]"
-                    f"({escape(_fmt_args(args))})",
-                    highlight=False,
+                    Text.assemble(
+                        "  ", ("→", "green"), " ", (str(name), "bold"), f"({_fmt_args(args)})"
+                    )
                 )
 
             if ev.get("error"):
-                console.print(f"  [red]error: {escape(str(ev['error']))}[/red]")
+                console.print(Text(f"  error: {ev['error']}", style="red"))
 
         elif kind in ("tool_call", "tool_result"):
             name = ev.get("tool_name") or ev.get("name") or "?"
             result = ev.get("result_text") or ev.get("result") or ""
             error = ev.get("error")
             if error:
-                console.print(f"  [red]✗ {escape(str(name))}: {escape(str(error))}[/red]")
+                console.print(Text(f"  ✗ {name}: {error}", style="red"))
             else:
-                console.print(f"  [dim]{escape(str(name))} →[/dim]")
+                console.print(Text(f"  {name} →", style="dim"))
                 for line in str(result).splitlines():
-                    console.print(f"    {escape(line)}")
+                    console.print(Text(f"    {line}"))
 
         elif kind == "environment":
             msg = ev.get("text") or ev.get("content") or ""
             if msg:
                 console.print(Rule("[yellow]env[/yellow]", style="yellow"))
-                console.print(escape(str(msg)))
+                console.print(Text(str(msg)))
 
 
 def _fmt_args(args: Any) -> str:
diff --git a/hud/environment/tests/test_workspace.py b/hud/environment/tests/test_workspace.py
index 90c10d0bc..ac18a50d8 100644
--- a/hud/environment/tests/test_workspace.py
+++ b/hud/environment/tests/test_workspace.py
@@ -285,7 +285,7 @@ async def test_a_symlinked_root_publishes_both_spellings(tmp_path: Path) -> None
     try:
         cap = ws.capability()
         assert cap.params["cwd"] == real.as_posix()
-        assert cap.params["cwd_alias"] == link.as_posix()
+        assert cap.params["cwd_aliases"] == [link.as_posix()]
 
         client = SSHClient(cap, cast("Any", None))
         assert client.map_path(f"{link}/calc.py") == f"{real}/calc.py"
@@ -298,6 +298,6 @@ async def test_a_plain_root_publishes_no_alias(tmp_path: Path) -> None:
     ws = Workspace(tmp_path / "root")
     await ws.start()
     try:
-        assert "cwd_alias" not in ws.capability().params
+        assert "cwd_aliases" not in ws.capability().params
     finally:
         await ws.stop()
diff --git a/hud/environment/workspace.py b/hud/environment/workspace.py
index ab3a2c26f..48c4f5a8e 100644
--- a/hud/environment/workspace.py
+++ b/hud/environment/workspace.py
@@ -209,7 +209,7 @@ def __init__(
         # workspace-relative address and re-anchors it somewhere that does not exist.
         given = Path(root).absolute().as_posix()
         real = self.root.as_posix()
-        self._cwd_alias = given if given != real and self._guest_path == real else None
+        self._cwd_aliases = [given] if given != real and self._guest_path == real else []
         # ssh config
         self._ssh_host = host
         self._ssh_port = port
@@ -437,7 +437,7 @@ def capability(self, name: str = "shell") -> Capability:
             client_key=key_path.read_text() if key_path else None,
             client_key_path=key_path,
             cwd=self._guest_path,
-            cwd_alias=self._cwd_alias,
+            cwd_aliases=self._cwd_aliases or None,
         )
 
     @property
diff --git a/hud/eval/run.py b/hud/eval/run.py
index 76075cb06..e8915c744 100644
--- a/hud/eval/run.py
+++ b/hud/eval/run.py
@@ -24,6 +24,7 @@
 import asyncio
 import contextlib
 import logging
+import traceback
 import uuid
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Self, cast
@@ -388,7 +389,10 @@ async def _drive() -> None:
             driver.add_done_callback(_consume_task_result)
             raise
         except Exception as exc:
-            detail = _detail(exc)
+            # format_exception_only keeps __notes__ — a provider attaches what
+            # only it can see there, like the sandbox's env output on a failed
+            # handshake — where str(exc) would drop them.
+            detail = "".join(traceback.format_exception_only(exc)).strip()
             if run is None:
                 logger.warning("rollout failed before launch (%s): %s", _phase, detail)
                 run = Run.failed(f"[{_phase}] {detail}")
@@ -405,16 +409,6 @@ async def _drive() -> None:
     return run
 
 
-def _detail(exc: BaseException) -> str:
-    """The failure text plus any notes attached to it.
-
-    Providers attach what only they can see — a sandbox's env output on a failed
-    handshake — as notes. ``str(exc)`` drops them, so a rollout that died at
-    import would report ``closed connection during 'hello'`` and nothing else.
-    """
-    return "\n".join([str(exc), *getattr(exc, "__notes__", [])])
-
-
 def _consume_task_result(task: asyncio.Future[Any]) -> None:
     with contextlib.suppress(asyncio.CancelledError, Exception):
         task.result()
diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py
index a8c38a572..08cbd64ed 100644
--- a/hud/eval/runtime.py
+++ b/hud/eval/runtime.py
@@ -32,16 +32,14 @@
 
 import asyncio
 import contextlib
-import hashlib
 import logging
-import re
 import sys
 import uuid
 from collections import deque
 from contextlib import AbstractAsyncContextManager, asynccontextmanager, nullcontext
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Protocol
+from typing import TYPE_CHECKING, Any, Protocol, cast
 from urllib.parse import urlsplit, urlunsplit
 
 import httpx
@@ -653,52 +651,33 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                 await sb.terminate.aio()
 
 
-# Build noise that must not perturb the snapshot digest: these are uploaded with
-# the context (Daytona's archiver has no .dockerignore) but never define the env,
-# so hashing them would rebuild the snapshot every time the host venv is touched.
-_DIGEST_PATTERN = re.compile(r"[0-9a-f]{12}")
+async def _snapshot_is_current(snapshot: Any, image: Any) -> bool:
+    """Whether *snapshot* was built from *image* as it exists right now.
 
-_DIGEST_SKIP = frozenset(
-    {".git", ".venv", "venv", "__pycache__", ".pytest_cache", ".ruff_cache", ".mypy_cache"}
-)
-
-
-def _snapshot_digest(context_digest: str, resources: Any) -> str:
-    """Name suffix for a snapshot: its content plus the sizing baked into it."""
-    return hashlib.sha256(f"{context_digest}{resources!r}".encode()).hexdigest()[:12]
-
-
-def _image_digest(image: Any) -> str:
-    """Content hash of a Daytona ``Image``: its Dockerfile plus every byte it copies in.
-
-    A Daytona snapshot is identified by name alone, so the name has to carry the
-    content — otherwise an edited env resolves to the snapshot built before the
-    edit and every rollout after it silently measures the old code. ``_context_list``
-    is the same private attribute ``snapshot.create`` reads to decide what to upload.
+    Daytona records what a snapshot was built from — the registry ref, or for a
+    built ``Image`` its Dockerfile text plus the hashes of the context it
+    uploads (``build_info``). The hashes are recomputed here with the SDK's own
+    hasher so they are comparable to what ``snapshot.create`` would upload.
+    ``_context_list`` is the same private attribute ``snapshot.create`` reads.
     """
-    digest = hashlib.sha256()
-    digest.update(image.dockerfile().encode())
-    for entry in image._context_list:
-        source = Path(entry.source_path)
-        is_dir = source.is_dir()
-        for file in sorted(source.rglob("*")) if is_dir else [source]:
-            # as_posix so the same tree digests identically on every host OS.
-            rel = file.relative_to(source).as_posix() if is_dir else file.name
-            if not file.is_file() or _DIGEST_SKIP & set(rel.split("/")):
-                continue
-            digest.update(rel.encode())
-            digest.update(file.read_bytes())
-    return digest.hexdigest()[:12]
-
-
-async def _daytona_serve_output(sandbox: Any, session: str, cmd_id: str) -> str:
-    """What the env process printed inside the sandbox, for a failed handshake."""
-    try:
-        logs = await sandbox.process.get_session_command_logs(session, cmd_id)
-    except Exception as e:  # the sandbox may already be gone
-        return f"env output unavailable: {e}"
-    output = (logs.stderr or logs.output or logs.stdout or "").strip()
-    return f"env output in sandbox {sandbox.id}:\n{output}" if output else "env printed nothing"
+    if isinstance(image, str):
+        return bool(snapshot.image_name == image)
+    build = snapshot.build_info
+    if build is None:
+        return False
+    from daytona._async.object_storage import AsyncObjectStorage
+
+    # The hasher is an instance method only for code organization; credentials
+    # are needed to upload, not to hash, so skip the credentialed __init__.
+    storage = cast("Any", AsyncObjectStorage.__new__(AsyncObjectStorage))
+    hashes = [
+        await storage._compute_hash_for_path_md5(entry.source_path, entry.archive_path)
+        for entry in image._context_list
+    ]
+    return bool(
+        build.dockerfile_content == image.dockerfile()
+        and list(build.context_hashes or []) == hashes
+    )
 
 
 class DaytonaRuntime:
@@ -712,13 +691,11 @@ class DaytonaRuntime:
     port. Yields its :class:`Runtime`, deletes the sandbox on exit.
 
     Pass a snapshot name — ``DaytonaRuntime("hud-libero-env")`` — optionally with an
-    ``image`` (Dockerfile/registry ref) to build that snapshot once if it is missing.
-    A built snapshot is content-addressed: its name carries a digest of the image
-    and everything the image copies in (``hud-libero-env-9f2c1d0ab3e4``), so editing
-    the env boots a rebuilt snapshot instead of silently reusing the stale one. That
-    digest is the name Daytona stores, so booting it later by name alone needs the
-    full ``hud-libero-env-9f2c1d0ab3e4`` — keep passing ``image`` instead, which
-    re-resolves to the same snapshot for free when nothing changed.
+    ``image`` (Dockerfile/registry ref) to build that snapshot if it is missing.
+    With *image*, an existing snapshot is compared against the image's recorded
+    build (Dockerfile plus context hashes) and rebuilt under the same name when
+    they differ, so editing the env never silently reuses the snapshot built
+    before the edit.
     Resources (cpu/memory/gpu) live on the snapshot, not here. *workdir* defaults to
     ``/app`` (the scaffolded ``Dockerfile.hud`` WORKDIR) since a Daytona session
     starts in ``~``, not the image's WORKDIR; override only for a non-standard layout.
@@ -751,43 +728,12 @@ def __init__(
         if runtime_config is not None:
             config = RuntimeConfig.model_validate(runtime_config)
         self.runtime_config = config
+        # Resolve each snapshot name against the image once; lock so concurrent
+        # first acquisitions resolve exactly once.
         self._image = image
-        # The context is walked once; resources come from the row, so they fold
-        # into the name per acquisition.
-        self._context_digest: str | None = None
-        self._ensured: set[str] = set()
+        self._resolved: set[str] = set()
         self._snapshot_lock = asyncio.Lock()
 
-    async def prune_snapshots(self) -> list[str]:
-        """Delete snapshots under this *snapshot_name* that this runtime has not booted.
-
-        Content-addressed names cost one snapshot per edit and nothing reaps them.
-        Every digest this instance booted is kept — a taskset with per-row resources
-        uses several at once — so reaping across edits means pruning from a fresh
-        runtime, not from one that already booted the old digest.
-
-        Only names this runtime could have minted are touched: the prefix plus a
-        digest, never a hand-named snapshot that shares it.
-        """
-        from daytona import AsyncDaytona
-
-        if not self._ensured:
-            raise RuntimeError("prune_snapshots needs a resolved snapshot: run a rollout first")
-
-        prefix = f"{self.snapshot_name}-"
-        deleted: list[str] = []
-        async with AsyncDaytona() as daytona:
-            page = await daytona.snapshot.list()
-            for snapshot in page.items:
-                name = str(snapshot.name)
-                if name in self._ensured or not name.startswith(prefix):
-                    continue
-                if _DIGEST_PATTERN.fullmatch(name[len(prefix) :]):
-                    await daytona.snapshot.delete(snapshot)
-                    deleted.append(name)
-        logger.info("pruned %d Daytona snapshot(s), kept %s", len(deleted), sorted(self._ensured))
-        return deleted
-
     @asynccontextmanager
     async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
         import asyncssh
@@ -852,18 +798,45 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                     )
                 snapshot_name = self.snapshot_name
                 if self._image is not None:
+                    if daytona_resources is not None:
+                        # Sizing is baked in at build time, so each sizing is its
+                        # own snapshot under a readable suffix (env-4cpu-8gb).
+                        sizing = []
+                        if daytona_resources.cpu:
+                            sizing.append(f"{daytona_resources.cpu}cpu")
+                        if daytona_resources.memory:
+                            sizing.append(f"{daytona_resources.memory}gb")
+                        if daytona_resources.gpu:
+                            sizing.append(f"{daytona_resources.gpu}gpu")
+                            sizing.extend(
+                                (t.value if isinstance(t, GpuType) else t).lower()
+                                for t in daytona_resources.gpu_type or []
+                            )
+                        snapshot_name = "-".join([snapshot_name, *sizing])
                     async with self._snapshot_lock:
-                        if self._context_digest is None:
-                            self._context_digest = _image_digest(self._image)
-                        snapshot_name = (
-                            f"{self.snapshot_name}-"
-                            f"{_snapshot_digest(self._context_digest, daytona_resources)}"
-                        )
-                        if snapshot_name not in self._ensured:
+                        if snapshot_name not in self._resolved:
                             try:
-                                await daytona.snapshot.get(snapshot_name)
-                                logger.info("reusing Daytona snapshot %s", snapshot_name)
+                                existing = await daytona.snapshot.get(snapshot_name)
                             except DaytonaNotFoundError:
+                                existing = None
+                            if existing is not None and not await _snapshot_is_current(
+                                existing, self._image
+                            ):
+                                logger.info(
+                                    "Daytona snapshot %s is stale; rebuilding", snapshot_name
+                                )
+                                await daytona.snapshot.delete(existing)
+                                # Deletion frees the name asynchronously (~10s
+                                # observed); creating before it lands conflicts.
+                                async with asyncio.timeout(120):
+                                    while True:
+                                        try:
+                                            await daytona.snapshot.get(snapshot_name)
+                                        except DaytonaNotFoundError:
+                                            break
+                                        await asyncio.sleep(0.5)
+                                existing = None
+                            if existing is None:
                                 logger.info("building Daytona snapshot %s", snapshot_name)
                                 await daytona.snapshot.create(
                                     CreateSnapshotParams(
@@ -872,7 +845,7 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                                         resources=daytona_resources,
                                     )
                                 )
-                            self._ensured.add(snapshot_name)
+                            self._resolved.add(snapshot_name)
                 sandbox_params = CreateSandboxFromSnapshotParams(
                     snapshot=snapshot_name,
                     ephemeral=True,
@@ -910,8 +883,21 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                             config=config if config.model_dump(exclude_none=True) else None,
                         )
                     except (EOFError, OSError) as exc:
-                        # Why it died only exists inside the sandbox.
-                        exc.add_note(await _daytona_serve_output(sandbox, session, started.cmd_id))
+                        # Why it died only exists inside the sandbox, and the
+                        # sandbox may already be gone.
+                        try:
+                            logs = await sandbox.process.get_session_command_logs(
+                                session, started.cmd_id
+                            )
+                        except Exception as log_exc:
+                            exc.add_note(f"env output unavailable: {log_exc}")
+                        else:
+                            output = (logs.stderr or logs.output or logs.stdout or "").strip()
+                            exc.add_note(
+                                f"env output in sandbox {sandbox.id}:\n{output}"
+                                if output
+                                else "env printed nothing"
+                            )
                         raise
             finally:
                 try:
diff --git a/hud/eval/tests/test_docker_provider.py b/hud/eval/tests/test_docker_provider.py
index e5f3a58c6..b0dabf6c5 100644
--- a/hud/eval/tests/test_docker_provider.py
+++ b/hud/eval/tests/test_docker_provider.py
@@ -9,12 +9,14 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import logging
 import os
 import sys
 from dataclasses import dataclass
-from pathlib import Path  # noqa: TC003  # runtime use in _install_fake_docker
+from pathlib import Path
 from types import ModuleType, SimpleNamespace
+from typing import Any
 
 import pytest
 
@@ -229,7 +231,7 @@ class _DaytonaResources:
 
 @dataclass(frozen=True)
 class _DaytonaGpuType:
-    name: str
+    value: str
 
 
 @dataclass(frozen=True)
@@ -269,41 +271,89 @@ async def create_ssh_access(self, *, expires_in_minutes: int) -> SimpleNamespace
         return SimpleNamespace(token="ssh-token")
 
 
+async def _tree_hash(path_str: str) -> str:
+    """Deterministic fingerprint of a context tree: same tree, same hash."""
+    digest = hashlib.md5(usedforsecurity=False)
+    root = Path(path_str)
+    for file in sorted(p for p in root.rglob("*") if p.is_file()):
+        digest.update(file.relative_to(root).as_posix().encode())
+        digest.update(file.read_bytes())
+    return digest.hexdigest()
+
+
+class _FakeObjectStorage:
+    """The SDK hasher the provider borrows to predict a context's upload hash."""
+
+    async def _compute_hash_for_path_md5(
+        self, path_str: str, archive_base_path: str | None = None
+    ) -> str:
+        return await _tree_hash(path_str)
+
+
 class _FakeSnapshotApi:
     """The registry a Daytona snapshot name resolves against: ``get`` 404s until
-    something with that exact name has been built."""
+    something with that exact name has been built, and each snapshot records what
+    it was built from (``image_name`` or ``build_info``), as the server does.
+    A deleted name stays taken for a couple more ``get``s — the live API frees
+    it asynchronously (~10s), and creating before then conflicts."""
 
-    def __init__(self, calls: dict[str, object]) -> None:
-        self.built: list[str] = []
-        calls["snapshot_create"] = self.built
+    def __init__(self) -> None:
+        self.snapshots: dict[str, SimpleNamespace] = {}
+        self.builds: list[str] = []
+        self._deleting: dict[str, SimpleNamespace] = {}
+        self._deleting_gets: dict[str, int] = {}
 
     async def get(self, name: str) -> SimpleNamespace:
-        if name not in self.built:
+        if name in self._deleting:
+            self._deleting_gets[name] -= 1
+            if self._deleting_gets[name] >= 0:
+                return self._deleting[name]
+            del self._deleting[name], self._deleting_gets[name]
+        if name not in self.snapshots:
             raise RuntimeError(f"snapshot {name} not found")
-        return SimpleNamespace(name=name)
+        return self.snapshots[name]
 
     async def create(self, params: _CreateSnapshotParams) -> None:
-        self.built.append(params.name)
-
-    async def list(self) -> SimpleNamespace:
-        return SimpleNamespace(items=[SimpleNamespace(name=name) for name in self.built])
+        if params.name in self.snapshots or params.name in self._deleting:
+            raise ValueError(f"snapshot with name {params.name} already exists")
+        image: Any = params.image
+        if isinstance(image, str):
+            record = SimpleNamespace(name=params.name, image_name=image, build_info=None)
+        else:
+            record = SimpleNamespace(
+                name=params.name,
+                image_name=None,
+                build_info=SimpleNamespace(
+                    dockerfile_content=image.dockerfile(),
+                    context_hashes=[
+                        await _tree_hash(entry.source_path) for entry in image._context_list
+                    ],
+                ),
+            )
+        self.snapshots[params.name] = record
+        self.builds.append(params.name)
 
     async def delete(self, snapshot: SimpleNamespace) -> None:
-        self.built.remove(snapshot.name)
+        self._deleting[snapshot.name] = self.snapshots.pop(snapshot.name)
+        self._deleting_gets[snapshot.name] = 1
 
 
 class _FakeDaytonaClient:
     def __init__(self, calls: dict[str, object]) -> None:
         self.calls = calls
         self.sandbox = _FakeDaytonaSandbox(calls)
-        self.snapshot = _FakeSnapshotApi(calls)
+        self.snapshot = _FakeSnapshotApi()
+        #: sandbox-create params in order; the last entry is the booted sandbox.
+        self.created: list[Any] = []
+        self.delete_fails = False
 
     async def create(self, params: object, **kwargs: object) -> _FakeDaytonaSandbox:
         self.calls["create"] = (params, kwargs["timeout"])
+        self.created.append(params)
         return self.sandbox
 
     async def delete(self, sandbox: _FakeDaytonaSandbox) -> None:
-        if self.calls.get("delete_fails"):
+        if self.delete_fails:
             raise RuntimeError("daytona API unreachable")
         self.calls["delete"] = sandbox.id
 
@@ -342,10 +392,12 @@ async def __aexit__(self, *exc_info: object) -> None:
         self._calls["ssh_closed"] = True
 
 
-def _install_fake_daytona(monkeypatch: pytest.MonkeyPatch) -> dict[str, object]:
+def _install_fake_daytona(monkeypatch: pytest.MonkeyPatch) -> _FakeDaytonaClient:
     calls: dict[str, object] = {}
     client = _FakeDaytonaClient(calls)
     daytona = ModuleType("daytona")
+    daytona_async = ModuleType("daytona._async")
+    object_storage = ModuleType("daytona._async.object_storage")
     asyncssh = ModuleType("asyncssh")
 
     class AsyncDaytona:
@@ -367,10 +419,15 @@ def connect(*args: object, **kwargs: object) -> _FakeSSHConnect:
     setattr(daytona, "Resources", _DaytonaResources)
     setattr(daytona, "GpuType", _DaytonaGpuType)
     setattr(daytona, "SessionExecuteRequest", _SessionExecuteRequest)
+    setattr(daytona, "_async", daytona_async)
+    setattr(daytona_async, "object_storage", object_storage)
+    setattr(object_storage, "AsyncObjectStorage", _FakeObjectStorage)
     setattr(asyncssh, "connect", connect)
     monkeypatch.setitem(sys.modules, "daytona", daytona)
+    monkeypatch.setitem(sys.modules, "daytona._async", daytona_async)
+    monkeypatch.setitem(sys.modules, "daytona._async.object_storage", object_storage)
     monkeypatch.setitem(sys.modules, "asyncssh", asyncssh)
-    return calls
+    return client
 
 
 async def test_acquisition_publishes_ephemeral_port_and_removes_container(
@@ -656,7 +713,7 @@ async def test_modal_runtime_applies_env_vars_to_image(
 async def test_daytona_runtime_config_flows_into_daytona_sdk(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    calls = _install_fake_daytona(monkeypatch)
+    calls = _install_fake_daytona(monkeypatch).calls
     config = RuntimeConfig(
         image="img:tag",
         resources=RuntimeResources(
@@ -711,7 +768,7 @@ async def test_daytona_runtime_config_flows_into_daytona_sdk(
 async def test_daytona_task_runtime_config_overlays_provider_defaults(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    calls = _install_fake_daytona(monkeypatch)
+    calls = _install_fake_daytona(monkeypatch).calls
     provider = DaytonaRuntime(
         runtime_config=RuntimeConfig(
             resources=RuntimeResources(cpu=2, memory_mb=4096),
@@ -742,7 +799,7 @@ async def test_daytona_task_runtime_config_overlays_provider_defaults(
 async def test_daytona_snapshot_sandboxes_disable_auto_stop(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    calls = _install_fake_daytona(monkeypatch)
+    calls = _install_fake_daytona(monkeypatch).calls
 
     async with DaytonaRuntime("snapshot")(_row()):
         pass
@@ -767,64 +824,57 @@ def _build_image(context: Path) -> SimpleNamespace:
     )
 
 
-async def _boot_snapshot(context: Path, calls: dict[str, object]) -> str:
+async def _boot_snapshot(context: Path, daytona: _FakeDaytonaClient) -> str:
     """Acquire once through a fresh provider; return the snapshot it booted."""
     async with DaytonaRuntime("env", image=_build_image(context))(_row()):
         pass
-    create_call = calls["create"]
-    assert isinstance(create_call, tuple)
-    snapshot = create_call[0].snapshot
-    assert isinstance(snapshot, str)
-    return snapshot
+    return daytona.created[-1].snapshot
 
 
-async def test_daytona_rebuilds_the_snapshot_when_the_env_changes(
+async def test_daytona_rebuilds_the_snapshot_in_place_when_the_env_changes(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    # A name that doesn't track content resolves to whatever was built first.
-    calls = _install_fake_daytona(monkeypatch)
+    # The name is the durable handle; content drift rebuilds it, never renames it —
+    # otherwise every rollout after an edit silently measures the old code.
+    daytona = _install_fake_daytona(monkeypatch)
     (tmp_path / "env.py").write_text("REWARD = 1.0\n")
 
-    first = await _boot_snapshot(tmp_path, calls)
-    unchanged = await _boot_snapshot(tmp_path, calls)
+    first = await _boot_snapshot(tmp_path, daytona)
+    unchanged = await _boot_snapshot(tmp_path, daytona)
     (tmp_path / "env.py").write_text("REWARD = 0.0\n")
-    edited = await _boot_snapshot(tmp_path, calls)
+    edited = await _boot_snapshot(tmp_path, daytona)
 
-    assert unchanged == first
-    assert edited != first
-    assert first.startswith("env-")
-    # Built once per distinct content; the unchanged re-acquisition reused it.
-    assert calls["snapshot_create"] == [first, edited]
+    assert first == unchanged == edited == "env"
+    # Built once per distinct content; the unchanged re-acquisition reused it,
+    # and the stale build was replaced, not left behind.
+    assert daytona.snapshot.builds == ["env", "env"]
+    assert list(daytona.snapshot.snapshots) == ["env"]
 
 
-async def test_daytona_snapshot_name_ignores_build_noise(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+async def test_daytona_registry_ref_image_builds_and_reuses_by_name(
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    # Daytona's archiver has no .dockerignore, so the host venv rides along in
-    # the context; digesting it would rebuild on every local `uv sync`.
-    calls = _install_fake_daytona(monkeypatch)
-    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
-    before = await _boot_snapshot(tmp_path, calls)
+    # image= is documented as "Dockerfile/registry ref"; a plain ref string must
+    # build once, reuse while unchanged, and rebuild when repointed.
+    daytona = _install_fake_daytona(monkeypatch)
 
-    (tmp_path / ".venv" / "bin").mkdir(parents=True)
-    (tmp_path / ".venv" / "bin" / "python").write_text("#!/host/python\n")
-    (tmp_path / "__pycache__").mkdir()
-    (tmp_path / "__pycache__" / "env.cpython-311.pyc").write_bytes(b"\x00\x01")
+    for ref in ("registry/env:1", "registry/env:1", "registry/env:2"):
+        async with DaytonaRuntime("env", image=ref)(_row()):
+            pass
 
-    assert await _boot_snapshot(tmp_path, calls) == before
+    assert daytona.snapshot.builds == ["env", "env"]
+    assert daytona.snapshot.snapshots["env"].image_name == "registry/env:2"
 
 
 async def test_daytona_sends_cpu_as_a_whole_number(monkeypatch: pytest.MonkeyPatch) -> None:
     # Daytona's API rejects 2.0, and no equality assertion can catch it: 2.0 == 2.
-    calls = _install_fake_daytona(monkeypatch)
+    daytona = _install_fake_daytona(monkeypatch)
     config = RuntimeConfig(image="img:tag", resources=RuntimeResources(cpu=2))
 
     async with DaytonaRuntime(runtime_config=config)(_row()):
         pass
 
-    create_call = calls["create"]
-    assert isinstance(create_call, tuple)
-    assert isinstance(create_call[0].resources.cpu, int)
+    assert isinstance(daytona.created[-1].resources.cpu, int)
 
 
 async def test_daytona_rejects_a_fractional_cpu_request(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -840,8 +890,8 @@ async def test_daytona_names_a_sandbox_it_could_not_delete(
     monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None:
     # Teardown must report the leak without shadowing the run's own error.
-    calls = _install_fake_daytona(monkeypatch)
-    calls["delete_fails"] = True
+    daytona = _install_fake_daytona(monkeypatch)
+    daytona.delete_fails = True
 
     with caplog.at_level(logging.WARNING, logger="hud.eval.runtime"):
         async with DaytonaRuntime("snapshot")(_row()):
@@ -853,12 +903,12 @@ async def test_daytona_names_a_sandbox_it_could_not_delete(
 async def test_daytona_sizes_each_row_from_its_own_runtime_config(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    # Resolving the name once would boot every row on the first row's cores.
-    calls = _install_fake_daytona(monkeypatch)
+    # Resources are baked into a snapshot at build time, so rows that request
+    # different sizes must boot distinct snapshots, not the first row's.
+    daytona = _install_fake_daytona(monkeypatch)
     (tmp_path / "env.py").write_text("REWARD = 1.0\n")
     provider = DaytonaRuntime("env", image=_build_image(tmp_path))
 
-    booted = []
     for cpu in (2, 4):
         task = Task(
             env="any-env",
@@ -867,58 +917,9 @@ async def test_daytona_sizes_each_row_from_its_own_runtime_config(
         )
         async with provider(task):
             pass
-        create_call = calls["create"]
-        assert isinstance(create_call, tuple)
-        booted.append(create_call[0].snapshot)
-
-    assert booted[0] != booted[1]
-    assert calls["snapshot_create"] == booted
-
-
-async def test_daytona_prune_leaves_hand_named_snapshots_alone(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    # "env-baseline" shares the prefix but is nothing this runtime minted.
-    calls = _install_fake_daytona(monkeypatch)
-    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
-    stale = await _boot_snapshot(tmp_path, calls)
-    created = calls["snapshot_create"]
-    assert isinstance(created, list)
-    created.append("env-baseline")
-
-    (tmp_path / "env.py").write_text("REWARD = 0.0\n")
-    provider = DaytonaRuntime("env", image=_build_image(tmp_path))
-    async with provider(_row()):
-        pass
-
-    assert await provider.prune_snapshots() == [stale]
-    assert "env-baseline" in created
-
-
-async def test_daytona_prune_keeps_only_the_snapshot_in_use(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    calls = _install_fake_daytona(monkeypatch)
-    (tmp_path / "env.py").write_text("REWARD = 1.0\n")
-    stale = await _boot_snapshot(tmp_path, calls)
-    (tmp_path / "env.py").write_text("REWARD = 0.0\n")
-
-    provider = DaytonaRuntime("env", image=_build_image(tmp_path))
-    async with provider(_row()):
-        pass
-    current = await _boot_snapshot(tmp_path, calls)
 
-    assert await provider.prune_snapshots() == [stale]
-    assert calls["snapshot_create"] == [current]
-
-
-async def test_daytona_prune_refuses_before_a_snapshot_is_resolved(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    # Nothing to keep means every digest looks prunable, including a live one.
-    _install_fake_daytona(monkeypatch)
-    with pytest.raises(RuntimeError, match="resolved snapshot"):
-        await DaytonaRuntime("env").prune_snapshots()
+    assert daytona.snapshot.builds == ["env-2cpu", "env-4cpu"]
+    assert [params.snapshot for params in daytona.created] == ["env-2cpu", "env-4cpu"]
 
 
 async def test_daytona_attaches_the_env_output_to_a_failed_handshake(
diff --git a/hud/train/client.py b/hud/train/client.py
index 2c9b001fe..7413b32c6 100644
--- a/hud/train/client.py
+++ b/hud/train/client.py
@@ -45,7 +45,7 @@
         tuple["torch.Tensor", dict[str, float]],
     ]
 
-logger = logging.getLogger("hud.train")
+logger = logging.getLogger("hud.train.client")
 
 
 def _run_to_input(run: Run) -> TrainInput:
@@ -88,45 +88,47 @@ def _check_groups(
     trajectories: Sequence[str | Run | TrajectoryPayload],
     group_size: int | None,
 ) -> None:
-    """Fail before a forward pass if the batch can't form full GRPO groups: an
-    incomplete final group gets a skewed advantage baseline. Cheap, client-side,
-    no round-trip — unlike per-group reward spread, which only the service sees."""
-    if group_size is None:
-        return
+    """Validate the batch's GRPO group structure client-side, before the round
+    trip. Groups are keyed by ``group_id`` when every item is a ``Run`` carrying
+    one, positional slices of ``group_size`` otherwise. An incomplete group is an
+    error: its advantage baseline is skewed. A batch where no group has any
+    reward spread only warns: every advantage normalizes to zero, which is a
+    legitimate sampling outcome for one batch but makes a later ``optim_step``
+    fail confusingly when it is the whole accumulation. Rewards are only visible
+    inline, so the spread check skips batches carrying ``trace_id`` strings."""
     n = len(trajectories)
-    if n % group_size != 0:
+    if group_size is not None and n % group_size != 0:
         raise ValueError(
             f"{n} trajectories do not divide evenly into groups of {group_size}; "
             "GRPO normalizes advantages within each group, so every group must be full"
         )
-    run_groups = [
+    run_group_ids = [
         item.group_id for item in trajectories if not isinstance(item, str | TrajectoryPayload)
     ]
-    if len(run_groups) != n or any(group_id is None for group_id in run_groups):
-        return
-    groups = Counter(run_groups)
-    if incomplete := [group_id for group_id, count in groups.items() if count != group_size]:
-        raise ValueError(f"incomplete GRPO groups: {incomplete}")
-
-
-def _check_reward_spread(
-    trajectories: Sequence[str | Run | TrajectoryPayload],
-    group_size: int | None,
-) -> None:
-    """Warn when no group has any reward spread: GRPO normalizes advantages within a
-    group, so identical rewards zero every one of them and the pass accumulates
-    nothing — which surfaces much later as a 409 from ``optim_step`` blaming the
-    wrong call. Only inline rewards are visible here; ``trace_id``s are skipped."""
-    rewards = [t.reward for t in trajectories if not isinstance(t, str)]
-    if len(rewards) != len(trajectories) or len(rewards) < 2:
+    keyed = len(run_group_ids) == n and all(g is not None for g in run_group_ids)
+    if keyed and group_size is not None:
+        counts = Counter(run_group_ids)
+        if incomplete := [group_id for group_id, count in counts.items() if count != group_size]:
+            raise ValueError(f"incomplete GRPO groups: {incomplete}")
+
+    inline = [item for item in trajectories if not isinstance(item, str)]
+    if len(inline) != n or n < 2:
         return
-    size = group_size or len(rewards)
-    if all(len(set(rewards[start : start + size])) == 1 for start in range(0, len(rewards), size)):
+    if keyed:
+        buckets: dict[object, list[float]] = {}
+        for group_id, item in zip(run_group_ids, inline, strict=True):
+            buckets.setdefault(group_id, []).append(item.reward)
+        reward_groups = list(buckets.values())
+    else:
+        size = group_size or n
+        reward_groups = [
+            [item.reward for item in inline[start : start + size]] for start in range(0, n, size)
+        ]
+    if all(len(set(group)) == 1 for group in reward_groups):
         logger.warning(
-            "every reward is identical within its GRPO group (%s), so all advantages are "
-            "zero and this pass accumulates no gradient — the next optim_step will fail. "
-            "Vary task difficulty until rollouts disagree.",
-            rewards[0],
+            "no GRPO group has any reward spread, so every advantage normalizes to "
+            "zero and this pass accumulates no gradient. Vary task difficulty "
+            "until rollouts within a group disagree."
         )
 
 
@@ -162,7 +164,6 @@ async def forward_backward(
         """
         inputs = _to_inputs(trajectories)
         _check_groups(trajectories, group_size)
-        _check_reward_spread(trajectories, group_size)
         request = ForwardBackwardRequest(
             inputs=inputs,
             loss_fn=loss_fn,
diff --git a/hud/train/tests/test_client.py b/hud/train/tests/test_client.py
index 6124bc7f0..e2305c572 100644
--- a/hud/train/tests/test_client.py
+++ b/hud/train/tests/test_client.py
@@ -1,13 +1,15 @@
 from __future__ import annotations
 
-import contextlib
 import logging
+from typing import Any
 
 import pytest
 
 from hud.eval.run import Run
 from hud.train import TrainingClient
 
+_MODEL_ID = "00000000-0000-0000-0000-000000000001"
+
 
 async def test_training_rejects_timed_out_run_without_explicit_reward() -> None:
     run = Run.failed("rollout timed out")
@@ -27,33 +29,58 @@ async def test_training_rejects_incomplete_run_groups() -> None:
         await TrainingClient("test-model").forward_backward(runs, group_size=2)
 
 
+def _grpo_runs(groups: list[str], rewards: list[float]) -> list[Run]:
+    runs = [Run(None, "", {}) for _ in rewards]
+    for index, (run, group_id, reward) in enumerate(zip(runs, groups, rewards, strict=True)):
+        run.trace.trace_id = str(index)
+        run.group_id = group_id
+        run.grade.reward = reward
+    return runs
+
+
+async def _forward_backward(runs: list[Run], monkeypatch: pytest.MonkeyPatch) -> None:
+    """Run forward_backward against a stubbed training service (a UUID model id
+    resolves without a catalog lookup)."""
+
+    async def fake_request(method: str, url: str, **kwargs: Any) -> dict[str, Any]:
+        return {"metrics": {}, "num_datums": len(runs)}
+
+    monkeypatch.setattr("hud.train.base.make_request", fake_request)
+    await TrainingClient(_MODEL_ID).forward_backward(runs, group_size=2)
+
+
 async def test_training_warns_when_no_group_has_reward_spread(
-    caplog: pytest.LogCaptureFixture,
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    # Identical rewards zero every GRPO advantage, so nothing accumulates and the
-    # next optim_step 409s with a message that blames the wrong call.
-    runs = [Run(None, "", {}) for _ in range(4)]
-    for index, run in enumerate(runs):
-        run.trace.trace_id = str(index)
-        run.group_id = "a" if index < 2 else "b"
-        run.grade.reward = 1.0
+    # Identical rewards within every group zero every GRPO advantage, so the
+    # pass accumulates nothing and the later optim_step fails confusingly.
+    runs = _grpo_runs(["a", "a", "b", "b"], [1.0, 1.0, 0.0, 0.0])
 
-    with caplog.at_level(logging.WARNING, logger="hud.train"), contextlib.suppress(Exception):
-        await TrainingClient("test-model").forward_backward(runs, group_size=2)
+    with caplog.at_level(logging.WARNING, logger="hud.train.client"):
+        await _forward_backward(runs, monkeypatch)
+
+    assert "no gradient" in caplog.text
+
+
+async def test_training_groups_rewards_by_group_id_not_position(
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # An interleaved batch is valid (groups are keyed by group_id), and its
+    # positional slices have spread even when the real groups have none.
+    runs = _grpo_runs(["a", "b", "a", "b"], [1.0, 0.0, 1.0, 0.0])
+
+    with caplog.at_level(logging.WARNING, logger="hud.train.client"):
+        await _forward_backward(runs, monkeypatch)
 
     assert "no gradient" in caplog.text
 
 
 async def test_training_is_quiet_when_a_group_has_reward_spread(
-    caplog: pytest.LogCaptureFixture,
+    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    runs = [Run(None, "", {}) for _ in range(4)]
-    for index, run in enumerate(runs):
-        run.trace.trace_id = str(index)
-        run.group_id = "a" if index < 2 else "b"
-        run.grade.reward = float(index % 2)
+    runs = _grpo_runs(["a", "a", "b", "b"], [0.0, 1.0, 0.0, 1.0])
 
-    with caplog.at_level(logging.WARNING, logger="hud.train"), contextlib.suppress(Exception):
-        await TrainingClient("test-model").forward_backward(runs, group_size=2)
+    with caplog.at_level(logging.WARNING, logger="hud.train.client"):
+        await _forward_backward(runs, monkeypatch)
 
     assert "no gradient" not in caplog.text