Promote CLI dev to main for 0.1.2

CHANGELOG.md

@@ -8,6 +8,16 @@ using [PEP 440](https://packaging.python.org/en/latest/specifications/version-sp
 
 ## [Unreleased]
 
+## 0.1.2 - 2026-10-11
+
+### Added
+
+- Supports HTTPS artifact and metadata inputs in task workers, with TLS validation and bounded downloads.
+
+### Changed
+
+- Imports pipeline runner modules only when running a pipeline, reducing startup time for plugin inspection and QAPI builds.
+
 ## 0.1.1 - 2026-10-04
 
 ### Added

README.md

@@ -80,6 +80,54 @@ These options are explicit and mutually exclusive. Omitting both leaves the
 plugin without a default; the Adagio app will flag that plugin until an
 environment is selected for a run.
 
+## Remote artifact inputs
+
+An artifact input accepts a local path or an HTTPS URL, both on the command
+line and in an arguments JSON file:
+
+```bash
+adagio run pipeline.adg --config runtime.toml --cache-dir ./cache \
+  --input-seqs 'https://example.org/data/seqs.qza' --log-dir ./logs
+```
+
+```json
+{"version": 1, "inputs": {"seqs": "https://example.org/data/seqs.qza"}}
+```
+
+Pass that file with `--arguments arguments.json`. The flag name (`seqs` here)
+comes from the pipeline's input name.
+
+The consuming task downloads the artifact inside its execution environment;
+the orchestration process does not fetch it. The original archive is loaded
+without re-importing, preserving its UUID and QIIME provenance. The downloaded
+type must match the declared input type and consuming action. Collections may
+contain both local paths and URLs; artifact URLs can also supply metadata when
+the artifact supports a metadata view. Metadata inputs also accept HTTPS URLs
+to TSV files, which QIIME loads directly without importing them as artifacts.
+
+Downloads stream to disk with bounded retries and share a locked staging
+directory for the run. Every new run fetches again before QIIME result-cache
+lookup, including runs with reuse enabled. To verify a specific version, append
+`#sha256=<64 hexadecimal digits>` to the URL. A mismatch fails the input.
+Transfer resumption and persistent download caching are not implemented yet.
+
+The staging directory lives in the run's temporary work directory, which is
+under the system temporary directory (`TMPDIR` on macOS and Linux). Point
+`TMPDIR` at a larger volume for large inputs. A download whose reported size
+exceeds the free space there fails before transfer; disk errors are not
+retried. Download time appears as `input_download_seconds` in the task timings.
+
+Raw data imports, remote manifests, visualizations, login flows, and embedded
+URL credentials are not supported.
+The URL need not end in `.qza`; its contents are validated. Workers need network
+access to the source. No `wget` or `curl` installation is needed.
+
+Task logs contain transfer progress. `--log-dir` also saves per-task
+`*_inputs.json` receipts with the source, final URL, byte count, SHA-256, type,
+and original UUID for artifacts (metadata TSVs have no artifact UUID).
+Connected runs include these receipts in their task-finished events. URL query strings are omitted from receipts and download logs; the full
+source is represented by a hash and remains in the supplied run arguments.
+
 ## Catalog pipelines
 
 Run a pipeline from the Adagio pipeline catalog:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "adagio-cli"
-version = "0.1.1"
+version = "0.1.2"
 description = "Command-line runner for Adagio pipeline files."
 readme = "README.md"
 requires-python = ">=3.10"

src/adagio/cli/main.py

@@ -1,31 +1,30 @@
+"""CLI entrypoint.
+
+Only the subcommand that runs gets imported. The pipeline runner pulls in
+pydantic, the executors and the catalog client, which `qapi list`/`build`
+never need, and the desktop app spawns those in a fresh interpreter per
+Inspect, so every module loaded here is paid for on each spawn.
+"""
+
 import json
 import sys
 from contextlib import ExitStack
 from functools import partial
 from pathlib import Path
-from typing import Annotated, Any
+from typing import TYPE_CHECKING, Annotated, Any
 
 from cyclopts import App, Group, Parameter
 from cyclopts.panel import CycloptsPanel
 from rich.console import Console
 
 from .. import __version__
-from ..app.parsers.pipeline import Input as InputSpec
-from ..app.parsers.pipeline import Output as OutputSpec
-from ..app.parsers.pipeline import Parameter as ParamSpec
-from ..app.parsers.pipeline import parse_inputs, parse_outputs, parse_parameters
-from ..executors.cache_support import CACHE_DIR_HELP, REUSE_HELP, resolve_cache_dir_path
 from .args import ShowParamsMode, extract_flag_value, promote_positional_pipeline
-from .config import load_run_config
-from .dynamic import build_dynamic_run
-from .pipeline import run_pipeline_cli
-from .pipeline_sources import (
-    PipelineResolution,
-    PipelineResolutionError,
-    resolve_pipeline_reference_details,
-)
-from .qapi import run_qapi
-from .runner import run_pipeline_from_kwargs
+
+if TYPE_CHECKING:
+    from ..app.parsers.pipeline import Input as InputSpec
+    from ..app.parsers.pipeline import Output as OutputSpec
+    from ..app.parsers.pipeline import Parameter as ParamSpec
+    from .pipeline_sources import PipelineResolution
 
 
 console = Console()
@@ -53,13 +52,27 @@ def main(argv: list[str] | None = None) -> None:
         return
 
     if argv and argv[0] == "qapi":
+        from .qapi import run_qapi
+
         run_qapi(argv[1:])
         return
 
     if argv and argv[0] == "pipeline":
+        from .pipeline import run_pipeline_cli
+
         run_pipeline_cli(argv[1:])
         return
 
+    _run_pipeline_app(argv)
+
+
+def _run_pipeline_app(argv: list[str]) -> None:
+    from ..app.parsers.pipeline import parse_inputs, parse_outputs, parse_parameters
+    from ..executors.cache_support import CACHE_DIR_HELP, REUSE_HELP
+    from .config import load_run_config
+    from .dynamic import build_dynamic_run
+    from .runner import run_pipeline_from_kwargs
+
     argv, positional_pipeline = promote_positional_pipeline(argv)
     pipeline_str = extract_flag_value(argv, "--pipeline", "-p")
     cache_dir_str = extract_flag_value(argv, "--cache-dir")
@@ -69,7 +82,11 @@ def main(argv: list[str] | None = None) -> None:
             ShowParamsMode(show_mode_str) if show_mode_str else ShowParamsMode.REQUIRED
         )
     except ValueError:
