Python: [Bug]: Workflow invocation kwargs collide with an executor ID of __global__

[ryo-whaletech]

Description

WorkflowInvocationKwargs exposes global and executor-specific kwargs as separate namespaces, but workflow invocation kwargs are currently normalized through a representation where __global__ can mean either the internal global slot or a real executor ID.

Because executor IDs do not reserve __global__, this affects two public input forms:

1. A typed WorkflowInvocationKwargs whose executor_kwargs contains an entry for executor ID __global__. The executor-specific entry replaces the real global mapping.
2. A plain per-executor mapping keyed by __global__. It is correctly recognized as targeting that real executor, but is later also interpreted as the global slot.

For executors __global__ and other, expected routing is:

typed wrapper:
  executor "__global__": {"shared": "G", "special": "A", "overridden": "specific"}
  executor "other":      {"shared": "G", "overridden": "global"}

plain per-executor mapping:
  executor "__global__": {"special": "A"}
  executor "other":      None / no targeted kwargs

Actual routing is:

typed wrapper:
  executor "__global__": {"special": "A", "overridden": "specific"}
  executor "other":      {"special": "A", "overridden": "specific"}

plain per-executor mapping:
  executor "__global__": {"special": "A"}
  executor "other":      {"special": "A"}

Executor-specific kwargs therefore leak to an unrelated executor. In the typed case, the genuine global kwargs are also lost. Both function_invocation_kwargs and client_kwargs are affected.

Code Sample

import asyncio
from typing import Any

from agent_framework import AgentResponse, BaseAgent, Message, WorkflowInvocationKwargs
from agent_framework.orchestrations import SequentialBuilder


class CapturingAgent(BaseAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name=name)
        self.seen: list[dict[str, Any]] = []

    def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
        self.seen.append(kwargs)

        async def result() -> AgentResponse[Any]:
            return AgentResponse(messages=[Message("assistant", ["ok"])])

        return result()


async def run_case(label: str, channel: str, invocation_kwargs: Any) -> None:
    global_agent = CapturingAgent("__global__")
    other_agent = CapturingAgent("other")
    workflow = SequentialBuilder(participants=[global_agent, other_agent]).build()

    await workflow.run("test", **{channel: invocation_kwargs})
    print(
        f"{label} / {channel}",
        (global_agent.seen[0][channel], other_agent.seen[0][channel]),
    )


async def main() -> None:
    cases = {
        "typed": WorkflowInvocationKwargs(
            global_kwargs={"shared": "G", "overridden": "global"},
            executor_kwargs={"__global__": {"special": "A", "overridden": "specific"}},
        ),
        "plain": {"__global__": {"special": "A"}},
    }
    for label, invocation_kwargs in cases.items():
        for channel in ("function_invocation_kwargs", "client_kwargs"):
            await run_case(label, channel, invocation_kwargs)


asyncio.run(main())

Error Messages / Stack Traces

No exception is raised. Current main prints:

typed / function_invocation_kwargs ({'special': 'A', 'overridden': 'specific'}, {'special': 'A', 'overridden': 'specific'})
typed / client_kwargs ({'special': 'A', 'overridden': 'specific'}, {'special': 'A', 'overridden': 'specific'})
plain / function_invocation_kwargs ({'special': 'A'}, {'special': 'A'})
plain / client_kwargs ({'special': 'A'}, {'special': 'A'})

Package Versions

agent-framework-core: 1.18.0; reproduced against main@4b4d849602d31471a0fa4721b7d34d40b6e42114

Python Version

Python 3.12.12

Additional Context

Reverified deterministically through the public Workflow.run(...) path against main@4b4d849602d31471a0fa4721b7d34d40b6e42114. Both the typed and plain collision cases reproduce for function_invocation_kwargs and client_kwargs. The latest Python release is 1.18.0. The reproductions use only local capturing agents—no provider, credentials, network, or timing dependency.

Historical context: review on PR #7963 explicitly identified the collision with an executor ID of __global__. A follow-up review called for separating internal global and executor-specific namespaces so plain mappings and executor IDs with that name would retain their behavior. The response confirmed that direction. Current main still uses the flattened representation in which those namespaces can collide.

A local fix using structurally separate global and executor-specific namespaces passes direct and nested regressions for both kwargs channels, including a plain global application kwarg named __global__, as well as an actual checkpoint save/restore cycle. Non-ambiguous Python 1.18.0-style state remains readable.

Old flattened state involving a real executor ID of __global__ is intrinsically ambiguous because the representation does not preserve whether that entry was global or executor-specific.

This appears to be ordinary functional correctness.

[github-actions]

🤖 Automated triage reproduction notes (agent-authored — trust but verify)

Agent analysis

python/packages/core/agent_framework/_workflows/_workflow.py::_resolve_invocation_kwargs flattens global and executor-specific mappings under the same __global__ key; _agent_executor.py::_resolve_executor_kwargs then treats that entry as global for every executor. Trigger it with executors named __global__ and other, passing either a typed wrapper with global plus __global__-specific kwargs or {"__global__": {...}} through either kwargs channel. The parameterized regression in test_workflow_kwargs.py reproduces all four failures through Workflow.run(...).

• Failing test: python/packages/core/tests/workflow/test_workflow_kwargs.py::test_executor_id_global_does_not_collide_with_global_kwargs
• Files examined: python/packages/core/agent_framework/_workflows/_workflow.py, python/packages/core/agent_framework/_workflows/_agent_executor.py, python/packages/core/agent_framework/_workflows/_workflow_executor.py, python/packages/core/agent_framework/_workflows/_const.py, python/packages/core/tests/workflow/test_workflow_kwargs.py, python/packages/core/pyproject.toml
• Tests run: test_executor_id_global_does_not_collide_with_global_kwargs (4 failed), focused existing kwargs-routing tests (5 passed)
• Reported version: 1.18.0
• Current version: 1.18.0

[he-yufeng]

Confirmed the mechanism. The collision comes from _resolve_invocation_kwargs normalizing both input forms into one flat dict where the global mapping and any executor ID share a single namespace, so an executor literally named __global__ overwrites (typed form) or impersonates (plain mapping) the global slot.

The route I'm taking: nest per-executor entries under a separate framework slot (__per_executor__) so the two namespaces can't overlap:

• typed wrapper -> {"__global__": ..., "__per_executor__": {...}}
• plain per-executor mapping -> {"__per_executor__": {...}} (global slot stays absent, so an untargeted executor still resolves to None)
• plain global mapping -> {"__global__": ..., "__per_executor__": {}}

_resolve_executor_kwargs reads the nested shape and falls back to the legacy flat shape for checkpoints serialized before the change, so in-flight state keeps routing as before. WorkflowExecutor unwraps the same way so subworkflows still see the plain mapping their own resolution expects.

Draft PR coming up.