diff --git a/python/packages/core/agent_framework/_workflows/_const.py b/python/packages/core/agent_framework/_workflows/_const.py index d939bd448cb..db99b45426a 100644 --- a/python/packages/core/agent_framework/_workflows/_const.py +++ b/python/packages/core/agent_framework/_workflows/_const.py @@ -22,6 +22,10 @@ # older first-party packages and checkpoints. RESOLVED_WORKFLOW_RUN_KWARGS_KEY = "_resolved_workflow_run_kwargs" +# Key used to record caller keys that targeted executors in the current workflow. +# Nested workflows use this to avoid reinterpreting parent-scoped entries. +ROUTED_WORKFLOW_RUN_KWARGS_KEY = "_routed_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" diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index be6d8ffc707..1fb3f54d947 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -33,6 +33,7 @@ RAW_CLIENT_KWARGS_KEY, RAW_FUNCTION_INVOCATION_KWARGS_KEY, RESOLVED_WORKFLOW_RUN_KWARGS_KEY, + ROUTED_WORKFLOW_RUN_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY, ) from ._edge import ( @@ -599,29 +600,38 @@ async def _run_workflow_with_tracing( if function_invocation_kwargs is not None or client_kwargs is not None: combined_kwargs: dict[str, Any] = {} resolved_combined_kwargs: dict[str, Any] = {} + routed_combined_kwargs: dict[str, list[str]] = {} if function_invocation_kwargs is not None: - resolved = self._resolve_invocation_kwargs( + resolved, routed_keys = self._resolve_invocation_kwargs( function_invocation_kwargs, "function_invocation_kwargs" ) resolved_combined_kwargs["function_invocation_kwargs"] = resolved + routed_combined_kwargs["function_invocation_kwargs"] = sorted(routed_keys) 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() + if ( + isinstance(function_invocation_kwargs, WorkflowInvocationKwargs) + or routed_keys + 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: - resolved = self._resolve_invocation_kwargs(client_kwargs, "client_kwargs") + resolved, routed_keys = self._resolve_invocation_kwargs(client_kwargs, "client_kwargs") resolved_combined_kwargs["client_kwargs"] = resolved + routed_combined_kwargs["client_kwargs"] = sorted(routed_keys) 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() + if ( + isinstance(client_kwargs, WorkflowInvocationKwargs) + or routed_keys + 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) + self._runner.state.set(ROUTED_WORKFLOW_RUN_KWARGS_KEY, routed_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.set(ROUTED_WORKFLOW_RUN_KWARGS_KEY, {}) self._runner.state.commit() # Commit immediately so kwargs are available # Explicitly set streaming mode per run @@ -1158,57 +1168,117 @@ def _resolve_invocation_kwargs( self, kwargs: WorkflowInvocationKwargs | Mapping[str, Any], param_name: str, - ) -> dict[str, Any]: + ) -> tuple[dict[str, Any], set[str]]: """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. 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. + Detects whether a plain kwargs dict uses per-executor targeting by checking if any + top-level key matches a known executor ID in the workflow. Explicit executor kwargs + may also target a unique wrapped agent name. A legacy ``"__global__"`` slot is + separated from matched executor entries unless that name is itself a real executor + ID. If no executor ID matches a plain mapping, the complete dict is treated as + global application kwargs. Args: kwargs: The raw invocation kwargs from the caller. param_name: The parameter name (for logging), e.g. ``"function_invocation_kwargs"``. Returns: - A dict containing normalized global or per-executor mappings. + The normalized global or per-executor mappings and the caller keys routed + to executors in this workflow. """ + executor_ids = set(self.executors.keys()) + agent_name_aliases: dict[str, str] = {} + ambiguous_agent_names: set[str] = set() + from ._agent_executor import AgentExecutor + + for executor_id, executor in self.executors.items(): + if not isinstance(executor, AgentExecutor): + continue + agent_name = executor.agent.name + if not agent_name or agent_name == executor_id or agent_name in executor_ids: + continue + if agent_name in agent_name_aliases: + ambiguous_agent_names.add(agent_name) + agent_name_aliases.pop(agent_name) + elif agent_name not in ambiguous_agent_names: + agent_name_aliases[agent_name] = executor_id + + is_explicit = isinstance(kwargs, WorkflowInvocationKwargs) + provided_executor_kwargs = kwargs.executor_kwargs if is_explicit else kwargs + matched_ids = provided_executor_kwargs.keys() & executor_ids + matched_aliases: set[str] = set() + if is_explicit: + ambiguous_matches = provided_executor_kwargs.keys() & ambiguous_agent_names + if ambiguous_matches: + names = ", ".join(sorted(ambiguous_matches)) + raise ValueError( + f"Ambiguous agent name(s) in {param_name}: {names}. " + "Use the corresponding executor IDs for per-executor kwargs." + ) + matched_aliases = provided_executor_kwargs.keys() & agent_name_aliases.keys() + routed_keys = matched_ids | matched_aliases + + def normalize_executor_kwargs( + values: Mapping[str, Any], + *, + copy_values: bool = False, + resolve_aliases: bool = False, + ) -> dict[str, Any]: + normalized: dict[str, Any] = {} + for key, value in values.items(): + executor_id = agent_name_aliases.get(key, key) if resolve_aliases else key + if executor_id in normalized: + raise ValueError(f"{param_name} targets executor '{executor_id}' by both its ID and agent name.") + normalized[executor_id] = dict(value) if copy_values else value + return normalized + if isinstance(kwargs, WorkflowInvocationKwargs): logger.info("Explicit global %s provided with executor-specific overrides.", param_name) - return { - "global_kwargs": dict(kwargs.global_kwargs), - "executor_kwargs": { - executor_id: dict(executor_kwargs) - for executor_id, executor_kwargs in kwargs.executor_kwargs.items() + return ( + { + "global_kwargs": dict(kwargs.global_kwargs), + "executor_kwargs": normalize_executor_kwargs( + kwargs.executor_kwargs, + copy_values=True, + resolve_aliases=True, + ), }, - } + routed_keys, + ) - executor_ids = set(self.executors.keys()) - matched_ids = kwargs.keys() & executor_ids - if matched_ids: + if routed_keys: executor_kwargs = dict(kwargs) - if GLOBAL_KWARGS_KEY not in executor_ids and GLOBAL_KWARGS_KEY in executor_kwargs: + if ( + GLOBAL_KWARGS_KEY not in executor_ids + and GLOBAL_KWARGS_KEY not in routed_keys + 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, + routed_keys, + ) + return ( + { + "global_kwargs": global_kwargs, + "executor_kwargs": normalize_executor_kwargs(executor_kwargs), + }, + routed_keys, ) - 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, + routed_keys, ) - return {"executor_kwargs": executor_kwargs} + return {"executor_kwargs": normalize_executor_kwargs(executor_kwargs)}, routed_keys logger.info( "No executor IDs found in %s keys; treating as global kwargs for all executors.", param_name, ) - return {"global_kwargs": dict(kwargs), "executor_kwargs": {}} + return {"global_kwargs": dict(kwargs), "executor_kwargs": {}}, set() @staticmethod def _to_legacy_invocation_kwargs(resolved: dict[str, Any]) -> dict[str, Any]: diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 691dafc03b8..7c2217edc00 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -14,6 +14,8 @@ GLOBAL_KWARGS_KEY, RAW_CLIENT_KWARGS_KEY, RAW_FUNCTION_INVOCATION_KWARGS_KEY, + RESOLVED_WORKFLOW_RUN_KWARGS_KEY, + ROUTED_WORKFLOW_RUN_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY, ) from ._edge_runner import gather_cancelling_siblings_on_error @@ -38,6 +40,44 @@ logger = logging.getLogger(__name__) +def _scope_invocation_kwargs_to_subworkflow( + invocation_kwargs: WorkflowInvocationKwargs | Mapping[str, Any], + parent_routed_keys: set[str], + parent_resolved_kwargs: Mapping[str, Any] | None, +) -> WorkflowInvocationKwargs | Mapping[str, Any] | None: + """Remove kwargs entries already routed within the parent workflow.""" + if not parent_routed_keys: + return invocation_kwargs + if isinstance(invocation_kwargs, WorkflowInvocationKwargs): + executor_kwargs = { + key: value for key, value in invocation_kwargs.executor_kwargs.items() if key not in parent_routed_keys + } + if not invocation_kwargs.global_kwargs and not executor_kwargs: + return None + return WorkflowInvocationKwargs( + global_kwargs=invocation_kwargs.global_kwargs, + executor_kwargs=executor_kwargs, + ) + scoped = {key: value for key, value in invocation_kwargs.items() if key not in parent_routed_keys} + parent_global_kwargs = parent_resolved_kwargs.get("global_kwargs") if parent_resolved_kwargs else None + if ( + GLOBAL_KWARGS_KEY in scoped + and GLOBAL_KWARGS_KEY not in parent_routed_keys + and isinstance(parent_global_kwargs, Mapping) + ): + executor_kwargs = {key: value for key, value in scoped.items() if key != GLOBAL_KWARGS_KEY} + if any(not isinstance(value, Mapping) for value in executor_kwargs.values()): + return scoped + return WorkflowInvocationKwargs( + global_kwargs=cast(Mapping[str, Any], parent_global_kwargs), + executor_kwargs=cast( + Mapping[str, Mapping[str, Any]], + executor_kwargs, + ), + ) + return scoped or None + + @dataclass class ExecutionContext: """Legacy per-execution bookkeeping. @@ -380,6 +420,8 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A # Get kwargs from parent workflow's State to propagate to subworkflow parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {}) + parent_resolved_kwargs: dict[str, Any] = ctx.get_state(RESOLVED_WORKFLOW_RUN_KWARGS_KEY, {}) + parent_routed_kwargs: dict[str, list[str]] = ctx.get_state(ROUTED_WORKFLOW_RUN_KWARGS_KEY, {}) # Use the caller's raw kwargs so legacy per-executor mappings are resolved # against the child workflow's executor IDs rather than the parent's. @@ -391,7 +433,23 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A ) raw_value = parent_kwargs.get(raw_key) if raw_value is not None: - resolved = cast(WorkflowInvocationKwargs | Mapping[str, Any], raw_value) + routed_keys = set(parent_routed_kwargs.get(key, ())) + transparent_alias = self._transparent_agent_name_alias() if self.id in routed_keys else None + resolved = _scope_invocation_kwargs_to_subworkflow( + cast(WorkflowInvocationKwargs | Mapping[str, Any], raw_value), + self._parent_routed_keys_to_exclude(routed_keys), + cast(Mapping[str, Any] | None, parent_resolved_kwargs.get(key)), + ) + if ( + transparent_alias is not None + and isinstance(resolved, Mapping) + and not isinstance(resolved, WorkflowInvocationKwargs) + and self.id in resolved + ): + resolved = dict(resolved) + if transparent_alias in resolved: + raise ValueError(f"{key} targets executor '{transparent_alias}' by both its ID and agent name.") + resolved[transparent_alias] = resolved.pop(self.id) else: normalized: Any = parent_kwargs.get(key) if isinstance(normalized, dict): @@ -418,6 +476,32 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A # Process the workflow result using shared logic await self._process_workflow_result(result, ctx) + def _parent_routed_keys_to_exclude(self, routed_keys: set[str]) -> set[str]: + """Return parent-routed keys that must not be reinterpreted by the child workflow.""" + if self.id in self.workflow.executors: + return routed_keys - {self.id} + + if self._transparent_agent_name_alias() is not None: + return routed_keys - {self.id} + + return routed_keys + + def _transparent_agent_name_alias(self) -> str | None: + """Return the child executor ID uniquely identified by this wrapper's ID.""" + if self.id in self.workflow.executors: + return None + + from ._agent_executor import AgentExecutor + + matching_agent_names = [ + executor_id + for executor_id, executor in self.workflow.executors.items() + if isinstance(executor, AgentExecutor) and executor.agent.name == self.id + ] + if len(matching_agent_names) == 1: + return matching_agent_names[0] + return None + @handler async def handle_message_wrapped_request_response( self, diff --git a/python/packages/core/tests/workflow/test_workflow_kwargs.py b/python/packages/core/tests/workflow/test_workflow_kwargs.py index 6685c70827a..ba91dc601f8 100644 --- a/python/packages/core/tests/workflow/test_workflow_kwargs.py +++ b/python/packages/core/tests/workflow/test_workflow_kwargs.py @@ -1237,6 +1237,216 @@ async def test_mixed_kwargs_route_through_subworkflow() -> None: } +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +@pytest.mark.parametrize("typed", [True, False], ids=["typed", "plain"]) +async def test_parent_executor_kwargs_do_not_route_to_same_id_in_subworkflow( + kwargs_channel: str, + typed: bool, +) -> None: + """Parent-targeted kwargs do not cross into a child executor with the same ID.""" + from agent_framework import AgentExecutor, Executor, WorkflowBuilder, WorkflowContext, handler + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + class _FanOut(Executor): + @handler + async def fan_out(self, message: str, ctx: WorkflowContext[str]) -> None: + await ctx.send_message(message) + + parent_agent = _KwargsCapturingAgent(name="parent_agent") + child_collision = _KwargsCapturingAgent(name="child_collision") + child_target = _KwargsCapturingAgent(name="child_target") + child = SequentialBuilder( + participants=[ + AgentExecutor(child_collision, id="shared"), + AgentExecutor(child_target, id="child"), + ] + ).build() + dispatcher = _FanOut(id="dispatcher") + parent_executor = AgentExecutor(parent_agent, id="shared") + nested_executor = WorkflowExecutor(child, id="nested") + parent = ( + WorkflowBuilder(start_executor=dispatcher) + .add_fan_out_edges(dispatcher, [parent_executor, nested_executor]) + .build() + ) + executor_kwargs = { + "shared": {"api_key": "parent-only"}, + "child": {"api_key": "child-only"}, + } + invocation_kwargs: Mapping[str, Any] | WorkflowInvocationKwargs + invocation_kwargs = WorkflowInvocationKwargs(executor_kwargs=executor_kwargs) if typed else executor_kwargs + + if kwargs_channel == "function_invocation_kwargs": + await parent.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await parent.run("test", client_kwargs=invocation_kwargs) + + assert parent_agent.captured_kwargs[0].get(kwargs_channel) == {"api_key": "parent-only"} + assert child_collision.captured_kwargs[0].get(kwargs_channel) == ({} if typed else None) + assert child_target.captured_kwargs[0].get(kwargs_channel) == {"api_key": "child-only"} + + +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_non_mapping_parent_executor_kwargs_do_not_become_child_globals( + kwargs_channel: str, + caplog: "LogCaptureFixture", +) -> None: + """Malformed parent-targeted values remain scoped to the parent graph.""" + from agent_framework import AgentExecutor, Executor, WorkflowBuilder, WorkflowContext, handler + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + class _FanOut(Executor): + @handler + async def fan_out(self, message: str, ctx: WorkflowContext[str]) -> None: + await ctx.send_message(message) + + parent_agent = _KwargsCapturingAgent(name="parent") + child_agent = _KwargsCapturingAgent(name="child") + child = SequentialBuilder(participants=[child_agent]).build() + dispatcher = _FanOut(id="dispatcher") + parent_executor = AgentExecutor(parent_agent, id="parent") + nested_executor = WorkflowExecutor(child, id="nested") + parent = ( + WorkflowBuilder(start_executor=dispatcher) + .add_fan_out_edges(dispatcher, [parent_executor, nested_executor]) + .build() + ) + invocation_kwargs = {"parent": "invalid"} + + if kwargs_channel == "function_invocation_kwargs": + await parent.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await parent.run("test", client_kwargs=invocation_kwargs) + + assert parent_agent.captured_kwargs[0].get(kwargs_channel) is None + assert child_agent.captured_kwargs[0].get(kwargs_channel) is None + assert any("expected a dict for its kwargs" in record.message for record in caplog.records) + + +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +@pytest.mark.parametrize("child_kwargs", [{"child_only": True}, None, "invalid"]) +async def test_legacy_mixed_global_survives_parent_key_filtering( + kwargs_channel: str, + child_kwargs: Mapping[str, Any] | str | None, + caplog: "LogCaptureFixture", +) -> None: + """A legacy global slot retains its meaning after parent-targeted entries are removed.""" + from agent_framework import AgentExecutor, Executor, WorkflowBuilder, WorkflowContext, handler + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + class _FanOut(Executor): + @handler + async def fan_out(self, message: str, ctx: WorkflowContext[str]) -> None: + await ctx.send_message(message) + + parent_agent = _KwargsCapturingAgent(name="parent") + child_agent = _KwargsCapturingAgent(name="child") + child = SequentialBuilder(participants=[child_agent]).build() + dispatcher = _FanOut(id="dispatcher") + parent_executor = AgentExecutor(parent_agent, id="parent") + nested_executor = WorkflowExecutor(child, id="nested") + parent = ( + WorkflowBuilder(start_executor=dispatcher) + .add_fan_out_edges(dispatcher, [parent_executor, nested_executor]) + .build() + ) + invocation_kwargs = { + "__global__": {"shared": "global"}, + "parent": {"parent_only": True}, + "child": child_kwargs, + } + + if kwargs_channel == "function_invocation_kwargs": + await parent.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await parent.run("test", client_kwargs=invocation_kwargs) + + assert parent_agent.captured_kwargs[0].get(kwargs_channel) == { + "shared": "global", + "parent_only": True, + } + if isinstance(child_kwargs, Mapping): + assert child_agent.captured_kwargs[0].get(kwargs_channel) == { + "shared": "global", + "child_only": True, + } + elif child_kwargs is None: + assert child_agent.captured_kwargs[0].get(kwargs_channel) == {"shared": "global"} + else: + assert child_agent.captured_kwargs[0].get(kwargs_channel) is None + assert any("expected a dict for its kwargs" in record.message for record in caplog.records) + + +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_transparent_subworkflow_preserves_only_its_own_parent_key(kwargs_channel: str) -> None: + """A transparent wrapper preserves its own key without leaking unrelated parent keys.""" + from agent_framework import AgentExecutor, Executor, WorkflowBuilder, WorkflowContext, handler + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + class _FanOut(Executor): + @handler + async def fan_out(self, message: str, ctx: WorkflowContext[str]) -> None: + await ctx.send_message(message) + + wrapped_agent = _KwargsCapturingAgent(name="wrapped") + internal_collision = _KwargsCapturingAgent(name="internal_collision") + outer_internal = _KwargsCapturingAgent(name="outer_internal") + child = SequentialBuilder( + participants=[ + AgentExecutor(wrapped_agent, id="wrapped"), + AgentExecutor(internal_collision, id="internal"), + ] + ).build() + dispatcher = _FanOut(id="dispatcher") + nested_executor = WorkflowExecutor(child, id="wrapped") + outer_internal_executor = AgentExecutor(outer_internal, id="internal") + parent = ( + WorkflowBuilder(start_executor=dispatcher) + .add_fan_out_edges(dispatcher, [nested_executor, outer_internal_executor]) + .build() + ) + invocation_kwargs = { + "wrapped": {"wrapper_only": True}, + "internal": {"outer_only": True}, + } + + if kwargs_channel == "function_invocation_kwargs": + await parent.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await parent.run("test", client_kwargs=invocation_kwargs) + + assert wrapped_agent.captured_kwargs[0].get(kwargs_channel) == {"wrapper_only": True} + assert internal_collision.captured_kwargs[0].get(kwargs_channel) is None + assert outer_internal.captured_kwargs[0].get(kwargs_channel) == {"outer_only": True} + + +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_transparent_subworkflow_preserves_agent_name_alias(kwargs_channel: str) -> None: + """A wrapper key routes through a unique child agent-name alias.""" + from agent_framework import AgentExecutor + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + wrapped_agent = _KwargsCapturingAgent(name="wrapped") + child = SequentialBuilder( + participants=[ + AgentExecutor(wrapped_agent, id="wrapped_executor"), + ] + ).build() + parent = SequentialBuilder( + participants=[ + WorkflowExecutor(child, id="wrapped"), + ] + ).build() + invocation_kwargs = {"wrapped": {"wrapper_only": True}} + + if kwargs_channel == "function_invocation_kwargs": + await parent.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await parent.run("test", client_kwargs=invocation_kwargs) + + assert wrapped_agent.captured_kwargs[0].get(kwargs_channel) == {"wrapper_only": True} + + @pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) async def test_legacy_mixed_kwargs_route_through_subworkflow(kwargs_channel: str) -> None: """A child graph reclassifies preserved legacy mixed input against its executor IDs.""" @@ -1486,6 +1696,82 @@ async def test_per_executor_function_invocation_kwargs_routes_to_correct_agent() assert agent2.captured_kwargs[0].get("function_invocation_kwargs") == {"tool_param": "value_for_agent2"} +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_explicit_agent_name_routes_to_executor_id(kwargs_channel: str) -> None: + """An explicitly executor-scoped agent name aliases its executor ID.""" + from agent_framework import AgentExecutor + + agent1 = _KwargsCapturingAgent(name="agent1") + agent2 = _KwargsCapturingAgent(name="agent2") + workflow = SequentialBuilder( + participants=[ + AgentExecutor(agent1, id="executor1"), + AgentExecutor(agent2, id="executor2"), + ] + ).build() + executor_kwargs = { + "agent1": {"api_key": "agent1-only"}, + "agent2": {"api_key": "agent2-only"}, + } + invocation_kwargs = WorkflowInvocationKwargs(executor_kwargs=executor_kwargs) + + if kwargs_channel == "function_invocation_kwargs": + await workflow.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await workflow.run("test", client_kwargs=invocation_kwargs) + + assert agent1.captured_kwargs[0].get(kwargs_channel) == {"api_key": "agent1-only"} + assert agent2.captured_kwargs[0].get(kwargs_channel) == {"api_key": "agent2-only"} + + +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_plain_agent_name_key_remains_global_with_explicit_executor_id(kwargs_channel: str) -> None: + """A plain application kwarg matching an agent name does not trigger executor routing.""" + from agent_framework import AgentExecutor + + agent1 = _KwargsCapturingAgent(name="agent1") + agent2 = _KwargsCapturingAgent(name="agent2") + workflow = SequentialBuilder( + participants=[ + AgentExecutor(agent1, id="executor1"), + AgentExecutor(agent2, id="executor2"), + ] + ).build() + invocation_kwargs = {"agent1": {"api_key": "shared"}} + + if kwargs_channel == "function_invocation_kwargs": + await workflow.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await workflow.run("test", client_kwargs=invocation_kwargs) + + assert agent1.captured_kwargs[0].get(kwargs_channel) == invocation_kwargs + assert agent2.captured_kwargs[0].get(kwargs_channel) == invocation_kwargs + + +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_explicit_global_agent_name_alias_targets_executor(kwargs_channel: str) -> None: + """An agent named __global__ can be explicitly targeted through a different executor ID.""" + from agent_framework import AgentExecutor + + global_agent = _KwargsCapturingAgent(name="__global__") + sibling = _KwargsCapturingAgent(name="sibling") + workflow = SequentialBuilder( + participants=[ + AgentExecutor(global_agent, id="global_agent"), + sibling, + ] + ).build() + invocation_kwargs = WorkflowInvocationKwargs(executor_kwargs={"__global__": {"targeted": True}}) + + if kwargs_channel == "function_invocation_kwargs": + await workflow.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await workflow.run("test", client_kwargs=invocation_kwargs) + + assert global_agent.captured_kwargs[0].get(kwargs_channel) == {"targeted": True} + assert sibling.captured_kwargs[0].get(kwargs_channel) == {} + + async def test_global_and_per_executor_function_invocation_kwargs_are_merged() -> None: """Global function kwargs are merged with executor-specific overrides.""" agent1 = _KwargsCapturingAgent(name="agent1")