feat(python-pipeline): support dynamic_secret runtime-secret arguments

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

@@ -20,6 +20,7 @@
 """
 from __future__ import annotations
 
+from .dynamic_data import dynamic_secret
 from .pipeline import pipeline
 from .raw import raw
 from .ref import ref
@@ -35,6 +36,7 @@
     "registered",
     "ref",
     "raw",
+    "dynamic_secret",
     "subpipeline",
     "TaskEnv",
     "In",

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

@@ -0,0 +1,44 @@
+"""Dynamic-data argument helpers for Python-authored pipelines.
+
+Tangle runnable arguments can be literals, graph/task edges, or dynamic data
+resolved by the runtime (for example a secret reference).  The Python pipeline
+emitter needs an explicit wrapper so author code can request a dynamicData
+argument without opening support for arbitrary dict constants.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+
+@dataclass(frozen=True)
+class DynamicData:
+    """A task argument value emitted as ``{"dynamicData": value}``.
+
+    Internal emit primitive — not part of the public authoring surface (see
+    ``__init__.__all__``). ``value`` is a general mapping because the runtime
+    resolves several dynamic-data kinds (secrets, run IDs, loop indices), but
+    the only public producer today is :func:`dynamic_secret`, which always
+    builds the ``{"secret": {"name": ...}}`` shape that the strict dehydrated
+    schema accepts. Any other shape emits but fails closed at compile
+    validation until a new kind is added to both the constructor surface and
+    the schema together — so the public API never promises more than the
+    schema enforces.
+    """
+
+    value: Mapping[str, Any]
+
+
+def dynamic_secret(name: str) -> DynamicData:
+    """Reference a runtime secret by name for a task argument.
+
+    Example emitted YAML::
+
+        openai_api_key:
+          dynamicData:
+            secret:
+              name: OPENAI_API_KEY
+    """
+    if not isinstance(name, str) or not name:
+        raise ValueError("dynamic_secret() requires a non-empty secret name string")
+    return DynamicData({"secret": {"name": name}})

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

@@ -20,11 +20,14 @@
   also recorded in the returned exempt-paths set so the no-template-delimiter
   output guard skips that one location (it is a legitimate RUNTIME
   placeholder, e.g. a run-query ``{{input_1}}`` sentinel).
+* a :class:`DynamicData` value    → ``{"dynamicData": value}`` for runtime-
+  resolved arguments such as secrets.
 
 Non-string constants are rejected: the runnable Tangle argument contract
-only supports string constants, ``graphInput``, or ``taskOutput``.
-Structured/non-string values must be stringified explicitly in pipeline
-code (e.g. ``json.dumps(...)``) before they reach the compiler.
+only supports string constants, ``graphInput``, ``taskOutput``, or explicit
+``dynamicData`` wrappers. Structured/non-string values must be stringified
+explicitly in pipeline code (e.g. ``json.dumps(...)``) before they reach the
+compiler.
 
 The literal key the user wrote (``wait_for``, ``depends_on``,
 ``project``, ``payload``, ...) is preserved verbatim as the dict key.
@@ -33,6 +36,7 @@
 
 from typing import Any
 
+from .dynamic_data import DynamicData
 from .errors import CompileError, InvalidArgumentTypeError
 from .graph import EdgeRef, GraphBuilder, TaskNode
 from .placeholders import GraphInputPlaceholder, TaskOutputProxy
