diff --git a/README.md b/README.md
index 81fe884..efefc21 100644
--- a/README.md
+++ b/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`:
diff --git a/examples/python_pipeline/execution_options_pipeline.py b/examples/python_pipeline/execution_options_pipeline.py
new file mode 100644
index 0000000..3f346ca
--- /dev/null
+++ b/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
diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py
index 3e558f9..6801f53 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.9"
+    __version__ = "0.1.10"
 
 __all__ = ["TangleDynamicDiscoveryClient", "__version__"]
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 fabbe17..a255f2d 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?, 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
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 9ef2446..1a2bc15 100644
--- a/packages/tangle-cli/src/tangle_cli/python_pipeline/graph.py
+++ b/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
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 8ed1adc..fa0891f 100644
--- a/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py
+++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py
@@ -16,7 +16,7 @@
 from typing import Any
 
 from .errors import CompileError
-from .graph import IS_ENABLED_UNSET
+from .graph import EXECUTION_OPTIONS_UNSET, IS_ENABLED_UNSET
 
 _UNWRAPPED_KEY_RE = re.compile(r"^[A-Za-z0-9_-]+$")
 
@@ -296,6 +296,8 @@ def __call__(
         self,
         *,
         is_enabled: Any = IS_ENABLED_UNSET,
+        execution_options: Any = EXECUTION_OPTIONS_UNSET,
+        max_cache_staleness: Any = EXECUTION_OPTIONS_UNSET,
         **kwargs: Any,
     ) -> Any:
         """Trace-mode invocation.
@@ -313,7 +315,21 @@ def __call__(
         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`` /
+        so both may be used on the same task.
+
+        ``execution_options`` and ``max_cache_staleness`` follow the same
+        reserved-keyword convention and are emitted as the canonical
+        ``executionOptions`` task field. ``execution_options`` takes the whole
+        ``ExecutionOptionsSpec`` mapping (``cachingStrategy`` and
+        ``retryStrategy`` — the only groups Tangle models);
+        ``max_cache_staleness`` is the narrow knob for
+        ``cachingStrategy.maxCacheStaleness`` (e.g. ``"P0D"`` to never reuse
+        cached results) and wins when both supply that field.
+        A component input named ``execution_options`` or
+        ``max_cache_staleness`` is bound the same way, with
+        ``ref(...).bind(max_cache_staleness=...)``.
+
+        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
@@ -377,6 +393,8 @@ def __call__(
             arguments=merged,
             annotations=dict(self.annotations) if self.annotations else None,
             is_enabled=is_enabled,
+            execution_options=execution_options,
+            max_cache_staleness=max_cache_staleness,
         )
         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 0f9968e..823f953 100644
--- a/packages/tangle-cli/src/tangle_cli/python_pipeline/subpipeline.py
+++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/subpipeline.py
@@ -16,8 +16,9 @@
 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`. Tangle only
-supports conditional execution for container-component tasks, so the reserved
-call-site ``is_enabled=`` metadata keyword is rejected on subpipeline boundary
+supports conditional execution and execution options for container-component
+tasks, so the reserved call-site ``is_enabled=``, ``execution_options=`` and
+``max_cache_staleness=`` metadata keywords are rejected on subpipeline boundary
 (graph-component) tasks.
 
 The child body is NOT executed into the parent's :class:`GraphBuilder`.
@@ -128,6 +129,10 @@ def __call__(self, **kwargs: Any) -> "TaskOutputProxy":
         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``.
