From b879432947b68b32cf3da601d57484cc9c550186 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 30 Sep 2026 09:58:13 -1000
Subject: [PATCH 1/7] feat(task): a depends_on entry may be a bare step id

TaskStepDependency.from_dict takes a step id or {"task_step_id": <id>}, and dependencies() reads a whole
depends_on: a lone string or any other shape is a ValueError rather than a list of characters. Steps are
still stored in the object form. _validate_dag now tells a step that comes later apart from an id that names
no step in the task.

The bundle resolver reads depends_on through the same parser, so a malformed one or an unknown id is
reported with its cause, not only as an output the step doesn't depend on.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 README.md                            |   2 +-
 src/agent_env/bundle/resolve.py      |  39 ++++++----
 src/agent_env/task/task.py           |  14 +++-
 src/agent_env/task_step/task_step.py |  31 ++++----
 tst/unit/bundle/materialize_test.py  |  11 +++
 tst/unit/bundle/resolve_test.py      |  23 ++++++
 tst/unit/task/depends_on_test.py     | 107 +++++++++++++++++++++++++++
 7 files changed, 197 insertions(+), 30 deletions(-)
 create mode 100644 tst/unit/task/depends_on_test.py

diff --git a/README.md b/README.md
index cebd52b7..02b62327 100644
--- a/README.md
+++ b/README.md
@@ -617,7 +617,7 @@ agent-env task create task.json --id <task-id> --project-id <label>
 
 `--project-id` is required by the CLI. Any string is accepted. The value is stored on the task and forwarded as a user/tag label on every model call; nothing else in a local run reads it.
 
