\dvx run\ now caches output blobs to local DVC cache

Previously, \`dvx run\` updated the \`.dvc\` file with new output hashes
but did NOT copy output blobs into the local cache. Historical
versions were lost as soon as the next run overwrote the working
copy — \`dvx checkout -R HEAD^\` would fail with "cache missing".

Added \`cache_blob()\` in \`dvx.cache\` (public wrapper over existing
\`_cache_file()\`) and call it in executor after \`compute_md5(out)\`
for both regular outputs and co-outputs. Idempotent — no-op if the
blob is already cached.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ryan-williams

specs/done/run-should-cache-outputs.md

@@ -0,0 +1,48 @@
+# `dvx run` should add outputs to local cache
+
+## Problem
+
+After `dvx run` completes a stage that produces an output file, it updates the `.dvc` file with the new output hash but does NOT copy the output blob into the local DVX cache (`.dvc/cache/files/md5/...`).
+
+Consequence: the output exists only in the working tree. If it gets overwritten by a subsequent stage (e.g. a daily pipeline overwriting `crash-log.parquet`), the previous hash's content is lost locally. Tools that look up historical versions by hash (e.g. `dxdc -R <commit>`, `dvx checkout`) fail with "cache missing".
+
+## Reproduction
+
+```bash
+# Initial state
+dvx run njsp/data/crash-log.parquet.dvc     # produces hash A
+git commit -am "crash log A"
+
+# Cache check
+ls .dvc/cache/files/md5/${A:0:2}/${A:2}      # MISSING — bug
+
+# Next day
+dvx run njsp/data/crash-log.parquet.dvc     # produces hash B, overwrites file
+git commit -am "crash log B"
+
+# Try to look up A
+dvx checkout -R HEAD^ njsp/data/crash-log.parquet
+# Error: Cache missing for 'HEAD^': dd0a841755c4afe199c3ca3f3853ec11
+```
+
+## Compare: `dvx add`
+
+`dvx add` (CLI) calls `add_to_cache()` in `dvx/cache.py`, which copies the file into the cache keyed by MD5. `dvx run` has no equivalent call in its executor path.
+
+## Fix
+
+After a non-side-effect stage produces its output successfully, call `add_to_cache()` on the output path before writing the `.dvc` file. Side-effect stages don't produce outputs, so no change there.
+
+In `executor.py` around line 531-544, after computing `md5 = compute_md5(out)`:
+
+```python
+from dvx.cache import add_to_cache
+try:
+    add_to_cache(out, force=True)  # idempotent if blob already present
+except Exception as e:
+    self._log(f"  ⚠ {path}: couldn't cache output: {e}")
+```
+
+## Related
+
+Also consider: should `dvx run --push` push cached blobs to the remote? Currently `--push` refers to git push, but the S3 blob push is ambiguous. Could be `--push-cache` or similar.

src/dvx/cache.py

@@ -483,6 +483,35 @@ def _hash_single_file(file_path) -> str:
     return md5.hexdigest()
 
 
+def cache_blob(file_path, md5: str, force: bool = False):
+    """Copy a file into the DVC cache keyed by its MD5.
+
+    Idempotent: returns immediately if the blob is already cached
+    (unless force=True).
+
+    Args:
+        file_path: Path to the file to cache
+        md5: MD5 hash of the file (must be precomputed)
+        force: Overwrite existing cache entry
+
+    Returns:
+        Path to the cached blob
+    """
+    from pathlib import Path
+
+    from dvc.repo import Repo as DVCRepo
+
+    try:
+        root = DVCRepo.find_root()
+    except Exception:
+        root = "."
+    cache_dir = Path(root) / ".dvc" / "cache" / "files" / "md5"
+    cache_dir.mkdir(parents=True, exist_ok=True)
+
+    _cache_file(file_path, md5, cache_dir, force)
+    return cache_dir / md5[:2] / md5[2:]
+
+
 def _cache_file(file_path, file_hash: str, cache_dir, force: bool = False):
     """Copy a file to DVC cache atomically."""
     import shutil

src/dvx/run/executor.py

@@ -531,6 +531,13 @@ def _execute_artifact(self, artifact: Artifact) -> ExecutionResult:
             md5 = compute_md5(out)
             size = compute_file_size(out)
 
+            # Cache the output blob so historical versions can be retrieved
+            try:
+                from dvx.cache import cache_blob
+                cache_blob(out, md5)
+            except Exception as e:
+                self._log(f"  ⚠ {path}: couldn't cache output: {e}")
+
             dvc_file = write_dvc_file(
                 output_path=out,
                 md5=md5,
@@ -586,6 +593,13 @@ def _handle_co_output(self, artifact: Artifact, cmd: str) -> ExecutionResult:
             md5 = compute_md5(out)
             size = compute_file_size(out)
 
+            # Cache the co-output blob
+            try:
+                from dvx.cache import cache_blob
+                cache_blob(out, md5)
+            except Exception as e:
+                self._log(f"  ⚠ {path}: couldn't cache co-output: {e}")
+
             deps_hashes = {}
             git_deps_hashes = {}
             if self.config.provenance and artifact.computation:

tests/test_executor.py

@@ -422,3 +422,61 @@ def test_after_ordering(tmp_path):
     assert a_level is not None
     assert b_level is not None
     assert a_level < b_level, f"stage_a (level {a_level}) should be before stage_b (level {b_level})"
+
+
+def test_run_caches_output_blob(tmp_path):
+    """dvx run copies output blobs to local cache so historical versions persist."""
+    import os
+
+    os.chdir(tmp_path)
+
+    # Initialize a DVC repo (need .dvc dir for cache_blob)
+    subprocess.run(["dvc", "init", "--no-scm"], cwd=tmp_path, capture_output=True, check=True)
+
+    output = tmp_path / "result.txt"
+    artifact = Artifact(
+        path=str(output),
+        computation=Computation(cmd=f"echo 'hello world' > {output}"),
+    )
+
+    config = ExecutionConfig(max_workers=1)
+    output_log = StringIO()
+    executor = ParallelExecutor([artifact], config, output_log)
+    results = executor.execute()
+
+    assert all(r.success for r in results)
+    assert output.exists()
+
+    # Verify the output blob was added to cache
+    from dvx.run.hash import compute_md5
+    md5 = compute_md5(output)
+    cache_path = tmp_path / ".dvc" / "cache" / "files" / "md5" / md5[:2] / md5[2:]
+    assert cache_path.exists(), f"Output blob should be cached at {cache_path}"
+    assert cache_path.read_text() == output.read_text()
+
+
+def test_run_cache_idempotent(tmp_path):
+    """Caching is idempotent — re-running doesn't error if blob already cached."""
+    import os
+
+    os.chdir(tmp_path)
+    subprocess.run(["dvc", "init", "--no-scm"], cwd=tmp_path, capture_output=True, check=True)
+
+    output = tmp_path / "data.txt"
+    artifact = Artifact(
+        path=str(output),
+        computation=Computation(cmd=f"echo 'data' > {output}"),
+    )
+
+    # Run twice
+    for _ in range(2):
+        config = ExecutionConfig(max_workers=1, force=True)
+        executor = ParallelExecutor([artifact], config, StringIO())
+        results = executor.execute()
+        assert all(r.success for r in results)
+
+    # Blob should still be cached
+    from dvx.run.hash import compute_md5
+    md5 = compute_md5(output)
+    cache_path = tmp_path / ".dvc" / "cache" / "files" / "md5" / md5[:2] / md5[2:]
+    assert cache_path.exists()