+        ``execution_options`` / ``max_cache_staleness`` are rejected for the
+        same reason: caching and retries apply to the container executions
+        inside the child graph, so silently emitting them on the boundary task
+        would look like a setting that is not actually applied.
         """
         import sys
 
@@ -154,6 +159,17 @@ def __call__(self, **kwargs: Any) -> "TaskOutputProxy":
                 "input with .bind(is_enabled=...)."
             )
 
+        for reserved in ("execution_options", "max_cache_staleness"):
+            if reserved in kwargs:
+                raise CompileError(
+                    f"subpipeline tasks do not support call-site {reserved}= "
+                    "because Tangle execution options (caching, retries) "
+                    "apply to container-component tasks. Set "
+                    f"{reserved}= on the tasks inside the child pipeline. If "
+                    f"the child declares a graph input named {reserved!r}, pass "
+                    f"that input with .bind({reserved}=...)."
+                )
+
         # 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/schemas/dehydrated_pipeline_schema.json b/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json
index d667ff1..a104bd0 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
@@ -263,8 +263,7 @@
       "additionalProperties": true,
       "properties": {
         "cachingStrategy": { "$ref": "#/$defs/CachingStrategySpec" },
-        "retryStrategy":   { "$ref": "#/$defs/RetryStrategySpec" },
-        "timeout":         { "type": "string", "description": "Go-style duration, e.g. `30m`, `2h`." }
+        "retryStrategy":   { "$ref": "#/$defs/RetryStrategySpec" }
       }
     },
 
@@ -274,7 +273,7 @@
       "properties": {
         "maxCacheStaleness": {
           "type": ["string", "null"],
-          "description": "Maximum allowed cache age, e.g. `P7D` (ISO-8601) or `7d`."
+          "description": "Maximum allowed cache age as an RFC3339 duration, e.g. `P7D`; `P0D` never reuses a cached result."
         }
       }
     },
@@ -282,9 +281,9 @@
     "RetryStrategySpec": {
       "type": "object",
       "additionalProperties": true,
+      "required": ["maxRetries"],
       "properties": {
-        "maxRetries": { "type": "integer", "minimum": 0 },
-        "backoff":    { "type": "string", "description": "Go-style duration, e.g. `30s`." }
+        "maxRetries": { "type": "integer", "minimum": 0 }
       }
     }
   }
diff --git a/pyproject.toml b/pyproject.toml
index e0dba17..9ad413e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "tangle-cli"
-version = "0.1.9"
+version = "0.1.10"
 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 5a9d414..80e401c 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.9" in metadata
+    assert "Version: 0.1.10" 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 bf0f083..eacb559 100644
--- a/tests/test_pipeline_compiler.py
+++ b/tests/test_pipeline_compiler.py
@@ -7,6 +7,7 @@
 cyclopts ``compile`` command (``tangle sdk pipelines compile``) and the
 ``compile_pipeline_file`` facade.
 """
+import json
 import shutil
 import sys
 from pathlib import Path
@@ -14,6 +15,7 @@
 import pytest
 import yaml
 
