Add Python task execution options authoring (v0.1.10)

README.md

@@ -413,6 +413,56 @@ def collision(runtime_condition: In[str]) -> Out[str]:
 
 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 execution options and caching
+
+Tangle caches task results, so a task that reads state which changes between runs (a run's `createdBy`, wall-clock time, an external table that the graph does not depend on) must opt out of caching explicitly. Use the reserved task-call metadata keyword `max_cache_staleness=`:
+
+```python
+@pipeline("Scheduled-run gate")
+def scheduled_gate() -> Out[str]:
+    is_scheduled = read_runtime_state(
+        name="CLOUD_PIPELINES_PIPELINE_RUN_CREATED_BY",
+        max_cache_staleness="P0D",
+    )
+    return is_scheduled.Output
+```
+
+This emits the canonical task field rather than a component argument:
+
+```yaml
+executionOptions:
+  cachingStrategy:
+    maxCacheStaleness: P0D
+```
+
+`P0D` means "never reuse a cached result"; any other ISO-8601 duration (`P7D`) caps how stale a reusable result may be. For the rest of `ExecutionOptionsSpec`, use the general `execution_options=` passthrough:
+
+```python
+uploaded = flaky_upload(
+    payload=data.Output,
+    execution_options={"retryStrategy": {"maxRetries": 3}},
+)
+```
+
+Both keywords may be combined; `max_cache_staleness=` wins over a `cachingStrategy.maxCacheStaleness` supplied through `execution_options=`, and every other passthrough field is preserved. A mapping passed as `execution_options=` is never mutated, so one shared constant can be reused across tasks.
+
+Tangle models exactly two execution-option groups today — `cachingStrategy.maxCacheStaleness` and `retryStrategy.maxRetries` (required whenever `retryStrategy` is present). Any other key is rejected at compile time: the backend ignores unmodeled keys silently, so accepting one would advertise a setting that never takes effect.
+
+Execution options are STATIC compile-time settings, so `max_cache_staleness` takes an RFC3339 duration string and `retryStrategy.maxRetries` a non-negative integer. Graph inputs, task outputs, `dynamic_secret(...)`, and `raw(...)` values are rejected because the backend does not resolve them for `executionOptions`. Passing an empty `execution_options={}` is an error — omit the keyword instead.
+
+If a component itself declares an input named `max_cache_staleness` or `execution_options`, bind that component argument separately while using the call-site keyword for task metadata:
+
+```python
+result = work.bind(max_cache_staleness="component-input-value")(
+    message="hello",
+    max_cache_staleness="P0D",
+)
+```
+
+The bound value remains under `arguments.max_cache_staleness`; the call-site value emits as `executionOptions`. Tangle applies caching and retries to container-component tasks, so `subpipeline(...)(max_cache_staleness=...)` and `subpipeline(...)(execution_options=...)` are rejected with guidance to set them on tasks inside the child pipeline. A child graph input with either name remains available through `subpipeline(...).bind(...)`.
+
+See `examples/python_pipeline/execution_options_pipeline.py` for a runnable example.
+
 ##### 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`:

examples/python_pipeline/execution_options_pipeline.py

@@ -0,0 +1,47 @@
+"""Runnable Python-authoring example for task-level execution options.
+
+Compile from the repository root with::
+
+    uv run tangle sdk pipelines compile \
+      examples/python_pipeline/execution_options_pipeline.py \
+      --pipeline execution_options_pipeline \
+      --output /tmp/tangle-execution-options-demo/pipeline.yaml
+"""
+
+from tangle_cli.python_pipeline import Out, pipeline, task
+
+
+@task(image="python:3.12")
+def read_runtime_state(name: str = "CREATED_BY") -> str:
+    """Read state that changes between runs, so caching must be disabled."""
+    import os
+
+    return os.environ.get(name, "")
+
+
+@task(image="python:3.12")
+def flaky_upload(payload: str) -> str:
+    """Stand-in for a task that benefits from retries."""
+    print(payload)
+    return payload
+
+
+@pipeline("Execution options demo")
+def execution_options_pipeline() -> Out[str]:
+    # ``max_cache_staleness="P0D"`` is the narrow knob for the common
+    # "never reuse a cached result for this task" case. Required whenever the
+    # task reads runtime state that a cached result would silently stale out.
+    runtime_state = read_runtime_state(
+        name="CLOUD_PIPELINES_PIPELINE_RUN_CREATED_BY",
+        max_cache_staleness="P0D",
+    )
+
+    # ``execution_options=`` is the general passthrough for the rest of
+    # ExecutionOptionsSpec. Tangle models exactly two groups today,
+    # ``cachingStrategy`` and ``retryStrategy``; anything else is rejected at
+    # compile time because the backend would silently ignore it.
+    uploaded = flaky_upload(
+        payload=runtime_state.Output,
+        execution_options={"retryStrategy": {"maxRetries": 3}},
+    )
+    return uploaded.Output

packages/tangle-cli/src/tangle_cli/__init__.py

@@ -14,6 +14,6 @@
 try:
     __version__ = metadata_version("tangle-cli")
 except PackageNotFoundError:
-    __version__ = "0.1.9"
+    __version__ = "0.1.10"
 
 __all__ = ["TangleDynamicDiscoveryClient", "__version__"]

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?, isEnabled?
+    annotations?, componentRef, arguments?, isEnabled?, executionOptions?
 
 Argument values are emitted in the runnable ``ArgumentValue`` shape,
 dispatched purely on the VALUE's runtime type — never on the argument
@@ -34,11 +34,18 @@
 """
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
 from .dynamic_data import DynamicData
 from .errors import CompileError, InvalidArgumentTypeError
-from .graph import IS_ENABLED_UNSET, EdgeRef, GraphBuilder, TaskNode
+from .graph import (
+    EXECUTION_OPTIONS_UNSET,
+    IS_ENABLED_UNSET,
+    EdgeRef,
+    GraphBuilder,
+    TaskNode,
+)
 from .placeholders import GraphInputPlaceholder, TaskOutputProxy
 from .raw import Raw
 
@@ -141,7 +148,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?, isEnabled?``.
+    ``annotations?, componentRef, arguments?, isEnabled?, executionOptions?``.
 
     ``task_path`` is this task's dot-delimited JSON path
     (``implementation.graph.tasks.<task_id>``); each argument's path is
@@ -167,6 +174,10 @@ def _emit_task(
     if node.is_enabled is not IS_ENABLED_UNSET:
         body["isEnabled"] = _emit_is_enabled(node.is_enabled)
 
+    execution_options = _emit_execution_options(node)
+    if execution_options is not None:
+        body["executionOptions"] = execution_options
+
     return body
 
 
@@ -316,6 +327,171 @@ def _emit_is_enabled(value: Any) -> Any:
     )
 
 
+# The fields the BACKEND actually models on ``ExecutionOptionsSpec`` —
+# mirrored from ``cloud_pipelines_backend.component_structures`` and vendored
+# into ``schemas/pipeline_schema.json`` (kept honest by
+# ``test_execution_option_fields_match_generated_schema``).
+#
+# This allowlist exists because the backend's pydantic models leave ``extra``
+# unset, i.e. ``extra="ignore"``: an unmodeled key such as ``timeout`` or
+# ``retryStrategy.backoff`` is SILENTLY DROPPED server-side rather than
+# rejected. Accepting one here would let an author believe a timeout/backoff
+# is in effect when nothing applies it, so the passthrough fails closed.
+_EXECUTION_OPTION_FIELDS: dict[str, frozenset[str]] = {
+    "cachingStrategy": frozenset({"maxCacheStaleness"}),
+    "retryStrategy": frozenset({"maxRetries"}),
+}
+
+# ``RetryStrategySpec.max_retries`` is a required (non-Optional) backend field,
+# so a retryStrategy without it is a submit-time validation failure.
+_REQUIRED_EXECUTION_OPTION_FIELDS: dict[str, frozenset[str]] = {
+    "retryStrategy": frozenset({"maxRetries"}),
+}
+
+
+def _emit_execution_options(node: TaskNode) -> dict[str, Any] | None:
+    """Merge the reserved execution-option keywords into ``executionOptions``.
+
+    ``execution_options=`` is the general passthrough for the modeled
+    ``ExecutionOptionsSpec`` (``cachingStrategy`` and ``retryStrategy``).
+    ``max_cache_staleness=`` is the narrow ergonomic knob for the common
+    "do not cache this task" case and WINS over any
+    ``cachingStrategy.maxCacheStaleness`` supplied through the passthrough.
+
+    Returns ``None`` when neither keyword was authored, so the key is omitted
+    entirely. Unlike arguments and ``isEnabled``, execution options are STATIC
+    compile-time settings: graph inputs, task outputs, dynamic data, and raw
+    values are rejected because the backend does not resolve them here. Keys
+    the backend does not model are rejected too — see
+    :data:`_EXECUTION_OPTION_FIELDS`.
+    """
+    options_value = node.execution_options
+    staleness = node.max_cache_staleness
+
+    if (
+        options_value is EXECUTION_OPTIONS_UNSET
+        and staleness is EXECUTION_OPTIONS_UNSET
+    ):
+        return None
+
+    options: dict[str, Any] = {}
+    if options_value is not EXECUTION_OPTIONS_UNSET:
+        if not isinstance(options_value, Mapping):
+            raise InvalidArgumentTypeError(
+                "unsupported execution_options value type "
+                f"{type(options_value).__name__!r}. Task execution options must "
+                "be a mapping such as "
+                '{"cachingStrategy": {"maxCacheStaleness": "P0D"}}.'
+            )
+        options = _normalize_execution_option_value(
+            options_value, "execution_options"
+        )
+        _validate_execution_option_fields(options)
+
+    if staleness is not EXECUTION_OPTIONS_UNSET:
+        if not isinstance(staleness, str):
+            raise InvalidArgumentTypeError(
+                "unsupported max_cache_staleness value type "
+                f"{type(staleness).__name__!r}. Task cache staleness must be a "
+                "duration string such as 'P0D' (never reuse cached results) or "
+                "'P7D'."
+            )
+        caching = options.get("cachingStrategy")
+        if not isinstance(caching, dict):
+            caching = {}
+        # The narrow keyword is the authoritative source for this one field.
+        caching["maxCacheStaleness"] = staleness
+        options["cachingStrategy"] = caching
+
+    if not options:
+        raise CompileError(
+            "execution_options={} is empty; omit the keyword instead of "
+            "passing an empty mapping so the task emits no executionOptions."
+        )
+    return options
+
+
+def _validate_execution_option_fields(options: dict[str, Any]) -> None:
+    """Reject execution-option keys the Tangle backend does not model.
+
+    The backend ignores (silently discards) unknown keys, so a typo or an
+    aspirational field would compile, submit, and quietly do nothing. Failing
+    at compile time keeps the passthrough as fail-closed as every other value
+    the Python authoring surface emits.
+    """
+    for group, value in options.items():
+        known_fields = _EXECUTION_OPTION_FIELDS.get(group)
+        if known_fields is None:
+            supported = ", ".join(sorted(_EXECUTION_OPTION_FIELDS))
+            raise InvalidArgumentTypeError(
+                f"unknown execution_options key {group!r}. Tangle models only "
+                f"{supported}; unmodeled keys are silently ignored by the "
+                "backend, so they are rejected here instead of looking like "
+                "a setting that never takes effect."
+            )
+        if not isinstance(value, dict):
+            raise InvalidArgumentTypeError(
+                f"execution_options.{group} must be a mapping; got "
+                f"{type(value).__name__!r}."
+            )
+        unknown = sorted(set(value) - known_fields)
+        if unknown:
+            supported = ", ".join(sorted(known_fields))
+            raise InvalidArgumentTypeError(
+                f"unknown execution_options.{group} field {unknown[0]!r}. "
+                f"Tangle models only {supported} here; unmodeled keys are "
+                "silently ignored by the backend, so they are rejected at "
+                "compile time."
+            )
+        required = _REQUIRED_EXECUTION_OPTION_FIELDS.get(group, frozenset())
+        missing = sorted(required - set(value))
+        if missing:
+            raise InvalidArgumentTypeError(
+                f"execution_options.{group} requires {missing[0]!r}. The "
+                "backend rejects a partial "
+                f"{group} spec."
+            )
+
+
+def _normalize_execution_option_value(value: Any, path: str) -> Any:
+    """Recursively validate/normalize one static execution-option value.
+
+    Mappings and sequences are copied (so a caller's dict is never mutated by
+    the ``max_cache_staleness`` merge) and tuples become lists so the result is
+    plain YAML-serializable data. Anything else — a task output, graph input,
+    ``dynamic_secret(...)``, ``raw(...)`` value, or an arbitrary object — is
+    rejected: ``executionOptions`` is applied at submit time, not resolved from
+    the running graph.
+    """
+    if isinstance(value, Mapping):
+        normalized: dict[str, Any] = {}
+        for key, item in value.items():
+            if not isinstance(key, str):
+                raise InvalidArgumentTypeError(
+                    f"unsupported execution_options key type "
+                    f"{type(key).__name__!r} at {path}. Execution option keys "
+                    "must be strings."
+                )
+            normalized[key] = _normalize_execution_option_value(
+                item, f"{path}.{key}"
+            )
+        return normalized
+    if isinstance(value, (list, tuple)):
+        return [
+            _normalize_execution_option_value(item, f"{path}[{index}]")
+            for index, item in enumerate(value)
+        ]
+    if value is None or isinstance(value, (str, bool, int, float)):
+        return value
+    raise InvalidArgumentTypeError(
+        f"unsupported execution_options value type {type(value).__name__!r} at "
+        f"{path}. Task execution options are static compile-time settings, so "
+        "only strings, numbers, booleans, null, lists, and nested mappings are "
+        "supported; graph inputs, task outputs, dynamic data, and raw values "
+        "are not evaluated for executionOptions."
+    )
+
+
 def _emit_edge_value(edge: EdgeRef) -> dict[str, Any]:
     """Render an :class:`EdgeRef` as a dehydrated ``ArgumentValue``
     sub-dict (``{taskOutput|graphInput: {...}}``) used in

packages/tangle-cli/src/tangle_cli/python_pipeline/graph.py

@@ -15,6 +15,11 @@
 # instead of being silently omitted.
 IS_ENABLED_UNSET = object()
 
+# Distinguishes omitted task execution options from an explicitly supplied
+# value, so ``execution_options=None`` / ``max_cache_staleness=None`` fail
+# closed in the emitter instead of being silently dropped.
+EXECUTION_OPTIONS_UNSET = object()
+
 
 @dataclass
 class EdgeRef:
@@ -37,6 +42,11 @@ class TaskNode:
     ``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.
+
+    ``execution_options`` (general passthrough) and ``max_cache_staleness``
+    (narrow ergonomic knob) are likewise task metadata, not component inputs.
+    The emitter merges them — the narrow keyword wins — and serializes the
+    result as the canonical ``executionOptions`` task field.
     """
 
     task_id: str
@@ -46,6 +56,8 @@ class TaskNode:
     arguments: dict[str, Any] = field(default_factory=dict)
     annotations: dict[str, str] | None = None
     is_enabled: Any = IS_ENABLED_UNSET
+    execution_options: Any = EXECUTION_OPTIONS_UNSET
+    max_cache_staleness: Any = EXECUTION_OPTIONS_UNSET
 
 
 @dataclass