@@ -219,7 +223,8 @@ def _emit_argument_value(
     Dispatch is purely on the VALUE's runtime type — the argument *key*
     is never inspected. Produces a ``taskOutput`` / ``graphInput`` wrapper
     for edges, or the RAW string for a constant (matching the runnable
-    Tangle argument contract). Non-string constants are rejected.
+    Tangle argument contract). Non-string constants are rejected unless they
+    are explicit dynamic-data wrappers.
 
     A :class:`Raw` value is emitted as its inner string verbatim — exactly
     like a plain ``str`` constant — and ``arg_path`` (this argument's
@@ -244,6 +249,8 @@ def _emit_argument_value(
         # recorded here, at the only point the Raw wrapper is still visible.
         exempt_paths.add(arg_path)
         return value.value
+    if isinstance(value, DynamicData):
+        return {"dynamicData": value.value}
     # Everything else is a constant. The runnable schema only accepts raw
     # string constants, so validate and emit the string verbatim.
     _validate_constant(value, key)
@@ -254,10 +261,10 @@ def _validate_constant(value: Any, key: str) -> None:
     """Assert ``value`` is a runnable string constant.
 
     Runnable Tangle pipeline arguments only support raw ``str`` constants
-    (alongside the ``graphInput`` / ``taskOutput`` wrappers). A non-string
-    constant (``int``, ``float``, ``bool``, ``None``, ``list``, ``dict``,
-    a tuple/set, a leftover helper object, a callable, ...) cannot be
-    represented under the runnable schema, so it is rejected with a
+    (alongside the ``graphInput`` / ``taskOutput`` / explicit ``dynamicData``
+    wrappers). A non-string constant (``int``, ``float``, ``bool``, ``None``,
+    ``list``, ``dict``, a tuple/set, a leftover helper object, a callable, ...)
+    cannot be represented under the runnable schema, so it is rejected with a
     GENERIC, operation-agnostic message — the compiler has no
     operation-specific knowledge (no SQL, BigQuery, or other domain
     awareness). Authors must stringify structured/non-string values
@@ -268,7 +275,8 @@ def _validate_constant(value: Any, key: str) -> None:
     raise InvalidArgumentTypeError(
         f"unsupported constant type {type(value).__name__!r} for "
         f"argument {key!r}. Runnable Tangle pipeline arguments only support "
-        "string constants, graphInput, or taskOutput. Convert structured or "
+        "string constants, graphInput, or taskOutput; dynamicData wrappers are also supported. "
+        "Convert structured or "
         "non-string values to a string explicitly in your pipeline code "
         "(for example json.dumps(...) or str(...)) before passing them as "
         "task arguments."

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

@@ -19,7 +19,7 @@
 * :func:`is_dehydrated_pipeline` — shape detector (no raise): top-level
   ``name`` + ``implementation.graph.tasks``, no ``template_file``, and
   task ``arguments`` values that are raw string constants or ``graphInput``
-  / ``taskOutput`` wrappers.
+  / ``taskOutput`` / ``dynamicData`` wrappers.
 * :func:`validate_dehydrated_pipeline` — JSON-Schema validation PLUS the
   deeper semantic checks jsonschema cannot express cleanly (dangling
   ``taskOutput.taskId``, undeclared ``graphInput.inputName``,
@@ -207,25 +207,25 @@ def assert_no_template_delimiters(
     {"name", "description", "metadata", "inputs", "outputs", "implementation"}
 )
 
-# The reference-only ArgumentValue wrappers. A constant is NOT a wrapper —
-# it is a raw string (matching the runnable Tangle argument contract).
-_REFERENCE_ARGUMENT_KEYS = ("graphInput", "taskOutput")
+# ArgumentValue wrappers. A constant is NOT a wrapper — it is a raw string
+# (matching the runnable Tangle argument contract).
+_ARGUMENT_WRAPPER_KEYS = ("graphInput", "taskOutput", "dynamicData")
 
 
 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``
-    wrapper.
+    string constant, or a mapping carrying a ``graphInput`` / ``taskOutput`` /
+    ``dynamicData`` wrapper.
 
     There is no ambiguity: a raw string constant (even one whose text is
     ``"graphInput"`` or JSON like ``'{"graphInput": ...}'``) is a string,
     never the object wrapper shapes — so it can never collide with a
-    ``graphInput`` / ``taskOutput`` mapping.
+    ``graphInput`` / ``taskOutput`` / ``dynamicData`` mapping.
     """
     if isinstance(value, str):
         return True
     return isinstance(value, Mapping) and any(
-        key in value for key in _REFERENCE_ARGUMENT_KEYS
+        key in value for key in _ARGUMENT_WRAPPER_KEYS
     )
 
 
@@ -238,7 +238,7 @@ def is_dehydrated_pipeline(data: Any) -> bool:
     * NO ``template_file`` (it is the final rendered form, not a wrapper);
     * task ``arguments`` values AND graph ``outputValues`` values (when
       present) that are raw string constants or ``graphInput`` /
-      ``taskOutput`` wrappers — a non-string raw value (a bare
+      ``taskOutput`` / ``dynamicData`` wrappers — a non-string raw value (a bare
       number/list/object) or a legacy ``{constantValue: ...}`` wrapper is
       not a runnable argument value, so it means the input is not yet
       dehydrated.

packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json

@@ -169,11 +169,12 @@
     },
 
     "ArgumentValue": {
-      "description": "Source of a task argument or graph output. Each argument has EXACTLY ONE source: a raw string constant, a `graphInput` wrapper, or a `taskOutput` wrapper (mixing sources in one object is rejected). The `graphInput` / `taskOutput` wrappers use the same shapes as runnable Tangle pipelines. The ONLY dehydrated-vs-runnable difference is `componentRef` (a lightweight ref) vs an inline `componentSpec`; argument values are identical to the runnable contract.",
+      "description": "Source of a task argument or graph output. Each argument has EXACTLY ONE source: a raw string constant, a `graphInput` wrapper, a `taskOutput` wrapper, or a `dynamicData` wrapper (mixing sources in one object is rejected). The `graphInput` / `taskOutput` / `dynamicData` wrappers use the same shapes as runnable Tangle pipelines. The ONLY dehydrated-vs-runnable difference is `componentRef` (a lightweight ref) vs an inline `componentSpec`; argument values are identical to the runnable contract.",
       "oneOf": [
         { "type": "string" },
         { "$ref": "#/$defs/GraphInputArgument" },
-        { "$ref": "#/$defs/TaskOutputArgument" }
+        { "$ref": "#/$defs/TaskOutputArgument" },
+        { "$ref": "#/$defs/DynamicDataArgument" }
       ]
     },
 
@@ -221,6 +222,33 @@
       }
     },
 
