feat(stage): republish upstream outputs so the chain survives (#26)

Nextflow hands each process only the previous one's /results, so a stage that
does not republish what it received removes it from the chain. The packaging
stage needs every stage's data_process.json and the refined reconstructions,
and the resample stage needs refinement/final-voxel and final-world for the
specimen pass.

The old capsules each had their own carry-forward helpers, which had already
drifted apart; this is one tested implementation. Files are hardlinked with a
copy fallback, so passing a multi-gigabyte tree along costs little.

Co-authored-by: Peter Grotz <199680633+peter-grotz@users.noreply.github.com>

Author: peter-grotz

src/exaspim_swc_processing/stage.py

@@ -26,12 +26,17 @@
 from aind_data_schema.core.processing import DataProcess, ProcessStage, ResourceUsage
 from aind_data_schema_models.process_names import ProcessName
 
+from exaspim_swc_processing.layout import place_artifact
+
 CAPSULE_URL_TEMPLATE = "https://codeocean.allenneuraldynamics.org/capsule/{capsule_id}"
 """Code Ocean capsule URL, used when no repository URL is given."""
 
 DATA_PROCESS_FILENAME = "data_process.json"
 """Name a stage writes its record under."""
 
+UPSTREAM_STAGES = ("dispatch", "refinement", "alignment", "final")
+"""Stage directories that are passed along the chain, in pipeline order."""
+
 
 def _resources() -> ResourceUsage:
     """Describe the machine the stage ran on.
@@ -224,3 +229,44 @@ def write_stage_process(process: DataProcess, output_dir: Path) -> Path:
     path = output_dir / DATA_PROCESS_FILENAME
     path.write_text(process.model_dump_json(indent=2), encoding="utf-8")
     return path
+
+
+def carry_forward(
+    data_dir: Path,
+    results_dir: Path,
+    stages: Sequence[str],
+) -> list[str]:
+    """Republish upstream stage outputs so later stages can still see them.
+
+    Nextflow hands each process only the previous one's ``/results``, so a stage that does
+    not republish what it received removes it from the chain. The terminal packaging stage
+    needs every stage's ``data_process.json`` and the refined reconstructions, so each
+    intermediate stage has to pass them along.
+
+    Files are hardlinked where the filesystem allows and copied otherwise, so passing a
+    multi-gigabyte tree through costs little beyond directory entries.
+
+    Parameters
+    ----------
+    data_dir : Path
+        Directory the upstream outputs are mounted at.
+    results_dir : Path
+        Directory this stage writes to.
+    stages : Sequence[str]
+        Stage directory names to republish, if present.
+
+    Returns
+    -------
+    list[str]
+        The stages that were found and republished, in the order given.
+    """
+    carried: list[str] = []
+    for stage in stages:
+        source = data_dir / stage
+        if not source.is_dir():
+            continue
+        for path in sorted(source.rglob("*")):
+            if path.is_file():
+                place_artifact(path, results_dir / stage / path.relative_to(source))
+        carried.append(stage)
+    return carried

tests/test_stage.py

@@ -9,7 +9,9 @@
 
 from exaspim_swc_processing.stage import (
     DATA_PROCESS_FILENAME,
+    UPSTREAM_STAGES,
     build_stage_process,
+    carry_forward,
     installed_version,
     resolve_code,
     write_stage_process,
@@ -151,3 +153,35 @@ def test_write_round_trips_through_the_schema(tmp_path: Path) -> None:
     assert path.name == DATA_PROCESS_FILENAME
     reloaded = DataProcess.model_validate_json(path.read_text(encoding="utf-8"))
     assert reloaded.name == "s"
+
+
+def test_carry_forward_republishes_upstream_stages(tmp_path: Path) -> None:
+    """Nextflow hands on only the previous stage's results, so each stage must republish."""
+    data = tmp_path / "data"
+    (data / "dispatch").mkdir(parents=True)
+    (data / "dispatch" / "data_process.json").write_text("{}", encoding="utf-8")
+    (data / "refinement" / "final-voxel").mkdir(parents=True)
+    (data / "refinement" / "final-voxel" / "N001-794492-HP.swc").write_text("x", encoding="utf-8")
+
+    results = tmp_path / "results"
+    carried = carry_forward(data, results, UPSTREAM_STAGES)
+
+    assert carried == ["dispatch", "refinement"]
+    assert (results / "dispatch" / "data_process.json").is_file()
+    assert (results / "refinement" / "final-voxel" / "N001-794492-HP.swc").is_file()
+
+
+def test_carry_forward_skips_absent_stages(tmp_path: Path) -> None:
+    """The first stage has no upstream outputs; that is not an error."""
+    assert carry_forward(tmp_path / "data", tmp_path / "results", UPSTREAM_STAGES) == []
+
+
+def test_carry_forward_preserves_nested_structure(tmp_path: Path) -> None:
+    """Deeply nested outputs survive the hop."""
+    data = tmp_path / "data"
+    nested = data / "final" / "ccf_space_reconstructions" / "jsons"
+    nested.mkdir(parents=True)
+    (nested / "N001-794492-HP.json").write_text("{}", encoding="utf-8")
+    results = tmp_path / "results"
+    carry_forward(data, results, ("final",))
+    assert (results / "final/ccf_space_reconstructions/jsons/N001-794492-HP.json").is_file()