pierrejeambrun · Oct 8, 2026 · Oct 5, 2026 · Oct 6, 2026 · Oct 6, 2026 · Oct 6, 2026
@@ -84,11 +84,25 @@ How it works
   Airflow 2 has no task-instance id, so there the key is
   ``<dag_id>/<run_id>/<task_id>/<map_index>/<try_number>``, and spans carry the five
   identity keys without ``airflow.task_instance.id``.
+* **Model name.** The model that actually answered is exposed on XCom under the
+  namespaced key ``__AIRFLOW__COMMON_AI_MODEL_NAME__``, separate from ``run_id`` /
+  ``usage`` above, so a downstream task, a Jinja template, or a REST client can read which
+  model responded without parsing the decision record or a span
+  (``ti.xcom_pull(task_ids="my_agent", key="__AIRFLOW__COMMON_AI_MODEL_NAME__")``).
+  Only ``AgentOperator`` /
+  ``@task.agent`` and ``LLMOperator`` / ``@task.llm`` push it today; like ``run_id`` and
+  ``usage``, with ``enable_hitl_review`` it is the initial run's model, not a
+  human-feedback regeneration's, since ``regenerate_with_feedback`` doesn't re-push it.
+  ``LLMOperator``'s subclasses that override ``execute()`` instead of calling
+  ``super().execute()`` -- ``LLMBranchOperator``, ``LLMSQLQueryOperator``,
+  ``LLMSchemaCompareOperator``, ``LLMFileAnalysisOperator`` -- don't publish it yet;
+  this is upcoming work.
 * **Scope.** The ``run_id`` / ``usage`` XComs come only from ``AgentOperator`` and
   ``@task.agent``, and so do the ``airflow.*`` identity attributes, apart from a Strands or
-  ADK agent run inside ``agent_framework_tracing`` (see below). The other LLM
+  ADK agent run inside ``agent_framework_tracing`` (see below). ``LLMOperator`` /
+  ``@task.llm`` additionally pushes the model name above. The other LLM
   operators still emit GenAI spans correlated to the task span by nesting, but
-  without the identity attributes or the run join key.
+  without the identity attributes, the run join key, or the model name.
 * **Content is off by default.** Only token counts, model id, latency, tool
   names, and finish reason are recorded. Prompt and completion text is never
   emitted unless you opt in (see below).

