diff --git a/docs/v6/reference/runtime.mdx b/docs/v6/reference/runtime.mdx
index e64f2a21c..cb378f708 100644
--- a/docs/v6/reference/runtime.mdx
+++ b/docs/v6/reference/runtime.mdx
@@ -152,10 +152,12 @@ 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 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).
 
+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`
 
 ```python
diff --git a/hud/agents/tests/test_provider_native_tools.py b/hud/agents/tests/test_provider_native_tools.py
index 3711b0832..014990f79 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,22 @@ 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:
+                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", ">"]:
             assert input is not None
             self._store[parts[2]] = input.encode()
@@ -520,3 +536,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_aliases": ["/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..a6c9ce9e7 100644
--- a/hud/agents/tools/ssh.py
+++ b/hud/agents/tools/ssh.py
@@ -7,13 +7,22 @@
 
 from __future__ import annotations
 
+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``."""
 
@@ -37,19 +46,26 @@ 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
-
 __all__ = ["SSHTool"]
diff --git a/hud/capabilities/base.py b/hud/capabilities/base.py
index 02ad1b199..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,6 +82,7 @@ def ssh(
         client_key_path: str | os.PathLike[str] | None = None,
         shell: str | None = None,
         cwd: str | None = None,
+        cwd_aliases: Sequence[str] | None = None,
     ) -> Capability:
         """``ssh/2`` — SSH daemon with publickey auth.
 
@@ -90,6 +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_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:
@@ -101,6 +108,8 @@ def ssh(
             params["client_key_path"] = os.fspath(client_key_path)
         if cwd is not None:
             params["cwd"] = cwd
+        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 d9a817cd6..a07fbed54 100644
--- a/hud/capabilities/ssh.py
+++ b/hud/capabilities/ssh.py
@@ -56,10 +56,17 @@ 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_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
+        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
@@ -70,8 +77,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, *aliases):
+                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/cli/trace.py b/hud/cli/trace.py
index 7b95486c9..077ea47d9 100644
--- a/hud/cli/trace.py
+++ b/hud/cli/trace.py
@@ -161,6 +161,8 @@ def _load_remote(trace_id: str) -> list[dict[str, Any]]:
 
 
 def _render_events(events: list[dict[str, Any]]) -> None:
+    # 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")
@@ -175,7 +177,7 @@ def _render_events(events: list[dict[str, Any]]) -> None:
 
             text = ev.get("text")
             if text:
-                console.print(text)
+                console.print(Text(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)})",
-                    highlight=False,
+                    Text.assemble(
+                        "  ", ("→", "green"), " ", (str(name), "bold"), f"({_fmt_args(args)})"
+                    )
                 )
 
             if ev.get("error"):
-                console.print(f"  [red]error: {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]✗ {name}: {error}[/red]")
+                console.print(Text(f"  ✗ {name}: {error}", style="red"))
             else:
-                console.print(f"  [dim]{name} →[/dim]")
+                console.print(Text(f"  {name} →", style="dim"))
                 for line in str(result).splitlines():
-                    console.print(f"    {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(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 66f6bc2eb..ac18a50d8 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_aliases"] == [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_aliases" not in ws.capability().params
+    finally:
+        await ws.stop()
diff --git a/hud/environment/workspace.py b/hud/environment/workspace.py
index 13ba752c0..48c4f5a8e 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_aliases = [given] if given != real and self._guest_path == real else []
         # 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_aliases=self._cwd_aliases or None,
         )
 
     @property
diff --git a/hud/eval/run.py b/hud/eval/run.py
index 396a84feb..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,13 +389,17 @@ async def _drive() -> None:
             driver.add_done_callback(_consume_task_result)
             raise
         except Exception as 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, 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
diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py
index bb674732a..08cbd64ed 100644
--- a/hud/eval/runtime.py
+++ b/hud/eval/runtime.py
@@ -39,7 +39,7 @@
 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
@@ -651,6 +651,35 @@ async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
                 await sb.terminate.aio()
 
 
