Promote dev to main — adagio-cli 0.1.3

CHANGELOG.md

@@ -6,7 +6,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 using [PEP 440](https://packaging.python.org/en/latest/specifications/version-specifiers/) version strings for Python releases.
 
-## [Unreleased]
+## 0.1.3 - 2026-10-17
+
+### Added
+
+- Applies the variable values and input paths saved with a pipeline (`configs` in the `.adg`) as the lowest argument layer, under `--arguments` and command-line flags. `--help` shows a variable's value as the parameter's default.
+- Runs each plugin and task in the execution environment saved with the pipeline (`configs[0].execution` in the `.adg`, the shape of a `--config` file), so a downloaded pipeline runs without a separate run-config file. A `--config` file overrides the saved environments. The run header lists each plugin's environment and whether it came from the pipeline file or `--config`.
+- Treats a per-output path ending in `/` as a directory: the output is written into it as `<output name>.qza` (`.qzv` for a visualization), matching `--output-dir`.
+
+### Changed
+
+- Reads variable values from the values saved with the pipeline rather than the pipeline signature; signatures written by older releases still supply their defaults.
+
+### Fixed
+
+- Uses the same framework type checking for URL and local artifact inputs,
+  avoiding parse failures on polymorphic type display labels.
 
 ## 0.1.2 - 2026-10-11
 

pyproject.toml

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

src/adagio/app/parsers/pipeline.py

@@ -9,7 +9,9 @@
 class Parameter(BaseModel):
     id: UUID
     name: str
-    required: bool
+    # Variables are declared without values; `default` is read only from specs
+    # written before configs existed.
+    required: bool = True
     default: Optional[Any] = None
     type: str
     description: Optional[str] = None

src/adagio/bundled_config.py

@@ -0,0 +1,134 @@
+"""The configuration bundled in a pipeline file.
+
+A pipeline's spec is its signature: which inputs, variables and outputs exist.
+Their values live in configs - the same shape as an ``--arguments`` file - and
+the .adg envelope carries them in a ``configs`` list beside ``spec``. The CLI
+applies the first bundled config as the lowest layer: an ``--arguments`` file
+overrides it, and command-line flags override both.
+
+Files written before configs existed kept variable defaults inside the spec
+(``signature.parameters[].default`` and ``meta.editor_variables[].value``);
+those read as a bundled config too, so an older file keeps working unchanged.
+
+The same config may carry an ``execution`` object: the environments and
+resources the app would run each plugin and task with, in the shape of a
+``--config`` file. The CLI applies them under any ``--config`` file the same
+way, so a downloaded pipeline runs without one.
+"""
+
+from typing import Any
+
+from pydantic import ValidationError
+
+from .cli.config import AdagioRunConfig
+from .model.arguments import AdagioArguments
+
+
+def _record(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}
+
+
+def _configured(value: Any) -> bool:
+    return value is not None and value != ""
+
+
+def bundled_config(data: Any) -> dict[str, Any] | None:
+    """The config a pipeline file carries, or ``None`` when it carries no values.
+
+    ``data`` is the parsed file: the .adg envelope, or a bare spec.
+    """
+    if not isinstance(data, dict):
+        return None
+    configs = data.get("configs")
+    if isinstance(configs, list):
+        for entry in configs:
+            if isinstance(entry, dict):
+                return {
+                    "inputs": _record(entry.get("inputs")),
+                    "parameters": _record(entry.get("parameters")),
+                    "outputs": _record(entry.get("outputs")),
+                }
+    return _legacy_config(data.get("spec", data))
+
+
+def saved_run_config(data: Any) -> AdagioRunConfig | None:
+    """The execution settings saved with a pipeline file, or ``None``.
+
+    Raises ``SystemExit`` with a readable message when the saved settings do
+    not have the shape of a ``--config`` file.
+    """
+    if not isinstance(data, dict) or not isinstance(data.get("configs"), list):
+        return None
+    for entry in data["configs"]:
+        if not isinstance(entry, dict):
+            continue
+        execution = entry.get("execution")
+        if execution is None:
+            return None
+        try:
+            return AdagioRunConfig.model_validate(execution)
+        except ValidationError as error:
+            raise SystemExit(
+                "The execution environments saved with this pipeline are not "
+                f"valid: {error}"
+            ) from error
+    return None
+
+
+def _legacy_config(spec: Any) -> dict[str, Any] | None:
+    """Variable values an older spec stored in its signature or editor metadata."""
+    if not isinstance(spec, dict):
+        return None
+    parameters: dict[str, Any] = {}
+    signature = _record(spec.get("signature"))
+    for parameter in signature.get("parameters") or []:
+        if (
+            isinstance(parameter, dict)
+            and isinstance(parameter.get("name"), str)
+            and _configured(parameter.get("default"))
+        ):
+            parameters[parameter["name"]] = parameter["default"]
+    meta = _record(spec.get("meta"))
+    for variable in meta.get("editor_variables") or []:
+        if (
+            isinstance(variable, dict)
+            and isinstance(variable.get("name"), str)
+            and variable.get("kind") != "Path or URL"
+            and _configured(variable.get("value"))
+            and variable["name"] not in parameters
+        ):
+            parameters[variable["name"]] = variable["value"]
+    if not parameters:
+        return None
+    return {"inputs": {}, "parameters": parameters, "outputs": {}}
+
+
+def apply_bundled_config(
+    arguments: AdagioArguments,
+    config: dict[str, Any] | None,
+    *,
+    input_names: set[str],
+    parameter_names: set[str],
+    output_names: set[str],
+) -> None:
+    """Lay the bundled config under ``arguments``, for the names the signature has.
+
+    The author's config may hold values for variables nothing uses yet; those
+    are not arguments of this pipeline and are skipped rather than refused,
+    unlike unknown names in a user's own ``--arguments`` file.
+    """
+    if config is None:
+        return
+    for name, value in config["inputs"].items():
+        if name in input_names and _configured(value):
+            arguments.inputs[name] = value
+    for name, value in config["parameters"].items():
+        if name in parameter_names and _configured(value):
+            arguments.parameters[name] = value
+    outputs = {
+        name: value
+        for name, value in config["outputs"].items()
+        if name in output_names and isinstance(value, str) and value
+    }
+    if outputs:
+        arguments.outputs = outputs

src/adagio/cli/config.py

@@ -7,6 +7,7 @@
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from ..executors.base import TaskEnvironmentOverride
+from ..executors.defaults import EnvironmentLayer
 
 try:
     import tomllib
@@ -181,3 +182,13 @@ def named_environment_overrides(
         if (override := raw_override.to_task_environment_override()) is not None
     }
     return resolved or None
+
+
+def environment_layer(run_config: AdagioRunConfig, *, source: str) -> EnvironmentLayer:
+    """The environments ``run_config`` sets, as one layer named ``source``."""
+    return EnvironmentLayer(
+        source=source,
+        default_override=default_environment_override(run_config),
+        plugin_overrides=named_environment_overrides(run_config.plugins) or {},
+        task_overrides=named_environment_overrides(run_config.tasks) or {},
+    )

src/adagio/cli/dynamic.py

@@ -92,11 +92,17 @@ def _display_type_label(*, spec_type: str | None, type_hint: Any, is_input: bool
     return _pipeline_type_label(type_hint)
 
 
+_OUTPUT_PATH_RULE = (
+    "Overrides --output-dir for this output. A path ending in / is a directory "
+    "the output is written into under its own name."
+)
+
+
 def _output_path_help(description: str | None) -> str:
     cleaned = (description or "").strip()
     if cleaned:
-        return f"{cleaned} Overrides --output-dir for this output."
-    return "Overrides --output-dir for this output."
+        return f"{cleaned} {_OUTPUT_PATH_RULE}"
+    return _OUTPUT_PATH_RULE
 
 
 def _render_pipeline_type(
@@ -334,7 +340,10 @@ def build_dynamic_run(
         CliParameter(
             name=("--config",),
             group=command_group,
-            help="Path to a TOML runtime config file.",
+            help=(
+                "Path to a TOML or JSON runtime config file. Its environments "
+                "override those saved with the pipeline."
+            ),
         ),
     ]
     annotations["cache_dir"] = Annotated[

src/adagio/cli/main.py

@@ -68,6 +68,7 @@ def main(argv: list[str] | None = None) -> None:
 
 def _run_pipeline_app(argv: list[str]) -> None:
     from ..app.parsers.pipeline import parse_inputs, parse_outputs, parse_parameters
+    from ..bundled_config import bundled_config, saved_run_config
     from ..executors.cache_support import CACHE_DIR_HELP, REUSE_HELP
     from .config import load_run_config
     from .dynamic import build_dynamic_run
@@ -210,6 +211,22 @@ def run(
             if arguments_path_str
             else None
         )
+        # What --help shows as defaults and what --show-params counts as
+        # missing follow the same layers as the run: the bundled config
+        # underneath, the arguments file over it.
+        bundled = bundled_config(data)
+        if bundled is not None:
+            arguments_data = {
+                "inputs": {
+                    **bundled["inputs"],
+                    **(arguments_data or {}).get("inputs", {}),
+                },
+                "parameters": {
+                    **bundled["parameters"],
+                    **(arguments_data or {}).get("parameters", {}),
+                },
+            }
+        saved_run_config(data)
         if config_path_str:
             load_run_config(Path(config_path_str))
         visible_inputs, visible_params, visible_outputs = _filter_visible_specs(

src/adagio/cli/runner.py

@@ -7,11 +7,14 @@
 
 from rich import box
 from rich.console import Console
+from rich.markup import escape
 from rich.panel import Panel
 from rich.text import Text
 
+from ..bundled_config import apply_bundled_config, bundled_config, saved_run_config
 from .config import (
     default_environment_override,
+    environment_layer,
     load_run_config,
     named_environment_overrides,
 )
@@ -87,12 +90,23 @@ def run_pipeline_from_kwargs(
     parsed_pipeline = AdagioPipeline.model_validate(pipeline_data)
     arguments = parsed_pipeline.signature.to_default_arguments()
     run_config = load_run_config(config_file)
+    saved_execution = saved_run_config(data)
     output_names = [output.name for output in parsed_pipeline.signature.outputs]
 
     input_names = {name for _, name in input_bindings}
     param_names = {name for _, name in param_bindings}
     output_name_set = set(output_names)
 
+    # Layers, lowest first: the config bundled in the file, the arguments
+    # file, then command-line flags.
+    apply_bundled_config(
+        arguments,
+        bundled_config(data),
+        input_names=input_names,
+        parameter_names=param_names,
+        output_names=output_name_set,
+    )
+
     if arguments_file is not None:
         file_data = json.loads(arguments_file.read_text(encoding="utf-8"))
         arguments_data = AdagioArgumentsFile.model_validate(file_data)
@@ -196,18 +210,32 @@ def run_pipeline_from_kwargs(
         console.print(f"[bold]Cache:[/bold] {describe_cache_config(cache_config)}")
 
     from ..executors import select_default_executor
+    from ..executors.defaults import ConfigurableTaskEnvironmentResolver
 
-    executor = select_default_executor(
-        default_override=default_environment_override(run_config),
-        plugin_overrides=named_environment_overrides(
+    # Environments, lowest first: those saved with the pipeline, then the
+    # --config file.
+    environments = {
+        "default_override": default_environment_override(run_config),
+        "plugin_overrides": named_environment_overrides(
             run_config.plugins if run_config is not None else {}
         ),
-        task_overrides=named_environment_overrides(
+        "task_overrides": named_environment_overrides(
             run_config.tasks if run_config is not None else {}
         ),
-    )
+        "lower_layers": (
+            [environment_layer(saved_execution, source="pipeline file")]
+            if saved_execution is not None
+            else []
+        ),
+    }
+    executor = select_default_executor(**environments)
 
     if not suppress_header:
+        _print_environments(
+            console,
+            parsed_pipeline,
+            ConfigurableTaskEnvironmentResolver(**environments),
+        )
         console.print(f"[bold]Executing pipeline[/bold] ({executor.mode_label})")
 
     target_ids: set[str] | None = None
@@ -226,6 +254,41 @@ def run_pipeline_from_kwargs(
     )
 
 
+def _print_environments(console: Console, pipeline: Any, resolver: Any) -> None:
+    """List each plugin's environment and where it came from.
+
+    A task whose environment differs from the rest of its plugin's gets a row
+    of its own, so a per-task choice is never hidden behind the plugin's.
+    """
+    from ..model.task import PluginActionTask
+
+    rows: dict[str, str] = {}
+    plugin_rows: dict[str, str] = {}
+    for task in pipeline.iter_tasks():
+        if not isinstance(task, PluginActionTask):
+            continue
+        try:
+            environment = resolver.resolve(task=task)
+        except ValueError:
+            description = "[red]not configured[/red]"
+        else:
+            source = resolver.source_of(task=task) or "default"
+            description = (
+                f"{environment.kind} {escape(environment.reference)} "
+                f"[dim]({source})[/dim]"
+            )
+        if task.plugin not in plugin_rows:
+            plugin_rows[task.plugin] = description
+            rows[task.plugin] = description
+        elif plugin_rows[task.plugin] != description:
+            rows[f"{task.plugin} ({task.name or task.id})"] = description
+    if not rows:
+        return
+    console.print("[bold]Environments:[/bold]")
+    for label, description in rows.items():
+        console.print(f"  {escape(label)}: {description}", highlight=False)
+
+
 def _is_missing(value: Any) -> bool:
     """Treat placeholders and null values as missing."""
     return value is None or value == "" or value == "<fill me>" or value == [] or value == {}