-        console.print(CycloptsPanel("Invalid --show-params value. Use one of: all, missing, required."))
+        console.print(
+            CycloptsPanel(
+                "Invalid --show-params value. Use one of: all, missing, required."
+            )
+        )
         sys.exit(1)
     if pipeline_str is None:
         pipeline_str = positional_pipeline
@@ -80,6 +97,7 @@ def main(argv: list[str] | None = None) -> None:
         help_format="rich",
         version=__version__,
     )
+
     @app.command
     def cache() -> None:
         """Manage the shared QIIME cache directory."""
@@ -188,7 +206,9 @@ def run(
         arguments_path_str = extract_flag_value(argv, "--arguments")
         config_path_str = extract_flag_value(argv, "--config")
         arguments_data = (
-            _load_arguments_data(Path(arguments_path_str), console) if arguments_path_str else None
+            _load_arguments_data(Path(arguments_path_str), console)
+            if arguments_path_str
+            else None
         )
         if config_path_str:
             load_run_config(Path(config_path_str))
@@ -207,8 +227,12 @@ def run(
             visible_input_names={spec.name for spec in visible_inputs},
             visible_param_names={spec.name for spec in visible_params},
             visible_output_names={spec.name for spec in visible_outputs},
-            argument_inputs=arguments_data.get("inputs", {}) if arguments_data else None,
-            argument_params=arguments_data.get("parameters", {}) if arguments_data else None,
+            argument_inputs=arguments_data.get("inputs", {})
+            if arguments_data
+            else None,
+            argument_params=arguments_data.get("parameters", {})
+            if arguments_data
+            else None,
             run_handler=partial(
                 run_pipeline_from_kwargs,
                 console=console,
@@ -221,12 +245,12 @@ def run(
 
 def _filter_visible_specs(
     *,
-    input_specs: list[InputSpec],
-    param_specs: list[ParamSpec],
-    output_specs: list[OutputSpec],
+    input_specs: "list[InputSpec]",
+    param_specs: "list[ParamSpec]",
+    output_specs: "list[OutputSpec]",
     show_mode: ShowParamsMode,
     arguments_data: dict[str, Any] | None,
-) -> tuple[list[InputSpec], list[ParamSpec], list[OutputSpec]]:
+) -> "tuple[list[InputSpec], list[ParamSpec], list[OutputSpec]]":
     if show_mode is ShowParamsMode.ALL:
         return input_specs, param_specs, output_specs
 
@@ -276,13 +300,23 @@ def _load_arguments_data(path: Path, _console: Console | None = None) -> dict[st
     if not isinstance(data.get("inputs"), dict) or not isinstance(
         data.get("parameters"), dict
     ):
-        _con.print(CycloptsPanel("Invalid arguments file: 'inputs' and 'parameters' must be objects."))
+        _con.print(
+            CycloptsPanel(
+                "Invalid arguments file: 'inputs' and 'parameters' must be objects."
+            )
+        )
         sys.exit(1)
     return data
 
 
 def _is_missing(value: Any) -> bool:
-    return value is None or value == "" or value == "<fill me>" or value == [] or value == {}
+    return (
+        value is None
+        or value == ""
+        or value == "<fill me>"
+        or value == []
+        or value == {}
+    )
 
 
 def _resolve_pipeline(
@@ -291,7 +325,12 @@ def _resolve_pipeline(
     console: Console,
     exit_stack: ExitStack,
     download_cache_dir: Path | None = None,
-) -> PipelineResolution:
+) -> "PipelineResolution":
+    from .pipeline_sources import (
+        PipelineResolutionError,
+        resolve_pipeline_reference_details,
+    )
+
     try:
         return resolve_pipeline_reference_details(
             reference,
@@ -306,6 +345,8 @@ def _resolve_pipeline(
 def _resolve_download_cache_dir(raw_value: str | None) -> Path | None:
     if raw_value is None:
         return None
+    from ..executors.cache_support import resolve_cache_dir_path
+
     return resolve_cache_dir_path(cwd=Path.cwd().resolve(), raw_value=raw_value)
 
 

src/adagio/cli/task_exec.py

@@ -1,6 +1,7 @@
 """Internal exec-task subcommand: runs a single QIIME action inside a plugin container."""
 
 import argparse
+import json
 from collections.abc import Mapping
 from contextlib import nullcontext
 import os
@@ -38,6 +39,9 @@ def run_task_exec(argv: list[str]) -> None:
 def _run_task(spec: dict[str, Any]) -> None:
     telemetry = TaskTelemetry(spec.get("progress_path"))
     telemetry.phase("preparing_cache" if spec.get("recycle_pool") else "preparing")
+    # Downloads are timed apart from plugin setup, which they would otherwise dwarf.
+    with telemetry.measure("input_download_seconds"):
+        spec, input_downloads = _stage_remote_inputs(spec)
     setup_started = time.monotonic()
     if spec.get("plugin") == DATA_IMPORT_PLUGIN:
         _run_data_import(spec, telemetry=telemetry)
@@ -172,10 +176,83 @@ def _run_task(spec: dict[str, Any]) -> None:
                 metadata_outputs=saved_metadata_outputs,
                 reused=reused,
                 timings=telemetry.finish(),
+                input_downloads=input_downloads,
             ),
         )
 
 
+def _stage_remote_inputs(spec: dict[str, Any]) -> tuple[dict[str, Any], list[dict]]:
+    """Resolve remote archives on the worker, before QIIME cache lookup.
+
+    Local inputs and raw import behavior remain owned by the normal loader.
+    URLs become paths only here; the orchestration process never fetches them.
+    """
+    from adagio.executors.container_support import is_uri
+    from adagio.remote_inputs import RemoteArtifactStore
+
+    result = dict(spec)
+    downloads = []
+
+    def resolve(name: str, source: str, *, metadata: bool = False) -> str:
+        if not is_uri(source):
+            return source
+        materialization = spec.get("archive_input_materializations", {}).get(name)
+        declared_type = spec.get("remote_input_types", {}).get(name)
+        if materialization or declared_type == "RawData":
+            raise ValueError("URL inputs support QIIME artifacts and metadata, not raw imports.")
+
+        from qiime2 import Artifact, Metadata
+        from qiime2.sdk.util import parse_type
+
+        expected = None
+        if not metadata and declared_type and declared_type != "Metadata":
+            expected = parse_type(declared_type)
+            if declared_type.startswith(("List[", "Collection[")):
+                expected = expected.fields[0]
+
+        def validate(path: str) -> dict[str, str]:
+            try:
+                if metadata and not zipfile.is_zipfile(path):
+                    Metadata.load(path)
+                    return {"type": "Metadata"}
+                artifact = Artifact.load(path)
+                artifact.validate(level="min")
+                if metadata:
+                    artifact.view(Metadata)
+            except Exception as exc:
+                expected_format = "QIIME metadata TSV or metadata-capable .qza" if metadata else "QIIME .qza artifact"
+                raise ValueError(
+                    f"Remote input {name!r} could not be loaded as {expected_format}. "
+                    f"QIIME reported: {type(exc).__name__}: {exc}"
+                ) from None
+            return {"uuid": str(artifact.uuid), "type": str(artifact.type)}
+
+        manifest = spec.get("result_manifest")
+        if not manifest:
+            raise ValueError("Remote inputs require a task result manifest in the run work directory.")
+        store = RemoteArtifactStore(Path(manifest).parent / ".adagio-downloads")
+        path, record = store.resolve(source, validate=validate)
+        if expected is not None and not parse_type(record["type"]) <= expected:
+            raise ValueError(
+                f"Remote input {name!r} has type {record['type']}; expected {declared_type}."
+            )
+        entry = {"input": name, **record}
+        downloads.append(entry)
+        print("Input download: " + json.dumps(entry), flush=True)
+        return path
+
+    for field in ("archive_inputs", "metadata_inputs"):
+        result[field] = {
+            name: resolve(name, source, metadata=field == "metadata_inputs")
+            for name, source in spec.get(field, {}).items()
+        }
+    result["archive_collection_inputs"] = {
+        name: [resolve(name, source) for source in sources]
+        for name, sources in spec.get("archive_collection_inputs", {}).items()
+    }
+    return result, downloads
+
+
 def _run_data_import(spec: dict[str, Any], *, telemetry: TaskTelemetry) -> None:
     """Import raw data into a single artifact and save it (no plugin action).
 

src/adagio/executors/apptainer.py

@@ -82,6 +82,7 @@ def launch(
         spec_path = task_spec_path(task_id=task.id, work_path=request.work_path)
         progress_path = request.work_path / f"{spec_path.stem}_progress.jsonl"
         task_spec = build_task_spec(
+            remote_input_types=dict(request.remote_input_types or {}),
             plugin=task.plugin,
             action=task.action,
             archive_inputs=archive_inputs,
@@ -237,6 +238,7 @@ def launch(
             metadata_outputs[output_name] = str(host_path_from_container(actual_path))
 
         return TaskExecutionResult(
+            input_downloads=output_manifest.get("input_downloads"),
             outputs=resolved_outputs,
             reused=reused,
             metadata_outputs=metadata_outputs,