+async def _snapshot_is_current(snapshot: Any, image: Any) -> bool:
+    """Whether *snapshot* was built from *image* as it exists right now.
+
+    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.
+    """
+    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:
     """The Daytona provider: each acquisition creates a fresh sandbox from a snapshot.
 
@@ -662,7 +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.
+    ``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.
@@ -695,10 +728,10 @@ 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
+        # Resolve each snapshot name against the image once; lock so concurrent
         # first acquisitions resolve exactly once.
         self._image = image
-        self._resolved = False
+        self._resolved: set[str] = set()
         self._snapshot_lock = asyncio.Lock()
 
     @asynccontextmanager
@@ -725,7 +758,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 +787,67 @@ 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:
+                    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 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 snapshot_name not in self._resolved:
+                            try:
+                                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(
+                                        name=snapshot_name,
+                                        image=self._image,
+                                        resources=daytona_resources,
                                     )
-                            self._resolved = True
+                                )
+                            self._resolved.add(snapshot_name)
                 sandbox_params = CreateSandboxFromSnapshotParams(
-                    snapshot=self.snapshot_name,
+                    snapshot=snapshot_name,
                     ephemeral=True,
                     auto_stop_interval=0,
                 )
@@ -793,7 +868,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 +876,39 @@ 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, 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:
-                # 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..b0dabf6c5 100644
--- a/hud/eval/tests/test_docker_provider.py
+++ b/hud/eval/tests/test_docker_provider.py
@@ -9,11 +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
 
@@ -202,6 +205,7 @@ class _CreateSandboxFromSnapshotParams:
 class _CreateSnapshotParams:
     name: str
     image: object
+    resources: object | None = None
 
 
 @dataclass(frozen=True)
@@ -227,7 +231,7 @@ class _DaytonaResources:
 
 @dataclass(frozen=True)
 class _DaytonaGpuType:
-    name: str
+    value: str
 
 
 @dataclass(frozen=True)
@@ -239,12 +243,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 +271,90 @@ 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, 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) -> 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 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 self.snapshots[name]
+
+    async def create(self, params: _CreateSnapshotParams) -> None:
+        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._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()
+        #: 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.delete_fails:
+            raise RuntimeError("daytona API unreachable")
         self.calls["delete"] = sandbox.id
 
 
@@ -306,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:
@@ -331,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(
@@ -620,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(
@@ -675,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),
@@ -706,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
@@ -722,6 +815,126 @@ 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, daytona: _FakeDaytonaClient) -> str:
+    """Acquire once through a fresh provider; return the snapshot it booted."""
+    async with DaytonaRuntime("env", image=_build_image(context))(_row()):
+        pass
+    return daytona.created[-1].snapshot
+
+
+async def test_daytona_rebuilds_the_snapshot_in_place_when_the_env_changes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # 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, daytona)
+    unchanged = await _boot_snapshot(tmp_path, daytona)
+    (tmp_path / "env.py").write_text("REWARD = 0.0\n")
+    edited = await _boot_snapshot(tmp_path, daytona)
+
+    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_registry_ref_image_builds_and_reuses_by_name(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # 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)
+
+    for ref in ("registry/env:1", "registry/env:1", "registry/env:2"):
+        async with DaytonaRuntime("env", image=ref)(_row()):
+            pass
+
+    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.
+    daytona = _install_fake_daytona(monkeypatch)
+    config = RuntimeConfig(image="img:tag", resources=RuntimeResources(cpu=2))
+
+    async with DaytonaRuntime(runtime_config=config)(_row()):
+        pass
+
+    assert isinstance(daytona.created[-1].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.
+    daytona = _install_fake_daytona(monkeypatch)
+    daytona.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:
+    # 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))
+
+    for cpu in (2, 4):
+        task = Task(
+            env="any-env",
+            id="t",
+            runtime_config=RuntimeConfig(resources=RuntimeResources(cpu=cpu)),
+        )
+        async with provider(task):
+            pass
+
+    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(
+    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..7413b32c6 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.client")
+
 
 def _run_to_input(run: Run) -> TrainInput:
     """Turn a graded :class:`hud.Run` into a training input: inline payload when
@@ -85,25 +88,48 @@ 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):
+    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
-    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}")
+    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(
+            "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."
+        )
 
 
 class TrainingClient(BaseTrainingClient):
diff --git a/hud/train/tests/test_client.py b/hud/train/tests/test_client.py
index 4f644baaa..e2305c572 100644
--- a/hud/train/tests/test_client.py
+++ b/hud/train/tests/test_client.py
@@ -1,10 +1,15 @@
 from __future__ import annotations
 
+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")
@@ -22,3 +27,60 @@ 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)
+
+
+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, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # 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.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, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    runs = _grpo_runs(["a", "a", "b", "b"], [0.0, 1.0, 0.0, 1.0])
+
+    with caplog.at_level(logging.WARNING, logger="hud.train.client"):
+        await _forward_backward(runs, monkeypatch)
+
+    assert "no gradient" not in caplog.text