From 93046e6db102f83193379d3a4a030781e24cd26d Mon Sep 17 00:00:00 2001 From: Volv G Date: Sat, 12 Sep 2026 06:23:44 -0700 Subject: [PATCH] Reject undeclared task arguments at validation time (v0.1.13) A task argument that no component input declares was accepted everywhere in this repo and then rejected by the Tangle UI validator, which blocks cloning or editing the stored run. Execution tolerates the extra key -- nothing binds it, so the pipeline runs on schedule for years -- and that is exactly why the defect survives: it only surfaces when a human opens the run in the UI, far from the commit that introduced it. This was a fail-open bug, not missing information. `pipeline_runner.py` hydrates (~343) and validates (~402), so `componentRef.spec` is inlined and readable by the time `validate_pipeline_for_run` runs. But `_validate_task_inputs` only iterated declared inputs -> arguments. It never computed the reverse set difference, so extra and misspelled keys passed silently even post-hydration. On a hydrated spec declaring `run_id` and `merchant_match_reference_snapshot_date`, a task passing a dead `snapshot_date` plus an invented `totally_made_up_arg` returned `validate_component_inputs(spec) -> []`. Compute `set(arguments) - set(declared_inputs)` beside the existing required-input loop and report each extra key as an error, mirroring the rule the compiler already enforces for `@pipeline` subpipeline children in `_validate_subpipeline_inputs`. This fails closed at `pipeline run` / deploy time -- in CI, hours to days before anyone hits the UI-clone block. Errors carry a nearest-match hint, because the real cases are near misses rather than nonsense. Three deterministic passes: separator- insensitive equality (`bq_table` -> `bq-table`), a difflib near match for ordinary typos, then token containment (`snapshot_date` -> `merchant_match_reference_snapshot_date`). A semantic rename with no lexical overlap (`wait` -> `timeout`) deliberately yields no hint; a confident wrong suggestion is worse than the declared-input list alone. The check fails OPEN wherever the declared set is unknowable, so it can never block a deploy over something it cannot see: * An unresolvable component spec is skipped. Pre-hydration `url:` refs, digest-pinned and name-pinned components whose spec is not inlined all return nothing from `_get_component_spec`, and the existing early return already covers them. * A malformed, non-list `inputs:` field is skipped, since the declared set cannot be trusted. * A well-formed empty `inputs: []` is NOT skipped. A component that declares no inputs genuinely accepts no arguments. No reserved-key allowlist is needed: the vendored `pipeline_schema.json` keeps `annotations`, `executionOptions` and `isEnabled` as siblings of `arguments`, so `arguments` is a pure input map with no metadata keys. Default is a hard error, matching the UI validator this exists to anticipate, with `TANGLE_ALLOW_UNDECLARED_TASK_ARGUMENTS=1` as an escape hatch for repositories that still carry undeclared arguments and need to deploy before cleaning them up. The env var follows the `TANGLE_*` boolean convention already used by `tangle_verbose_enabled` and `TANGLE_TRUSTED_HYDRATION_ALLOW_ALL`; a warning channel was rejected because `collect_pipeline_spec_errors` is error-only and the compiler's `warnings` list is a separate `CompileResult` surface that submit-time validation does not reach. One existing test fixture passed `arguments: {config: ...}` to a component declaring no inputs. It exercises payload sanitization, not validation, so the fixture now declares the input and its assertions are unchanged -- a small demonstration that the check catches real drift. Co-Authored-By: Claude Opus 4.8 (1M context) Assisted-By: devx/1bc39432-2e61-4aeb-8d86-e2f19fd326df --- .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../src/tangle_cli/pipeline_validation.py | 177 ++++++++++++++- pyproject.toml | 2 +- tests/test_packaging.py | 2 +- tests/test_pipeline_runs_cli.py | 5 +- tests/test_pipelines_cli.py | 210 ++++++++++++++++++ uv.lock | 2 +- 7 files changed, 383 insertions(+), 17 deletions(-) diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index 11c9185..f102116 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.12" + __version__ = "0.1.13" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_validation.py b/packages/tangle-cli/src/tangle_cli/pipeline_validation.py index d2fc128..3d45595 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_validation.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_validation.py @@ -8,7 +8,9 @@ from __future__ import annotations +import difflib import json +import os from functools import lru_cache from importlib import resources from typing import Any, Iterable, Mapping @@ -20,7 +22,15 @@ PIPELINE_GRAPH_PATH = "implementation.graph" TASKS_PATH = f"{PIPELINE_GRAPH_PATH}.tasks" +# Escape hatch for repositories that still carry undeclared task arguments. +# Set to a truthy value to downgrade the undeclared-argument check to a no-op. +# Follows the ``TANGLE_*`` boolean env-var convention used elsewhere in the CLI +# (see ``tangle_verbose_enabled`` / ``TANGLE_TRUSTED_HYDRATION_ALLOW_ALL``). +ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV = "TANGLE_ALLOW_UNDECLARED_TASK_ARGUMENTS" +_TRUTHY_ENV_VALUES = ("1", "true", "yes", "on") + __all__ = [ + "ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV", "PipelineValidationError", "collect_pipeline_spec_errors", "load_pipeline_schema", @@ -186,6 +196,138 @@ def _get_component_spec(task: Mapping[str, Any]) -> Mapping[str, Any] | None: return None +def _undeclared_arguments_allowed() -> bool: + """Return True when the undeclared-argument check is disabled by env var.""" + + value = os.environ.get(ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV, "") + return value.strip().lower() in _TRUTHY_ENV_VALUES + + +def _name_tokens(name: str) -> list[str]: + """Split an input name into lowercase alphanumeric tokens. + + Component authors mix ``snake_case``, ``kebab-case`` and spaces for the + same concept, so tokenizing lets the hint survive separator drift. + """ + + token = "" + tokens: list[str] = [] + for char in name: + if char.isalnum(): + token += char.lower() + elif token: + tokens.append(token) + token = "" + if token: + tokens.append(token) + return tokens + + +def _contains_tokens(haystack: list[str], needle: list[str]) -> bool: + """Return True when needle appears as a contiguous run inside haystack.""" + + if not needle or len(needle) > len(haystack): + return False + return any( + haystack[start : start + len(needle)] == needle + for start in range(len(haystack) - len(needle) + 1) + ) + + +def _nearest_declared_input(name: str, declared_names: Iterable[str]) -> str | None: + """Return the declared input ``name`` most plausibly meant, if any. + + Three deterministic passes over sorted candidates, strongest first: + + 1. separator-insensitive equality (``bq_table`` vs ``bq-table``); + 2. ``difflib`` near-match, which catches ordinary typos; + 3. token containment, which catches qualification renames such as + ``snapshot_date`` -> ``merchant_match_reference_snapshot_date``. + + Semantic renames with no lexical overlap (``wait`` -> ``timeout``) return + None on purpose: a wrong hint is worse than none. + """ + + candidates = sorted(declared_names) + tokens = _name_tokens(name) + if not tokens: + return None + + for candidate in candidates: + if _name_tokens(candidate) == tokens: + return candidate + + close = difflib.get_close_matches(name, candidates, n=1, cutoff=0.8) + if close: + return close[0] + + containing = [ + candidate + for candidate in candidates + if _contains_tokens(_name_tokens(candidate), tokens) + or _contains_tokens(tokens, _name_tokens(candidate)) + ] + if containing: + return min(containing, key=lambda candidate: (len(_name_tokens(candidate)), candidate)) + return None + + +def _suggest_declared_input(name: str, declared_names: Iterable[str]) -> str: + """Return a ``Did you mean ...`` clause, or an empty string when unsure. + + Real-world undeclared arguments are overwhelmingly typos or renames, so a + nearest-match hint is far more actionable than the declared-input list + alone. + """ + + nearest = _nearest_declared_input(name, declared_names) + return f" Did you mean '{nearest}'?" if nearest else "" + + +def _validate_undeclared_arguments( + full_task_name: str, + component_spec: Mapping[str, Any], + component_inputs: Any, + task_arguments: Mapping[str, Any], + declared_names: set[str], +) -> list[str]: + """Return errors for task arguments the component does not declare. + + Deliberately fails OPEN in the cases where the answer is unknowable rather + than wrong: + + * the caller only invokes this when ``_get_component_spec`` resolved a spec, + so unhydrated / digest-pinned / URL-only component refs are skipped; + * a malformed (non-list) ``inputs`` field is skipped, because the declared + set cannot be trusted. + + An empty-but-well-formed ``inputs`` list is NOT skipped: a component that + declares no inputs genuinely accepts no arguments. + + ``TaskSpec`` keeps ``annotations``, ``executionOptions`` and ``isEnabled`` + as siblings of ``arguments`` (see the vendored pipeline schema), so there + are no reserved/metadata keys inside ``arguments`` to allowlist. + """ + + if _undeclared_arguments_allowed(): + return [] + if component_inputs is not None and not isinstance(component_inputs, list): + return [] + + undeclared = sorted(str(name) for name in task_arguments if str(name) not in declared_names) + if not undeclared: + return [] + + component_name = component_spec.get("name") or "" + declared_display = sorted(declared_names) + return [ + f"Task '{full_task_name}': argument '{name}' is not a declared input of " + f"component '{component_name}'.{_suggest_declared_input(name, declared_names)} " + f"Declared inputs: {declared_display}" + for name in undeclared + ] + + def _is_input_required(input_spec: Mapping[str, Any]) -> bool: """Return True when a component input is non-optional and lacks a default.""" @@ -283,7 +425,8 @@ def _validate_task_inputs( ) -> list[str]: """Return component-input wiring errors for one task. - Required component inputs must be present in task arguments. When declared + Required component inputs must be present in task arguments, and supplied + arguments must correspond to a declared component input. When declared inputs are explicitly supplied, graph/task output references are checked regardless of whether the input is required. Nested graph component specs are validated recursively. @@ -309,13 +452,27 @@ def _validate_task_inputs( if not component_spec: return errors - component_inputs = component_spec.get("inputs", []) - if not isinstance(component_inputs, list): - component_inputs = [] + raw_component_inputs = component_spec.get("inputs", []) + component_inputs = raw_component_inputs if isinstance(raw_component_inputs, list) else [] task_arguments = task_spec.get("arguments", {}) or {} if not isinstance(task_arguments, Mapping): task_arguments = {} + declared_names = { + str(spec["name"]) + for spec in component_inputs + if isinstance(spec, Mapping) and spec.get("name") + } + errors.extend( + _validate_undeclared_arguments( + full_task_name, + component_spec, + raw_component_inputs, + task_arguments, + declared_names, + ) + ) + for input_spec in component_inputs: if not isinstance(input_spec, Mapping): continue @@ -344,12 +501,7 @@ def _validate_task_inputs( implementation = component_spec.get("implementation", {}) nested_graph = implementation.get("graph") if isinstance(implementation, Mapping) else None if isinstance(nested_graph, Mapping): - subgraph_inputs = { - str(inp.get("name")) - for inp in component_inputs - if isinstance(inp, Mapping) and inp.get("name") - } - errors.extend(_validate_graph_inputs(nested_graph, subgraph_inputs, f"{full_task_name} > ")) + errors.extend(_validate_graph_inputs(nested_graph, declared_names, f"{full_task_name} > ")) return errors @@ -389,7 +541,10 @@ def _validate_graph_inputs( def validate_component_inputs(pipeline_spec: Mapping[str, Any]) -> list[str]: - """Return required-input and reference-wiring errors for a pipeline. + """Return input-wiring errors for a pipeline. + + Covers missing required inputs, undeclared task arguments, and dangling + graph-input / task-output references. Validation uses embedded component specs when present. If the pipeline has no object-shaped implementation graph, this component-input pass returns no diff --git a/pyproject.toml b/pyproject.toml index d30d701..9c05dcf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.12" +version = "0.1.13" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 345fc66..5bcb1a1 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.12" in metadata + assert "Version: 0.1.13" 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 165a266..ce86fb5 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -550,8 +550,9 @@ def test_pipeline_runs_submit_dry_run_prints_sanitized_payload(monkeypatch, tmp_ "componentRef": { "name": "text-component", "text": ( - "name: Text Component\n_source_dir: /tmp/private\nimplementation:\n" - " container:\n image: busybox\n" + "name: Text Component\n_source_dir: /tmp/private\n" + "inputs:\n- {name: config, type: JsonObject}\n" + "implementation:\n container:\n image: busybox\n" ), } } diff --git a/tests/test_pipelines_cli.py b/tests/test_pipelines_cli.py index c9f28f7..e026ab0 100644 --- a/tests/test_pipelines_cli.py +++ b/tests/test_pipelines_cli.py @@ -11,6 +11,7 @@ from tangle_cli import cli from tangle_cli.pipeline_hydrator import PipelineHydrator +from tangle_cli.pipeline_validation import ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV from tangle_cli.pipelines import ( _dependency_edges, collect_pipeline_spec_errors, @@ -493,6 +494,215 @@ def test_component_input_validation_rejects_bad_graph_input_ref(): ] +def test_component_input_validation_rejects_undeclared_argument(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec(inputs=[{"name": "query", "type": "String"}]) + }, + "arguments": {"query": "shoes", "totally_made_up_arg": "1"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'totally_made_up_arg' is not a declared input of " + "component 'Component'. Declared inputs: ['query']" + ) + ] + + +def test_component_input_validation_suggests_nearest_declared_input(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[{"name": "timeout", "type": "Integer", "default": "1000"}] + ) + }, + "arguments": {"timeouts": "1440"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'timeouts' is not a declared input of " + "component 'Component'. Did you mean 'timeout'? Declared inputs: ['timeout']" + ) + ] + + +def test_component_input_validation_suggests_qualified_rename(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[ + { + "name": "merchant_match_reference_snapshot_date", + "type": "String", + "optional": True, + } + ] + ) + }, + "arguments": {"snapshot_date": "2026-09-12"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'snapshot_date' is not a declared input of " + "component 'Component'. Did you mean " + "'merchant_match_reference_snapshot_date'? Declared inputs: " + "['merchant_match_reference_snapshot_date']" + ) + ] + + +def test_component_input_validation_suggests_across_separator_drift(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec(inputs=[{"name": "bq-table", "type": "String"}]) + }, + "arguments": {"bq_table": "t"}, + } + ) + + errors = validate_component_inputs(pipeline) + + assert "Did you mean 'bq-table'?" in errors[0] + + +def test_component_input_validation_omits_hint_for_semantic_rename(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[{"name": "timeout", "type": "Integer", "default": "1000"}] + ) + }, + "arguments": {"wait": "1440"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'wait' is not a declared input of " + "component 'Component'. Declared inputs: ['timeout']" + ) + ] + + +def test_component_input_validation_accepts_fully_declared_arguments(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[ + {"name": "query", "type": "String"}, + {"name": "timeout", "type": "Integer", "default": "1000"}, + ] + ) + }, + "arguments": {"query": "shoes", "timeout": "1440"}, + } + ) + + assert validate_component_inputs(pipeline) == [] + + +@pytest.mark.parametrize( + "component_ref", + [ + {"url": "resolve://components.resolve.yaml#Component"}, + {"name": "Component", "digest": "879a8ddf"}, + {"text": "not: [valid"}, + ], + ids=["url-only", "digest-pinned", "unparsable-text"], +) +def test_component_input_validation_skips_unresolvable_component_spec(component_ref): + pipeline = _single_task_pipeline( + { + "componentRef": component_ref, + "arguments": {"totally_made_up_arg": "1"}, + } + ) + + assert validate_component_inputs(pipeline) == [] + + +def test_component_input_validation_skips_undeclared_check_on_malformed_inputs(): + spec = _component_spec() + spec["inputs"] = "not-a-list" + pipeline = _single_task_pipeline( + {"componentRef": {"spec": spec}, "arguments": {"anything": "1"}} + ) + + assert validate_component_inputs(pipeline) == [] + + +def test_component_input_validation_rejects_arguments_to_input_less_component(): + pipeline = _single_task_pipeline( + {"componentRef": {"spec": _component_spec()}, "arguments": {"anything": "1"}} + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'anything' is not a declared input of " + "component 'Component'. Declared inputs: []" + ) + ] + + +def test_component_input_validation_flags_undeclared_arguments_in_nested_graphs(): + inner = _component_spec( + inputs=[{"name": "query", "type": "String"}], + implementation={ + "graph": { + "tasks": { + "inner": { + "componentRef": { + "spec": _component_spec( + inputs=[{"name": "query", "type": "String"}] + ) + }, + "arguments": { + "query": {"graphInput": {"inputName": "query"}}, + "snapshot_date": "2026-09-12", + }, + } + } + } + }, + ) + pipeline = _single_task_pipeline( + {"componentRef": {"spec": inner}, "arguments": {"query": "shoes"}} + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task > inner': argument 'snapshot_date' is not a declared input of " + "component 'Component'. Declared inputs: ['query']" + ) + ] + + +def test_component_input_validation_undeclared_check_can_be_disabled(monkeypatch): + monkeypatch.setenv(ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV, "1") + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec(inputs=[{"name": "query", "type": "String"}]) + }, + "arguments": {"query": "shoes", "totally_made_up_arg": "1"}, + } + ) + + assert validate_component_inputs(pipeline) == [] + + def test_collect_pipeline_errors_combines_shape_schema_and_input_wiring(): pipeline = _single_task_pipeline( { diff --git a/uv.lock b/uv.lock index c495089..cf7bae0 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.12" +version = "0.1.13" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },