StepRunner: walk transitive deps before scheduling

Add a post-order walk (deduped by output_path) so callers pass only
terminal steps; runner resolves the full graph, and already-succeeded
deps short-circuit via the existing cache check. The prior "Iterable
exhausted" branch is unreachable under the new traversal and is
removed.

Fixes #5146

Author: ravwojdyla

lib/marin/src/marin/execution/step_runner.py

@@ -88,6 +88,34 @@ def _write_executor_info(step: StepSpec) -> None:
 # ---------------------------------------------------------------------------
 
 
+def _flatten_transitive_deps(steps: Iterable[StepSpec]) -> list[StepSpec]:
+    """Walk transitive deps in post-order, deduping by ``output_path``.
+
+    Callers can pass only terminal steps; the runner resolves the full graph.
+    A cycle in the graph is a construction-time invariant violation and raises.
+    """
+    seen: set[str] = set()
+    in_stack: set[str] = set()
+    ordered: list[StepSpec] = []
+
+    def visit(step: StepSpec) -> None:
+        path = step.output_path
+        if path in seen:
+            return
+        if path in in_stack:
+            raise ValueError(f"Cycle detected in step graph involving {step.name_with_hash}")
+        in_stack.add(path)
+        for dep in step.deps:
+            visit(dep)
+        in_stack.remove(path)
+        seen.add(path)
+        ordered.append(step)
+
+    for step in steps:
+        visit(step)
+    return ordered
+
+
 class StepRunner:
     """Runs ``StepSpec`` objects respecting their dependencies.
 
@@ -106,10 +134,11 @@ def run(
     ) -> None:
         """Eagerly run steps, launching each as soon as its deps are satisfied.
 
-        Concurrency is bounded by the thread pool (``max_concurrent`` workers,
-        default 8). If a step's dependencies haven't been seen yet, it is
-        buffered until they complete. Raises if the iterable is exhausted and
-        buffered steps still have unsatisfied dependencies.
+        The input iterable may contain only the terminal steps you want to
+        reach; the runner walks transitive deps (deduped by ``output_path``)
+        before scheduling. Concurrency is bounded by the thread pool
+        (``max_concurrent`` workers, default 8). Already-succeeded deps
+        (``STATUS_SUCCESS`` on disk) resolve via the cache check.
         """
         max_workers = max_concurrent or 8
         if max_workers < 1:
@@ -196,7 +225,8 @@ def _do_launch(step: StepSpec) -> None:
             else:
                 completed.add(path)
 
-        for step in steps:
+        flattened = _flatten_transitive_deps(steps)
+        for step in flattened:
             path_to_name[step.output_path] = step.name_with_hash
 
             _harvest()
@@ -212,22 +242,6 @@ def _do_launch(step: StepSpec) -> None:
 
         # Drain remaining running and waiting steps
         while running or waiting:
-            if not running and waiting:
-                _flush_waiting()
-                if not running and waiting:
-                    missing = []
-                    for s in waiting:
-                        unmet = [
-                            _display_name(d.output_path)
-                            for d in s.deps
-                            if d.output_path not in completed and d.output_path not in failed
-                        ]
-                        missing.append(f"  {s.name_with_hash}: needs {unmet}")
-                    raise RuntimeError(
-                        f"Iterable exhausted with {len(waiting)} step(s) with unsatisfied dependencies:\n"
-                        + "\n".join(missing)
-                    )
-
             _harvest(block=True)
             _flush_waiting()
 

tests/execution/test_step_runner.py

@@ -372,29 +372,98 @@ def test_runner_max_concurrent(tmp_path: Path):
     assert train_artifact.tokens_seen > 0
 
 
-def test_runner_raises_clear_error_for_unmet_deps(tmp_path: Path):
-    """When the iterable omits a dependency, the runner must name the offending step and
-    its unmet dep paths — not crash with ``TypeError: unhashable type: 'list'``."""
+def test_runner_walks_transitive_deps(tmp_path: Path):
+    """Passing only terminal steps should cause the runner to walk and run transitive deps."""
+    executed: list[str] = []
+
+    def record(name: str):
+        def _fn(output_path: str) -> PathMetadata:
+            executed.append(name)
+            return PathMetadata(path=output_path)
+
+        return _fn
 
     dep = StepSpec(
-        name="missing_upstream",
-        override_output_path=(tmp_path / "missing_upstream").as_posix(),
+        name="dep",
+        override_output_path=(tmp_path / "dep").as_posix(),
+        fn=record("dep"),
+    )
+    mid = StepSpec(
+        name="mid",
+        override_output_path=(tmp_path / "mid").as_posix(),
+        deps=[dep],
+        fn=record("mid"),
+    )
+    terminal = StepSpec(
+        name="terminal",
+        override_output_path=(tmp_path / "terminal").as_posix(),
+        deps=[mid],
+        fn=record("terminal"),
+    )
+
+    StepRunner().run([terminal])
+
+    assert executed == ["dep", "mid", "terminal"]
+
+
+def test_runner_walks_transitive_deps_with_cache_hit(tmp_path: Path):
+    """Deps already succeeded on disk must be recognized via cache-hit during the walk."""
+    dep = StepSpec(
+        name="dep",
+        override_output_path=(tmp_path / "dep").as_posix(),
         fn=lambda output_path: PathMetadata(path=output_path),
     )
+    downstream_ran: list[str] = []
+
+    def run_downstream(output_path: str) -> PathMetadata:
+        downstream_ran.append(output_path)
+        return PathMetadata(path=output_path)
+
     downstream = StepSpec(
         name="downstream",
         override_output_path=(tmp_path / "downstream").as_posix(),
         deps=[dep],
+        fn=run_downstream,
+    )
+
+    # Prime the cache for ``dep`` only.
+    StepRunner().run([dep])
+    assert downstream_ran == []
+
+    # Pass only ``downstream``; the runner walks deps and cache-hits ``dep``.
+    StepRunner().run([downstream])
+    assert downstream_ran == [(tmp_path / "downstream").as_posix()]
+
+
+def test_runner_dedups_shared_deps(tmp_path: Path):
+    """A dep shared by multiple terminals must be executed exactly once."""
+    dep_runs: list[str] = []
+
+    def run_dep(output_path: str) -> PathMetadata:
+        dep_runs.append(output_path)
+        return PathMetadata(path=output_path)
+
+    dep = StepSpec(
+        name="shared_dep",
+        override_output_path=(tmp_path / "shared_dep").as_posix(),
+        fn=run_dep,
+    )
+    a = StepSpec(
+        name="a",
+        override_output_path=(tmp_path / "a").as_posix(),
+        deps=[dep],
+        fn=lambda output_path: PathMetadata(path=output_path),
+    )
+    b = StepSpec(
+        name="b",
+        override_output_path=(tmp_path / "b").as_posix(),
+        deps=[dep],
         fn=lambda output_path: PathMetadata(path=output_path),
     )
 
-    runner = StepRunner()
-    with pytest.raises(RuntimeError, match=r"Iterable exhausted .* unsatisfied dependencies") as exc_info:
-        runner.run([downstream])
+    StepRunner().run([a, b])
 
-    message = str(exc_info.value)
-    assert downstream.name_with_hash in message
-    assert dep.output_path in message
+    assert dep_runs == [(tmp_path / "shared_dep").as_posix()]
 
 
 def test_runner_preserves_underlying_step_exception(tmp_path: Path):