diff --git a/README.md b/README.md index 4a67052..54af38b 100644 --- a/README.md +++ b/README.md @@ -323,6 +323,45 @@ def greeting_pipeline(who: In[str], cfg) -> Out[str]: Task IDs default from the left-hand variable name at the call site, converted to title case. If there is no simple left-hand variable, or if you want a stable explicit label, call `.named("Task Id")` before invoking the task. Use `.bind(...)` to pre-fill task arguments and `.with_annotations({...})` to add per-task annotations. +##### Conditional task execution + +Pipeline inputs used as conditions are ordinary `In[str]` values; there is no special conditional input annotation. Pass the value through the reserved task-call metadata keyword `is_enabled=`: + +```python +@pipeline("Conditional greeting") +def conditional_greeting(enabled: In[str]) -> Out[str]: + greeting = write_greeting(who="world", is_enabled=enabled) + return greeting.out +``` + +This emits the canonical task field rather than a component argument: + +```yaml +isEnabled: + graphInput: + inputName: enabled +``` + +`is_enabled=` supports Python booleans (serialized as lowercase `"true"` / `"false"` strings), string constants, `In[...]` graph inputs, and previous task outputs such as `is_enabled=gate.Output`. It is container-component task metadata; component function parameters are not implicitly conditions. + +If a component itself declares an input named `is_enabled`, bind that component argument separately while using the call-site keyword for task metadata: + +```python +@task(image="python:3.12") +def work(is_enabled: str, message: str) -> str: + return message + +@pipeline("Input-name collision") +def collision(runtime_condition: In[str]) -> Out[str]: + result = work.bind(is_enabled="component-input-value")( + message="hello", + is_enabled=runtime_condition, + ) + return result.Output +``` + +The bound value remains under `arguments.is_enabled`; the call-site value emits as `isEnabled`. Tangle does not evaluate conditions on graph-component tasks, so `subpipeline(...)(is_enabled=...)` is rejected with guidance to condition tasks inside the child pipeline. A child graph input with that name remains available through `subpipeline(...).bind(is_enabled=...)(...)`. There is no `condition` alias. + ##### Task images, dependencies, and image IDs Use `@task(image="...")` to write the component image directly. Use `dependencies_from="pyproject.toml"` when generated components need to install Python dependencies. Several tasks can share one authoring-only `TaskEnv`: diff --git a/examples/python_pipeline/is_enabled_pipeline.py b/examples/python_pipeline/is_enabled_pipeline.py new file mode 100644 index 0000000..23d0d77 --- /dev/null +++ b/examples/python_pipeline/is_enabled_pipeline.py @@ -0,0 +1,42 @@ +"""Runnable Python-authoring example for task-level conditional execution. + +Compile from the repository root with:: + + uv run tangle sdk pipelines compile \ + examples/python_pipeline/is_enabled_pipeline.py \ + --pipeline conditional_pipeline \ + --output /tmp/tangle-is-enabled-demo/pipeline.yaml +""" + +from tangle_cli.python_pipeline import In, Out, pipeline, task + + +@task(image="python:3.12") +def condition_value(value: str = "true") -> str: + """Produce a string value that another task can use as its condition.""" + return value + + +@task(image="python:3.12") +def show_message(message: str) -> str: + """Print and return a message when this task is enabled.""" + print(message) + return message + + +@pipeline("Conditional execution demo") +def conditional_pipeline(enabled: In[str]) -> Out[str]: + constant_false = show_message( + message="This task is always skipped", + is_enabled=False, + ) + graph_input_condition = show_message( + message="This task follows the runtime graph input", + is_enabled=enabled, + ) + computed_condition = condition_value(value="true") + task_output_condition = show_message( + message="This task follows another task's output", + is_enabled=computed_condition.Output, + ) + return task_output_condition.Output diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index ce1e6aa..1804ad6 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.7" + __version__ = "0.1.8" __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 79e0a3b..d2fc128 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_validation.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_validation.py @@ -291,6 +291,20 @@ def _validate_task_inputs( errors: list[str] = [] full_task_name = f"{path_prefix}{task_name}" if path_prefix else task_name + + if "isEnabled" in task_spec: + condition = task_spec["isEnabled"] + error = _validate_graph_input_ref( + condition, graph_inputs, full_task_name, "isEnabled" + ) + if error: + errors.append(error) + error = _validate_task_output_ref( + condition, tasks, task_outputs, full_task_name, "isEnabled" + ) + if error: + errors.append(error) + component_spec = _get_component_spec(task_spec) if not component_spec: return errors @@ -501,6 +515,20 @@ def _validate_graph_spec( else: edges.add((referenced_task, str(task_name))) + # A task-output condition is a real scheduling dependency, just like a + # task-output argument. Include it in dangling-reference and cycle + # checks so local validation matches backend ordering semantics. + if "isEnabled" in raw_task: + for referenced_task in _extract_task_output_refs( + raw_task["isEnabled"] + ): + if referenced_task not in task_names: + errors.append( + f"{task_path}.isEnabled references unknown task {referenced_task!r}" + ) + else: + edges.add((referenced_task, str(task_name))) + if isinstance(component_ref, Mapping): nested_spec = component_ref.get("spec") if isinstance(nested_spec, Mapping): diff --git a/packages/tangle-cli/src/tangle_cli/pipelines.py b/packages/tangle-cli/src/tangle_cli/pipelines.py index b843bdc..d21f67d 100644 --- a/packages/tangle-cli/src/tangle_cli/pipelines.py +++ b/packages/tangle-cli/src/tangle_cli/pipelines.py @@ -419,6 +419,11 @@ def _dependency_edges(tasks: Mapping[str, Any]) -> set[tuple[str, str]]: for referenced_task in _extract_task_output_refs(task_spec.get("arguments", {})): if referenced_task in task_names: edges.add((referenced_task, target)) + for referenced_task in _extract_task_output_refs( + task_spec.get("isEnabled") + ): + if referenced_task in task_names: + edges.add((referenced_task, target)) return edges diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py index 8887ffe..fabbe17 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py @@ -4,7 +4,7 @@ name, description, metadata, inputs, outputs, implementation Per-task key order: - annotations?, componentRef, arguments + annotations?, componentRef, arguments?, isEnabled? Argument values are emitted in the runnable ``ArgumentValue`` shape, dispatched purely on the VALUE's runtime type — never on the argument @@ -38,7 +38,7 @@ from .dynamic_data import DynamicData from .errors import CompileError, InvalidArgumentTypeError -from .graph import EdgeRef, GraphBuilder, TaskNode +from .graph import IS_ENABLED_UNSET, EdgeRef, GraphBuilder, TaskNode from .placeholders import GraphInputPlaceholder, TaskOutputProxy from .raw import Raw @@ -141,7 +141,7 @@ def _emit_task( node: TaskNode, task_path: str, exempt_paths: set[str] ) -> dict[str, Any]: """Build the per-task body dict in canonical key order: - ``annotations?, componentRef, arguments``. + ``annotations?, componentRef, arguments?, isEnabled?``. ``task_path`` is this task's dot-delimited JSON path (``implementation.graph.tasks.``); each argument's path is @@ -164,6 +164,9 @@ def _emit_task( for k, v in node.arguments.items() } + if node.is_enabled is not IS_ENABLED_UNSET: + body["isEnabled"] = _emit_is_enabled(node.is_enabled) + return body @@ -283,6 +286,36 @@ def _validate_constant(value: Any, key: str) -> None: ) +def _emit_is_enabled(value: Any) -> Any: + """Render task-level conditional metadata in the backend contract. + + Python booleans are normalized to lowercase strings because runnable + Tangle schemas and the backend evaluator do not accept raw JSON/YAML + booleans. String constants and graph/task references use the same wire + shapes as runnable argument values. Other value forms, including + ``DynamicData`` and ``Raw``, are intentionally unsupported by the backend + condition evaluator and fail at compile time. + """ + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, TaskOutputProxy): + return { + "taskOutput": { + "taskId": value._task_id, + "outputName": value._resolved_output_name(), + } + } + if isinstance(value, GraphInputPlaceholder): + return {"graphInput": {"inputName": value.input_name}} + if isinstance(value, str): + return value + raise InvalidArgumentTypeError( + f"unsupported is_enabled value type {type(value).__name__!r}. " + "Task conditions only support bool, string constants, graphInput, " + "or taskOutput; booleans are serialized as lowercase strings." + ) + + def _emit_edge_value(edge: EdgeRef) -> dict[str, Any]: """Render an :class:`EdgeRef` as a dehydrated ``ArgumentValue`` sub-dict (``{taskOutput|graphInput: {...}}``) used in diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/graph.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/graph.py index 4723c6f..9ef2446 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/graph.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/graph.py @@ -10,6 +10,12 @@ from typing import Any, Literal +# Distinguishes an omitted task condition from an explicitly supplied value. +# The latter is validated by the emitter, so ``is_enabled=None`` fails closed +# instead of being silently omitted. +IS_ENABLED_UNSET = object() + + @dataclass class EdgeRef: """How one task's input wires to a producer. @@ -28,9 +34,9 @@ class EdgeRef: class TaskNode: """A single emitted task in the graph. - ``arguments`` values may be plain strings, TaskOutputProxy objects - (for taskOutput edges in non-``wait_for`` argument positions — not - used in the PoC), or GraphInputPlaceholder objects. + ``arguments`` values may be plain strings, TaskOutputProxy objects, or + GraphInputPlaceholder objects. ``is_enabled`` is separate task metadata; + the emitter normalizes and serializes it as ``isEnabled`` when supplied. """ task_id: str @@ -39,6 +45,7 @@ class TaskNode: ref_digest: str | None = None arguments: dict[str, Any] = field(default_factory=dict) annotations: dict[str, str] | None = None + is_enabled: Any = IS_ENABLED_UNSET @dataclass diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py index 2edbcb5..8ed1adc 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py @@ -16,6 +16,7 @@ from typing import Any from .errors import CompileError +from .graph import IS_ENABLED_UNSET _UNWRAPPED_KEY_RE = re.compile(r"^[A-Za-z0-9_-]+$") @@ -291,7 +292,12 @@ def materialize(self, output_path: Path | None = None) -> Path: # ------------------------------------------------------------------ # Trace-mode call site - def __call__(self, **kwargs: Any) -> Any: + def __call__( + self, + *, + is_enabled: Any = IS_ENABLED_UNSET, + **kwargs: Any, + ) -> Any: """Trace-mode invocation. Records a :class:`TaskNode` into the active :class:`GraphBuilder` @@ -301,11 +307,17 @@ def __call__(self, **kwargs: Any) -> Any: call site (resolved via the AST pre-pass map stashed on the builder). - Edge kwargs (``wait_for`` / ``depends_on``) and regular kwargs - share one ``arguments`` dict in the IR; the value-vs-key - dispatch happens at emit time. ``.bind(...)`` kwargs are merged - in last so call-site kwargs win on conflict (same key) and come - first in insertion order (the bind block is appended). + ``is_enabled`` is task metadata, not a component input. It accepts a + boolean, string, graph input, or task output and is emitted as the + canonical ``isEnabled`` task field. If a component itself declares an + input named ``is_enabled``, bind that input with + ``ref(...).bind(is_enabled=...)``; bound kwargs remain component + arguments while the reserved call-site keyword remains task metadata, + so both may be used on the same task. Edge kwargs (``wait_for`` / + ``depends_on``) and regular kwargs share one ``arguments`` dict in the + IR; the value-vs-key dispatch happens at emit time. ``.bind(...)`` + kwargs are merged in last so call-site kwargs win on conflict (same + key) and come first in insertion order (the bind block is appended). """ # Local import keeps the @ref shell importable during early # bootstrap (and avoids the circular dep at module load). @@ -364,6 +376,7 @@ def __call__(self, **kwargs: Any) -> Any: ref_digest=self.ref_digest, arguments=merged, annotations=dict(self.annotations) if self.annotations else None, + is_enabled=is_enabled, ) builder.add_task(node) diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/subpipeline.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/subpipeline.py index 549082a..0f9968e 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/subpipeline.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/subpipeline.py @@ -15,7 +15,10 @@ ergonomics (``.bind`` / ``.named`` / ``.with_annotations`` and call-site kwargs). Calling the handle inside an active ``@pipeline`` trace records ONE parent task (never the child's internals) and returns a -:class:`tangle_cli.python_pipeline.placeholders.TaskOutputProxy`. +:class:`tangle_cli.python_pipeline.placeholders.TaskOutputProxy`. Tangle only +supports conditional execution for container-component tasks, so the reserved +call-site ``is_enabled=`` metadata keyword is rejected on subpipeline boundary +(graph-component) tasks. The child body is NOT executed into the parent's :class:`GraphBuilder`. The compile driver (a later milestone) reads the recorded child @@ -120,6 +123,11 @@ def __call__(self, **kwargs: Any) -> "TaskOutputProxy": declared outputs (derived from the child's return annotation) so unknown named access fails early and a bare proxy resolves to a default output only when unambiguous. + + ``is_enabled`` at the call site is rejected because Tangle does not + evaluate conditions on graph-component tasks. A child graph input with + that name remains available through ``.bind(is_enabled=...)``, matching + the reserved-metadata collision convention used by ``CallableRef``. """ import sys @@ -137,6 +145,15 @@ def __call__(self, **kwargs: Any) -> "TaskOutputProxy": "@pipeline, or compile the script with `tangle sdk pipelines compile`." ) + if "is_enabled" in kwargs: + raise CompileError( + "subpipeline tasks do not support call-site is_enabled= because " + "Tangle conditional execution is limited to container-component " + "tasks. Apply conditions to tasks inside the child pipeline. If " + "the child declares a graph input named 'is_enabled', pass that " + "input with .bind(is_enabled=...)." + ) + # Resolve the parent task ID. ``.named(...)`` always wins over the # AST-derived auto ID. if self.task_id_hint is not None: diff --git a/packages/tangle-cli/src/tangle_cli/schema_validation.py b/packages/tangle-cli/src/tangle_cli/schema_validation.py index c33c15c..fffc3c2 100644 --- a/packages/tangle-cli/src/tangle_cli/schema_validation.py +++ b/packages/tangle-cli/src/tangle_cli/schema_validation.py @@ -212,6 +212,15 @@ def assert_no_template_delimiters( _ARGUMENT_WRAPPER_KEYS = ("graphInput", "taskOutput", "dynamicData") +def _is_condition_value(value: Any) -> bool: + """True for a backend-supported serialized ``isEnabled`` value.""" + if isinstance(value, str): + return True + return isinstance(value, Mapping) and any( + key in value for key in ("graphInput", "taskOutput") + ) + + def _is_argument_value(value: Any) -> bool: """True when ``value`` looks like a runnable ArgumentValue — a raw string constant, or a mapping carrying a ``graphInput`` / ``taskOutput`` / @@ -267,6 +276,8 @@ def is_dehydrated_pipeline(data: Any) -> bool: for task in tasks.values(): if not isinstance(task, Mapping): return False + if "isEnabled" in task and not _is_condition_value(task["isEnabled"]): + return False arguments = task.get("arguments") if arguments is None: continue @@ -376,6 +387,13 @@ def _validate_semantics(data: Mapping[str, Any]) -> None: input_names, loc=f"tasks.{task_id}.arguments.{arg_name}", ) + if "isEnabled" in task: + _check_argument_refs( + task["isEnabled"], + task_ids, + input_names, + loc=f"tasks.{task_id}.isEnabled", + ) output_values = graph.get("outputValues") if isinstance(graph, Mapping) else None if isinstance(output_values, Mapping): diff --git a/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json b/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json index fdf311c..d667ff1 100644 --- a/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json +++ b/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json @@ -118,7 +118,7 @@ "additionalProperties": { "$ref": "#/$defs/ArgumentValue" }, "description": "Map of component input name -> argument source." }, - "isEnabled": { "type": "boolean" }, + "isEnabled": { "$ref": "#/$defs/ConditionValue" }, "executionOptions": { "$ref": "#/$defs/ExecutionOptionsSpec" }, "annotations": { "type": "object", @@ -178,6 +178,15 @@ ] }, + "ConditionValue": { + "description": "Task execution condition evaluated by Tangle. Supports a string constant or a graph/task output reference. Raw JSON/YAML booleans and dynamicData are not accepted.", + "oneOf": [ + { "type": "string" }, + { "$ref": "#/$defs/GraphInputArgument" }, + { "$ref": "#/$defs/TaskOutputArgument" } + ] + }, + "GraphInputArgument": { "type": "object", "additionalProperties": false, diff --git a/packages/tangle-cli/src/tangle_cli/schemas/pipeline_schema.json b/packages/tangle-cli/src/tangle_cli/schemas/pipeline_schema.json index 7679b27..59e4c34 100644 --- a/packages/tangle-cli/src/tangle_cli/schemas/pipeline_schema.json +++ b/packages/tangle-cli/src/tangle_cli/schemas/pipeline_schema.json @@ -933,9 +933,6 @@ { "$ref": "#/$defs/TaskOutputArgument" }, - { - "$ref": "#/$defs/DynamicDataArgument" - }, { "type": "null" } @@ -1893,9 +1890,6 @@ { "$ref": "#/$defs/TaskOutputArgument" }, - { - "$ref": "#/$defs/DynamicDataArgument" - }, { "type": "null" } diff --git a/pyproject.toml b/pyproject.toml index 970362b..2a64ca0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.7" +version = "0.1.8" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/scripts/refresh_pipeline_schema.py b/scripts/refresh_pipeline_schema.py index 74c9703..ef77014 100755 --- a/scripts/refresh_pipeline_schema.py +++ b/scripts/refresh_pipeline_schema.py @@ -42,6 +42,36 @@ def fetch_tangle_structures() -> str: return response.text +def _narrow_task_condition_schema(value: Any) -> None: + """Align generated ``isEnabled`` schemas with backend evaluation support. + + The pinned Pydantic model types task conditions like general arguments and + therefore includes ``DynamicDataArgument``. The backend condition evaluator + only supports constants, graph inputs, and task outputs, so keep refreshes + from widening the vendored validation contract accidentally. + """ + if isinstance(value, dict): + properties = value.get("properties") + if isinstance(properties, dict): + condition = properties.get("isEnabled") + if isinstance(condition, dict) and isinstance( + condition.get("anyOf"), list + ): + condition["anyOf"] = [ + variant + for variant in condition["anyOf"] + if not ( + isinstance(variant, dict) + and variant.get("$ref") == "#/$defs/DynamicDataArgument" + ) + ] + for nested in value.values(): + _narrow_task_condition_schema(nested) + elif isinstance(value, list): + for nested in value: + _narrow_task_condition_schema(nested) + + def generate_schema(source_text: str) -> dict[str, Any]: import pydantic @@ -64,7 +94,7 @@ def generate_schema(source_text: str) -> dict[str, Any]: adapter = pydantic.TypeAdapter(graph_spec) adapter.rebuild(_types_namespace=namespace) graph_schema = adapter.json_schema() - return { + schema = { "$schema": "http://json-schema.org/draft-07/schema#", "title": "Tangle Pipeline Schema (generated from TangleML)", "type": "object", @@ -90,6 +120,8 @@ def generate_schema(source_text: str) -> dict[str, Any]: "generatedBy": "scripts/refresh_pipeline_schema.py", }, } + _narrow_task_condition_schema(schema) + return schema def main() -> None: diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 4092937..e40e838 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -178,7 +178,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.7" in metadata + assert "Version: 0.1.8" 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_compiler.py b/tests/test_pipeline_compiler.py index 052b151..bf0f083 100644 --- a/tests/test_pipeline_compiler.py +++ b/tests/test_pipeline_compiler.py @@ -733,6 +733,91 @@ def test_compile_validates_schema(multi_arg_args): validate_dehydrated_data(data) +def test_compile_emits_first_class_is_enabled_values_and_collision_escape(tmp_path): + """Task conditions use canonical ``isEnabled`` without stealing a bound + component input of the same Python name.""" + out = tmp_path / "compiled.yaml" + _provide_noop(out) + src = tmp_path / "conditional_pipeline.py" + src.write_text( + "from tangle_cli.python_pipeline import In, Out, pipeline, ref\n" + "\n" + "@pipeline('Conditional Pipeline')\n" + "def conditional_pipeline(enabled: In[str]) -> Out[str]:\n" + " producer = ref(url='file://./noop.yaml')(message='produce')\n" + " by_bool = ref(url='file://./noop.yaml')(is_enabled=False)\n" + " by_string = ref(url='file://./noop.yaml')(is_enabled='TRUE')\n" + " by_input = ref(url='file://./noop.yaml')(is_enabled=enabled)\n" + " by_task = ref(url='file://./noop.yaml')(is_enabled=producer.should_run)\n" + " collision = ref(url='file://./noop.yaml').bind(\n" + " is_enabled='component-value'\n" + " )(is_enabled=by_task)\n" + " alias_is_argument = ref(url='file://./noop.yaml')(condition='false')\n" + " return collision\n", + encoding="utf-8", + ) + + compile_pipeline(src, out) + tasks = yaml.safe_load(out.read_text())["implementation"]["graph"]["tasks"] + + assert tasks["By Bool"]["isEnabled"] == "false" + assert tasks["By String"]["isEnabled"] == "TRUE" + assert tasks["By Input"]["isEnabled"] == { + "graphInput": {"inputName": "enabled"} + } + assert tasks["By Task"]["isEnabled"] == { + "taskOutput": {"taskId": "Producer", "outputName": "should_run"} + } + assert tasks["Collision"]["isEnabled"] == { + "taskOutput": {"taskId": "By Task", "outputName": "wait_for_output"} + } + assert tasks["Collision"]["arguments"]["is_enabled"] == "component-value" + assert tasks["Alias Is Argument"]["arguments"]["condition"] == "false" + assert "isEnabled" not in tasks["Alias Is Argument"] + + +def test_compile_omits_is_enabled_when_not_authored(multi_arg_args): + data, _produce, _consume = multi_arg_args + tasks = data["implementation"]["graph"]["tasks"] + assert all("isEnabled" not in task for task in tasks.values()) + + +@pytest.mark.parametrize( + ("expression", "type_name"), + [ + ("None", "NoneType"), + ("1", "int"), + ("{'graphInput': {'inputName': 'enabled'}}", "dict"), + ("dynamic_secret('TOKEN')", "DynamicData"), + ("raw('{{runtime}}')", "Raw"), + ], +) +def test_compile_rejects_unsupported_is_enabled_values( + expression, type_name, tmp_path +): + out = tmp_path / "compiled.yaml" + _provide_noop(out) + src = tmp_path / "bad_condition_pipeline.py" + src.write_text( + "from tangle_cli.python_pipeline import (\n" + " In, Out, dynamic_secret, pipeline, raw, ref,\n" + ")\n" + "\n" + "@pipeline('Bad Condition Pipeline')\n" + "def bad_condition_pipeline(enabled: In[str]) -> Out[str]:\n" + f" bad = ref(url='file://./noop.yaml')(is_enabled={expression})\n" + " return bad\n", + encoding="utf-8", + ) + + with pytest.raises(CompileError) as exc: + compile_pipeline(src, out) + + assert "unsupported is_enabled value type" in str(exc.value) + assert type_name in str(exc.value) + assert not out.exists() + + # --------------------------------------------------------------------------- # PipelineCompiler handler + ZONE_ROOT_MARKERS seam. @@ -1053,6 +1138,60 @@ def test_compile_subpipeline_emits_child_sidecar(tmp_path): validate_dehydrated_data(yaml.safe_load(child.read_text())) +def test_compile_subpipeline_rejects_is_enabled_task_metadata(tmp_path): + src = tmp_path / "conditional_subpipeline.py" + source = (FIXTURES / "subpipeline_pipeline.py").read_text(encoding="utf-8") + src.write_text( + source.replace( + "(seed=parent_wait_token)", + "(seed=parent_wait_token, is_enabled='false')", + ), + encoding="utf-8", + ) + shutil.copy(FIXTURES / "config.yaml", tmp_path / "config.yaml") + + with pytest.raises(CompileError) as exc: + compile_pipeline( + src, tmp_path / "compiled.yaml", pipeline_name="Parent Pipeline" + ) + + message = str(exc.value) + assert "subpipeline tasks do not support call-site is_enabled=" in message + assert "container-component tasks" in message + assert ".bind(is_enabled=...)" in message + + +def test_compile_subpipeline_preserves_bound_is_enabled_graph_input(tmp_path): + src = tmp_path / "bound_is_enabled_subpipeline.py" + src.write_text( + "from tangle_cli.python_pipeline import In, Out, pipeline, subpipeline, task\n" + "\n" + "@task(image='python:3.12')\n" + "def child_task(value: str):\n" + " print(value)\n" + "\n" + "@pipeline('Child')\n" + "def child(is_enabled: In[str]) -> Out[str]:\n" + " result = child_task(value=is_enabled)\n" + " return result\n" + "\n" + "@pipeline('Parent')\n" + "def parent(seed: In[str]) -> Out[str]:\n" + " child_result = subpipeline(child).bind(\n" + " is_enabled='component-input'\n" + " )()\n" + " return child_result\n", + encoding="utf-8", + ) + + out = tmp_path / "compiled.yaml" + compile_pipeline(src, out, pipeline_name="Parent") + + tasks = yaml.safe_load(out.read_text())["implementation"]["graph"]["tasks"] + assert tasks["Child Result"]["arguments"]["is_enabled"] == "component-input" + assert "isEnabled" not in tasks["Child Result"] + + def test_compile_subpipeline_override_config_wins(tmp_path): """``.override_config`` on a subpipeline edge sets the child's COMPILE-TIME cfg, and the overridden value wins over the child's own config.yaml. The diff --git a/tests/test_pipeline_dehydrator.py b/tests/test_pipeline_dehydrator.py index e9aed29..ea515cb 100644 --- a/tests/test_pipeline_dehydrator.py +++ b/tests/test_pipeline_dehydrator.py @@ -65,6 +65,33 @@ def test_pipeline_dehydrator_replaces_refs_by_explicit_choice(tmp_path: Path) -> assert "spec" in keep_result["implementation"]["graph"]["tasks"]["task"]["componentRef"] +def test_pipeline_dehydrator_preserves_is_enabled_round_trip(tmp_path: Path) -> None: + task = _task( + "Leaf Component", + "digest-1", + canonical_url="https://example.test/leaf.yaml", + ) + task["isEnabled"] = { + "taskOutput": {"taskId": "gate", "outputName": "enabled"} + } + data = _pipeline( + { + "gate": _task( + "Gate", "digest-2", canonical_url="https://example.test/gate.yaml" + ), + "task": task, + } + ) + + result = PipelineDehydrator( + {"": DehydrateChoice.URL}, output_file=tmp_path / "out.yaml" + ).dehydrate(data) + + assert result["implementation"]["graph"]["tasks"]["task"]["isEnabled"] == { + "taskOutput": {"taskId": "gate", "outputName": "enabled"} + } + + def test_pipeline_dehydrator_construction_is_auth_env_safe(monkeypatch: pytest.MonkeyPatch) -> None: """Auth-free dehydration construction must not require TANGLE_API_URL.""" diff --git a/tests/test_pipelines_cli.py b/tests/test_pipelines_cli.py index e87690c..d9f001e 100644 --- a/tests/test_pipelines_cli.py +++ b/tests/test_pipelines_cli.py @@ -12,6 +12,7 @@ from tangle_cli import cli from tangle_cli.pipeline_hydrator import PipelineHydrator from tangle_cli.pipelines import ( + _dependency_edges, collect_pipeline_spec_errors, load_pipeline_schema, validate_component_inputs, @@ -199,6 +200,32 @@ def test_pipelines_validate_fails_for_invalid_yaml(tmp_path: Path): assert "unknown task 'missing'" in str(exc_info.value) +def test_pipeline_validation_treats_is_enabled_task_output_as_dependency(): + pipeline = _minimal_valid_pipeline() + tasks = pipeline["implementation"]["graph"]["tasks"] + tasks["extract"]["isEnabled"] = { + "taskOutput": {"taskId": "load", "outputName": "enabled"} + } + + errors = collect_pipeline_spec_errors(pipeline) + + assert ("load", "extract") in _dependency_edges(tasks) + assert any("dependency cycle" in error for error in errors) + + +def test_pipeline_validation_rejects_unknown_is_enabled_reference(): + pipeline = _minimal_valid_pipeline() + pipeline["implementation"]["graph"]["tasks"]["load"]["isEnabled"] = { + "taskOutput": {"taskId": "missing", "outputName": "enabled"} + } + + errors = collect_pipeline_spec_errors(pipeline) + + assert any( + "isEnabled references unknown task 'missing'" in error for error in errors + ) + + def test_pipelines_validate_rejects_non_string_task_ids(tmp_path: Path): pipeline_path = _write_pipeline( tmp_path / "pipeline.yaml", @@ -256,6 +283,17 @@ def test_pipeline_schema_validation_rejects_wrong_root_input_type(): assert any("'inputs' must be array, got str" in error for error in errors) +@pytest.mark.parametrize( + "condition", + [False, {"dynamicData": {"secret": {"name": "TOKEN"}}}], +) +def test_pipeline_schema_rejects_backend_unsupported_is_enabled(condition): + pipeline = _minimal_valid_pipeline() + pipeline["implementation"]["graph"]["tasks"]["load"]["isEnabled"] = condition + + assert validate_pipeline_schema(pipeline) + + def _component_spec( *, inputs: list[dict] | None = None, diff --git a/tests/test_schema_validation.py b/tests/test_schema_validation.py index 536a574..1f067db 100644 --- a/tests/test_schema_validation.py +++ b/tests/test_schema_validation.py @@ -101,6 +101,36 @@ def test_validate_dehydrated_data_rejects_empty_tasks(): validate_dehydrated_data(data) +@pytest.mark.parametrize( + "condition", + [ + "false", + {"graphInput": {"inputName": "in1"}}, + {"taskOutput": {"taskId": "extract", "outputName": "enabled"}}, + ], +) +def test_validate_dehydrated_data_accepts_supported_is_enabled(condition): + data = _valid_pipeline() + data["implementation"]["graph"]["tasks"]["load"]["isEnabled"] = condition + validate_dehydrated_data(data) + + +@pytest.mark.parametrize( + "condition", + [ + False, + {"dynamicData": {"secret": {"name": "TOKEN"}}}, + {"graphInput": "in1"}, + {"taskOutput": {"taskId": "extract"}}, + ], +) +def test_validate_dehydrated_data_rejects_unsupported_is_enabled(condition): + data = _valid_pipeline() + data["implementation"]["graph"]["tasks"]["load"]["isEnabled"] = condition + with pytest.raises(SchemaValidationError): + validate_dehydrated_data(data) + + # --------------------------------------------------------------------------- # Template-delimiter output contract. @@ -209,6 +239,17 @@ def test_validate_dehydrated_pipeline_rejects_undeclared_graph_input(): assert "nope" in str(exc.value) +def test_validate_dehydrated_pipeline_checks_is_enabled_references(): + data = _valid_pipeline() + data["implementation"]["graph"]["tasks"]["load"]["isEnabled"] = { + "taskOutput": {"taskId": "missing", "outputName": "enabled"} + } + with pytest.raises(SchemaValidationError) as exc: + validate_dehydrated_pipeline(data) + assert "tasks.load.isEnabled" in str(exc.value) + assert "missing" in str(exc.value) + + def test_validate_dehydrated_pipeline_rejects_output_value_without_output(): data = _valid_pipeline() data["implementation"]["graph"]["outputValues"]["ghost"] = { diff --git a/uv.lock b/uv.lock index acd5ed7..2a49bac 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.7" +version = "0.1.8" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },