Python: Fix workflow kwargs collision with __global__ executor IDs

python/packages/core/agent_framework/_workflows/_agent_executor.py

@@ -15,7 +15,7 @@
 from .._sessions import AgentSession
 from .._types import AgentResponse, AgentResponseUpdate, Message, ResponseStream
 from ._agent_utils import prepare_agent_run_args, resolve_agent_id, resolve_executor_kwargs
-from ._const import INTERNAL_SOURCE_ID, WORKFLOW_RUN_KWARGS_KEY
+from ._const import INTERNAL_SOURCE_ID, RESOLVED_WORKFLOW_RUN_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY
 from ._executor import Executor, handler
 from ._message_utils import normalize_messages_input
 from ._request_info_mixin import response_handler
@@ -441,7 +441,8 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR
             The complete AgentResponse, or None if waiting for user input.
         """
         raw_run_kwargs = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {})
-        function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs)
+        resolved_run_kwargs = ctx.get_state(RESOLVED_WORKFLOW_RUN_KWARGS_KEY)
+        function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs, resolved_run_kwargs)
         tools = ctx.get_runtime_tools()
 
         if not self._cache:
@@ -497,7 +498,8 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp
             The complete AgentResponse, or None if waiting for user input.
         """
         raw_run_kwargs = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {})
-        function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs)
+        resolved_run_kwargs = ctx.get_state(RESOLVED_WORKFLOW_RUN_KWARGS_KEY)
+        function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs, resolved_run_kwargs)
         tools = ctx.get_runtime_tools()
 
         if not self._cache:
@@ -582,6 +584,7 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp
     def _prepare_agent_run_args(
         self,
         raw_run_kwargs: dict[str, Any],
+        resolved_run_kwargs: Any = None,
     ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
         """Prepare function_invocation_kwargs and client_kwargs for agent.run().
 
@@ -594,7 +597,7 @@ def _prepare_agent_run_args(
         Returns:
             A 2-tuple of (function_invocation_kwargs, client_kwargs).
         """
-        return prepare_agent_run_args(self.id, raw_run_kwargs)
+        return prepare_agent_run_args(self.id, raw_run_kwargs, resolved_run_kwargs)
 
     def _resolve_executor_kwargs(self, resolved: dict[str, Any] | None) -> dict[str, Any] | None:
         """Extract this executor's kwargs from a resolved invocation kwargs dict.

python/packages/core/agent_framework/_workflows/_agent_utils.py

@@ -4,7 +4,7 @@
 from typing import Any, cast
 
 from .._agents import SupportsAgentRun
-from ._const import GLOBAL_KWARGS_KEY
+from ._const import GLOBAL_KWARGS_KEY, RAW_CLIENT_KWARGS_KEY, RAW_FUNCTION_INVOCATION_KWARGS_KEY
 
 logger = logging.getLogger(__name__)
 
@@ -23,22 +23,12 @@ def resolve_agent_id(agent: SupportsAgentRun) -> str:
     return agent.name if agent.name else agent.id
 
 
-def resolve_executor_kwargs(executor_id: str, resolved: dict[str, Any] | None) -> dict[str, Any] | None:
-    """Extract one executor's kwargs from a resolved invocation kwargs dict.
-
-    Args:
-        executor_id: The id of the executor whose kwargs are wanted.
-        resolved: The resolved dict produced by ``Workflow._resolve_invocation_kwargs``,
-            containing either a ``__global__`` key (global kwargs) or executor-ID keys
-            (per-executor kwargs). May also be ``None``.
-
-    Returns:
-        The kwargs for that executor, or ``None`` if not applicable.
-    """
-    if not isinstance(resolved, dict):
-        return None
-    global_kwargs: Any = resolved.get(GLOBAL_KWARGS_KEY)
-    executor_kwargs: Any = resolved.get(executor_id)
+def _merge_executor_kwargs(
+    executor_id: str,
+    global_kwargs: Any,
+    executor_kwargs: Any,
+) -> dict[str, Any] | None:
+    """Merge validated global and executor-specific kwargs for one executor."""
     if global_kwargs is None and executor_kwargs is None:
         return None
 
@@ -62,26 +52,102 @@ def resolve_executor_kwargs(executor_id: str, resolved: dict[str, Any] | None) -
     return {**(global_kwargs or {}), **(executor_kwargs or {})}
 
 
+def resolve_executor_kwargs(executor_id: str, resolved: dict[str, Any] | None) -> dict[str, Any] | None:
+    """Extract one executor's kwargs from a resolved invocation kwargs dict.
+
+    Args:
+        executor_id: The id of the executor whose kwargs are wanted.
+        resolved: A legacy-compatible resolved dict containing either a ``__global__``
+            key (global kwargs) or executor-ID keys (per-executor kwargs). May also be
+            ``None``.
+
+    Returns:
+        The kwargs for that executor, or ``None`` if not applicable.
+    """
+    if not isinstance(resolved, dict):
+        return None
+    return _merge_executor_kwargs(executor_id, resolved.get(GLOBAL_KWARGS_KEY), resolved.get(executor_id))
+
+
+def _resolve_structured_executor_kwargs(
+    executor_id: str,
+    resolved: Any,
+) -> dict[str, Any] | None:
+    """Extract one executor's kwargs from collision-free workflow run state."""
+    if not isinstance(resolved, dict):
+        raise TypeError("Resolved workflow invocation kwargs state must be a dict.")
+    resolved_dict = cast(dict[str, Any], resolved)
+    executor_kwargs = resolved_dict.get("executor_kwargs")
+    specific_kwargs = (
+        cast(dict[str, Any], executor_kwargs).get(executor_id) if isinstance(executor_kwargs, dict) else None
+    )
+    return _merge_executor_kwargs(executor_id, resolved_dict.get("global_kwargs"), specific_kwargs)
+
+
+def prepare_executor_run_kwargs(
+    executor_id: str,
+    raw_run_kwargs: dict[str, Any],
+    resolved_run_kwargs: Any = None,
+) -> dict[str, Any]:
+    """Prepare sanitized, executor-ready kwargs from workflow run state.
+
+    New collision-free state takes precedence when present. Legacy dict state remains
+    readable for checkpoint and package-version compatibility. Internal raw-routing
+    snapshots are never forwarded to agents.
+
+    Args:
+        executor_id: The id of the executor about to invoke an agent.
+        raw_run_kwargs: Legacy-compatible state stored under ``WORKFLOW_RUN_KWARGS_KEY``.
+        resolved_run_kwargs: Collision-free state stored under the resolved run-state key.
+
+    Returns:
+        A copy of the run kwargs containing only values applicable to ``executor_id``.
+    """
+    run_kwargs = dict(raw_run_kwargs)
+    run_kwargs.pop(RAW_FUNCTION_INVOCATION_KWARGS_KEY, None)
+    run_kwargs.pop(RAW_CLIENT_KWARGS_KEY, None)
+
+    if resolved_run_kwargs is not None and not isinstance(resolved_run_kwargs, dict):
+        raise TypeError("Resolved workflow run kwargs state must be a dict.")
+
+    for key in ("function_invocation_kwargs", "client_kwargs"):
+        if resolved_run_kwargs is not None:
+            executor_value = (
+                _resolve_structured_executor_kwargs(executor_id, resolved_run_kwargs[key])
+                if key in resolved_run_kwargs
+                else None
+            )
+        else:
+            executor_value = resolve_executor_kwargs(executor_id, raw_run_kwargs.get(key))
+        if executor_value is None:
+            run_kwargs.pop(key, None)
+        else:
+            run_kwargs[key] = executor_value
+
+    return run_kwargs
+
+
 def prepare_agent_run_args(
     executor_id: str,
     raw_run_kwargs: dict[str, Any],
+    resolved_run_kwargs: Any = None,
 ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
     """Prepare function_invocation_kwargs and client_kwargs for agent.run().
 
-    Extracts ``function_invocation_kwargs`` and ``client_kwargs`` from the workflow state
-    dict, resolving per-executor entries using ``executor_id``. The ``__global__`` sentinel
-    key (set by ``Workflow._resolve_invocation_kwargs``) denotes global kwargs that apply to
-    all executors. Per-executor dicts use executor IDs as keys; only the entry for
-    ``executor_id`` is extracted.
+    Extracts ``function_invocation_kwargs`` and ``client_kwargs`` from workflow state,
+    preferring collision-free state when present and otherwise reading the legacy dict
+    representation. Global and executor-specific values are merged for ``executor_id``.
 
     Args:
         executor_id: The id of the executor about to invoke the agent.
-        raw_run_kwargs: The workflow state dict stored under ``WORKFLOW_RUN_KWARGS_KEY``.
+        raw_run_kwargs: The legacy-compatible workflow run-state dict.
+        resolved_run_kwargs: Optional collision-free workflow run-state dict.
 
     Returns:
         A 2-tuple of (function_invocation_kwargs, client_kwargs).
     """
-    function_invocation_kwargs = resolve_executor_kwargs(executor_id, raw_run_kwargs.get("function_invocation_kwargs"))
-    client_kwargs = resolve_executor_kwargs(executor_id, raw_run_kwargs.get("client_kwargs"))
+    run_kwargs = prepare_executor_run_kwargs(executor_id, raw_run_kwargs, resolved_run_kwargs)
+    function_invocation_kwargs = run_kwargs.get("function_invocation_kwargs")
+    client_kwargs = run_kwargs.get("client_kwargs")
 
     return function_invocation_kwargs, client_kwargs

python/packages/core/agent_framework/_workflows/_const.py

@@ -17,12 +17,18 @@
 # to pass kwargs from workflow.run() through to agent.run() and @tool functions.
 WORKFLOW_RUN_KWARGS_KEY = "_workflow_run_kwargs"
 
+# Key used to store collision-free, executor-aware workflow invocation kwargs.
+# WORKFLOW_RUN_KWARGS_KEY remains in the legacy dict format for compatibility with
+# older first-party packages and checkpoints.
+RESOLVED_WORKFLOW_RUN_KWARGS_KEY = "_resolved_workflow_run_kwargs"
+
 # State keys used to preserve caller-provided kwargs for nested workflow routing.
 RAW_FUNCTION_INVOCATION_KWARGS_KEY = "_raw_function_invocation_kwargs"
 RAW_CLIENT_KWARGS_KEY = "_raw_client_kwargs"
 
-# Sentinel key used in resolved invocation kwargs dicts to denote global kwargs
-# that apply to all executors (as opposed to per-executor keyed entries).
+# Legacy sentinel used by plain mixed input and the compatibility state stored at
+# WORKFLOW_RUN_KWARGS_KEY. New resolution state keeps global and executor-specific
+# namespaces separate.
 GLOBAL_KWARGS_KEY = "__global__"
 
 

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -13,7 +13,7 @@
 import weakref
 from collections.abc import AsyncIterable, Awaitable, Callable, Collection, Mapping, Sequence
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any, Literal, overload
+from typing import TYPE_CHECKING, Any, Literal, cast, overload
 
 from .._sessions import ContextProvider
 from .._tools import ToolTypes, normalize_tools
@@ -27,6 +27,7 @@
     INTERNAL_SOURCE_ID,
     RAW_CLIENT_KWARGS_KEY,
     RAW_FUNCTION_INVOCATION_KWARGS_KEY,
+    RESOLVED_WORKFLOW_RUN_KWARGS_KEY,
     WORKFLOW_RUN_KWARGS_KEY,
 )
 from ._edge import (
@@ -576,25 +577,30 @@ async def _run_workflow_with_tracing(
                 #   explicitly provides new kwargs.
                 if function_invocation_kwargs is not None or client_kwargs is not None:
                     combined_kwargs: dict[str, Any] = {}
+                    resolved_combined_kwargs: dict[str, Any] = {}
                     if function_invocation_kwargs is not None:
-                        combined_kwargs["function_invocation_kwargs"] = self._resolve_invocation_kwargs(
+                        resolved = self._resolve_invocation_kwargs(
                             function_invocation_kwargs, "function_invocation_kwargs"
                         )
+                        resolved_combined_kwargs["function_invocation_kwargs"] = resolved
+                        combined_kwargs["function_invocation_kwargs"] = self._to_legacy_invocation_kwargs(resolved)
                         if isinstance(function_invocation_kwargs, WorkflowInvocationKwargs) or any(
                             isinstance(value, Mapping) for value in function_invocation_kwargs.values()
                         ):
                             combined_kwargs[RAW_FUNCTION_INVOCATION_KWARGS_KEY] = function_invocation_kwargs
                     if client_kwargs is not None:
-                        combined_kwargs["client_kwargs"] = self._resolve_invocation_kwargs(
-                            client_kwargs, "client_kwargs"
-                        )
+                        resolved = self._resolve_invocation_kwargs(client_kwargs, "client_kwargs")
+                        resolved_combined_kwargs["client_kwargs"] = resolved
+                        combined_kwargs["client_kwargs"] = self._to_legacy_invocation_kwargs(resolved)
                         if isinstance(client_kwargs, WorkflowInvocationKwargs) or any(
                             isinstance(value, Mapping) for value in client_kwargs.values()
                         ):
                             combined_kwargs[RAW_CLIENT_KWARGS_KEY] = client_kwargs
                     self._runner.state.set(WORKFLOW_RUN_KWARGS_KEY, combined_kwargs)
+                    self._runner.state.set(RESOLVED_WORKFLOW_RUN_KWARGS_KEY, resolved_combined_kwargs)
                 elif not is_continuation:
                     self._runner.state.set(WORKFLOW_RUN_KWARGS_KEY, {})
+                    self._runner.state.set(RESOLVED_WORKFLOW_RUN_KWARGS_KEY, {})
                 self._runner.state.commit()  # Commit immediately so kwargs are available
 
                 # Explicitly set streaming mode per run
@@ -1125,13 +1131,13 @@ def _resolve_invocation_kwargs(
         kwargs: WorkflowInvocationKwargs | Mapping[str, Any],
         param_name: str,
     ) -> dict[str, Any]:
-        """Resolve invocation kwargs into a normalized per-executor or global format.
+        """Resolve invocation kwargs into collision-free global and executor namespaces.
 
         Detects whether the provided kwargs dict uses per-executor targeting by checking
-        if any top-level key matches a known executor ID in the workflow. If at least one
-        key matches, all entries are treated as per-executor. Otherwise the dict is treated
-        as global kwargs that apply to every executor. The ``"__global__"`` key can be used
-        explicitly to combine global kwargs with per-executor overrides.
+        if any top-level key matches a known executor ID in the workflow. A legacy
+        ``"__global__"`` slot is separated from matched executor entries unless that name
+        is itself a real executor ID. If no executor ID matches, the complete dict is
+        treated as global application kwargs.
 
         Args:
             kwargs: The raw invocation kwargs from the caller.
@@ -1141,29 +1147,51 @@ def _resolve_invocation_kwargs(
             A dict containing normalized global or per-executor mappings.
         """
         if isinstance(kwargs, WorkflowInvocationKwargs):
-            resolved = {GLOBAL_KWARGS_KEY: dict(kwargs.global_kwargs)}
-            resolved.update({
-                executor_id: dict(executor_kwargs) for executor_id, executor_kwargs in kwargs.executor_kwargs.items()
-            })
             logger.info("Explicit global %s provided with executor-specific overrides.", param_name)
-            return resolved
+            return {
+                "global_kwargs": dict(kwargs.global_kwargs),
+                "executor_kwargs": {
+                    executor_id: dict(executor_kwargs)
+                    for executor_id, executor_kwargs in kwargs.executor_kwargs.items()
+                },
+            }
 
         executor_ids = set(self.executors.keys())
         matched_ids = kwargs.keys() & executor_ids
         if matched_ids:
+            executor_kwargs = dict(kwargs)
+            if GLOBAL_KWARGS_KEY not in executor_ids and GLOBAL_KWARGS_KEY in executor_kwargs:
+                global_kwargs = executor_kwargs.pop(GLOBAL_KWARGS_KEY)
+                logger.info(
+                    "Detected legacy mixed %s with global values and executor ID(s) %s.",
+                    param_name,
+                    matched_ids,
+                )
+                return {"global_kwargs": global_kwargs, "executor_kwargs": executor_kwargs}
             logger.info(
                 "Detected per-executor %s: executor ID(s) %s found in keys. "
                 "All entries will be treated as per-executor.",
                 param_name,
                 matched_ids,
             )
-            return dict(kwargs)
+            return {"executor_kwargs": executor_kwargs}
 
         logger.info(
             "No executor IDs found in %s keys; treating as global kwargs for all executors.",
             param_name,
         )
-        return {GLOBAL_KWARGS_KEY: dict(kwargs)}
+        return {"global_kwargs": dict(kwargs), "executor_kwargs": {}}
+
+    @staticmethod
+    def _to_legacy_invocation_kwargs(resolved: dict[str, Any]) -> dict[str, Any]:
+        """Encode collision-free state in the existing best-effort legacy format."""
+        legacy: dict[str, Any] = {}
+        if "global_kwargs" in resolved:
+            legacy[GLOBAL_KWARGS_KEY] = resolved["global_kwargs"]
+        executor_kwargs = resolved.get("executor_kwargs")
+        if isinstance(executor_kwargs, dict):
+            legacy.update(cast(dict[str, Any], executor_kwargs))
+        return legacy
 
     # Graph signature helpers
 

python/packages/core/tests/workflow/test_agent_executor.py

@@ -20,8 +20,13 @@
     WorkflowRunState,
 )
 from agent_framework._workflows._agent_executor import AgentExecutorResponse
+from agent_framework._workflows._agent_utils import prepare_executor_run_kwargs
 from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
-from agent_framework._workflows._const import GLOBAL_KWARGS_KEY
+from agent_framework._workflows._const import (
+    GLOBAL_KWARGS_KEY,
+    RAW_CLIENT_KWARGS_KEY,
+    RAW_FUNCTION_INVOCATION_KWARGS_KEY,
+)
 
 
 class _CountingAgent(BaseAgent):
@@ -330,6 +335,34 @@ async def test_prepare_agent_run_args_returns_none_when_no_kwargs() -> None:
     assert ci_kwargs is None
 
 
+async def test_prepare_executor_run_kwargs_resolves_channels_and_removes_internal_state() -> None:
+    """Executor-ready kwargs preserve options without leaking raw-routing snapshots."""
+    raw = {
+        "function_invocation_kwargs": {GLOBAL_KWARGS_KEY: {"legacy": True}},
+        "client_kwargs": {GLOBAL_KWARGS_KEY: {"legacy": True}},
+        RAW_FUNCTION_INVOCATION_KWARGS_KEY: {"agent": {"raw": True}},
+        RAW_CLIENT_KWARGS_KEY: {"agent": {"raw": True}},
+        "options": {"temperature": 0.5},
+    }
+    resolved = {
+        "function_invocation_kwargs": {
+            "global_kwargs": {"shared": "G"},
+            "executor_kwargs": {"agent": {"specific": "A"}},
+        },
+        "client_kwargs": {"executor_kwargs": {"other": {"ignored": True}}},
+    }
+
+    actual = prepare_executor_run_kwargs("agent", raw, resolved)
+
+    assert actual == {
+        "function_invocation_kwargs": {"shared": "G", "specific": "A"},
+        "options": {"temperature": 0.5},
+    }
+    assert prepare_executor_run_kwargs("agent", raw, {}) == {"options": {"temperature": 0.5}}
+    with pytest.raises(TypeError, match="Resolved workflow run kwargs state must be a dict"):
+        prepare_executor_run_kwargs("agent", raw, "invalid")
+
+
 class _NonCopyableRaw:
     """Simulates an LLM SDK response object that cannot be deep-copied (e.g., proto/gRPC)."""