-Ordering is declared per step. With `depends_on` omitted, a step runs after every earlier step (a sequential chain). With `depends_on: [{"task_step_id": "<id>"}]` the steps form a DAG and steps with no edge between them run concurrently; forward references and duplicate step ids are rejected at create time. `fail_task_on_error` defaults to `true`: a failure cancels in-flight steps and fails the run. Set it to `false` on a tolerant step and the failure is recorded in `context.metadata.failed_steps` (step id, type, error, `is_fatal: false`), dependents still run, and the run can complete with exit code 0.
+Ordering is declared per step. With `depends_on` omitted, a step runs after every earlier step (a sequential chain). With `depends_on: ["<id>"]`, each entry a step id or the object form `{"task_step_id": "<id>"}`, the steps form a DAG and steps with no edge between them run concurrently; forward references, ids that name no step and duplicate step ids are rejected at create time. A step is stored with the object form. `fail_task_on_error` defaults to `true`: a failure cancels in-flight steps and fails the run. Set it to `false` on a tolerant step and the failure is recorded in `context.metadata.failed_steps` (step id, type, error, `is_fatal: false`), dependents still run, and the run can complete with exit code 0.
 
 Every step reads and writes the same `TaskStepContext`: `deployed_envs`, `deployed_agents`, `deployed_sandboxes`, `prompt_responses`, and a free-form `metadata` map. The run writes this context to `--output-dir` as one JSON file per run; see [The run record](#the-run-record).
 
diff --git a/src/agent_env/bundle/resolve.py b/src/agent_env/bundle/resolve.py
index 4dc3018f..baff8dd3 100644
--- a/src/agent_env/bundle/resolve.py
+++ b/src/agent_env/bundle/resolve.py
@@ -23,7 +23,7 @@
 from agent_env.plugins import _registration
 from agent_env.store.ids import LOCAL_PREFIX, validate_local_id
 from agent_env.task_step.registry import get_task_step_registry
-from agent_env.task_step.task_step import TaskStep, attach_retry_config
+from agent_env.task_step.task_step import TaskStep, attach_retry_config, dependencies
 
 from ._fs import fold, relative, show
 from .parse import NAMED_BY, Bundle, BundleEntry, BundleError, BundleKind
@@ -173,7 +173,7 @@ def _task(self, entry: BundleEntry) -> ResolvedEntry:
             elif self._derived_outputs(entry, step, cls):
                 declared.append((index, step, cls))
         outputs = self._outputs(entry, declared)
-        upstream = _upstream(steps)
+        upstream = _upstream(steps, self._depends_on(entry, steps))
         references: list[Reference] = []
         for index, step, cls in declared:
             where = f"step {step.get('id')!r}: "
@@ -283,6 +283,25 @@ def _outputs(self, entry: BundleEntry, declared: list[tuple[int, dict, type]]) -
                 outputs[kind, name] = Output(kind, output_id, index, step.get("id"), site.ref.artifact_type)
         return outputs
 
+    def _depends_on(self, entry: BundleEntry, steps: list[dict]) -> list[list[str] | None]:
+        """Each step's ``depends_on`` as the step ids it names, read as building the step reads it; None when it
+        has none. One that can't be read, or names no step in the task, is a problem here: a step reading an
+        earlier step's output is checked against these edges before any step is built."""
+        step_ids = {step.get("id") for step in steps}
+        depends = []
+        for step in steps:
+            try:
+                read = dependencies(step.get("depends_on"))
+            except ValueError as e:
+                self._problem(entry, f"step {step.get('id')!r}: {e}")
+                read = []
+            ids = None if read is None else [dep.task_step_id for dep in read]
+            for dep_id in ids or ():
+                if dep_id not in step_ids:
+                    self._problem(entry, f"step {step.get('id')!r}: depends_on {dep_id!r} names no step in this task")
+            depends.append(ids)
+        return depends
+
     def _resolve(self, entry: BundleEntry, site: RefSite, where: str, name: str, version: int | None,
                  outputs: dict[tuple[EntityKind, str], Output], upstream: set[int],
                  references: list[Reference]) -> None:
@@ -358,21 +377,15 @@ def _nfc(name: str) -> str:
     return unicodedata.normalize("NFC", name)
 
 
-def _upstream(steps: list[dict]) -> list[set[int]]:
-    """The indexes of the steps each step runs after: its ``depends_on`` or, when it has none, every
-    step before it, followed transitively; the rule the task scheduler runs steps by."""
+def _upstream(steps: list[dict], depends: list[list[str] | None]) -> list[set[int]]:
+    """The indexes of the steps each step runs after: the step ids ``depends`` holds for it or, when it has
+    none, every step before it, followed transitively; the rule the task scheduler runs steps by."""
     index_of: dict[str, int] = {}
     for index, step in enumerate(steps):
         if isinstance(step.get("id"), str):
             index_of.setdefault(step["id"], index)
-    direct = []
-    for index, step in enumerate(steps):
-        raw = step.get("depends_on")
-        if raw is None:
-            direct.append(set(range(index)))
-            continue
-        ids = [dep.get("task_step_id") for dep in raw if isinstance(dep, dict)] if isinstance(raw, list) else []
-        direct.append({index_of[dep_id] for dep_id in ids if dep_id in index_of})
+    direct = [set(range(index)) if ids is None else {index_of[dep_id] for dep_id in ids if dep_id in index_of}
+              for index, ids in enumerate(depends)]
     upstream = []
     for index in range(len(steps)):
         seen: set[int] = set()
diff --git a/src/agent_env/task/task.py b/src/agent_env/task/task.py
index 699645ff..d6cb9301 100644
--- a/src/agent_env/task/task.py
+++ b/src/agent_env/task/task.py
@@ -315,6 +315,7 @@ def __init__(
 
     @classmethod
     def _validate_dag(cls, steps: list[TaskStep]) -> None:
+        ids = {s.id for s in steps}
         seen: set[str] = set()
         for i, step in enumerate(steps):
             if step.id in seen:
@@ -342,12 +343,19 @@ def _validate_dag(cls, steps: list[TaskStep]) -> None:
             if step.depends_on is None:
                 continue
             for dep in step.depends_on:
-                if dep.task_step_id not in prior:
+                if dep.task_step_id in prior:
+                    continue
+                if dep.task_step_id in ids:
                     raise ValueError(
                         f"Step '{step.id}' at position {i} declares depends_on "
-                        f"'{dep.task_step_id}' which does not appear in a prior step "
-                        f"(forward refs are not allowed; known prior ids: {sorted(prior)})"
+                        f"'{dep.task_step_id}', which doesn't come before it "
+                        "(forward refs are not allowed: a step depends only on the steps before it)"
                     )
+                raise ValueError(
+                    f"Step '{step.id}' at position {i} declares depends_on "
+                    f"'{dep.task_step_id}', which names no step in this task "
+                    f"(steps before it: {sorted(prior)})"
+                )
 
     def preflight(self) -> list[str]:
         """Every step's resolvable-config problems, in step order; empty when fine.
diff --git a/src/agent_env/task_step/task_step.py b/src/agent_env/task_step/task_step.py
index 6ea8250e..f591e25b 100644
--- a/src/agent_env/task_step/task_step.py
+++ b/src/agent_env/task_step/task_step.py
@@ -19,8 +19,22 @@ class TaskStepDependency:
     task_step_id: str
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> "TaskStepDependency":
-        return cls(task_step_id=data["task_step_id"])
+    def from_dict(cls, data: str | dict[str, Any]) -> "TaskStepDependency":
+        """One ``depends_on`` entry: a step id, or ``{"task_step_id": <step id>}``."""
+        step_id = data.get("task_step_id") if isinstance(data, dict) else data
+        if not isinstance(step_id, str):
+            raise ValueError(f'a depends_on entry is a step id or {{"task_step_id": "<step id>"}}, not {data!r}')
+        return cls(task_step_id=step_id)
+
+
+def dependencies(raw: Any) -> list[TaskStepDependency] | None:
+    """``depends_on`` as written, read into dependencies. None, which runs a step after every earlier one,
+    stays None; a lone step id is refused rather than read as a list of its characters."""
+    if raw is None:
+        return None
+    if not isinstance(raw, (list, tuple)):
+        raise ValueError(f"depends_on is a list of step ids, not {raw!r}")
+    return [dep if isinstance(dep, TaskStepDependency) else TaskStepDependency.from_dict(dep) for dep in raw]
 
 
 def attach_retry_config(step: "TaskStep", data: dict[str, Any]) -> "TaskStep":
@@ -91,12 +105,7 @@ def __init__(
     ):
         self.id = id
         self.version = version
-        if depends_on is not None:
-            depends_on = [
-                d if isinstance(d, TaskStepDependency) else TaskStepDependency.from_dict(d)
-                for d in depends_on
-            ]
-        self.depends_on = depends_on
+        self.depends_on = dependencies(depends_on)
         self.fail_task_on_error = fail_task_on_error
         if retry_config is not None and not isinstance(retry_config, RetryConfig):
             retry_config = RetryConfig.from_dict(retry_config)
@@ -132,13 +141,9 @@ def to_dict(self) -> dict[str, Any]:
 
     @classmethod
     def _base_from_dict(cls, data: dict) -> dict:
-        raw = data.get("depends_on")
-        depends_on = (
-            [TaskStepDependency.from_dict(d) for d in raw] if raw is not None else None
-        )
         return {
             "id": data["id"], "version": data.get("version"),
-            "depends_on": depends_on,
+            "depends_on": dependencies(data.get("depends_on")),
             "fail_task_on_error": data.get("fail_task_on_error", True),
         }
 
diff --git a/tst/unit/bundle/materialize_test.py b/tst/unit/bundle/materialize_test.py
index 04e634d7..d3c971c3 100644
--- a/tst/unit/bundle/materialize_test.py
+++ b/tst/unit/bundle/materialize_test.py
@@ -399,6 +399,17 @@ def test_a_step_reading_its_own_tasks_output_is_not_preflighted(bundle_dir, monk
     assert task.steps[1].artifact_id == f"{ROOT}/t/made"
 
 
+def test_a_bare_step_id_in_depends_on_is_written_in_the_object_form_and_reused(bundle_dir):
+    box, greet, docs = json.loads(LAYOUT["tasks/t.json"])
+    _steps(bundle_dir, [box, {**greet, "depends_on": ["box"]}, {**docs, "depends_on": ["greet", {"task_step_id": "box"}]}])
+    _run(bundle_dir)
+
+    stored = local_store().find_one("tasks", Filter.of(id=f"{ROOT}/t"))
+    assert [step.get("depends_on") for step in stored["steps"]] == [
+        None, [{"task_step_id": "box"}], [{"task_step_id": "greet"}, {"task_step_id": "box"}]]
+    assert {done.reused for done in _run(bundle_dir).writes} == {True}
+
+
 def test_a_step_writing_an_output_is_still_preflighted(bundle_dir, monkeypatch):
     _steps(bundle_dir, [{"id": "write", "type": "writes_materialize_test", "artifact_id": "made"}])
     monkeypatch.setattr(_Checked, "problems", ["the writer's script isn't there"])
diff --git a/tst/unit/bundle/resolve_test.py b/tst/unit/bundle/resolve_test.py
index 1c2378c2..6385498d 100644
--- a/tst/unit/bundle/resolve_test.py
+++ b/tst/unit/bundle/resolve_test.py
@@ -230,6 +230,9 @@ def test_an_output_must_come_from_one_earlier_step_of_the_right_type(make, steps
 @pytest.mark.parametrize("depends_on", [
     [{"task_step_id": "cli"}],
     [{"task_step_id": "tag"}],
+    ["cli"],
+    ["tag"],
+    [{"task_step_id": "tag"}, "cli"],
 ])
 def test_an_output_resolves_for_a_step_that_depends_on_its_writer(make, depends_on):
     entry = resolved(make(tasks={"t": [
@@ -241,6 +244,24 @@ def test_an_output_resolves_for_a_step_that_depends_on_its_writer(make, depends_
     assert entry.config[2]["artifact_id"] == f"{ROOT}/t/tickets-cli"
 
 
+@pytest.mark.parametrize(("depends_on", "problem"), [
+    ("cli", "depends_on is a list of step ids, not 'cli'"),
+    ([7], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not 7'),
+    ([{"id": "cli"}], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not {\'id\': \'cli\'}'),
+    (["clii"], "depends_on 'clii' names no step in this task"),
+])
+def test_a_depends_on_that_cant_be_read_is_named_beside_the_output_it_then_cant_reach(make, depends_on, problem):
+    assert set(problems(make(tasks={"t": [
+        {"id": "cli", "type": "build_mcp_cli", "env_id": "tickets", "command_name": "c", "cli_artifact_id": "tickets-cli"},
+        {"id": "install", "type": "load_artifact", "agent_name": "solver", "artifact_id": "tickets-cli",
+         "depends_on": depends_on},
+    ]}))) == {
+        "tasks/t.json: step 'install': artifact_id: 'tickets-cli' is written by step 'cli', which this step doesn't "
+        "depend on",
+        f"tasks/t.json: step 'install': {problem}",
+    }
+
+
 def test_a_name_differing_only_in_unicode_form_still_resolves(make):
     nfd = "cafe\u0301"
     entry = resolved(make(files={"envs/caf\u00e9/Dockerfile": "FROM scratch\n"}, tasks={"t": [
@@ -257,6 +278,8 @@ def test_a_name_differing_only_in_unicode_form_still_resolves(make):
 def test_an_undeclared_step_passes_through_unless_it_repeats_a_bundle_name(make):
     clean = {"id": "tickets", "type": "plugin_test", "note": "hello", "after_step_id": "greeting"}
     assert resolved(make(tasks={"t": [clean], "hello": []}, evals={"regression": 'tasks = ["t"]'}), "t").config == [clean]
+    after = {"id": "after", "type": "plugin_test", "depends_on": ["tickets"]}
+    assert resolved(make(tasks={"t": [clean, after]}), "t").config == [clean, after]
     assert resolved(make(tasks={"u": [{**clean, "note": "regression"}]}), "u").config == [{**clean, "note": "regression"}]
     assert problems(make(tasks={"u": [{"id": "p", "type": "plugin_test", "target": {"name": "Tickets"}}]})) == (
         "tasks/u.json: step 'p': target.name = 'Tickets' names this bundle's env 'tickets', but plugin_test "
diff --git a/tst/unit/task/depends_on_test.py b/tst/unit/task/depends_on_test.py
new file mode 100644
index 00000000..bad4cff1
--- /dev/null
+++ b/tst/unit/task/depends_on_test.py
@@ -0,0 +1,107 @@
+"""A step's ``depends_on``: each entry a step id or ``{"task_step_id": <step id>}``, what else is refused, the
+form a step is stored in, and the edges a task checks and runs its steps by."""
+
+import asyncio
+import re
+
+import pytest
+
+from agent_env.config import get_config
+from agent_env.config.runtime import Config
+from agent_env.store import Filter
+from agent_env.task import Task
+from agent_env.task.store import get_task_store
+from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.task_step import TaskStep, TaskStepDependency, dependencies
+
+
+class _Step(TaskStep):
+    """Logs when it starts and ends."""
+
+    type = "depends_on_test"
+    entity_refs = ()
+    log: list[str] = []
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(**cls._base_from_dict(data))
+
+    async def execute(self, context):
+        _Step.log.append(f"start {self.id}")
+        await asyncio.sleep(0.02)
+        _Step.log.append(f"end {self.id}")
+        return context
+
+
+@pytest.fixture(autouse=True)
+def registries(monkeypatch):
+    steps = {**Config().task_step_registry(), _Step.type: _Step}
+    monkeypatch.setattr(Config, "task_step_registry", lambda self: steps)
+    monkeypatch.setattr(_Step, "log", [])
+
+
+def _step(id, depends_on=None):
+    return _Step(id=id, version=None, depends_on=depends_on)
+
+
+@pytest.mark.parametrize("entry", ["a", {"task_step_id": "a"}, TaskStepDependency("a")])
+def test_an_entry_is_a_step_id_or_its_object_form(entry):
+    assert dependencies([entry]) == [TaskStepDependency("a")]
+    assert _step("b", [entry]).depends_on == _Step.from_dict({"id": "b", "depends_on": [entry]}).depends_on == [
+        TaskStepDependency("a")]
+
+
+@pytest.mark.parametrize(("depends_on", "problem"), [
+    ("a", "depends_on is a list of step ids, not 'a'"),
+    ({"task_step_id": "a"}, "depends_on is a list of step ids, not {'task_step_id': 'a'}"),
+    ([7], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not 7'),
+    ([{"id": "a"}], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not {\'id\': \'a\'}'),
+    ([{"task_step_id": 7}], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not '
+     "{'task_step_id': 7}"),
+])
+def test_anything_else_is_refused_saying_what_it_takes(depends_on, problem):
+    with pytest.raises(ValueError, match=re.escape(problem)):
+        _step("b", depends_on)
+    with pytest.raises(ValueError, match=re.escape(problem)):
+        _Step.from_dict({"id": "b", "depends_on": depends_on})
+
+
+def test_a_step_id_is_stored_in_the_object_form_and_reads_back(local_stores):
+    steps = [_step("a"), _Step.from_dict({"id": "b", "depends_on": ["a"]}), _step("c", ["a", {"task_step_id": "b"}])]
+    get_task_store().put_document(Task(id="t", version=None, steps=steps))
+
+    stored = get_config().get_document_store().find_one("tasks", Filter.of(id="t"))
+    assert [step.get("depends_on") for step in stored["steps"]] == [
+        None, [{"task_step_id": "a"}], [{"task_step_id": "a"}, {"task_step_id": "b"}]]
+    assert Task.get("t").steps[2].depends_on == [TaskStepDependency("a"), TaskStepDependency("b")]
+
+
+@pytest.mark.parametrize(("steps", "problem"), [
+    ([_step("ask", ["deploy"]), _step("deploy")],
+     "Step 'ask' at position 0 declares depends_on 'deploy', which doesn't come before it (forward refs are not "
+     "allowed: a step depends only on the steps before it)"),
+    ([_step("deploy"), _step("ask", ["ask"])],
+     "Step 'ask' at position 1 declares depends_on 'ask', which doesn't come before it (forward refs are not "
+     "allowed: a step depends only on the steps before it)"),
+    ([_step("deploy"), _step("ask", ["deplyo"])],
+     "Step 'ask' at position 1 declares depends_on 'deplyo', which names no step in this task (steps before it: "
+     "['deploy'])"),
+], ids=["later", "itself", "unknown"])
+def test_a_step_depends_only_on_a_step_before_it_and_a_typo_is_told_apart_from_a_forward_ref(steps, problem):
+    with pytest.raises(ValueError) as caught:
+        Task(id="t", version=None, steps=steps)
+    assert str(caught.value) == problem
+
+
+@pytest.mark.asyncio
+async def test_bare_and_object_entries_gate_the_scheduler_alike(local_stores):
+    task = Task(id="t", version=None, steps=[
+        _step("a"), _step("b", ["a"]), _step("c", [{"task_step_id": "a"}]), _step("d", ["b", {"task_step_id": "c"}]),
+    ])
+
+    await task.run(context=TaskStepContext())
+
+    assert _Step.log[:2] == ["start a", "end a"]
+    assert sorted(_Step.log[2:4]) == ["start b", "start c"]
+    assert sorted(_Step.log[4:6]) == ["end b", "end c"]
+    assert _Step.log[6:] == ["start d", "end d"]

From 2ca5747e6428a90d069d74b6424d4820b516391b Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 30 Sep 2026 09:58:14 -1000
Subject: [PATCH 2/7] feat(bundle): agent-env run --dry-run

materialize(plan, dry_run=True) makes every check and preflight a run makes before its first task and
writes nothing: no entity, ledger row, lock or instance. Each write is reported at the version it would
reuse or the store's next one. dry_run_bundle() returns the same to Python as a DryRun.

Ledger.check(write, versions) hashes the versions the walk passes in for what an env or agent needs, so a
need that would be written anew rewrites its dependent in the prediction as in the run. Versions are
hashed as strings, as before, so the rows a run recorded stay reused.

A step that reads what the run would write first can't be preflighted against the store; the dry run
lists those whose type has a preflight.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 README.md                           |   4 +-
 src/agent_env/bundle/__init__.py    |   2 +-
 src/agent_env/bundle/ledger.py      |  36 ++++++----
 src/agent_env/bundle/materialize.py |  71 +++++++++++++------
 src/agent_env/bundle/run.py         |  81 ++++++++++++++++++----
 src/agent_env/cli/run.py            |  50 ++++++++++++--
 tst/unit/bundle/installed_test.py   |  16 ++++-
 tst/unit/bundle/ledger_test.py      |  63 +++++++++++++----
 tst/unit/bundle/materialize_test.py |  89 ++++++++++++++++++++----
 tst/unit/bundle/run_test.py         | 102 +++++++++++++++++++++++++++-
 10 files changed, 429 insertions(+), 85 deletions(-)

diff --git a/README.md b/README.md
index 02b62327..5b56bbca 100644
--- a/README.md
+++ b/README.md
@@ -960,6 +960,7 @@ A rerun writes a new version only of what changed since the bundle last wrote it
 - **Overrides.**
   - `--model` sets the agent's model. The judge keeps its own, so scores stay comparable across models.
   - `--sandbox` sets the provider for `deploy_env`, `deploy_agent`, `deploy_sandbox` and the judge's deploy. It takes a name or a comma-separated fallback chain, and an unknown one is refused before anything is written. A chain can't create a VM, so under one a `deploy_sandbox` step in `vm` mode fails, and so does an environment behind a gateway.
+- **Dry run.** `--dry-run` makes every check the run makes before its first task, preflights included, and writes and runs nothing: no entity, ledger row, lock or instance. It prints the lines the run prints as it writes, each with the version the write would reuse or get, then the store ids the plan read (`Store refs`), the tasks that would run, each eval's tasks, and any step it couldn't preflight (`Not preflighted`), because that step reads what the run would write first and the store doesn't hold it yet. It exits 0, or 1 on a problem the run would stop at. `--model` and `--keep` change nothing it shows. `agent_env.bundle.dry_run_bundle()` returns the same to Python, as a `DryRun`.
 - **Results.** The summary gives each task's outcome, its scores labelled by the step that recorded each, its duration and its instance id, then each eval's pass count.
   - A task passes when it recorded a score and every score is at least 1. A task that recorded no score is `unscored`.
   - The command exits 1 when a run raised or left a failed step; a score below 1 doesn't change the exit status. With `agent-env --verbose run`, each run that raised also prints its traceback.
@@ -972,9 +973,10 @@ Known limits:
 - An eval in a bundle runs only the bundle's own tasks. One that names a store task is refused; run that task with `agent-env task run`.
 - A step that writes an entity under an id it makes up is refused when it runs, since that id isn't under `@local/`. This refuses `run_container_unit_tests_verifier`'s stdout artifacts and `collect_artifacts`, and `snapshot_env` unless the task names its `snapshot_id`. `verify_sandbox` writes nothing, so it runs.
 - A step's reference to one of the bundle's entities names no version, so it reads the latest version when the step runs. Another run of the same folder, made after an edit, can write a newer one first.
+- A dry run's versions and reasons hold for the stores as they are when it runs. It takes no lock, so another run writing the same ids can change them first, and it doesn't see what another bundle's pending run would write.
 - Teardown reaches only the sandboxes a run's context records. A `rubrics_verifier`'s judge, and an agent whose deploy was cancelled, terminate their own sandboxes but leave their local work folders. A run killed outright (`kill -9`) leaves everything it deployed.
 - A process a local sandbox's command detaches from itself (a double fork, `setsid`) outlives the command and the run.
-- A run still opens the configured stores: it reads store entities through them and sets up their indexes. Under a config whose stores are remote, their credentials must be available, even for a bundle, such as `hello`, that writes nothing there.
+- A run still opens the configured stores: it reads store entities through them and sets up their indexes. Under a config whose stores are remote, their credentials must be available, even for a bundle, such as `hello`, that writes nothing there. A dry run reads them the same way, so a read can create an empty default local store, or set up a remote one's indexes, as any command that reads a store does.
 
 ### The local explorer and runner
 
diff --git a/src/agent_env/bundle/__init__.py b/src/agent_env/bundle/__init__.py
index 640edd5a..5d5c6452 100644
--- a/src/agent_env/bundle/__init__.py
+++ b/src/agent_env/bundle/__init__.py
@@ -1,4 +1,4 @@
 """Bundles: folders of entities that run without a registry."""
 
 from .parse import Bundle, BundleEntry, BundleError, BundleKind, is_ignored, parse_bundle
-from .run import BundleRun, EvalRun, Outcome, RunInterrupted, TaskRun, run_bundle
+from .run import BundleRun, DryRun, EvalRun, Outcome, RunInterrupted, TaskRun, dry_run_bundle, run_bundle
diff --git a/src/agent_env/bundle/ledger.py b/src/agent_env/bundle/ledger.py
index 70a36649..7bff620b 100644
--- a/src/agent_env/bundle/ledger.py
+++ b/src/agent_env/bundle/ledger.py
@@ -66,11 +66,18 @@ class Check:
     digest: Digest | None  # None when the write's inputs aren't tracked, so it is always written
     version: int | None  # the version to reuse; None when the write needs a new one
     reasons: tuple[str, ...]
+    stored: int | None  # the store's latest version of the id when checked
+    needs: Mapping[tuple[str, str], int]  # the version hashed for each earlier write it needs, by (store, id)
 
     @property
     def unchanged(self) -> bool:
         return self.version is not None
 
+    @property
+    def next_version(self) -> int:
+        """The version a new write lands on, unless another run writes the id first."""
+        return (self.stored or 0) + 1
+
 
 class Ledger:
     """One plan's rows in the ``@local`` namespace's store."""
@@ -84,15 +91,18 @@ def __init__(self, store: LocalSqliteDocumentStore, plan: Plan):
     def for_plan(cls, plan: Plan) -> Ledger:
         return cls(get_config().local_namespace_document_store(), plan)
 
-    def check(self, write: Write) -> Check:
-        """Compare ``write`` with the latest version the ledger records for it. Checks run in the plan's
-        order, after each earlier write is recorded, so the versions a dependent records are current."""
-        digest = self.digest(write)
+    def check(self, write: Write, versions: Mapping[tuple[str, str], int]) -> Check:
+        """Compare ``write`` with the latest version the ledger records for it. ``versions`` holds, by (store, id),
+        the version each earlier write of the plan left or, in a dry run, would leave; an env or agent is hashed
+        with those of what it needs, so one written anew rewrites it too."""
+        needs = {need: versions[need] for need in write.needs}
+        stored = self._stored_version(write)
+        digest = self.digest(write, needs)
         if digest is None:
-            return Check(write, None, None, ("its inputs aren't tracked yet, so it is written every run",))
+            untracked = ("its inputs aren't tracked yet, so it is written every run",)
+            return Check(write, None, None, untracked, stored, needs)
         row = self._latest(write.kind.store, write.id)
         recorded = row["version"] if row else None
-        stored = self._stored_version(write)
         reasons = []
         if row is not None and row["bundle"] != self._bundle:
             reasons.append(f"last written by the bundle {row['bundle']}")
@@ -102,10 +112,11 @@ def check(self, write: Write) -> Check:
             reasons.append(f"the store's latest, v{stored}, wasn't recorded by this bundle")
         if row is not None:
             reasons.extend(_changes(row, digest))
-        return Check(write, digest, None if reasons else recorded, tuple(reasons))
+        return Check(write, digest, None if reasons else recorded, tuple(reasons), stored, needs)
 
-    def digest(self, write: Write) -> Digest | None:
-        """What ``write`` is made from, or None when the ledger can't tell it all, so it is written every run."""
+    def digest(self, write: Write, needs: Mapping[tuple[str, str], int]) -> Digest | None:
+        """What ``write`` is made from, given the version of each write it needs, or None when the ledger can't
+        tell it all, so it is written every run."""
         if not _tracked(write):
             return None
         inputs = {"type": _type(write), "config": _sha256(_canonical(write.source.config)),
@@ -118,8 +129,7 @@ def digest(self, write: Write) -> Digest | None:
             # a store entity it names without a version getting a new one, means it must be written again. A
             # task or eval names its references without a version.
             for store, id in write.needs:
-                row = self._latest(store, id)
-                inputs["needs"][f"{store} {id}"] = str(row["version"]) if row else "unwritten"
+                inputs["needs"][f"{store} {id}"] = str(needs[store, id])
             for ref in write.source.references:
                 if ref.local is None and ref.version is None:
                     inputs["store_refs"][f"{ref.kind} {ref.id}"] = str(self._plan.store_latest[ref.kind, ref.id])
@@ -139,7 +149,7 @@ def record(self, check: Check, write: Callable[[], int]) -> int:
         self._store.insert(LEDGER_COLLECTION, pending)
         version = write()
         done = {**pending, "status": "done", "version": version, "at": _now()}
-        if check.digest is not None and self.digest(check.write) != check.digest:
+        if check.digest is not None and self.digest(check.write, check.needs) != check.digest:
             # A file changed during the write, so what the version holds is unknown: the next run writes it again.
             done.update(digest=None, inputs=None)
         self._store.replace(LEDGER_COLLECTION, Filter.of(**key, status="pending"), done, upsert=True)
@@ -228,7 +238,7 @@ def _changes(row: Mapping[str, Any], digest: Digest) -> list[str]:
             reasons.append(f"{label}: {shown}")
     needs_before, needs_after = before["needs"], after["needs"]
     added, removed, changed = _diff(needs_before, needs_after)
-    reasons.extend(f"{name} was written anew (v{needs_before[name]} → v{needs_after[name]})" for name in changed)
+    reasons.extend(f"{name} is written anew (v{needs_before[name]} → v{needs_after[name]})" for name in changed)
     reasons.extend(f"now needs {name}" for name in added)
     reasons.extend(f"no longer needs {name}" for name in removed)
     refs_before, refs_after = before["store_refs"], after["store_refs"]
diff --git a/src/agent_env/bundle/materialize.py b/src/agent_env/bundle/materialize.py
index 5cb31a8d..e6da717c 100644
--- a/src/agent_env/bundle/materialize.py
+++ b/src/agent_env/bundle/materialize.py
@@ -6,14 +6,15 @@
 builds and preflights every task before writing any of them, and writes the evals last, since they name the
 tasks. Each write goes through the ledger, so one whose inputs haven't changed reuses the version the bundle
 last wrote. A reused task keeps whatever its steps took from config when it was first written, such as a
-rubrics verifier's default judge model.
+rubrics verifier's default judge model. A dry run takes the same path, refusals, checks and preflights
+included, without the lock or a single write.
 """
 
 from __future__ import annotations
 
 import copy
-from collections.abc import Callable, Iterator
-from contextlib import contextmanager
+from collections.abc import Callable, Iterator, Mapping
+from contextlib import contextmanager, nullcontext
 from dataclasses import dataclass
 from typing import Any
 
@@ -24,6 +25,7 @@
 from agent_env.store.routing import namespace_routing_enabled
 from agent_env.task import Task
 from agent_env.task_step.registry import get_task_step_registry
+from agent_env.task_step.task_step import TaskStep
 
 from ._fs import relative, with_article
 from .authoring import AuthoringContext
@@ -34,7 +36,8 @@
 
 @dataclass(frozen=True)
 class Materialized:
-    """One write: the version it left in the store, and whether that is the one the bundle last wrote."""
+    """One write: the version it left in the store, or in a dry run would leave, and whether that is the one the
+    bundle last wrote."""
 
     write: Write
     version: int
@@ -44,10 +47,12 @@ class Materialized:
 
 @dataclass(frozen=True)
 class Materialization:
-    """What materializing a plan left in the store, one ``Materialized`` per write."""
+    """What materializing a plan left in the store, or in a dry run would leave, one ``Materialized`` per write."""
 
     plan: Plan
     writes: tuple[Materialized, ...]  # in the plan's order
+    # A dry run's steps with a preflight that read what it would write first, so the store can't check them yet.
+    not_preflighted: tuple[tuple[Write, TaskStep], ...] = ()
 
     def version_of(self, store: str, id: str) -> int:
         for done in self.writes:
@@ -59,12 +64,17 @@ def version_of(self, store: str, id: str) -> int:
 def materialize(
     plan: Plan,
     *,
+    dry_run: bool = False,
     on_wait: Callable[[], None] | None = None,
     on_write: Callable[[Materialized], None] | None = None,
 ) -> Materialization:
     """Write ``plan``'s entities, tasks and evals. The first write that fails stops it, and every earlier
     one stays: they're in the ledger, so the next run reuses them. ``on_wait`` is called when another run
-    holds a lock this one needs, and ``on_write`` after each write, reused or not."""
+    holds a lock this one needs, and ``on_write`` after each write, reused or not.
+
+    ``dry_run`` checks and preflights what a run would, and writes nothing: no entity, ledger row or lock.
+    Each write gets the version it would reuse or the store's next, which another run can take first. A step
+    reading what the run would write first isn't preflighted, since the store doesn't hold it yet."""
     _refuse_unwritable(plan)
     if not namespace_routing_enabled():
         raise RuntimeError("materializing a bundle needs namespace routing, which the agent-env CLI turns on; "
@@ -74,27 +84,32 @@ def materialize(
     tasks = [write for write in plan.writes if write.kind is BundleKind.TASK]
     evals = [write for write in plan.writes if write.kind is BundleKind.EVAL]
     done: dict[tuple[str, str], Materialized] = {}
+    versions: dict[tuple[str, str], int] = {}
 
     def through_ledger(write: Write, write_fn: Callable[[], int]) -> None:
-        done[_key(write)] = _through_ledger(plan, ledger, write, write_fn)
+        done[_key(write)] = _through_ledger(plan, ledger, write, versions, write_fn, dry_run)
+        versions[_key(write)] = done[_key(write)].version
         if on_write is not None:
             on_write(done[_key(write)])
 
-    with materializing(plan.bundle.bundle, on_wait):
+    with nullcontext() if dry_run else materializing(plan.bundle.bundle, on_wait):
         for write in entities:
             through_ledger(write, lambda: _WRITERS[write.kind](plan, write))
-        built, problems = {}, []
+        unwritten = {item.write.id for item in done.values() if not item.reused} if dry_run else set()
+        built, problems, unchecked = {}, [], []
         for write in tasks:
             with _noted(plan, write, "preflighting"):
                 built[write.id] = _build(write)
-                problems.extend(f"{_path(plan, write)}: {problem}" for problem in _preflight(write, built[write.id]))
+                found, skipped = _preflight(write, built[write.id], unwritten)
+            problems.extend(f"{_path(plan, write)}: {problem}" for problem in found)
+            unchecked.extend((write, step) for step in skipped)
         if problems:
             raise BundleError(problems)
         for write in tasks:
             through_ledger(write, lambda: Task.put(id=write.id, steps=built[write.id].steps).version)
         for write in evals:
             through_ledger(write, lambda: _WRITERS[write.kind](plan, write))
-    return Materialization(plan, tuple(done[_key(write)] for write in plan.writes))
+    return Materialization(plan, tuple(done[_key(write)] for write in plan.writes), tuple(unchecked))
 
 
 def _write_artifact(plan: Plan, write: Write) -> int:
@@ -160,10 +175,16 @@ def _unwritable(write: Write) -> str | None:
     return None
 
 
-def _through_ledger(plan: Plan, ledger: Ledger, write: Write, write_fn: Callable[[], int]) -> Materialized:
-    with _noted(plan, write, "writing"):
-        check = ledger.check(write)
-        version = check.version if check.unchanged else ledger.record(check, write_fn)
+def _through_ledger(plan: Plan, ledger: Ledger, write: Write, versions: Mapping[tuple[str, str], int],
+                    write_fn: Callable[[], int], dry_run: bool) -> Materialized:
+    with _noted(plan, write, "checking" if dry_run else "writing"):
+        check = ledger.check(write, versions)
+        if check.unchanged:
+            version = check.version
+        elif dry_run:
+            version = check.next_version
+        else:
+            version = ledger.record(check, write_fn)
     return Materialized(write, version, check.unchanged, check.reasons)
 
 
@@ -180,15 +201,21 @@ def _build(write: Write) -> Task:
     return Task(id=write.id, version=None, steps=[build_step(step) for step in write.source.config])
 
 
-def _preflight(write: Write, task: Task) -> list[str]:
-    """``task``'s preflight problems. A step reading one of the task's own outputs is skipped: the output
-    only exists once the task runs."""
+def _preflight(write: Write, task: Task, unwritten: set[str]) -> tuple[list[str], list[TaskStep]]:
+    """``task``'s preflight problems, and the steps with a preflight left unchecked because they read an id in
+    ``unwritten``, which a dry run would write first. A step reading one of the task's own outputs is skipped
+    too, and not listed: the output only exists once the task runs."""
     outputs = {output.id for output in write.source.outputs}
-    problems = []
+    problems, unchecked = [], []
     for config, step in zip(write.source.config, task.steps):
-        if not _reads_any(config, outputs):
-            problems.extend(step.preflight())
-    return problems
+        if _reads_any(config, outputs):
+            continue
+        if _reads_any(config, unwritten):
+            if type(step).preflight is not TaskStep.preflight:
+                unchecked.append(step)
+            continue
+        problems.extend(step.preflight())
+    return problems, unchecked
 
 
 def _reads_any(step: dict, ids: set[str]) -> bool:
diff --git a/src/agent_env/bundle/run.py b/src/agent_env/bundle/run.py
index 01b2d043..01dd1888 100644
--- a/src/agent_env/bundle/run.py
+++ b/src/agent_env/bundle/run.py
@@ -5,7 +5,8 @@
 and the others go on. Each run's sandboxes are torn down as it ends, unless ``keep`` holds them up. Ctrl-C or
 SIGTERM cancels the runs: each one that started is marked cancelled and torn down, and a second one stops the
 teardown. A bundle's evals run only the bundle's own tasks for now, so one naming a store task is refused
-before anything is written.
+before anything is written. A dry run makes every check the run makes before its first task, and writes and
+runs nothing.
 """
 
 from __future__ import annotations
@@ -139,6 +140,19 @@ def swapped(runs: tuple[TaskRun, ...]) -> tuple[TaskRun, ...]:
                        evals=tuple(replace(eval_run, runs=swapped(eval_run.runs)) for eval_run in self.evals))
 
 
+@dataclass(frozen=True)
+class DryRun:
+    """What ``run_bundle`` would write and run, found without writing or running anything."""
+
+    materialization: Materialization  # each write at the version it would leave, and why
+    runs: tuple[BundleEntry, ...]  # the tasks that would run, each once, in the plan's order
+    skipped: tuple[BundleEntry, ...]  # the tasks no eval names, which running every eval leaves out
+
+    def path(self, entry: BundleEntry) -> str:
+        """``entry``'s path in the bundle (``tasks/hello.json``)."""
+        return relative(self.materialization.plan.bundle.bundle.root, entry.path)
+
+
 class RunInterrupted(KeyboardInterrupt):
     """Ctrl-C or SIGTERM stopped ``run_bundle``. ``result`` holds every run, those it cancelled included, each
     torn down unless a second signal stopped that."""
@@ -184,21 +198,18 @@ def run_bundle(
         build_sandbox_provider(sandbox)
     say = _progress(on_progress)
     with namespace_routing():
-        plan = plan_bundle(resolve_bundle(parse_bundle(Path(root), id_root=id_root)), tasks=tasks, evals=evals)
-        refuse_store_tasks(plan)
+        plan = _planned(root, tasks, evals, id_root)
         materialization = materialize(
             plan,
             on_wait=lambda: say("waiting for another agent-env run to finish writing this bundle's ids"),
             on_write=lambda done: say(_written(plan, done)),
         )
-        wanted = {entry.entry.id for entry in plan.tasks}
-        wanted.update(ref.id for entry in plan.evals for ref in entry.references)
-        entries = [write.source.entry for write in plan.writes if write.kind is BundleKind.TASK and write.id in wanted]
+        entries = _to_run(plan)
         to_run = [(entry, Task.get(entry.id, materialization.version_of("task", entry.id))) for entry in entries]
         with Interrupts() as interrupts:
             runs = interrupts.run(_run_all(plan, to_run, model, sandbox, keep, say, interrupts))
             _mark_cancelled(runs, interrupts.reason)
-            result = _bundle_run(plan, materialization, runs, wanted, every=not tasks and not evals)
+            result = _bundle_run(plan, materialization, runs, _skipped(plan, entries, every=not tasks and not evals))
             if not interrupts.count:
                 return result
             if interrupts.count == 1:
@@ -209,6 +220,32 @@ def run_bundle(
             raise RunInterrupted(result, interrupts.signum)
 
 
+def dry_run_bundle(
+    root: Path | str,
+    *,
+    tasks: Sequence[str] = (),
+    evals: Sequence[str] = (),
+    sandbox: str | None = None,
+    on_progress: Callable[[str], None] | None = None,
+    id_root: str | None = None,
+) -> DryRun:
+    """Check what ``run_bundle`` with these arguments checks before its first task runs, and say what it would
+    write and run, writing and running nothing: no entity, ledger row, lock or instance. Each write's version is
+    the one it would reuse or the store's next, which another run writing the id first can take. ``on_progress``
+    is given the line ``run_bundle`` gives for each write, as it is checked.
+
+    It raises what ``run_bundle`` raises before anything is written, except that it runs no event loop, so a
+    running one isn't refused."""
+    if sandbox:
+        build_sandbox_provider(sandbox)
+    say = _progress(on_progress)
+    with namespace_routing():
+        plan = _planned(root, tasks, evals, id_root)
+        materialization = materialize(plan, dry_run=True, on_write=lambda done: say(_written(plan, done)))
+        runs = _to_run(plan)
+        return DryRun(materialization, runs, _skipped(plan, runs, every=not tasks and not evals))
+
+
 def _mark_cancelled(runs: tuple[TaskRun, ...], reason: str) -> None:
     """Mark each run a signal stopped mid-way as cancelled. It runs once the loop has closed, which waits for every
     store write the runs' unwinding started, so the mark lands last: a failure write a second signal cut short
@@ -219,21 +256,39 @@ def _mark_cancelled(runs: tuple[TaskRun, ...], reason: str) -> None:
                 record_task_cancelled(run.instance_id, reason, Task._utc_now_str())
 
 
-def _bundle_run(plan: Plan, materialization: Materialization, runs: tuple[TaskRun, ...], wanted: set[str],
-                every: bool) -> BundleRun:
+def _bundle_run(plan: Plan, materialization: Materialization, runs: tuple[TaskRun, ...],
+                skipped: tuple[BundleEntry, ...]) -> BundleRun:
     by_id = {run.task.id: run for run in runs}
     eval_runs = tuple(
         EvalRun(entry.entry, materialization.version_of("eval", entry.entry.id),
                 tuple(by_id[ref.id] for ref in entry.references))
         for entry in plan.evals
     )
-    skipped = ()
-    if every:
-        skipped = tuple(entry.entry for entry in plan.bundle.entries
-                        if entry.entry.kind is BundleKind.TASK and entry.entry.id not in wanted)
     return BundleRun(materialization, runs, eval_runs, skipped)
 
 
+def _planned(root: Path | str, tasks: Sequence[str], evals: Sequence[str], id_root: str | None) -> Plan:
+    plan = plan_bundle(resolve_bundle(parse_bundle(Path(root), id_root=id_root)), tasks=tasks, evals=evals)
+    refuse_store_tasks(plan)
+    return plan
+
+
+def _to_run(plan: Plan) -> tuple[BundleEntry, ...]:
+    """The tasks selected on their own and every task a selected eval names, each once, in the plan's order."""
+    wanted = {entry.entry.id for entry in plan.tasks}
+    wanted.update(ref.id for entry in plan.evals for ref in entry.references)
+    return tuple(write.source.entry for write in plan.writes if write.kind is BundleKind.TASK and write.id in wanted)
+
+
+def _skipped(plan: Plan, runs: Sequence[BundleEntry], every: bool) -> tuple[BundleEntry, ...]:
+    """When every eval runs, the tasks none of them names."""
+    if not every:
+        return ()
+    ran = {entry.id for entry in runs}
+    return tuple(entry.entry for entry in plan.bundle.entries
+                 if entry.entry.kind is BundleKind.TASK and entry.entry.id not in ran)
+
+
 def refuse_store_tasks(plan: Plan) -> None:
     """Refuse a selected eval that names a store task: a bundle's evals run only the bundle's own tasks."""
     problems = [
diff --git a/src/agent_env/cli/run.py b/src/agent_env/cli/run.py
index 4545a4ea..0a0673de 100644
--- a/src/agent_env/cli/run.py
+++ b/src/agent_env/cli/run.py
@@ -10,7 +10,8 @@
 
 import click
 
-from agent_env.bundle import BundleError, BundleKind, BundleRun, Outcome, RunInterrupted, TaskRun, run_bundle
+from agent_env.bundle import (BundleError, BundleKind, BundleRun, DryRun, Outcome, RunInterrupted, TaskRun,
+                              dry_run_bundle, run_bundle)
 from agent_env.bundle.installed import InstalledBundle, checked, find_bundle, installed_bundles, run_name
 from agent_env.plugins._discovery import discovery_error
 from agent_env.providers.sandbox_providers.local_sandbox import LocalSandbox
@@ -23,6 +24,7 @@
            Outcome.CANCELLED: "yellow"}
 _ENV_ENDPOINTS = (("mcp", "mcp_url"), ("gateway", "gateway_url"), ("pgweb", "db_web_url"), ("db-mcp", "db_mcp_url"),
                   ("vnc", "vnc_url"), ("expires", "expires_at_utc"))
+_DRY_RUN = "Dry run: nothing is written or run."
 
 
 @click.command()
@@ -35,9 +37,11 @@
                    "fallback chain.")
 @click.option("--keep", is_flag=True,
               help="Keep the sandboxes up after the runs, print their endpoints, and tear them down on Ctrl-C.")
+@click.option("--dry-run", is_flag=True,
+              help="Check the bundle as the run does and show what it would write and run; write and run nothing.")
 @click.pass_context
 def run(ctx: click.Context, bundle: str | None, tasks: tuple[str, ...], evals: tuple[str, ...], model: str | None,
-        sandbox: str | None, keep: bool):
+        sandbox: str | None, keep: bool, dry_run: bool):
     """Run a bundle's tasks and evals, or list the installed bundles.
 
     BUNDLE is a folder, or the name of a bundle an installed package provides (PACKAGE/NAME when several
@@ -47,18 +51,27 @@ def run(ctx: click.Context, bundle: str | None, tasks: tuple[str, ...], evals: t
     once, at most four at a time, and the run exits 1 if any of them failed. With --verbose, a run that
     raised prints its traceback.
 
+    --dry-run makes the checks the run makes before its first task, reading the stores as the run does, and
+    prints what it would write and run, writing and running nothing. It exits 1 on a problem the run would
+    stop at, and 0 otherwise; --model and --keep change nothing it shows.
+
     Each run's sandboxes are torn down as it ends; its instance and outputs stay. --keep holds them up
     until Ctrl-C instead. Ctrl-C or SIGTERM mid-run cancels the runs, tears them down, prints what ran and
     exits 130 (143 for SIGTERM); a second one stops the teardown and prints what is still up.
     """
     if bundle is None:
-        if tasks or evals or model or sandbox or keep:
-            raise click.UsageError("--task, --eval, --model, --sandbox and --keep need a BUNDLE to run")
+        if tasks or evals or model or sandbox or keep or dry_run:
+            raise click.UsageError("--task, --eval, --model, --sandbox, --keep and --dry-run need a BUNDLE to run")
         _list_installed()
         return
     root, id_root = _locate(bundle)
     verbose = ctx.find_root().params.get("verbose")
     try:
+        if dry_run:
+            click.echo(_DRY_RUN)
+            _report_dry_run(dry_run_bundle(root, tasks=tasks, evals=evals, sandbox=sandbox, on_progress=click.echo,
+                                           id_root=id_root))
+            return
         result = run_bundle(root, tasks=tasks, evals=evals, model=model, sandbox=sandbox, on_progress=click.echo,
                             id_root=id_root, keep=keep)
     except BundleError as e:
@@ -174,6 +187,35 @@ def _summarize(result: BundleRun, verbose: bool) -> None:
         _print_tracebacks(result)
 
 
+def _report_dry_run(dry: DryRun) -> None:
+    materialization = dry.materialization
+    plan = materialization.plan
+    click.echo()
+    if plan.store_refs:
+        click.echo("Store refs:")
+    for ref in plan.store_refs:
+        if ref.version is None:
+            click.echo(f"  {ref.kind} {ref.id} v{plan.store_latest[ref.kind, ref.id]}, the latest")
+        else:
+            click.echo(f"  {ref.kind} {ref.id} v{ref.version}")
+    click.echo("Would run:")
+    for entry in dry.runs:
+        click.echo(f"  {dry.path(entry)} v{materialization.version_of('task', entry.id)}")
+    if plan.evals:
+        click.echo("Evals:")
+    for entry in plan.evals:
+        named = ", ".join(dry.path(ref.local) for ref in entry.references)
+        click.echo(f"  {dry.path(entry.entry)} v{materialization.version_of('eval', entry.entry.id)}: {named}")
+    for entry in dry.skipped:
+        click.echo(f"{dry.path(entry)} isn't named by any eval, so it wouldn't run; run it with "
+                   f"--task {shlex.quote(entry.name)}")
+    if materialization.not_preflighted:
+        click.echo("Not preflighted, since each reads what the run would write first:")
+    for write, step in materialization.not_preflighted:
+        click.echo(f"  {dry.path(write.source.entry)}: step {step.id!r} ({step.type})")
+    click.echo(_DRY_RUN)
+
+
 def _print_tracebacks(result: BundleRun) -> None:
     for run in result.runs:
         if run.error is not None:
diff --git a/tst/unit/bundle/installed_test.py b/tst/unit/bundle/installed_test.py
index 51e596e1..9af92eb7 100644
--- a/tst/unit/bundle/installed_test.py
+++ b/tst/unit/bundle/installed_test.py
@@ -235,10 +235,10 @@ def test_the_cli_says_so_when_no_bundles_are_installed_and_needs_a_bundle_for_it
     _registered(monkeypatch)
 
     assert CliRunner().invoke(cli, ["run"]).output == "No bundles are installed. agent-env run PATH runs a folder.\n"
-    for option in (["--task", "t"], ["--keep"]):
+    for option in (["--task", "t"], ["--keep"], ["--dry-run"]):
         usage = CliRunner().invoke(cli, ["run", *option])
         assert usage.exit_code == 2
-        assert "--task, --eval, --model, --sandbox and --keep need a BUNDLE to run" in usage.output
+        assert "--task, --eval, --model, --sandbox, --keep and --dry-run need a BUNDLE to run" in usage.output
 
 
 def test_the_cli_runs_an_installed_bundle_by_name_with_ids_rooted_at_its_package(site, monkeypatch, quiet_logs,
@@ -253,6 +253,18 @@ def test_the_cli_runs_an_installed_bundle_by_name_with_ids_rooted_at_its_package
     assert "instance @local/demo-bundles/hello/t-" in result.output
 
 
+def test_the_cli_dry_runs_an_installed_bundle_by_name_and_writes_nothing(site, monkeypatch, quiet_logs, local_stores):
+    _package(site, "demo_bundles", ["hello"])
+    _registered(monkeypatch, _ep("hello", "demo_bundles"))
+
+    result = CliRunner().invoke(cli, ["run", "hello", "--dry-run"])
+
+    assert result.exit_code == 0, result.output
+    assert "tasks/t.json: v1 (new)\n" in result.output
+    assert "Would run:\n  tasks/t.json v1\n" in result.output
+    assert not local_stores.local_namespace_document_store().path.exists()
+
+
 def test_a_folder_wins_over_an_installed_name_and_a_folder_that_isnt_a_bundle_names_it(site, monkeypatch, tmp_path,
                                                                                        quiet_logs, local_stores):
     _package(site, "demo_bundles", ["hello"])
diff --git a/tst/unit/bundle/ledger_test.py b/tst/unit/bundle/ledger_test.py
index 32580fec..8b3cb062 100644
--- a/tst/unit/bundle/ledger_test.py
+++ b/tst/unit/bundle/ledger_test.py
@@ -120,11 +120,21 @@ def _written(write):
 
 def _record_changed(ledger, writes):
     """Check and record ``writes`` in order, as the materializer does; the checks, by id."""
-    checks = {}
+    checks, versions = {}, {}
     for write in writes:
-        check = ledger.check(write)
-        if not check.unchanged:
-            ledger.record(check, lambda: _written(write))
+        check = ledger.check(write, versions)
+        version = check.version if check.unchanged else ledger.record(check, lambda: _written(write))
+        versions[write.kind.store, write.id] = version
+        checks[write.id] = check
+    return checks
+
+
+def _checked(ledger, writes):
+    """Check ``writes`` in order without recording any, as a dry run does; the checks, by id."""
+    checks, versions = {}, {}
+    for write in writes:
+        check = ledger.check(write, versions)
+        versions[write.kind.store, write.id] = check.version if check.unchanged else check.next_version
         checks[write.id] = check
     return checks
 
@@ -160,10 +170,35 @@ def test_a_rewritten_dependency_rewrites_an_env_that_records_its_version_but_not
     checks = _run(bundle_dir)
 
     assert checks[GREETING].reasons == ("files changed: hello.txt",)
-    assert checks[f"{ROOT}/shelf"].reasons == (f"artifact {GREETING} was written anew (v1 → v2)",)
+    assert checks[f"{ROOT}/shelf"].reasons == (f"artifact {GREETING} is written anew (v1 → v2)",)
     assert checks[f"{ROOT}/t"].unchanged
 
 
+def test_a_dry_check_hashes_the_versions_its_needs_would_get_so_it_predicts_the_run(bundle_dir):
+    _run(bundle_dir)
+    (bundle_dir / "artifacts/greeting/hello.txt").write_text("hello again\n")
+    plan = plan_of(bundle_dir)
+
+    dry = _checked(Ledger.for_plan(plan), plan.writes)
+    real = _record_changed(Ledger.for_plan(plan), plan.writes)
+
+    assert dry[f"{ROOT}/shelf"].reasons == (f"artifact {GREETING} is written anew (v1 → v2)",)
+    assert {id: (check.unchanged, check.reasons) for id, check in dry.items()} == {
+        id: (check.unchanged, check.reasons) for id, check in real.items()}
+    assert {id: check.version if check.unchanged else check.next_version for id, check in dry.items()} == {
+        id: _latest_version(_COLLECTIONS[check.write.kind.store], id) for id, check in real.items()}
+
+
+def test_a_need_is_hashed_as_its_version_string_so_the_rows_a_run_recorded_stay_reused(bundle_dir):
+    _run(bundle_dir)
+    plan = plan_of(bundle_dir)
+
+    checks = _checked(Ledger.for_plan(plan), plan.writes)
+
+    assert checks[f"{ROOT}/shelf"].digest.inputs["needs"] == {f"artifact {GREETING}": "1"}
+    assert all(check.unchanged for check in checks.values())
+
+
 def test_a_second_file_flips_the_inferred_type_and_says_so(bundle_dir):
     _run(bundle_dir)
     (bundle_dir / "artifacts/greeting/extra.txt").write_text("x")
@@ -212,7 +247,7 @@ def test_a_version_the_bundle_didnt_record_is_written_over(bundle_dir):
     check = _run(bundle_dir)[GREETING]
 
     assert check.reasons == ("the store's latest, v3, wasn't recorded by this bundle",)
-    assert _latest_version("artifacts", GREETING) == 4
+    assert check.next_version == _latest_version("artifacts", GREETING) == 4
 
 
 def test_a_version_written_by_another_bundle_is_written_over(bundle_dir):
@@ -245,10 +280,10 @@ def edited_meanwhile():
         hello.write_text("edited\n")
         return _written(write)
 
-    ledger.record(ledger.check(write), edited_meanwhile)
+    ledger.record(ledger.check(write, {}), edited_meanwhile)
     hello.write_text("hello\n")
 
-    assert ledger.check(write).reasons == ("the ledger doesn't know what its last version was made from",)
+    assert ledger.check(write, {}).reasons == ("the ledger doesn't know what its last version was made from",)
 
 
 def test_a_new_digest_scheme_rewrites_everything_and_says_only_that(bundle_dir, monkeypatch):
@@ -269,10 +304,10 @@ def crash():
         raise RuntimeError("the write crashed")
 
     with pytest.raises(RuntimeError):
-        ledger.record(ledger.check(write), crash)
-    assert ledger.check(write).reasons == ("new",)
+        ledger.record(ledger.check(write, {}), crash)
+    assert ledger.check(write, {}).reasons == ("new",)
 
-    ledger.record(ledger.check(write), lambda: _written(write))
+    ledger.record(ledger.check(write, {}), lambda: _written(write))
     rows = local_store().query(LEDGER_COLLECTION, Filter.of(id=GREETING))
     assert sorted((row["bundle"], row["status"], row.get("version")) for row in rows) == [
         ("@local/~/elsewhere/triage", "pending", None), (plan.bundle.bundle.id_root, "done", 1),
@@ -334,9 +369,7 @@ def _image_document(version):
 
 def test_reads_never_create_the_local_namespaces_store(bundle_dir):
     plan = plan_of(bundle_dir)
-    ledger = Ledger.for_plan(plan)
-    for write in plan.writes:
-        ledger.check(write)
+    _checked(Ledger.for_plan(plan), plan.writes)
 
     assert not local_store().path.exists()
 
@@ -347,7 +380,7 @@ def test_the_ledger_never_touches_the_configured_store(bundle_dir, cli_routing):
     ledger = Ledger.for_plan(plan)
 
     _record_changed(ledger, plan.writes)
-    assert all(ledger.check(write).unchanged for write in plan.writes)
+    assert all(check.unchanged for check in _checked(ledger, plan.writes).values())
     (bundle_dir / "artifacts/greeting/hello.txt").write_text("hello again\n")
     rewritten = _record_changed(ledger, plan.writes)
 
diff --git a/tst/unit/bundle/materialize_test.py b/tst/unit/bundle/materialize_test.py
index d3c971c3..a9f8acf5 100644
--- a/tst/unit/bundle/materialize_test.py
+++ b/tst/unit/bundle/materialize_test.py
@@ -16,9 +16,11 @@
 from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse
 from agent_env.artifact.store import get_artifact_store
 from agent_env.bundle import BundleError
-from agent_env.bundle.ledger import LEDGER_COLLECTION
+from agent_env.bundle import materialize as materialize_module
+from agent_env.bundle.ledger import LEDGER_COLLECTION, Ledger
 from agent_env.bundle.materialize import materialize
 from agent_env.config import configure
+from agent_env.config.paths import state_root
 from agent_env.config.runtime import Config
 from agent_env.entity_refs import EntityRef, RefRole
 from agent_env.env.env import Env
@@ -127,12 +129,15 @@ def bundle_dir(tmp_path, monkeypatch, local_stores, cli_routing):
     return layout(tmp_path / "triage", LAYOUT)
 
 
+_RUN_OR_DRY_RUN = pytest.mark.parametrize("dry_run", [False, True], ids=["run", "dry-run"])
+
+
 def _steps(root, steps):
     (root / "tasks/t.json").write_text(json.dumps(steps))
 
 
-def _run(root):
-    return materialize(plan_of(root))
+def _run(root, dry_run=False):
+    return materialize(plan_of(root), dry_run=dry_run)
 
 
 def _summary(materialization):
@@ -172,7 +177,8 @@ def test_a_rerun_reuses_every_version_and_an_edit_rewrites_only_what_it_changed(
     assert third.version_of("task", f"{ROOT}/t") == first.version_of("task", f"{ROOT}/t")
 
 
-def test_what_has_no_writer_yet_is_refused_before_anything_is_written(bundle_dir):
+@_RUN_OR_DRY_RUN
+def test_what_has_no_writer_yet_is_refused_before_anything_is_written(bundle_dir, dry_run):
     layout(bundle_dir, {
         "envs/tickets/Dockerfile": "FROM scratch\n",
         "envs/imaged/env.toml": 'type = "imaged_materialize_test"\n',
@@ -192,7 +198,7 @@ def test_what_has_no_writer_yet_is_refused_before_anything_is_written(bundle_dir
         {"id": "snap", "type": "load_artifact", "env_id": "tickets", "artifact_id": "snap"},
     ])
 
-    assert sorted(_problems(lambda: _run(bundle_dir))) == [
+    assert sorted(_problems(lambda: _run(bundle_dir, dry_run))) == [
         "agents/solver: writing an image built from Dockerfile isn't supported yet",
         "artifacts/base-mcp: writing a docker_image artifact isn't supported yet",
         "artifacts/snap: writing an environment artifact isn't supported yet",
@@ -349,12 +355,13 @@ def test_an_eval_may_share_its_tasks_name(bundle_dir):
     assert Eval.get(f"{ROOT}/t").tasks == [EvalTask(f"{ROOT}/t")]
 
 
-def test_without_namespace_routing_nothing_is_written(bundle_dir):
+@_RUN_OR_DRY_RUN
+def test_without_namespace_routing_nothing_is_written(bundle_dir, dry_run):
     plan = plan_of(bundle_dir)
     disable_namespace_routing()
 
     with pytest.raises(RuntimeError, match="needs namespace routing"):
-        materialize(plan)
+        materialize(plan, dry_run=dry_run)
     assert not local_store().path.exists()
 
 
@@ -376,12 +383,14 @@ def test_tasks_are_preflighted_together_after_the_entities_and_none_is_written_i
     assert fixed[f"{ROOT}/t"] == (1, False, ("new",))
 
 
-def test_a_task_the_ledger_would_reuse_is_still_preflighted(bundle_dir, monkeypatch):
+@_RUN_OR_DRY_RUN
+def test_a_task_the_ledger_would_reuse_is_still_preflighted(bundle_dir, monkeypatch, dry_run):
+    """In a dry run too: the store holds the reused artifact the step reads, as the run will read it."""
     _steps(bundle_dir, [{"id": "check", "type": "checked_materialize_test", "artifact_id": "greeting"}])
     _run(bundle_dir)
     monkeypatch.setattr(_Checked, "problems", ["the script is gone"])
 
-    assert _problems(lambda: _run(bundle_dir)) == ("tasks/t.json: the script is gone",)
+    assert _problems(lambda: _run(bundle_dir, dry_run)) == ("tasks/t.json: the script is gone",)
 
 
 def test_a_step_reading_its_own_tasks_output_is_not_preflighted(bundle_dir, monkeypatch):
@@ -417,7 +426,8 @@ def test_a_step_writing_an_output_is_still_preflighted(bundle_dir, monkeypatch):
     assert _problems(lambda: _run(bundle_dir)) == ("tasks/t.json: the writer's script isn't there",)
 
 
-def test_a_preflight_that_raises_names_its_task(bundle_dir, monkeypatch):
+@_RUN_OR_DRY_RUN
+def test_a_preflight_that_raises_names_its_task(bundle_dir, monkeypatch, dry_run):
     _steps(bundle_dir, [{"id": "check", "type": "checked_materialize_test"}])
 
     def unreachable(self):
@@ -425,7 +435,7 @@ def unreachable(self):
 
     monkeypatch.setattr(_Checked, "preflight", unreachable)
     with pytest.raises(ConnectionError) as caught:
-        _run(bundle_dir)
+        _run(bundle_dir, dry_run)
     assert caught.value.__notes__ == [f"while preflighting tasks/t.json ({ROOT}/t)"]
 
 
@@ -458,19 +468,72 @@ def from_toml(cls, data, ctx):
     assert (after[f"{ROOT}/docs"], after[f"{ROOT}/greeting"][:2]) == ((1, True, ()), (1, False))
 
 
-def test_each_write_is_reported_reused_or_not(bundle_dir):
+@_RUN_OR_DRY_RUN
+def test_each_write_is_reported_reused_or_not(bundle_dir, dry_run):
     reported = []
     (bundle_dir / "artifacts/greeting/hello.txt").write_text("hello again\n")
     _run(bundle_dir)
     (bundle_dir / "artifacts/greeting/hello.txt").write_text("hello\n")
 
-    done = materialize(plan_of(bundle_dir), on_write=reported.append)
+    done = materialize(plan_of(bundle_dir), dry_run=dry_run, on_write=reported.append)
 
     assert sorted(reported, key=lambda item: item.write.id) == sorted(done.writes, key=lambda item: item.write.id)
     assert [(item.write.id, item.reused) for item in reported] == [
         (f"{ROOT}/docs", True), (f"{ROOT}/greeting", False), (f"{ROOT}/t", True)]
 
 
+def test_a_dry_run_predicts_what_materializing_then_writes(bundle_dir):
+    def agree():
+        predicted = _summary(_run(bundle_dir, dry_run=True))
+        assert predicted == _summary(_run(bundle_dir))
+        return predicted
+
+    assert set(agree().values()) == {(1, False, ("new",))}
+    assert {reused for _, reused, _ in agree().values()} == {True}
+    (bundle_dir / "artifacts/docs/b.md").write_text("b2\n")
+    assert agree()[f"{ROOT}/docs"] == (2, False, ("files changed: b.md",))
+    FileArtifact.put_bytes(f"{ROOT}/greeting", description="by hand", filename="hello.txt", content=b"by hand")
+    assert agree()[f"{ROOT}/greeting"] == (3, False, ("the store's latest, v2, wasn't recorded by this bundle",))
+
+
+def test_a_dry_run_calls_no_writer_and_takes_no_lock(bundle_dir, monkeypatch):
+    layout(bundle_dir, {"evals/regression.toml": 'tasks = ["t"]\n'})
+
+    def refused(*_, **__):
+        raise AssertionError("a dry run wrote")
+
+    monkeypatch.setattr(Ledger, "record", refused)
+    monkeypatch.setattr(materialize_module, "materializing", refused)
+    monkeypatch.setattr(Task, "put", refused)
+    monkeypatch.setattr(Eval, "put", refused)
+
+    dry = _run(bundle_dir, dry_run=True)
+
+    assert {done.version for done in dry.writes} == {1}
+    assert list(state_root().rglob("*")) == []
+
+
+def test_a_dry_run_doesnt_preflight_a_step_reading_what_it_would_write_and_lists_it(bundle_dir, monkeypatch):
+    box, greet, _ = json.loads(LAYOUT["tasks/t.json"])
+    _steps(bundle_dir, [
+        box, greet, {"id": "check", "type": "checked_materialize_test", "artifact_id": "greeting"},
+        {"id": "write", "type": "writes_materialize_test", "artifact_id": "made"},
+        {"id": "read", "type": "checked_materialize_test", "artifact_id": "made", "depends_on": [{"task_step_id": "write"}]},
+    ])
+    monkeypatch.setattr(_Writes, "preflight", lambda self: [])
+    monkeypatch.setattr(_Checked, "problems", ["greeting isn't in the store yet"])
+    new = _run(bundle_dir, dry_run=True)
+    monkeypatch.setattr(_Checked, "problems", [])
+    assert _run(bundle_dir).not_preflighted == ()
+    (bundle_dir / "artifacts/greeting/hello.txt").write_text("hello again\n")
+    monkeypatch.setattr(_Checked, "problems", ["the store holds the greeting it is about to replace"])
+
+    rewritten = _run(bundle_dir, dry_run=True)
+
+    assert [(write.id, step.id) for write, step in new.not_preflighted] == [(f"{ROOT}/t", "check")]
+    assert [(write.id, step.id) for write, step in rewritten.not_preflighted] == [(f"{ROOT}/t", "check")]
+
+
 def test_a_second_run_of_the_bundle_waits_for_the_first_to_finish_writing(bundle_dir, tmp_path):
     held, done, waited = tmp_path / "held", tmp_path / "done", []
     holder = subprocess.Popen([sys.executable, "-c", textwrap.dedent(f"""
diff --git a/tst/unit/bundle/run_test.py b/tst/unit/bundle/run_test.py
index afa786b7..c75eeee1 100644
--- a/tst/unit/bundle/run_test.py
+++ b/tst/unit/bundle/run_test.py
@@ -10,9 +10,16 @@
 from click.testing import CliRunner
 
 import agent_env.bundle.run as run_module
-from agent_env.bundle import BundleError, Outcome, TaskRun, run_bundle
+from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
+from agent_env.artifact.artifacts.file import FileArtifact
+from agent_env.artifact.store import get_artifact_store
+from agent_env.bundle import BundleError, Outcome, TaskRun, dry_run_bundle, parse_bundle, run_bundle
+from agent_env.bundle.materialize import materialize
+from agent_env.bundle.plan import plan_bundle
+from agent_env.bundle.resolve import resolve_bundle
 from agent_env.cli import cli
 from agent_env.config.runtime import Config
+from agent_env.store import Filter
 from agent_env.store.routing import namespace_routing
 from agent_env.task import Task
 from agent_env.task_step.context import TaskStepContext
@@ -20,6 +27,7 @@
 from tst.unit.bundle._support import layout, local_store, plan_of
 
 ROOT = "@local/~/triage"
+DRY_RUN = "Dry run: nothing is written or run.\n"
 
 
 class _Scored(TaskStep):
@@ -344,3 +352,95 @@ def test_the_cli_prints_a_bundle_problem_as_the_answer_it_is(bundle_dir, quiet_l
 
     assert result.exit_code == 1
     assert result.output.startswith("Error: --task 'nope': this bundle has no task with that name or id")
+
+
+def _put_image(id):
+    get_artifact_store().put_document(DockerImageArtifact(
+        id=id, description=id, image_name="solver:v1", tar_gz_s3_url="file:///solver.tar.gz"))
+
+
+def test_a_dry_run_selects_what_a_run_would_and_runs_nothing(bundle_dir):
+    every = dry_run_bundle(bundle_dir)
+    selected = dry_run_bundle(bundle_dir, tasks=["unnamed", "a"], evals=["smoke"])
+
+    assert [entry.name for entry in every.runs] == ["a", "b", "c"]
+    assert [every.path(entry) for entry in every.skipped] == ["tasks/unnamed.json"]
+    assert ([entry.name for entry in selected.runs], selected.skipped) == (["a", "b", "unnamed"], ())
+    assert _Scored.seen == []
+    assert not local_store().path.exists()
+
+
+def test_a_dry_run_refuses_what_a_run_refuses_before_anything_is_written(bundle_dir):
+    with namespace_routing():
+        Task.put(id="shared-task", steps=[_Scored(id="check", version=None)])
+    layout(bundle_dir, {"evals/other.toml": 'tasks = ["shared-task"]\n'})
+
+    with pytest.raises(BundleError, match="evals/other.toml: tasks\\[0\\]: 'shared-task' is a store task"):
+        dry_run_bundle(bundle_dir)
+    with pytest.raises(ValueError, match="Unknown sandbox backend: 'nowhere'"):
+        dry_run_bundle(bundle_dir, evals=["smoke"], sandbox="nowhere")
+    with pytest.raises(BundleError, match="--task 'nope': this bundle has no task with that name or id"):
+        dry_run_bundle(bundle_dir, tasks=["nope"])
+    assert not local_store().path.exists()
+
+
+def test_the_cli_dry_run_prints_what_the_run_would_write_and_run_and_exits_0(bundle_dir, quiet_logs):
+    with namespace_routing():
+        FileArtifact.put_bytes("shared-notes", description="notes", filename="notes.txt", content=b"notes")
+    layout(bundle_dir, {
+        "artifacts/script/run.py": "print(1)\n",
+        "artifacts/script/lib.py": "x = 1\n",
+        "tasks/code.json": json.dumps([
+            {"id": "code", "type": "run_code", "script_artifact_id": "script", "script_file": "run.py"},
+            {"id": "notes", "type": "load_artifact", "sandbox_name": "box", "artifact_id": "shared-notes"},
+        ]),
+        "evals/smoke.toml": 'tasks = ["a", "b", "code"]\n',
+    })
+
+    result = CliRunner().invoke(cli, ["run", str(bundle_dir), "--dry-run"])
+
+    assert result.exit_code == 0, result.output
+    assert result.output == (
+        DRY_RUN
+        + "artifacts/script: v1 (new)\n"
+        "tasks/a.json: v1 (new)\n"
+        "tasks/b.json: v1 (new)\n"
+        "tasks/c.json: v1 (new)\n"
+        "tasks/code.json: v1 (new)\n"
+        "evals/full.toml: v1 (new)\n"
+        "evals/smoke.toml: v1 (new)\n"
+        "\n"
+        "Store refs:\n"
+        "  artifact shared-notes v1, the latest\n"
+        "Would run:\n"
+        "  tasks/a.json v1\n"
+        "  tasks/b.json v1\n"
+        "  tasks/c.json v1\n"
+        "  tasks/code.json v1\n"
+        "Evals:\n"
+        "  evals/full.toml v1: tasks/a.json, tasks/c.json\n"
+        "  evals/smoke.toml v1: tasks/a.json, tasks/b.json, tasks/code.json\n"
+        "tasks/unnamed.json isn't named by any eval, so it wouldn't run; run it with --task unnamed\n"
+        "Not preflighted, since each reads what the run would write first:\n"
+        "  tasks/code.json: step 'code' (run_code)\n"
+        + DRY_RUN
+    )
+    assert not local_store().path.exists()
+    assert _Scored.seen == []
+
+
+def test_the_cli_dry_run_prints_a_problem_as_one_line_and_exits_1(bundle_dir, quiet_logs):
+    result = CliRunner().invoke(cli, ["run", str(bundle_dir), "--task", "nope", "--dry-run"])
+
+    assert result.exit_code == 1
+    assert result.output.startswith(f"{DRY_RUN}Error: --task 'nope': this bundle has no task with that name or id")
+    assert "Traceback" not in result.output
+
+
+def test_the_cli_dry_run_accepts_keep_and_model_and_ignores_them(bundle_dir, quiet_logs):
+    plain = CliRunner().invoke(cli, ["run", str(bundle_dir), "--eval", "smoke", "--dry-run"])
+    ignoring = CliRunner().invoke(cli, ["run", str(bundle_dir), "--eval", "smoke", "--dry-run", "--keep", "--model", "m"])
+
+    assert (plain.exit_code, ignoring.exit_code) == (0, 0)
+    assert ignoring.output == plain.output
+    assert _Scored.seen == []

From 993bfc0686fe3ad5e73d35cb4313453fbae0bc83 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 30 Sep 2026 09:58:14 -1000
Subject: [PATCH 3/7] feat(bundle): name another bundle's entities by their
 @local id

An explicit @local/... id outside the bundle is checked against the @local namespace's store like any
store id. An agent is written with the version the plan read, and again once that bundle writes a new
one; tasks and evals stay bare. An id under the bundle's own root that it doesn't write is still refused
without a store read. Planning such a read with namespace routing off raises.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 README.md                           |  7 +++--
 src/agent_env/bundle/plan.py        | 48 +++++++++++++++++------------
 tst/unit/bundle/ledger_test.py      | 17 ++++++++++
 tst/unit/bundle/materialize_test.py | 23 ++++++++++++++
 tst/unit/bundle/plan_test.py        | 42 ++++++++++++++++++++++---
 tst/unit/bundle/run_test.py         | 39 +++++++++++++++++++++++
 6 files changed, 151 insertions(+), 25 deletions(-)

diff --git a/README.md b/README.md
index 5b56bbca..d7b5a345 100644
--- a/README.md
+++ b/README.md
@@ -910,6 +910,8 @@ A bundle is a folder holding tasks and what they need, which `agent-env run` wri
 - `tasks/<name>.json`, a list of steps that refer to the bundle's entities by name;
 - `evals/<name>.toml`, with `tasks = [...]` naming the bundle's tasks.
 
+Anything else is named by its store id. Another bundle's entity is named by its `@local/…` id, such as `@local/~/other/greeting` for the `greeting` of a bundle at `~/other`, and read from the local store that bundle's runs write, like any store id.
+
 An `agent.toml` takes `image`, a store id or `{ artifact = "<id>", version = <n> }` to pin one, and optionally `default_env_vars` (string values) and a `[metadata]` table of `default_model` and `min_disk_size_gb`. The agent's card isn't authored: the image serves it when the agent deploys.
 
 ```toml
@@ -970,9 +972,10 @@ A rerun writes a new version only of what changed since the bundle last wrote it
   - Ctrl-C or SIGTERM during the runs cancels them: each run that started is marked `cancelled` in the store and torn down, a run that was waiting shows `didn't start`, the summary prints, and the command exits 130 (143 for SIGTERM). A second Ctrl-C stops the teardown and prints what is still up. From Python, `run_bundle()` raises `RunInterrupted`, a `KeyboardInterrupt` whose `result` holds the runs.
 
 Known limits:
-- An eval in a bundle runs only the bundle's own tasks. One that names a store task is refused; run that task with `agent-env task run`.
+- An eval in a bundle runs only the bundle's own tasks. One that names a store task, another bundle's included, is refused; run that task with `agent-env task run`.
 - A step that writes an entity under an id it makes up is refused when it runs, since that id isn't under `@local/`. This refuses `run_container_unit_tests_verifier`'s stdout artifacts and `collect_artifacts`, and `snapshot_env` unless the task names its `snapshot_id`. `verify_sandbox` writes nothing, so it runs.
-- A step's reference to one of the bundle's entities names no version, so it reads the latest version when the step runs. Another run of the same folder, made after an edit, can write a newer one first.
+- A step's reference to one of the bundle's entities names no version, so it reads the latest version when the step runs. Another run of the same folder, made after an edit, can write a newer one first. A step's reference to another bundle's entity reads its latest when the step runs too, while an agent is written with the version the run read, and is written again once that bundle writes a new one.
+- An `@local/…` id under the bundle's own root that the bundle doesn't write is refused without a store read, since it is most likely left over from a rename. So a bundle in a subfolder of another can't be named from it.
 - A dry run's versions and reasons hold for the stores as they are when it runs. It takes no lock, so another run writing the same ids can change them first, and it doesn't see what another bundle's pending run would write.
 - Teardown reaches only the sandboxes a run's context records. A `rubrics_verifier`'s judge, and an agent whose deploy was cancelled, terminate their own sandboxes but leave their local work folders. A run killed outright (`kill -9`) leaves everything it deployed.
 - A process a local sandbox's command detaches from itself (a double fork, `setsid`) outlives the command and the run.
diff --git a/src/agent_env/bundle/plan.py b/src/agent_env/bundle/plan.py
index 38ad8505..b9ba3358 100644
--- a/src/agent_env/bundle/plan.py
+++ b/src/agent_env/bundle/plan.py
@@ -22,6 +22,7 @@
 from agent_env.env.env import Env
 from agent_env.store.base import NotFoundError
 from agent_env.store.ids import LOCAL_PREFIX
+from agent_env.store.routing import namespace_routing_enabled
 from agent_env.task import Task
 from agent_env.task_step.registry import get_task_step_registry
 
@@ -197,8 +198,7 @@ def _closure(self, roots: list[ResolvedEntry]) -> set[_Key]:
     def _check_store_refs(
         self, closure: set[_Key], *, stores: bool = True
     ) -> tuple[tuple[Reference, ...], dict[tuple[EntityKind, str], int]]:
-        """Check each store id the closure references; without ``stores``, only the ``@local`` ones, which need no
-        store read."""
+        """Check each store id the closure references; without ``stores``, only what needs no store read."""
         read: dict[tuple[EntityKind, str, int | None], Any] = {}
         checked: dict[tuple[EntityKind, str, int | None], Reference] = {}
         latest: dict[tuple[EntityKind, str], int] = {}
@@ -209,31 +209,41 @@ def _check_store_refs(
             for ref in source.references:
                 if ref.local is not None:
                     continue
-                if not stores and not ref.id.startswith(LOCAL_PREFIX):
-                    continue
-                problem = self._store_ref_problem(ref, read)
+                problem = self._bundle_ref_problem(ref)
+                if problem is None and stores:
+                    problem = self._store_ref_problem(ref, read)
                 if problem:
                     self.problems.append(f"{self._path(source.entry)}: {ref.where}: {problem}")
-                else:
+                elif stores:
                     checked.setdefault((ref.kind, ref.id, ref.version), ref)
                     if ref.version is None:
                         latest[ref.kind, ref.id] = read[ref.kind, ref.id, None].version
         return tuple(checked.values()), latest
 
-    def _store_ref_problem(self, ref: Reference, read: dict) -> str | None:
+    def _bundle_ref_problem(self, ref: Reference) -> str | None:
+        """What makes an ``@local`` id wrong without reading a store: it names one of this bundle's outputs by id,
+        one of its entities of another kind, or an id under its root that it doesn't write. Another bundle's id is
+        read like any store id."""
+        if not ref.id.startswith(LOCAL_PREFIX):
+            return None
         kind, key = ref.kind.value, _nfc(ref.id)
-        if ref.id.startswith(LOCAL_PREFIX):
-            if (kind, key) in self.outputs:
-                entry, output = self.outputs[kind, key]
-                return (f"{ref.id!r} is written by step {output.step_id!r} of {self._path(entry)}; refer to an output "
-                        "by its name, in the task that writes it")
-            other = self.identified.get(key)
-            if other is not None:
-                return (f"{ref.id!r} is this bundle's {other.kind.store} {other.name!r}, but this field takes "
-                        f"{with_article(kind)}")
-            if key.startswith(f"{self.resolved.bundle.id_root}/"):
-                return f"{ref.id!r} isn't in this bundle"
-            return f"{ref.id!r} isn't in this bundle, and references to another bundle's entities aren't supported yet"
+        if (kind, key) in self.outputs:
+            entry, output = self.outputs[kind, key]
+            return (f"{ref.id!r} is written by step {output.step_id!r} of {self._path(entry)}; refer to an output "
+                    "by its name, in the task that writes it")
+        other = self.identified.get(key)
+        if other is not None:
+            return (f"{ref.id!r} is this bundle's {other.kind.store} {other.name!r}, but this field takes "
+                    f"{with_article(kind)}")
+        if key.startswith(f"{self.resolved.bundle.id_root}/"):
+            return f"{ref.id!r} isn't in this bundle"
+        return None
+
+    def _store_ref_problem(self, ref: Reference, read: dict) -> str | None:
+        if ref.id.startswith(LOCAL_PREFIX) and not namespace_routing_enabled():
+            raise RuntimeError("reading an @local id from the store needs namespace routing, which the agent-env CLI "
+                               "turns on; call plan_bundle inside agent_env.store.routing.namespace_routing()")
+        kind = ref.kind.value
         lookup = (ref.kind, ref.id, ref.version)
         if lookup not in read:
             try:
diff --git a/tst/unit/bundle/ledger_test.py b/tst/unit/bundle/ledger_test.py
index 8b3cb062..48e6379d 100644
--- a/tst/unit/bundle/ledger_test.py
+++ b/tst/unit/bundle/ledger_test.py
@@ -342,6 +342,23 @@ def test_an_env_naming_a_store_env_without_a_version_is_rewritten_when_the_store
     assert checks[f"{ROOT}/t"].unchanged
 
 
+def test_an_env_naming_another_bundles_env_without_a_version_is_rewritten_when_that_one_gets_a_new_one(
+        bundle_dir, monkeypatch, cli_routing):
+    monkeypatch.setattr("agent_env.env.registry.get_env_registry", lambda: {_NoteEnv.type: _NoteEnv})
+    rocket = "@local/~/other/rocket"
+    local_store().insert("envs", {"id": rocket, "version": 1, "type": _NoteEnv.type})
+    (bundle_dir / "envs/rack").mkdir()
+    (bundle_dir / "envs/rack/env.toml").write_text(f'type = "rack_ledger_test"\nenvs = ["{rocket}"]\n')
+    (bundle_dir / "tasks/t.json").write_text(_steps({"id": "rack", "type": "deploy_env", "env_id": "rack"}))
+    _run(bundle_dir)
+    local_store().insert("envs", {"id": rocket, "version": 2, "type": _NoteEnv.type})
+
+    checks = _run(bundle_dir)
+
+    assert checks[f"{ROOT}/rack"].reasons == (f"env {rocket} has a new version in the store (v1 → v2)",)
+    assert checks[f"{ROOT}/t"].unchanged
+
+
 def test_an_agent_is_rewritten_when_its_toml_changes_or_its_unpinned_store_image_does(bundle_dir):
     registry = get_config().get_document_store()
     registry.insert("artifacts", _image_document(1))
diff --git a/tst/unit/bundle/materialize_test.py b/tst/unit/bundle/materialize_test.py
index a9f8acf5..67a71124 100644
--- a/tst/unit/bundle/materialize_test.py
+++ b/tst/unit/bundle/materialize_test.py
@@ -287,6 +287,29 @@ def test_an_unpinned_store_image_is_written_at_the_version_the_plan_checked(bund
     assert A2AAgent.get(f"{ROOT}/solver").docker_image_artifact.version == 1
 
 
+def test_another_bundles_entities_are_read_like_store_ids_and_an_agent_pins_the_version_it_read(bundle_dir):
+    other = layout(bundle_dir.parent / "other", LAYOUT)
+    _run(other)
+    image = "@local/~/other/solver-image"
+    _put_image(image)
+    box, greet, _ = json.loads(LAYOUT["tasks/t.json"])
+    layout(bundle_dir, {"agents/solver/agent.toml": f'image = "{image}"\n'})
+    _steps(bundle_dir, [box, {**greet, "artifact_id": "@local/~/other/greeting"},
+                        {"id": "agent", "type": "deploy_agent", "env_ids": [], "a2a_agent_id": "solver"}])
+
+    assert _summary(_run(bundle_dir)) == {f"{ROOT}/solver": (1, False, ("new",)), f"{ROOT}/t": (1, False, ("new",))}
+    assert Task.get(f"{ROOT}/t").steps[1].artifacts == [{"id": "@local/~/other/greeting", "version": None}]
+    _put_image(image)
+    (other / "artifacts/greeting/hello.txt").write_text("hello again\n")
+    _run(other)
+
+    assert _summary(_run(bundle_dir)) == {
+        f"{ROOT}/solver": (2, False, (f"artifact {image} has a new version in the store (v1 → v2)",)),
+        f"{ROOT}/t": (1, True, ()),
+    }
+    assert A2AAgent.get(f"{ROOT}/solver").docker_image_artifact.version == 2
+
+
 def _put_image(id):
     get_artifact_store().put_document(DockerImageArtifact(
         id=id, description=id, image_name=f"{id}:v1", tar_gz_s3_url=f"file:///{id}.tar.gz"))
diff --git a/tst/unit/bundle/plan_test.py b/tst/unit/bundle/plan_test.py
index bbb22f64..4035e9d7 100644
--- a/tst/unit/bundle/plan_test.py
+++ b/tst/unit/bundle/plan_test.py
@@ -14,12 +14,14 @@
 from agent_env.bundle import resolve as resolve_module
 from agent_env.bundle.plan import check_bundle, plan_bundle
 from agent_env.bundle.resolve import resolve_bundle
+from agent_env.config import configure
 from agent_env.entity_refs import EntityKind, EntityRef
 from agent_env.env.env import Env
 from agent_env.env.store import get_env_store, reset_env_store
 from agent_env.store import reset_config, set_document_store
 from agent_env.task import Task
 from agent_env.task.store import get_task_store, reset_task_store
+from tst.unit.bundle._support import RefusingStore
 from tst.unit.store.fakes import FakeDocumentStore
 
 ROOT = "@local/~/triage"
@@ -253,8 +255,6 @@ def test_store_ids_the_selection_reaches_are_read_once_and_kept(make, monkeypatc
      "'tickets' has that name"),
     ([load(f"{ROOT}/tickets")], f"step 'load': artifact_id: '{ROOT}/tickets' is this bundle's env 'tickets', but "
      "this field takes an artifact"),
-    ([deploy("@local/~/other/crm")], "step 'deploy': env_id: '@local/~/other/crm' isn't in this bundle, and "
-     "references to another bundle's entities aren't supported yet"),
     ([deploy(f"{ROOT}/tickts")], f"step 'deploy': env_id: '{ROOT}/tickts' isn't in this bundle"),
 ])
 def test_a_store_id_must_exist(make, steps, problem):
@@ -262,6 +262,42 @@ def test_a_store_id_must_exist(make, steps, problem):
     assert problems(make(tasks={"t": steps})) == (f"tasks/t.json: {problem}",)
 
 
+def test_another_bundles_local_id_is_read_from_the_local_namespace_like_a_store_id(make, local_stores, cli_routing):
+    put_env("@local/~/other/crm")
+    VMImageArtifact.put(id="@local/~/other/golden", description="golden", ecr_url="ecr/x")
+    configure(document_store=RefusingStore())
+
+    plan = plan_bundle(make(tasks={"t": [deploy("@local/~/other/crm")]}))
+
+    assert [(ref.kind, ref.id, ref.version) for ref in plan.store_refs] == [(EntityKind.ENV, "@local/~/other/crm", None)]
+    assert plan.store_latest == {(EntityKind.ENV, "@local/~/other/crm"): 1}
+    assert problems(make(tasks={"t": [deploy("@local/~/other/gone")]})) == (
+        "tasks/t.json: step 'deploy': env_id: there is no env '@local/~/other/gone' in the store",
+    )
+    assert problems(make(tasks={"t": [deploy("both")]}, files={
+        "envs/both/env.toml": composite(image="@local/~/other/golden")})) == (
+        "envs/both: image: '@local/~/other/golden' is a vm_image in the store, but this field takes docker_image",
+    )
+
+
+def test_an_id_under_the_bundles_own_root_that_it_doesnt_write_is_refused_without_a_read(
+        make, local_stores, cli_routing, monkeypatch):
+    put_env(f"{ROOT}/crm")
+
+    def read(*_):
+        raise AssertionError("the plan read a store")
+
+    monkeypatch.setattr(plan_module, "_GETTERS", dict.fromkeys(plan_module._GETTERS, read))
+    assert problems(make(tasks={"t": [deploy(f"{ROOT}/crm")]})) == (
+        f"tasks/t.json: step 'deploy': env_id: '{ROOT}/crm' isn't in this bundle",
+    )
+
+
+def test_planning_a_read_of_another_bundles_local_id_without_namespace_routing_is_refused(make):
+    with pytest.raises(RuntimeError, match="reading an @local id from the store needs namespace routing"):
+        plan_bundle(make(tasks={"t": [deploy("@local/~/other/crm")]}))
+
+
 def test_an_output_is_only_readable_in_its_own_task(make):
     resolved = make(tasks={"t": [snapshot("snap")], "u": [load(f"{ROOT}/t/snap")]})
     assert problems(resolved, tasks=["u"]) == (
@@ -317,8 +353,6 @@ def read(*_):
     assert sorted(caught.value.problems) == [
         "artifacts/note: artifact.toml: unknown key 'descripton'; a file artifact takes description, type and id",
         "tasks/u.json: Duplicate step id 'd' at position 1; step ids must be unique within a task",
-        "tasks/v.json: step 'deploy': env_id: '@local/~/other/crm' isn't in this bundle, and references to another "
-        "bundle's entities aren't supported yet",
     ]
 
 
diff --git a/tst/unit/bundle/run_test.py b/tst/unit/bundle/run_test.py
index c75eeee1..64d2ae52 100644
--- a/tst/unit/bundle/run_test.py
+++ b/tst/unit/bundle/run_test.py
@@ -213,6 +213,20 @@ def test_an_eval_naming_a_store_task_is_refused_before_anything_is_written(bundl
     assert _Scored.seen == []
 
 
+def test_an_eval_naming_another_bundles_task_is_refused_as_a_store_task(bundle_dir):
+    run_bundle(layout(bundle_dir.parent / "other", {"tasks/shared.json": _scored(1.0)}))
+    layout(bundle_dir, {"evals/other.toml": 'tasks = ["@local/~/other/shared"]\n'})
+
+    with pytest.raises(BundleError) as caught:
+        run_bundle(bundle_dir, evals=["other"])
+
+    assert caught.value.problems == (
+        "evals/other.toml: tasks[0]: '@local/~/other/shared' is a store task, and a bundle's evals run only the "
+        "bundle's own tasks for now; run it on its own with agent-env task run --id '@local/~/other/shared'",
+    )
+    assert local_store().find_one("evals", Filter.of(id=f"{ROOT}/other")) is None
+
+
 def test_an_eval_that_isnt_selected_isnt_refused(bundle_dir):
     with namespace_routing():
         Task.put(id="shared-task", steps=[_Scored(id="check", version=None)])
@@ -429,6 +443,31 @@ def test_the_cli_dry_run_prints_what_the_run_would_write_and_run_and_exits_0(bun
     assert _Scored.seen == []
 
 
+def test_the_cli_dry_run_reads_another_bundles_entity_and_says_why_an_agent_pinning_it_would_be_rewritten(
+        bundle_dir, quiet_logs):
+    image = "@local/~/other/solver-image"
+    layout(bundle_dir, {"agents/solver/agent.toml": f'image = "{image}"\n', "tasks/agent.json": json.dumps(
+        [{"id": "agent", "type": "deploy_agent", "env_ids": [], "a2a_agent_id": "solver"}])})
+    with namespace_routing():
+        _put_image(image)
+        materialize(plan_bundle(resolve_bundle(parse_bundle(bundle_dir)), tasks=["agent"]))
+        _put_image(image)
+
+    result = CliRunner().invoke(cli, ["run", str(bundle_dir), "--task", "agent", "--dry-run"])
+
+    assert result.exit_code == 0, result.output
+    assert result.output.startswith(
+        DRY_RUN
+        + f"agents/solver: v2 (artifact {image} has a new version in the store (v1 → v2))\n"
+        "tasks/agent.json: v1, unchanged\n"
+        "\n"
+        "Store refs:\n"
+        f"  artifact {image} v2, the latest\n"
+        "Would run:\n"
+        "  tasks/agent.json v1\n"
+    )
+
+
 def test_the_cli_dry_run_prints_a_problem_as_one_line_and_exits_1(bundle_dir, quiet_logs):
     result = CliRunner().invoke(cli, ["run", str(bundle_dir), "--task", "nope", "--dry-run"])
 

From 89f4957a49688354d93340d537941dd685d8c58a Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 30 Sep 2026 11:30:41 -1000
Subject: [PATCH 4/7] fix(bundle): a dry run matches what it would write by
 store and id

An artifact and an agent can share an id, so a dry run rewriting the agent skipped the preflight of a step
over the reused artifact, and exited 0 where the run then stopped. It now compares (store, id); a step
reading its own task's output is still matched by id, so the real run is unchanged.

- Deriving a step's output id reads the step without its depends_on, so a malformed one is reported once.
- dry_run_bundle's docstring says a failed preflight raises and that routing is on process-wide while it
  runs; run --dry-run's help says it takes no lock.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/agent_env/bundle/materialize.py | 18 ++++++++++--------
 src/agent_env/bundle/resolve.py     |  2 +-
 src/agent_env/bundle/run.py         |  5 +++--
 src/agent_env/cli/run.py            |  3 ++-
 tst/unit/bundle/materialize_test.py | 13 +++++++++++++
 tst/unit/bundle/resolve_test.py     | 10 ++++++++++
 6 files changed, 39 insertions(+), 12 deletions(-)

diff --git a/src/agent_env/bundle/materialize.py b/src/agent_env/bundle/materialize.py
index e6da717c..9198aca8 100644
--- a/src/agent_env/bundle/materialize.py
+++ b/src/agent_env/bundle/materialize.py
@@ -95,7 +95,7 @@ def through_ledger(write: Write, write_fn: Callable[[], int]) -> None:
     with nullcontext() if dry_run else materializing(plan.bundle.bundle, on_wait):
         for write in entities:
             through_ledger(write, lambda: _WRITERS[write.kind](plan, write))
-        unwritten = {item.write.id for item in done.values() if not item.reused} if dry_run else set()
+        unwritten = {_key(item.write) for item in done.values() if not item.reused} if dry_run else set()
         built, problems, unchecked = {}, [], []
         for write in tasks:
             with _noted(plan, write, "preflighting"):
@@ -201,16 +201,17 @@ def _build(write: Write) -> Task:
     return Task(id=write.id, version=None, steps=[build_step(step) for step in write.source.config])
 
 
-def _preflight(write: Write, task: Task, unwritten: set[str]) -> tuple[list[str], list[TaskStep]]:
-    """``task``'s preflight problems, and the steps with a preflight left unchecked because they read an id in
-    ``unwritten``, which a dry run would write first. A step reading one of the task's own outputs is skipped
+def _preflight(write: Write, task: Task, unwritten: set[tuple[str, str]]) -> tuple[list[str], list[TaskStep]]:
+    """``task``'s preflight problems, and the steps with a preflight left unchecked because they read a
+    (store, id) in ``unwritten``, which a dry run would write first; an artifact and an agent can share an id. A step reading one of the task's own outputs is skipped
     too, and not listed: the output only exists once the task runs."""
     outputs = {output.id for output in write.source.outputs}
     problems, unchecked = [], []
     for config, step in zip(write.source.config, task.steps):
-        if _reads_any(config, outputs):
+        reads = _reads(config)
+        if any(id in outputs for _, id in reads):
             continue
-        if _reads_any(config, unwritten):
+        if reads & unwritten:
             if type(step).preflight is not TaskStep.preflight:
                 unchecked.append(step)
             continue
@@ -218,9 +219,10 @@ def _preflight(write: Write, task: Task, unwritten: set[str]) -> tuple[list[str]
     return problems, unchecked
 
 
-def _reads_any(step: dict, ids: set[str]) -> bool:
+def _reads(step: dict) -> set[tuple[str, str]]:
+    """The (store, id) of each entity ``step`` reads."""
     refs = get_task_step_registry()[step["type"]].entity_refs or ()
-    return any(site.value in ids for site in ref_sites(refs, step) if site.ref.role is RefRole.INPUT)
+    return {(site.ref.kind.value, site.value) for site in ref_sites(refs, step) if site.ref.role is RefRole.INPUT}
 
 
 def _key(write: Write) -> tuple[str, str]:
diff --git a/src/agent_env/bundle/resolve.py b/src/agent_env/bundle/resolve.py
index baff8dd3..5b92a98a 100644
--- a/src/agent_env/bundle/resolve.py
+++ b/src/agent_env/bundle/resolve.py
@@ -244,7 +244,7 @@ def _derived_outputs(self, entry: BundleEntry, step: dict, cls: type) -> bool:
         if not absent:
             return True
         try:
-            read = build_step(step).to_dict()
+            read = build_step({key: value for key, value in step.items() if key != "depends_on"}).to_dict()
         except Exception as e:
             self._problem(entry, f"step {step.get('id')!r}: {cls.type} can't read it ({type(e).__name__}: {e})")
             return False
diff --git a/src/agent_env/bundle/run.py b/src/agent_env/bundle/run.py
index 01dd1888..1f533748 100644
--- a/src/agent_env/bundle/run.py
+++ b/src/agent_env/bundle/run.py
@@ -234,8 +234,9 @@ def dry_run_bundle(
     the one it would reuse or the store's next, which another run writing the id first can take. ``on_progress``
     is given the line ``run_bundle`` gives for each write, as it is checked.
 
-    It raises what ``run_bundle`` raises before anything is written, except that it runs no event loop, so a
-    running one isn't refused."""
+    It raises what ``run_bundle`` raises before its first task runs, a failed preflight included. Like
+    ``run_bundle``, it turns namespace routing on for the whole process while it runs, so it suits the CLI and
+    scripts, not a service; it runs no event loop, so a running one isn't refused."""
     if sandbox:
         build_sandbox_provider(sandbox)
     say = _progress(on_progress)
diff --git a/src/agent_env/cli/run.py b/src/agent_env/cli/run.py
index 0a0673de..d43d636e 100644
--- a/src/agent_env/cli/run.py
+++ b/src/agent_env/cli/run.py
@@ -53,7 +53,8 @@ def run(ctx: click.Context, bundle: str | None, tasks: tuple[str, ...], evals: t
 
     --dry-run makes the checks the run makes before its first task, reading the stores as the run does, and
     prints what it would write and run, writing and running nothing. It exits 1 on a problem the run would
-    stop at, and 0 otherwise; --model and --keep change nothing it shows.
+    stop at, and 0 otherwise; --model and --keep change nothing it shows. It takes no lock, so another run
+    can write first and change what it shows.
 
     Each run's sandboxes are torn down as it ends; its instance and outputs stay. --keep holds them up
     until Ctrl-C instead. Ctrl-C or SIGTERM mid-run cancels the runs, tears them down, prints what ran and
diff --git a/tst/unit/bundle/materialize_test.py b/tst/unit/bundle/materialize_test.py
index 67a71124..9480c042 100644
--- a/tst/unit/bundle/materialize_test.py
+++ b/tst/unit/bundle/materialize_test.py
@@ -583,3 +583,16 @@ def test_a_second_run_of_the_bundle_waits_for_the_first_to_finish_writing(bundle
 def test_version_of_names_what_isnt_a_write(bundle_dir):
     with pytest.raises(KeyError, match="isn't one of this plan's writes"):
         _run(bundle_dir).version_of("env", f"{ROOT}/nothing")
+
+
+def test_a_dry_run_preflights_a_step_over_a_reused_artifact_named_like_a_rewritten_agent(bundle_dir, monkeypatch):
+    _put_image("solver-image")
+    layout(bundle_dir, {"agents/greeting/agent.toml": 'image = "solver-image"\n'})
+    _steps(bundle_dir, [{"id": "check", "type": "checked_materialize_test", "artifact_id": "greeting"},
+                        {"id": "agent", "type": "deploy_agent", "env_ids": [], "a2a_agent_id": "greeting"}])
+    _run(bundle_dir)
+    (bundle_dir / "agents/greeting/agent.toml").write_text('image = "solver-image"\n[default_env_vars]\nA = "1"\n')
+    monkeypatch.setattr(_Checked, "problems", ["the script is gone"])
+
+    assert _problems(lambda: _run(bundle_dir, dry_run=True)) == ("tasks/t.json: the script is gone",)
+    assert _problems(lambda: _run(bundle_dir)) == ("tasks/t.json: the script is gone",)
diff --git a/tst/unit/bundle/resolve_test.py b/tst/unit/bundle/resolve_test.py
index 6385498d..6e5721ee 100644
--- a/tst/unit/bundle/resolve_test.py
+++ b/tst/unit/bundle/resolve_test.py
@@ -430,3 +430,13 @@ def test_entities_without_references_resolve_to_a_copy_of_their_config(make):
         (BundleKind.ARTIFACT, "base-mcp"): (), (BundleKind.SKILL, "pdf"): (),
     }
     assert [image.entry.kind for image in result.built_images] == [BundleKind.AGENT]
+
+
+@pytest.mark.parametrize(("depends_on", "problem"), [
+    ("tickets", "depends_on is a list of step ids, not 'tickets'"),
+    ([7], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not 7'),
+])
+def test_a_bad_depends_on_on_a_step_deriving_its_output_is_named_once(make, depends_on, problem):
+    assert problems(make(tasks={"t": [
+        {"id": "cli", "type": "build_mcp_cli", "env_id": "tickets", "command_name": "c", "depends_on": depends_on},
+    ]})) == (f"tasks/t.json: step 'cli': {problem}",)

From 555f7f7e0872d303b8710a0ac76dc89d56a00d43 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 30 Sep 2026 11:41:43 -1000
Subject: [PATCH 5/7] fix(task): read the depends_on object form as before

{"task_step_id": <id>} is read as it always was, whatever the id holds, so every stored task main loads
still loads; only the new bare form must be a string. A lone string and any other shape are still refused.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/agent_env/task_step/task_step.py | 9 +++++----
 tst/unit/task/depends_on_test.py     | 6 ++++--
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/src/agent_env/task_step/task_step.py b/src/agent_env/task_step/task_step.py
index f591e25b..5dae12e8 100644
--- a/src/agent_env/task_step/task_step.py
+++ b/src/agent_env/task_step/task_step.py
@@ -21,10 +21,11 @@ class TaskStepDependency:
     @classmethod
     def from_dict(cls, data: str | dict[str, Any]) -> "TaskStepDependency":
         """One ``depends_on`` entry: a step id, or ``{"task_step_id": <step id>}``."""
-        step_id = data.get("task_step_id") if isinstance(data, dict) else data
-        if not isinstance(step_id, str):
-            raise ValueError(f'a depends_on entry is a step id or {{"task_step_id": "<step id>"}}, not {data!r}')
-        return cls(task_step_id=step_id)
+        if isinstance(data, str):
+            return cls(task_step_id=data)
+        if isinstance(data, dict) and "task_step_id" in data:
+            return cls(task_step_id=data["task_step_id"])
+        raise ValueError(f'a depends_on entry is a step id or {{"task_step_id": "<step id>"}}, not {data!r}')
 
 
 def dependencies(raw: Any) -> list[TaskStepDependency] | None:
diff --git a/tst/unit/task/depends_on_test.py b/tst/unit/task/depends_on_test.py
index bad4cff1..2e58178c 100644
--- a/tst/unit/task/depends_on_test.py
+++ b/tst/unit/task/depends_on_test.py
@@ -56,8 +56,6 @@ def test_an_entry_is_a_step_id_or_its_object_form(entry):
     ({"task_step_id": "a"}, "depends_on is a list of step ids, not {'task_step_id': 'a'}"),
     ([7], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not 7'),
     ([{"id": "a"}], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not {\'id\': \'a\'}'),
-    ([{"task_step_id": 7}], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not '
-     "{'task_step_id': 7}"),
 ])
 def test_anything_else_is_refused_saying_what_it_takes(depends_on, problem):
     with pytest.raises(ValueError, match=re.escape(problem)):
@@ -66,6 +64,10 @@ def test_anything_else_is_refused_saying_what_it_takes(depends_on, problem):
         _Step.from_dict({"id": "b", "depends_on": depends_on})
 
 
+def test_the_object_form_reads_whatever_id_it_holds_as_it_always_has():
+    assert _Step.from_dict({"id": "b", "depends_on": [{"task_step_id": 7}]}).depends_on == [TaskStepDependency(7)]
+
+
 def test_a_step_id_is_stored_in_the_object_form_and_reads_back(local_stores):
     steps = [_step("a"), _Step.from_dict({"id": "b", "depends_on": ["a"]}), _step("c", ["a", {"task_step_id": "b"}])]
     get_task_store().put_document(Task(id="t", version=None, steps=steps))

From 67a957150629437785ff3476e0b06849f253b239 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 30 Sep 2026 11:50:16 -1000
Subject: [PATCH 6/7] fix(bundle): a depends_on naming a step by anything but
 its id is a problem

The object form reads any id, as before, so a bundle task could reach the resolver with
{"task_step_id": ["load"]} and crash on the lookup. The resolver now reports such an entry and
leaves it out of the step's edges.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/agent_env/bundle/resolve.py | 19 +++++++++++++------
 tst/unit/bundle/resolve_test.py |  1 +
 2 files changed, 14 insertions(+), 6 deletions(-)

diff --git a/src/agent_env/bundle/resolve.py b/src/agent_env/bundle/resolve.py
index 5b92a98a..32fea60c 100644
--- a/src/agent_env/bundle/resolve.py
+++ b/src/agent_env/bundle/resolve.py
@@ -285,8 +285,9 @@ def _outputs(self, entry: BundleEntry, declared: list[tuple[int, dict, type]]) -
 
     def _depends_on(self, entry: BundleEntry, steps: list[dict]) -> list[list[str] | None]:
         """Each step's ``depends_on`` as the step ids it names, read as building the step reads it; None when it
-        has none. One that can't be read, or names no step in the task, is a problem here: a step reading an
-        earlier step's output is checked against these edges before any step is built."""
+        has none. One that can't be read, names a step by anything but its id, or names no step in the task, is a
+        problem here: a step reading an earlier step's output is checked against these edges before any step is
+        built."""
         step_ids = {step.get("id") for step in steps}
         depends = []
         for step in steps:
@@ -295,10 +296,16 @@ def _depends_on(self, entry: BundleEntry, steps: list[dict]) -> list[list[str] |
             except ValueError as e:
                 self._problem(entry, f"step {step.get('id')!r}: {e}")
                 read = []
-            ids = None if read is None else [dep.task_step_id for dep in read]
-            for dep_id in ids or ():
-                if dep_id not in step_ids:
-                    self._problem(entry, f"step {step.get('id')!r}: depends_on {dep_id!r} names no step in this task")
+            ids = None if read is None else []
+            for dep in read or ():
+                if not isinstance(dep.task_step_id, str):
+                    self._problem(entry, f"step {step.get('id')!r}: depends_on names a step by its id, not "
+                                         f"{dep.task_step_id!r}")
+                    continue
+                if dep.task_step_id not in step_ids:
+                    self._problem(entry, f"step {step.get('id')!r}: depends_on {dep.task_step_id!r} names no step in "
+                                         "this task")
+                ids.append(dep.task_step_id)
             depends.append(ids)
         return depends
 
diff --git a/tst/unit/bundle/resolve_test.py b/tst/unit/bundle/resolve_test.py
index 6e5721ee..a05aa8d4 100644
--- a/tst/unit/bundle/resolve_test.py
+++ b/tst/unit/bundle/resolve_test.py
@@ -249,6 +249,7 @@ def test_an_output_resolves_for_a_step_that_depends_on_its_writer(make, depends_
     ([7], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not 7'),
     ([{"id": "cli"}], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not {\'id\': \'cli\'}'),
     (["clii"], "depends_on 'clii' names no step in this task"),
+    ([{"task_step_id": ["cli"]}], "depends_on names a step by its id, not ['cli']"),
 ])
 def test_a_depends_on_that_cant_be_read_is_named_beside_the_output_it_then_cant_reach(make, depends_on, problem):
     assert set(problems(make(tasks={"t": [

From 868be5ccb9493df17e82b3600b71bfbb734406ab Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 30 Sep 2026 12:25:20 -1000
Subject: [PATCH 7/7] refactor(bundle): simpler ledger walk and dependency
 checks

- materialize checks each write in its own closure and gives the ledger only the versions of what the
  write needs, so Ledger.check(write, needs) matches digest(write, needs) and no versions map shadows done.
- The resolver keeps a dependency id as an edge only when it names a step, with one message otherwise, so
  _upstream no longer filters.
- _validate_dag raises one error with the clause that applies, and _preflight's docstring is rewrapped.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/agent_env/bundle/ledger.py      |  9 ++++----
 src/agent_env/bundle/materialize.py | 32 ++++++++++++-----------------
 src/agent_env/bundle/resolve.py     | 16 ++++++---------
 src/agent_env/task/task.py          | 15 ++++----------
 tst/unit/bundle/ledger_test.py      |  4 ++--
 tst/unit/bundle/resolve_test.py     |  2 +-
 6 files changed, 30 insertions(+), 48 deletions(-)

diff --git a/src/agent_env/bundle/ledger.py b/src/agent_env/bundle/ledger.py
index 7bff620b..aeb1de3f 100644
--- a/src/agent_env/bundle/ledger.py
+++ b/src/agent_env/bundle/ledger.py
@@ -91,11 +91,10 @@ def __init__(self, store: LocalSqliteDocumentStore, plan: Plan):
     def for_plan(cls, plan: Plan) -> Ledger:
         return cls(get_config().local_namespace_document_store(), plan)
 
-    def check(self, write: Write, versions: Mapping[tuple[str, str], int]) -> Check:
-        """Compare ``write`` with the latest version the ledger records for it. ``versions`` holds, by (store, id),
-        the version each earlier write of the plan left or, in a dry run, would leave; an env or agent is hashed
-        with those of what it needs, so one written anew rewrites it too."""
-        needs = {need: versions[need] for need in write.needs}
+    def check(self, write: Write, needs: Mapping[tuple[str, str], int]) -> Check:
+        """Compare ``write`` with the latest version the ledger records for it. ``needs`` holds, by (store, id),
+        the version each earlier write it needs left or, in a dry run, would leave; an env or agent is hashed with
+        them, so one written anew rewrites it too."""
         stored = self._stored_version(write)
         digest = self.digest(write, needs)
         if digest is None:
diff --git a/src/agent_env/bundle/materialize.py b/src/agent_env/bundle/materialize.py
index 9198aca8..d13e1481 100644
--- a/src/agent_env/bundle/materialize.py
+++ b/src/agent_env/bundle/materialize.py
@@ -13,7 +13,7 @@
 from __future__ import annotations
 
 import copy
-from collections.abc import Callable, Iterator, Mapping
+from collections.abc import Callable, Iterator
 from contextlib import contextmanager, nullcontext
 from dataclasses import dataclass
 from typing import Any
@@ -84,11 +84,17 @@ def materialize(
     tasks = [write for write in plan.writes if write.kind is BundleKind.TASK]
     evals = [write for write in plan.writes if write.kind is BundleKind.EVAL]
     done: dict[tuple[str, str], Materialized] = {}
-    versions: dict[tuple[str, str], int] = {}
 
     def through_ledger(write: Write, write_fn: Callable[[], int]) -> None:
-        done[_key(write)] = _through_ledger(plan, ledger, write, versions, write_fn, dry_run)
-        versions[_key(write)] = done[_key(write)].version
+        with _noted(plan, write, "checking" if dry_run else "writing"):
+            check = ledger.check(write, {need: done[need].version for need in write.needs})
+            if check.unchanged:
+                version = check.version
+            elif dry_run:
+                version = check.next_version
+            else:
+                version = ledger.record(check, write_fn)
+        done[_key(write)] = Materialized(write, version, check.unchanged, check.reasons)
         if on_write is not None:
             on_write(done[_key(write)])
 
@@ -175,19 +181,6 @@ def _unwritable(write: Write) -> str | None:
     return None
 
 
-def _through_ledger(plan: Plan, ledger: Ledger, write: Write, versions: Mapping[tuple[str, str], int],
-                    write_fn: Callable[[], int], dry_run: bool) -> Materialized:
-    with _noted(plan, write, "checking" if dry_run else "writing"):
-        check = ledger.check(write, versions)
-        if check.unchanged:
-            version = check.version
-        elif dry_run:
-            version = check.next_version
-        else:
-            version = ledger.record(check, write_fn)
-    return Materialized(write, version, check.unchanged, check.reasons)
-
-
 @contextmanager
 def _noted(plan: Plan, write: Write, doing: str) -> Iterator[None]:
     try:
@@ -203,8 +196,9 @@ def _build(write: Write) -> Task:
 
 def _preflight(write: Write, task: Task, unwritten: set[tuple[str, str]]) -> tuple[list[str], list[TaskStep]]:
     """``task``'s preflight problems, and the steps with a preflight left unchecked because they read a
-    (store, id) in ``unwritten``, which a dry run would write first; an artifact and an agent can share an id. A step reading one of the task's own outputs is skipped
-    too, and not listed: the output only exists once the task runs."""
+    (store, id) in ``unwritten``, which a dry run would write first; an artifact and an agent can share an id.
+    A step reading one of the task's own outputs is skipped too, and not listed: the output only exists once the
+    task runs."""
     outputs = {output.id for output in write.source.outputs}
     problems, unchecked = [], []
     for config, step in zip(write.source.config, task.steps):
diff --git a/src/agent_env/bundle/resolve.py b/src/agent_env/bundle/resolve.py
index 32fea60c..2fcc2ac4 100644
--- a/src/agent_env/bundle/resolve.py
+++ b/src/agent_env/bundle/resolve.py
@@ -285,9 +285,8 @@ def _outputs(self, entry: BundleEntry, declared: list[tuple[int, dict, type]]) -
 
     def _depends_on(self, entry: BundleEntry, steps: list[dict]) -> list[list[str] | None]:
         """Each step's ``depends_on`` as the step ids it names, read as building the step reads it; None when it
-        has none. One that can't be read, names a step by anything but its id, or names no step in the task, is a
-        problem here: a step reading an earlier step's output is checked against these edges before any step is
-        built."""
+        has none. One that can't be read, or names no step in the task, is a problem here: a step reading an
+        earlier step's output is checked against these edges before any step is built."""
         step_ids = {step.get("id") for step in steps}
         depends = []
         for step in steps:
@@ -298,14 +297,11 @@ def _depends_on(self, entry: BundleEntry, steps: list[dict]) -> list[list[str] |
                 read = []
             ids = None if read is None else []
             for dep in read or ():
-                if not isinstance(dep.task_step_id, str):
-                    self._problem(entry, f"step {step.get('id')!r}: depends_on names a step by its id, not "
-                                         f"{dep.task_step_id!r}")
-                    continue
-                if dep.task_step_id not in step_ids:
+                if isinstance(dep.task_step_id, str) and dep.task_step_id in step_ids:
+                    ids.append(dep.task_step_id)
+                else:
                     self._problem(entry, f"step {step.get('id')!r}: depends_on {dep.task_step_id!r} names no step in "
                                          "this task")
-                ids.append(dep.task_step_id)
             depends.append(ids)
         return depends
 
@@ -391,7 +387,7 @@ def _upstream(steps: list[dict], depends: list[list[str] | None]) -> list[set[in
     for index, step in enumerate(steps):
         if isinstance(step.get("id"), str):
             index_of.setdefault(step["id"], index)
-    direct = [set(range(index)) if ids is None else {index_of[dep_id] for dep_id in ids if dep_id in index_of}
+    direct = [set(range(index)) if ids is None else {index_of[dep_id] for dep_id in ids}
               for index, ids in enumerate(depends)]
     upstream = []
     for index in range(len(steps)):
diff --git a/src/agent_env/task/task.py b/src/agent_env/task/task.py
index d6cb9301..1e638440 100644
--- a/src/agent_env/task/task.py
+++ b/src/agent_env/task/task.py
@@ -345,17 +345,10 @@ def _validate_dag(cls, steps: list[TaskStep]) -> None:
             for dep in step.depends_on:
                 if dep.task_step_id in prior:
                     continue
-                if dep.task_step_id in ids:
-                    raise ValueError(
-                        f"Step '{step.id}' at position {i} declares depends_on "
-                        f"'{dep.task_step_id}', which doesn't come before it "
-                        "(forward refs are not allowed: a step depends only on the steps before it)"
-                    )
-                raise ValueError(
-                    f"Step '{step.id}' at position {i} declares depends_on "
-                    f"'{dep.task_step_id}', which names no step in this task "
-                    f"(steps before it: {sorted(prior)})"
-                )
+                why = ("doesn't come before it (forward refs are not allowed: a step depends only on the steps "
+                       "before it)" if dep.task_step_id in ids else
+                       f"names no step in this task (steps before it: {sorted(prior)})")
+                raise ValueError(f"Step '{step.id}' at position {i} declares depends_on '{dep.task_step_id}', which {why}")
 
     def preflight(self) -> list[str]:
         """Every step's resolvable-config problems, in step order; empty when fine.
diff --git a/tst/unit/bundle/ledger_test.py b/tst/unit/bundle/ledger_test.py
index 48e6379d..7ca7b332 100644
--- a/tst/unit/bundle/ledger_test.py
+++ b/tst/unit/bundle/ledger_test.py
@@ -122,7 +122,7 @@ def _record_changed(ledger, writes):
     """Check and record ``writes`` in order, as the materializer does; the checks, by id."""
     checks, versions = {}, {}
     for write in writes:
-        check = ledger.check(write, versions)
+        check = ledger.check(write, {need: versions[need] for need in write.needs})
         version = check.version if check.unchanged else ledger.record(check, lambda: _written(write))
         versions[write.kind.store, write.id] = version
         checks[write.id] = check
@@ -133,7 +133,7 @@ def _checked(ledger, writes):
     """Check ``writes`` in order without recording any, as a dry run does; the checks, by id."""
     checks, versions = {}, {}
     for write in writes:
-        check = ledger.check(write, versions)
+        check = ledger.check(write, {need: versions[need] for need in write.needs})
         versions[write.kind.store, write.id] = check.version if check.unchanged else check.next_version
         checks[write.id] = check
     return checks
diff --git a/tst/unit/bundle/resolve_test.py b/tst/unit/bundle/resolve_test.py
index a05aa8d4..eed82765 100644
--- a/tst/unit/bundle/resolve_test.py
+++ b/tst/unit/bundle/resolve_test.py
@@ -249,7 +249,7 @@ def test_an_output_resolves_for_a_step_that_depends_on_its_writer(make, depends_
     ([7], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not 7'),
     ([{"id": "cli"}], 'a depends_on entry is a step id or {"task_step_id": "<step id>"}, not {\'id\': \'cli\'}'),
     (["clii"], "depends_on 'clii' names no step in this task"),
-    ([{"task_step_id": ["cli"]}], "depends_on names a step by its id, not ['cli']"),
+    ([{"task_step_id": ["cli"]}], "depends_on ['cli'] names no step in this task"),
 ])
 def test_a_depends_on_that_cant_be_read_is_named_beside_the_output_it_then_cant_reach(make, depends_on, problem):
     assert set(problems(make(tasks={"t": [