+import tangle_cli
 from tangle_cli import cli
 from tangle_cli.pipeline_compiler import (
     IMAGE_IDS,
@@ -818,6 +820,252 @@ def test_compile_rejects_unsupported_is_enabled_values(
     assert not out.exists()
 
 
+def test_compile_emits_execution_options_and_collision_escape(tmp_path):
+    """Task execution options use canonical ``executionOptions`` without
+    stealing a bound component input of the same Python name."""
+    out = tmp_path / "compiled.yaml"
+    _provide_noop(out)
+    src = tmp_path / "execution_options_pipeline.py"
+    src.write_text(
+        "from tangle_cli.python_pipeline import Out, pipeline, ref\n"
+        "\n"
+        "@pipeline('Execution Options Pipeline')\n"
+        "def execution_options_pipeline() -> Out[str]:\n"
+        "    no_cache = ref(url='file://./noop.yaml')(\n"
+        "        max_cache_staleness='P0D'\n"
+        "    )\n"
+        "    passthrough = ref(url='file://./noop.yaml')(\n"
+        "        execution_options={'retryStrategy': {'maxRetries': 3}}\n"
+        "    )\n"
+        "    narrow_wins = ref(url='file://./noop.yaml')(\n"
+        "        execution_options={\n"
+        "            'retryStrategy': {'maxRetries': 1},\n"
+        "            'cachingStrategy': {'maxCacheStaleness': 'P7D'},\n"
+        "        },\n"
+        "        max_cache_staleness='P0D',\n"
+        "    )\n"
+        "    plain = ref(url='file://./noop.yaml')(message='plain')\n"
+        "    collision = ref(url='file://./noop.yaml').bind(\n"
+        "        max_cache_staleness='component-value'\n"
+        "    )(max_cache_staleness='P0D', depends_on=plain)\n"
+        "    return collision\n",
+        encoding="utf-8",
+    )
+
+    compile_pipeline(src, out)
+    tasks = yaml.safe_load(out.read_text())["implementation"]["graph"]["tasks"]
+
+    assert tasks["No Cache"]["executionOptions"] == {
+        "cachingStrategy": {"maxCacheStaleness": "P0D"}
+    }
+    assert tasks["Passthrough"]["executionOptions"] == {
+        "retryStrategy": {"maxRetries": 3},
+    }
+    # The narrow keyword overrides the passthrough's cache field and leaves
+    # every other execution option intact.
+    assert tasks["Narrow Wins"]["executionOptions"] == {
+        "retryStrategy": {"maxRetries": 1},
+        "cachingStrategy": {"maxCacheStaleness": "P0D"},
+    }
+    # A bound component input of the same name stays a component argument.
+    assert (
+        tasks["Collision"]["arguments"]["max_cache_staleness"]
+        == "component-value"
+    )
+    assert tasks["Collision"]["executionOptions"] == {
+        "cachingStrategy": {"maxCacheStaleness": "P0D"}
+    }
+    assert "executionOptions" not in tasks["Plain"]
+
+
+def test_compile_omits_execution_options_when_not_authored(multi_arg_args):
+    data, _produce, _consume = multi_arg_args
+    tasks = data["implementation"]["graph"]["tasks"]
+    assert all("executionOptions" not in task for task in tasks.values())
+
+
+@pytest.mark.parametrize(
+    ("expression", "message_fragment"),
+    [
+        ("max_cache_staleness=None", "unsupported max_cache_staleness value type"),
+        ("max_cache_staleness=0", "unsupported max_cache_staleness value type"),
+        (
+            "max_cache_staleness=dynamic_secret('TOKEN')",
+            "unsupported max_cache_staleness value type",
+        ),
+        ("execution_options=None", "unsupported execution_options value type"),
+        ("execution_options='P0D'", "unsupported execution_options value type"),
+        (
+            "execution_options={'cachingStrategy': raw('{{runtime}}')}",
+            "unsupported execution_options value type",
+        ),
+        (
+            "execution_options={'cachingStrategy': dynamic_secret('TOKEN')}",
+            "unsupported execution_options value type",
+        ),
+        ("execution_options={}", "execution_options={} is empty"),
+        # Keys the backend does not model are silently dropped server-side,
+        # so the authoring surface must reject them instead of advertising a
+        # setting that never takes effect.
+        (
+            "execution_options={'timeout': '30m'}",
+            "unknown execution_options key 'timeout'",
+        ),
+        (
+            "execution_options={'retryStrategy': {'maxRetries': 3, "
+            "'backoff': '30s'}}",
+            "unknown execution_options.retryStrategy field 'backoff'",
+        ),
+        (
+            "execution_options={'cachingStrategy': {'maxStaleness': 'P0D'}}",
+            "unknown execution_options.cachingStrategy field 'maxStaleness'",
+        ),
+        # ``maxRetries`` is a required backend field.
+        (
+            "execution_options={'retryStrategy': {}}",
+            "execution_options.retryStrategy requires 'maxRetries'",
+        ),
+        # A malformed group must be reported, never silently replaced by the
+        # narrow keyword's merge.
+        (
+            "execution_options={'cachingStrategy': 'bad'}",
+            "execution_options.cachingStrategy must be a mapping",
+        ),
+        (
+            "execution_options={'cachingStrategy': None}, "
+            "max_cache_staleness='P0D'",
+            "execution_options.cachingStrategy must be a mapping",
+        ),
+        (
+            "execution_options={'cachingStrategy': 'bad'}, "
+            "max_cache_staleness='P0D'",
+            "execution_options.cachingStrategy must be a mapping",
+        ),
+    ],
+)
+def test_compile_rejects_unsupported_execution_options(
+    expression, message_fragment, tmp_path
+):
+    out = tmp_path / "compiled.yaml"
+    _provide_noop(out)
+    src = tmp_path / "bad_execution_options_pipeline.py"
+    src.write_text(
+        "from tangle_cli.python_pipeline import (\n"
+        "    Out, dynamic_secret, pipeline, raw, ref,\n"
+        ")\n"
+        "\n"
+        "@pipeline('Bad Execution Options Pipeline')\n"
+        "def bad_execution_options_pipeline() -> Out[str]:\n"
+        f"    bad = ref(url='file://./noop.yaml')({expression})\n"
+        "    return bad\n",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(CompileError) as exc:
+        compile_pipeline(src, out)
+
+    assert message_fragment in str(exc.value)
+    assert not out.exists()
+
+
+def test_compile_execution_options_does_not_mutate_authored_mapping(tmp_path):
+    """The ``max_cache_staleness`` merge must not write back into a shared
+    mapping the pipeline author reuses across tasks."""
+    out = tmp_path / "compiled.yaml"
+    _provide_noop(out)
+    src = tmp_path / "shared_execution_options_pipeline.py"
+    src.write_text(
+        "from tangle_cli.python_pipeline import Out, pipeline, ref\n"
+        "\n"
+        "SHARED = {'retryStrategy': {'maxRetries': 3}}\n"
+        "\n"
+        "@pipeline('Shared Execution Options Pipeline')\n"
+        "def shared_execution_options_pipeline() -> Out[str]:\n"
+        "    first = ref(url='file://./noop.yaml')(\n"
+        "        execution_options=SHARED, max_cache_staleness='P0D'\n"
+        "    )\n"
+        "    second = ref(url='file://./noop.yaml')(\n"
+        "        execution_options=SHARED, depends_on=first\n"
+        "    )\n"
+        "    return second\n",
+        encoding="utf-8",
+    )
+
+    compile_pipeline(src, out)
+    tasks = yaml.safe_load(out.read_text())["implementation"]["graph"]["tasks"]
+
+    assert tasks["First"]["executionOptions"] == {
+        "retryStrategy": {"maxRetries": 3},
+        "cachingStrategy": {"maxCacheStaleness": "P0D"},
+    }
+    assert tasks["Second"]["executionOptions"] == {
+        "retryStrategy": {"maxRetries": 3}
+    }
+
+
+def test_execution_option_fields_match_generated_schema():
+    """The emit-time allowlist must track the schema generated from the pinned
+    backend models, so a backend field addition cannot silently stay
+    unauthorable (and a removed one cannot stay advertised).
+
+    ``pipeline_schema.json`` is regenerated by
+    ``scripts/refresh_pipeline_schema.py`` from
+    ``cloud_pipelines_backend.component_structures``, so it — not the
+    hand-maintained dehydrated schema — is the source of truth here.
+    """
+    from tangle_cli.python_pipeline.emit import (
+        _EXECUTION_OPTION_FIELDS,
+        _REQUIRED_EXECUTION_OPTION_FIELDS,
+    )
+
+    schema_path = (
+        Path(tangle_cli.__file__).parent / "schemas" / "pipeline_schema.json"
+    )
+    defs = json.loads(schema_path.read_text(encoding="utf-8"))["$defs"]
+
+    expected_groups = set(defs["ExecutionOptionsSpec"]["properties"])
+    assert set(_EXECUTION_OPTION_FIELDS) == expected_groups
+
+    group_to_spec = {
+        "cachingStrategy": "CachingStrategySpec",
+        "retryStrategy": "RetryStrategySpec",
+    }
+    for group, spec_name in group_to_spec.items():
+        spec = defs[spec_name]
+        assert _EXECUTION_OPTION_FIELDS[group] == frozenset(spec["properties"])
+        assert _REQUIRED_EXECUTION_OPTION_FIELDS.get(
+            group, frozenset()
+        ) == frozenset(spec.get("required", ()))
+
+
+def test_compile_emits_execution_options_for_task_authored_components(tmp_path):
+    """``@task``-authored tasks take the same reserved keyword as ``ref(...)``."""
+    src = tmp_path / "task_execution_options_pipeline.py"
+    src.write_text(
+        "from tangle_cli.python_pipeline import Out, pipeline, task\n"
+        "\n"
+        "@task(image='python:3.12')\n"
+        "def read_runtime_env(name: str) -> str:\n"
+        "    return name\n"
+        "\n"
+        "@pipeline('Task Execution Options Pipeline')\n"
+        "def task_execution_options_pipeline() -> Out[str]:\n"
+        "    gate = read_runtime_env(\n"
+        "        name='CREATED_BY', max_cache_staleness='P0D'\n"
+        "    )\n"
+        "    return gate.Output\n",
+        encoding="utf-8",
+    )
+
+    out = tmp_path / "compiled.yaml"
+    compile_pipeline(src, out)
+    tasks = yaml.safe_load(out.read_text())["implementation"]["graph"]["tasks"]
+
+    assert tasks["Gate"]["executionOptions"] == {
+        "cachingStrategy": {"maxCacheStaleness": "P0D"}
+    }
+
+
 # ---------------------------------------------------------------------------
 # PipelineCompiler handler + ZONE_ROOT_MARKERS seam.
 
@@ -1161,6 +1409,38 @@ def test_compile_subpipeline_rejects_is_enabled_task_metadata(tmp_path):
     assert ".bind(is_enabled=...)" in message
 
 
+@pytest.mark.parametrize(
+    ("reserved", "expression"),
+    [
+        ("execution_options", "execution_options={'timeout': '30m'}"),
+        ("max_cache_staleness", "max_cache_staleness='P0D'"),
+    ],
+)
+def test_compile_subpipeline_rejects_execution_options_task_metadata(
+    reserved, expression, tmp_path
+):
+    src = tmp_path / "execution_options_subpipeline.py"
+    source = (FIXTURES / "subpipeline_pipeline.py").read_text(encoding="utf-8")
+    src.write_text(
+        source.replace(
+            "(seed=parent_wait_token)",
+            f"(seed=parent_wait_token, {expression})",
+        ),
+        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 f"subpipeline tasks do not support call-site {reserved}=" in message
+    assert "container-component tasks" in message
+    assert f".bind({reserved}=...)" 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(
diff --git a/tests/test_pipeline_dehydrator.py b/tests/test_pipeline_dehydrator.py
index ea515cb..cafdf57 100644
--- a/tests/test_pipeline_dehydrator.py
+++ b/tests/test_pipeline_dehydrator.py
@@ -92,6 +92,32 @@ def test_pipeline_dehydrator_preserves_is_enabled_round_trip(tmp_path: Path) ->
     }
 
 
+def test_pipeline_dehydrator_preserves_execution_options_round_trip(
+    tmp_path: Path,
+) -> None:
+    task = _task(
+        "Leaf Component",
+        "digest-1",
+        canonical_url="https://example.test/leaf.yaml",
+    )
+    task["executionOptions"] = {
+        "cachingStrategy": {"maxCacheStaleness": "P0D"},
+        "retryStrategy": {"maxRetries": 3},
+    }
+    data = _pipeline({"task": task})
+
+    result = PipelineDehydrator(
+        {"": DehydrateChoice.URL}, output_file=tmp_path / "out.yaml"
+    ).dehydrate(data)
+
+    assert result["implementation"]["graph"]["tasks"]["task"][
+        "executionOptions"
+    ] == {
+        "cachingStrategy": {"maxCacheStaleness": "P0D"},
+        "retryStrategy": {"maxRetries": 3},
+    }
+
+
 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_schema_validation.py b/tests/test_schema_validation.py
index 1f067db..d81939a 100644
--- a/tests/test_schema_validation.py
+++ b/tests/test_schema_validation.py
@@ -131,6 +131,52 @@ def test_validate_dehydrated_data_rejects_unsupported_is_enabled(condition):
         validate_dehydrated_data(data)
 
 
+@pytest.mark.parametrize(
+    "execution_options",
+    [
+        {"cachingStrategy": {"maxCacheStaleness": "P0D"}},
+        {"cachingStrategy": {"maxCacheStaleness": None}},
+        {"retryStrategy": {"maxRetries": 0}},
+        {
+            "cachingStrategy": {"maxCacheStaleness": "P7D"},
+            "retryStrategy": {"maxRetries": 3},
+        },
+    ],
+)
+def test_validate_dehydrated_data_accepts_supported_execution_options(
+    execution_options,
+):
+    data = _valid_pipeline()
+    data["implementation"]["graph"]["tasks"]["load"][
+        "executionOptions"
+    ] = execution_options
+    validate_dehydrated_data(data)
+
+
+@pytest.mark.parametrize(
+    "execution_options",
+    [
+        # ``maxRetries`` is required by the backend's RetryStrategySpec, so a
+        # partial retry strategy must not pass the dehydrated schema either.
+        {"retryStrategy": {}},
+        {"retryStrategy": {"backoff": "30s"}},
+        {"retryStrategy": {"maxRetries": -1}},
+        {"retryStrategy": {"maxRetries": "3"}},
+        {"cachingStrategy": {"maxCacheStaleness": 0}},
+        {"cachingStrategy": "P0D"},
+    ],
+)
+def test_validate_dehydrated_data_rejects_unsupported_execution_options(
+    execution_options,
+):
+    data = _valid_pipeline()
+    data["implementation"]["graph"]["tasks"]["load"][
+        "executionOptions"
+    ] = execution_options
+    with pytest.raises(SchemaValidationError):
+        validate_dehydrated_data(data)
+
+
 # ---------------------------------------------------------------------------
 # Template-delimiter output contract.
 
diff --git a/uv.lock b/uv.lock
index bc24095..63a6d17 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.9"
+version = "0.1.10"
 source = { editable = "." }
 dependencies = [
     { name = "cloud-pipelines" },