fix: _wait_for_co_outputs skips cross-level co-outputs

`ParallelExecutor._wait_for_co_outputs` had the primary block on every
co-output of its cmd, drawn from `_cmd_artifact_paths[cmd]` which
spans **all** artifacts across all levels. If a co-output was scheduled
for a later level, its `_execute_artifact` hadn't been called yet, its
`_dvc_done_events[co_path]` could never be set, and the primary parked
on `Event.wait()` forever. Main thread parked in `as_completed`
waiting for the primary's future. Deadlock; the pool ran to timeout.

Concrete failure mode: hudcostreets/path's `path-data months` produces
8 co-outputs — 6 land in Level 1 (they share the same per-year `.pqt`
deps as the primary `www/public/all.pqt`) but 2 land in Level 2
(`data/all.pqt` + `data/all.xlsx`, whose `.dvc` deps reference the
`data/*.pqt` outputs). The `path-data` daily-updater GHA has hung at
its 6h timeout every day since 2026-06-20 as a result.

Fix: track `_scheduled_paths` — a set populated at the start of each
`_execute_level` with that level's paths. `_wait_for_co_outputs`
filters `_cmd_artifact_paths[cmd]` down to co-outputs already in
`_scheduled_paths`, so cross-level entries are silently skipped.
Their .dvcs still get written when their own level runs (via
`_handle_co_output`); the primary just doesn't wait for them.

`test_cross_level_co_output_does_not_deadlock` sets up A (Level 1) +
B (Level 2, depends on A) sharing a cmd, runs the executor in a
thread, and asserts `t.join(timeout=15)` returns. Without the fix
it deadlocks and hits the 15s timeout; with it, both artifacts
complete cleanly.

Full suite: 238 passed, 4 skipped (S3 integration). No regressions.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: ryan-williams

src/dvx/run/executor.py

@@ -162,6 +162,11 @@ def __init__(
         self._dvc_done_events: dict[str, threading.Event] = {
             a.path: threading.Event() for a in artifacts if a.computation is not None
         }
+        # Populated at the start of each level with that level's artifact
+        # paths. `_wait_for_co_outputs` filters against this so the primary
+        # only blocks on same-level co-outputs; cross-level co-outputs
+        # haven't been submitted to the pool yet and can never signal.
+        self._scheduled_paths: set[str] = set()
 
     def execute(self) -> list[ExecutionResult]:
         """Execute all artifacts, respecting dependencies.
@@ -311,6 +316,12 @@ def _execute_level(self, artifacts: list[Artifact]) -> list[ExecutionResult]:
         Returns:
             List of ExecutionResult, one per artifact
         """
+        # Track which artifacts are in-flight this level so `_wait_for_co_outputs`
+        # can filter out cross-level co-outputs (their `_dvc_done_events` won't
+        # be armed until a later level submits them — waiting would deadlock).
+        for a in artifacts:
+            self._scheduled_paths.add(a.path)
+
         if len(artifacts) == 1:
             # Single artifact - run directly without thread pool overhead
             return [self._execute_artifact(artifacts[0])]
@@ -686,16 +697,26 @@ def _signal_dvc_done(self, path: str) -> None:
             ev.set()
 
     def _wait_for_co_outputs(self, cmd: str | None, my_path: str) -> list[str]:
-        """Block until every co-output of ``cmd`` (other than ``my_path``) has
-        finished writing its .dvc. Returns the list of co-output paths.
+        """Block until every same-level co-output of ``cmd`` (other than
+        ``my_path``) has finished writing its .dvc. Returns those paths.
 
         Called from the primary stage's thread before commit + cache push so
         that ``git add -u`` captures every co-output's md5 update and the
         push manifest includes every co-output's blob.
+
+        Cross-level co-outputs are filtered out via ``_scheduled_paths``:
+        their ``_execute_artifact`` won't be invoked until a later level
+        submits them, so their ``_dvc_done_events`` can never be set —
+        waiting would deadlock the pool (as it did on hccs/path when
+        ``data/all.{pqt,xlsx}`` landed a level below their ``www/public/``
+        siblings despite sharing ``path-data months``).
         """
         if not cmd:
             return []
-        co_paths = [p for p in self._cmd_artifact_paths.get(cmd, []) if p != my_path]
+        co_paths = [
+            p for p in self._cmd_artifact_paths.get(cmd, [])
+            if p != my_path and p in self._scheduled_paths
+        ]
         for co_path in co_paths:
             ev = self._dvc_done_events.get(co_path)
             if ev is not None:

tests/test_executor.py

@@ -178,6 +178,59 @@ def test_multi_output_partial_failure(tmp_workdir):
     assert "not created" in failures[0].reason or "not produced" in failures[0].reason
 
 
+def test_cross_level_co_output_does_not_deadlock(tmp_workdir):
+    """Co-outputs of one cmd that end up in different levels must not deadlock.
+
+    When artifact B declares artifact A as a dep AND they share the same cmd,
+    ``_group_into_levels`` places A in Level 1 and B in Level 2 despite
+    them being co-outputs. Before the fix, Level 1's primary called
+    ``_wait_for_co_outputs`` and blocked on B's ``_dvc_done_event`` — which
+    could never fire because B's ``_execute_artifact`` doesn't run until
+    Level 2 is submitted. Deadlock: primary parked on B's event, main
+    thread parked in ``as_completed`` waiting for the primary. Repro
+    matched hccs/path's ``path-data months`` outputs (see 2026-06-19+
+    daily-update runs that hung at 6h GHA timeout).
+    """
+    script = tmp_workdir / "multi.sh"
+    script.write_text(
+        "#!/bin/bash\n"
+        "echo l1 > a.txt\n"
+        "echo l2 > b.txt\n"
+    )
+    script.chmod(0o755)
+    cmd = f"bash {script}"
+
+    # a.txt: no deps → Level 1. b.txt: dep on a.txt → Level 2. Same cmd.
+    a = Artifact(path="a.txt", computation=Computation(cmd=cmd, deps=[]))
+    b = Artifact(
+        path="b.txt",
+        computation=Computation(cmd=cmd, deps=[a]),
+    )
+
+    # Run in a thread so a deadlock surfaces as a timeout, not a hung suite.
+    import threading
+    executor = ParallelExecutor([a, b], ExecutionConfig(max_workers=2), StringIO())
+    results: list = []
+    errors: list = []
+
+    def _run():
+        try:
+            results.extend(executor.execute())
+        except Exception as e:  # pragma: no cover - surfaced via assertion
+            errors.append(e)
+
+    t = threading.Thread(target=_run, daemon=True)
+    t.start()
+    t.join(timeout=15)
+    assert not t.is_alive(), (
+        "executor deadlocked — primary waited on a cross-level co-output "
+        "whose _dvc_done_event was never set"
+    )
+    assert not errors, f"executor raised: {errors[0]!r}"
+    assert len(results) == 2
+    assert all(r.success for r in results), [r.reason for r in results]
+
+
 def test_multi_output_command_failure(tmp_workdir):
     """Test handling when the shared command fails."""
     cmd = "exit 1"