From 82ca2412fd1159a20eb7ca5cffc622d47d5bcaa1 Mon Sep 17 00:00:00 2001 From: Volv G Date: Fri, 11 Sep 2026 22:27:17 -0700 Subject: [PATCH] Submit Python-authored pipelines in one step (v0.1.12) Shipping a Python-authored pipeline took two commands and left a mess. `pipelines compile` wrote `pipeline.yaml`, a `pipeline.components.yaml` sidecar and sometimes a `pipeline.subgraphs/` directory next to the source, then `pipeline-runs submit` consumed them. The artifacts are pure build output, but nothing removed them, so they were committed by accident, went stale against the script they were generated from, and turned a one-line change into a two-command ritual that could silently submit yesterday's graph. Add `pipeline-runs submit-from-python SCRIPT.py`, which compiles, hydrates, submits and then removes the bundle -- on dry runs and on failures too: tangle sdk pipeline-runs submit-from-python pipeline.py \ --override batch_size=100 \ --image eval-slim=registry.example/img@sha256:... \ --arg shop=acme --annotation owner=team The two value tiers stay distinct and are documented as such: `--override` and `--image` are compile-time (`cfg` values and `@task(image_id=...)` resolution), while `--arg` / `--args-json` / `--arg-secret` are run-time arguments. Run-tier flags are otherwise identical to `submit`, so the new command is not a second dialect to learn. The bundle is compiled beside the script, never in a temp directory. The compiler resolves the root output path up front and derives every relative URL from it, so an author-written `ref(url="file://./component.yaml")` only resolves when the bundle sits in the script's own directory. The script is resolved physically first, so a symlinked script compiles next to its real target where its `cfg` file and siblings live. The name comes from an exclusive create, so two concurrent compiles of one script cannot collide or overwrite a hand-written sibling YAML, and cleanup is stem-scoped: only `.yaml`, `.components.yaml` and `.subgraphs/` are removed, so a neighbour is never caught in it. Allocation failures are translated into the CLI's own error. Once the temporary file exists, every exit -- a failed initial close, a compile error, a failed submit -- passes through best-effort stem-scoped cleanup, so the original error is what the user sees. Hydration is forced. The compiled bundle references local sidecars the server cannot read, so `--no-hydrate` would only produce a run that fails later and further away. A multi-entry `--config` prepares *every* entry before creating any run: each is compiled exactly once, hydrated, merged with its run arguments and secrets, validated, and frozen into a submit body; only then are the frozen bodies submitted in order, with no recompilation. A typo or a broken graph in the last entry therefore creates no runs at all, instead of landing entry one and failing afterwards. Bundles stay on disk until the final submit returns, because run-lifecycle hooks are handed the compiled path and may read it. Submission still goes through the existing run lifecycle, so submission-id stamping and ambiguous-submit recovery are unchanged. Config keys are checked before anything is compiled. A key that merely looks like a supported field (`overide:`) is rejected rather than dropped, since dropping it submits a valid run that is not the one authored, while unrelated keys in a shared config file are left alone. `hydrate:` and `pipeline_path:` are rejected outright, and the two safety-sensitive values are type-checked: a non-integer `submit_recovery_attempts` would crash an ambiguous submit before its recovery lookup, and a string `trusted_hydration_cli` would quietly enable allow-all hydration. `--override` / `--image` parsing moves verbatim from `pipelines compile` into a shared helper so both commands accept the same syntax and reject the same mistakes. `pipeline-runs submit` is unchanged. Assisted-By: devx/01a0930b-82ad-779b-afde-8cefb3614443 --- README.md | 54 ++ .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../tangle-cli/src/tangle_cli/cli_helpers.py | 33 + .../src/tangle_cli/pipeline_runs_cli.py | 412 ++++++++- .../src/tangle_cli/pipelines_cli.py | 36 +- pyproject.toml | 2 +- skills/tangent/OSS-CONVENTIONS.md | 1 + skills/tangent/references/tangle-tools.md | 23 + tests/test_packaging.py | 2 +- tests/test_pipeline_runs_cli.py | 789 ++++++++++++++++++ tests/test_pipelines_cli.py | 68 ++ uv.lock | 2 +- 12 files changed, 1392 insertions(+), 32 deletions(-) diff --git a/README.md b/README.md index b21d280..f8fad0e 100644 --- a/README.md +++ b/README.md @@ -386,6 +386,7 @@ Pipeline run API/submit commands live under `sdk pipeline-runs`: ```bash uv run tangle sdk pipeline-runs submit pipeline.yaml --dry-run uv run tangle sdk pipeline-runs submit pipeline.yaml --arg key=value --annotation owner=team +uv run tangle sdk pipeline-runs submit-from-python pipeline.py --arg key=value uv run tangle sdk pipeline-runs wait RUN_ID --max-wait 600 --poll-interval 10 uv run tangle sdk pipeline-runs logs EXECUTION_ID uv run tangle sdk pipeline-runs annotations set RUN_ID key value @@ -401,6 +402,59 @@ uv run tangle sdk pipelines compile pipeline.py -o pipeline.yaml uv run tangle sdk pipelines compile pipeline.py -o pipeline.yaml --pipeline pipeline_fn_name ``` +To compile and submit in one step, without keeping the compiled YAML around, use +`pipeline-runs submit-from-python`: + +```bash +uv run tangle sdk pipeline-runs submit-from-python pipeline.py \ + --override batch_size=100 \ + --image eval-slim=registry.example/eval-slim@sha256:... \ + --arg shop=acme --annotation owner=team +``` + +It compiles the script, hydrates the bundle, submits the run, and removes the +compiled artifacts again — including on dry runs and failures. Its run-tier +flags are the same as `pipeline-runs submit`; the extra compile-tier flags are +`--pipeline`, repeatable `--override KEY=VALUE`, and repeatable `--image ID=REF`. +Use `pipelines compile` instead when the compiled YAML itself is what you want. + +The two value tiers are distinct and easy to confuse: + +| Flag | Tier | Meaning | +| --- | --- | --- | +| `--override KEY=VALUE` | compile | `cfg` value used while the graph is built | +| `--image ID=REF` | compile | resolves `@task(image_id=ID)` to a registry ref | +| `--arg` / `--args-json` / `--arg-secret` | run | pipeline arguments for the created run | + +Notes: + +- Hydration is always on: the compiled bundle references local sidecars + (`resolve://./.components.yaml#…`, `file://.subgraphs/…`) that the + server cannot read. There is no `--no-hydrate` here. +- The bundle is compiled next to the script (never in `/tmp`) under a unique + hidden `.tangle-submit-*.yaml` name, because relative refs such as + `ref(url="file://./component.yaml")` resolve against the *output* directory. + The name is allocated with an exclusive create, so concurrent compiles of the + same script never collide or overwrite a sibling YAML. A symlinked script + compiles next to its resolved target, where its `config.yaml` and relative + refs actually live. +- The bundle stays readable until every run has been submitted, so + run-lifecycle hooks still see an existing `pipeline_path`; cleanup then + removes exactly that stem's files (`.yaml`, `.components.yaml`, + `.subgraphs/`) and nothing else. +- Submission never waits; use `pipeline-runs wait RUN_ID`. +- With a multi-entry `--config` file, entries may carry different `--override` / + `--image` values, and **every entry is fully prepared before any run is + created**: each is compiled exactly once, hydrated, merged with its run + arguments and secrets, validated, and frozen into a submit body; only then are + the frozen bodies submitted in order (no recompilation). An unsupported config + key (`hydrate:`, `pipeline_path:`), a malformed `override` / `image` / + `arg-secret` value, an input given as both `--arg` and `--arg-secret`, a + missing script, or a compile/hydrate error in *any* entry therefore creates no + runs and leaves no artifacts behind. A runtime failure while submitting entry + N can still follow the successful submits of entries 1..N-1 — that is inherent + to creating N runs. + A minimal graph uses `@pipeline` for the graph and `@task` for local Python components. `@task` functions are not executed at compile time; the compiler records call sites, emits a sibling `.components.yaml` with `local_from_python` entries, and rewrites task component refs to that sidecar. Hydrate later regenerates the same component YAML from the Python source. ```python diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index a578094..11c9185 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.11" + __version__ = "0.1.12" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/cli_helpers.py b/packages/tangle-cli/src/tangle_cli/cli_helpers.py index 7868fb4..558069c 100644 --- a/packages/tangle-cli/src/tangle_cli/cli_helpers.py +++ b/packages/tangle-cli/src/tangle_cli/cli_helpers.py @@ -46,6 +46,39 @@ def optional_path(value: str | pathlib.Path | object | None) -> pathlib.Path | N return None +def parse_overrides(values: list[str] | None) -> dict[str, str]: + """Parse repeatable ``--override KEY=VALUE`` compile-time cfg overrides. + + Shared verbatim by ``sdk pipelines compile`` and ``sdk pipeline-runs + submit-from-python`` so both accept exactly the same syntax and reject the + same mistakes. An empty VALUE is allowed (``--override note=``). + """ + + parsed: dict[str, str] = {} + for value in values or []: + if "=" not in value: + raise SystemExit("--override entries must use KEY=VALUE syntax") + key, parsed_value = value.split("=", 1) + if not key: + raise SystemExit("--override entries must use KEY=VALUE syntax") + parsed[key] = parsed_value + return parsed + + +def parse_image_overrides(values: list[str] | None) -> dict[str, str]: + """Parse repeatable ``--image ID=REF`` compile-time image-id overrides.""" + + parsed: dict[str, str] = {} + for value in values or []: + if "=" not in value: + raise SystemExit("--image entries must use ID=REF syntax") + image_id, image_ref = value.split("=", 1) + if not image_id or not image_ref: + raise SystemExit("--image entries must use ID=REF syntax") + parsed[image_id] = image_ref + return parsed + + def api_arg_specs( *, base_url: str | None = None, diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py index 3df4df4..c607bca 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py @@ -2,11 +2,15 @@ from __future__ import annotations +import difflib import json import os import pathlib +import shutil import sys -from typing import Annotated, Any +import tempfile +from contextlib import ExitStack, contextmanager, suppress +from typing import TYPE_CHECKING, Annotated, Any, Callable, Iterator from cyclopts import App, Parameter @@ -17,6 +21,8 @@ include_env_credentials_for_args, load_args_or_exit, optional_path, + parse_image_overrides, + parse_overrides, print_json, ) from .cli_options import ( @@ -42,6 +48,10 @@ parse_key_value_entries, ) from .pipeline_run_search import normalize_query_input, parse_annotation +from .pipelines import PipelineValidationError, compile_pipeline_file + +if TYPE_CHECKING: # pragma: no cover - typing only + from .pipeline_compiler import CompileResult app = App(name="pipeline-runs", help="Submit and inspect Tangle pipeline runs.") annotations_app = App(name="annotations", help="Work with pipeline-run annotations.") @@ -120,6 +130,56 @@ def _run_manager_action(config: str | None, cli_base_url: str | None, specs: dic finalize_logs() +def _run_prepared_manager_actions( + config: str | None, + cli_base_url: str | None, + specs: dict[str, tuple[Any, ...]], + prepare, + submit, + *, + precheck: Callable[[ArgsContainer], None] | None = None, +): + """Prepare EVERY config entry, then submit the prepared results in order. + + ``prepare(manager, args, artifacts)`` returns one entry's frozen payload and + may register cleanup on the shared ``artifacts`` stack; ``submit`` performs + the single API write. Splitting the phases means a failure while preparing + any entry submits nothing. Artifacts outlive preparation because run + lifecycle hooks read the compiled path during submission. + """ + + loaded = load_args_or_exit(config, **specs) + if precheck is not None: + for args in loaded: + precheck(args) + + with ExitStack() as log_finalizers: + entries: list[tuple[ArgsContainer, PipelineRunManager]] = [] + for args in loaded: + try: + logger, finalize_logs = logger_for_log_type(getattr(args, "log_type", "console")) + except ValueError as exc: + raise SystemExit(str(exc)) from exc + log_finalizers.callback(finalize_logs) + entries.append((args, _manager(args, cli_base_url=cli_base_url, logger=logger))) + + prepared: list[Any] = [] + with ExitStack() as artifacts: + for args, manager in entries: + try: + prepared.append(prepare(manager, args, artifacts)) + except PipelineRunError as exc: + raise SystemExit(str(exc)) from exc + + for (args, manager), payload in zip(entries, prepared): + try: + result = submit(manager, args, payload) + except PipelineRunError as exc: + raise SystemExit(str(exc)) from exc + if result is not None: + print_json(result) + + def _run_annotation_action(config: str | None, cli_base_url: str | None, specs: dict[str, tuple[Any, ...]], fn): for args in load_args_or_exit(config, **specs): try: @@ -247,6 +307,356 @@ def action(manager: PipelineRunManager, args: ArgsContainer) -> dict[str, Any]: _run_manager_action(config, base_url, specs, action) +#: Config fields this command understands. Used only to catch MISSPELLINGS of +#: supported keys (`overide:`), which would otherwise be dropped silently and +#: submit something other than what was authored. Unrelated keys in a shared +#: config file are deliberately left alone. +_SUBMIT_FROM_PYTHON_CONFIG_KEYS = frozenset( + "script_path pipeline override image arg args args_json arg_secret arg_secrets " + "annotation dry_run run_as trusted_source trusted_hydration trusted_hydration_cli " + "submit_recovery_attempts log_type base_url token auth_header header".split() +) + +#: Keys that are valid for other commands but must never be honored here, +#: because silently ignoring them would change what actually gets submitted. +_SUBMIT_FROM_PYTHON_REJECTED_CONFIG_KEYS = { + "hydrate": ( + "submit-from-python always hydrates the compiled bundle; " + "its relative refs are meaningless to the server otherwise" + ), + "pipeline_path": ( + "submit-from-python compiles a Python script; " + "use script_path (and --pipeline to pick the root @pipeline function)" + ), +} + +_LOG_TYPES = ("console", "none", "file") + +#: Dot-prefixed so the throwaway bundle stays hidden next to the user's sources. +_TEMP_BUNDLE_PREFIX = ".tangle-submit-" + + +@contextmanager +def _compiled_bundle( + script: pathlib.Path, + *, + overrides: dict[str, str], + image_overrides: dict[str, str], + pipeline_name: str | None, + logger: Logger, +) -> Iterator[CompileResult]: + """Compile *script* to a throwaway bundle beside it, always removing it. + + The bundle cannot live in a temp directory: the compiler derives every + relative URL from the output path, so author-written refs such as + ``ref(url="file://./component.yaml")`` only resolve when the bundle sits in + the script's own (physically resolved) directory. The name is allocated + with an exclusive create, so concurrent compiles never collide. + """ + + script_path = script.resolve() + try: + handle = tempfile.NamedTemporaryFile( + dir=script_path.parent, prefix=_TEMP_BUNDLE_PREFIX, suffix=".yaml", delete=False + ) + except OSError as exc: + raise PipelineValidationError( + f"Cannot create a compile output in {script_path.parent}: {exc}" + ) from exc + root = pathlib.Path(handle.name) + closed = False + try: + handle.close() + closed = True + result = compile_pipeline_file( + script_path, + root, + overrides=overrides, + pipeline_name=pipeline_name, + image_overrides=image_overrides, + logger=logger, + ) + for warning in result.warnings: + logger.warn(warning) + yield result + finally: + # Cleanup is stem-scoped and must never mask the primary error. + if not closed: + with suppress(Exception): + handle.close() + for path in (root, root.with_name(root.stem + ".components.yaml")): + try: + path.unlink() + except FileNotFoundError: + pass + except OSError as exc: + logger.warn(f"Could not remove compile artifact {path}: {exc}") + shutil.rmtree(root.with_name(root.stem + ".subgraphs"), ignore_errors=True) + + +def _misspelled_config_key(key: str) -> str | None: + """Return the supported field *key* looks like a typo of, if any.""" + + if key.startswith("_") or key in _SUBMIT_FROM_PYTHON_CONFIG_KEYS: + return None + normalized = "".join(char for char in key.lower() if char.isalnum()) + for known in _SUBMIT_FROM_PYTHON_CONFIG_KEYS: + if normalized == known.replace("_", ""): + return known + close = difflib.get_close_matches(key, sorted(_SUBMIT_FROM_PYTHON_CONFIG_KEYS), n=1, cutoff=0.85) + return close[0] if close else None + + +def _resolve_run_arguments(args: ArgsContainer) -> dict[str, Any]: + """Resolve one entry's run arguments: literals plus secret bindings. + + Pure, so the precheck and the preparation phase share one implementation + and an ``--arg``/``--arg-secret`` conflict is caught before any submit. + """ + + run_args = parse_json_or_key_values(args.args_json or args.args_config, args.arg) + secret_names = normalize_arg_secret_config(args.arg_secrets_config) + secret_names.update(parse_arg_secret_entries(args.arg_secret)) + return merge_secret_run_args(run_args, secret_names) + + +def _check_submit_from_python_inputs(args: ArgsContainer) -> None: + """Validate one entry before ANY entry is compiled or submitted. + + Only deterministic, side-effect-free checks belong here: they run for every + config entry up front, so a malformed value in the last entry cannot land + after earlier entries were already submitted. + """ + + config = getattr(args, "_config", {}) + if isinstance(config, dict): + for key, reason in _SUBMIT_FROM_PYTHON_REJECTED_CONFIG_KEYS.items(): + if key in config: + raise SystemExit( + f"Config error: '{key}' is not supported by submit-from-python: {reason}" + ) + for key in config: + if (suggestion := _misspelled_config_key(str(key))) is not None: + raise SystemExit( + f"Config error: unknown submit-from-python field '{key}'; " + f"did you mean '{suggestion}'?" + ) + + # Safety-sensitive values reach retry/trust logic, where a wrong type is + # worse than a rejected config: a non-int recovery budget crashes an + # ambiguous submit before its recovery lookup, and a truthy string silently + # enables allow-all hydration. + attempts = getattr(args, "submit_recovery_attempts", None) + if isinstance(attempts, bool) or not isinstance(attempts, int) or attempts < 0: + raise SystemExit("Config error: submit_recovery_attempts must be a non-negative integer") + trusted_hydration = getattr(args, "trusted_hydration_cli", None) + if trusted_hydration is not None and not isinstance(trusted_hydration, bool): + raise SystemExit("Config error: trusted_hydration_cli must be a boolean") + + if getattr(args, "log_type", "console") not in _LOG_TYPES: + raise SystemExit(f"--log-type must be one of: {', '.join(_LOG_TYPES)}") + + try: + parse_overrides(args.override) + parse_image_overrides(args.image) + _resolve_run_arguments(args) + parse_key_value_entries(args.annotation) + except PipelineRunError as exc: + raise SystemExit(str(exc)) from exc + + if args.script_path is not None and not pathlib.Path(args.script_path).exists(): + raise SystemExit(f"Pipeline script not found: {args.script_path}") + + +@app.command(name="submit-from-python") +def pipeline_runs_submit_from_python( + script_path: pathlib.Path | None = None, + *, + pipeline: Annotated[ + str | None, + Parameter( + name="--pipeline", + help="Select the root @pipeline function by name when the file defines several.", + ), + ] = None, + override: Annotated[ + list[str] | None, + Parameter( + name="--override", + help="Compile-time config override as KEY=VALUE. Repeat for multiple.", + negative_iterable=(), + ), + ] = None, + image: Annotated[ + list[str] | None, + Parameter( + name="--image", + help=( + "Compile-time image-id override as ID=REF for @task(image_id=ID). " + "Repeat for multiple IDs." + ), + negative_iterable=(), + ), + ] = None, + arg: Annotated[ + list[str] | None, + Parameter(help="Pipeline argument as KEY=VALUE. Repeat for multiple.", negative_iterable=()), + ] = None, + args_json: Annotated[str | None, Parameter(help="Pipeline arguments as a JSON object.")] = None, + arg_secret: Annotated[ + list[str] | None, + Parameter( + name="--arg-secret", + help=( + "Pipeline argument bound to a Tangle secret as INPUT=SECRET_NAME. " + "Repeat for multiple." + ), + negative_iterable=(), + ), + ] = None, + annotation: Annotated[ + list[str] | None, + Parameter(help="Run annotation as KEY=VALUE. Repeat for multiple.", negative_iterable=()), + ] = None, + dry_run: Annotated[ + bool | None, + Parameter(help="Compile, hydrate and print the submit payload without creating a run."), + ] = None, + run_as: Annotated[ + str | None, + Parameter(help="Downstream extension point; unsupported by the OSS default hooks."), + ] = None, + trusted_source: Annotated[ + list[str] | None, + Parameter( + name="--trusted-source", + help="Trusted local_from_python source root or glob. Repeat for multiple.", + negative_iterable=(), + ), + ] = None, + trusted_hydration: Annotated[ + bool | None, + Parameter( + name="--trusted-hydration", + help="Allow all local_from_python execution during hydration for trusted inputs.", + ), + ] = None, + base_url: BaseUrlOption = None, + token: TokenOption = None, + auth_header: AuthHeaderOption = None, + header: HeaderOption = None, + config: ConfigOption = None, + submit_recovery_attempts: Annotated[ + int, + Parameter( + help=( + "Number of post-failed-submit recovery lookups before resubmitting; " + "higher values wait longer for delayed run registration." + ) + ), + ] = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, + log_type: LogTypeOption = "console", +) -> None: + """Compile a Python-authored pipeline and submit it as a run. + + Equivalent to ``pipelines compile`` followed by ``pipeline-runs submit``, + without leaving compiled YAML behind. The bundle is compiled next to the + script (so relative ``file://`` refs keep resolving), hydrated, submitted, + and then removed — also on dry runs and failures. Use ``pipelines compile`` + when the compiled YAML itself is what you want. + + Compile-time values and run-time values are distinct: ``--override + KEY=VALUE`` sets ``cfg`` values consumed while the graph is built and + ``--image ID=REF`` resolves ``@task(image_id=ID)``, while ``--arg`` / + ``--args-json`` / ``--arg-secret`` set pipeline arguments for the run. + + Hydration is always on: the compiled bundle's refs point at local sidecars + the server cannot read. Submission never waits; use ``pipeline-runs wait``. + + A multi-entry ``--config`` prepares EVERY entry first — compile once, + hydrate, merge run arguments and secrets, validate, freeze the submit body + — and only then submits the frozen bodies in order. A failure while + preparing any entry therefore creates no runs and leaves no artifacts. + """ + + specs = { + "script_path": ("script_path", script_path, None, False, True, optional_path), + "pipeline": (pipeline, None), + "override": (override, None), + "image": (image, None), + "arg": (arg, None), + "args_json": (args_json, None), + "args_config": ("args", None, None, True), + "arg_secret": (arg_secret, None), + "arg_secrets_config": ("arg_secrets", None, None, True), + "annotation": (annotation, None), + "dry_run": (dry_run, None), + "run_as": (run_as, None), + "trusted_source": (trusted_source, None), + "trusted_hydration_cli": ("trusted_hydration_cli", trusted_hydration, None, False), + "submit_recovery_attempts": (submit_recovery_attempts, _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS), + "log_type": (log_type, "console"), + **api_arg_specs(base_url=base_url, token=token, auth_header=auth_header, header=header), + } + + def prepare( + manager: PipelineRunManager, args: ArgsContainer, artifacts: ExitStack + ) -> tuple[dict[str, Any], pathlib.Path]: + """Compile once, hydrate, validate, and freeze this entry's submit body. + + The compiled path travels with the body so run lifecycle hooks keep the + path-backed context they had when submission happened inside the + bundle; that is why the bundle outlives preparation. + """ + + try: + compiled = artifacts.enter_context( + _compiled_bundle( + args.script_path, + overrides=parse_overrides(args.override), + image_overrides=parse_image_overrides(args.image), + pipeline_name=args.pipeline, + logger=manager.logger, + ) + ) + body = manager.build_submit_body( + compiled.pipeline_path, + run_args=_resolve_run_arguments(args), + annotations=parse_key_value_entries(args.annotation), + hydrate=True, + run_as=args.run_as, + ) + return body, compiled.pipeline_path + except PipelineValidationError as exc: + raise SystemExit(str(exc)) from exc + + def submit( + manager: PipelineRunManager, + args: ArgsContainer, + prepared: tuple[dict[str, Any], pathlib.Path], + ) -> dict[str, Any]: + body, pipeline_path = prepared + if args.dry_run: + return body + # The body is already hydrated and validated, so the run is created + # without recompiling; submission-id stamping and the ambiguous-submit + # recovery lookup still happen inside the run lifecycle. + return manager.run_prepared_body( + body, + pipeline_path=pipeline_path, + submit_recovery_attempts=args.submit_recovery_attempts, + )["response"] + + _run_prepared_manager_actions( + config, + base_url, + specs, + prepare, + submit, + precheck=_check_submit_from_python_inputs, + ) + + @app.command(name="details") def pipeline_runs_details( run_id: str | None = None, diff --git a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py index 4ce9006..533fb5f 100644 --- a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py @@ -7,7 +7,13 @@ from cyclopts import App, Parameter -from .cli_helpers import LazyTangleApiClient, load_config_or_exit, optional_path +from .cli_helpers import ( + LazyTangleApiClient, + load_config_or_exit, + optional_path, + parse_image_overrides, + parse_overrides, +) from .cli_options import ( AuthHeaderOption, BaseUrlOption, @@ -126,30 +132,6 @@ def _parse_vars(values: list[str] | dict[str, object] | None) -> dict[str, str]: return parsed -def _parse_overrides(values: list[str] | None) -> dict[str, str]: - parsed: dict[str, str] = {} - for value in values or []: - if "=" not in value: - raise SystemExit("--override entries must use KEY=VALUE syntax") - key, parsed_value = value.split("=", 1) - if not key: - raise SystemExit("--override entries must use KEY=VALUE syntax") - parsed[key] = parsed_value - return parsed - - -def _parse_image_overrides(values: list[str] | None) -> dict[str, str]: - parsed: dict[str, str] = {} - for value in values or []: - if "=" not in value: - raise SystemExit("--image entries must use ID=REF syntax") - image_id, image_ref = value.split("=", 1) - if not image_id or not image_ref: - raise SystemExit("--image entries must use ID=REF syntax") - parsed[image_id] = image_ref - return parsed - - @app.command(name="hydrate") def pipelines_hydrate( pipeline_path: pathlib.Path, @@ -299,8 +281,8 @@ def pipelines_compile( result = compile_pipeline_file( pipeline_path, output, - overrides=_parse_overrides(override), - image_overrides=_parse_image_overrides(image), + overrides=parse_overrides(override), + image_overrides=parse_image_overrides(image), pipeline_name=pipeline, logger=logger, ) diff --git a/pyproject.toml b/pyproject.toml index 9e51c5c..d30d701 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.11" +version = "0.1.12" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/skills/tangent/OSS-CONVENTIONS.md b/skills/tangent/OSS-CONVENTIONS.md index 4576a73..bc64315 100644 --- a/skills/tangent/OSS-CONVENTIONS.md +++ b/skills/tangent/OSS-CONVENTIONS.md @@ -84,6 +84,7 @@ OSS replacement. **Verbs/flags below were verified against the `tangle-cli` sour | ` pipeline-run submit p.yaml -f c.yaml --hydrate --no-wait` | `tangle sdk pipeline-runs submit p.yaml [--arg K=V \| --args-json JSON] [--annotation K=V]` — **hydrate is the default; there is NO `--no-wait` (submit never waits); there is NO `-f config.yaml` (use `--arg`/`--args-json`, or `--config` for CLI-option defaults)** | | `… submit … (submit-as-is, no version resolution)` | `tangle sdk pipeline-runs submit p.yaml --no-hydrate` | | `… submit … --dry-run` (preview payload) | `tangle sdk pipeline-runs submit p.yaml --dry-run` (prints the submit body, creates no run) | +| `… ship/submit a Python-authored pipeline` | `tangle sdk pipeline-runs submit-from-python p.py [--pipeline NAME] [--override K=V] [--image ID=REF]` plus the usual run flags — compiles, hydrates (always), submits, then deletes the compiled bundle; never waits (use `pipelines compile` to keep the YAML) | | ` pipeline-run details RUN_ID --state` | `tangle sdk pipeline-runs details RUN_ID --include-execution-state` | | `… details … --implementations` | `… details … --include-implementations` | | `… details … --include-annotations` | `… details … --include-annotations` (unchanged) | diff --git a/skills/tangent/references/tangle-tools.md b/skills/tangent/references/tangle-tools.md index 485cc3c..5f310f7 100644 --- a/skills/tangent/references/tangle-tools.md +++ b/skills/tangent/references/tangle-tools.md @@ -157,6 +157,29 @@ tangle sdk pipeline-runs submit pipeline.yaml \ --arg K=V --annotation session=YYYY-MM-DD-scenario ``` +### Submitting a Python-authored pipeline directly + +`submit-from-python` compiles a Python pipeline script, hydrates it, submits the +run, and removes the compiled YAML again (also on `--dry-run` and on failure): + +```bash +tangle sdk pipeline-runs submit-from-python pipeline.py \ + --override batch_size=100 \ + --arg model=baseline --annotation session=YYYY-MM-DD-scenario +``` + +- Compile-tier flags: `--pipeline NAME` (pick the root `@pipeline` when the file + defines several), `--override KEY=VALUE` (`cfg` values), `--image ID=REF` + (`@task(image_id=…)`). The bundle is always removed; use `pipelines compile` + when you want to keep the YAML. +- Run-tier flags are identical to `submit` (`--arg`, `--args-json`, + `--arg-secret`, `--annotation`, `--dry-run`, auth/config/log options). +- Hydration is forced (no `--no-hydrate`) and submit still never waits. +- A multi-entry `--config` prepares every entry (compile once, hydrate, merge + args/secrets, validate, freeze the body) before creating any run, so a bad + value or compile error in the last entry submits nothing. +- Do not confuse `--override` (compile time) with `--arg` (run time). + ## Validating & Editing Pipelines Local pipeline operations live under `pipelines` (NOT `pipeline-runs`): diff --git a/tests/test_packaging.py b/tests/test_packaging.py index b6d4295..345fc66 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -183,7 +183,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.11" in metadata + assert "Version: 0.1.12" in metadata assert "Requires-Dist: tangle-api==0.1.1" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py index ae03a19..165a266 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -5,7 +5,9 @@ import json import math import os +import shutil import sys +import textwrap import time from contextlib import nullcontext from pathlib import Path @@ -3945,3 +3947,790 @@ def iter_execution_container_log_lines(self, id: str): message = str(exc_info.value) assert "404" in message assert "/api/executions/exec-1/stream_container_log" in message + + +# --------------------------------------------------------------------------- +# submit-from-python: compile + hydrate + submit in one step. +# +# The adversarial surface for this command is artifact lifecycle (never leave +# compiled YAML behind, never delete anything else), artifact locality (the +# bundle must compile next to the script or relative refs break), and config +# handling (a silently dropped config key would submit something the author +# did not author). + +COMPILE_FIXTURES = Path(__file__).parent / "fixtures" / "python_pipeline" + +_REF_PIPELINE_SOURCE = """ +from tangle_cli.python_pipeline import Out, pipeline, ref + + +@pipeline("Noop Pipeline") +def noop_pipeline(cfg) -> Out[str]: + return ref(url="file://./noop.yaml").named(cfg.task_id)() +""" + +_INPUT_PIPELINE_SOURCE = """ +from tangle_cli.python_pipeline import In, Out, pipeline, ref + + +@pipeline("Input Pipeline") +def input_pipeline(parent_wait_token: In[str], cfg) -> Out[str]: + return ref(url="file://./noop.yaml").named(cfg.task_id)(wait_for=parent_wait_token) +""" + +_TWO_PIPELINE_SOURCE = """ +from tangle_cli.python_pipeline import Out, pipeline, ref + + +@pipeline("First Pipeline") +def first_pipeline(cfg) -> Out[str]: + return ref(url="file://./noop.yaml").named("first")() + + +@pipeline("Second Pipeline") +def second_pipeline(cfg) -> Out[str]: + return ref(url="file://./noop.yaml").named("second")() +""" + + +def _python_project( + tmp_path: Path, + source: str = _REF_PIPELINE_SOURCE, + *, + name: str = "pipeline.py", + cfg: str = "task_id: noop-task\n", +) -> Path: + """Write a compilable project: script + config.yaml + referenced noop.yaml.""" + + project = tmp_path / "project" + project.mkdir(parents=True, exist_ok=True) + shutil.copy(COMPILE_FIXTURES / "noop.yaml", project / "noop.yaml") + (project / "config.yaml").write_text(cfg, encoding="utf-8") + script = project / name + script.write_text(textwrap.dedent(source), encoding="utf-8") + return script + + +def _entries(directory: Path) -> set[str]: + """Directory contents, ignoring the interpreter's own bytecode cache.""" + + return {entry.name for entry in directory.iterdir() if entry.name != "__pycache__"} + + +def _submitted_tasks(body: dict[str, Any]) -> dict[str, Any]: + return body["root_task"]["componentRef"]["spec"]["implementation"]["graph"]["tasks"] + + +def test_submit_from_python_is_registered_with_compile_and_run_flags(capsys): + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "--help"]) + assert "submit-from-python" in capsys.readouterr().out + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", "--help"]) + help_text = capsys.readouterr().out + for flag in ( + "--pipeline", + "--override", + "--image", + "--arg", + "--args-json", + "--arg-secret", + "--annotation", + "--dry-run", + "--run-as", + "--trusted-source", + "--trusted-hydration", + "--submit-recovery-attempts", + "--config", + "--log-type", + ): + assert flag in help_text + # Hydration is forced (compiled refs are local-only) and submit never waits. + assert "--no-hydrate" not in help_text + assert "--wait" not in help_text + + +def test_submit_from_python_compiles_hydrates_submits_and_cleans_up(monkeypatch, tmp_path: Path, capsys): + script = _python_project(tmp_path) + before = _entries(script.parent) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", str(script), "--log-type", "none"]) + + assert json.loads(capsys.readouterr().out) == {"id": "run-1", "root_execution_id": "exec-1"} + body = fake_client.created[0] + assert body["root_task"]["componentRef"]["spec"]["name"] == "Noop Pipeline" + # Forced hydration inlined the sibling component instead of shipping the + # server a local file:// URL it could never read. + task = _submitted_tasks(body)["noop-task"] + assert task["componentRef"]["spec"]["name"] == "Noop" + assert "url" not in task["componentRef"] + # Nothing compiled is left behind. + assert _entries(script.parent) == before + + +def test_submit_from_python_cleanup_is_stem_scoped_and_keeps_neighbors( + monkeypatch, tmp_path: Path, capsys +): + script = _python_project(tmp_path) + # A sibling bundle that merely looks like compiler output must survive. + decoy = script.parent / "pipeline.yaml" + decoy.write_text("name: hand written\n", encoding="utf-8") + decoy_sidecar = script.parent / "pipeline.components.yaml" + decoy_sidecar.write_text("noop: {}\n", encoding="utf-8") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", str(script), "--log-type", "none"]) + + capsys.readouterr() + assert decoy.read_text(encoding="utf-8") == "name: hand written\n" + assert decoy_sidecar.read_text(encoding="utf-8") == "noop: {}\n" + assert len(fake_client.created) == 1 + + +def test_submit_from_python_removes_artifacts_when_submit_fails(monkeypatch, tmp_path: Path): + script = _python_project(tmp_path) + before = _entries(script.parent) + + class FailingClient(FakeClient): + def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: + self.created.append(body) + raise ValueError("submit exploded") + + def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: + # No run was registered, so post-failure recovery finds nothing and + # the original submit error is what reaches the caller. + self.list_calls.append(kwargs) + return {"pipeline_runs": [], "next_page_token": None} + + fake_client = FailingClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + # The transport error propagates exactly as it does for `submit`; what this + # command adds is that the compiled bundle is still gone afterwards. + with pytest.raises(ValueError, match="submit exploded"): + app(["sdk", "pipeline-runs", "submit-from-python", str(script), "--log-type", "none"]) + + assert _entries(script.parent) == before + + +def test_submit_from_python_compile_failure_exits_without_submitting_or_leftovers( + monkeypatch, tmp_path: Path +): + script = _python_project( + tmp_path, + """ + from tangle_cli.python_pipeline import ref + + some_ref = ref(url="file://./noop.yaml") + """, + ) + before = _entries(script.parent) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", str(script), "--log-type", "none"]) + + assert exc_info.value.code != 0 + assert "no @pipeline" in str(exc_info.value).lower() + assert fake_client.created == [] + assert _entries(script.parent) == before + + +def test_submit_from_python_dry_run_prints_body_without_creating_a_run( + monkeypatch, tmp_path: Path, capsys +): + script = _python_project(tmp_path) + before = _entries(script.parent) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit-from-python", + str(script), + "--dry-run", + "--log-type", + "none", + ], + ) + + body = json.loads(capsys.readouterr().out) + assert body["root_task"]["componentRef"]["spec"]["name"] == "Noop Pipeline" + assert fake_client.created == [] + assert _entries(script.parent) == before + + +def test_submit_from_python_compile_overrides_and_pipeline_selection_reach_the_compiler( + monkeypatch, tmp_path: Path, capsys +): + script = _python_project(tmp_path) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit-from-python", + str(script), + "--override", + "task_id=overridden-task", + "--log-type", + "none", + ], + ) + + capsys.readouterr() + assert set(_submitted_tasks(fake_client.created[0])) == {"overridden-task"} + + multi = _python_project(tmp_path, _TWO_PIPELINE_SOURCE, name="multi.py") + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit-from-python", + str(multi), + "--pipeline", + "second_pipeline", + "--log-type", + "none", + ], + ) + + capsys.readouterr() + assert fake_client.created[1]["root_task"]["componentRef"]["spec"]["name"] == "Second Pipeline" + + +def test_submit_from_python_image_override_reaches_the_compiled_bundle( + monkeypatch, tmp_path: Path, capsys +): + script = _python_project( + tmp_path, + """ + from tangle_cli.python_pipeline import Out, pipeline, task + + + @task(image_id="eval-slim") + def scored(greeting: str = "hello"): + print(greeting) + + + @pipeline("Image Pipeline") + def image_pipeline(cfg) -> Out[str]: + run_scored = scored() + return run_scored + """, + name="image_pipeline.py", + ) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit-from-python", + str(script), + "--image", + "eval-slim=registry.example/eval-slim@sha256:abc", + "--trusted-hydration", + "--log-type", + "none", + ], + ) + + capsys.readouterr() + # The bundle is gone, so the override has to be visible in what was + # actually submitted: hydration inlines the component sidecar. + submitted = json.dumps(fake_client.created[0]) + assert "registry.example/eval-slim@sha256:abc" in submitted + # (`generated/` is the compiler's own python-task output, not bundle state.) + assert [name for name in _entries(script.parent) if name.startswith(".tangle-submit-")] == [] + + +def test_submit_from_python_run_flags_reach_the_submit_body(monkeypatch, tmp_path: Path, capsys): + script = _python_project(tmp_path, _INPUT_PIPELINE_SOURCE, name="input_pipeline.py") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit-from-python", + str(script), + "--arg", + "parent_wait_token=token", + "--annotation", + "team=oss", + "--log-type", + "none", + ], + ) + + capsys.readouterr() + body = fake_client.created[0] + assert body["root_task"]["arguments"] == {"parent_wait_token": "token"} + assert body["annotations"]["team"] == "oss" + + +def test_submit_from_python_multi_config_compiles_and_submits_each_entry( + monkeypatch, tmp_path: Path, capsys +): + script = _python_project(tmp_path) + before = _entries(script.parent) + config = tmp_path / "submit.yaml" + config.write_text( + yaml.safe_dump( + { + "_defaults": {"script_path": str(script), "log_type": "none"}, + "configs": [ + {"override": ["task_id=first-task"], "annotation": ["run=one"]}, + {"override": ["task_id=second-task"], "annotation": ["run=two"]}, + ], + }, + sort_keys=False, + ), + encoding="utf-8", + ) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + capsys.readouterr() + assert len(fake_client.created) == 2 + assert set(_submitted_tasks(fake_client.created[0])) == {"first-task"} + assert set(_submitted_tasks(fake_client.created[1])) == {"second-task"} + assert fake_client.created[0]["annotations"]["run"] == "one" + assert fake_client.created[1]["annotations"]["run"] == "two" + assert _entries(script.parent) == before + + +@pytest.mark.parametrize( + "key, value, expected", + [ + ("hydrate", False, "always hydrates"), + ("pipeline_path", "pipeline.yaml", "use script_path"), + ], +) +def test_submit_from_python_config_rejects_unsupported_keys( + monkeypatch, tmp_path: Path, key: str, value: Any, expected: str +): + script = _python_project(tmp_path) + config = tmp_path / "submit.yaml" + config.write_text( + yaml.safe_dump({"script_path": str(script), "log_type": "none", key: value}), + encoding="utf-8", + ) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + assert expected in str(exc_info.value) + assert fake_client.created == [] + + +def test_submit_from_python_uses_selected_config_branch(monkeypatch, tmp_path: Path, capsys): + script = _python_project(tmp_path) + config = tmp_path / "submit.yaml" + config.write_text( + yaml.safe_dump( + { + "_select": { + "env": "TANGLE_ENV", + "cases": { + "prod": { + "script_path": str(script), + "log_type": "none", + "override": ["task_id=prod-task"], + }, + "dev": { + "script_path": str(script), + "log_type": "none", + "override": ["task_id=dev-task"], + }, + }, + } + }, + sort_keys=False, + ), + encoding="utf-8", + ) + monkeypatch.setenv("TANGLE_ENV", "dev") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + capsys.readouterr() + assert set(_submitted_tasks(fake_client.created[0])) == {"dev-task"} + + +def test_submit_from_python_requires_a_script(monkeypatch): + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient()) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", "--log-type", "none"]) + + assert "script_path is required" in str(exc_info.value) + + +def _multi_config(tmp_path: Path, script: Path, second: dict[str, Any]) -> Path: + config = tmp_path / "submit.yaml" + config.write_text( + yaml.safe_dump( + { + "_defaults": {"log_type": "none"}, + "configs": [ + {"script_path": str(script), "override": ["task_id=first-task"]}, + {"script_path": str(script), **second}, + ], + }, + sort_keys=False, + ), + encoding="utf-8", + ) + return config + + +@pytest.mark.parametrize( + "second, expected", + [ + ({"override": ["malformed"]}, "--override entries must use KEY=VALUE syntax"), + ({"image": ["eval-slim="]}, "--image entries must use ID=REF syntax"), + ({"arg_secret": ["broken"]}, "INPUT=SECRET"), + # An input given as BOTH a literal and a secret reference is only + # detected once the two are merged. + ( + {"arg": ["api_key=value"], "arg_secret": ["api_key=SECRET"]}, + "both a value and a secret reference", + ), + ({"log_type": "verbose"}, "--log-type must be one of"), + ({"script_path": "/nonexistent/pipeline.py"}, "Pipeline script not found"), + ], +) +def test_submit_from_python_late_entry_input_errors_block_every_submit( + monkeypatch, tmp_path: Path, second: dict[str, Any], expected: str +): + """A bad value in the LAST entry must not land after the first has submitted.""" + + script = _python_project(tmp_path) + config = _multi_config(tmp_path, script, second) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + assert expected in str(exc_info.value) + assert fake_client.created == [] + + +def test_submit_from_python_late_compile_error_blocks_every_submit( + monkeypatch, tmp_path: Path, capsys +): + """Every entry is compiled/hydrated/frozen before the first run is created. + + A compile failure in the LAST entry must therefore create no runs at all, + not just fail after the earlier entries were already submitted. + """ + + script = _python_project(tmp_path) + before = _entries(script.parent) + config = _multi_config(tmp_path, script, {"pipeline": "does_not_exist"}) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + capsys.readouterr() + assert "does_not_exist" in str(exc_info.value) + assert fake_client.created == [] + assert _entries(script.parent) == before + + +def test_submit_from_python_prepares_every_entry_before_the_first_submit( + monkeypatch, tmp_path: Path, capsys +): + """Preparation of all entries strictly precedes the first API write.""" + + script = _python_project(tmp_path) + config = _multi_config(tmp_path, script, {"override": ["task_id=second-task"]}) + events: list[str] = [] + real_build_submit_body = PipelineRunManager.build_submit_body + + def spy_build_submit_body(self, pipeline_path, **kwargs): + events.append("prepare") + return real_build_submit_body(self, pipeline_path, **kwargs) + + class RecordingClient(FakeClient): + def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: + events.append("submit") + return super().pipeline_runs_create(body) + + monkeypatch.setattr(PipelineRunManager, "build_submit_body", spy_build_submit_body) + fake_client = RecordingClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + capsys.readouterr() + assert events == ["prepare", "prepare", "submit", "submit"] + assert set(_submitted_tasks(fake_client.created[0])) == {"first-task"} + assert set(_submitted_tasks(fake_client.created[1])) == {"second-task"} + + +def test_submit_from_python_accepts_a_symlinked_script(monkeypatch, tmp_path: Path, capsys): + script = _python_project(tmp_path) + links = tmp_path / "links" + links.mkdir() + alias = links / "pipeline.py" + alias.symlink_to(script) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", str(alias), "--log-type", "none"]) + + capsys.readouterr() + assert fake_client.created[0]["root_task"]["componentRef"]["spec"]["name"] == "Noop Pipeline" + assert _entries(links) == {"pipeline.py"} + + +class _PathRecordingHooks(PipelineRunHooks): + """Records what every run-lifecycle callback sees as ``pipeline_path``.""" + + seen: list[tuple[str, Path | None, bool]] = [] + + @classmethod + def _record(cls, stage: str, context: Any) -> None: + raw = getattr(context, "pipeline_path", None) + path = Path(raw) if raw else None + cls.seen.append((stage, path, bool(path and path.exists()))) + + def before_run_lifecycle(self, context): + self._record("before", context) + return super().before_run_lifecycle(context) + + def around_run(self, context): + self._record("around", context) + return super().around_run(context) + + def before_submit_context(self, context): + self._record("before_submit", context) + return super().before_submit_context(context) + + def after_submit_context(self, context): + self._record("after_submit", context) + return super().after_submit_context(context) + + def on_submit_error(self, error, *, context): + self._record("submit_error", context) + return super().on_submit_error(error, context=context) + + def after_run_lifecycle(self, context, *, success, error=None): + self._record("after", context) + return super().after_run_lifecycle(context, success=success, error=error) + + +@pytest.fixture +def path_recording_hooks(monkeypatch): + _PathRecordingHooks.seen = [] + monkeypatch.setattr(pipeline_runs_cli, "PipelineRunHooks", _PathRecordingHooks) + return _PathRecordingHooks + + +def test_submit_from_python_removes_the_bundle_when_closing_it_fails(monkeypatch, tmp_path: Path): + """The temp file exists from allocation on, so even a failing close cleans up.""" + + import tempfile as tempfile_module + + script = _python_project(tmp_path) + before = _entries(script.parent) + real_factory = tempfile_module.NamedTemporaryFile + created: list[str] = [] + + class ClosesBadly: + def __init__(self, handle): + self._handle = handle + self.name = handle.name + created.append(handle.name) + + def close(self): + self._handle.close() + raise OSError("close failed") + + monkeypatch.setattr( + tempfile_module, "NamedTemporaryFile", lambda *a, **k: ClosesBadly(real_factory(*a, **k)) + ) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient()) + app = cli.build_app() + + with pytest.raises(OSError, match="close failed"): + app(["sdk", "pipeline-runs", "submit-from-python", str(script), "--log-type", "none"]) + + assert not Path(created[0]).exists() + assert _entries(script.parent) == before + + +def test_submit_from_python_hooks_see_an_existing_bundle_for_every_entry( + monkeypatch, tmp_path: Path, capsys, path_recording_hooks +): + """Freezing the body must not strip the path-backed lifecycle context. + + Submission used to run inside ``compiled_bundle``, so downstream hooks + (mutex, notification, source policy) could read ``context.pipeline_path`` + off disk. Preparing all entries first must preserve that, for the LAST + entry as much as the first. + """ + + script = _python_project(tmp_path) + before = _entries(script.parent) + config = _multi_config(tmp_path, script, {"override": ["task_id=second-task"]}) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient()) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + capsys.readouterr() + stages = {stage for stage, _, _ in path_recording_hooks.seen} + assert stages == {"before", "around", "before_submit", "after_submit", "after"} + missing = [ + (stage, path) for stage, path, existed in path_recording_hooks.seen if not existed + ] + assert missing == [], f"hooks saw missing pipeline_path values: {missing}" + observed = {path for _, path, _ in path_recording_hooks.seen} + # Two entries -> two distinct temp bundles, each visible to its own run, + # each hidden and beside the script so relative refs resolved. + assert len(observed) == 2 + assert all(path.name.startswith(".tangle-submit-") for path in observed) + assert all(path.parent == script.parent for path in observed) + # ...and every one of them is cleaned up once the command returns. + assert _entries(script.parent) == before + + +def test_submit_from_python_hooks_see_an_existing_bundle_when_submit_fails( + monkeypatch, tmp_path: Path, path_recording_hooks +): + script = _python_project(tmp_path) + before = _entries(script.parent) + + class FailingClient(FakeClient): + def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: + self.created.append(body) + raise ValueError("submit exploded") + + def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: + self.list_calls.append(kwargs) + return {"pipeline_runs": [], "next_page_token": None} + + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FailingClient()) + app = cli.build_app() + + with pytest.raises(ValueError, match="submit exploded"): + app(["sdk", "pipeline-runs", "submit-from-python", str(script), "--log-type", "none"]) + + # The error callbacks are as entitled to a readable bundle as the happy path. + assert [stage for stage, _, _ in path_recording_hooks.seen][-1] == "after" + assert all(existed for _, _, existed in path_recording_hooks.seen) + assert _entries(script.parent) == before + + +def test_submit_from_python_rejects_a_misspelled_config_key_in_a_later_entry( + monkeypatch, tmp_path: Path +): + """A dropped `overide:` would submit the ORIGINAL pipeline, silently.""" + + script = _python_project(tmp_path) + config = _multi_config(tmp_path, script, {"overide": ["task_id=intended"]}) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + assert "did you mean 'override'" in str(exc_info.value) + assert fake_client.created == [] + + +@pytest.mark.parametrize( + "entry, expected", + [ + # A non-int budget crashes an ambiguous submit before its recovery + # lookup, so a possibly-created run is never reconciled. + ({"submit_recovery_attempts": "oops"}, "submit_recovery_attempts must be"), + ({"submit_recovery_attempts": -1}, "submit_recovery_attempts must be"), + # bool("false") is True, which would silently allow all hydration. + ({"trusted_hydration_cli": "false"}, "trusted_hydration_cli must be a boolean"), + ], +) +def test_submit_from_python_rejects_unsafe_config_types( + monkeypatch, tmp_path: Path, entry: dict[str, Any], expected: str +): + script = _python_project(tmp_path) + config = tmp_path / "submit.yaml" + config.write_text( + yaml.safe_dump({"script_path": str(script), "log_type": "none", **entry}), + encoding="utf-8", + ) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", "--config", str(config)]) + + assert expected in str(exc_info.value) + assert fake_client.created == [] + + +def test_submit_from_python_reports_an_unwritable_script_directory(monkeypatch, tmp_path: Path): + """Allocation failures use the command's error contract, not a traceback.""" + + import tempfile as tempfile_module + + script = _python_project(tmp_path) + + def refuse(*args: Any, **kwargs: Any): + raise PermissionError("read-only file system") + + monkeypatch.setattr(tempfile_module, "NamedTemporaryFile", refuse) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient()) + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app(["sdk", "pipeline-runs", "submit-from-python", str(script), "--log-type", "none"]) + + assert "Cannot create a compile output" in str(exc_info.value) diff --git a/tests/test_pipelines_cli.py b/tests/test_pipelines_cli.py index 7b3378a..c9f28f7 100644 --- a/tests/test_pipelines_cli.py +++ b/tests/test_pipelines_cli.py @@ -1967,3 +1967,71 @@ def test_pipelines_layout_preserves_tasks_and_updates_coordinates(tmp_path: Path load_position = json.loads(updated_tasks["load"]["annotations"]["editor.position"]) assert extract_position == {"x": 0, "y": 0} assert load_position["x"] > extract_position["x"] + + +# --------------------------------------------------------------------------- +# Shared compile-flag parsers. +# +# ``--override`` / ``--image`` parsing moved to cli_helpers so +# ``pipeline-runs submit-from-python`` reuses it instead of copying it. The +# CLI-visible behavior of ``pipelines compile`` must be unchanged. + + +def test_compile_cli_rejects_malformed_override(tmp_path: Path): + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app( + [ + "sdk", + "pipelines", + "compile", + str(tmp_path / "pipeline.py"), + "-o", + str(tmp_path / "out.yaml"), + "--override", + "not-a-pair", + ] + ) + + assert "--override entries must use KEY=VALUE syntax" in str(exc_info.value) + + +def test_compile_cli_rejects_malformed_image(tmp_path: Path): + app = cli.build_app() + + with pytest.raises(SystemExit) as exc_info: + app( + [ + "sdk", + "pipelines", + "compile", + str(tmp_path / "pipeline.py"), + "-o", + str(tmp_path / "out.yaml"), + "--image", + "eval-slim=", + ] + ) + + assert "--image entries must use ID=REF syntax" in str(exc_info.value) + + +def test_shared_compile_flag_parsers(): + from tangle_cli.cli_helpers import parse_image_overrides, parse_overrides + + # An empty override value is meaningful ("set to empty string"); an empty + # image ref is not. + assert parse_overrides(["a=1", "b=x=y", "c="]) == {"a": "1", "b": "x=y", "c": ""} + assert parse_image_overrides(["eval-slim=registry.example/img@sha256:abc"]) == { + "eval-slim": "registry.example/img@sha256:abc" + } + assert parse_overrides(None) == {} + assert parse_image_overrides(None) == {} + + for bad in (["nope"], ["=1"]): + with pytest.raises(SystemExit): + parse_overrides(bad) + for bad in (["nope"], ["id="], ["=ref"]): + with pytest.raises(SystemExit): + parse_image_overrides(bad) diff --git a/uv.lock b/uv.lock index 1a009be..c495089 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.11" +version = "0.1.12" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },