Publish copies of pipeline outputs

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)

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
 

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),
                 )

src/adagio/model/arguments.py

@@ -9,13 +9,15 @@ 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."""
         return '\n'.join([
             *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)

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": "<fill me>"}
+            ),
+        )
+        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()

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)