Adds community pipeline support [WIP]

README.md

@@ -19,3 +19,27 @@ Set up the project and run the test suite with:
 uv sync --group dev
 uv run pytest
 ```
+
+## Catalog pipelines
+
+Run a pipeline from the Adagio pipeline catalog:
+
+```bash
+adagio pipeline show @adagio/microbial-diversity
+adagio run @adagio/microbial-diversity --cache-dir /path/to/cache --arguments run-arguments.json
+```
+
+`@adagio/<slug>` first resolves against a nearby local `adagio-pipelines`
+checkout when one is available. If no local catalog is found, Adagio fetches
+`pipeline.adg` from `cymis/adagio-pipelines` on GitHub, checking `official`
+before `community`.
+
+During `adagio run`, remote catalog pipelines are downloaded under the selected
+`--cache-dir` and reused by source name and slug on later runs. `adagio pipeline
+show` uses a temporary download when it fetches from GitHub because it does not
+take a cache directory.
+
+Private GitHub access is explicit: set `GITHUB_TOKEN` or `GH_TOKEN` to a token
+that can read `cymis/adagio-pipelines`; with a token, the CLI fetches through
+the GitHub contents API. The CLI does not read browser, git, or `gh` credentials
+automatically.

src/adagio/cli/dynamic.py

@@ -304,7 +304,7 @@ def build_dynamic_run(
             CliParameter(
                 name=("--pipeline", "-p"),
                 group=command_group,
-                help="Path to the pipeline JSON file.",
+                help="Path to the pipeline file or a catalog reference like @adagio/slug.",
             ),
         ]
     }
@@ -589,7 +589,7 @@ def run(
     run.__doc__ = (
         "Run an Adagio pipeline.\n\n"
         "Dynamic inputs, parameters, and outputs are loaded from the pipeline file and exposed as CLI options.\n"
-        "Use: adagio run --pipeline PATH --help"
+        "Use: adagio run --pipeline PATH-OR-@ADAGIO/SLUG --help"
     )
     return run
 

src/adagio/cli/main.py

@@ -1,5 +1,6 @@
 import json
 import sys
+from contextlib import ExitStack
 from functools import partial
 from pathlib import Path
 from typing import Annotated, Any
@@ -13,11 +14,16 @@
 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
+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
 
@@ -56,6 +62,7 @@ def main(argv: list[str] | None = None) -> None:
 
     argv, positional_pipeline = promote_positional_pipeline(argv)
     pipeline_str = extract_flag_value(argv, "--pipeline", "-p")
+    cache_dir_str = extract_flag_value(argv, "--cache-dir")
     show_mode_str = extract_flag_value(argv, "--show-params")
     try:
         show_mode = (
@@ -108,7 +115,7 @@ def run(
                 Parameter(
                     name=("--pipeline", "-p"),
                     group=command_group,
-                    help="Path to the pipeline JSON file.",
+                    help="Path to the pipeline file or a catalog reference like @adagio/slug.",
                 ),
             ],
             arguments: Annotated[
@@ -155,45 +162,61 @@ def run(
         ):
             """Run a pipeline (requires --pipeline; dynamic options come from that file)."""
             _ = (config, show_params, cache_dir, reuse)
-            console.print(CycloptsPanel("Missing --pipeline. Try:\n  adagio run --pipeline pipeline.json --help"))
+            console.print(
+                CycloptsPanel(
+                    "Missing --pipeline. Try:\n"
+                    "  adagio run --pipeline pipeline.adg --help\n"
+                    "  adagio run @adagio/microbial-diversity --help"
+                )
+            )
             sys.exit(1)
 
         app(argv)
         return
 
-    pipeline_path = Path(pipeline_str)
-    data = json.loads(pipeline_path.read_text(encoding="utf-8"))
-    input_specs = parse_inputs(data)
-    param_specs = parse_parameters(data)
-    output_specs = parse_outputs(data)
-    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
-    )
-    if config_path_str:
-        load_run_config(Path(config_path_str))
-    visible_inputs, visible_params, visible_outputs = _filter_visible_specs(
-        input_specs=input_specs,
-        param_specs=param_specs,
-        output_specs=output_specs,
-        show_mode=show_mode,
-        arguments_data=arguments_data,
-    )
+    with ExitStack() as exit_stack:
+        pipeline_resolution = _resolve_pipeline(
+            pipeline_str,
+            console=console,
+            exit_stack=exit_stack,
+            download_cache_dir=_resolve_download_cache_dir(cache_dir_str),
+        )
+        data = json.loads(pipeline_resolution.path.read_text(encoding="utf-8"))
+        input_specs = parse_inputs(data)
+        param_specs = parse_parameters(data)
+        output_specs = parse_outputs(data)
+        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
+        )
+        if config_path_str:
+            load_run_config(Path(config_path_str))
+        visible_inputs, visible_params, visible_outputs = _filter_visible_specs(
+            input_specs=input_specs,
+            param_specs=param_specs,
+            output_specs=output_specs,
+            show_mode=show_mode,
+            arguments_data=arguments_data,
+        )
 
-    dynamic_run = build_dynamic_run(
-        input_specs=input_specs,
-        param_specs=param_specs,
-        output_specs=output_specs,
-        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,
-        run_handler=partial(run_pipeline_from_kwargs, console=console),
-    )
-    app.command(dynamic_run, name="run")
-    app(argv)
+        dynamic_run = build_dynamic_run(
+            input_specs=input_specs,
+            param_specs=param_specs,
+            output_specs=output_specs,
+            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,
+            run_handler=partial(
+                run_pipeline_from_kwargs,
+                console=console,
+                resolved_pipeline=pipeline_resolution,
+            ),
+        )
+        app.command(dynamic_run, name="run")
+        app(argv)
 
 
 def _filter_visible_specs(
@@ -262,5 +285,29 @@ def _is_missing(value: Any) -> bool:
     return value is None or value == "" or value == "<fill me>" or value == [] or value == {}
 
 
+def _resolve_pipeline(
+    reference: str,
+    *,
+    console: Console,
+    exit_stack: ExitStack,
+    download_cache_dir: Path | None = None,
+) -> PipelineResolution:
+    try:
+        return resolve_pipeline_reference_details(
+            reference,
+            exit_stack=exit_stack,
+            download_cache_dir=download_cache_dir,
+        )
+    except PipelineResolutionError as error:
+        console.print(CycloptsPanel(str(error)))
+        sys.exit(1)
+
+
+def _resolve_download_cache_dir(raw_value: str | None) -> Path | None:
+    if raw_value is None:
+        return None
+    return resolve_cache_dir_path(cwd=Path.cwd().resolve(), raw_value=raw_value)
+
+
 if __name__ == "__main__":
     main()

src/adagio/cli/pipeline.py

@@ -1,11 +1,13 @@
 import json
+from contextlib import ExitStack
 from pathlib import Path
 
 from cyclopts import App
 from rich.console import Console
 
 from ..describe import render_pipeline_text
 from ..model.pipeline import AdagioPipeline
+from .pipeline_sources import PipelineResolutionError, resolve_pipeline_reference
 
 console = Console()
 
@@ -21,7 +23,13 @@ def run_pipeline_cli(argv: list[str]) -> None:
 
 def show_pipeline(pipeline: Path) -> None:
     """Print a pipeline summary to the terminal."""
-    data = json.loads(pipeline.read_text(encoding="utf-8"))
-    pipeline_data = data.get("spec", data) if isinstance(data, dict) else data
-    parsed_pipeline = AdagioPipeline.model_validate(pipeline_data)
-    console.print(render_pipeline_text(parsed_pipeline), soft_wrap=True)
+    with ExitStack() as exit_stack:
+        try:
+            pipeline_path = resolve_pipeline_reference(pipeline, exit_stack=exit_stack)
+        except PipelineResolutionError as error:
+            raise SystemExit(str(error)) from error
+
+        data = json.loads(pipeline_path.read_text(encoding="utf-8"))
+        pipeline_data = data.get("spec", data) if isinstance(data, dict) else data
+        parsed_pipeline = AdagioPipeline.model_validate(pipeline_data)
+        console.print(render_pipeline_text(parsed_pipeline), soft_wrap=True)