From f1f6ebfb7fba30eddeacf40e4f0d2de83560b481 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 29 Oct 2026 11:30:51 -0700
Subject: [PATCH 1/7] fix(sandbox): quote signed URLs, agent env names and CLI
 paths in host scripts

A signed GET or PUT URL went into the VM host's script inside double quotes,
where bash still expands $(...) and backticks; a Cloud Storage URL carries the
bucket name of a stored artifact as written. The agent container's run script
pasted env var names, the container name and the image name bare, and
install_cli pasted the CLI's entrypoint path bare into its chmod. Each is now
shell-quoted; a value with no shell syntax renders exactly as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/agent_env/a2a_agent/a2a_agent.py          |  8 +++---
 .../providers/sandbox_providers/sandbox.py    |  4 +--
 .../sandbox_providers/local_sandbox_test.py   | 15 +++++++++++
 .../vm_file_transfer_test.py                  | 15 ++++++++++-
 .../sandbox_providers/vm_sandbox_test.py      | 22 +++++++++++----
 .../test_deploy_agent_enable_docker.py        |  2 +-
 tst/unit/task_step/test_load_artifact_file.py | 19 ++++++++++++-
 tst/util/exec_scripts.py                      | 27 ++++++++++++++++++-
 8 files changed, 97 insertions(+), 15 deletions(-)

diff --git a/src/agent_env/a2a_agent/a2a_agent.py b/src/agent_env/a2a_agent/a2a_agent.py
index 05947fdba..cc11a1198 100644
--- a/src/agent_env/a2a_agent/a2a_agent.py
+++ b/src/agent_env/a2a_agent/a2a_agent.py
@@ -305,7 +305,7 @@ async def install_cli(
 
         if isinstance(sandbox, VmSandbox):
             await sandbox.exec_script(
-                f"docker exec -u 0 {shlex.quote(container or sandbox.container_name)} chmod +x {entrypoint_path}")
+                f"docker exec -u 0 {shlex.quote(container or sandbox.container_name)} chmod +x {shlex.quote(entrypoint_path)}")
         else:
             await sandbox.exec("chmod", "+x", entrypoint_path)
 
