From 9a09d0c9ab153c244d2323b4a1440a9ff43e051f Mon Sep 17 00:00:00 2001 From: John Chase Date: Tue, 14 Jul 2026 22:41:43 -0700 Subject: [PATCH] Publish secondary copies of pipeline outputs --- src/adagio/cli/runner.py | 8 +++++ src/adagio/cli/runtime.py | 12 +++++++ src/adagio/executors/task_environments.py | 15 ++++++++- src/adagio/model/arguments.py | 3 ++ tests/test_runtime_flags.py | 23 +++++++++++++ tests/test_serial_runner.py | 40 +++++++++++++++++++++++ 6 files changed, 100 insertions(+), 1 deletion(-) diff --git a/src/adagio/cli/runner.py b/src/adagio/cli/runner.py index 8c40166..9301912 100644 --- a/src/adagio/cli/runner.py +++ b/src/adagio/cli/runner.py @@ -120,10 +120,18 @@ def run_pipeline_from_kwargs( "Unknown outputs in arguments file: " + ", ".join(unknown_outputs), ) + unknown_publish = sorted(set(arguments_data.publish) - output_name_set) + if unknown_publish: + _error_exit( + console, + "Unknown publish outputs in arguments file: " + ", ".join(unknown_publish), + ) + arguments.inputs.update(arguments_data.inputs) arguments.parameters.update(arguments_data.parameters) if arguments_data.outputs is not None: arguments.outputs = arguments_data.outputs + arguments.publish = arguments_data.publish for ident, original in input_bindings: value = kwargs.get(ident) diff --git a/src/adagio/cli/runtime.py b/src/adagio/cli/runtime.py index 1eb3bde..d959312 100644 --- a/src/adagio/cli/runtime.py +++ b/src/adagio/cli/runtime.py @@ -268,6 +268,18 @@ def _build_arguments( if resolved_outputs is not None: arguments.outputs = resolved_outputs + resolved_publish = _resolve_outputs( + runtime_arguments.get("publish"), storage_root=storage_root + ) + if isinstance(resolved_publish, dict): + output_names = {output.name for output in pipeline.signature.outputs} + unknown = sorted(set(resolved_publish) - output_names) + if unknown: + raise SystemExit( + "Unknown publish outputs in arguments file: " + ", ".join(unknown) + ) + arguments.publish = resolved_publish + if _outputs_need_default(arguments.outputs): arguments.outputs = output_dir diff --git a/src/adagio/executors/task_environments.py b/src/adagio/executors/task_environments.py index ff836f7..34e4bca 100644 --- a/src/adagio/executors/task_environments.py +++ b/src/adagio/executors/task_environments.py @@ -588,14 +588,27 @@ def _save_outputs( monitor.start_save_output() state.save_output_started = True + publish_destination = None try: shutil.copy2(source_path, destination) + if output.name in arguments.publish: + publish_destination = resolve_output_destination( + output_name=output.name, + output_names=[item.name for item in sig.outputs], + outputs=arguments.publish, + source_path=Path(destination), + ) + publish_parent = os.path.dirname(publish_destination) + if publish_parent: + os.makedirs(publish_parent, exist_ok=True) + if Path(publish_destination).resolve() != Path(destination).resolve(): + shutil.copy2(destination, publish_destination) except Exception as exc: # noqa: BLE001 if monitor is not None: monitor.finish_output( output_id=output.id, output_name=output.name, - destination=destination, + destination=publish_destination or destination, status="failed", error=str(exc), ) diff --git a/src/adagio/model/arguments.py b/src/adagio/model/arguments.py index 8982ff5..fac4f9c 100644 --- a/src/adagio/model/arguments.py +++ b/src/adagio/model/arguments.py @@ -9,6 +9,7 @@ class AdagioArguments(BaseModel): inputs: dict[str, InputValue] parameters: dict[str, AllowableValue] outputs: str | dict[str, str] + publish: dict[str, str] = Field(default_factory=dict) def __repr__(self): """Format arguments for display.""" @@ -16,6 +17,7 @@ def __repr__(self): *self._format_repr_sect(self.inputs, 'inputs'), *self._format_repr_sect(self.parameters, 'parameters'), *self._format_repr_sect(self.outputs, 'outputs'), + *self._format_repr_sect(self.publish, 'publish'), ]) def _format_repr_sect(self, section, name): @@ -38,3 +40,4 @@ class AdagioArgumentsFile(BaseModel): inputs: dict[str, InputValue] = Field(default_factory=dict) parameters: dict[str, AllowableValue] = Field(default_factory=dict) outputs: str | dict[str, str] | None = None + publish: dict[str, str] = Field(default_factory=dict) diff --git a/tests/test_runtime_flags.py b/tests/test_runtime_flags.py index 5f489ac..fb49714 100644 --- a/tests/test_runtime_flags.py +++ b/tests/test_runtime_flags.py @@ -3,11 +3,13 @@ import unittest import urllib.error from pathlib import Path +from types import SimpleNamespace from unittest.mock import patch from rich.console import Console from adagio.cli import runtime as runtime_cli +from adagio.model.arguments import AdagioArguments _SPEC = { @@ -37,6 +39,27 @@ def _write(root: Path, name: str, payload) -> Path: class RuntimeFlagsTests(unittest.TestCase): + def test_publish_paths_are_kept_separate_from_normal_outputs(self) -> None: + signature = SimpleNamespace( + outputs=[SimpleNamespace(name="table")], + to_default_arguments=lambda: AdagioArguments( + inputs={}, parameters={}, outputs={"table": ""} + ), + ) + pipeline = SimpleNamespace(signature=signature) + + arguments = runtime_cli._build_arguments( + pipeline=pipeline, + runtime_arguments={ + "inputs": {}, + "publish": {"table": "/exports/table.qza"}, + }, + output_dir="/jobs/current/outputs", + ) + + self.assertEqual(arguments.outputs, "/jobs/current/outputs") + self.assertEqual(arguments.publish, {"table": "/exports/table.qza"}) + def _run(self, extra_argv, config_payload): console = Console() executor = _CapturingExecutor() diff --git a/tests/test_serial_runner.py b/tests/test_serial_runner.py index ff6cc62..f69dbb8 100644 --- a/tests/test_serial_runner.py +++ b/tests/test_serial_runner.py @@ -120,6 +120,46 @@ def launch(self, *, environment, request, console=None): # noqa: ANN001 class SerialRunnerOutputTests(unittest.TestCase): + def test_publish_copies_output_without_replacing_normal_destination(self) -> None: + output_def = FakeOutputDef(id="out-1", name="result") + pipeline = FakePipeline( + tasks=[FakeTask(id="task-1", outputs={"result": FakeEndpoint("out-1")})], + outputs=[output_def], + ) + monitor = RecordingMonitor() + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + output_dir = root / "outputs" + publish_path = root / "published" / "named-result" + arguments = AdagioArguments( + inputs={}, + parameters={}, + outputs=str(output_dir), + publish={"result": str(publish_path)}, + ) + + def resolve_task(task, state, console): # noqa: ANN001 + del task, console + produced = state.work_path / "task-result.qza" + produced.write_text("published", encoding="utf-8") + state.scope["out-1"] = str(produced) + return False + + run_serial_pipeline( + pipeline=pipeline, + arguments=arguments, + resolve_task=resolve_task, + finish_outputs=_save_outputs, + monitor=monitor, + ) + + normal = output_dir / "result.qza" + published = publish_path.with_suffix(".qza") + self.assertEqual(normal.read_text(encoding="utf-8"), "published") + self.assertEqual(published.read_text(encoding="utf-8"), "published") + self.assertEqual(monitor.saved_outputs[0][2], str(normal)) + def test_collection_input_manifest_expands_to_paths(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir)