This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new b476b22f264 Support TaskFlow call syntax on stub tasks for the Lang
SDK (#69757)
b476b22f264 is described below
commit b476b22f264a5de9736319ee9fc5843ac0b27eb2
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Wed Aug 12 15:25:16 2026 +0800
Support TaskFlow call syntax on stub tasks for the Lang SDK (#69757)
* Support TaskFlow call syntax on @task.stub tasks
Stub tasks silently ignored TaskFlow call arguments, so a Dag author
could not hand literals or upstream XCom results to a lang-SDK runtime.
The decorator now binds the call to the stub's signature at parse time
and captures an ordered arg spec (literal values and direct upstream
XCom references, with pydantic-derived JSON value schemas) that
serializes with the Dag, while rejecting what cannot cross the language
boundary: custom XCom keys, aggregated mapped outputs, non-JSON
literals, and stubs with arguments inside mapped task groups. Mapped
(.expand()) stubs capture no spec and keep the legacy behavior until a
follow-up delivers per-map-index bindings.
* Ship stub arg_bindings in a new execution API version
TIRunContext gains an arg_bindings field so a lang-SDK runtime receives
the stub task's TaskFlow arg spec at startup. ti_run derives it from the
serialized Dag only for stub operators, so regular tasks never pay for
the lookup, and only for clients on the new API version -- gated on the
Cadwyn VersionChangeWithSideEffects.is_applied check rather than a date
comparison -- so stub Dags that predate arg bindings keep running
against older clients, for which the version migration strips the field.
* Deliver stub arg bindings to SDK runtimes via the supervisor schema
StartupDetails in the supervisor wire schema carries the new
arg_bindings so foreign runtimes receive the spec at task startup, with
a version migration that strips it for runtimes pinned to the previous
schema. The Go and TS SDKs regenerate against the new schema version;
the Go arg-binding runtime itself lands in a stacked follow-up PR.
* Reject upstream outputs nested inside stub literal collections
An XComArg buried in a list or dict literal fell through to the JSON
check, whose "pass it in its JSON form instead" advice is impossible to
follow for a task output. Detect nested references up front and point
the author at the working alternative: pass the upstream output as its
own argument.
* Materialize stub TaskFlow arg bindings in core Dag serialization
Review feedback flagged two issues with keeping this in the provider:
the execution API gated on @task.stub's operator class name, which any
future lang-SDK operator would have had to duplicate, and generating
JSON-schema fragments from Python type hints is tightly coupled to the
execution API and didn't belong in the standard provider's decorator.
A generic inherits_from_stub_operator flag (mirroring EmptyOperator's
inherits_from_empty_operator) now drives arg-binding materialization
from OperatorSerialization._serialize_node, so the provider's
_StubOperator shrinks to just its own structural checks and any future
foreign-runtime operator gets the same treatment for free.
* Simplify the stub-task marker to a plain is_stub attribute
inherits_from_stub_operator mirrored EmptyOperator's private-field-plus-
property pattern, but that indirection exists there to let a live
EmptyOperator instance override an inherited property; @task.stub has no
such need for a settable default. A plain public is_stub attribute is
simpler and matches the sibling lang-SDK marker landing in parallel.
Propagating it onto MappedOperator too (not just the non-mapped stub)
keeps the flag meaningful for any future consumer that needs to know a
task is stub-backed regardless of whether it's mapped; arg-binding
materialization itself still skips mapped operators, which have no
op_args/op_kwargs to bind against.
* Extract stub arg-binding materialization into its own module
serialized_objects.py was carrying the entire TaskFlow arg-binding builder
inline, pulling in pydantic schema generation and several SDK-only imports
for logic that only fires for @task.stub tasks. Splitting it into
stub_arg_bindings.py keeps the core serialization module's import surface
focused, and a dedicated arg_bindings property replaces direct access to
the private _arg_bindings attribute from outside SerializedBaseOperator.
The builder is imported lazily too, so Python-only deployments never pay for
pydantic's JSON-schema machinery just to serialize a Dag. Review also caught
that schema.json described the binding in the {__type, __var} encoded form
the
serializer never emits, and that the test asserting it validated nothing --
schema.json's tasks array never reaches the operator subschema, so the
definition is now asserted directly instead.
---
.../execution_api/datamodels/task_arg_binding.py | 93 +++++++
.../execution_api/datamodels/taskinstance.py | 8 +
.../execution_api/routes/task_instances.py | 30 ++-
.../api_fastapi/execution_api/services/__init__.py | 16 ++
.../execution_api/services/task_instances.py | 61 +++++
.../api_fastapi/execution_api/versions/__init__.py | 2 +
.../execution_api/versions/v2026_10_30.py | 42 +++
.../serialization/definitions/baseoperator.py | 7 +
.../serialization/definitions/mappedoperator.py | 7 +
airflow-core/src/airflow/serialization/schema.json | 20 +-
.../airflow/serialization/serialized_objects.py | 27 +-
.../src/airflow/serialization/stub_arg_bindings.py | 286 +++++++++++++++++++++
.../versions/head/test_task_instances.py | 113 ++++++++
.../execution_api/versions/v2026_10_30/__init__.py | 16 ++
.../versions/v2026_10_30/test_task_instances.py | 100 +++++++
.../unit/serialization/test_dag_serialization.py | 117 +++++++++
.../unit/serialization/test_stub_arg_bindings.py | 173 +++++++++++++
.../src/tests_common/test_utils/version_compat.py | 2 +
generated/known_sdk_imports_in_core.txt | 3 +-
.../cmd/airflow-go-pack/pack_integration_test.go | 3 +-
go-sdk/pkg/execution/messages.go | 2 +-
.../airflow/providers/standard/decorators/stub.py | 3 +
.../tests/unit/standard/decorators/test_stub.py | 265 ++++++++++++++++++-
.../src/airflow/sdk/api/datamodels/_generated.py | 34 ++-
task-sdk/src/airflow/sdk/bases/decorator.py | 1 +
.../sdk/definitions/_internal/abstractoperator.py | 2 +
.../src/airflow/sdk/definitions/mappedoperator.py | 3 +
.../airflow/sdk/execution_time/schema/schema.json | 125 ++++++++-
.../sdk/execution_time/schema/versions/__init__.py | 5 +
.../execution_time/schema/versions/v2026_10_30.py | 36 +++
.../execution_time/schema/test_migrator.py | 113 +++++++-
ts-sdk/src/generated/supervisor.ts | 103 +++++---
32 files changed, 1772 insertions(+), 46 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/task_arg_binding.py
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/task_arg_binding.py
new file mode 100644
index 00000000000..94c653d577c
--- /dev/null
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/task_arg_binding.py
@@ -0,0 +1,93 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Positional-argument binding spec for stub (foreign-runtime) tasks.
+
+Captured at parse time from the ``@task.stub`` TaskFlow call, stored in the
serialized
+Dag, and delivered to the lang-SDK runtime via ``TIRunContext.arg_bindings``.
+"""
+
+from __future__ import annotations
+
+from functools import cache
+from typing import Annotated, Literal
+
+from pydantic import Field, JsonValue, TypeAdapter
+from typing_extensions import TypeAliasType
+
+from airflow.api_fastapi.core_api.base import BaseModel
+
+# A named, titled alias (like TaskArgBinding below) kept as free-form JSON
rather than a
+# typed model, so unknown JSON-schema keywords survive re-serialization along
the way.
+ArgValueSchema = TypeAliasType(
+ "ArgValueSchema", Annotated[dict[str, JsonValue],
Field(title="ArgValueSchema")]
+)
+"""JSON-schema fragment constraining the value a stub-task argument binds to;
generated
+by pydantic from the stub annotation, carried verbatim, unknown keywords
ignored."""
+
+
+class _ArgBindingBase(BaseModel):
+ """Fields every :class:`TaskArgBinding` variant carries, regardless of
``kind``."""
+
+ name: str
+ """The stub function's parameter name this binding fills, in declaration
order."""
+
+ value_schema: ArgValueSchema | None = None
+ """Schema fragment from the stub function's annotation; omitted when
unconstrained."""
+
+
+class XComArgBinding(_ArgBindingBase):
+ """One positional stub-task argument pulled from an upstream task's
XCom."""
+
+ # No default: it would drop ``kind`` from ``required``, and the generated
task-sdk
+ # client then types it ``Literal | None``, invalid as a tagged-union
discriminator.
+ kind: Literal["xcom"]
+
+ task_id: str
+ """Upstream task id whose ``return_value`` XCom is pulled."""
+
+
+class LiteralArgBinding(_ArgBindingBase):
+ """One positional stub-task argument carrying an inline literal from the
Dag file."""
+
+ kind: Literal["literal"]
+ """No default, for the same generated-client reason as
``XComArgBinding.kind``."""
+
+ value: JsonValue | None = None
+ """The literal value from the Dag file."""
+
+ from_default: bool = False
+ """True when the value was filled from the stub signature's default rather
than passed in the call."""
+
+
+# A named alias with an explicit title so the union lands in every schema as
its own
+# named definition, which the supervisor-schema dump dedups with its task-sdk
twin by title.
+TaskArgBinding = TypeAliasType(
+ "TaskArgBinding",
+ Annotated[XComArgBinding | LiteralArgBinding, Field(discriminator="kind",
title="TaskArgBinding")],
+)
+"""One positional argument of a stub (foreign-runtime) task, in declaration
order."""
+
+
+@cache
+def get_arg_bindings_adapter() -> TypeAdapter[list[TaskArgBinding]]:
+ """
+ Build (lazily, then cache) the adapter validating serialized dicts into
``TaskArgBinding``.
+
+ Only the stub-task path in the execution API needs it, so regular runs
never pay for it.
+ """
+ return TypeAdapter(list[TaskArgBinding])
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
index 0064bdf1e73..ddf31db9718 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
@@ -36,6 +36,7 @@ from airflow.api_fastapi.common.types import UtcDateTime
from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
from airflow.api_fastapi.execution_api.datamodels.asset import AssetProfile
from airflow.api_fastapi.execution_api.datamodels.connection import
ConnectionResponse
+from airflow.api_fastapi.execution_api.datamodels.task_arg_binding import
TaskArgBinding
from airflow.api_fastapi.execution_api.datamodels.variable import
VariableResponse
from airflow.utils.state import (
DagRunState,
@@ -439,6 +440,13 @@ class TIRunContext(BaseModel):
always reflects when the task *first* started, not when it was
rescheduled/resumed.
"""
+ arg_bindings: list[TaskArgBinding] | None = None
+ """
+ Ordered positional-argument binding spec for stub (foreign-runtime) tasks.
+
+ ``None`` for regular tasks and for stub tasks that declare no parameters.
+ """
+
class PrevSuccessfulDagRunResponse(BaseModel):
"""Schema for response with previous successful DagRun information for
Task Template Context."""
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 8f79c89808b..02d723a8218 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -32,7 +32,7 @@ from fastapi import Body, HTTPException, Query, Response,
Security, status
from opentelemetry import trace
from opentelemetry.trace import StatusCode
from opentelemetry.trace.propagation.tracecontext import
TraceContextTextMapPropagator
-from pydantic import JsonValue
+from pydantic import JsonValue, ValidationError
from sqlalchemy import and_, func, or_, tuple_, update
from sqlalchemy.engine import CursorResult
from sqlalchemy.exc import DataError, NoResultFound, SQLAlchemyError
@@ -50,6 +50,7 @@ from airflow.api_fastapi.common.db.dags import
eager_load_teams
from airflow.api_fastapi.common.types import UtcDateTime
from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
+from airflow.api_fastapi.execution_api.datamodels.task_arg_binding import
get_arg_bindings_adapter
from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
InactiveAssetsResponse,
PreviousTIResponse,
@@ -76,6 +77,10 @@ from airflow.api_fastapi.execution_api.security import (
get_team_name_for_ti,
require_auth,
)
+from airflow.api_fastapi.execution_api.services.task_instances import (
+ client_supports_arg_bindings,
+ get_arg_bindings,
+)
from airflow.configuration import conf
from airflow.exceptions import InvalidPartitionKeyError, TaskNotFound
from airflow.models.asset import AssetActive
@@ -164,6 +169,7 @@ def ti_run(
TI.hostname,
TI.unixname,
TI.pid,
+ TI.dag_version_id,
# This selects the raw JSON value, bypassing the deserialization
-- we want that to happen on the
# client
column("next_kwargs", JSON),
@@ -309,6 +315,28 @@ def ti_run(
should_retry=_is_eligible_to_retry(previous_state, ti.try_number,
ti.max_tries),
)
+ # Only set for lang-SDK (foreign-runtime) tasks with a captured
TaskFlow arg
+ # spec; the route excludes unset fields, keeping regular responses
lean.
+ if client_supports_arg_bindings() and (
+ arg_bindings := get_arg_bindings(dag_bag, ti, session=session)
+ ):
+ try:
+ context.arg_bindings =
get_arg_bindings_adapter().validate_python(arg_bindings)
+ except ValidationError:
+ log.exception(
+ "Serialized arg_bindings spec failed validation",
+ dag_id=ti.dag_id,
+ task_id=ti.task_id,
+ dag_version_id=ti.dag_version_id,
+ )
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ detail={
+ "reason": "invalid_arg_bindings",
+ "message": "The serialized TaskFlow arg spec for this
stub task is not valid.",
+ },
+ )
+
# Only set if they are non-null
if ti.next_method:
context.next_method = ti.next_method
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/services/__init__.py
b/airflow-core/src/airflow/api_fastapi/execution_api/services/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/services/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
new file mode 100644
index 00000000000..ab67aaa9814
--- /dev/null
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py
@@ -0,0 +1,61 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Business logic backing the task-instance execution routes."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
+ from airflow.models.dagbag import DBDagBag
+
+
+def client_supports_arg_bindings() -> bool:
+ """
+ Whether the request's negotiated API version can receive ``arg_bindings``.
+
+ Clients on older versions never see the field (the version migration
strips it from
+ the response), so the derivation must not run for them.
+
+ Rather than comparing the negotiated version by date, we check the
+ ``VersionChangeWithSideEffects`` subclass's ``is_applied`` flag; see
+
https://docs.cadwyn.dev/concepts/version_changes/#version-changes-with-side-effects
+ """
+ # Imported locally: the versions package transitively imports the routes,
which import
+ # this module, so a top-level import here would be circular.
+ from airflow.api_fastapi.execution_api.versions.v2026_10_30 import
AddArgBindingsToTIRunContext
+
+ return AddArgBindingsToTIRunContext.is_applied
+
+
+def get_arg_bindings(dag_bag: DBDagBag, ti: Any, *, session: Session) -> list
| None:
+ """
+ Extract the stub task's TaskFlow arg spec from its Dag version.
+
+ Mapped (``.expand()``) stubs never capture a parse-time spec, so they
resolve to
+ ``None`` here and keep the legacy ignored-args behavior; per-map-index
delivery
+ lands in a follow-up.
+ """
+ if ti.dag_version_id is None:
+ return None
+ if (dag := dag_bag.get_dag(ti.dag_version_id, session=session)) is None:
+ return None
+ if (task := dag.task_dict.get(ti.task_id)) is None or not task.is_stub:
+ return None
+ return task.arg_bindings
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
index dc7035d31e3..d56ec735c8f 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
@@ -51,9 +51,11 @@ from airflow.api_fastapi.execution_api.versions.v2026_06_30
import (
AddTeamNameField,
AddVariableKeysEndpoint,
)
+from airflow.api_fastapi.execution_api.versions.v2026_10_30 import
AddArgBindingsToTIRunContext
bundle = VersionBundle(
HeadVersion(),
+ Version("2026-10-30", AddArgBindingsToTIRunContext),
Version(
"2026-06-30",
AddVariableKeysEndpoint,
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
new file mode 100644
index 00000000000..1c85aed252c
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py
@@ -0,0 +1,42 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from cadwyn import (
+ ResponseInfo,
+ VersionChangeWithSideEffects,
+ convert_response_to_previous_version_for,
+ schema,
+)
+
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import
TIRunContext
+
+
+class AddArgBindingsToTIRunContext(VersionChangeWithSideEffects):
+ """Add the ``arg_bindings`` argument-binding spec for stub
(foreign-runtime) tasks."""
+
+ description = __doc__
+
+ # A side-effect change, not just a schema one, so ti_run can gate the
server-side spec
+ # derivation on ``is_applied``: clients older than this version never
receive the field.
+ instructions_to_migrate_to_previous_version =
(schema(TIRunContext).field("arg_bindings").didnt_exist,)
+
+ @convert_response_to_previous_version_for(TIRunContext) # type:
ignore[arg-type]
+ def remove_arg_bindings_field(response: ResponseInfo) -> None: # type:
ignore[misc]
+ """Strip ``arg_bindings`` from the run context for older clients."""
+ response.body.pop("arg_bindings", None)
diff --git a/airflow-core/src/airflow/serialization/definitions/baseoperator.py
b/airflow-core/src/airflow/serialization/definitions/baseoperator.py
index 6bafc589123..7c8915470e7 100644
--- a/airflow-core/src/airflow/serialization/definitions/baseoperator.py
+++ b/airflow-core/src/airflow/serialization/definitions/baseoperator.py
@@ -68,6 +68,7 @@ class SerializedBaseOperator(DAGNode):
_can_skip_downstream: bool
_is_empty: bool
+ _arg_bindings: list[dict[str, Any]] | None = None
_needs_expansion: bool
_task_display_name: str | None = None
_weight_rule: str | PriorityWeightStrategy = "downstream"
@@ -95,6 +96,7 @@ class SerializedBaseOperator(DAGNode):
inlets: Sequence = []
is_setup: bool = False
+ is_stub: bool = False
is_teardown: bool = False
map_index_template: str | None = None
@@ -295,6 +297,11 @@ class SerializedBaseOperator(DAGNode):
def inherits_from_skipmixin(self) -> bool:
return self._can_skip_downstream
+ @property
+ def arg_bindings(self) -> list[dict[str, Any]] | None:
+ """The stub task's materialized TaskFlow arg-binding spec, or None for
regular tasks."""
+ return self._arg_bindings
+
@property
def operator_name(self) -> str:
# Overwrites operator_name of BaseOperator to use _operator_name
instead of
diff --git
a/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
b/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
index 348e8423100..8762c6734be 100644
--- a/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
+++ b/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
@@ -105,6 +105,8 @@ class SerializedMappedOperator(DAGNode):
_is_empty: bool = attrs.field(alias="is_empty", init=False, default=False)
_can_skip_downstream: bool = attrs.field(alias="can_skip_downstream")
_is_sensor: bool = attrs.field(alias="is_sensor", default=False)
+ is_stub: bool = attrs.field(init=False, default=False)
+ _arg_bindings: list[dict[str, Any]] | None =
attrs.field(alias="arg_bindings", init=False, default=None)
_task_module: str
task_type: str
_operator_name: str
@@ -186,6 +188,11 @@ class SerializedMappedOperator(DAGNode):
def inherits_from_skipmixin(self) -> bool:
return self._can_skip_downstream
+ @property
+ def arg_bindings(self) -> list[dict[str, Any]] | None:
+ """The stub task's materialized TaskFlow arg-binding spec, or None for
regular tasks."""
+ return self._arg_bindings
+
@property
def owner(self) -> str:
return self._get_partial_kwargs_or_operator_default("owner")
diff --git a/airflow-core/src/airflow/serialization/schema.json
b/airflow-core/src/airflow/serialization/schema.json
index 872c3a1331e..bbb78a8e618 100644
--- a/airflow-core/src/airflow/serialization/schema.json
+++ b/airflow-core/src/airflow/serialization/schema.json
@@ -142,6 +142,19 @@
"description": "A python dictionary containing values of any type",
"type": "object"
},
+ "arg_binding": {
+ "$comment": "One captured TaskFlow call argument of a @task.stub task.
Materialized directly by _serialize_node, so it stays plain JSON with no
{__type, __var} encoding. The object stays open so future binding fields keep
validating on older cores",
+ "type": "object",
+ "properties": {
+ "name": { "type": "string" },
+ "kind": { "type": "string", "enum": [ "xcom", "literal" ] },
+ "value_schema": { "type": "object" },
+ "task_id": { "type": "string" },
+ "value": {},
+ "from_default": { "type": "boolean" }
+ },
+ "required": [ "name", "kind" ]
+ },
"color": {
"type": "string",
"pattern": "^#[a-fA-F0-9]{3,6}$"
@@ -345,7 +358,12 @@
"is_teardown": {"type": "boolean", "default": false},
"on_failure_fail_dagrun": {"type": "boolean", "default": false},
"max_active_tis_per_dag": {"type": "integer"},
- "max_active_tis_per_dagrun": {"type": "integer"}
+ "max_active_tis_per_dagrun": {"type": "integer"},
+ "_arg_bindings": {
+ "$comment": "Only present on @task.stub tasks called with TaskFlow
arguments",
+ "type": "array",
+ "items": { "$ref": "#/definitions/arg_binding" }
+ }
},
"dependencies": {
"expand_input": ["partial_kwargs", "_is_mapped"],
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py
b/airflow-core/src/airflow/serialization/serialized_objects.py
index f42e51e28f7..3a0a9278ae2 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -31,7 +31,7 @@ import sys
import weakref
from collections.abc import Collection, Iterable, Mapping
from functools import cache, cached_property, lru_cache
-from inspect import signature
+from inspect import Parameter, signature
from textwrap import dedent
from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple, TypeVar, cast,
overload
@@ -112,8 +112,6 @@ from airflow.utils.db import LazySelectSequence
from airflow.utils.sqlalchemy import deserialize_pod_dict
if TYPE_CHECKING:
- from inspect import Parameter
-
from kubernetes.client import models as k8s # noqa: TC004
from kubernetes.client.api_client import ApiClient # noqa: TC004
@@ -1063,6 +1061,20 @@ class OperatorSerialization(DAGNode, BaseSerialization):
if op.inherits_from_skipmixin:
serialize_op["_can_skip_downstream"] = True
+ if op.is_stub:
+ # Imported here, not at module scope: this pulls pydantic's
JSON-schema machinery,
+ # which only lang-SDK (non-Python) workloads ever need.
+ from airflow.sdk.bases.decorator import DecoratedOperator
+ from airflow.serialization.stub_arg_bindings import
build_arg_bindings
+
+ serialize_op["is_stub"] = True
+ if (
+ not op.is_mapped
+ and isinstance(op, DecoratedOperator)
+ and (arg_bindings := build_arg_bindings(op))
+ ):
+ serialize_op["_arg_bindings"] = arg_bindings
+
if op.start_trigger_args:
serialize_op["start_trigger_args"] =
_encode_start_trigger_args(op.start_trigger_args)
@@ -1177,6 +1189,10 @@ class OperatorSerialization(DAGNode, BaseSerialization):
raise RuntimeError("_is_sensor=False should never have
been serialized!")
object.__setattr__(op, "deps", op.deps |
{ReadyToRescheduleDep()})
continue
+ elif k in ("is_stub", "_arg_bindings"):
+ # Both are restored unconditionally below: is_stub must fail
closed rather than go
+ # through generic decoding, and _arg_bindings is plain JSON,
not {__type, __var}.
+ continue
elif (
k in cls._decorated_fields
or k not in op.get_serialized_fields()
@@ -1235,6 +1251,11 @@ class OperatorSerialization(DAGNode, BaseSerialization):
# Used to determine if an Operator is inherited from SkipMixin
setattr(op, "_can_skip_downstream",
bool(encoded_op.get("_can_skip_downstream", False)))
+ # Fails closed like the Dag-level flag: a non-Python producer's blob
is never schema-validated
+ # on this path, so anything that is not JSON ``true`` means "not a
stub".
+ setattr(op, "is_stub", encoded_op.get("is_stub") is True)
+ setattr(op, "_arg_bindings", encoded_op.get("_arg_bindings"))
+
start_trigger_args = None
encoded_start_trigger_args = encoded_op.get("start_trigger_args", None)
if encoded_start_trigger_args:
diff --git a/airflow-core/src/airflow/serialization/stub_arg_bindings.py
b/airflow-core/src/airflow/serialization/stub_arg_bindings.py
new file mode 100644
index 00000000000..864848c49ae
--- /dev/null
+++ b/airflow-core/src/airflow/serialization/stub_arg_bindings.py
@@ -0,0 +1,286 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Materialize the TaskFlow arg-binding spec for lang-SDK stub tasks, at
Dag-serialization time.
+
+Any non-mapped operator flagged ``is_stub`` (currently only ``@task.stub``'s
``_StubOperator``)
+gets its ordered positional-argument spec built here, from the live operator's
already-bound
+``python_callable``/``op_args``/``op_kwargs`` -- called from
+``OperatorSerialization._serialize_node`` the same way ``is_stub`` itself is
derived, so no
+provider needs to duplicate this against the execution API's
``TaskArgBinding`` schema.
+"""
+
+from __future__ import annotations
+
+import copy
+import datetime
+import json
+import types
+import typing
+from functools import cache
+from inspect import Parameter, Signature, signature
+from typing import TYPE_CHECKING, Any
+
+from pydantic import PydanticUserError, TypeAdapter
+from pydantic.json_schema import GenerateJsonSchema
+
+from airflow.models.xcom import XCOM_RETURN_KEY
+from airflow.sdk import XComArg
+from airflow.sdk.definitions.context import KNOWN_CONTEXT_KEYS
+from airflow.sdk.definitions.mappedoperator import MappedOperator
+from airflow.sdk.definitions.xcom_arg import PlainXComArg
+
+if TYPE_CHECKING:
+ from airflow.sdk.bases.decorator import DecoratedOperator
+
+
+class _ValueSchemaGenerator(GenerateJsonSchema):
+ """
+ Pydantic's stock JSON-schema generation plus OpenAPI's fixed-width numeric
formats.
+
+ A foreign runtime decodes numbers into machine types, which the bare
+ ``integer``/``number`` type names cannot convey; ``format`` is an
annotation per
+ JSON schema, so runtimes that don't know these names simply skip them.
+ """
+
+ def int_schema(self, schema):
+ return {**super().int_schema(schema), "format": "int64"}
+
+ def float_schema(self, schema):
+ return {**super().float_schema(schema), "format": "double"}
+
+
+# Most-derived first: datetime subclasses date, so it must be matched before
date.
+_TEMPORAL_BASES = (datetime.datetime, datetime.date, datetime.time,
datetime.timedelta)
+
+
+def _normalize_temporal_annotation(annotation: Any) -> Any:
+ """
+ Map temporal subclasses (e.g. ``pendulum.DateTime``) to their stdlib base.
+
+ Applied recursively through unions and containers, and only as a retry
when direct
+ schema generation fails, so temporal types carrying their own pydantic
schema keep it.
+ """
+ # Parametrized generics must be detected before the plain-class branch: on
Python
+ # 3.10, isinstance(list[X], type) is True and issubclass silently consults
the
+ # origin, so the class branch would return list[X] unnormalized.
+ origin = typing.get_origin(annotation)
+ args = typing.get_args(annotation)
+ if origin is not None and args:
+ normalized = tuple(_normalize_temporal_annotation(arg) for arg in args)
+ if normalized == args:
+ return annotation
+ if origin in (typing.Union, types.UnionType):
+ return typing.Union[normalized] # noqa: UP007 -- runtime
construction from a tuple
+ return origin[normalized]
+ if isinstance(annotation, type):
+ return next((base for base in _TEMPORAL_BASES if
issubclass(annotation, base)), annotation)
+ return annotation
+
+
+def _infer_value_schema(annotation: Any) -> dict[str, Any] | None:
+ """
+ Build the JSON-schema fragment for one stub parameter annotation, via
pydantic.
+
+ The pydantic-generated schema ships verbatim, so runtimes must treat it as
+ open-vocabulary JSON schema. Returns ``None`` when the annotation
constrains nothing
+ (missing, ``Any``, bare ``None``) or pydantic cannot generate a schema for
it; the
+ binding then omits ``value_schema`` and the foreign runtime falls back to a
+ decode-only check.
+ """
+ if annotation is Parameter.empty or annotation is None or annotation is
Any:
+ return None
+ if annotation is type(None):
+ # get_type_hints normalizes a bare ``None`` annotation to NoneType; a
parameter
+ # that can only ever be None constrains nothing worth shipping.
+ return None
+ try:
+ schema = _generate_value_schema(annotation)
+ except TypeError:
+ # Unhashable annotations cannot key the cache; generate directly. Any
pydantic
+ # failure inside the body degrades to None there, so this retry never
re-raises.
+ schema = _generate_value_schema.__wrapped__(annotation)
+ # Deep-copy so callers embedding the fragment never alias the cached dict.
+ return copy.deepcopy(schema) if schema else None
+
+
+@cache
+def _generate_value_schema(annotation: Any) -> dict[str, Any] | None:
+ """
+ Generate the schema for one annotation, cached for the process lifetime.
+
+ TypeAdapter construction is one of pydantic's most expensive operations and
+ annotations are static, so re-serializations of the same Dag must not
re-pay it.
+ """
+ # PydanticUserError/TypeError cover annotations pydantic can't schema;
either way,
+ # that degrades to no schema rather than failing Dag serialization.
+ try:
+ return
TypeAdapter(annotation).json_schema(schema_generator=_ValueSchemaGenerator)
+ except (PydanticUserError, TypeError):
+ normalized = _normalize_temporal_annotation(annotation)
+ if normalized is annotation:
+ return None
+ try:
+ return
TypeAdapter(normalized).json_schema(schema_generator=_ValueSchemaGenerator)
+ except (PydanticUserError, TypeError):
+ return None
+
+
+def _validate_stub_signature(sig: Signature, task_id: str) -> None:
+ for param in sig.parameters.values():
+ if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
+ raise ValueError(
+ f"@task.stub task {task_id!r} must declare a fixed number of
parameters for the "
+ f"foreign runtime to bind against; *{param.name} is not
supported"
+ )
+ if param.name in KNOWN_CONTEXT_KEYS:
+ raise ValueError(
+ f"@task.stub task {task_id!r} parameter {param.name!r} is an
Airflow context key; "
+ "stub signatures declare only data parameters -- the lang-SDK
runtime injects its "
+ "own task context natively (e.g. the Go SDK's sdk.TIRunContext
parameter)"
+ )
+
+
+def _resolve_param_annotations(python_callable: Any, sig: Signature) ->
dict[str, Any]:
+ """Map each parameter to its serialization-time-resolvable annotation
(``Parameter.empty`` when not)."""
+ try:
+ hints = typing.get_type_hints(python_callable)
+ except (NameError, TypeError):
+ # Annotations that cannot be resolved (e.g. names behind TYPE_CHECKING
with
+ # ``from __future__ import annotations``) degrade to "any".
+ hints = {}
+
+ def _resolve(name: str, param: Parameter) -> Any:
+ if name in hints:
+ return hints[name]
+ if isinstance(param.annotation, str):
+ return Parameter.empty
+ return param.annotation
+
+ return {name: _resolve(name, param) for name, param in
sig.parameters.items()}
+
+
+def _ensure_json_literal(value: Any, task_id: str, name: str) -> None:
+ if next(XComArg.iter_xcom_references(value), None) is not None:
+ raise ValueError(
+ f"@task.stub task {task_id!r} parameter {name!r} received a
collection with an "
+ "upstream task output nested inside it; only a direct XComArg
argument can cross "
+ "the language boundary -- pass the upstream output as its own
argument"
+ )
+ try:
+ json.dumps(value, allow_nan=False)
+ except (TypeError, ValueError):
+ raise ValueError(
+ f"@task.stub task {task_id!r} parameter {name!r} received a
literal of type "
+ f"{type(value).__name__} that is not JSON-serializable, so it
cannot be passed "
+ "to the foreign runtime; pass it in its JSON form instead"
+ )
+
+
+def _validate_xcom_value(value: Any, task_id: str, name: str) -> bool:
+ """Validate an XComArg argument, returning True when it is a bindable
direct upstream output."""
+ if isinstance(value, PlainXComArg):
+ if value.key != XCOM_RETURN_KEY:
+ raise ValueError(
+ f"@task.stub task {task_id!r} parameter {name!r} references
the XCom key "
+ f"{value.key!r}; only an upstream task's return value can
cross the language "
+ "boundary -- indexing an output by a custom key is not
supported"
+ )
+ # isinstance, not .is_mapped: Airflow 2.11 operators have no is_mapped
attribute.
+ if isinstance(value.operator, MappedOperator):
+ raise ValueError(
+ f"@task.stub task {task_id!r} parameter {name!r} references
the aggregated "
+ f"output of the mapped task {value.operator.task_id!r}; a
foreign runtime "
+ "pulls single XCom rows, so a mapped upstream's combined
output is not "
+ "supported"
+ )
+ return True
+ if isinstance(value, XComArg):
+ raise ValueError(
+ f"@task.stub task {task_id!r} parameter {name!r} received a "
+ f"{type(value).__name__}; only direct upstream task outputs can
cross the "
+ "language boundary -- .map()/.zip()/.concat() results are not
supported"
+ )
+ return False
+
+
+def build_arg_bindings(op: DecoratedOperator) -> list[dict[str, Any]] | None:
+ """
+ Bind the TaskFlow call arguments to the stub signature and build the
ordered arg spec.
+
+ The caller owns the precondition:
``OperatorSerialization._serialize_node`` calls this only for
+ a non-mapped ``DecoratedOperator`` flagged ``is_stub``, and nothing here
re-checks it.
+
+ Each spec entry is a plain dict matching one variant of the execution API's
+ ``TaskArgBinding`` union: an ``XComArgBinding`` (``kind="xcom"``) for
upstream TaskFlow
+ outputs, or a ``LiteralArgBinding`` (``kind="literal"``) for everything
else. ``name`` is
+ always the stub function's parameter name, so a foreign runtime can bind
by name in
+ addition to the existing positional order.
+
+ Returns ``None`` for argless calls: the binding contract (including the
signature checks
+ below) applies only once a TaskFlow call actually passes arguments, so
pre-TaskFlow stub
+ Dags whose call arguments were always ignored keep serializing.
+ """
+ python_callable = op.python_callable
+ op_args = op.op_args
+ op_kwargs = op.op_kwargs
+ task_id = op.task_id
+
+ if not op_args and not op_kwargs:
+ return None
+
+ # Direct .expand() on the stub needs no spec here (ti_run derives
per-map-index
+ # bindings from the serialized expand input), but a mapped task group
creates
+ # per-map-index instances of the tasks inside it with no expand input of
their own,
+ # so their arg values are unresolvable both here and server-side.
+ if op.get_closest_mapped_task_group() is not None:
+ raise ValueError(
+ f"@task.stub task {task_id!r} passes TaskFlow call arguments
inside a mapped "
+ "task group; the captured spec cannot carry values that resolve
per map index at "
+ "runtime, so stub tasks with arguments are not supported under a
task group's "
+ ".expand()"
+ )
+
+ op_signature = signature(python_callable)
+ _validate_stub_signature(op_signature, task_id)
+
+ bound = op_signature.bind(*op_args, **op_kwargs)
+ explicitly_bound = set(bound.arguments)
+ bound.apply_defaults()
+
+ annotations = _resolve_param_annotations(python_callable, op_signature)
+
+ spec: list[dict[str, Any]] = []
+ for name in op_signature.parameters:
+ value = bound.arguments[name]
+ value_schema = _infer_value_schema(annotations[name])
+ if _validate_xcom_value(value, task_id, name):
+ xcom_entry: dict[str, Any] = {"name": name, "kind": "xcom",
"task_id": value.operator.task_id}
+ if value_schema is not None:
+ xcom_entry["value_schema"] = value_schema
+ spec.append(xcom_entry)
+ continue
+ _ensure_json_literal(value, task_id, name)
+ entry: dict[str, Any] = {"name": name, "kind": "literal", "value":
value}
+ if value_schema is not None:
+ # Key omission (never ``None``) is the wire contract for
"unconstrained":
+ # ti_run responds with ``exclude_unset``, so an absent key stays
absent.
+ entry["value_schema"] = value_schema
+ if name not in explicitly_bound:
+ entry["from_default"] = True
+ spec.append(entry)
+ return spec
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 36ed804e647..80392fd17e2 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -32,6 +32,7 @@ from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import
InMemorySpanExporter
from opentelemetry.trace import StatusCode
from opentelemetry.trace.propagation.tracecontext import
TraceContextTextMapPropagator
+from pydantic import ValidationError
from sqlalchemy import select, update
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
@@ -161,6 +162,14 @@ def test_id_matches_sub_claim(client, session,
create_task_instance):
class TestTIRunState:
+ RUN_PAYLOAD = {
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": "2024-09-30T12:00:00Z",
+ }
+
def setup_method(self):
clear_db_logs()
clear_db_runs()
@@ -372,6 +381,110 @@ class TestTIRunState:
assert extras["scope"] == "execution"
assert extras["sub"] == str(ti.id)
+ def test_ti_run_returns_arg_bindings_for_stub_task(self, client,
dag_maker):
+ """A stub task's TaskFlow arg spec is extracted from the serialized
Dag and returned."""
+ with dag_maker("test_arg_bindings_dag", serialized=True):
+
+ @task.stub
+ def extract(): ...
+
+ @task.stub
+ def transform(country: str, extracted: dict, limit: int = 10): ...
+
+ transform("uk", extract())
+
+ dr = dag_maker.create_dagrun()
+ tis = {ti.task_id: ti for ti in dr.get_task_instances()}
+ for ti in tis.values():
+ ti.set_state(State.QUEUED)
+ dag_maker.session.flush()
+
+ response =
client.patch(f"/execution/task-instances/{tis['transform'].id}/run",
json=self.RUN_PAYLOAD)
+ assert response.status_code == 200
+ assert response.json()["arg_bindings"] == [
+ {"name": "country", "kind": "literal", "value_schema": {"type":
"string"}, "value": "uk"},
+ {
+ "name": "extracted",
+ "kind": "xcom",
+ "value_schema": {"type": "object", "additionalProperties":
True},
+ "task_id": "extract",
+ },
+ {
+ "name": "limit",
+ "kind": "literal",
+ "value_schema": {"type": "integer", "format": "int64"},
+ "value": 10,
+ "from_default": True,
+ },
+ ]
+
+ # An argless stub has no captured spec, so the field stays unset.
+ response =
client.patch(f"/execution/task-instances/{tis['extract'].id}/run",
json=self.RUN_PAYLOAD)
+ assert response.status_code == 200
+ assert "arg_bindings" not in response.json()
+
+ @mock.patch(
+
"airflow.api_fastapi.execution_api.routes.task_instances.get_arg_bindings",
+ autospec=True,
+ return_value=[{"name": "country", "kind": "hologram", "value": "uk"}],
+ )
+ def test_ti_run_reports_invalid_arg_bindings_spec(self, _, client,
dag_maker):
+ """A serialized spec this core version cannot validate fails with a
structured error, not a bare 500."""
+ with dag_maker("test_invalid_arg_bindings_dag", serialized=True):
+
+ @task.stub
+ def transform(country: str): ...
+
+ transform("uk")
+
+ dr = dag_maker.create_dagrun()
+ (ti,) = dr.get_task_instances()
+ ti.set_state(State.QUEUED)
+ dag_maker.session.flush()
+
+ response = client.patch(f"/execution/task-instances/{ti.id}/run",
json=self.RUN_PAYLOAD)
+
+ assert response.status_code == 500
+ assert response.json()["detail"]["reason"] == "invalid_arg_bindings"
+
+ def test_ti_run_returns_no_arg_bindings_for_mapped_stub(self, client,
dag_maker):
+ """Mapped stubs keep the legacy ignored-args behavior until
per-map-index delivery lands."""
+ with dag_maker("test_mapped_stub_ignored_args", serialized=True):
+
+ @task.stub
+ def transform(country: str): ...
+
+ transform.expand(country=["uk", "fr"])
+
+ dr = dag_maker.create_dagrun()
+ ti = next(t for t in dr.get_task_instances() if t.map_index == 0)
+ ti.set_state(State.QUEUED)
+ dag_maker.session.flush()
+
+ response = client.patch(f"/execution/task-instances/{ti.id}/run",
json=self.RUN_PAYLOAD)
+ assert response.status_code == 200
+ assert "arg_bindings" not in response.json()
+
+ def test_arg_bindings_adapter_rejects_unknown_kind(self):
+ """The discriminated union refuses serialized specs with an
unrecognised kind."""
+ from airflow.api_fastapi.execution_api.datamodels.task_arg_binding
import get_arg_bindings_adapter
+
+ with pytest.raises(ValidationError, match="does not match any of the
expected tags"):
+ get_arg_bindings_adapter().validate_python(
+ [{"name": "country", "kind": "template", "value": "x"}]
+ )
+
+ def
test_arg_bindings_adapter_carries_value_schema_fragments_verbatim(self):
+ """The fragment is free-form JSON schema: every keyword the provider
generated must
+ survive validation untouched -- a typed model would silently strip
what it doesn't know."""
+ from airflow.api_fastapi.execution_api.datamodels.task_arg_binding
import get_arg_bindings_adapter
+
+ fragment = {"anyOf": [{"type": "array", "items": {"type": "string"}},
{"type": "null"}]}
+ (binding,) = get_arg_bindings_adapter().validate_python(
+ [{"name": "tags", "kind": "literal", "value_schema": fragment,
"value": ["a"]}]
+ )
+ assert binding.value_schema == fragment
+
def test_dynamic_task_mapping_with_parse_time_value(self, client,
dag_maker):
"""Test that dynamic task mapping works correctly with parse-time
values."""
with dag_maker("test_dynamic_task_mapping_with_parse_time_value",
serialized=True):
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/__init__.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_task_instances.py
new file mode 100644
index 00000000000..a4b98bd1020
--- /dev/null
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_10_30/test_task_instances.py
@@ -0,0 +1,100 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow.sdk import task
+from airflow.utils.state import State
+
+from tests_common.test_utils.db import clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+TIMESTAMP_STR = "2024-09-30T12:00:00Z"
+
+RUN_PATCH_BODY = {
+ "state": "running",
+ "hostname": "h",
+ "unixname": "u",
+ "pid": 1,
+ "start_date": TIMESTAMP_STR,
+}
+
+
[email protected]
+def old_ver_client(client):
+ """Execution API version immediately before ``arg_bindings`` was added."""
+ client.headers["Airflow-API-Version"] = "2026-06-30"
+ return client
+
+
+class TestArgBindingsFieldBackwardCompat:
+ @pytest.fixture(autouse=True)
+ def _freeze_time(self, time_machine):
+ time_machine.move_to(TIMESTAMP_STR, tick=False)
+
+ def setup_method(self):
+ clear_db_runs()
+
+ def teardown_method(self):
+ clear_db_runs()
+
+ @pytest.fixture
+ def stub_ti(self, dag_maker):
+ with dag_maker("test_arg_bindings_compat_dag", serialized=True):
+
+ @task.stub
+ def extract(): ...
+
+ @task.stub
+ def transform(country: str, extracted: dict, limit: int = 10): ...
+
+ transform("uk", extract())
+
+ dr = dag_maker.create_dagrun()
+ tis = {ti.task_id: ti for ti in dr.get_task_instances()}
+ for ti in tis.values():
+ ti.set_state(State.QUEUED)
+ dag_maker.session.flush()
+ return tis["transform"]
+
+ def test_old_version_strips_arg_bindings_even_when_set(self,
old_ver_client, stub_ti):
+ response =
old_ver_client.patch(f"/execution/task-instances/{stub_ti.id}/run",
json=RUN_PATCH_BODY)
+ assert response.status_code == 200
+ assert "arg_bindings" not in response.json()
+
+ def test_head_version_includes_arg_bindings(self, client, stub_ti):
+ response = client.patch(f"/execution/task-instances/{stub_ti.id}/run",
json=RUN_PATCH_BODY)
+ assert response.status_code == 200
+ assert response.json()["arg_bindings"] == [
+ {"name": "country", "kind": "literal", "value_schema": {"type":
"string"}, "value": "uk"},
+ {
+ "name": "extracted",
+ "kind": "xcom",
+ "value_schema": {"type": "object", "additionalProperties":
True},
+ "task_id": "extract",
+ },
+ {
+ "name": "limit",
+ "kind": "literal",
+ "value_schema": {"type": "integer", "format": "int64"},
+ "value": 10,
+ "from_default": True,
+ },
+ ]
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py
b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 7852c25dc5e..cf9c5c34c17 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -41,6 +41,7 @@ from typing import TYPE_CHECKING
from unittest import mock
import attrs
+import jsonschema
import pendulum
import pytest
from dateutil.relativedelta import FR, relativedelta
@@ -3524,6 +3525,122 @@ def
test_python_callable_name_uses_qualname_exclude_module():
assert serialized3["python_callable_name"] == "empty_function"
+def test_stub_task_args_round_trip():
+ """The stub task's TaskFlow arg spec (``_arg_bindings``) is materialized
by Dag serialization
+ and survives the round trip."""
+ from airflow.sdk import task
+
+ with DAG(dag_id="arg_bindings_dag", schedule=None) as dag:
+
+ @task.stub
+ def extract(): ...
+
+ @task.stub
+ def transform(country: str, extracted: dict): ...
+
+ # Nested value_schema (dict[str, int] re-encodes its
additionalProperties) plus
+ # dict/list literal values, whose contents must not collide with the
{__type,__var}
+ # encoding during round-trip.
+ @task.stub
+ def aggregate(counts: dict[str, int], tags: list, config: dict): ...
+
+ data = extract()
+ transform("uk", data)
+ aggregate(data, ["metrics", "hourly"], {"threshold": {"warn": 1}})
+
+ ser_dag = DagSerialization.to_dict(dag)
+ DagSerialization.validate_schema(ser_dag)
+
+ encoded_tasks = {t[Encoding.VAR]["task_id"]: t[Encoding.VAR] for t in
ser_dag["dag"]["tasks"]}
+ assert "_arg_bindings" not in encoded_tasks["extract"], "argless stubs
must not serialize a spec"
+ # validate_schema above never reaches task objects: schema.json's `tasks`
array hangs the
+ # operator sub-schema off `additionalProperties`, which JSON Schema
ignores for arrays. Assert
+ # the `arg_binding` definition directly so it stays honest about the wire
form.
+ jsonschema.validate(
+ encoded_tasks["transform"]["_arg_bindings"],
+ {
+ "definitions": load_dag_schema_dict()["definitions"],
+ "type": "array",
+ "items": {"$ref": "#/definitions/arg_binding"},
+ },
+ )
+ # Materialized directly by _serialize_node (like _is_empty), not routed
through the
+ # generic per-field encoder, so the wire form stays plain JSON with no
{__type, __var}
+ # wrapping -- the execution API validates it straight off the serialized
Dag.
+ assert encoded_tasks["transform"]["_arg_bindings"] == [
+ {
+ "name": "country",
+ "kind": "literal",
+ "value_schema": {"type": "string"},
+ "value": "uk",
+ },
+ {
+ "name": "extracted",
+ "kind": "xcom",
+ "value_schema": {"type": "object", "additionalProperties": True},
+ "task_id": "extract",
+ },
+ ]
+
+ round_tripped = DagSerialization.from_dict(ser_dag)
+ assert round_tripped.task_dict["transform"].is_stub is True
+ assert round_tripped.task_dict["transform"].arg_bindings == [
+ {"name": "country", "kind": "literal", "value_schema": {"type":
"string"}, "value": "uk"},
+ {
+ "name": "extracted",
+ "kind": "xcom",
+ "value_schema": {"type": "object", "additionalProperties": True},
+ "task_id": "extract",
+ },
+ ]
+ # The nested value_schema and dict/list literal values survive the
round-trip intact.
+ assert round_tripped.task_dict["aggregate"].arg_bindings == [
+ {
+ "name": "counts",
+ "kind": "xcom",
+ "value_schema": {
+ "type": "object",
+ "additionalProperties": {"type": "integer", "format": "int64"},
+ },
+ "task_id": "extract",
+ },
+ {
+ "name": "tags",
+ "kind": "literal",
+ "value_schema": {"type": "array", "items": {}},
+ "value": ["metrics", "hourly"],
+ },
+ {
+ "name": "config",
+ "kind": "literal",
+ "value_schema": {"type": "object", "additionalProperties": True},
+ "value": {"threshold": {"warn": 1}},
+ },
+ ]
+ assert round_tripped.task_dict["extract"].is_stub is True
+ assert round_tripped.task_dict["extract"].arg_bindings is None
+
+ # The deserialized spec must be plain JSON (no {__type, __var} encoding
sentinels) so the
+ # execution API can validate it straight off the serialized Dag -- this is
the contract
+ # ti_run relies on when it feeds get_arg_bindings() into the
TaskArgBinding adapter.
+ from airflow.api_fastapi.execution_api.datamodels.task_arg_binding import
get_arg_bindings_adapter
+
+ for task_id in ("transform", "aggregate"):
+
get_arg_bindings_adapter().validate_python(round_tripped.task_dict[task_id].arg_bindings)
+
+
[email protected]("raw", ["false", "true", 0, 1, None, [], {"a": 1}])
+def test_task_is_stub_fails_closed_on_non_boolean(raw):
+ """A task flag from a non-Python producer is never schema-validated, so it
must fail closed."""
+ with DAG(dag_id="test_task_is_stub_non_boolean", schedule=None) as dag:
+ BaseOperator(task_id="simple_task", start_date=datetime(2019, 8, 1))
+
+ ser_dag = DagSerialization.to_dict(dag)
+ ser_dag["dag"]["tasks"][0][Encoding.VAR]["is_stub"] = raw
+
+ assert
DagSerialization.from_dict(ser_dag).task_dict["simple_task"].is_stub is False
+
+
def test_handle_v1_serdag():
v1 = {
"__version": 1,
diff --git a/airflow-core/tests/unit/serialization/test_stub_arg_bindings.py
b/airflow-core/tests/unit/serialization/test_stub_arg_bindings.py
new file mode 100644
index 00000000000..f6c8b73a123
--- /dev/null
+++ b/airflow-core/tests/unit/serialization/test_stub_arg_bindings.py
@@ -0,0 +1,173 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import contextlib
+import datetime
+import typing
+from typing import Any
+
+import pendulum
+import pytest
+
+from airflow.serialization.stub_arg_bindings import _infer_value_schema
+
+
[email protected](
+ ("annotation", "expected"),
+ [
+ pytest.param(str, {"type": "string"}, id="str"),
+ pytest.param(bool, {"type": "boolean"}, id="bool"),
+ pytest.param(int, {"type": "integer", "format": "int64"}, id="int"),
+ pytest.param(float, {"type": "number", "format": "double"},
id="float"),
+ pytest.param(dict, {"type": "object", "additionalProperties": True},
id="dict"),
+ pytest.param(
+ dict[str, int],
+ {"type": "object", "additionalProperties": {"type": "integer",
"format": "int64"}},
+ id="dict-parameterized",
+ ),
+ pytest.param(
+ typing.Mapping[str, int],
+ {"type": "object", "additionalProperties": {"type": "integer",
"format": "int64"}},
+ id="mapping",
+ ),
+ pytest.param(list, {"type": "array", "items": {}}, id="list"),
+ pytest.param(
+ list[int],
+ {"type": "array", "items": {"type": "integer", "format": "int64"}},
+ id="list-parameterized",
+ ),
+ pytest.param(tuple, {"type": "array", "items": {}}, id="tuple"),
+ pytest.param(set, {"type": "array", "items": {}, "uniqueItems": True},
id="set"),
+ pytest.param(
+ typing.Sequence[int],
+ {"type": "array", "items": {"type": "integer", "format": "int64"}},
+ id="sequence",
+ ),
+ pytest.param(datetime.datetime, {"type": "string", "format":
"date-time"}, id="datetime"),
+ pytest.param(datetime.date, {"type": "string", "format": "date"},
id="date"),
+ pytest.param(datetime.time, {"type": "string", "format": "time"},
id="time"),
+ pytest.param(datetime.timedelta, {"type": "string", "format":
"duration"}, id="timedelta"),
+ pytest.param(bytes, {"type": "string", "format": "binary"},
id="bytes"),
+ pytest.param(
+ typing.Literal["a", "b"],
+ {"type": "string", "enum": ["a", "b"]},
+ id="literal",
+ ),
+ pytest.param(Any, None, id="any"),
+ pytest.param(None, None, id="none"),
+ pytest.param(type(None), None, id="nonetype"),
+ pytest.param(
+ pendulum.DateTime,
+ {"type": "string", "format": "date-time"},
+ id="pendulum-datetime",
+ ),
+ pytest.param(
+ pendulum.DateTime | None,
+ {"anyOf": [{"type": "string", "format": "date-time"}, {"type":
"null"}]},
+ id="optional-pendulum-datetime",
+ ),
+ pytest.param(
+ list[pendulum.DateTime],
+ {"type": "array", "items": {"type": "string", "format":
"date-time"}},
+ id="list-pendulum-datetime",
+ ),
+ pytest.param(pendulum.Duration, {"type": "string", "format":
"duration"}, id="pendulum-duration"),
+ pytest.param(
+ typing.Optional[str], # noqa: UP045 -- legacy form on purpose
+ {"anyOf": [{"type": "string"}, {"type": "null"}]},
+ id="optional-str",
+ ),
+ pytest.param(
+ typing.Union[int, str], # noqa: UP007 -- legacy form on purpose
+ {"anyOf": [{"type": "integer", "format": "int64"}, {"type":
"string"}]},
+ id="union",
+ ),
+ pytest.param(str | None, {"anyOf": [{"type": "string"}, {"type":
"null"}]}, id="pep604-optional"),
+ pytest.param(
+ int | None,
+ {"anyOf": [{"type": "integer", "format": "int64"}, {"type":
"null"}]},
+ id="optional-int",
+ ),
+ pytest.param(
+ datetime.datetime | None,
+ {"anyOf": [{"type": "string", "format": "date-time"}, {"type":
"null"}]},
+ id="optional-datetime",
+ ),
+ pytest.param(
+ dict | bool,
+ {"anyOf": [{"type": "object", "additionalProperties": True},
{"type": "boolean"}]},
+ id="union-dict-bool",
+ ),
+ pytest.param(
+ str | int | None,
+ {"anyOf": [{"type": "string"}, {"type": "integer", "format":
"int64"}, {"type": "null"}]},
+ id="union-with-null",
+ ),
+ pytest.param(list | tuple, {"type": "array", "items": {}},
id="union-dedupes-equal-members"),
+ pytest.param(
+ datetime.datetime | str,
+ {"anyOf": [{"type": "string", "format": "date-time"}, {"type":
"string"}]},
+ id="mixed-format-union-keeps-both",
+ ),
+ pytest.param(
+ str | contextlib.AbstractContextManager,
+ None,
+ id="union-unclassifiable-member",
+ ),
+ pytest.param(contextlib.AbstractContextManager, None,
id="custom-class"),
+ # pydantic raises PydanticUserError (not the JSON-schema subclasses)
for these; they
+ # must still degrade to no schema rather than crash Dag serialization.
+ pytest.param(typing.ClassVar, None, id="pydantic-user-error"),
+ pytest.param(typing.Callable[[int], str], None,
id="callable-invalid-for-json-schema"),
+ pytest.param(
+ pendulum.DateTime | contextlib.AbstractContextManager,
+ None,
+ id="union-temporal-and-unclassifiable",
+ ),
+ ],
+)
+def test_infer_value_schema(annotation, expected):
+ assert _infer_value_schema(annotation) == expected
+
+
+def test_infer_value_schema_cache_returns_isolated_copies():
+ first = _infer_value_schema(dict)
+ second = _infer_value_schema(dict)
+ assert first == second
+ assert first is not second, "callers embed and serialize the fragment, so
it must not alias the cache"
+
+
+def test_infer_value_schema_unhashable_annotation_generates_uncached():
+ annotation = typing.Annotated[int, {"unhashable": True}]
+ assert _infer_value_schema(annotation) == {"type": "integer", "format":
"int64"}
+
+
+def test_infer_value_schema_degrades_on_pydantic_typeerror(monkeypatch):
+ """A bare TypeError from pydantic degrades to no schema rather than
crashing Dag serialization."""
+ from airflow.serialization import stub_arg_bindings
+
+ def _raise_type_error(_annotation):
+ raise TypeError("pydantic cannot build a schema for this")
+
+ monkeypatch.setattr(stub_arg_bindings, "TypeAdapter", _raise_type_error)
+
+ # A fresh class dodges the process-lifetime schema cache and exercises the
hashable-but-
+ # unschemable path, where a naive ``except TypeError`` retry would
re-raise and crash.
+ class _Unschemable: ...
+
+ assert _infer_value_schema(_Unschemable) is None
diff --git a/devel-common/src/tests_common/test_utils/version_compat.py
b/devel-common/src/tests_common/test_utils/version_compat.py
index 7eb25dec2b3..e1b51ebe034 100644
--- a/devel-common/src/tests_common/test_utils/version_compat.py
+++ b/devel-common/src/tests_common/test_utils/version_compat.py
@@ -42,6 +42,7 @@ AIRFLOW_V_3_1_9_PLUS = get_base_airflow_version_tuple() >=
(3, 1, 9)
AIRFLOW_V_3_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 0)
AIRFLOW_V_3_2_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 2)
AIRFLOW_V_3_3_PLUS = get_base_airflow_version_tuple() >= (3, 3, 0)
+AIRFLOW_V_3_4_PLUS = get_base_airflow_version_tuple() >= (3, 4, 0)
if AIRFLOW_V_3_1_PLUS:
from airflow.sdk import PokeReturnValue, timezone
@@ -64,6 +65,7 @@ __all__ = [
"AIRFLOW_V_3_1_PLUS",
"AIRFLOW_V_3_2_PLUS",
"AIRFLOW_V_3_3_PLUS",
+ "AIRFLOW_V_3_4_PLUS",
"NOTSET",
"XCOM_RETURN_KEY",
"ArgNotSet",
diff --git a/generated/known_sdk_imports_in_core.txt
b/generated/known_sdk_imports_in_core.txt
index b93815ae6d9..f0aaae24ac2 100644
--- a/generated/known_sdk_imports_in_core.txt
+++ b/generated/known_sdk_imports_in_core.txt
@@ -29,7 +29,8 @@ airflow-core/src/airflow/serialization/definitions/dag.py::2
airflow-core/src/airflow/serialization/definitions/deadline.py::1
airflow-core/src/airflow/serialization/definitions/mappedoperator.py::5
airflow-core/src/airflow/serialization/encoders.py::11
-airflow-core/src/airflow/serialization/serialized_objects.py::16
+airflow-core/src/airflow/serialization/serialized_objects.py::17
+airflow-core/src/airflow/serialization/stub_arg_bindings.py::5
airflow-core/src/airflow/settings.py::1
airflow-core/src/airflow/stats.py::1
airflow-core/src/airflow/timetables/simple.py::1
diff --git a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
index 77725b0ac4e..84e9a1045f4 100644
--- a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
+++ b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
@@ -34,6 +34,7 @@ import (
"github.com/apache/airflow/go-sdk/internal/airflowmetadata"
"github.com/apache/airflow/go-sdk/internal/bundlefooter"
+ "github.com/apache/airflow/go-sdk/pkg/execution"
)
// crossArchFor returns an architecture different from the host that the Go
@@ -142,7 +143,7 @@ func TestPack_CrossArchExecutableWithMetadataFile(t
*testing.T) {
sdk:
language: "go"
version: "` + sdkVersion + `"
- supervisor_schema_version: "2026-06-16"
+ supervisor_schema_version: "` + execution.SupervisorSchemaVersion + `"
source: "main.go"
dags:
concurrent_xcom_dag:
diff --git a/go-sdk/pkg/execution/messages.go b/go-sdk/pkg/execution/messages.go
index 72d451a8666..bb81d60c0a4 100644
--- a/go-sdk/pkg/execution/messages.go
+++ b/go-sdk/pkg/execution/messages.go
@@ -32,7 +32,7 @@ import (
// reported in a bundle's airflow-metadata manifest as
// sdk.supervisor_schema_version so the supervisor can down/upgrade messages to
// a shape the bundle understands.
-const SupervisorSchemaVersion = "2026-06-16"
+const SupervisorSchemaVersion = "2026-10-30"
// The message-type discriminator strings (genmodels.Type*) are generated from
the
// schema's "type" consts in discriminators.gen.go; outbound messages stamp the
diff --git
a/providers/standard/src/airflow/providers/standard/decorators/stub.py
b/providers/standard/src/airflow/providers/standard/decorators/stub.py
index 08bcf163a56..b2ccb8c249a 100644
--- a/providers/standard/src/airflow/providers/standard/decorators/stub.py
+++ b/providers/standard/src/airflow/providers/standard/decorators/stub.py
@@ -34,6 +34,9 @@ if TYPE_CHECKING:
class _StubOperator(DecoratedOperator):
custom_operator_name: str = "@task.stub"
+ # Read by core Dag serialization to materialize the TaskFlow arg-binding
spec.
+ is_stub: bool = True
+
def __init__(
self,
*,
diff --git a/providers/standard/tests/unit/standard/decorators/test_stub.py
b/providers/standard/tests/unit/standard/decorators/test_stub.py
index 2a17c3fdd82..b4b55994381 100644
--- a/providers/standard/tests/unit/standard/decorators/test_stub.py
+++ b/providers/standard/tests/unit/standard/decorators/test_stub.py
@@ -17,12 +17,29 @@
from __future__ import annotations
import contextlib
+import datetime
import pytest
+from airflow.exceptions import SerializationError
+from airflow.providers.common.compat.sdk import DAG, task_group
from airflow.providers.standard.decorators.stub import stub
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS,
AIRFLOW_V_3_4_PLUS
+
+
+def _to_dict(dag):
+ """Serialize a Dag through core Dag serialization -- arg_bindings
materializes here."""
+ from airflow.serialization.serialized_objects import DagSerialization
+
+ return DagSerialization.to_dict(dag)
+
+
+def _round_trip(dag):
+ """Round-trip a Dag through core Dag serialization -- arg_bindings
materializes here."""
+ from airflow.serialization.serialized_objects import DagSerialization
+
+ return DagSerialization.from_dict(DagSerialization.to_dict(dag))
def fn_ellipsis(): ...
@@ -69,3 +86,249 @@ def test_stub_rejects_retry_policy():
def test_stub_allows_retries():
stub(fn_pass, retries=5)()
+
+
+def fn_extract(): ...
+
+
+def fn_transform(country: str, extracted: dict, retries_num: int = 3): ...
+
+
+def fn_untyped(a, b): ...
+
+
+def fn_varargs(*args): ...
+
+
+def fn_kwonly_varkw(**kwargs): ...
+
+
+def fn_context_key(ti): ...
+
+
[email protected](
+ not AIRFLOW_V_3_4_PLUS,
+ reason="arg-binding materialization was added to core Dag serialization in
Airflow 3.4",
+)
+class TestStubTaskflowArgs:
+ """The TaskFlow call on a stub captures the ordered positional-arg spec
(``arg_bindings``),
+ materialized by core Dag serialization from the stub's bound TaskFlow call
args."""
+
+ def test_literal_and_xcom_spec(self):
+ with DAG(dag_id="d") as dag:
+ extracted = stub(fn_extract)()
+ result = stub(fn_transform)("uk", extracted)
+
+ assert _round_trip(dag).task_dict["fn_transform"].arg_bindings == [
+ {"name": "country", "kind": "literal", "value_schema": {"type":
"string"}, "value": "uk"},
+ {
+ "name": "extracted",
+ "kind": "xcom",
+ "value_schema": {"type": "object", "additionalProperties":
True},
+ "task_id": "fn_extract",
+ },
+ {
+ "name": "retries_num",
+ "kind": "literal",
+ "value_schema": {"type": "integer", "format": "int64"},
+ "value": 3,
+ "from_default": True,
+ },
+ ]
+ assert result.operator.upstream_task_ids == {"fn_extract"}
+
+ def test_kwargs_normalize_to_declaration_order(self):
+ with DAG(dag_id="d") as dag:
+ extracted = stub(fn_extract)()
+ stub(fn_transform)(extracted=extracted, country="fr",
retries_num=7)
+
+ assert _round_trip(dag).task_dict["fn_transform"].arg_bindings == [
+ {"name": "country", "kind": "literal", "value_schema": {"type":
"string"}, "value": "fr"},
+ {
+ "name": "extracted",
+ "kind": "xcom",
+ "value_schema": {"type": "object", "additionalProperties":
True},
+ "task_id": "fn_extract",
+ },
+ {
+ "name": "retries_num",
+ "kind": "literal",
+ "value_schema": {"type": "integer", "format": "int64"},
+ "value": 7,
+ },
+ ]
+
+ def test_explicitly_passing_the_default_value_is_not_from_default(self):
+ """The flag tracks provenance, not value equality: an author-passed
argument is explicit
+ even when it equals the signature default, so keyword-style consumers
must still claim it."""
+ with DAG(dag_id="d") as dag:
+ extracted = stub(fn_extract)()
+ stub(fn_transform)("uk", extracted, retries_num=3)
+
+ assert _round_trip(dag).task_dict["fn_transform"].arg_bindings[2] == {
+ "name": "retries_num",
+ "kind": "literal",
+ "value_schema": {"type": "integer", "format": "int64"},
+ "value": 3,
+ }
+
+ def test_custom_xcom_key_rejected(self):
+ with DAG(dag_id="d") as dag:
+ extracted = stub(fn_extract)()
+ stub(fn_transform)("uk", extracted["part"])
+
+ with pytest.raises(SerializationError, match="indexing an output by a
custom key"):
+ _to_dict(dag)
+
+ def test_zero_param_stub_has_no_spec(self):
+ with DAG(dag_id="d") as dag:
+ stub(fn_pass)()
+
+ assert _round_trip(dag).task_dict["fn_pass"].arg_bindings is None
+
+ def test_untyped_params_omit_value_schema(self):
+ """Key absence (never ``None``) is the wire contract for an
unconstrained argument."""
+ with DAG(dag_id="d") as dag:
+ stub(fn_untyped)(1, "x")
+
+ assert _round_trip(dag).task_dict["fn_untyped"].arg_bindings == [
+ {"name": "a", "kind": "literal", "value": 1},
+ {"name": "b", "kind": "literal", "value": "x"},
+ ]
+
+ def test_unresolvable_annotation_omits_value_schema(self):
+ def fn(x): ...
+
+ fn.__annotations__ = {"x": "NotARealType"}
+ with DAG(dag_id="d") as dag:
+ stub(fn)("v")
+
+ assert _round_trip(dag).task_dict["fn"].arg_bindings == [
+ {"name": "x", "kind": "literal", "value": "v"}
+ ]
+
+ def test_varargs_rejected(self):
+ with DAG(dag_id="d") as dag:
+ stub(fn_varargs)(1, 2)
+
+ with pytest.raises(SerializationError, match="fixed number of
parameters"):
+ _to_dict(dag)
+
+ def test_varkw_rejected(self):
+ with DAG(dag_id="d") as dag:
+ stub(fn_kwonly_varkw)(x=1)
+
+ with pytest.raises(SerializationError, match="fixed number of
parameters"):
+ _to_dict(dag)
+
+ def test_context_key_param_rejected(self):
+ with DAG(dag_id="d") as dag:
+ stub(fn_context_key)(1)
+
+ with pytest.raises(SerializationError, match="is an Airflow context
key"):
+ _to_dict(dag)
+
+ @pytest.mark.parametrize("fn", [fn_varargs, fn_kwonly_varkw,
fn_context_key], ids=lambda f: f.__name__)
+ def test_argless_call_skips_signature_checks(self, fn):
+ """Pre-TaskFlow stub Dags never passed arguments; their signatures
must keep serializing."""
+ with DAG(dag_id="d") as dag:
+ stub(fn)()
+
+ assert _round_trip(dag).task_dict[fn.__name__].arg_bindings is None
+
+ def test_argless_call_captures_no_spec_for_defaulted_params(self):
+ def fn(limit: int = 10): ...
+
+ with DAG(dag_id="d") as dag:
+ stub(fn)()
+
+ assert _round_trip(dag).task_dict["fn"].arg_bindings is None
+
+ def test_non_json_literal_rejected(self):
+ with DAG(dag_id="d") as dag:
+ stub(fn_transform)("uk", object())
+
+ with pytest.raises(SerializationError, match="not JSON-serializable"):
+ _to_dict(dag)
+
+ def test_nan_literal_rejected(self):
+ with DAG(dag_id="d") as dag:
+ stub(fn_transform)("uk", {"ratio": float("nan")})
+
+ with pytest.raises(SerializationError, match="not JSON-serializable"):
+ _to_dict(dag)
+
+ def test_temporal_literal_rejected(self):
+ def fn(when: datetime.datetime): ...
+
+ with DAG(dag_id="d") as dag:
+ stub(fn)(datetime.datetime(2020, 1, 1))
+
+ with pytest.raises(SerializationError, match="not JSON-serializable"):
+ _to_dict(dag)
+
+ @pytest.mark.parametrize("wrap", [lambda x: [x], lambda x: {"data": x}],
ids=["list", "dict"])
+ def test_xcom_nested_in_collection_literal_rejected(self, wrap):
+ with DAG(dag_id="d") as dag:
+ extracted = stub(fn_extract)()
+ stub(fn_transform)("uk", wrap(extracted))
+
+ with pytest.raises(SerializationError, match="nested inside"):
+ _to_dict(dag)
+
+ def test_mapped_xcom_arg_rejected(self):
+ with DAG(dag_id="d") as dag:
+ extracted = stub(fn_extract)()
+ stub(fn_transform)("uk", extracted.map(lambda v: v))
+
+ with pytest.raises(SerializationError, match="only direct upstream
task outputs"):
+ _to_dict(dag)
+
+ def test_mapped_upstream_aggregated_output_rejected(self):
+ def fn_produce(n: int): ...
+
+ with DAG(dag_id="d") as dag:
+ vals = stub(fn_produce).expand(n=[1, 2])
+ stub(fn_transform)("uk", vals)
+
+ with pytest.raises(SerializationError, match="aggregated output of the
mapped task"):
+ _to_dict(dag)
+
+ def test_expand_builds_mapped_stub_without_parse_time_bindings(self):
+ """Mapped stubs capture no spec: their call args keep the legacy
ignored behavior for now."""
+ with DAG(dag_id="d") as dag:
+ result = stub(fn_transform).expand(country=["uk", "fr"],
extracted=[{}, {}])
+ # op_kwargs_expand_input/partial_kwargs (not is_mapped) so the
assertions also
+ # hold on the Airflow 2.x MappedOperator, which the provider still
supports.
+ assert result.operator.op_kwargs_expand_input.value == {
+ "country": ["uk", "fr"],
+ "extracted": [{}, {}],
+ }
+ assert "_arg_bindings" not in result.operator.partial_kwargs
+
+ # The wrapping MappedOperator carries the is_stub marker too, but has
no op_args/op_kwargs
+ # to bind against, so it serializes cleanly with no materialized spec
of its own.
+ round_tripped = _round_trip(dag).task_dict["fn_transform"]
+ assert round_tripped.is_stub is True
+ assert round_tripped.arg_bindings is None
+
+ def test_stub_with_args_inside_mapped_task_group_rejected(self):
+ @task_group
+ def group(n):
+ stub(fn_transform)("uk", {})
+
+ with DAG(dag_id="d") as dag:
+ group.expand(n=[1, 2])
+
+ with pytest.raises(SerializationError, match="mapped task group"):
+ _to_dict(dag)
+
+ def test_argless_stub_inside_mapped_task_group_allowed(self):
+ @task_group
+ def group(n):
+ stub(fn_extract)()
+
+ with DAG(dag_id="d") as dag:
+ group.expand(n=[1, 2])
+
+ _to_dict(dag) # must not raise
diff --git a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
index cc3c7eb0a8f..201f218c3c9 100644
--- a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
+++ b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
@@ -27,7 +27,7 @@ from uuid import UUID
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, JsonValue,
RootModel
-API_VERSION: Final[str] = "2026-06-30"
+API_VERSION: Final[str] = "2026-10-30"
class AssetAliasReferenceAssetEventDagRun(BaseModel):
@@ -608,6 +608,10 @@ class DagAttributeTypes(str, Enum):
TASK_GROUP = "taskgroup"
+class ArgValueSchema(RootModel[dict[str, JsonValue | None]]):
+ root: dict[str, JsonValue | None]
+
+
class AssetReferenceAssetEventDagRun(BaseModel):
"""
Schema for AssetModel used in AssetEventDagRunReference.
@@ -697,6 +701,18 @@ class HTTPValidationError(BaseModel):
detail: Annotated[list[ValidationError] | None, Field(title="Detail")] =
None
+class LiteralArgBinding(BaseModel):
+ """
+ One positional stub-task argument carrying an inline literal from the Dag
file.
+ """
+
+ name: Annotated[str, Field(title="Name")]
+ value_schema: ArgValueSchema | None = None
+ kind: Annotated[Literal["literal"], Field(title="Kind")]
+ value: JsonValue | None = None
+ from_default: Annotated[bool | None, Field(title="From Default")] = False
+
+
class TITerminalStatePayload(BaseModel):
"""
Schema for updating TaskInstance to a terminal state except SUCCESS state.
@@ -710,6 +726,17 @@ class TITerminalStatePayload(BaseModel):
rendered_map_index: Annotated[str | None, Field(title="Rendered Map
Index")] = None
+class XComArgBinding(BaseModel):
+ """
+ One positional stub-task argument pulled from an upstream task's XCom.
+ """
+
+ name: Annotated[str, Field(title="Name")]
+ value_schema: ArgValueSchema | None = None
+ kind: Annotated[Literal["xcom"], Field(title="Kind")]
+ task_id: Annotated[str, Field(title="Task Id")]
+
+
class AssetEventDagRunReference(BaseModel):
"""
Schema for AssetEvent model used in DagRun.
@@ -782,6 +809,10 @@ class DagRun(BaseModel):
team_name: Annotated[str | None, Field(title="Team Name")] = None
+class TaskArgBinding(RootModel[XComArgBinding | LiteralArgBinding]):
+ root: Annotated[XComArgBinding | LiteralArgBinding,
Field(discriminator="kind", title="TaskArgBinding")]
+
+
class TIRunContext(BaseModel):
"""
Response schema for TaskInstance run context.
@@ -797,3 +828,4 @@ class TIRunContext(BaseModel):
xcom_keys_to_clear: Annotated[list[str] | None, Field(title="Xcom Keys To
Clear")] = None
should_retry: Annotated[bool | None, Field(title="Should Retry")] = False
start_date: Annotated[AwareDatetime | None, Field(title="Start Date")] =
None
+ arg_bindings: Annotated[list[TaskArgBinding] | None, Field(title="Arg
Bindings")] = None
diff --git a/task-sdk/src/airflow/sdk/bases/decorator.py
b/task-sdk/src/airflow/sdk/bases/decorator.py
index e45778725cc..e3c64eb0a8a 100644
--- a/task-sdk/src/airflow/sdk/bases/decorator.py
+++ b/task-sdk/src/airflow/sdk/bases/decorator.py
@@ -670,6 +670,7 @@ class _TaskDecorator(ExpandableFactory, Generic[FParams,
FReturn, OperatorSubcla
is_empty=False,
is_sensor=self.operator_class._is_sensor,
can_skip_downstream=self.operator_class._can_skip_downstream,
+ is_stub=self.operator_class.is_stub,
task_module=self.operator_class.__module__,
task_type=self.operator_class.__name__,
operator_name=operator_name,
diff --git a/task-sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py
b/task-sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py
index 2d4ade82127..93d20ba43dc 100644
--- a/task-sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py
+++ b/task-sdk/src/airflow/sdk/definitions/_internal/abstractoperator.py
@@ -130,6 +130,8 @@ class AbstractOperator(Templater, DAGNode):
_is_sensor: bool = False
_is_mapped: bool = False
_can_skip_downstream: bool = False
+ # Declared in Python, implemented by a Lang-SDK runtime. Set by
``@task.stub``.
+ is_stub: bool = False
@property
def dag_id(self) -> str:
diff --git a/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
b/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
index 4c030be4beb..3a68ecadfc2 100644
--- a/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
+++ b/task-sdk/src/airflow/sdk/definitions/mappedoperator.py
@@ -258,6 +258,7 @@ class OperatorPartial:
is_empty=issubclass(self.operator_class, EmptyOperator),
is_sensor=issubclass(self.operator_class, BaseSensorOperator),
can_skip_downstream=issubclass(self.operator_class, SkipMixin),
+ is_stub=self.operator_class.is_stub,
task_module=self.operator_class.__module__,
task_type=self.operator_class.__name__,
operator_name=operator_name,
@@ -309,6 +310,7 @@ class MappedOperator(AbstractOperator):
_is_empty: bool = attrs.field(alias="is_empty")
_can_skip_downstream: bool = attrs.field(alias="can_skip_downstream")
_is_sensor: bool = attrs.field(alias="is_sensor", default=False)
+ is_stub: bool = False
_task_module: str
task_type: str
_operator_name: str
@@ -378,6 +380,7 @@ class MappedOperator(AbstractOperator):
"partial_kwargs",
"operator_extra_links",
"returns_dag_result",
+ "is_stub",
}
@property
diff --git a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
index 8d606cf9680..4524c74ff79 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/schema.json
@@ -1,6 +1,6 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema";,
- "api_version": "2026-06-16",
+ "api_version": "2026-10-30",
"description": "Apache Airflow SDK Supervisor Schema",
"$defs": {
"AssetAliasReferenceAssetEventDagRun": {
@@ -4590,6 +4590,114 @@
"title": "XComSequenceSliceResult",
"type": "object"
},
+ "ArgValueSchema": {
+ "additionalProperties": {
+ "$ref": "#/$defs/JsonValue"
+ },
+ "title": "ArgValueSchema",
+ "type": "object"
+ },
+ "LiteralArgBinding": {
+ "description": "One positional stub-task argument carrying an inline
literal from the Dag file.",
+ "properties": {
+ "name": {
+ "title": "Name",
+ "type": "string"
+ },
+ "value_schema": {
+ "anyOf": [
+ {
+ "$ref": "#/$defs/ArgValueSchema"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null
+ },
+ "kind": {
+ "const": "literal",
+ "title": "Kind",
+ "type": "string"
+ },
+ "value": {
+ "anyOf": [
+ {
+ "$ref": "#/$defs/JsonValue"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null
+ },
+ "from_default": {
+ "default": false,
+ "title": "From Default",
+ "type": "boolean"
+ }
+ },
+ "required": [
+ "name",
+ "kind"
+ ],
+ "title": "LiteralArgBinding",
+ "type": "object"
+ },
+ "TaskArgBinding": {
+ "discriminator": {
+ "mapping": {
+ "literal": "#/$defs/LiteralArgBinding",
+ "xcom": "#/$defs/XComArgBinding"
+ },
+ "propertyName": "kind"
+ },
+ "oneOf": [
+ {
+ "$ref": "#/$defs/XComArgBinding"
+ },
+ {
+ "$ref": "#/$defs/LiteralArgBinding"
+ }
+ ],
+ "title": "TaskArgBinding"
+ },
+ "XComArgBinding": {
+ "description": "One positional stub-task argument pulled from an
upstream task's XCom.",
+ "properties": {
+ "name": {
+ "title": "Name",
+ "type": "string"
+ },
+ "value_schema": {
+ "anyOf": [
+ {
+ "$ref": "#/$defs/ArgValueSchema"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null
+ },
+ "kind": {
+ "const": "xcom",
+ "title": "Kind",
+ "type": "string"
+ },
+ "task_id": {
+ "title": "Task Id",
+ "type": "string"
+ }
+ },
+ "required": [
+ "name",
+ "kind",
+ "task_id"
+ ],
+ "title": "XComArgBinding",
+ "type": "object"
+ },
"AssetEventDagRunReference": {
"additionalProperties": false,
"description": "Schema for AssetEvent model used in DagRun.",
@@ -4981,6 +5089,21 @@
],
"default": null,
"title": "Start Date"
+ },
+ "arg_bindings": {
+ "anyOf": [
+ {
+ "items": {
+ "$ref": "#/$defs/TaskArgBinding"
+ },
+ "type": "array"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Arg Bindings"
}
},
"required": [
diff --git
a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py
b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py
index 9491a8993fd..7e5ce93f86b 100644
--- a/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py
@@ -37,8 +37,13 @@ def get_bundle() -> VersionBundle:
"""
from cadwyn import HeadVersion, Version, VersionBundle
+ from airflow.sdk.execution_time.schema.versions.v2026_10_30 import (
+ AddArgBindingsToSupervisorTIRunContext,
+ )
+
return VersionBundle(
HeadVersion(),
+ Version("2026-10-30", AddArgBindingsToSupervisorTIRunContext),
Version("2026-06-16"),
)
diff --git
a/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py
b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py
new file mode 100644
index 00000000000..e6b93f5dea8
--- /dev/null
+++ b/task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py
@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from cadwyn import VersionChange, schema
+
+from airflow.sdk.api.datamodels._generated import TIRunContext
+
+
+class AddArgBindingsToSupervisorTIRunContext(VersionChange):
+ """
+ Add the ``arg_bindings`` argument-binding spec for stub (foreign-runtime)
tasks.
+
+ Each entry is a discriminated union of ``XComArgBinding`` and
``LiteralArgBinding``
+ keyed on ``kind``. The supervisor-schema mirror of the execution API's
+ ``AddArgBindingsToTIRunContext``, named apart so the two migrations are
not confused.
+ """
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version =
(schema(TIRunContext).field("arg_bindings").didnt_exist,)
diff --git a/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py
b/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py
index 05218aded3d..cd5f5fff5fb 100644
--- a/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py
+++ b/task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py
@@ -105,12 +105,9 @@ _LATEST_VERSION = "3026-06-16"
class TestSchemaVersionMigratorDowngrade:
"""
Drive the downgrade direction against a mock bundle so we can pin
- *field-level* migration behaviour. The real supervisor bundle has
- no schema-level migrations on the IPC bodies yet, so it would no-op
- every version -- which proves nothing about the migration chain.
- The mock bundle's mechanism is identical to the real one, so what
- we prove about it applies to the real bundle the moment a
- ``schema(...)`` instruction lands.
+ *field-level* migration behaviour independent of the real bundle's
+ contents. The real bundle's ``arg_bindings`` migration is covered by
+ :class:`TestRealBundleArgBindingsDowngrade` below.
"""
@pytest.fixture
@@ -369,3 +366,107 @@ class TestLazyCadwynImport:
"assert 'cadwyn' in sys.modules, 'cadwyn should load when the
bundle is accessed'"
)
subprocess.run([sys.executable, "-c", code], check=True,
capture_output=True, text=True)
+
+
+class TestRealBundleArgBindingsDowngrade:
+ """
+ Drive the *real* supervisor bundle through the ``arg_bindings`` migration.
+
+ ``AddArgBindingsToSupervisorTIRunContext`` is the bundle's first
``schema(...)``
+ instruction on a model *nested* inside a registered body
+ (``StartupDetails.ti_context``); this pins that the downgrade
+ re-validation strips the nested field on the wire for a runtime
+ pinned to the previous version, and keeps it at head.
+ """
+
+ @pytest.fixture
+ def startup_details(self):
+ import datetime
+ import uuid
+
+ from airflow.sdk.api.datamodels._generated import (
+ BundleInfo,
+ DagRun,
+ DagRunState,
+ DagRunType,
+ TaskInstance,
+ TIRunContext,
+ )
+ from airflow.sdk.execution_time.comms import StartupDetails
+
+ now = datetime.datetime.now(datetime.timezone.utc)
+ return StartupDetails(
+ ti=TaskInstance(
+ id=uuid.uuid4(),
+ task_id="transform",
+ dag_id="d",
+ run_id="r",
+ try_number=1,
+ dag_version_id=uuid.uuid4(),
+ ),
+ dag_rel_path="d.py",
+ bundle_info=BundleInfo(name="b", version=None),
+ start_date=now,
+ ti_context=TIRunContext(
+ dag_run=DagRun(
+ dag_id="d",
+ run_id="r",
+ logical_date=now,
+ data_interval_start=None,
+ data_interval_end=None,
+ start_date=now,
+ end_date=None,
+ run_type=DagRunType.MANUAL,
+ state=DagRunState.RUNNING,
+ run_after=now,
+ consumed_asset_events=[],
+ partition_key=None,
+ ),
+ max_tries=1,
+ arg_bindings=[
+ # No value_schema: the unconstrained ("any") case rides
through the migrator too.
+ {"name": "country", "kind": "literal", "value": "uk"},
+ {
+ "name": "extracted",
+ "kind": "xcom",
+ "value_schema": {"type": "object"},
+ "task_id": "extract",
+ },
+ {
+ "name": "limit",
+ "kind": "literal",
+ "value_schema": {"type": "integer", "format": "int64"},
+ "value": 10,
+ "from_default": True,
+ },
+ ],
+ ),
+ sentry_integration="",
+ )
+
+ @pytest.fixture
+ def real_migrator(self) -> SchemaVersionMigrator:
+ return get_schema_version_migrator()
+
+ def test_downgrade_strips_arg_bindings_for_previous_version(self,
real_migrator, startup_details):
+ out = real_migrator.downgrade(startup_details,
"2026-06-16").model_dump()
+ assert "arg_bindings" not in out["ti_context"]
+
+ def test_head_version_keeps_arg_bindings(self, real_migrator,
startup_details):
+ from airflow.sdk.api.datamodels._generated import LiteralArgBinding,
XComArgBinding
+
+ out = real_migrator.downgrade(startup_details, "2026-10-30")
+ assert out.ti_context.arg_bindings is not None
+ literal, xcom, defaulted = (a.root for a in
out.ti_context.arg_bindings)
+ assert isinstance(literal, LiteralArgBinding)
+ assert literal.value == "uk"
+ assert literal.name == "country"
+ assert literal.from_default is False
+ assert literal.value_schema is None
+ assert isinstance(xcom, XComArgBinding)
+ assert xcom.task_id == "extract"
+ assert xcom.name == "extracted"
+ assert xcom.value_schema.root == {"type": "object"}
+ assert isinstance(defaulted, LiteralArgBinding)
+ assert defaulted.from_default is True
+ assert defaulted.value_schema.root == {"type": "integer", "format":
"int64"}
diff --git a/ts-sdk/src/generated/supervisor.ts
b/ts-sdk/src/generated/supervisor.ts
index 049b0c1ce92..e12e5d815ae 100644
--- a/ts-sdk/src/generated/supervisor.ts
+++ b/ts-sdk/src/generated/supervisor.ts
@@ -245,6 +245,18 @@ export type NextKwargs1 =
export type XcomKeysToClear = string[];
export type ShouldRetry = boolean;
export type StartDate2 = string | null;
+export type ArgBindings = TaskArgBinding[] | null;
+/**
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "TaskArgBinding".
+ */
+export type TaskArgBinding = XComArgBinding | LiteralArgBinding;
+export type Name8 = string;
+export type Kind = "xcom";
+export type TaskId1 = string;
+export type Name9 = string;
+export type Kind1 = "literal";
+export type FromDefault = boolean;
export type Type13 = "TaskCallbackRequest";
export type Filepath2 = string;
export type BundleName3 = string;
@@ -310,7 +322,7 @@ export type NextKwargs2 = {
} | null;
export type RenderedMapIndex1 = string | null;
export type Type20 = "DeferTask";
-export type Name8 = string;
+export type Name10 = string;
export type Key1 = string;
export type Type21 = "DeleteAssetStateStoreByName";
export type Uri5 = string;
@@ -324,7 +336,7 @@ export type Type24 = "DeleteVariable";
export type Key5 = string;
export type DagId6 = string;
export type RunId5 = string;
-export type TaskId1 = string;
+export type TaskId2 = string;
export type MapIndex1 = number | null;
export type Type25 = "DeleteXCom";
/**
@@ -362,11 +374,11 @@ export type ErrorType1 =
| "PERMISSION_DENIED"
| "GENERIC_ERROR"
| "API_SERVER_ERROR";
-export type Name9 = string;
+export type Name11 = string;
export type Type27 = "GetAssetByName";
export type Uri6 = string;
export type Type28 = "GetAssetByUri";
-export type Name10 = string | null;
+export type Name12 = string | null;
export type Uri7 = string | null;
export type After = string | null;
export type Before = string | null;
@@ -389,7 +401,7 @@ export type Extra8 = {
[k: string]: string;
} | null;
export type Type30 = "GetAssetEventByAssetAlias";
-export type Name11 = string;
+export type Name13 = string;
export type Key6 = string;
export type Type31 = "GetAssetStateStoreByName";
export type Uri8 = string;
@@ -421,7 +433,7 @@ export type LogicalDate3 = string;
export type State3 = string | null;
export type Type41 = "GetPreviousDagRun";
export type DagId12 = string;
-export type TaskId2 = string;
+export type TaskId3 = string;
export type LogicalDate4 = string | null;
export type MapIndex2 = number;
export type Type42 = "GetPreviousTI";
@@ -458,25 +470,25 @@ export type Type49 = "GetVariableKeys";
export type Key10 = string;
export type DagId16 = string;
export type RunId9 = string;
-export type TaskId3 = string;
+export type TaskId4 = string;
export type MapIndex5 = number | null;
export type IncludePriorDates = boolean;
export type Type50 = "GetXCom";
export type Key11 = string;
export type DagId17 = string;
export type RunId10 = string;
-export type TaskId4 = string;
+export type TaskId5 = string;
export type Type51 = "GetXComCount";
export type Key12 = string;
export type DagId18 = string;
export type RunId11 = string;
-export type TaskId5 = string;
+export type TaskId6 = string;
export type Offset1 = number;
export type Type52 = "GetXComSequenceItem";
export type Key13 = string;
export type DagId19 = string;
export type RunId12 = string;
-export type TaskId6 = string;
+export type TaskId7 = string;
export type Start = number | null;
export type Stop = number | null;
export type Step = number | null;
@@ -498,7 +510,7 @@ export type AssignedUsers1 = HITLUser[] | null;
export type Type54 = "HITLDetailRequestResult";
export type InactiveAssets = AssetProfile[] | null;
export type Type55 = "InactiveAssetsResult";
-export type Name12 = string | null;
+export type Name14 = string | null;
export type Type56 = "MaskSecret";
export type Ok = boolean;
export type Type57 = "OKResponse";
@@ -508,7 +520,7 @@ export type StartDate4 = string | null;
export type EndDate3 = string | null;
export type Type58 = "PrevSuccessfulDagRunResult";
export type Type59 = "PreviousDagRunResult";
-export type TaskId7 = string;
+export type TaskId8 = string;
export type DagId20 = string;
export type RunId13 = string;
export type LogicalDate5 = string | null;
@@ -536,7 +548,7 @@ export type RetryReason = string | null;
export type Type64 = "RetryTask";
export type Type65 = "SentFDs";
export type Fds = number[];
-export type Name13 = string;
+export type Name15 = string;
export type Key15 = string;
export type Type66 = "SetAssetStateStoreByName";
export type Uri9 = string;
@@ -552,7 +564,7 @@ export type Type70 = "SetTaskStateStore";
export type Key18 = string;
export type DagId21 = string;
export type RunId14 = string;
-export type TaskId8 = string;
+export type TaskId9 = string;
export type MapIndex7 = number | null;
export type DagResult1 = boolean;
export type MappedLength = number | null;
@@ -1019,6 +1031,7 @@ export interface TIRunContext {
xcom_keys_to_clear?: XcomKeysToClear;
should_retry?: ShouldRetry;
start_date?: StartDate2;
+ arg_bindings?: ArgBindings;
}
/**
* Variable schema for responses with fields that are needed for Runtime.
@@ -1030,6 +1043,38 @@ export interface VariableResponse {
key: Key;
value: Value;
}
+/**
+ * One positional stub-task argument pulled from an upstream task's XCom.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "XComArgBinding".
+ */
+export interface XComArgBinding {
+ name: Name8;
+ value_schema?: ArgValueSchema | null;
+ kind: Kind;
+ task_id: TaskId1;
+}
+/**
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "ArgValueSchema".
+ */
+export interface ArgValueSchema {
+ [k: string]: JsonValue;
+}
+/**
+ * One positional stub-task argument carrying an inline literal from the Dag
file.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "LiteralArgBinding".
+ */
+export interface LiteralArgBinding {
+ name: Name9;
+ value_schema?: ArgValueSchema | null;
+ kind: Kind1;
+ value?: unknown;
+ from_default?: FromDefault;
+}
/**
* Email notification request for task failures/retries.
*
@@ -1149,7 +1194,7 @@ export interface DeferTask {
* via the `definition` "DeleteAssetStateStoreByName".
*/
export interface DeleteAssetStateStoreByName {
- name: Name8;
+ name: Name10;
key: Key1;
type?: Type21;
}
@@ -1187,7 +1232,7 @@ export interface DeleteXCom {
key: Key5;
dag_id: DagId6;
run_id: RunId5;
- task_id: TaskId1;
+ task_id: TaskId2;
map_index?: MapIndex1;
type?: Type25;
}
@@ -1205,7 +1250,7 @@ export interface ErrorResponse {
* via the `definition` "GetAssetByName".
*/
export interface GetAssetByName {
- name: Name9;
+ name: Name11;
type?: Type27;
}
/**
@@ -1221,7 +1266,7 @@ export interface GetAssetByUri {
* via the `definition` "GetAssetEventByAsset".
*/
export interface GetAssetEventByAsset {
- name: Name10;
+ name: Name12;
uri: Uri7;
after?: After;
before?: Before;
@@ -1252,7 +1297,7 @@ export interface GetAssetEventByAssetAlias {
* via the `definition` "GetAssetStateStoreByName".
*/
export interface GetAssetStateStoreByName {
- name: Name11;
+ name: Name13;
key: Key6;
type?: Type31;
}
@@ -1354,7 +1399,7 @@ export interface GetPreviousDagRun {
*/
export interface GetPreviousTI {
dag_id: DagId12;
- task_id: TaskId2;
+ task_id: TaskId3;
logical_date?: LogicalDate4;
map_index?: MapIndex2;
state?: TaskInstanceState | null;
@@ -1440,7 +1485,7 @@ export interface GetXCom {
key: Key10;
dag_id: DagId16;
run_id: RunId9;
- task_id: TaskId3;
+ task_id: TaskId4;
map_index?: MapIndex5;
include_prior_dates?: IncludePriorDates;
type?: Type50;
@@ -1455,7 +1500,7 @@ export interface GetXComCount {
key: Key11;
dag_id: DagId17;
run_id: RunId10;
- task_id: TaskId4;
+ task_id: TaskId5;
type?: Type51;
}
/**
@@ -1466,7 +1511,7 @@ export interface GetXComSequenceItem {
key: Key12;
dag_id: DagId18;
run_id: RunId11;
- task_id: TaskId5;
+ task_id: TaskId6;
offset: Offset1;
type?: Type52;
}
@@ -1478,7 +1523,7 @@ export interface GetXComSequenceSlice {
key: Key13;
dag_id: DagId19;
run_id: RunId12;
- task_id: TaskId6;
+ task_id: TaskId7;
start: Start;
stop: Stop;
step: Step;
@@ -1520,7 +1565,7 @@ export interface InactiveAssetsResult {
*/
export interface MaskSecret {
value: JsonValue;
- name?: Name12;
+ name?: Name14;
type?: Type56;
}
/**
@@ -1559,7 +1604,7 @@ export interface PreviousDagRunResult {
* via the `definition` "PreviousTIResponse".
*/
export interface PreviousTIResponse {
- task_id: TaskId7;
+ task_id: TaskId8;
dag_id: DagId20;
run_id: RunId13;
logical_date?: LogicalDate5;
@@ -1636,7 +1681,7 @@ export interface SentFDs {
* via the `definition` "SetAssetStateStoreByName".
*/
export interface SetAssetStateStoreByName {
- name: Name13;
+ name: Name15;
key: Key15;
value: JsonValue;
type?: Type66;
@@ -1694,7 +1739,7 @@ export interface SetXCom {
value: JsonValue;
dag_id: DagId21;
run_id: RunId14;
- task_id: TaskId8;
+ task_id: TaskId9;
map_index?: MapIndex7;
dag_result?: DagResult1;
mapped_length?: MappedLength;
@@ -1896,4 +1941,4 @@ export interface XComSequenceSliceResult {
* (e.g. bundle metadata) and runs the migrator accordingly.
* Exposed so the SDK author / operator can confirm which schema
* version their build is pinned to. */
-export const SUPERVISOR_API_VERSION = "2026-06-16" as const;
+export const SUPERVISOR_API_VERSION = "2026-10-30" as const;
