TangleML · Volv-G · Aug 5, 2026 · Aug 5, 2026 · Aug 5, 2026 · Aug 5, 2026
@@ -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`:

@@ -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
@@ -14,6 +14,6 @@
 try:
     __version__ = metadata_version("tangle-cli")
 except PackageNotFoundError:
-    __version__ = "0.1.7"
+    __version__ = "0.1.8"
 
 __all__ = ["TangleDynamicDiscoveryClient", "__version__"]
@@ -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):

@@ -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
 

@@ -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.<task_id>``); 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

@@ -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

@@ -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)
 

@@ -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:

@@ -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):