@@ -661,7 +661,7 @@ async def _run_container(self, image_name: str, a2a_port: int, merged_env: dict[
         env_flags = []
         for key, value in agent_env.items():
             escaped_value = value.replace("'", "'\\''")
-            env_flags.append(f"-e {key}='{escaped_value}'")
+            env_flags.append(f"-e {shlex.quote(key)}='{escaped_value}'")
         env_str = " \\\n    ".join(env_flags)
         publish = " ".join(
             f"-p {spec}" for spec in port_bindings(self._sandbox.host_ips, self._sandbox.host_port(a2a_port), a2a_port)
@@ -670,11 +670,11 @@ async def _run_container(self, image_name: str, a2a_port: int, merged_env: dict[
         run_script = f"""#!/bin/bash
 set -e
 {setup_script}docker {"run -d" if trust_dir is None else "create"} \\
-    --name {self._sandbox.container_name} \\
+    --name {shlex.quote(self._sandbox.container_name)} \\
     --label {shlex.quote(f"{SANDBOX_LABEL}={self._sandbox.sandbox_id}")} \\
     {publish} \\
     {network_flag}{env_str} \\
-    {image_name} > /dev/null
+    {shlex.quote(image_name)} > /dev/null
 {"sleep 2" if trust_dir is None else ""}
 """
         await self._sandbox.exec_script(run_script)
diff --git a/src/agent_env/providers/sandbox_providers/sandbox.py b/src/agent_env/providers/sandbox_providers/sandbox.py
index 86c740656..3a37b3726 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/sandbox.py
@@ -476,7 +476,7 @@ async def _load_docker_images(self, artifacts: list, signed_urls: list[str | Non
                 # Download to a file first (retry-safe with -o); a `curl | ... docker load`
                 # pipe can't be retried without corrupting the stream (curl won't rewind).
                 load_commands.append(
-                    f'(curl -fsSL {CURL_RETRY_FLAGS} "{signed}" -o {shlex.quote(tmp_tar)} '
+                    f"(curl -fsSL {CURL_RETRY_FLAGS} {shlex.quote(signed)} -o {shlex.quote(tmp_tar)} "
                     f"&& gunzip -c {shlex.quote(tmp_tar)} | docker load && rm -f {shlex.quote(tmp_tar)})"
                 )
             else:
@@ -757,7 +757,7 @@ async def upload_vm_file(sandbox: VmSandbox, vm_path: str, store: ObjectStore, o
     machine."""
     put_url = await asyncio.to_thread(store.signed_put_url, object_url)
     if put_url is not None:
-        await sandbox.exec_script(f'curl -fsSL -X PUT --upload-file {shlex.quote(vm_path)} "{put_url}"')
+        await sandbox.exec_script(f"curl -fsSL -X PUT --upload-file {shlex.quote(vm_path)} {shlex.quote(put_url)}")
         return
     with tempfile.TemporaryDirectory(prefix="agentenv-vm-file-") as directory:
         local_path = os.path.join(directory, posixpath.basename(vm_path))
diff --git a/tst/unit/providers/sandbox_providers/local_sandbox_test.py b/tst/unit/providers/sandbox_providers/local_sandbox_test.py
index 9d5819240..4c257e446 100644
--- a/tst/unit/providers/sandbox_providers/local_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/local_sandbox_test.py
@@ -24,6 +24,7 @@
 from agent_env.a2a_agent import a2a_agent as a2a_agent_module
 from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
 from tst.unit.store.fakes import FakeObjectStore
+from tst.util.exec_scripts import run_with_stubs
 
 
 def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
@@ -1292,6 +1293,20 @@ async def test_an_agent_placed_on_a_local_vm_sandbox_maps_the_host_alias(tmp_pat
     assert "--add-host host.docker.internal:host-gateway" in sandbox.scripts[0]
 
 
+@pytest.mark.asyncio
+async def test_an_agent_env_var_whose_name_holds_shell_syntax_reaches_docker_as_it_is(tmp_path, monkeypatch):
+    monkeypatch.setattr(a2a_agent_module, "local_grant_trust", lambda: None)
+    sandbox = _RecordingLocalSandbox(work_dir=tmp_path / "work")
+    agent = A2AAgent.__new__(A2AAgent)
+    agent._sandbox = sandbox
+
+    await agent._run_container("img:v1", 8000, {"$(touch pwned)": "x';touch pwned;'"})
+
+    docker, = run_with_stubs(sandbox.scripts[0], tmp_path, "docker", "sleep")["docker"]
+    assert "$(touch pwned)=x';touch pwned;'" in docker and docker[-1] == "img:v1"
+    assert not (tmp_path / "pwned").exists()
+
+
 async def _no_sleep(_seconds):
     return None
 
diff --git a/tst/unit/providers/sandbox_providers/vm_file_transfer_test.py b/tst/unit/providers/sandbox_providers/vm_file_transfer_test.py
index 2ecb83cc3..bea4ec70d 100644
--- a/tst/unit/providers/sandbox_providers/vm_file_transfer_test.py
+++ b/tst/unit/providers/sandbox_providers/vm_file_transfer_test.py
@@ -11,6 +11,7 @@
 from agent_env.providers.sandbox_providers import sandbox as sandbox_module
 from agent_env.providers.sandbox_providers.sandbox import read_vm_file, upload_vm_file
 from agent_env.store.object_store.local import LocalFilesystemObjectStore
+from tst.util.exec_scripts import run_with_stubs
 
 
 class _ShellVm:
@@ -104,7 +105,19 @@ async def test_a_store_that_signs_uploads_is_uploaded_to_by_the_vm(tmp_path):
 
     await upload_vm_file(vm, "/tmp/image.tar.gz", _SigningStore(), "s3://bucket/builds/image.tar.gz")
 
-    assert vm.scripts == ['curl -fsSL -X PUT --upload-file /tmp/image.tar.gz "https://put.example/image.tar.gz?sig=1"']
+    assert vm.scripts == ["curl -fsSL -X PUT --upload-file /tmp/image.tar.gz 'https://put.example/image.tar.gz?sig=1'"]
+
+
+@pytest.mark.asyncio
+async def test_a_presigned_url_with_shell_syntax_reaches_curl_as_it_is(tmp_path):
+    vm = _RecordingVm()
+    name = "a$(touch pwned)`touch pwned`;touch pwned"
+
+    await upload_vm_file(vm, "/tmp/image.tar.gz", _SigningStore(), f"s3://bucket/builds/{name}")
+
+    curl, = run_with_stubs(vm.scripts[0], tmp_path, "curl")["curl"]
+    assert curl[-1] == f"https://put.example/{name}?sig=1"
+    assert not (tmp_path / "pwned").exists()
 
 
 @pytest.mark.asyncio
diff --git a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
index 2e77e3540..7d9fd10bb 100644
--- a/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/vm_sandbox_test.py
@@ -14,7 +14,7 @@
 from agent_env.store import ImageStore, RegistryAuth, set_object_store
 from agent_env.config import get_config, reset_config
 from tst.unit.store.fakes import ConfiguredObjectStore
-from tst.util.exec_scripts import script_run
+from tst.util.exec_scripts import run_with_stubs, script_run
 
 
 def _image(tar_gz_object_url: str | None, image_name: str = "img:1") -> DockerImageArtifact:
@@ -131,13 +131,25 @@ async def test_a_load_of_tarballs_runs_the_script_it_always_has(signing_store):
 
     flags = "--retry 5 --retry-all-errors --retry-delay 1"
     assert sandbox.scripts == [
-        f'(curl -fsSL {flags} "https://signed/a.tar.gz" -o /tmp/_docker_image_vm-test_0.tar.gz '
+        f"(curl -fsSL {flags} https://signed/a.tar.gz -o /tmp/_docker_image_vm-test_0.tar.gz "
         "&& gunzip -c /tmp/_docker_image_vm-test_0.tar.gz | docker load && rm -f /tmp/_docker_image_vm-test_0.tar.gz) & "
-        f'(curl -fsSL {flags} "https://signed/b.tar.gz" -o /tmp/_docker_image_vm-test_1.tar.gz '
+        f"(curl -fsSL {flags} https://signed/b.tar.gz -o /tmp/_docker_image_vm-test_1.tar.gz "
         "&& gunzip -c /tmp/_docker_image_vm-test_1.tar.gz | docker load && rm -f /tmp/_docker_image_vm-test_1.tar.gz) & wait"
     ]
 
 
+@pytest.mark.asyncio
+async def test_a_signed_url_with_shell_syntax_reaches_curl_as_it_is(signing_store, tmp_path):
+    name = "a$(touch pwned)`touch pwned`;touch pwned;&.tar.gz"
+    sandbox = _RecordingVmSandbox(images_stdout="a\n")
+
+    await sandbox.load_docker_images([_image(f"s3://bucket/{name}", "a:1")])
+
+    curl, = run_with_stubs(sandbox.scripts[0], tmp_path, "curl", "gunzip", "docker")["curl"]
+    assert curl[-3:] == [f"https://signed/{name}", "-o", "/tmp/_docker_image_vm-test_0.tar.gz"]
+    assert not (tmp_path / "pwned").exists()
+
+
 class _Registry(ImageStore):
     """Holds credentials for registry.example only."""
 
@@ -177,7 +189,7 @@ async def test_tarballs_are_loaded_and_the_rest_pulled(signing_store, registry):
     await sandbox.load_docker_images([_image(None, PUBLIC), _image("s3://bucket/a.tar.gz", "a:1")])
 
     load, pull = sandbox.scripts
-    assert '"https://signed/a.tar.gz"' in load and PUBLIC not in load
+    assert " https://signed/a.tar.gz " in load and PUBLIC not in load
     assert pull == f"docker pull {PUBLIC}"
 
 
@@ -972,7 +984,7 @@ async def test_tarballs_are_loaded_registry_images_pulled_and_contexts_built(sig
     await sandbox.load_docker_images([_image("s3://bucket/a.tar.gz", "a:1"), _image(None, PUBLIC),
                                       _context_image("c", "sha256:" + "3" * 64)])
 
-    assert '"https://signed/a.tar.gz"' in sandbox.scripts[0]
+    assert " https://signed/a.tar.gz " in sandbox.scripts[0]
     assert sandbox.scripts[1] == f"docker pull {PUBLIC}"
     assert not any("local/c-0123456789ab" in script for script in sandbox.scripts if script.startswith("docker pull"))
     assert sandbox.scripts[-2].endswith("-t local/c-0123456789ab:v1 .")
diff --git a/tst/unit/task_step/test_deploy_agent_enable_docker.py b/tst/unit/task_step/test_deploy_agent_enable_docker.py
index d20886583..4ad355917 100644
--- a/tst/unit/task_step/test_deploy_agent_enable_docker.py
+++ b/tst/unit/task_step/test_deploy_agent_enable_docker.py
@@ -37,7 +37,7 @@ async def test_enable_docker_never_exposes_host_socket():
     # Security contract: never bind-mount the VM socket; redirect the agent to the
     # isolated daemon. The sandbox (VM) is the external boundary we assert against.
     agent = _agent()
-    agent._sandbox = AsyncMock(ON_THIS_MACHINE=False)
+    agent._sandbox = AsyncMock(ON_THIS_MACHINE=False, container_name="agent-api")
     await agent._run_container("img:1", 8000, {"LITELLM_API_KEY": "k"}, enable_docker=True)
     launch = agent._sandbox.exec_script.await_args.args[0]
     assert "/var/run/docker.sock" not in launch
diff --git a/tst/unit/task_step/test_load_artifact_file.py b/tst/unit/task_step/test_load_artifact_file.py
index 036a70b78..cde1c353e 100644
--- a/tst/unit/task_step/test_load_artifact_file.py
+++ b/tst/unit/task_step/test_load_artifact_file.py
@@ -18,7 +18,7 @@
 from agent_env.providers.sandbox_providers.sandbox import CURL_RETRY_FLAGS, VmSandbox
 from agent_env.task_step.context import DeployedAgent, DeployedSandbox, TaskStepContext
 from agent_env.task_step.task_steps.load_artifact import LoadArtifactTaskStep
-from tst.util.exec_scripts import script_run
+from tst.util.exec_scripts import run_with_stubs, script_run
 
 CHECK = FileArtifact(
     id="check", version=2, description="checker", filename="check.py",
@@ -228,6 +228,23 @@ async def test_into_an_agent_installed_on_the_vm_host(monkeypatch, load, onto_th
         "mkdir -p /tmp/file_artifacts", onto_the_host]
 
 
+@pytest.mark.asyncio
+async def test_a_cli_whose_names_hold_shell_syntax_is_made_executable_as_it_is(monkeypatch, tmp_path):
+    entrypoint = "bin/x;touch pwned;"
+    tool = SimpleNamespace(id="tool", version=1, command_name="t$(touch pwned)", entrypoint=entrypoint,
+                           get_cli_files=lambda: SimpleNamespace(get_file_artifacts=lambda: {entrypoint: CHECK}))
+    sandbox = _Vm()
+    monkeypatch.setattr(sandbox_provider, "get_agent_sandbox_provider", _provider(sandbox))
+
+    path = await A2AAgent.install_cli(SimpleNamespace(sandbox_type=None, sandbox_id="sb-agent"), tool, "https://gateway",
+                                      "task")
+
+    chmod = next(script for script in sandbox.scripts if " chmod +x " in script)
+    assert run_with_stubs(chmod, tmp_path, "docker")["docker"] == [["exec", "-u", "0", "task", "chmod", "+x", path]]
+    assert path == "/opt/cli/t$(touch pwned)/bin/x;touch pwned;"
+    assert not (tmp_path / "pwned").exists()
+
+
 @pytest.mark.asyncio
 async def test_a_cli_reaches_an_agent_installed_into_a_container(monkeypatch):
     async def _install(installed_in: str | None) -> list[str]:
diff --git a/tst/util/exec_scripts.py b/tst/util/exec_scripts.py
index bb54f28bb..6855de45a 100644
--- a/tst/util/exec_scripts.py
+++ b/tst/util/exec_scripts.py
@@ -1,4 +1,11 @@
-"""What a recording fake sandbox saw a ``sudo bash -c`` exec run."""
+"""What a recording fake sandbox saw a ``sudo bash -c`` exec run, and what that script does when bash runs it."""
+
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+from pathlib import Path
 
 
 def script_run(args: tuple[str, ...]) -> str:
@@ -7,3 +14,21 @@ def script_run(args: tuple[str, ...]) -> str:
     for n, param in enumerate(args[5:], 1):
         script = script.replace(f'"${n}"', param)
     return script
+
+
+def run_with_stubs(script: str, directory: Path, *commands: str) -> dict[str, list[list[str]]]:
+    """Run ``script`` under bash in ``directory``, each of ``commands`` a stub on PATH that only records its arguments,
+    and return each command's calls."""
+    bin_dir = directory / "bin"
+    bin_dir.mkdir(exist_ok=True)
+    for name in commands:
+        log = shlex.quote(str(directory / f"{name}.log"))
+        (bin_dir / name).write_text(f'#!/bin/sh\nfor a in "$@"; do printf "%s\\n" "$a"; done >> {log}\necho --- >> {log}\n')
+        (bin_dir / name).chmod(0o755)
+    subprocess.run(["bash", "-c", script], cwd=directory, check=True,
+                   env={**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"})
+    return {name: _calls(directory / f"{name}.log") for name in commands}
+
+
+def _calls(log: Path) -> list[list[str]]:
+    return [call.strip("\n").split("\n") for call in log.read_text().split("---\n")[:-1]] if log.exists() else []

From df94e067e88fdd15c6d745b94b588e3109b481d8 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 29 Oct 2026 12:03:24 -0700
Subject: [PATCH 2/7] fix(task): render seeds in one pass and keep shell syntax
 out of shell fields

substitute_seed_in_steps replaced each <key> raw and key by key, so a seed
row's text became part of a step's command before the step's own quoting
could see a placeholder, and substituted text was re-scanned for other keys.
Placeholders are now rendered in one left-to-right pass. A step declares the
fields it runs as shell text (TaskStep.shell_fields); a seed value bound for
one must be text the shell reads literally, or the run is refused before any
step starts. Every other field takes any value, as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/agent_env/cli/task/run.py                 |  5 +-
 src/agent_env/task/task.py                    | 36 +++++++----
 src/agent_env/task_step/task_step.py          |  3 +
 .../task_steps/run_docker_container.py        |  1 +
 .../run_container_unit_tests_verifier.py      |  1 +
 .../task_steps/verifiers/verify_sandbox.py    |  1 +
 tst/unit/task/seed_substitution_test.py       | 61 +++++++++++++++++++
 7 files changed, 94 insertions(+), 14 deletions(-)
 create mode 100644 tst/unit/task/seed_substitution_test.py

diff --git a/src/agent_env/cli/task/run.py b/src/agent_env/cli/task/run.py
index 7473da0a2..2d190d29d 100644
--- a/src/agent_env/cli/task/run.py
+++ b/src/agent_env/cli/task/run.py
@@ -515,8 +515,9 @@ def run_batch(task_id: str, task_version: int | None, seeds: str, concurrency: i
     waiting.
 
     Each row in the CSV becomes a seed dict passed to the task via
-    context.metadata["seed"]. Prompt templates with <placeholder> variables
-    are rendered using the seed values before being sent to the agent.
+    context.metadata["seed"]. Each <placeholder> in a step is rendered with
+    the seed's value before the run starts. A value bound for a shell command
+    may hold only letters, digits and _.:/@%+=,-, or the run is refused.
     """
     if output_dir is None:
         output_dir = f"/tmp/taskid={fs_safe(task_id)}-batch{uuid4().hex[:6]}"
diff --git a/src/agent_env/task/task.py b/src/agent_env/task/task.py
index 605947ba7..c7ee46c6a 100644
--- a/src/agent_env/task/task.py
+++ b/src/agent_env/task/task.py
@@ -9,6 +9,8 @@
 import itertools
 import json
 import logging
+import re
+import shlex
 import time
 import uuid
 from dataclasses import dataclass
@@ -94,33 +96,44 @@ def _overlapping_survivors(
 
 def substitute_seed_in_steps(steps: list["TaskStep"], seed: dict) -> list["TaskStep"]:
     """Round-trip each step through its own to_dict/from_dict with ``<key>``
-    seed placeholders replaced in every string leaf. Returns a NEW list, never
-    mutating the input; idempotent (an already-substituted string has no tokens
-    left). Used by Task.run(); a retry re-dispatches the same substituted step
+    seed placeholders replaced in every string leaf, in one left-to-right pass
+    so substituted text is never re-scanned. A value going into one of a step's
+    ``shell_fields`` must be text the shell reads literally, or the run is
+    refused before any step runs. Returns a NEW list, never mutating the input.
+    Used by Task.run(); a retry re-dispatches the same substituted step
     objects, so it executes exactly what the original run executed."""
     if not seed:
         return list(steps)
+    placeholders = re.compile("|".join(re.escape(f"<{key}>") for key in seed))
 
-    def _sub(value):
+    def _sub(value, step, field):
         if isinstance(value, str):
-            for k, v in seed.items():
-                value = value.replace(f"<{k}>", str(v))
-            return value
+            return placeholders.sub(lambda match: _seed_value(seed, match.group(0)[1:-1], step, field), value)
         if isinstance(value, list):
-            return [_sub(x) for x in value]
+            return [_sub(x, step, field) for x in value]
         if isinstance(value, dict):
-            return {k: _sub(v) for k, v in value.items()}
+            return {k: _sub(v, step, field) for k, v in value.items()}
         return value
 
     from agent_env.task_step.task_step import attach_retry_config
 
     out: list["TaskStep"] = []
     for step in steps:
-        data = _sub(step.to_dict())
+        data = {field: _sub(value, step, field) for field, value in step.to_dict().items()}
         out.append(attach_retry_config(type(step).from_dict(data), data))
     return out
 
 
+def _seed_value(seed: dict, key: str, step: "TaskStep", field: str) -> str:
+    value = str(seed[key])
+    if field in step.shell_fields and shlex.quote(value) != value:
+        raise ValueError(
+            f"Seed {key!r} can't go into step {step.id!r} field {field!r}: a value substituted into shell text may hold "
+            f"only letters, digits and _.:/@%+=,- (got {value!r})"
+        )
+    return value
+
+
 def recorded_scores(steps: list["TaskStep"], context: TaskStepContext) -> dict[str, float]:
     """Every score a run of ``steps`` recorded, by the id of the step that recorded it, or by its verification's key
     when that isn't a step's verifier id."""
@@ -488,8 +501,7 @@ async def _run(
         # and concurrent runs must not race on a shared rewrite. And applied on
         # EVERY start_step, not only 0: a resume reloads the task with its
         # placeholders intact, so skipping substitution would hand later steps
-        # literal `<multienv_id>` strings. Substitution is idempotent — an
-        # already-replaced string has no tokens left.
+        # literal `<multienv_id>` strings.
         seed = context.metadata.get("seed") or {}
         steps = substitute_seed_in_steps(self.steps, seed) if seed else self.steps
         if start_step < 0 or start_step >= len(steps):
diff --git a/src/agent_env/task_step/task_step.py b/src/agent_env/task_step/task_step.py
index 44c56c723..fe709c848 100644
--- a/src/agent_env/task_step/task_step.py
+++ b/src/agent_env/task_step/task_step.py
@@ -94,6 +94,9 @@ class TaskStep(ABC):
     # Each field of a scope other than the task's, by its path in the stored step; any other field is task-scoped. A
     # subclass merges its parent's. Such a field is optional, so a step written out without it still builds and runs.
     field_scopes: ClassVar[Mapping[str, FieldScope]] = {"metadata.attribution": FieldScope.USER}
+    # Fields holding shell text, by their key in the stored step: a seed value substituted into one must be text the
+    # shell reads literally.
+    shell_fields: ClassVar[frozenset[str]] = frozenset()
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
diff --git a/src/agent_env/task_step/task_steps/run_docker_container.py b/src/agent_env/task_step/task_steps/run_docker_container.py
index e61f1ff93..c27be64b8 100644
--- a/src/agent_env/task_step/task_steps/run_docker_container.py
+++ b/src/agent_env/task_step/task_steps/run_docker_container.py
@@ -29,6 +29,7 @@ class RunDockerContainerTaskStep(TaskStep):
         ),
     )
     field_scopes = {**TaskStep.field_scopes, "docker_context_url": FieldScope.USER}
+    shell_fields = frozenset({"command_override", "ready_command"})
 
     def __init__(
         self,
diff --git a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
index 5e3580e9d..98ce55d8d 100644
--- a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
+++ b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
@@ -88,6 +88,7 @@
 class RunContainerUnitTestsVerifierTaskStep(TaskStep):
     type: ClassVar[str] = "run_container_unit_tests_verifier"
     entity_refs = (EntityRef.env("env_id"),)
+    shell_fields = frozenset({"command", "setup_commands"})
 
     def __init__(
         self,
diff --git a/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py b/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
index 9144fbc58..f6bbccf1e 100644
--- a/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
+++ b/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
@@ -59,6 +59,7 @@ def _outcome(passed: bool, justification: str) -> dict:
 class VerifySandboxTaskStep(TaskStep):
     type: ClassVar[str] = "verify_sandbox"
     entity_refs = ()
+    shell_fields = frozenset({"criteria"})
 
     def __init__(
         self,
diff --git a/tst/unit/task/seed_substitution_test.py b/tst/unit/task/seed_substitution_test.py
new file mode 100644
index 000000000..c900a7777
--- /dev/null
+++ b/tst/unit/task/seed_substitution_test.py
@@ -0,0 +1,61 @@
+"""A run's seed fills each step's ``<key>`` placeholders in one pass. A value bound for a field a step runs as shell
+text must be one the shell reads literally, so a seed row can name an id there but can't add a command."""
+
+from __future__ import annotations
+
+import pytest
+
+from agent_env.task.task import substitute_seed_in_steps
+from agent_env.task_step.task_steps.run_docker_container import RunDockerContainerTaskStep
+from agent_env.task_step.task_steps.verifiers.run_container_unit_tests_verifier import (
+    RunContainerUnitTestsVerifierTaskStep,
+)
+from agent_env.task_step.task_steps.verifiers.verify_sandbox import VerifySandboxTaskStep
+
+
+def _verifier(**fields):
+    return RunContainerUnitTestsVerifierTaskStep(id="check", version=None, sandbox_name="h", container_name="c",
+                                                 **{"command": "true", **fields})
+
+
+_SHELL_TEXT = [
+    (_verifier(command="scrape <url>"), "command"),
+    (_verifier(setup_commands=["pip install <url>"]), "setup_commands"),
+    (VerifySandboxTaskStep(id="check", version=None, base_dir="/app", verifier_id="v",
+                           criteria=[{"type": "bash_cmd_succeeds", "bash_cmd": "test -e <url>"}]), "criteria"),
+    (RunDockerContainerTaskStep(id="box", version=None, sandbox_name="h", docker_context_artifact_id="u",
+                                command_override="serve <url>"), "command_override"),
+    (RunDockerContainerTaskStep(id="box", version=None, sandbox_name="h", docker_context_artifact_id="u",
+                                ready_command="curl <url>"), "ready_command"),
+]
+
+
+@pytest.mark.parametrize("value", ["https://x/a; rm -rf /", "$(id)", "`id`", "a b", "x'y", 'x"y', "a&b", "*", ""])
+@pytest.mark.parametrize(("step", "field"), _SHELL_TEXT, ids=[field for _, field in _SHELL_TEXT])
+def test_a_seed_value_with_shell_syntax_is_refused_for_shell_text(step, field, value):
+    with pytest.raises(ValueError, match=rf"Seed 'url' can't go into step '{step.id}' field '{field}'"):
+        substitute_seed_in_steps([step], {"url": value})
+
+
+def test_an_id_is_substituted_into_shell_text_as_it_is():
+    step = _verifier(command='qc --task-id <task_id> --universe-id "<universe_id>" --turns <turns>')
+
+    seeded, = substitute_seed_in_steps([step], {"task_id": "task-12", "universe_id": "hg4/v3.1", "turns": 8})
+
+    assert seeded.command == 'qc --task-id task-12 --universe-id "hg4/v3.1" --turns 8'
+
+
+def test_any_value_fills_a_field_that_isnt_shell_text():
+    step = _verifier(env_vars={"URL": "<url>"})
+
+    seeded, = substitute_seed_in_steps([step], {"url": "https://x/a?b=1&c=$(id)"})
+
+    assert seeded.env_vars == {"URL": "https://x/a?b=1&c=$(id)"}
+
+
+def test_substituted_text_is_never_rescanned():
+    step = _verifier(env_vars={"A": "<a>", "B": "<b>"})
+
+    seeded, = substitute_seed_in_steps([step], {"a": "<b>", "b": "$(id)"})
+
+    assert seeded.env_vars == {"A": "<b>", "B": "$(id)"}

From 262cf88dfd9a6606821fed3467159d088a77389e Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 29 Oct 2026 13:03:49 -0700
Subject: [PATCH 3/7] docs(task): name shell_fields among the step extension
 rules; trim the refused seed in its error

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 AGENTS.md                  | 4 +++-
 src/agent_env/task/task.py | 2 +-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/AGENTS.md b/AGENTS.md
index fdb41c0a2..dfec6bfe0 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -137,7 +137,9 @@ Out-of-tree code plugs in either through `config.toml`, as a `module:Class` poin
 - `[envs]`, `[artifacts]`, `[task_steps]` with `impls = ["module:Class", ...]`: extra classes for the
   registries. The identity is the class's `type` (a `ClassVar` on envs and steps, the Pydantic `type`
   default on artifacts); a class that keeps the base default, collides with a registered type, or
-  leaves a method its base requires unimplemented fails at registry build with `ConfigError`.
+  leaves a method its base requires unimplemented fails at registry build with `ConfigError`. A step
+  that runs one of its fields as shell text names it in `shell_fields`, so a run's seed can't put shell
+  syntax there.
 - For sandbox and state providers the table name must equal the `type` the provider produces,
   because that string is persisted and used to reconnect.
 - Entry points register the same kinds from an installed package, with the entry-point name as the
diff --git a/src/agent_env/task/task.py b/src/agent_env/task/task.py
index c7ee46c6a..f59a053fd 100644
--- a/src/agent_env/task/task.py
+++ b/src/agent_env/task/task.py
@@ -129,7 +129,7 @@ def _seed_value(seed: dict, key: str, step: "TaskStep", field: str) -> str:
     if field in step.shell_fields and shlex.quote(value) != value:
         raise ValueError(
             f"Seed {key!r} can't go into step {step.id!r} field {field!r}: a value substituted into shell text may hold "
-            f"only letters, digits and _.:/@%+=,- (got {value!r})"
+            f"only letters, digits and _.:/@%+=,- (got {value[:80]!r}{'...' if len(value) > 80 else ''})"
         )
     return value
 

From 82880f0e8986c4dca5aa4f0869a55d94a7ac705d Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 29 Oct 2026 13:14:28 -0700
Subject: [PATCH 4/7] fix(task): name nested shell fields by path; quote the
 docker step's failure messages

verify_sandbox declared all of criteria as shell text, so a seed bound for a
probe_file_contains regex or a path was refused although it never reaches a
shell; shell_fields now names fields by their path in the stored step, and
verify_sandbox declares only criteria.bash_cmd. run_docker_container's
missing-Dockerfile and readiness messages put dockerfile_path and
container_name inside single quotes in a host script; each message is quoted
whole. The verifier's timeout_sec is quoted in its timeout call.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/agent_env/task/task.py                    | 16 +++++++-------
 src/agent_env/task_step/task_step.py          |  4 ++--
 .../task_steps/run_docker_container.py        |  5 +++--
 .../run_container_unit_tests_verifier.py      |  2 +-
 .../task_steps/verifiers/verify_sandbox.py    |  2 +-
 tst/unit/task/seed_substitution_test.py       | 14 +++++++++++-
 .../task_step/local_step_containers_test.py   | 22 +++++++++++++++++++
 7 files changed, 50 insertions(+), 15 deletions(-)

diff --git a/src/agent_env/task/task.py b/src/agent_env/task/task.py
index f59a053fd..c531bf7b2 100644
--- a/src/agent_env/task/task.py
+++ b/src/agent_env/task/task.py
@@ -106,29 +106,29 @@ def substitute_seed_in_steps(steps: list["TaskStep"], seed: dict) -> list["TaskS
         return list(steps)
     placeholders = re.compile("|".join(re.escape(f"<{key}>") for key in seed))
 
-    def _sub(value, step, field):
+    def _sub(value, step, path):
         if isinstance(value, str):
-            return placeholders.sub(lambda match: _seed_value(seed, match.group(0)[1:-1], step, field), value)
+            return placeholders.sub(lambda match: _seed_value(seed, match.group(0)[1:-1], step, path), value)
         if isinstance(value, list):
-            return [_sub(x, step, field) for x in value]
+            return [_sub(x, step, path) for x in value]
         if isinstance(value, dict):
-            return {k: _sub(v, step, field) for k, v in value.items()}
+            return {k: _sub(v, step, f"{path}.{k}" if path else k) for k, v in value.items()}
         return value
 
     from agent_env.task_step.task_step import attach_retry_config
 
     out: list["TaskStep"] = []
     for step in steps:
-        data = {field: _sub(value, step, field) for field, value in step.to_dict().items()}
+        data = _sub(step.to_dict(), step, "")
         out.append(attach_retry_config(type(step).from_dict(data), data))
     return out
 
 
-def _seed_value(seed: dict, key: str, step: "TaskStep", field: str) -> str:
+def _seed_value(seed: dict, key: str, step: "TaskStep", path: str) -> str:
     value = str(seed[key])
-    if field in step.shell_fields and shlex.quote(value) != value:
+    if path in step.shell_fields and shlex.quote(value) != value:
         raise ValueError(
-            f"Seed {key!r} can't go into step {step.id!r} field {field!r}: a value substituted into shell text may hold "
+            f"Seed {key!r} can't go into step {step.id!r} field {path!r}: a value substituted into shell text may hold "
             f"only letters, digits and _.:/@%+=,- (got {value[:80]!r}{'...' if len(value) > 80 else ''})"
         )
     return value
diff --git a/src/agent_env/task_step/task_step.py b/src/agent_env/task_step/task_step.py
index fe709c848..d6d6385ba 100644
--- a/src/agent_env/task_step/task_step.py
+++ b/src/agent_env/task_step/task_step.py
@@ -94,8 +94,8 @@ class TaskStep(ABC):
     # Each field of a scope other than the task's, by its path in the stored step; any other field is task-scoped. A
     # subclass merges its parent's. Such a field is optional, so a step written out without it still builds and runs.
     field_scopes: ClassVar[Mapping[str, FieldScope]] = {"metadata.attribution": FieldScope.USER}
-    # Fields holding shell text, by their key in the stored step: a seed value substituted into one must be text the
-    # shell reads literally.
+    # Fields holding shell text, by their path in the stored step (a list's items share its path): a seed value
+    # substituted into one must be text the shell reads literally.
     shell_fields: ClassVar[frozenset[str]] = frozenset()
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
diff --git a/src/agent_env/task_step/task_steps/run_docker_container.py b/src/agent_env/task_step/task_steps/run_docker_container.py
index c27be64b8..97cdfef43 100644
--- a/src/agent_env/task_step/task_steps/run_docker_container.py
+++ b/src/agent_env/task_step/task_steps/run_docker_container.py
@@ -197,7 +197,8 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         dockerfile_abs = posixpath.join(work_dir, self.dockerfile_path)
         await sandbox.exec_script(
             f"test -f {shlex.quote(dockerfile_abs)} "
-            f"|| (echo 'Dockerfile not found at {self.dockerfile_path} inside the build context' >&2; exit 1)"
+            f"|| (echo {shlex.quote(f'Dockerfile not found at {self.dockerfile_path} inside the build context')} >&2; "
+            f"exit 1)"
         )
 
         build_arg_flags = ""
@@ -287,7 +288,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 f"for i in $(seq 1 60); do "
                 f"if docker exec {shlex.quote(container)} sh -c {ready} >/dev/null 2>&1; then exit 0; fi; "
                 f"sleep 2; done; "
-                f"echo 'container {self.container_name} failed readiness check within 120s' >&2; exit 1"
+                f"echo {shlex.quote(f'container {self.container_name} failed readiness check within 120s')} >&2; exit 1"
             )
 
         context.metadata.setdefault("deployed_docker_containers", []).append({
diff --git a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
index 98ce55d8d..6885cdb85 100644
--- a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
+++ b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
@@ -433,7 +433,7 @@ async def _run_in_container(self, sandbox, merged_env: dict[str, str],
         )
         wrapped = (
             f"docker exec -u {user} {env_flags} {container} "
-            f"timeout --kill-after=10 {self.timeout_sec} bash -c {shlex.quote(command)}"
+            f"timeout --kill-after=10 {shlex.quote(str(self.timeout_sec))} bash -c {shlex.quote(command)}"
         )
         exit_code, stdout, stderr = await sandbox.exec_with_output("sudo", "bash", "-c", wrapped)
         return exit_code, stdout, stderr, await self._extracted(lambda path: self._extract_file(sandbox, path))
diff --git a/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py b/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
index f6bbccf1e..7a7dcd9f6 100644
--- a/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
+++ b/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
@@ -59,7 +59,7 @@ def _outcome(passed: bool, justification: str) -> dict:
 class VerifySandboxTaskStep(TaskStep):
     type: ClassVar[str] = "verify_sandbox"
     entity_refs = ()
-    shell_fields = frozenset({"criteria"})
+    shell_fields = frozenset({"criteria.bash_cmd"})
 
     def __init__(
         self,
diff --git a/tst/unit/task/seed_substitution_test.py b/tst/unit/task/seed_substitution_test.py
index c900a7777..b6c64cae7 100644
--- a/tst/unit/task/seed_substitution_test.py
+++ b/tst/unit/task/seed_substitution_test.py
@@ -22,7 +22,7 @@ def _verifier(**fields):
     (_verifier(command="scrape <url>"), "command"),
     (_verifier(setup_commands=["pip install <url>"]), "setup_commands"),
     (VerifySandboxTaskStep(id="check", version=None, base_dir="/app", verifier_id="v",
-                           criteria=[{"type": "bash_cmd_succeeds", "bash_cmd": "test -e <url>"}]), "criteria"),
+                           criteria=[{"type": "bash_cmd_succeeds", "bash_cmd": "test -e <url>"}]), "criteria.bash_cmd"),
     (RunDockerContainerTaskStep(id="box", version=None, sandbox_name="h", docker_context_artifact_id="u",
                                 command_override="serve <url>"), "command_override"),
     (RunDockerContainerTaskStep(id="box", version=None, sandbox_name="h", docker_context_artifact_id="u",
@@ -53,6 +53,18 @@ def test_any_value_fills_a_field_that_isnt_shell_text():
     assert seeded.env_vars == {"URL": "https://x/a?b=1&c=$(id)"}
 
 
+def test_a_criterion_that_isnt_a_shell_command_takes_any_value():
+    step = VerifySandboxTaskStep(id="check", version=None, base_dir="/app", verifier_id="v", criteria=[
+        {"type": "probe_file_contains", "paths": ["<report>"], "expected": "<pattern>"},
+        {"type": "bash_cmd_succeeds", "bash_cmd": "test -s <name>"}])
+
+    seeded, = substitute_seed_in_steps([step], {"report": "out dir/report.txt", "pattern": "[0-9]+ passed", "name": "ok"})
+
+    assert seeded.criteria == [
+        {"type": "probe_file_contains", "paths": ["out dir/report.txt"], "expected": "[0-9]+ passed"},
+        {"type": "bash_cmd_succeeds", "bash_cmd": "test -s ok"}]
+
+
 def test_substituted_text_is_never_rescanned():
     step = _verifier(env_vars={"A": "<a>", "B": "<b>"})
 
diff --git a/tst/unit/task_step/local_step_containers_test.py b/tst/unit/task_step/local_step_containers_test.py
index 565a08d33..9150c3c10 100644
--- a/tst/unit/task_step/local_step_containers_test.py
+++ b/tst/unit/task_step/local_step_containers_test.py
@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import asyncio
+import subprocess
 import types
 from unittest.mock import AsyncMock
 
@@ -24,6 +25,7 @@
 from agent_env.task_step.task_steps.verifiers.run_container_unit_tests_verifier import (
     RunContainerUnitTestsVerifierTaskStep,
 )
+from tst.util.exec_scripts import run_with_stubs
 
 SANDBOX_ID = "local-ab12cd34"
 HOST_PORT = 50123
@@ -115,6 +117,26 @@ async def test_a_tag_and_a_container_network_named_by_the_step_are_the_sandboxs_
     assert f"--network container:worker-{SANDBOX_ID} " in run
 
 
+@pytest.mark.asyncio
+async def test_a_dockerfile_path_and_container_name_holding_shell_syntax_run_nothing_when_reported(
+    tmp_path, monkeypatch,
+):
+    sandbox = _RecordingSandbox(tmp_path)
+    context = _on(monkeypatch, sandbox)
+    monkeypatch.setattr(RunDockerContainerTaskStep, "_stage_from_universe", AsyncMock())
+
+    await RunDockerContainerTaskStep(
+        id="box", version=None, sandbox_name="h", docker_context_artifact_id="u", docker_context_artifact_version=1,
+        dockerfile_path="missing'; touch pwned; #", container_name="c';touch pwned;'", ready_command="true",
+    ).execute(context)
+
+    reports = [s for s in sandbox.scripts if s.startswith("test -f ") or "failed readiness check" in s]
+    for script in reports:
+        with pytest.raises(subprocess.CalledProcessError):
+            run_with_stubs(script, tmp_path, "seq", "docker")
+    assert len(reports) == 2 and not (tmp_path / "pwned").exists()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("error, falls_back", [
     ("ERROR: failed to solve: no match for platform in manifest: not found", True),

From e8e16e3434c21ad025d9d47b883978c1baaad979 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 29 Oct 2026 13:56:58 -0700
Subject: [PATCH 5/7] refactor(task): drop the shell-field seed check; keep
 one-pass substitution

The per-step shell_fields declaration and the refusal of shell syntax in seeds
bound for them are removed: they added a concept every step author would have
to know. Seeds are still rendered in one left-to-right pass, so substituted
text is never re-scanned, and every quoting fix at the script sinks stays.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 AGENTS.md                                     |  4 +-
 src/agent_env/cli/task/run.py                 |  5 +-
 src/agent_env/task/task.py                    | 29 +++--------
 src/agent_env/task_step/task_step.py          |  3 --
 .../task_steps/run_docker_container.py        |  1 -
 .../run_container_unit_tests_verifier.py      |  1 -
 .../task_steps/verifiers/verify_sandbox.py    |  1 -
 tst/unit/task/seed_substitution_test.py       | 49 ++-----------------
 8 files changed, 14 insertions(+), 79 deletions(-)

diff --git a/AGENTS.md b/AGENTS.md
index dfec6bfe0..fdb41c0a2 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -137,9 +137,7 @@ Out-of-tree code plugs in either through `config.toml`, as a `module:Class` poin
 - `[envs]`, `[artifacts]`, `[task_steps]` with `impls = ["module:Class", ...]`: extra classes for the
   registries. The identity is the class's `type` (a `ClassVar` on envs and steps, the Pydantic `type`
   default on artifacts); a class that keeps the base default, collides with a registered type, or
-  leaves a method its base requires unimplemented fails at registry build with `ConfigError`. A step
-  that runs one of its fields as shell text names it in `shell_fields`, so a run's seed can't put shell
-  syntax there.
+  leaves a method its base requires unimplemented fails at registry build with `ConfigError`.
 - For sandbox and state providers the table name must equal the `type` the provider produces,
   because that string is persisted and used to reconnect.
 - Entry points register the same kinds from an installed package, with the entry-point name as the
diff --git a/src/agent_env/cli/task/run.py b/src/agent_env/cli/task/run.py
index 2d190d29d..7473da0a2 100644
--- a/src/agent_env/cli/task/run.py
+++ b/src/agent_env/cli/task/run.py
@@ -515,9 +515,8 @@ def run_batch(task_id: str, task_version: int | None, seeds: str, concurrency: i
     waiting.
 
     Each row in the CSV becomes a seed dict passed to the task via
-    context.metadata["seed"]. Each <placeholder> in a step is rendered with
-    the seed's value before the run starts. A value bound for a shell command
-    may hold only letters, digits and _.:/@%+=,-, or the run is refused.
+    context.metadata["seed"]. Prompt templates with <placeholder> variables
+    are rendered using the seed values before being sent to the agent.
     """
     if output_dir is None:
         output_dir = f"/tmp/taskid={fs_safe(task_id)}-batch{uuid4().hex[:6]}"
diff --git a/src/agent_env/task/task.py b/src/agent_env/task/task.py
index c531bf7b2..4c2beb166 100644
--- a/src/agent_env/task/task.py
+++ b/src/agent_env/task/task.py
@@ -10,7 +10,6 @@
 import json
 import logging
 import re
-import shlex
 import time
 import uuid
 from dataclasses import dataclass
@@ -97,43 +96,31 @@ def _overlapping_survivors(
 def substitute_seed_in_steps(steps: list["TaskStep"], seed: dict) -> list["TaskStep"]:
     """Round-trip each step through its own to_dict/from_dict with ``<key>``
     seed placeholders replaced in every string leaf, in one left-to-right pass
-    so substituted text is never re-scanned. A value going into one of a step's
-    ``shell_fields`` must be text the shell reads literally, or the run is
-    refused before any step runs. Returns a NEW list, never mutating the input.
-    Used by Task.run(); a retry re-dispatches the same substituted step
-    objects, so it executes exactly what the original run executed."""
+    so substituted text is never re-scanned. Returns a NEW list, never mutating
+    the input. Used by Task.run(); a retry re-dispatches the same substituted
+    step objects, so it executes exactly what the original run executed."""
     if not seed:
         return list(steps)
     placeholders = re.compile("|".join(re.escape(f"<{key}>") for key in seed))
 
-    def _sub(value, step, path):
+    def _sub(value):
         if isinstance(value, str):
-            return placeholders.sub(lambda match: _seed_value(seed, match.group(0)[1:-1], step, path), value)
+            return placeholders.sub(lambda match: str(seed[match.group(0)[1:-1]]), value)
         if isinstance(value, list):
-            return [_sub(x, step, path) for x in value]
+            return [_sub(x) for x in value]
         if isinstance(value, dict):
-            return {k: _sub(v, step, f"{path}.{k}" if path else k) for k, v in value.items()}
+            return {k: _sub(v) for k, v in value.items()}
         return value
 
     from agent_env.task_step.task_step import attach_retry_config
 
     out: list["TaskStep"] = []
     for step in steps:
-        data = _sub(step.to_dict(), step, "")
+        data = _sub(step.to_dict())
         out.append(attach_retry_config(type(step).from_dict(data), data))
     return out
 
 
-def _seed_value(seed: dict, key: str, step: "TaskStep", path: str) -> str:
-    value = str(seed[key])
-    if path in step.shell_fields and shlex.quote(value) != value:
-        raise ValueError(
-            f"Seed {key!r} can't go into step {step.id!r} field {path!r}: a value substituted into shell text may hold "
-            f"only letters, digits and _.:/@%+=,- (got {value[:80]!r}{'...' if len(value) > 80 else ''})"
-        )
-    return value
-
-
 def recorded_scores(steps: list["TaskStep"], context: TaskStepContext) -> dict[str, float]:
     """Every score a run of ``steps`` recorded, by the id of the step that recorded it, or by its verification's key
     when that isn't a step's verifier id."""
diff --git a/src/agent_env/task_step/task_step.py b/src/agent_env/task_step/task_step.py
index d6d6385ba..44c56c723 100644
--- a/src/agent_env/task_step/task_step.py
+++ b/src/agent_env/task_step/task_step.py
@@ -94,9 +94,6 @@ class TaskStep(ABC):
     # Each field of a scope other than the task's, by its path in the stored step; any other field is task-scoped. A
     # subclass merges its parent's. Such a field is optional, so a step written out without it still builds and runs.
     field_scopes: ClassVar[Mapping[str, FieldScope]] = {"metadata.attribution": FieldScope.USER}
-    # Fields holding shell text, by their path in the stored step (a list's items share its path): a seed value
-    # substituted into one must be text the shell reads literally.
-    shell_fields: ClassVar[frozenset[str]] = frozenset()
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
diff --git a/src/agent_env/task_step/task_steps/run_docker_container.py b/src/agent_env/task_step/task_steps/run_docker_container.py
index 97cdfef43..71a88780a 100644
--- a/src/agent_env/task_step/task_steps/run_docker_container.py
+++ b/src/agent_env/task_step/task_steps/run_docker_container.py
@@ -29,7 +29,6 @@ class RunDockerContainerTaskStep(TaskStep):
         ),
     )
     field_scopes = {**TaskStep.field_scopes, "docker_context_url": FieldScope.USER}
-    shell_fields = frozenset({"command_override", "ready_command"})
 
     def __init__(
         self,
diff --git a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
index 6885cdb85..b286d8769 100644
--- a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
+++ b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
@@ -88,7 +88,6 @@
 class RunContainerUnitTestsVerifierTaskStep(TaskStep):
     type: ClassVar[str] = "run_container_unit_tests_verifier"
     entity_refs = (EntityRef.env("env_id"),)
-    shell_fields = frozenset({"command", "setup_commands"})
 
     def __init__(
         self,
diff --git a/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py b/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
index 7a7dcd9f6..9144fbc58 100644
--- a/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
+++ b/src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py
@@ -59,7 +59,6 @@ def _outcome(passed: bool, justification: str) -> dict:
 class VerifySandboxTaskStep(TaskStep):
     type: ClassVar[str] = "verify_sandbox"
     entity_refs = ()
-    shell_fields = frozenset({"criteria.bash_cmd"})
 
     def __init__(
         self,
diff --git a/tst/unit/task/seed_substitution_test.py b/tst/unit/task/seed_substitution_test.py
index b6c64cae7..c4f523939 100644
--- a/tst/unit/task/seed_substitution_test.py
+++ b/tst/unit/task/seed_substitution_test.py
@@ -1,16 +1,12 @@
-"""A run's seed fills each step's ``<key>`` placeholders in one pass. A value bound for a field a step runs as shell
-text must be one the shell reads literally, so a seed row can name an id there but can't add a command."""
+"""A run's seed fills each step's ``<key>`` placeholders in one left-to-right pass, so text a value brings in is
+never read again as another placeholder."""
 
 from __future__ import annotations
 
-import pytest
-
 from agent_env.task.task import substitute_seed_in_steps
-from agent_env.task_step.task_steps.run_docker_container import RunDockerContainerTaskStep
 from agent_env.task_step.task_steps.verifiers.run_container_unit_tests_verifier import (
     RunContainerUnitTestsVerifierTaskStep,
 )
-from agent_env.task_step.task_steps.verifiers.verify_sandbox import VerifySandboxTaskStep
 
 
 def _verifier(**fields):
@@ -18,26 +14,7 @@ def _verifier(**fields):
                                                  **{"command": "true", **fields})
 
 
-_SHELL_TEXT = [
-    (_verifier(command="scrape <url>"), "command"),
-    (_verifier(setup_commands=["pip install <url>"]), "setup_commands"),
-    (VerifySandboxTaskStep(id="check", version=None, base_dir="/app", verifier_id="v",
-                           criteria=[{"type": "bash_cmd_succeeds", "bash_cmd": "test -e <url>"}]), "criteria.bash_cmd"),
-    (RunDockerContainerTaskStep(id="box", version=None, sandbox_name="h", docker_context_artifact_id="u",
-                                command_override="serve <url>"), "command_override"),
-    (RunDockerContainerTaskStep(id="box", version=None, sandbox_name="h", docker_context_artifact_id="u",
-                                ready_command="curl <url>"), "ready_command"),
-]
-
-
-@pytest.mark.parametrize("value", ["https://x/a; rm -rf /", "$(id)", "`id`", "a b", "x'y", 'x"y', "a&b", "*", ""])
-@pytest.mark.parametrize(("step", "field"), _SHELL_TEXT, ids=[field for _, field in _SHELL_TEXT])
-def test_a_seed_value_with_shell_syntax_is_refused_for_shell_text(step, field, value):
-    with pytest.raises(ValueError, match=rf"Seed 'url' can't go into step '{step.id}' field '{field}'"):
-        substitute_seed_in_steps([step], {"url": value})
-
-
-def test_an_id_is_substituted_into_shell_text_as_it_is():
+def test_each_placeholder_takes_its_value():
     step = _verifier(command='qc --task-id <task_id> --universe-id "<universe_id>" --turns <turns>')
 
     seeded, = substitute_seed_in_steps([step], {"task_id": "task-12", "universe_id": "hg4/v3.1", "turns": 8})
@@ -45,26 +22,6 @@ def test_an_id_is_substituted_into_shell_text_as_it_is():
     assert seeded.command == 'qc --task-id task-12 --universe-id "hg4/v3.1" --turns 8'
 
 
-def test_any_value_fills_a_field_that_isnt_shell_text():
-    step = _verifier(env_vars={"URL": "<url>"})
-
-    seeded, = substitute_seed_in_steps([step], {"url": "https://x/a?b=1&c=$(id)"})
-
-    assert seeded.env_vars == {"URL": "https://x/a?b=1&c=$(id)"}
-
-
-def test_a_criterion_that_isnt_a_shell_command_takes_any_value():
-    step = VerifySandboxTaskStep(id="check", version=None, base_dir="/app", verifier_id="v", criteria=[
-        {"type": "probe_file_contains", "paths": ["<report>"], "expected": "<pattern>"},
-        {"type": "bash_cmd_succeeds", "bash_cmd": "test -s <name>"}])
-
-    seeded, = substitute_seed_in_steps([step], {"report": "out dir/report.txt", "pattern": "[0-9]+ passed", "name": "ok"})
-
-    assert seeded.criteria == [
-        {"type": "probe_file_contains", "paths": ["out dir/report.txt"], "expected": "[0-9]+ passed"},
-        {"type": "bash_cmd_succeeds", "bash_cmd": "test -s ok"}]
-
-
 def test_substituted_text_is_never_rescanned():
     step = _verifier(env_vars={"A": "<a>", "B": "<b>"})
 

From feca0a1fa6adc083c66049eaf6df4acc3c77da35 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 29 Oct 2026 15:32:44 -0700
Subject: [PATCH 6/7] docs(verifier): say a stored command gets seeds as
 written, and data belongs in the env vars

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../verifiers/run_container_unit_tests_verifier.py  | 13 +++++--------
 1 file changed, 5 insertions(+), 8 deletions(-)

diff --git a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
index b286d8769..61c3c4838 100644
--- a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
+++ b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
@@ -180,14 +180,11 @@ def _resolve_command(self, context: TaskStepContext) -> tuple[str, dict[str, str
         new task version: ``step_params`` for a targeted or resumed run, ``seed`` for
         ``task run-batch --seeds`` fan-out. Both are additive to the stored ``env_vars``.
 
-        Seed values reach the container as env vars rather than by string substitution,
-        because ``-e K=<quoted>`` is a separate docker argument and bash expands ``"$K"``
-        without re-parsing metacharacters. ``<key>`` placeholders are still rendered for
-        rows that vary a flag rather than a value, in one left-to-right pass so
-        substituted text is never re-scanned: replacing key-by-key, a value of ``<b>``
-        quotes to ``'<b>'`` and a later key ``b`` rewrites the placeholder *inside* those
-        quotes, yielding ``''$(id)''`` — quotes that close each other and leave the
-        value bare for bash.
+        Seed values reach the container as env vars, ``-e K=<quoted>``, and bash expands
+        ``"$K"`` without re-parsing them, so a command that takes data reads it that way.
+        A ``<key>`` in the stored command was already filled with the value as written
+        when the run started (``substitute_seed_in_steps``); this pass shell-quotes the
+        placeholders of a ``step_params`` command only.
         """
         overrides = self.step_param_overrides(context)
         seed = context.metadata.get("seed") or {}

From b6fc1710552eedc89929058cd373b9cd29af1f47 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Thu, 29 Oct 2026 15:59:09 -0700
Subject: [PATCH 7/7] test(store): run the DynamoDB step-journal suite one AWS
 call at a time

moto checks a conditional write's condition and stores the item as two steps,
so two threads can both pass the same condition and one write is lost; DynamoDB
evaluates the condition atomically. The document store's tests already
serialized botocore calls for their racing cases; the step-journal suite's
concurrent-completions test did not, and lost a completion now and then in CI.
The serialization moves to the AWS tests' conftest as a fixture both use.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tst/unit/store/aws/conftest.py                   | 16 ++++++++++++++++
 .../store/aws/dynamodb_document_store_test.py    | 15 +--------------
 tst/unit/store/aws/dynamodb_step_journal_test.py |  2 +-
 3 files changed, 18 insertions(+), 15 deletions(-)

diff --git a/tst/unit/store/aws/conftest.py b/tst/unit/store/aws/conftest.py
index c3bd1ea79..dff5376ec 100644
--- a/tst/unit/store/aws/conftest.py
+++ b/tst/unit/store/aws/conftest.py
@@ -6,10 +6,12 @@
 """
 
 import os
+import threading
 
 import pytest
 
 pytest.importorskip("boto3")
+botocore_client = pytest.importorskip("botocore.client")
 
 _PLACEHOLDER_AWS_ENV = {
     "AWS_ACCESS_KEY_ID": "testing",
@@ -25,3 +27,17 @@ def _placeholder_aws_credentials(monkeypatch):
         for name, value in _PLACEHOLDER_AWS_ENV.items():
             if name not in os.environ:
                 monkeypatch.setenv(name, value)
+
+
+@pytest.fixture
+def one_aws_call_at_a_time(monkeypatch):
+    """moto checks a write's condition and applies the write as two steps any thread can come
+    between, where DynamoDB applies each call atomically; one call at a time restores that for
+    tests that race writers."""
+    serial, make_api_call = threading.Lock(), botocore_client.BaseClient._make_api_call
+
+    def call(client, operation, params):
+        with serial:
+            return make_api_call(client, operation, params)
+
+    monkeypatch.setattr(botocore_client.BaseClient, "_make_api_call", call)
diff --git a/tst/unit/store/aws/dynamodb_document_store_test.py b/tst/unit/store/aws/dynamodb_document_store_test.py
index e29f1be66..350e0716b 100644
--- a/tst/unit/store/aws/dynamodb_document_store_test.py
+++ b/tst/unit/store/aws/dynamodb_document_store_test.py
@@ -1,8 +1,5 @@
 """DynamoDbDocumentStore runs the full DocumentStore conformance suite against moto."""
 
-import threading
-
-import botocore.client
 import pytest
 from moto import mock_aws
 
@@ -12,17 +9,7 @@
 
 
 @pytest.fixture
-def store_coll(monkeypatch):
-    """moto checks a write's condition and applies the write as two steps any thread can come
-    between, where DynamoDB applies each call atomically; one call at a time restores that for
-    the kit's racing cases."""
-    serial, make_api_call = threading.Lock(), botocore.client.BaseClient._make_api_call
-
-    def one_call_at_a_time(client, operation, params):
-        with serial:
-            return make_api_call(client, operation, params)
-
-    monkeypatch.setattr(botocore.client.BaseClient, "_make_api_call", one_call_at_a_time)
+def store_coll(one_aws_call_at_a_time):
     with mock_aws():
         yield DynamoDbDocumentStore(table_prefix="test_", region="us-west-2"), "coll"
 
diff --git a/tst/unit/store/aws/dynamodb_step_journal_test.py b/tst/unit/store/aws/dynamodb_step_journal_test.py
index 7c5551de1..f84512238 100644
--- a/tst/unit/store/aws/dynamodb_step_journal_test.py
+++ b/tst/unit/store/aws/dynamodb_step_journal_test.py
@@ -14,7 +14,7 @@
 
 
 @pytest.fixture
-def store():
+def store(one_aws_call_at_a_time):
     with mock_aws():
         set_document_store(DynamoDbDocumentStore(table_prefix="test_", region="us-west-2"))
         s = TaskInstanceStore()