From 2d826f3fb09e60e35a48555086295f5d8151f53d Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 12 Oct 2026 11:16:55 +0200 Subject: [PATCH 1/3] Python: Fix restored subworkflow sibling requests --- .../_workflows/_workflow_executor.py | 10 +- .../core/tests/workflow/test_sub_workflow.py | 110 ++++++++++++++++++ 2 files changed, 119 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 2fe0ea5c145..7b87c17ad3e 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -729,7 +729,8 @@ async def _handle_response( # The sub-workflow is the source of truth for what it is awaiting. Validate the response # against its pending requests and ignore anything unknown or already handled. pending_requests = await self.workflow._runner_context.get_pending_request_info_events() # pyright: ignore[reportPrivateUsage] - if request_id not in pending_requests: + previously_pending_request_ids = set(pending_requests) + if request_id not in previously_pending_request_ids: logger.warning( f"WorkflowExecutor {self.id} received a response for an unknown or already-handled " f"request_id: {request_id}. This response will be ignored." @@ -742,4 +743,11 @@ async def _handle_response( responses={request_id: response}, tools=ctx.get_runtime_tools(), ) + # A restored sub-workflow can re-emit sibling requests before consuming this response. + # They were already forwarded to the parent, so only propagate requests created by this run. + result[:] = [ + event + for event in result + if event.type != "request_info" or event.request_id not in previously_pending_request_ids + ] await self._process_workflow_result(result, ctx) diff --git a/python/packages/core/tests/workflow/test_sub_workflow.py b/python/packages/core/tests/workflow/test_sub_workflow.py index 7e541470121..27553a17e9e 100644 --- a/python/packages/core/tests/workflow/test_sub_workflow.py +++ b/python/packages/core/tests/workflow/test_sub_workflow.py @@ -11,6 +11,7 @@ from agent_framework import ( Executor, + FunctionExecutor, SubWorkflowRequestMessage, SubWorkflowResponseMessage, Workflow, @@ -663,6 +664,80 @@ def _build_checkpoint_test_workflow(storage: InMemoryCheckpointStorage) -> Workf ) +@dataclass +class SiblingCheckpointRequest: + """Request emitted by one branch of a checkpointed sub-workflow.""" + + field: str + data: dict[str, int] + + +class SiblingCheckpointRequestExecutor(Executor): + """Add one response value to the shared result.""" + + def __init__(self, field: str) -> None: + super().__init__(id=f"ask_{field}") + self._field = field + + @handler + async def ask(self, data: dict[str, int], ctx: WorkflowContext[dict[str, int]]) -> None: + await ctx.request_info(SiblingCheckpointRequest(field=self._field, data=data), int) + + @response_handler + async def answer( + self, + request: SiblingCheckpointRequest, + response: int, + ctx: WorkflowContext[dict[str, int]], + ) -> None: + await ctx.send_message({**request.data, request.field: response}) + + +async def _fork_checkpoint_sibling_requests( + data: dict[str, int], + ctx: WorkflowContext[dict[str, int]], +) -> None: + await ctx.send_message(data) + + +async def _join_checkpoint_sibling_responses( + parts: list[dict[str, int]], + ctx: WorkflowContext[dict[str, int], dict[str, int]], +) -> None: + await ctx.yield_output({key: value for part in parts for key, value in part.items()}) + + +async def _finish_checkpoint_sibling_responses( + data: dict[str, int], + ctx: WorkflowContext[dict[str, int], dict[str, int]], +) -> None: + await ctx.yield_output(data) + + +def _build_sibling_checkpoint_test_workflow() -> Workflow: + """Build a parent whose child fans out to two requests before a later request.""" + fork = FunctionExecutor(_fork_checkpoint_sibling_requests, id="fork") + request_a = SiblingCheckpointRequestExecutor("a") + request_b = SiblingCheckpointRequestExecutor("b") + join = FunctionExecutor(_join_checkpoint_sibling_responses, id="join") + sub_workflow = ( + WorkflowBuilder(start_executor=fork) + .add_fan_out_edges(fork, [request_a, request_b]) + .add_fan_in_edges([request_a, request_b], join) + .build() + ) + + sub_workflow_executor = WorkflowExecutor(sub_workflow, id="sub_workflow", propagate_request=True) + request_c = SiblingCheckpointRequestExecutor("c") + finish = FunctionExecutor(_finish_checkpoint_sibling_responses, id="finish") + return ( + WorkflowBuilder(start_executor=sub_workflow_executor) + .add_edge(sub_workflow_executor, request_c) + .add_edge(request_c, finish) + .build() + ) + + async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None: """Test that resuming a sub-workflow from checkpoint does not emit duplicate requests. @@ -711,6 +786,41 @@ async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None: assert request_events[0].data.prompt == "Second request" +async def test_sub_workflow_checkpoint_restore_clears_answered_sibling_requests() -> None: + """Restoring and answering sibling requests must not retain either request.""" + storage = InMemoryCheckpointStorage() + workflow = _build_sibling_checkpoint_test_workflow() + + result = await workflow.run({}, checkpoint_storage=storage) + pending = {event.request_id: event.data.field for event in result.get_request_info_events()} + assert sorted(pending.values()) == ["a", "b"] + + responses = {request_id: {"a": 1, "b": 2}[field] for request_id, field in pending.items()} + checkpoint_id = await workflow.resolve_pause_checkpoint_id(responses, checkpoint_storage=storage) + workflow = _build_sibling_checkpoint_test_workflow() + result = await workflow.run( + checkpoint_id=checkpoint_id, + checkpoint_storage=storage, + responses=responses, + ) + + pending = {event.request_id: event.data.field for event in result.get_request_info_events()} + assert list(pending.values()) == ["c"] + + responses = {request_id: 3 for request_id in pending} + checkpoint_id = await workflow.resolve_pause_checkpoint_id(responses, checkpoint_storage=storage) + workflow = _build_sibling_checkpoint_test_workflow() + result = await workflow.run( + checkpoint_id=checkpoint_id, + checkpoint_storage=storage, + responses=responses, + ) + + assert result.get_outputs() == [{"a": 1, "b": 2, "c": 3}] + assert result.get_request_info_events() == [] + assert result.get_final_state() == WorkflowRunState.IDLE + + async def test_sub_workflow_checkpoint_restore_preserves_sub_workflow_state() -> None: """Resuming a sub-workflow mid-progress must restore its internal executor state. From 910cf0d39efcee874bb7da23e4e4999d94fd7d53 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 12 Oct 2026 11:25:20 +0200 Subject: [PATCH 2/3] Python: Narrow restored request filtering --- .../agent_framework/_workflows/_workflow_executor.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 7b87c17ad3e..452867ff07d 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -729,8 +729,7 @@ async def _handle_response( # The sub-workflow is the source of truth for what it is awaiting. Validate the response # against its pending requests and ignore anything unknown or already handled. pending_requests = await self.workflow._runner_context.get_pending_request_info_events() # pyright: ignore[reportPrivateUsage] - previously_pending_request_ids = set(pending_requests) - if request_id not in previously_pending_request_ids: + if request_id not in pending_requests: logger.warning( f"WorkflowExecutor {self.id} received a response for an unknown or already-handled " f"request_id: {request_id}. This response will be ignored." @@ -743,11 +742,11 @@ async def _handle_response( responses={request_id: response}, tools=ctx.get_runtime_tools(), ) - # A restored sub-workflow can re-emit sibling requests before consuming this response. - # They were already forwarded to the parent, so only propagate requests created by this run. + # A restored sub-workflow can re-emit the exact pending event objects before consuming + # this response. Suppress those objects, but preserve new requests that reuse an ID. result[:] = [ event for event in result - if event.type != "request_info" or event.request_id not in previously_pending_request_ids + if event.type != "request_info" or event is not pending_requests.get(event.request_id) ] await self._process_workflow_result(result, ctx) From b382e746cff99671c4f6f0e9f315ee3b84d75c35 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Tue, 13 Oct 2026 13:24:09 +0200 Subject: [PATCH 3/3] Python: Clear restored requests during cancellation --- .../_workflows/_workflow_executor.py | 22 +++++++++++----- .../core/tests/workflow/test_sub_workflow.py | 26 +++++++++++++++++++ 2 files changed, 41 insertions(+), 7 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 452867ff07d..7313e91f322 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -567,6 +567,7 @@ async def handle_propagated_request_response( @override async def _cancel_pending_request(self, request_id: str, ctx: WorkflowContext[Any, Any]) -> None: """Propagate cancellation into the wrapped workflow.""" + pending_requests = await self.workflow._runner_context.get_pending_request_info_events() # pyright: ignore[reportPrivateUsage] fi_kwargs, ci_kwargs = self._get_child_invocation_kwargs(ctx, preserve_empty=True) result = await self.workflow.cancel_pending_requests( [request_id], @@ -574,6 +575,7 @@ async def _cancel_pending_request(self, request_id: str, ctx: WorkflowContext[An function_invocation_kwargs=fi_kwargs, client_kwargs=ci_kwargs, ) + self._remove_reemitted_pending_requests(result, pending_requests) await self._process_workflow_result(result, ctx) @override @@ -625,6 +627,18 @@ async def on_checkpoint_restore(self, state: dict[str, Any]) -> None: ) ) + @staticmethod + def _remove_reemitted_pending_requests( + result: WorkflowRunResult, + pending_requests: Mapping[str, WorkflowEvent[Any]], + ) -> None: + """Remove restored request event instances that the parent already received.""" + result[:] = [ + event + for event in result + if event.type != "request_info" or event is not pending_requests.get(event.request_id) + ] + async def _process_workflow_result( self, result: WorkflowRunResult, @@ -742,11 +756,5 @@ async def _handle_response( responses={request_id: response}, tools=ctx.get_runtime_tools(), ) - # A restored sub-workflow can re-emit the exact pending event objects before consuming - # this response. Suppress those objects, but preserve new requests that reuse an ID. - result[:] = [ - event - for event in result - if event.type != "request_info" or event is not pending_requests.get(event.request_id) - ] + self._remove_reemitted_pending_requests(result, pending_requests) await self._process_workflow_result(result, ctx) diff --git a/python/packages/core/tests/workflow/test_sub_workflow.py b/python/packages/core/tests/workflow/test_sub_workflow.py index 27553a17e9e..cfd144b816b 100644 --- a/python/packages/core/tests/workflow/test_sub_workflow.py +++ b/python/packages/core/tests/workflow/test_sub_workflow.py @@ -821,6 +821,32 @@ async def test_sub_workflow_checkpoint_restore_clears_answered_sibling_requests( assert result.get_final_state() == WorkflowRunState.IDLE +async def test_sub_workflow_checkpoint_restore_clears_cancelled_sibling_requests() -> None: + """Restoring and cancelling sibling requests must not retain either request.""" + storage = InMemoryCheckpointStorage() + workflow = _build_sibling_checkpoint_test_workflow() + + result = await workflow.run({}, checkpoint_storage=storage) + pending = {event.request_id: event.data.field for event in result.get_request_info_events()} + assert sorted(pending.values()) == ["a", "b"] + + checkpoint_id = await workflow.resolve_pause_checkpoint_id(pending, checkpoint_storage=storage) + workflow = _build_sibling_checkpoint_test_workflow() + result = await workflow.cancel_pending_requests( + pending, + checkpoint_id=checkpoint_id, + checkpoint_storage=storage, + ) + + assert result.get_request_info_events() == [] + assert result.get_final_state() == WorkflowRunState.IDLE + assert await workflow._runner_context.get_pending_request_info_events() == {} # pyright: ignore[reportPrivateUsage] + + checkpoints = await storage.list_checkpoints(workflow_name=workflow.name) + latest_checkpoint = max(checkpoints, key=lambda checkpoint: checkpoint.timestamp) + assert latest_checkpoint.pending_request_info_events == {} + + async def test_sub_workflow_checkpoint_restore_preserves_sub_workflow_state() -> None: """Resuming a sub-workflow mid-progress must restore its internal executor state.