@@ -52,6 +52,7 @@
 )
 from airflow.providers.common.ai.toolsets.sandbox import SandboxToolset
 from airflow.providers.common.ai.utils.logging import (
+    MODEL_NAME_XCOM_KEY,
     format_usage_for_xcom,
     log_run_summary,
     log_run_usage,
@@ -1434,12 +1435,14 @@ def _emit_message_history(self, context: Context, result: Any) -> None:
         context["task_instance"].xcom_push(key="message_history", value=transcript)
 
     def _emit_run_metadata(self, context: Context, result: Any, *, usage: RunUsage) -> None:
-        """Expose the pydantic-ai run id and token usage on XCom for downstream tasks."""
+        """Expose the pydantic-ai run id, resolved model name, and token usage on XCom."""
         if not self.do_xcom_push:
             return
         ti = context["task_instance"]
         ti.xcom_push(key="run_id", value=result.run_id)
         ti.xcom_push(key="usage", value=format_usage_for_xcom(usage))
+        if (model_name := getattr(result.response, "model_name", None)) is not None:
+            ti.xcom_push(key=MODEL_NAME_XCOM_KEY, value=model_name)
 
     def regenerate_with_feedback(self, *, feedback: str, message_history: Any) -> tuple[str, Any]:
         """

@@ -43,7 +43,7 @@
     timed_out_record,
     validate_decision_policy,
 )
-from airflow.providers.common.ai.utils.logging import log_run_summary
+from airflow.providers.common.ai.utils.logging import MODEL_NAME_XCOM_KEY, log_run_summary
 from airflow.providers.common.ai.utils.output_type import rehydrate_pydantic_output
 from airflow.providers.common.ai.utils.usage import coerce_usage_limits
 from airflow.providers.common.compat.notifier import BaseNotifier
@@ -308,6 +308,8 @@ def execute(self, context: Context) -> Any:
         output = result.output
 
         model_confidence = ModelConfidence.from_result(result)
+        if model_confidence.model is not None:
+            self._push_xcom(context, MODEL_NAME_XCOM_KEY, model_confidence.model)
         # Gated by the least confident of the fields that reported a confidence; a field whose type
         # reports none is not gated. A bare output type is the one field ``response``.
         fields = list(model_confidence.confidence) or ["response"]
@@ -399,8 +401,8 @@ def _resume_after_review(
                 self._push_decision(context, timed_out_record(decision))
             raise
 
-    def _push_decision(self, context: Context, record: dict[str, Any]) -> None:
-        """Expose what the model proposed, its confidence, and what the gate decided, on XCom."""
+    def _push_xcom(self, context: Context, key: str, value: Any) -> None:
+        """Push ``value`` under ``key`` on XCom, honoring ``do_xcom_push`` and a missing task instance."""
         if not self.do_xcom_push:
             return
         try:
@@ -410,10 +412,14 @@ def _push_decision(self, context: Context, record: dict[str, Any]) -> None:
         push = getattr(ti, "xcom_push", None)
         if not callable(push):
             # A hand-built context (a dict, or no task instance at all) has nowhere to push to; the
-            # record is inspection output, so the run goes on without it.
-            self.log.warning("No task instance in the context; the decision record was not pushed to XCom.")
+            # value is inspection output, so the run goes on without it.
+            self.log.warning("No task instance in the context; %r was not pushed to XCom.", key)
             return
-        push(key=DECISION_XCOM_KEY, value=record)
+        push(key=key, value=value)
+
+    def _push_decision(self, context: Context, record: dict[str, Any]) -> None:
+        """Expose what the model proposed, its confidence, and what the gate decided, on XCom."""
+        self._push_xcom(context, DECISION_XCOM_KEY, record)
 
     def _finalize_decision(
         self, context: Context, event: dict[str, Any], decision: dict[str, Any] | None, *, action: Any

@@ -94,6 +94,12 @@ def _log_cache_and_cost(logger: Logger | logging.Logger, usage: RunUsage) -> Non
         logger.info("LLM run cost: $%s (USD, best-effort)", format(usage.cost, "f"))
 
 
+# XCom key the run's resolved model name is published under, so downstream tasks and the UI
+# can read which model actually answered without parsing the decision record. See the "Model
+# name" entry in docs/observability.rst for exactly which operators publish it.
+MODEL_NAME_XCOM_KEY = "__AIRFLOW__COMMON_AI_MODEL_NAME__"
+
+
 def format_usage_for_xcom(usage: RunUsage) -> dict[str, Any]:
     """Build the XCom ``usage`` payload -- shared by the success and failure paths."""
     return {

@@ -77,6 +77,7 @@
 from airflow.providers.common.ai.toolsets.mcp import MCPToolset
 from airflow.providers.common.ai.toolsets.sandbox import SandboxToolset
 from airflow.providers.common.ai.toolsets.sql import SQLToolset
+from airflow.providers.common.ai.utils.logging import MODEL_NAME_XCOM_KEY
 from airflow.providers.common.ai.utils.prompt_cache import PromptCaching
 from airflow.providers.common.ai.utils.toolset_base import MaskingToolset
 from airflow.providers.common.ai.utils.toolsets import find_toolset
@@ -1852,10 +1853,10 @@ def test_none_is_single_turn_no_history_no_emit(self, mock_hook_cls, make_mock_r
         op.execute(context=context)
 
         assert "message_history" not in mock_agent.run_sync.call_args.kwargs
-        # The transcript is not emitted without history, but run id + usage always are.
+        # The transcript is not emitted without history, but run id + usage + model always are.
         pushed_keys = {c.kwargs["key"] for c in context["task_instance"].xcom_push.call_args_list}
         assert "message_history" not in pushed_keys
-        assert pushed_keys == {"run_id", "usage"}
+        assert pushed_keys == {"run_id", "usage", MODEL_NAME_XCOM_KEY}
 
     @patch("airflow.providers.common.ai.operators.agent.PydanticAIHook", autospec=True)
     def test_transcript_emitted_to_xcom_when_history_set(self, mock_hook_cls, make_mock_run_result):
@@ -1870,7 +1871,7 @@ def test_transcript_emitted_to_xcom_when_history_set(self, mock_hook_cls, make_m
 
         ti = context["task_instance"]
         pushes = {c.kwargs["key"]: c.kwargs["value"] for c in ti.xcom_push.call_args_list}
-        assert set(pushes) == {"run_id", "usage", "message_history"}
+        assert set(pushes) == {"run_id", "usage", "message_history", MODEL_NAME_XCOM_KEY}
         restored = ModelMessagesTypeAdapter.validate_json(pushes["message_history"])
         assert len(restored) == 2
 
@@ -2258,8 +2259,8 @@ def model_fn(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
 
 class TestAgentOperatorRunIdentity:
     @patch("airflow.providers.common.ai.operators.agent.PydanticAIHook", autospec=True)
-    def test_run_id_and_usage_pushed_to_xcom(self, mock_hook_cls, make_mock_run_result):
-        """The pydantic-ai run id and token usage are exposed on XCom for downstream tasks."""
+    def test_run_id_usage_and_model_pushed_to_xcom(self, mock_hook_cls, make_mock_run_result):
+        """The pydantic-ai run id, resolved model name, and token usage are exposed on XCom."""
         mock_agent = _make_mock_agent("ok", make_mock_run_result)
         mock_agent.run_sync.return_value.run_id = "the-run-id"
         mock_hook_cls.get_hook.return_value.create_agent.return_value = mock_agent
@@ -2272,6 +2273,7 @@ def test_run_id_and_usage_pushed_to_xcom(self, mock_hook_cls, make_mock_run_resu
             c.kwargs["key"]: c.kwargs["value"] for c in context["task_instance"].xcom_push.call_args_list
         }
         assert pushes["run_id"] == "the-run-id"
+        assert pushes[MODEL_NAME_XCOM_KEY] == "test-model"
         assert pushes["usage"] == {
             "requests": 1,
             "input_tokens": 0,

@@ -35,6 +35,7 @@
 )
 from airflow.providers.common.ai.operators import llm as llm_module
 from airflow.providers.common.ai.operators.llm import DecisionPolicy, LLMOperator
+from airflow.providers.common.ai.utils.logging import MODEL_NAME_XCOM_KEY
 
 from tests_common.test_utils.version_compat import AIRFLOW_V_3_1_PLUS, AIRFLOW_V_3_3_PLUS
 
@@ -403,7 +404,23 @@ def test_hand_built_context_skips_the_decision_push_with_a_warning(
             output = op.execute(context)
 
         assert Summary.model_validate(output).text == "t"
-        assert "the decision record was not pushed to XCom" in caplog.text
+        assert "'decision' was not pushed to XCom" in caplog.text
+
+    @patch("airflow.providers.common.ai.operators.llm.PydanticAIHook", autospec=True)
+    def test_resolved_model_name_pushed_to_xcom(self, mock_hook_cls, make_mock_run_result):
+        """The model that actually answered is exposed on its own namespaced XCom key."""
+        mock_agent = MagicMock(spec=["run_sync"])
+        mock_agent.run_sync.return_value = self._result(make_mock_run_result, Summary(text="t"), None)
+        mock_hook_cls.get_hook.return_value.create_agent.return_value = mock_agent
+        op = LLMOperator(task_id="t", prompt="p", llm_conn_id="c", output_type=Summary)
+        context = MagicMock(spec=dict)
+
+        op.execute(context)
+
+        pushes = {
+            c.kwargs["key"]: c.kwargs["value"] for c in context["task_instance"].xcom_push.call_args_list
+        }
+        assert pushes[MODEL_NAME_XCOM_KEY] == "jev-1.13.0"
 
     @pytest.mark.skipif(
         not AIRFLOW_V_3_1_PLUS, reason="a reviewing decision_policy needs the HITL flow, Airflow >= 3.1"