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..aeb1de3f 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,17 @@ 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, 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:
-            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 +111,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 +128,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 +148,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 +237,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..d13e1481 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 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; "
@@ -76,25 +86,36 @@ def materialize(
     done: dict[tuple[str, str], Materialized] = {}
 
     def through_ledger(write: Write, write_fn: Callable[[], int]) -> None:
-        done[_key(write)] = _through_ledger(plan, ledger, write, write_fn)
+        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)])
 
-    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 = {_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"):
                 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,13 +181,6 @@ 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)
-    return Materialized(write, version, check.unchanged, check.reasons)
-
-
 @contextmanager
 def _noted(plan: Plan, write: Write, doing: str) -> Iterator[None]:
     try:
@@ -180,20 +194,29 @@ 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[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 = []
+    problems, unchecked = [], []
     for config, step in zip(write.source.config, task.steps):
-        if not _reads_any(config, outputs):
-            problems.extend(step.preflight())
-    return problems
+        reads = _reads(config)
+        if any(id in outputs for _, id in reads):
+            continue
+        if reads & 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:
+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/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/src/agent_env/bundle/resolve.py b/src/agent_env/bundle/resolve.py
index 4dc3018f..2fcc2ac4 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}: "
@@ -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
@@ -283,6 +283,28 @@ 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 []
+            for dep in read or ():
+                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")
+            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 +380,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}
+              for index, ids in enumerate(depends)]
     upstream = []
     for index in range(len(steps)):
         seen: set[int] = set()
diff --git a/src/agent_env/bundle/run.py b/src/agent_env/bundle/run.py
index 01b2d043..1f533748 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,33 @@ 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 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)
+    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 +257,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..d43d636e 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,28 @@ 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. 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
     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 +188,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/src/agent_env/task/task.py b/src/agent_env/task/task.py
index 699645ff..1e638440 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,12 @@ 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:
-                    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)})"
-                    )
+                if dep.task_step_id in prior:
+                    continue
+                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/src/agent_env/task_step/task_step.py b/src/agent_env/task_step/task_step.py
index 6ea8250e..5dae12e8 100644
--- a/src/agent_env/task_step/task_step.py
+++ b/src/agent_env/task_step/task_step.py
@@ -19,8 +19,23 @@ 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>}``."""
+        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:
+    """``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 +106,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 +142,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/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..7ca7b332 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, {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
+    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, {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
 
@@ -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),
@@ -307,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))
@@ -334,9 +386,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 +397,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 04e634d7..9480c042 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",
@@ -281,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"))
@@ -349,12 +378,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 +406,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):
@@ -399,6 +431,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"])
@@ -406,7 +449,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):
@@ -414,7 +458,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)"]
 
 
@@ -447,19 +491,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"""
@@ -486,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/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/resolve_test.py b/tst/unit/bundle/resolve_test.py
index 1c2378c2..eed82765 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,25 @@ 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"),
+    ([{"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": [
+        {"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 +279,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 "
@@ -407,3 +431,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}",)
diff --git a/tst/unit/bundle/run_test.py b/tst/unit/bundle/run_test.py
index afa786b7..64d2ae52 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):
@@ -205,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)])
@@ -344,3 +366,120 @@ 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_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"])
+
+    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 == []
diff --git a/tst/unit/task/depends_on_test.py b/tst/unit/task/depends_on_test.py
new file mode 100644
index 00000000..2e58178c
--- /dev/null
+++ b/tst/unit/task/depends_on_test.py
@@ -0,0 +1,109 @@
+"""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\'}'),
+])
+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_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))
+
+    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"]