+    "DynamicDataArgument": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["dynamicData"],
+      "properties": {
+        "dynamicData": { "$ref": "#/$defs/DynamicSecretData" }
+      }
+    },
+
+    "DynamicSecretData": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["secret"],
+      "properties": {
+        "secret": { "$ref": "#/$defs/DynamicSecretReference" }
+      }
+    },
+
+    "DynamicSecretReference": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["name"],
+      "properties": {
+        "name": { "type": "string", "minLength": 1 }
+      }
+    },
+
     "ExecutionOptionsSpec": {
       "type": "object",
       "additionalProperties": true,

tests/fixtures/python_pipeline/dynamic_secret_pipeline.py

@@ -0,0 +1,10 @@
+"""Minimal compile fixture for a task argument sourced from a runtime secret."""
+
+from tangle_cli.python_pipeline import dynamic_secret, pipeline, ref
+
+MODEL = ref(name="model")
+
+
+@pipeline(name="Dynamic Secret")
+def dynamic_secret_pipeline() -> None:
+    MODEL.named("Call Model")(openai_api_key=dynamic_secret("OPENAI_API_KEY"))

tests/test_python_pipeline_dsl.py

@@ -56,6 +56,7 @@ def test_all_names_are_exported(self):
             "registered",
             "ref",
             "raw",
+            "dynamic_secret",
             "subpipeline",
             "TaskEnv",
             "In",

tests/test_python_pipeline_dynamic_data.py

@@ -0,0 +1,79 @@
+"""Tests for ``dynamicData`` argument wrappers in Python-authored pipelines.
+
+The dehydrated-pipeline schema is deliberately strict: a ``dynamicData``
+argument must be a non-empty ``secret.name`` reference and nothing else.
+(The runnable ``pipeline_schema.json`` stays intentionally permissive — it
+mirrors the real Tangle runtime, which resolves arbitrary dynamic data such
+as run IDs and loop indices — so strictness is enforced here, at compile /
+dehydrated-validation time.)
+"""
+import copy
+from pathlib import Path
+
+import pytest
+import yaml
+
+from tangle_cli.pipeline_compiler import compile_pipeline
+from tangle_cli.python_pipeline.dynamic_data import dynamic_secret
+from tangle_cli.python_pipeline.emit import emit_pipeline
+from tangle_cli.python_pipeline.graph import GraphBuilder, TaskNode
+from tangle_cli.schema_validation import SchemaValidationError, validate_dehydrated_pipeline
+
+FIXTURES = Path(__file__).parent / "fixtures" / "python_pipeline"
+
+
+def _body_with_dynamic_secret() -> dict:
+    builder = GraphBuilder(name="Demo")
+    builder.add_task(
+        TaskNode(
+            task_id="Call Model",
+            ref_url="file://./noop.yaml",
+            arguments={"openai_api_key": dynamic_secret("OPENAI_API_KEY")},
+        )
+    )
+    return emit_pipeline(builder)[0]
+
+
+def test_dynamic_secret_emits_dynamic_data_argument():
+    body = _body_with_dynamic_secret()
+
+    emitted = body["implementation"]["graph"]["tasks"]["Call Model"]["arguments"]["openai_api_key"]
+    assert emitted == {"dynamicData": {"secret": {"name": "OPENAI_API_KEY"}}}
+    validate_dehydrated_pipeline(body)
+
+
+def test_dynamic_secret_fixture_compiles_and_validates(tmp_path):
+    """Keep a real Python pipeline example alongside the dynamic-data tests."""
+    output = tmp_path / "dynamic_secret.yaml"
+    compile_pipeline(FIXTURES / "dynamic_secret_pipeline.py", output)
+    body = yaml.safe_load(output.read_text())
+
+    argument = body["implementation"]["graph"]["tasks"]["Call Model"]["arguments"]
+    assert argument["openai_api_key"] == {
+        "dynamicData": {"secret": {"name": "OPENAI_API_KEY"}}
+    }
+    validate_dehydrated_pipeline(body)
+
+
+@pytest.mark.parametrize(
+    "dynamic_data",
+    [
+        {},
+        {"secret": {}},
+        {"secret": {"name": ""}},
+        {"arbitrary": [1]},
+        {"secret": {"name": "OPENAI_API_KEY", "extra": "value"}},
+    ],
+)
+def test_dynamic_data_rejects_invalid_secret_references(dynamic_data: dict):
+    body = _body_with_dynamic_secret()
+    argument = body["implementation"]["graph"]["tasks"]["Call Model"]["arguments"]["openai_api_key"]
+    argument["dynamicData"] = copy.deepcopy(dynamic_data)
+
+    with pytest.raises(SchemaValidationError):
+        validate_dehydrated_pipeline(body)
+
+
+def test_dynamic_secret_requires_non_empty_name():
+    with pytest.raises(ValueError, match="non-empty secret name"):
+        dynamic_secret("")