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" },