From bc6d51b8a7bcbde00e5cd38e09dc429e542cec11 Mon Sep 17 00:00:00 2001 From: WhaleTech <304937387+ryo-whaletech@users.noreply.github.com> Date: Mon, 14 Sep 2026 21:00:51 +0900 Subject: [PATCH 1/2] Python: Fix workflow kwargs collision with __global__ executor IDs --- .../_workflows/_agent_executor.py | 11 +- .../_workflows/_agent_utils.py | 116 +++++-- .../core/agent_framework/_workflows/_const.py | 10 +- .../agent_framework/_workflows/_workflow.py | 62 +++- .../tests/workflow/test_agent_executor.py | 35 +- .../tests/workflow/test_workflow_kwargs.py | 302 +++++++++++++++++- .../_workflows/_executors_agents.py | 16 +- .../declarative/tests/test_graph_coverage.py | 1 + .../declarative/tests/test_graph_executors.py | 175 ++++++++++ .../_group_chat.py | 5 +- .../orchestrations/tests/test_group_chat.py | 26 ++ 11 files changed, 704 insertions(+), 55 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index cd69504c71d..6867bd73a05 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/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. diff --git a/python/packages/core/agent_framework/_workflows/_agent_utils.py b/python/packages/core/agent_framework/_workflows/_agent_utils.py index 08034123328..4d7277d6258 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_utils.py +++ b/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 diff --git a/python/packages/core/agent_framework/_workflows/_const.py b/python/packages/core/agent_framework/_workflows/_const.py index f01b45d8407..d939bd448cb 100644 --- a/python/packages/core/agent_framework/_workflows/_const.py +++ b/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__" diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 11d95fe5a79..84a43101c11 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/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 diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py index 9e124db1d9c..18d93cf37d6 100644 --- a/python/packages/core/tests/workflow/test_agent_executor.py +++ b/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).""" diff --git a/python/packages/core/tests/workflow/test_workflow_kwargs.py b/python/packages/core/tests/workflow/test_workflow_kwargs.py index d91d224c54f..15cea3adfad 100644 --- a/python/packages/core/tests/workflow/test_workflow_kwargs.py +++ b/python/packages/core/tests/workflow/test_workflow_kwargs.py @@ -20,7 +20,13 @@ WorkflowInvocationKwargs, WorkflowRunState, ) -from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY +from agent_framework._workflows._const import ( + GLOBAL_KWARGS_KEY, + RAW_CLIENT_KWARGS_KEY, + RAW_FUNCTION_INVOCATION_KWARGS_KEY, + RESOLVED_WORKFLOW_RUN_KWARGS_KEY, + WORKFLOW_RUN_KWARGS_KEY, +) from agent_framework.orchestrations import ( ConcurrentBuilder, GroupChatBuilder, @@ -864,6 +870,137 @@ async def test_runtime_tools_are_not_written_to_checkpoints(tmp_path) -> None: assert "tools" not in run_kwargs +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_collision_free_kwargs_survive_file_checkpoint_restore(tmp_path: Any, kwargs_channel: str) -> None: + """A real __global__ executor remains distinct after file serialization and restore.""" + from agent_framework_orchestrations._orchestration_request_info import AgentRequestInfoResponse + + from agent_framework import FileCheckpointStorage + + storage = FileCheckpointStorage( + tmp_path, + allowed_checkpoint_types=[ + "agent_framework_orchestrations._orchestration_request_info:AgentRequestInfoResponse" + ], + ) + + def build_workflow() -> tuple[Any, _KwargsCapturingAgent, _KwargsCapturingAgent]: + global_agent = _KwargsCapturingAgent(name="__global__") + sibling = _KwargsCapturingAgent(name="sibling") + workflow = ( + SequentialBuilder(participants=[global_agent, sibling], checkpoint_storage=storage) + .with_request_info(agents=[global_agent]) + .build() + ) + return workflow, global_agent, sibling + + workflow, global_agent, _ = build_workflow() + invocation_kwargs = WorkflowInvocationKwargs( + global_kwargs={"shared": "G", "overridden": "global"}, + executor_kwargs={"__global__": {"special": "A", "overridden": "specific"}}, + ) + if kwargs_channel == "function_invocation_kwargs": + paused = await workflow.run("test", function_invocation_kwargs=invocation_kwargs) + else: + paused = await workflow.run("test", client_kwargs=invocation_kwargs) + [request] = paused.get_request_info_events() + assert global_agent.captured_kwargs[0].get(kwargs_channel) == { + "shared": "G", + "special": "A", + "overridden": "specific", + } + + checkpoints = await storage.list_checkpoints(workflow_name=workflow.name) + checkpoint = max( + (item for item in checkpoints if item.pending_request_info_events), + key=lambda item: item.timestamp, + ) + resolved_state = checkpoint.state[RESOLVED_WORKFLOW_RUN_KWARGS_KEY] + assert resolved_state[kwargs_channel] == { + "global_kwargs": {"shared": "G", "overridden": "global"}, + "executor_kwargs": {"__global__": {"special": "A", "overridden": "specific"}}, + } + + resumed_workflow, _, resumed_sibling = build_workflow() + resumed = await resumed_workflow.run(checkpoint_id=checkpoint.checkpoint_id) + [resumed_request] = resumed.get_request_info_events() + assert resumed_request.request_id == request.request_id + await resumed_workflow.run( + responses={resumed_request.request_id: AgentRequestInfoResponse.approve()}, + ) + assert resumed_sibling.captured_kwargs[0].get(kwargs_channel) == { + "shared": "G", + "overridden": "global", + } + + +@pytest.mark.parametrize( + ("legacy_resolved", "expected"), + [ + ({GLOBAL_KWARGS_KEY: {"shared": "G"}}, {"shared": "G"}), + ({"sibling": {"specific": "B"}}, {"specific": "B"}), + ], + ids=["global", "per-executor"], +) +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_legacy_kwargs_file_checkpoint_remains_readable( + tmp_path: Any, + kwargs_channel: str, + legacy_resolved: dict[str, Any], + expected: dict[str, Any], +) -> None: + """Non-ambiguous legacy dict state remains readable after file restore.""" + from agent_framework_orchestrations._orchestration_request_info import AgentRequestInfoResponse + + from agent_framework import FileCheckpointStorage + + storage = FileCheckpointStorage( + tmp_path, + allowed_checkpoint_types=[ + "agent_framework_orchestrations._orchestration_request_info:AgentRequestInfoResponse" + ], + ) + + def build_workflow() -> tuple[Any, _KwargsCapturingAgent]: + first = _KwargsCapturingAgent(name="first") + sibling = _KwargsCapturingAgent(name="sibling") + workflow = ( + SequentialBuilder(participants=[first, sibling], checkpoint_storage=storage) + .with_request_info(agents=[first]) + .build() + ) + return workflow, sibling + + workflow, _ = build_workflow() + if kwargs_channel == "function_invocation_kwargs": + paused = await workflow.run("test", function_invocation_kwargs={"seed": True}) + else: + paused = await workflow.run("test", client_kwargs={"seed": True}) + [request] = paused.get_request_info_events() + checkpoints = await storage.list_checkpoints(workflow_name=workflow.name) + checkpoint = max( + (item for item in checkpoints if item.pending_request_info_events), + key=lambda item: item.timestamp, + ) + run_kwargs = checkpoint.state[WORKFLOW_RUN_KWARGS_KEY] + run_kwargs[kwargs_channel] = legacy_resolved + run_kwargs.pop( + RAW_FUNCTION_INVOCATION_KWARGS_KEY if kwargs_channel == "function_invocation_kwargs" else RAW_CLIENT_KWARGS_KEY, + None, + ) + checkpoint.state.pop(RESOLVED_WORKFLOW_RUN_KWARGS_KEY, None) + await storage.save(checkpoint) + + resumed_workflow, resumed_sibling = build_workflow() + resumed = await resumed_workflow.run(checkpoint_id=checkpoint.checkpoint_id) + [resumed_request] = resumed.get_request_info_events() + assert resumed_request.request_id == request.request_id + await resumed_workflow.run( + responses={resumed_request.request_id: AgentRequestInfoResponse.approve()}, + ) + assert resumed_sibling.captured_kwargs[0].get(kwargs_channel) == expected + + # endregion @@ -1100,6 +1237,65 @@ async def test_mixed_kwargs_route_through_subworkflow() -> None: } +@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.""" + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + inner1 = _KwargsCapturingAgent(name="inner1") + inner2 = _KwargsCapturingAgent(name="inner2") + child = SequentialBuilder(participants=[inner1, inner2]).build() + parent = SequentialBuilder(participants=[WorkflowExecutor(child, id="subworkflow")]).build() + invocation_kwargs = { + "__global__": {"shared": "G", "overridden": "global"}, + "inner1": {"specific": "A", "overridden": "specific"}, + } + + if kwargs_channel == "function_invocation_kwargs": + await parent.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await parent.run("test", client_kwargs=invocation_kwargs) + + actual = tuple(agent.captured_kwargs[0].get(kwargs_channel) for agent in (inner1, inner2)) + assert actual == ( + {"shared": "G", "specific": "A", "overridden": "specific"}, + {"shared": "G", "overridden": "global"}, + ) + + +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +@pytest.mark.parametrize("typed", [True, False], ids=["typed", "plain"]) +async def test_real_global_executor_collision_routes_through_subworkflow(kwargs_channel: str, typed: bool) -> None: + """A child workflow preserves a real __global__ executor namespace.""" + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + actual_global = _KwargsCapturingAgent(name="__global__") + sibling = _KwargsCapturingAgent(name="sibling") + child = SequentialBuilder(participants=[actual_global, sibling]).build() + parent = SequentialBuilder(participants=[WorkflowExecutor(child, id="subworkflow")]).build() + expected: tuple[dict[str, Any] | None, dict[str, Any] | None] + if typed: + invocation_kwargs: Mapping[str, Any] | WorkflowInvocationKwargs = WorkflowInvocationKwargs( + global_kwargs={"shared": "G", "overridden": "global"}, + executor_kwargs={"__global__": {"special": "A", "overridden": "specific"}}, + ) + expected = ( + {"shared": "G", "special": "A", "overridden": "specific"}, + {"shared": "G", "overridden": "global"}, + ) + else: + invocation_kwargs = {"__global__": {"special": "A"}} + expected = ({"special": "A"}, None) + + if kwargs_channel == "function_invocation_kwargs": + await parent.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await parent.run("test", client_kwargs=invocation_kwargs) + + actual = tuple(agent.captured_kwargs[0].get(kwargs_channel) for agent in (actual_global, sibling)) + assert actual == expected + + # endregion @@ -1131,6 +1327,110 @@ async def test_function_and_client_kwargs_together() -> None: assert agent.captured_kwargs[0].get("client_kwargs") == ci_kwargs +@pytest.mark.parametrize( + ("executor_ids", "invocation_kwargs", "expected"), + [ + ( + ("agent1", "sibling"), + { + "__global__": {"shared": "G", "overridden": "global"}, + "agent1": {"specific": "A", "overridden": "specific"}, + }, + ( + {"shared": "G", "specific": "A", "overridden": "specific"}, + {"shared": "G", "overridden": "global"}, + ), + ), + ( + ("agent1", "sibling"), + {"__global__": "tenant-a", "agent1": {"specific": "A"}}, + (None, None), + ), + ( + ("agent1", "sibling"), + {"__global__": {"tenant": "tenant-a"}, "other_option": 123}, + ( + {"__global__": {"tenant": "tenant-a"}, "other_option": 123}, + {"__global__": {"tenant": "tenant-a"}, "other_option": 123}, + ), + ), + ( + ("__global__", "agent1"), + {"__global__": {"global_executor": True}, "agent1": {"agent1": True}}, + ({"global_executor": True}, {"agent1": True}), + ), + ( + ("__global__", "agent1"), + {"__global__": "invalid", "agent1": {"specific": "A"}}, + (None, {"specific": "A"}), + ), + ( + ("__global__", "sibling"), + WorkflowInvocationKwargs( + global_kwargs={"shared": "G", "overridden": "global"}, + executor_kwargs={"__global__": {"special": "A", "overridden": "specific"}}, + ), + ( + {"shared": "G", "special": "A", "overridden": "specific"}, + {"shared": "G", "overridden": "global"}, + ), + ), + ( + ("agent1", "sibling"), + WorkflowInvocationKwargs( + global_kwargs={"shared": "G"}, + executor_kwargs={"__global__": {"special": "A"}}, + ), + ({"shared": "G"}, {"shared": "G"}), + ), + ( + ("__global__", "sibling"), + {"__global__": {"special": "A"}}, + ({"special": "A"}, None), + ), + ], + ids=[ + "legacy-mixed", + "legacy-mixed-invalid-global", + "mapping-valued-application-global", + "real-global-plus-target", + "malformed-real-global-plus-target", + "typed-global-collision", + "typed-unmatched-global-executor", + "plain-global-collision", + ], +) +@pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) +async def test_invocation_kwargs_compatibility_boundaries( + kwargs_channel: str, + executor_ids: tuple[str, str], + invocation_kwargs: Mapping[str, Any] | WorkflowInvocationKwargs, + expected: tuple[dict[str, Any] | None, dict[str, Any] | None], + caplog: "LogCaptureFixture", +) -> None: + """Preserve legacy boundaries while separating a real __global__ executor.""" + agents = [_KwargsCapturingAgent(name=executor_id) for executor_id in executor_ids] + workflow = SequentialBuilder(participants=agents).build() + + if kwargs_channel == "function_invocation_kwargs": + await workflow.run("test", function_invocation_kwargs=invocation_kwargs) + else: + await workflow.run("test", client_kwargs=invocation_kwargs) + + actual = tuple(agent.captured_kwargs[0].get(kwargs_channel) for agent in agents) + assert actual == expected + if isinstance(invocation_kwargs, Mapping) and invocation_kwargs.get("__global__") == "tenant-a": + assert sum("expected a dict for global kwargs" in record.message for record in caplog.records) == 2 + if ( + isinstance(invocation_kwargs, Mapping) + and executor_ids[0] == "__global__" + and invocation_kwargs.get("__global__") == "invalid" + ): + assert [record.message for record in caplog.records] == [ + "Executor __global__ expected a dict for its kwargs, but got . Ignoring." + ] + + async def test_global_function_invocation_kwargs_flow_to_all_agents() -> None: """Global function_invocation_kwargs should be received by all agents in a sequential workflow.""" agent1 = _KwargsCapturingAgent(name="agent1") diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py index 190393f6510..ee15de2dc5b 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py @@ -788,9 +788,19 @@ async def _invoke_agent_and_store_results( _validate_conversation_history(messages_for_agent, agent_name) # Retrieve kwargs passed to workflow.run() so they propagate to agent tools - from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY - - run_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {}) + from agent_framework._workflows import _agent_utils as workflow_agent_utils + from agent_framework._workflows import _const as workflow_const + + run_kwargs: dict[str, Any] = ctx.get_state(workflow_const.WORKFLOW_RUN_KWARGS_KEY, {}) + prepare_run_kwargs = getattr(workflow_agent_utils, "prepare_executor_run_kwargs", None) + resolved_state_key = getattr(workflow_const, "RESOLVED_WORKFLOW_RUN_KWARGS_KEY", None) + if callable(prepare_run_kwargs) and isinstance(resolved_state_key, str): + missing_resolved_state = object() + resolved_run_kwargs: Any = ctx.get_state(resolved_state_key, missing_resolved_state) + if resolved_run_kwargs is not missing_resolved_state: + if not isinstance(resolved_run_kwargs, dict): + raise TypeError("Resolved workflow run kwargs state must be a dict.") + run_kwargs = cast(Any, prepare_run_kwargs)(self.id, run_kwargs, resolved_run_kwargs) options: dict[str, Any] | None = None if run_kwargs: # Merge caller-provided options to avoid duplicate keyword argument diff --git a/python/packages/declarative/tests/test_graph_coverage.py b/python/packages/declarative/tests/test_graph_coverage.py index 90a72c0c3da..43662c82c2c 100644 --- a/python/packages/declarative/tests/test_graph_coverage.py +++ b/python/packages/declarative/tests/test_graph_coverage.py @@ -68,6 +68,7 @@ def mock_context(mock_state: MagicMock) -> MagicMock: """Create a mock workflow context.""" ctx = MagicMock() ctx.state = mock_state + ctx.get_state = MagicMock(side_effect=mock_state.get) ctx.send_message = AsyncMock() ctx.yield_output = AsyncMock() ctx.request_info = AsyncMock() diff --git a/python/packages/declarative/tests/test_graph_executors.py b/python/packages/declarative/tests/test_graph_executors.py index 81579740be0..4fb9e6e1c85 100644 --- a/python/packages/declarative/tests/test_graph_executors.py +++ b/python/packages/declarative/tests/test_graph_executors.py @@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from agent_framework import WorkflowInvocationKwargs try: import powerfx # noqa: F401 @@ -1746,6 +1747,178 @@ def test_eval_passes_through_plain_strings_without_engine(self): class TestExecutorKwargsForwarding: """Workflow run kwargs should be forwarded through executor agent invocations.""" + @pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) + @pytest.mark.parametrize( + ("executor_ids", "invocation_kwargs", "expected"), + [ + ( + ("agent1", "sibling"), + { + "__global__": {"shared": "G", "overridden": "global"}, + "agent1": {"specific": "A", "overridden": "specific"}, + }, + ( + {"shared": "G", "specific": "A", "overridden": "specific"}, + {"shared": "G", "overridden": "global"}, + ), + ), + ( + ("__global__", "sibling"), + WorkflowInvocationKwargs( + global_kwargs={"shared": "G", "overridden": "global"}, + executor_kwargs={"__global__": {"specific": "A", "overridden": "specific"}}, + ), + ( + {"shared": "G", "specific": "A", "overridden": "specific"}, + {"shared": "G", "overridden": "global"}, + ), + ), + ], + ids=["legacy-mixed", "global-executor-collision"], + ) + async def test_workflow_run_resolves_executor_kwargs( + self, + kwargs_channel: str, + executor_ids: tuple[str, str], + invocation_kwargs: dict[str, Any] | WorkflowInvocationKwargs, + expected: tuple[dict[str, Any], dict[str, Any]], + ) -> None: + """The public declarative path resolves legacy mixed and collision-free state.""" + agents: dict[str, Any] = {} + actions: list[dict[str, Any]] = [] + for index, executor_id in enumerate(executor_ids): + agent_name = f"agent_{index}" + response = MagicMock(text="response", messages=[], tool_calls=[]) + agent = MagicMock(name=agent_name) + agent.run = AsyncMock(return_value=response) + agents[agent_name] = agent + actions.append({"kind": "InvokeAzureAgent", "id": executor_id, "agent": agent_name, "input": "hello"}) + + workflow = DeclarativeWorkflowBuilder( + {"name": "kwargs_workflow", "actions": actions}, + agents=agents, + ).build() + if kwargs_channel == "function_invocation_kwargs": + await workflow.run(ActionTrigger(), function_invocation_kwargs=invocation_kwargs) + else: + await workflow.run(ActionTrigger(), client_kwargs=invocation_kwargs) + + for agent, expected_kwargs in zip(agents.values(), expected, strict=True): + call_kwargs = agent.run.call_args.kwargs + assert call_kwargs[kwargs_channel] == expected_kwargs + assert call_kwargs["options"]["additional_function_arguments"] == {kwargs_channel: expected_kwargs} + assert "_raw_function_invocation_kwargs" not in call_kwargs + assert "_raw_client_kwargs" not in call_kwargs + + @pytest.mark.parametrize("kwargs_channel", ["function_invocation_kwargs", "client_kwargs"]) + @pytest.mark.parametrize("legacy_checkpoint", [False, True], ids=["new-state", "legacy-state"]) + async def test_workflow_run_kwargs_survive_file_checkpoint_restore( + self, + tmp_path: Any, + kwargs_channel: str, + legacy_checkpoint: bool, + ) -> None: + """New checkpoints resolve kwargs while legacy checkpoints keep historical forwarding.""" + from agent_framework import FileCheckpointStorage + from agent_framework._workflows._const import ( + RESOLVED_WORKFLOW_RUN_KWARGS_KEY, + WORKFLOW_RUN_KWARGS_KEY, + ) + + from agent_framework_declarative._workflows._executors_external_input import ExternalInputResponse + + storage = FileCheckpointStorage( + tmp_path, + allowed_checkpoint_types=[ + "agent_framework_declarative._workflows._declarative_base:ActionComplete", + "agent_framework_declarative._workflows._declarative_base:ActionTrigger", + "agent_framework_declarative._workflows._executors_external_input:ExternalInputRequest", + "agent_framework_declarative._workflows._executors_external_input:ExternalInputResponse", + ], + ) + executor_id = "sibling" if legacy_checkpoint else "__global__" + + def build_workflow() -> tuple[Any, Any]: + response = MagicMock(text="response", messages=[], tool_calls=[]) + agent = MagicMock(name="checkpoint_agent") + agent.run = AsyncMock(return_value=response) + workflow = DeclarativeWorkflowBuilder( + { + "name": "declarative_kwargs_checkpoint", + "actions": [ + { + "kind": "RequestExternalInput", + "id": "pause", + "prompt": "Continue?", + "variable": "Local.answer", + }, + { + "kind": "InvokeAzureAgent", + "id": executor_id, + "agent": "checkpoint_agent", + "input": "hello", + }, + ], + }, + agents={"checkpoint_agent": agent}, + checkpoint_storage=storage, + ).build() + return workflow, agent + + if legacy_checkpoint: + invocation_kwargs: dict[str, Any] | WorkflowInvocationKwargs = { + "__global__": {"shared": "G"}, + "sibling": {"specific": "S"}, + } + else: + invocation_kwargs = WorkflowInvocationKwargs( + global_kwargs={"shared": "G"}, + executor_kwargs={"__global__": {"specific": "S"}}, + ) + + workflow, _ = build_workflow() + if kwargs_channel == "function_invocation_kwargs": + paused = await workflow.run(ActionTrigger(), function_invocation_kwargs=invocation_kwargs) + else: + paused = await workflow.run(ActionTrigger(), client_kwargs=invocation_kwargs) + [request] = paused.get_request_info_events() + checkpoints = await storage.list_checkpoints(workflow_name=workflow.name) + checkpoint = max( + (item for item in checkpoints if item.pending_request_info_events), + key=lambda item: item.timestamp, + ) + if legacy_checkpoint: + checkpoint.state.pop(RESOLVED_WORKFLOW_RUN_KWARGS_KEY, None) + await storage.save(checkpoint) + else: + assert checkpoint.state[RESOLVED_WORKFLOW_RUN_KWARGS_KEY][kwargs_channel] + assert isinstance(checkpoint.state[WORKFLOW_RUN_KWARGS_KEY][kwargs_channel], dict) + + resumed_workflow, resumed_agent = build_workflow() + resumed = await resumed_workflow.run(checkpoint_id=checkpoint.checkpoint_id) + [resumed_request] = resumed.get_request_info_events() + assert resumed_request.request_id == request.request_id + await resumed_workflow.run( + responses={resumed_request.request_id: ExternalInputResponse(user_input="yes")}, + ) + + call_kwargs = resumed_agent.run.call_args.kwargs + if legacy_checkpoint: + raw_key = ( + "_raw_function_invocation_kwargs" + if kwargs_channel == "function_invocation_kwargs" + else "_raw_client_kwargs" + ) + expected_run_kwargs = {kwargs_channel: invocation_kwargs, raw_key: invocation_kwargs} + assert call_kwargs[kwargs_channel] == invocation_kwargs + assert call_kwargs[raw_key] == invocation_kwargs + else: + expected_run_kwargs = {kwargs_channel: {"shared": "G", "specific": "S"}} + assert call_kwargs[kwargs_channel] == {"shared": "G", "specific": "S"} + assert "_raw_function_invocation_kwargs" not in call_kwargs + assert "_raw_client_kwargs" not in call_kwargs + assert call_kwargs["options"]["additional_function_arguments"] == expected_run_kwargs + @pytest.mark.asyncio async def test_invoke_agent_forwards_kwargs(self): """InvokeAzureAgentExecutor should forward run_kwargs to agent.run().""" @@ -1791,6 +1964,7 @@ def mock_set(key, value): mock_ctx.yield_output = AsyncMock() executor = InvokeAzureAgentExecutor.__new__(InvokeAzureAgentExecutor) + executor.id = "test_agent" executor._agents = {"test_agent": mock_agent} await executor._invoke_agent_and_store_results( @@ -1861,6 +2035,7 @@ def mock_set(key, value): mock_ctx.yield_output = AsyncMock() executor = InvokeAzureAgentExecutor.__new__(InvokeAzureAgentExecutor) + executor.id = "test_agent" executor._agents = {"test_agent": mock_agent} await executor._invoke_agent_and_store_results( diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py index 879e09236cf..9e364989e5a 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py @@ -34,7 +34,7 @@ from agent_framework._workflows._agent_executor import AgentExecutor, AgentExecutorRequest, AgentExecutorResponse from agent_framework._workflows._agent_utils import prepare_agent_run_args, resolve_agent_id from agent_framework._workflows._checkpoint import CheckpointStorage -from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY +from agent_framework._workflows._const import RESOLVED_WORKFLOW_RUN_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY from agent_framework._workflows._executor import Executor from agent_framework._workflows._workflow import Workflow from agent_framework._workflows._workflow_context import WorkflowContext @@ -496,7 +496,8 @@ async def _invoke_agent(self, ctx: WorkflowContext[Any, Any]) -> AgentOrchestrat those itself or it is the only agent in the group chat that does not see them. """ raw_run_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {}) - function_invocation_kwargs, client_kwargs = prepare_agent_run_args(self.id, raw_run_kwargs) + resolved_run_kwargs: Any = ctx.get_state(RESOLVED_WORKFLOW_RUN_KWARGS_KEY) + function_invocation_kwargs, client_kwargs = prepare_agent_run_args(self.id, raw_run_kwargs, resolved_run_kwargs) async def _invoke_agent_helper(conversation: list[Message]) -> AgentOrchestrationOutput: # Run the agent in non-streaming mode for simplicity diff --git a/python/packages/orchestrations/tests/test_group_chat.py b/python/packages/orchestrations/tests/test_group_chat.py index 92d600ba9ce..725666ac6cb 100644 --- a/python/packages/orchestrations/tests/test_group_chat.py +++ b/python/packages/orchestrations/tests/test_group_chat.py @@ -16,6 +16,7 @@ Content, Message, WorkflowEvent, + WorkflowInvocationKwargs, WorkflowRunState, ) from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage @@ -324,6 +325,31 @@ async def test_agent_manager_receives_workflow_run_kwargs() -> None: assert call.get("client_kwargs") == {"trace_id": "trace-abc"} +async def test_agent_manager_receives_resolved_global_and_specific_run_kwargs() -> None: + """The #8312 GroupChat path consumes collision-free state through the shared resolver.""" + manager = KwargsRecordingManagerAgent() + worker = StubAgent("agent", "worker response") + workflow = GroupChatBuilder(participants=[worker], orchestrator_agent=manager).build() + invocation_kwargs = WorkflowInvocationKwargs( + global_kwargs={"shared": "G", "overridden": "global"}, + executor_kwargs={"manager_agent": {"specific": "M", "overridden": "specific"}}, + ) + + async for _ in workflow.run( + "coordinate task", + stream=True, + function_invocation_kwargs=invocation_kwargs, + client_kwargs=invocation_kwargs, + ): + pass + + expected = {"shared": "G", "specific": "M", "overridden": "specific"} + assert manager.seen_kwargs + for call in manager.seen_kwargs: + assert call.get("function_invocation_kwargs") == expected + assert call.get("client_kwargs") == expected + + async def test_agent_manager_receives_no_run_kwargs_when_none_supplied() -> None: """With nothing declared on the run, the orchestrator is invoked with both kwargs as None. From 670092265db5ae7dae848f8f3de95a41f2a91e96 Mon Sep 17 00:00:00 2001 From: WhaleTech <304937387+ryo-whaletech@users.noreply.github.com> Date: Mon, 14 Sep 2026 22:45:30 +0900 Subject: [PATCH 2/2] Python: Scope workflow kwargs warning assertion --- .../packages/core/tests/workflow/test_workflow_kwargs.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/python/packages/core/tests/workflow/test_workflow_kwargs.py b/python/packages/core/tests/workflow/test_workflow_kwargs.py index 15cea3adfad..6685c70827a 100644 --- a/python/packages/core/tests/workflow/test_workflow_kwargs.py +++ b/python/packages/core/tests/workflow/test_workflow_kwargs.py @@ -1426,7 +1426,12 @@ async def test_invocation_kwargs_compatibility_boundaries( and executor_ids[0] == "__global__" and invocation_kwargs.get("__global__") == "invalid" ): - assert [record.message for record in caplog.records] == [ + matching_warnings = [ + record.message + for record in caplog.records + if record.levelname == "WARNING" and record.name == "agent_framework._workflows._agent_utils" + ] + assert matching_warnings == [ "Executor __global__ expected a dict for its kwargs, but got . Ignoring." ]