From 0599d8eb232913d06e4a8f69f603380e69168d34 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 16:13:38 +0900 Subject: [PATCH 01/10] fix(a2a): reject empty invocations explicitly Key decisions: - Keep A2A continuation authority explicit; durable session task state only enriches diagnostics. - Raise AgentInvalidRequestException with participant and available task context instead of inventing input. - Leave AgentExecutor and Group Chat production contracts unchanged. Files changed: - packages/a2a/agent_framework_a2a/_agent.py - packages/a2a/tests/test_a2a_agent.py - packages/a2a/tests/test_a2a_group_chat.py Notes for next iteration: - No blockers. INPUT_REQUIRED pause/resume remains a separate task. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../a2a/agent_framework_a2a/_agent.py | 12 +- python/packages/a2a/tests/test_a2a_agent.py | 37 +++- .../packages/a2a/tests/test_a2a_group_chat.py | 192 ++++++++++++++++++ 3 files changed, 238 insertions(+), 3 deletions(-) create mode 100644 python/packages/a2a/tests/test_a2a_group_chat.py diff --git a/python/packages/a2a/agent_framework_a2a/_agent.py b/python/packages/a2a/agent_framework_a2a/_agent.py index 19966f94069..12b9c44b647 100644 --- a/python/packages/a2a/agent_framework_a2a/_agent.py +++ b/python/packages/a2a/agent_framework_a2a/_agent.py @@ -44,6 +44,7 @@ ) from agent_framework._telemetry import mark_feature_used from agent_framework._types import AgentRunInputs +from agent_framework.exceptions import AgentInvalidRequestException from agent_framework.observability import AgentTelemetryLayer from google.protobuf.json_format import MessageToDict @@ -486,7 +487,16 @@ def run( ) else: if not normalized_messages: - raise ValueError("At least one message is required when starting a new task (no continuation_token).") + context_id, task_id, task_state = self._extract_a2a_session_state(session) + message = f"A2A agent {self.name!r} requires a real message or an explicit continuation token." + if context_id is not None or task_id is not None or task_state is not None: + task_context = [ + f"context_id={context_id!r}", + f"task_id={task_id!r}", + f"task_state={TaskState.Name(task_state) if task_state is not None else None}", + ] + message = f"{message} Session context: {', '.join(task_context)}." + raise AgentInvalidRequestException(message) a2a_message = self._prepare_message_for_a2a(normalized_messages[-1], session=session) request = SendMessageRequest(message=a2a_message) if background and not stream: diff --git a/python/packages/a2a/tests/test_a2a_agent.py b/python/packages/a2a/tests/test_a2a_agent.py index 1e180fcd12f..0cbfd9b3e7c 100644 --- a/python/packages/a2a/tests/test_a2a_agent.py +++ b/python/packages/a2a/tests/test_a2a_agent.py @@ -29,6 +29,7 @@ SessionContext, ) from agent_framework.a2a import A2AAgent +from agent_framework.exceptions import AgentInvalidRequestException from pytest import fixture, mark, raises, warns from agent_framework_a2a import A2AAgentSession, A2AContinuationToken, A2AServiceSessionId @@ -1234,17 +1235,49 @@ async def test_run_creates_session_for_providers_when_none_provided(mock_a2a_cli async def test_run_raises_when_no_messages_and_no_continuation_token( mock_a2a_client: MockA2AClient, messages: list[str] | None ) -> None: - """Test that run() raises ValueError when messages is None/empty and no continuation_token is provided.""" + """Empty A2A input requires a real message or an explicit continuation token.""" agent = A2AAgent( name="Test Agent", client=cast(Any, mock_a2a_client), http_client=None, ) - with raises(ValueError, match="At least one message is required"): + with raises( + AgentInvalidRequestException, + match="A2A agent 'Test Agent' requires a real message or an explicit continuation token", + ): await agent.run(messages) +async def test_empty_input_error_includes_session_task_context_without_resuming( + mock_a2a_client: MockA2AClient, +) -> None: + """Durable A2A task state explains an invalid call but never authorizes continuation.""" + agent = A2AAgent( + name="Remote specialist", + client=cast(Any, mock_a2a_client), + http_client=None, + ) + session = AgentSession( + service_session_id=A2AServiceSessionId( + context_id="customer-42", + task_id="task-17", + task_state=TaskState.TASK_STATE_COMPLETED, + ) + ) + + with raises( + AgentInvalidRequestException, + match=( + "A2A agent 'Remote specialist' requires a real message or an explicit continuation token. " + "Session context: context_id='customer-42', task_id='task-17', task_state=TASK_STATE_COMPLETED." + ), + ): + await agent.run([], session=session) + + assert mock_a2a_client.call_count == 0 + + async def test_run_with_continuation_token_does_not_require_messages(mock_a2a_client: MockA2AClient) -> None: """Test that run() does not raise when messages is None but a continuation_token is provided.""" task = Task( diff --git a/python/packages/a2a/tests/test_a2a_group_chat.py b/python/packages/a2a/tests/test_a2a_group_chat.py new file mode 100644 index 00000000000..b7f23ff22a1 --- /dev/null +++ b/python/packages/a2a/tests/test_a2a_group_chat.py @@ -0,0 +1,192 @@ +# Copyright (c) Microsoft. All rights reserved. + +from collections.abc import AsyncIterator, Sequence +from typing import Any, cast + +import pytest +from a2a.types import Artifact, Part, StreamResponse, Task, TaskState, TaskStatus +from agent_framework import ( + AgentResponse, + AgentResponseUpdate, + AgentSession, + BaseAgent, + Content, + Message, + ResponseStream, +) +from agent_framework.exceptions import AgentInvalidRequestException +from agent_framework.orchestrations import GroupChatBuilder, GroupChatState + +from agent_framework_a2a import A2AAgent + + +class RecordingA2AClient: + """Minimal A2A transport that records real remote invocations.""" + + def __init__(self) -> None: + self.call_count = 0 + + async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: + self.call_count += 1 + yield StreamResponse( + task=Task( + id=f"task-{self.call_count}", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED), + artifacts=[Artifact(artifact_id="answer", parts=[Part(text="Remote answer")])], + ) + ) + + +class TextlessAgent(BaseAgent): + """Participant whose response projects to no Group Chat messages.""" + + def __init__(self) -> None: + super().__init__(name="textless", description="Returns framework control content only") + self.call_count = 0 + + def run( # type: ignore[override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + stream: bool = False, + session: AgentSession | None = None, + **kwargs: Any, + ) -> Any: + self.call_count += 1 + function_call = Content.from_function_call(call_id="control-1", name="internal_control") + if stream: + + async def _stream() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=[function_call], role="assistant", author_name=self.name) + + return ResponseStream(_stream(), finalizer=AgentResponse.from_updates) + + async def _run() -> AgentResponse[Any]: + return AgentResponse(messages=[Message("assistant", [function_call], author_name=self.name)]) + + return _run() + + +class SessionBackedAgent(BaseAgent): + """Non-A2A participant that supports empty turns through its session.""" + + def __init__(self) -> None: + super().__init__(name="session-backed", description="Continues from session state") + self.invocations: list[Any] = [] + + def run( # type: ignore[override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + stream: bool = False, + session: AgentSession | None = None, + **kwargs: Any, + ) -> Any: + assert session is not None + self.invocations.append(messages) + turn = int(session.state.get("turn", 0)) + 1 + session.state["turn"] = turn + text = f"Session turn {turn}" + if stream: + + async def _stream() -> AsyncIterator[AgentResponseUpdate]: + yield AgentResponseUpdate( + contents=[Content.from_text(text=text)], + role="assistant", + author_name=self.name, + ) + + return ResponseStream(_stream(), finalizer=AgentResponse.from_updates) + + async def _run() -> AgentResponse[Any]: + return AgentResponse(messages=[Message("assistant", [text], author_name=self.name)]) + + return _run() + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_consecutive_a2a_selection_rejects_empty_invocation_without_remote_call(stream: bool) -> None: + """A consecutive A2A turn fails instead of inventing continuation input.""" + client = RecordingA2AClient() + remote = A2AAgent(name="remote", client=cast(Any, client), http_client=None) + + def select_remote(state: GroupChatState) -> str: + return "remote" + + workflow = GroupChatBuilder( + participants=[remote], + selection_func=select_remote, + max_rounds=2, + ).build() + + with pytest.raises( + AgentInvalidRequestException, + match="A2A agent 'remote' requires a real message or an explicit continuation token", + ): + if stream: + async for _ in workflow.run("Investigate the incident", stream=True): + pass + else: + await workflow.run("Investigate the incident", stream=False) + + assert client.call_count == 1 + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_a2a_reselection_after_textless_peer_rejects_empty_invocation(stream: bool) -> None: + """An intervening response with no projected messages cannot activate A2A.""" + client = RecordingA2AClient() + remote = A2AAgent(name="remote", client=cast(Any, client), http_client=None) + textless = TextlessAgent() + speakers = ["remote", "textless", "remote"] + + def select_in_sequence(state: GroupChatState) -> str: + return speakers[state.current_round] + + workflow = GroupChatBuilder( + participants=[remote, textless], + selection_func=select_in_sequence, + max_rounds=3, + ).build() + + with pytest.raises( + AgentInvalidRequestException, + match="A2A agent 'remote' requires a real message or an explicit continuation token", + ): + if stream: + async for _ in workflow.run("Investigate the incident", stream=True): + pass + else: + await workflow.run("Investigate the incident", stream=False) + + assert client.call_count == 1 + assert textless.call_count == 1 + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_consecutive_session_backed_participant_still_receives_empty_turn(stream: bool) -> None: + """Group Chat preserves valid empty-input behavior for non-A2A agents.""" + participant = SessionBackedAgent() + selection_count = 0 + + def select_participant(state: GroupChatState) -> str: + nonlocal selection_count + selection_count += 1 + return "session-backed" + + workflow = GroupChatBuilder( + participants=[participant], + selection_func=select_participant, + max_rounds=2, + ).build() + + if stream: + async for _ in workflow.run("Investigate the incident", stream=True): + pass + else: + await workflow.run("Investigate the incident", stream=False) + + assert selection_count == 2 + assert len(participant.invocations) == 2 + assert participant.invocations[1] == [] From e8fa3b7d9d5fe68762c0568690c202014a2dbc2f Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 16:25:33 +0900 Subject: [PATCH 02/10] fix(a2a): pause group chat for remote input Key decisions: - Translate A2A INPUT_REQUIRED task content into the existing Content user-input-request contract. - Use the remote task ID as stable request correlation for streamed and finalized responses. - Reuse AgentExecutor request handling so caller input resumes the same task without a workflow-specific A2A path. Files changed: - packages/a2a/agent_framework_a2a/_agent.py - packages/a2a/tests/test_a2a_agent.py - packages/a2a/tests/test_a2a_group_chat.py Notes for next iteration: - Checkpoint restoration of pending A2A input is now unblocked. - The local issue file could not be moved because repository issue files are restricted by content exclusion policy. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../a2a/agent_framework_a2a/_agent.py | 18 ++++ python/packages/a2a/tests/test_a2a_agent.py | 41 ++++++++- .../packages/a2a/tests/test_a2a_group_chat.py | 85 +++++++++++++++++++ 3 files changed, 143 insertions(+), 1 deletion(-) diff --git a/python/packages/a2a/agent_framework_a2a/_agent.py b/python/packages/a2a/agent_framework_a2a/_agent.py index 12b9c44b647..75b9440ca9e 100644 --- a/python/packages/a2a/agent_framework_a2a/_agent.py +++ b/python/packages/a2a/agent_framework_a2a/_agent.py @@ -730,6 +730,21 @@ def _updates_from_task( status = task.status task_metadata = MessageToDict(task.metadata) if task.metadata else None + if status.state == TaskState.TASK_STATE_INPUT_REQUIRED and status.HasField("message") and status.message.parts: + contents = self._parse_contents_from_a2a(status.message.parts) + if contents: + contents[0].id = task.id + contents[0].user_input_request = True + return [ + AgentResponseUpdate( + contents=contents, + role="assistant" if status.message.role == A2ARole.ROLE_AGENT else "user", + response_id=task.id, + additional_properties={"a2a_metadata": task_metadata} if task_metadata else None, + raw_representation=task, + ) + ] + if status.state in TERMINAL_TASK_STATES: task_messages = self._parse_messages_from_task(task) if task.artifacts and streamed_artifact_ids: @@ -834,6 +849,9 @@ def _updates_from_task_update_event( contents = self._parse_contents_from_a2a(message.parts) if not contents: return [] + if state == TaskState.TASK_STATE_INPUT_REQUIRED: + contents[0].id = update_event.task_id + contents[0].user_input_request = True msg_meta = MessageToDict(message.metadata) if message.metadata else {} event_meta = MessageToDict(update_event.metadata) if update_event.metadata else {} diff --git a/python/packages/a2a/tests/test_a2a_agent.py b/python/packages/a2a/tests/test_a2a_agent.py index 0cbfd9b3e7c..467ec4cc999 100644 --- a/python/packages/a2a/tests/test_a2a_agent.py +++ b/python/packages/a2a/tests/test_a2a_agent.py @@ -1510,13 +1510,21 @@ async def test_streaming_input_required_emits_content(a2a_agent: A2AAgent, mock_ ) mock_a2a_client.responses.append(StreamResponse(status_update=update_event)) + stream = a2a_agent.run("Hello", stream=True) updates: list[AgentResponseUpdate] = [] - async for update in a2a_agent.run("Hello", stream=True): + async for update in stream: updates.append(update) + response = await stream.get_final_response() assert len(updates) == 1 assert updates[0].text == "What is your name?" assert updates[0].message_id == "msg-input-req" + assert [(request.id, request.text) for request in updates[0].user_input_requests] == [ + ("task-status", "What is your name?") + ] + assert [(request.id, request.text) for request in response.user_input_requests] == [ + ("task-status", "What is your name?") + ] @mark.asyncio @@ -2142,6 +2150,37 @@ async def test_task_state_tracked_on_session(mock_a2a_client: MockA2AClient) -> assert session.task_state == TaskState.TASK_STATE_INPUT_REQUIRED +@mark.asyncio +@mark.parametrize("stream", [False, True]) +async def test_input_required_exposes_stable_user_input_request( + mock_a2a_client: MockA2AClient, + stream: bool, +) -> None: + """INPUT_REQUIRED preserves the remote question and task identity.""" + agent = A2AAgent(name="Test Agent", id="test-agent", client=cast(Any, mock_a2a_client), http_client=None) + mock_a2a_client.add_in_progress_task_response( + "task-input", + context_id="ctx-input", + state=TaskState.TASK_STATE_INPUT_REQUIRED, + text="What is your name?", + ) + + if stream: + response_stream = agent.run("Start", stream=True) + updates = [update async for update in response_stream] + response = await response_stream.get_final_response() + assert len(updates) == 1 + assert [(request.id, request.text) for request in updates[0].user_input_requests] == [ + ("task-input", "What is your name?") + ] + else: + response = await agent.run("Start") + + assert [(request.id, request.text) for request in response.user_input_requests] == [ + ("task-input", "What is your name?") + ] + + @mark.asyncio async def test_plain_agent_session_tracks_structured_service_session_id(mock_a2a_client: MockA2AClient) -> None: """Plain AgentSession should persist A2A continuation state in structured service_session_id.""" diff --git a/python/packages/a2a/tests/test_a2a_group_chat.py b/python/packages/a2a/tests/test_a2a_group_chat.py index b7f23ff22a1..d300f14a50e 100644 --- a/python/packages/a2a/tests/test_a2a_group_chat.py +++ b/python/packages/a2a/tests/test_a2a_group_chat.py @@ -5,6 +5,8 @@ import pytest from a2a.types import Artifact, Part, StreamResponse, Task, TaskState, TaskStatus +from a2a.types import Message as A2AMessage +from a2a.types import Role as A2ARole from agent_framework import ( AgentResponse, AgentResponseUpdate, @@ -38,6 +40,40 @@ async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: ) +class InputRequiredA2AClient: + """A2A transport that pauses once, then completes the same task.""" + + def __init__(self) -> None: + self.messages: list[Any] = [] + + async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: + self.messages.append(request.message) + if len(self.messages) == 1: + yield StreamResponse( + task=Task( + id="task-input", + context_id="group-chat-context", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id="input-request", + role=A2ARole.ROLE_AGENT, + parts=[Part(text="What is your name?")], + ), + ), + ) + ) + return + yield StreamResponse( + task=Task( + id="task-input", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED), + artifacts=[Artifact(artifact_id="answer", parts=[Part(text="Thanks, Alice")])], + ) + ) + + class TextlessAgent(BaseAgent): """Participant whose response projects to no Group Chat messages.""" @@ -190,3 +226,52 @@ def select_participant(state: GroupChatState) -> str: assert selection_count == 2 assert len(participant.invocations) == 2 assert participant.invocations[1] == [] + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_input_required_pauses_group_chat_and_resumes_same_task(stream: bool) -> None: + """Only caller input can resume the paused A2A participant.""" + client = InputRequiredA2AClient() + remote = A2AAgent(name="remote", client=cast(Any, client), http_client=None) + peer = SessionBackedAgent() + speakers = ["remote", "session-backed"] + + def select_in_sequence(state: GroupChatState) -> str: + return speakers[state.current_round] + + workflow = GroupChatBuilder( + participants=[remote, peer], + selection_func=select_in_sequence, + max_rounds=2, + ).build() + + if stream: + initial_stream = workflow.run("Start", stream=True) + async for _ in initial_stream: + pass + initial_result = await initial_stream.get_final_response() + else: + initial_result = await workflow.run("Start") + + requests = initial_result.get_request_info_events() + assert len(requests) == 1 + assert requests[0].data.id == "task-input" + assert requests[0].data.text == "What is your name?" + assert peer.invocations == [] + + caller_response = Content.from_text(text="Alice") + if stream: + resumed_stream = workflow.run( + stream=True, + responses={requests[0].request_id: caller_response}, + ) + async for _ in resumed_stream: + pass + await resumed_stream.get_final_response() + else: + await workflow.run(responses={requests[0].request_id: caller_response}) + + assert len(client.messages) == 2 + assert client.messages[1].task_id == "task-input" + assert client.messages[1].parts[0].text == "Alice" + assert len(peer.invocations) == 1 From a80481a4d6d3727b3a3111cbb587415f96de9ec3 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 16:33:19 +0900 Subject: [PATCH 03/10] fix(a2a): restore pending input from checkpoints Key decisions: - Keep normalized INPUT_REQUIRED content durable by excluding transport-only protobuf raw representations. - Restore through the existing AgentExecutor checkpoint and request-response path without a new schema or continuation API. - Cover file-backed restoration in streaming and non-streaming Group Chat runs, including unrelated-response rejection and exact task resumption. Files changed: - packages/a2a/agent_framework_a2a/_agent.py - packages/a2a/tests/test_a2a_group_chat.py Notes for next iteration: - The local issue file could not be moved because repository issue files are restricted by content exclusion policy. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../a2a/agent_framework_a2a/_agent.py | 2 + .../packages/a2a/tests/test_a2a_group_chat.py | 74 +++++++++++++++++++ 2 files changed, 76 insertions(+) diff --git a/python/packages/a2a/agent_framework_a2a/_agent.py b/python/packages/a2a/agent_framework_a2a/_agent.py index 75b9440ca9e..14d0de6a10e 100644 --- a/python/packages/a2a/agent_framework_a2a/_agent.py +++ b/python/packages/a2a/agent_framework_a2a/_agent.py @@ -735,6 +735,7 @@ def _updates_from_task( if contents: contents[0].id = task.id contents[0].user_input_request = True + contents[0].raw_representation = None return [ AgentResponseUpdate( contents=contents, @@ -852,6 +853,7 @@ def _updates_from_task_update_event( if state == TaskState.TASK_STATE_INPUT_REQUIRED: contents[0].id = update_event.task_id contents[0].user_input_request = True + contents[0].raw_representation = None msg_meta = MessageToDict(message.metadata) if message.metadata else {} event_meta = MessageToDict(update_event.metadata) if update_event.metadata else {} diff --git a/python/packages/a2a/tests/test_a2a_group_chat.py b/python/packages/a2a/tests/test_a2a_group_chat.py index d300f14a50e..522c4b75069 100644 --- a/python/packages/a2a/tests/test_a2a_group_chat.py +++ b/python/packages/a2a/tests/test_a2a_group_chat.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. from collections.abc import AsyncIterator, Sequence +from pathlib import Path from typing import Any, cast import pytest @@ -13,6 +14,7 @@ AgentSession, BaseAgent, Content, + FileCheckpointStorage, Message, ResponseStream, ) @@ -275,3 +277,75 @@ def select_in_sequence(state: GroupChatState) -> str: assert client.messages[1].task_id == "task-input" assert client.messages[1].parts[0].text == "Alice" assert len(peer.invocations) == 1 + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_input_required_survives_group_chat_checkpoint_restoration(stream: bool, tmp_path: Path) -> None: + """Restoration preserves caller authority and the original remote task.""" + client = InputRequiredA2AClient() + storage = FileCheckpointStorage(tmp_path) + peers: list[SessionBackedAgent] = [] + + def select_in_sequence(state: GroupChatState) -> str: + return ["remote", "session-backed"][state.current_round] + + def build_workflow() -> Any: + peer = SessionBackedAgent() + peers.append(peer) + return GroupChatBuilder( + participants=[ + A2AAgent(name="remote", client=cast(Any, client), http_client=None), + peer, + ], + selection_func=select_in_sequence, + max_rounds=2, + checkpoint_storage=storage, + ).build() + + workflow = build_workflow() + if stream: + initial_stream = workflow.run("Start", stream=True) + async for _ in initial_stream: + pass + initial_result = await initial_stream.get_final_response() + else: + initial_result = await workflow.run("Start") + + [request] = initial_result.get_request_info_events() + assert request.data.id == "task-input" + checkpoints = await storage.list_checkpoints(workflow_name=workflow.name) + checkpoint = next( + checkpoint for checkpoint in checkpoints if request.request_id in checkpoint.pending_request_info_events + ) + assert checkpoint.pending_request_info_events[request.request_id].data.id == "task-input" + + restored = build_workflow() + with pytest.raises(ValueError, match="unknown request ID"): + await restored.run( + checkpoint_id=checkpoint.checkpoint_id, + responses={"unrelated-request": Content.from_text(text="peer message")}, + ) + assert len(client.messages) == 1 + assert all(peer.invocations == [] for peer in peers) + + caller_response = Content.from_text(text="Alice") + if stream: + resumed_stream = restored.run( + checkpoint_id=checkpoint.checkpoint_id, + stream=True, + responses={request.request_id: caller_response}, + ) + async for _ in resumed_stream: + pass + resumed_result = await resumed_stream.get_final_response() + else: + resumed_result = await restored.run( + checkpoint_id=checkpoint.checkpoint_id, + responses={request.request_id: caller_response}, + ) + + assert resumed_result.get_request_info_events() == [] + assert len(client.messages) == 2 + assert client.messages[1].task_id == "task-input" + assert client.messages[1].parts[0].text == "Alice" + assert sum(len(peer.invocations) for peer in peers) == 1 From e2864e650ecb510c915dc85aa0a6ad2a6b64db1b Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 16:37:40 +0900 Subject: [PATCH 04/10] test(handoff): lock textless target context Key decisions: - Exercise the built Handoff workflow in streaming and non-streaming modes instead of bypassing routing, sessions, or termination. - Keep the slice test-only because current production already carries the initial task to a textless handoff target without synthetic user input. - Revisit the source to verify its handoff function call retains a matching result and user-turn termination sees only caller messages. Files changed: - packages/orchestrations/tests/test_handoff.py Notes for next iteration: - No production defect was reproduced. - The local issue file could not be moved because repository issue files are restricted by content exclusion policy. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../orchestrations/tests/test_handoff.py | 132 ++++++++++++++++++ 1 file changed, 132 insertions(+) diff --git a/python/packages/orchestrations/tests/test_handoff.py b/python/packages/orchestrations/tests/test_handoff.py index 21964ebddb4..767467fa316 100644 --- a/python/packages/orchestrations/tests/test_handoff.py +++ b/python/packages/orchestrations/tests/test_handoff.py @@ -577,6 +577,138 @@ async def _get() -> ChatResponse: assert requests[-1].source_executor_id == triage.name +@pytest.mark.parametrize("stream", [False, True]) +async def test_textless_handoff_preserves_target_context_without_synthetic_user_turn(stream: bool) -> None: + """Textless handoffs route accumulated context without inventing user input.""" + + class TextlessHandoffClient(FunctionInvocationLayer[Any], ChatMiddlewareLayer[Any], BaseChatClient[Any]): + def __init__(self, name: str, handoff_sequence: list[str | None]) -> None: + ChatMiddlewareLayer.__init__(self) + FunctionInvocationLayer.__init__(self) + BaseChatClient.__init__(self) + self._name = name + self._handoff_sequence = handoff_sequence + self.received_messages: list[list[Message]] = [] + + def _inner_get_response( + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + del options + del kwargs + + self.received_messages.append(list(messages)) + call_index = len(self.received_messages) - 1 + handoff_to = self._handoff_sequence[call_index] + if handoff_to is None: + contents = [Content.from_text(text=f"{self._name} complete")] + else: + contents = [ + Content.from_function_call( + call_id=f"{self._name}-handoff-{call_index}", + name=get_handoff_tool_name(handoff_to), + arguments={}, + ) + ] + + if stream: + + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate(contents=contents, role="assistant", finish_reason="stop") + + return ResponseStream(_stream(), finalizer=lambda updates: ChatResponse.from_updates(updates)) + + async def _get() -> ChatResponse: + return ChatResponse( + messages=[Message(role="assistant", contents=contents)], + response_id=f"{self._name}-{call_index}", + ) + + return _get() + + initial_task = "Investigate order 1234." + source_client = TextlessHandoffClient("source", ["target", None]) + target_client = TextlessHandoffClient("target", ["source"]) + source = Agent( + id="source", + name="source", + client=source_client, + require_per_service_call_history_persistence=True, + ) + target = Agent( + id="target", + name="target", + client=target_client, + require_per_service_call_history_persistence=True, + ) + observed_user_turns: list[list[str]] = [] + + def terminate_after_second_real_user_turn(conversation: list[Message]) -> bool: + user_turns = [message.text or "" for message in conversation if message.role == "user"] + observed_user_turns.append(user_turns) + return len(user_turns) >= 2 + + workflow = ( + HandoffBuilder( + participants=_as_handoff_agents(source, target), + termination_condition=terminate_after_second_real_user_turn, + ) + .with_start_agent(_as_handoff_agent(source)) + .build() + ) + + if stream: + events = await _drain(workflow.run(initial_task, stream=True)) + final_state = [event.state for event in events if event.type == "status"][-1] + else: + result = await workflow.run(initial_task) + events = list(result) + final_state = result.get_final_state() + + assert len(target_client.received_messages) == 1 + assert [message.text for message in target_client.received_messages[0] if message.role == "user"] == [initial_task] + assert len(source_client.received_messages) == 2 + + revisited_source_messages = source_client.received_messages[1] + source_call_ids = { + content.call_id + for message in revisited_source_messages + for content in message.contents + if content.type == "function_call" + } + source_result_ids = { + content.call_id + for message in revisited_source_messages + for content in message.contents + if content.type == "function_result" + } + assert "source-handoff-0" in source_call_ids + assert "source-handoff-0" in source_result_ids + + all_received_messages = [ + message + for invocation in [*source_client.received_messages, *target_client.received_messages] + for message in invocation + ] + assert all("continue the conversation" not in (message.text or "").lower() for message in all_received_messages) + assert observed_user_turns + assert all(user_turns == [initial_task] for user_turns in observed_user_turns) + + handoffs = [event.data for event in events if event.type == "handoff_sent"] + assert handoffs == [ + HandoffSentEvent(source=source.name, target=target.name), + HandoffSentEvent(source=target.name, target=source.name), + ] + requests = [event for event in events if event.type == "request_info"] + assert len(requests) == 1 + assert requests[0].source_executor_id == source.name + assert final_state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS + + async def test_handoff_resume_preserves_approved_tool_output_for_stateless_runs() -> None: """Approved calls must keep function_call/function_result pairs for later replays.""" submit_call_id = "call_submit_refund_approved" From 12a3c75e53a73e59ea447f6538d93128bef64a26 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 18:18:56 +0900 Subject: [PATCH 05/10] test(handoff): use resolved IDs in event assertions --- python/packages/orchestrations/tests/test_handoff.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/python/packages/orchestrations/tests/test_handoff.py b/python/packages/orchestrations/tests/test_handoff.py index 767467fa316..ee4b62c5708 100644 --- a/python/packages/orchestrations/tests/test_handoff.py +++ b/python/packages/orchestrations/tests/test_handoff.py @@ -700,12 +700,12 @@ def terminate_after_second_real_user_turn(conversation: list[Message]) -> bool: handoffs = [event.data for event in events if event.type == "handoff_sent"] assert handoffs == [ - HandoffSentEvent(source=source.name, target=target.name), - HandoffSentEvent(source=target.name, target=source.name), + HandoffSentEvent(source=resolve_agent_id(source), target=resolve_agent_id(target)), + HandoffSentEvent(source=resolve_agent_id(target), target=resolve_agent_id(source)), ] requests = [event for event in events if event.type == "request_info"] assert len(requests) == 1 - assert requests[0].source_executor_id == source.name + assert requests[0].source_executor_id == resolve_agent_id(source) assert final_state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS From 3d232be0236afc6ab91a10c2fc3b0937cd0489e9 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 19:12:52 +0900 Subject: [PATCH 06/10] fix(workflows): preserve A2A input request semantics --- .../a2a/agent_framework_a2a/_agent.py | 91 ++++-- python/packages/a2a/tests/test_a2a_agent.py | 262 +++++++++++++++++- .../packages/a2a/tests/test_a2a_group_chat.py | 128 ++++++++- .../core/agent_framework/_workflows/_agent.py | 8 +- .../_workflows/_agent_executor.py | 16 +- 5 files changed, 458 insertions(+), 47 deletions(-) diff --git a/python/packages/a2a/agent_framework_a2a/_agent.py b/python/packages/a2a/agent_framework_a2a/_agent.py index 14d0de6a10e..78d105b3779 100644 --- a/python/packages/a2a/agent_framework_a2a/_agent.py +++ b/python/packages/a2a/agent_framework_a2a/_agent.py @@ -631,7 +631,7 @@ async def _map_a2a_stream( if status_event.context_id: last_context_id = status_event.context_id last_task_state = status_event.status.state - updates = self._updates_from_task_update_event(status_event) + updates = self._updates_from_task_update_event(status_event, background=background) is_terminal = status_event.status.state in TERMINAL_TASK_STATES is_input_required = status_event.status.state == TaskState.TASK_STATE_INPUT_REQUIRED if emit_intermediate: @@ -709,6 +709,27 @@ async def _map_a2a_stream( # Task helpers # ------------------------------------------------------------------ + def _user_input_request_id(self, task_id: str) -> str: + """Return a stable workflow request ID scoped to this A2A participant.""" + scoped_task_id = f"{len(self.id)}:{self.id}{task_id}" + return f"a2a-input-{uuid.uuid5(uuid.NAMESPACE_URL, scoped_task_id)}" + + def _input_required_request(self, task_id: str, message: A2AMessage | None) -> Content: + """Normalize an A2A input requirement into one durable caller request.""" + contents = self._parse_contents_from_a2a(message.parts) if message is not None else [] + prompt_parts = [ + value for content in contents if (value := content.text or (content.uri if content.type == "uri" else None)) + ] + request = Content.from_text( + text="\n".join(prompt_parts) or "Remote A2A task requires input.", + additional_properties=( + {"a2a_input_required_message": MessageToDict(message)} if message is not None else None + ), + ) + request.id = self._user_input_request_id(task_id) + request.user_input_request = True + return request + def _updates_from_task( self, task: Task, @@ -730,21 +751,18 @@ def _updates_from_task( status = task.status task_metadata = MessageToDict(task.metadata) if task.metadata else None - if status.state == TaskState.TASK_STATE_INPUT_REQUIRED and status.HasField("message") and status.message.parts: - contents = self._parse_contents_from_a2a(status.message.parts) - if contents: - contents[0].id = task.id - contents[0].user_input_request = True - contents[0].raw_representation = None - return [ - AgentResponseUpdate( - contents=contents, - role="assistant" if status.message.role == A2ARole.ROLE_AGENT else "user", - response_id=task.id, - additional_properties={"a2a_metadata": task_metadata} if task_metadata else None, - raw_representation=task, - ) - ] + if status.state == TaskState.TASK_STATE_INPUT_REQUIRED: + message = status.message if status.HasField("message") and status.message.parts else None + return [ + AgentResponseUpdate( + contents=[self._input_required_request(task.id, message)], + role="assistant" if message is None or message.role == A2ARole.ROLE_AGENT else "user", + response_id=task.id, + continuation_token=self._build_continuation_token(task) if background else None, + additional_properties={"a2a_metadata": task_metadata} if task_metadata else None, + raw_representation=task, + ) + ] if status.state in TERMINAL_TASK_STATES: task_messages = self._parse_messages_from_task(task) @@ -815,7 +833,10 @@ def _updates_from_task( return [] def _updates_from_task_update_event( - self, update_event: TaskStatusUpdateEvent | TaskArtifactUpdateEvent + self, + update_event: TaskStatusUpdateEvent | TaskArtifactUpdateEvent, + *, + background: bool = False, ) -> list[AgentResponseUpdate]: """Convert A2A task update events into streaming AgentResponseUpdates.""" if isinstance(update_event, TaskArtifactUpdateEvent): @@ -839,22 +860,43 @@ def _updates_from_task_update_event( if not isinstance(update_event, TaskStatusUpdateEvent): return [] + state = update_event.status.state + continuation_token = ( + A2AContinuationToken(task_id=update_event.task_id, context_id=update_event.context_id) + if background and state in IN_PROGRESS_TASK_STATES + else None + ) + if state == TaskState.TASK_STATE_INPUT_REQUIRED: + message = ( + update_event.status.message + if update_event.status.HasField("message") and update_event.status.message.parts + else None + ) + message_meta = MessageToDict(message.metadata) if message is not None and message.metadata else {} + event_meta = MessageToDict(update_event.metadata) if update_event.metadata else {} + merged_metadata = {**message_meta, **event_meta} or None + return [ + AgentResponseUpdate( + contents=[self._input_required_request(update_event.task_id, message)], + role="assistant" if message is None or message.role == A2ARole.ROLE_AGENT else "user", + response_id=update_event.task_id, + message_id=message.message_id if message is not None else None, + continuation_token=continuation_token, + additional_properties={"a2a_metadata": merged_metadata} if merged_metadata else None, + raw_representation=update_event, + ) + ] + if not update_event.status.HasField("message") or not update_event.status.message.parts: return [] - state = update_event.status.state - if state not in TERMINAL_TASK_STATES and state != TaskState.TASK_STATE_INPUT_REQUIRED: + if state not in TERMINAL_TASK_STATES: return [] message = update_event.status.message contents = self._parse_contents_from_a2a(message.parts) if not contents: return [] - if state == TaskState.TASK_STATE_INPUT_REQUIRED: - contents[0].id = update_event.task_id - contents[0].user_input_request = True - contents[0].raw_representation = None - msg_meta = MessageToDict(message.metadata) if message.metadata else {} event_meta = MessageToDict(update_event.metadata) if update_event.metadata else {} merged_metadata = {**msg_meta, **event_meta} or None @@ -865,6 +907,7 @@ def _updates_from_task_update_event( role="assistant" if message.role == A2ARole.ROLE_AGENT else "user", response_id=update_event.task_id, message_id=message.message_id, + continuation_token=continuation_token, additional_properties={"a2a_metadata": merged_metadata} if merged_metadata else None, raw_representation=update_event, ) diff --git a/python/packages/a2a/tests/test_a2a_agent.py b/python/packages/a2a/tests/test_a2a_agent.py index 467ec4cc999..8ff147f98aa 100644 --- a/python/packages/a2a/tests/test_a2a_agent.py +++ b/python/packages/a2a/tests/test_a2a_agent.py @@ -838,11 +838,35 @@ async def test_input_required_task_emits_continuation_token( response = await a2a_agent.run("Need input", background=True) + [request] = response.user_input_requests + assert request.text == "Remote A2A task requires input." assert response.continuation_token is not None token = cast(dict[str, Any], response.continuation_token) assert token["task_id"] == "task-input" +async def test_background_input_required_preserves_request_and_continuation_token( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """Background callers retain both caller input and polling contracts.""" + mock_a2a_client.add_in_progress_task_response( + "task-input-prompt", + context_id="ctx-input", + state=TaskState.TASK_STATE_INPUT_REQUIRED, + text="Approve deployment?", + ) + + response = await a2a_agent.run("Need input", background=True) + + [request] = response.user_input_requests + assert request.text == "Approve deployment?" + assert response.continuation_token == { + "task_id": "task-input-prompt", + "context_id": "ctx-input", + } + + async def test_working_task_no_token_without_background(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None: """Test that background=False (default) does not emit continuation tokens for in-progress tasks.""" mock_a2a_client.add_in_progress_task_response("task-fg", context_id="ctx-fg", state=TaskState.TASK_STATE_WORKING) @@ -1029,6 +1053,34 @@ async def test_poll_task_in_progress(a2a_agent: A2AAgent, mock_a2a_client: MockA assert response_token["task_id"] == "task-poll" +async def test_poll_task_input_required_preserves_request_and_continuation_token( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """Polling preserves both caller input and later resubscription paths.""" + mock_a2a_client.get_task_response = Task( + id="task-poll-input", + context_id="ctx-poll-input", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id="poll-input-request", + role=A2ARole.ROLE_AGENT, + parts=[Part(text="Approve deployment?")], + ), + ), + ) + + response = await a2a_agent.poll_task(A2AContinuationToken(task_id="task-poll-input", context_id="ctx-poll-input")) + + [request] = response.user_input_requests + assert request.text == "Approve deployment?" + assert response.continuation_token == { + "task_id": "task-poll-input", + "context_id": "ctx-poll-input", + } + + async def test_poll_task_completed(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None: """Test poll_task returns result with no continuation token when task is complete.""" status = TaskStatus(state=TaskState.TASK_STATE_COMPLETED, message=None) @@ -1519,12 +1571,122 @@ async def test_streaming_input_required_emits_content(a2a_agent: A2AAgent, mock_ assert len(updates) == 1 assert updates[0].text == "What is your name?" assert updates[0].message_id == "msg-input-req" - assert [(request.id, request.text) for request in updates[0].user_input_requests] == [ - ("task-status", "What is your name?") - ] - assert [(request.id, request.text) for request in response.user_input_requests] == [ - ("task-status", "What is your name?") - ] + [update_request] = updates[0].user_input_requests + [response_request] = response.user_input_requests + assert update_request.id == response_request.id + assert update_request.id != "task-status" + assert update_request.text == "What is your name?" + + +async def test_streaming_background_input_required_preserves_request_and_token( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """Streaming background status retains caller and resubscription contracts.""" + update_event = TaskStatusUpdateEvent( + task_id="task-status-background", + context_id="ctx-status", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id="msg-input-req", + role=A2ARole.ROLE_AGENT, + parts=[Part(text="Approve deployment?")], + ), + ), + ) + mock_a2a_client.responses.append(StreamResponse(status_update=update_event)) + + stream = a2a_agent.run("Hello", stream=True, background=True) + updates = [update async for update in stream] + + assert len(updates) == 1 + [request] = updates[0].user_input_requests + assert request.text == "Approve deployment?" + assert updates[0].continuation_token == { + "task_id": "task-status-background", + "context_id": "ctx-status", + } + + +async def test_streaming_input_required_without_message_emits_generic_request( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """A message-less INPUT_REQUIRED status update still pauses for the caller.""" + update_event = TaskStatusUpdateEvent( + task_id="task-status-no-message", + context_id="ctx-status", + status=TaskStatus(state=TaskState.TASK_STATE_INPUT_REQUIRED), + ) + mock_a2a_client.responses.append(StreamResponse(status_update=update_event)) + + stream = a2a_agent.run("Hello", stream=True) + updates = [update async for update in stream] + response = await stream.get_final_response() + + assert len(updates) == 1 + [request] = response.user_input_requests + assert request.text == "Remote A2A task requires input." + assert request.id != "task-status-no-message" + + +async def test_streaming_input_required_multipart_prompt_remains_one_complete_request( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """All status prompt parts survive stream finalization as one request.""" + update_event = TaskStatusUpdateEvent( + task_id="task-status-multipart", + context_id="ctx-status", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id="msg-input-req", + role=A2ARole.ROLE_AGENT, + parts=[Part(text="Choose a deployment."), Part(text="Options: blue or green.")], + ), + ), + ) + mock_a2a_client.responses.append(StreamResponse(status_update=update_event)) + + stream = a2a_agent.run("Hello", stream=True) + updates = [update async for update in stream] + response = await stream.get_final_response() + + assert len(updates) == 1 + [request] = response.user_input_requests + assert request.text == "Choose a deployment.\nOptions: blue or green." + + +async def test_streaming_input_required_prompt_preserves_non_text_parts( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """Streaming caller requests retain links and durable remote message data.""" + update_event = TaskStatusUpdateEvent( + task_id="task-status-file", + context_id="ctx-status", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id="msg-input-req", + role=A2ARole.ROLE_AGENT, + parts=[Part(text="Review the attached document."), Part(url="hosted://files/report.pdf")], + ), + ), + ) + mock_a2a_client.responses.append(StreamResponse(status_update=update_event)) + + stream = a2a_agent.run("Hello", stream=True) + updates = [update async for update in stream] + response = await stream.get_final_response() + + assert len(updates) == 1 + [request] = response.user_input_requests + assert request.text == "Review the attached document.\nhosted://files/report.pdf" + serialized_message = request.additional_properties["a2a_input_required_message"] + assert serialized_message["parts"][1]["url"] == "hosted://files/report.pdf" @mark.asyncio @@ -2165,20 +2327,96 @@ async def test_input_required_exposes_stable_user_input_request( text="What is your name?", ) + updates: list[AgentResponseUpdate] = [] if stream: response_stream = agent.run("Start", stream=True) updates = [update async for update in response_stream] response = await response_stream.get_final_response() assert len(updates) == 1 - assert [(request.id, request.text) for request in updates[0].user_input_requests] == [ - ("task-input", "What is your name?") - ] else: response = await agent.run("Start") - assert [(request.id, request.text) for request in response.user_input_requests] == [ - ("task-input", "What is your name?") - ] + [request] = response.user_input_requests + assert request.id != "task-input" + assert request.text == "What is your name?" + if stream: + assert updates[0].user_input_requests[0].id == request.id + + +async def test_input_required_without_message_exposes_generic_user_input_request( + mock_a2a_client: MockA2AClient, +) -> None: + """INPUT_REQUIRED pauses even when the remote omits its optional prompt.""" + agent = A2AAgent(name="Test Agent", id="test-agent", client=cast(Any, mock_a2a_client), http_client=None) + mock_a2a_client.add_in_progress_task_response( + "task-input-no-message", + context_id="ctx-input", + state=TaskState.TASK_STATE_INPUT_REQUIRED, + ) + + response = await agent.run("Start") + + [request] = response.user_input_requests + assert request.text == "Remote A2A task requires input." + assert request.id != "task-input-no-message" + + +async def test_input_required_multipart_prompt_remains_one_complete_request( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """All prompt parts survive non-streaming response finalization.""" + mock_a2a_client.responses.append( + StreamResponse( + task=Task( + id="task-input-multipart", + context_id="ctx-input", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id="input-request", + role=A2ARole.ROLE_AGENT, + parts=[Part(text="Choose a deployment."), Part(text="Options: blue or green.")], + ), + ), + ) + ) + ) + + response = await a2a_agent.run("Start") + + [request] = response.user_input_requests + assert request.text == "Choose a deployment.\nOptions: blue or green." + + +async def test_input_required_prompt_preserves_non_text_parts( + a2a_agent: A2AAgent, + mock_a2a_client: MockA2AClient, +) -> None: + """Caller-visible requests retain links and durable remote message data.""" + mock_a2a_client.responses.append( + StreamResponse( + task=Task( + id="task-input-file", + context_id="ctx-input", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id="input-request", + role=A2ARole.ROLE_AGENT, + parts=[Part(text="Review the attached document."), Part(url="hosted://files/report.pdf")], + ), + ), + ) + ) + ) + + response = await a2a_agent.run("Start") + + [request] = response.user_input_requests + assert request.text == "Review the attached document.\nhosted://files/report.pdf" + serialized_message = request.additional_properties["a2a_input_required_message"] + assert serialized_message["parts"][1]["url"] == "hosted://files/report.pdf" @mark.asyncio diff --git a/python/packages/a2a/tests/test_a2a_group_chat.py b/python/packages/a2a/tests/test_a2a_group_chat.py index 522c4b75069..055635e1936 100644 --- a/python/packages/a2a/tests/test_a2a_group_chat.py +++ b/python/packages/a2a/tests/test_a2a_group_chat.py @@ -19,7 +19,7 @@ ResponseStream, ) from agent_framework.exceptions import AgentInvalidRequestException -from agent_framework.orchestrations import GroupChatBuilder, GroupChatState +from agent_framework.orchestrations import ConcurrentBuilder, GroupChatBuilder, GroupChatState from agent_framework_a2a import A2AAgent @@ -76,6 +76,30 @@ async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: ) +class MessageLessInputRequiredA2AClient(InputRequiredA2AClient): + """A2A transport whose first input request omits the optional prompt.""" + + async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: + self.messages.append(request.message) + if len(self.messages) == 1: + yield StreamResponse( + task=Task( + id="task-input-no-message", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_INPUT_REQUIRED), + ) + ) + return + yield StreamResponse( + task=Task( + id="task-input-no-message", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED), + artifacts=[Artifact(artifact_id="answer", parts=[Part(text="Thanks, Alice")])], + ) + ) + + class TextlessAgent(BaseAgent): """Participant whose response projects to no Group Chat messages.""" @@ -143,6 +167,94 @@ async def _run() -> AgentResponse[Any]: return _run() +async def test_concurrent_a2a_requests_with_same_remote_task_id_remain_isolated() -> None: + """Caller responses remain scoped to the participant that requested them.""" + first_client = InputRequiredA2AClient() + second_client = InputRequiredA2AClient() + workflow = ConcurrentBuilder( + participants=[ + A2AAgent(name="first-remote", client=cast(Any, first_client), http_client=None), + A2AAgent(name="second-remote", client=cast(Any, second_client), http_client=None), + ] + ).build() + + initial_result = await workflow.run("Start") + + requests = {event.source_executor_id: event for event in initial_result.get_request_info_events()} + assert set(requests) == {"first-remote", "second-remote"} + assert requests["first-remote"].request_id != requests["second-remote"].request_id + + await workflow.run( + responses={requests["first-remote"].request_id: "Alice"}, + ) + assert len(first_client.messages) == 2 + assert len(second_client.messages) == 1 + + await workflow.run( + responses={requests["second-remote"].request_id: "Bob"}, + ) + assert len(second_client.messages) == 2 + + +async def test_input_required_round_trips_through_group_chat_as_agent() -> None: + """A workflow agent exposes and accepts the generic request-info envelope.""" + client = InputRequiredA2AClient() + workflow = GroupChatBuilder( + participants=[A2AAgent(name="remote", client=cast(Any, client), http_client=None)], + selection_func=lambda state: "remote", + max_rounds=1, + ).build() + workflow_agent = workflow.as_agent(name="group-chat-agent") + + initial_response = await workflow_agent.run("Start") + + request_calls = [ + content + for message in initial_response.messages + for content in message.contents + if content.type == "function_call" and content.name == "request_info" + ] + [request_call] = request_calls + assert request_call.call_id is not None + + final_response = await workflow_agent.run( + Message( + role="tool", + contents=[Content.from_function_result(call_id=request_call.call_id, result="Alice")], + ) + ) + + assert final_response.user_input_requests == [] + assert len(client.messages) == 2 + assert client.messages[1].task_id == "task-input" + assert client.messages[1].parts[0].text == "Alice" + + +async def test_message_less_input_required_pauses_and_resumes_group_chat() -> None: + """Remote caller authority survives an omitted A2A prompt message.""" + client = MessageLessInputRequiredA2AClient() + remote = A2AAgent(name="remote", client=cast(Any, client), http_client=None) + peer = SessionBackedAgent() + workflow = GroupChatBuilder( + participants=[remote, peer], + selection_func=lambda state: ["remote", "session-backed"][state.current_round], + max_rounds=2, + ).build() + + initial_result = await workflow.run("Start") + + [request] = initial_result.get_request_info_events() + assert request.data.text == "Remote A2A task requires input." + assert peer.invocations == [] + + await workflow.run(responses={request.request_id: "Alice"}) + + assert len(client.messages) == 2 + assert client.messages[1].task_id == "task-input-no-message" + assert client.messages[1].parts[0].text == "Alice" + assert len(peer.invocations) == 1 + + @pytest.mark.parametrize("stream", [False, True]) async def test_consecutive_a2a_selection_rejects_empty_invocation_without_remote_call(stream: bool) -> None: """A consecutive A2A turn fails instead of inventing continuation input.""" @@ -257,11 +369,12 @@ def select_in_sequence(state: GroupChatState) -> str: requests = initial_result.get_request_info_events() assert len(requests) == 1 - assert requests[0].data.id == "task-input" + assert requests[0].data.id == requests[0].request_id + assert requests[0].request_id != "task-input" assert requests[0].data.text == "What is your name?" assert peer.invocations == [] - caller_response = Content.from_text(text="Alice") + caller_response = "Alice" if stream: resumed_stream = workflow.run( stream=True, @@ -312,23 +425,24 @@ def build_workflow() -> Any: initial_result = await workflow.run("Start") [request] = initial_result.get_request_info_events() - assert request.data.id == "task-input" + assert request.data.id == request.request_id + assert request.request_id != "task-input" checkpoints = await storage.list_checkpoints(workflow_name=workflow.name) checkpoint = next( checkpoint for checkpoint in checkpoints if request.request_id in checkpoint.pending_request_info_events ) - assert checkpoint.pending_request_info_events[request.request_id].data.id == "task-input" + assert checkpoint.pending_request_info_events[request.request_id].data.id == request.request_id restored = build_workflow() with pytest.raises(ValueError, match="unknown request ID"): await restored.run( checkpoint_id=checkpoint.checkpoint_id, - responses={"unrelated-request": Content.from_text(text="peer message")}, + responses={"unrelated-request": "peer message"}, ) assert len(client.messages) == 1 assert all(peer.invocations == [] for peer in peers) - caller_response = Content.from_text(text="Alice") + caller_response = "Alice" if stream: resumed_stream = restored.run( checkpoint_id=checkpoint.checkpoint_id, diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py index 01bd70d300d..a41c32dd28a 100644 --- a/python/packages/core/agent_framework/_workflows/_agent.py +++ b/python/packages/core/agent_framework/_workflows/_agent.py @@ -714,8 +714,12 @@ def _process_request_info_event( Note: If the event data is already a FunctionApprovalRequestContent, it will be returned as-is. """ - if isinstance(event.data, Content) and event.data.user_input_request: - # Return the event data as-is if it's already a properly formed FunctionApprovalRequestContent + if ( + isinstance(event.data, Content) + and event.data.user_input_request + and event.data.type in {"function_approval_request", "function_call"} + ): + # Preserve request contents that already have a matching response envelope. return event.data request_id = event.request_id diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index b7787736fa5..e3aebb403bc 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -316,6 +316,16 @@ async def handle_user_input_response( self._pending_responses_to_agent.clear() await self._run_agent_and_emit(ctx) + @response_handler + async def handle_text_user_input_response( + self, + original_request: Content, + response: str, + ctx: WorkflowContext[AgentExecutorResponse, AgentResponse | AgentResponseUpdate], + ) -> None: + """Normalize a text response before resuming agent execution.""" + await self.handle_user_input_response(original_request, Content.from_text(text=response), ctx) + @override async def on_checkpoint_save(self) -> dict[str, Any]: """Capture current executor state for checkpointing. @@ -446,7 +456,8 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR ) for user_input_request in response.user_input_requests: self._pending_agent_requests[user_input_request.id] = user_input_request # type: ignore[index] - await ctx.request_info(user_input_request, Content, request_id=user_input_request.id) + response_type = str if user_input_request.type == "text" else Content + await ctx.request_info(user_input_request, response_type, request_id=user_input_request.id) return None # Only yield output if the response is complete and not waiting for user input. @@ -540,7 +551,8 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp if user_input_requests: for user_input_request in user_input_requests: self._pending_agent_requests[user_input_request.id] = user_input_request # type: ignore[index] - await ctx.request_info(user_input_request, Content, request_id=user_input_request.id) + response_type = str if user_input_request.type == "text" else Content + await ctx.request_info(user_input_request, response_type, request_id=user_input_request.id) return None return response From dff091d1eceb84fedbc6046a887a74937b098d43 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 19:47:51 +0900 Subject: [PATCH 07/10] fix(workflows): preserve input request correlation --- .../a2a/agent_framework_a2a/_agent.py | 11 ++- .../packages/a2a/tests/test_a2a_group_chat.py | 97 ++++++++++++++++++- .../_workflows/_agent_executor.py | 16 +-- .../agent_framework/_workflows/_workflow.py | 4 +- 4 files changed, 107 insertions(+), 21 deletions(-) diff --git a/python/packages/a2a/agent_framework_a2a/_agent.py b/python/packages/a2a/agent_framework_a2a/_agent.py index 78d105b3779..5d0abc44527 100644 --- a/python/packages/a2a/agent_framework_a2a/_agent.py +++ b/python/packages/a2a/agent_framework_a2a/_agent.py @@ -709,10 +709,10 @@ async def _map_a2a_stream( # Task helpers # ------------------------------------------------------------------ - def _user_input_request_id(self, task_id: str) -> str: - """Return a stable workflow request ID scoped to this A2A participant.""" - scoped_task_id = f"{len(self.id)}:{self.id}{task_id}" - return f"a2a-input-{uuid.uuid5(uuid.NAMESPACE_URL, scoped_task_id)}" + def _user_input_request_id(self, task_id: str, occurrence_id: str) -> str: + """Return a workflow request ID scoped to one remote prompt occurrence.""" + scoped_occurrence = f"{len(self.id)}:{self.id}{len(task_id)}:{task_id}{occurrence_id}" + return f"a2a-input-{uuid.uuid5(uuid.NAMESPACE_URL, scoped_occurrence)}" def _input_required_request(self, task_id: str, message: A2AMessage | None) -> Content: """Normalize an A2A input requirement into one durable caller request.""" @@ -726,7 +726,8 @@ def _input_required_request(self, task_id: str, message: A2AMessage | None) -> C {"a2a_input_required_message": MessageToDict(message)} if message is not None else None ), ) - request.id = self._user_input_request_id(task_id) + occurrence_id = message.message_id if message is not None and message.message_id else str(uuid.uuid4()) + request.id = self._user_input_request_id(task_id, occurrence_id) request.user_input_request = True return request diff --git a/python/packages/a2a/tests/test_a2a_group_chat.py b/python/packages/a2a/tests/test_a2a_group_chat.py index 055635e1936..25038490126 100644 --- a/python/packages/a2a/tests/test_a2a_group_chat.py +++ b/python/packages/a2a/tests/test_a2a_group_chat.py @@ -100,6 +100,38 @@ async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: ) +class RepeatedInputRequiredA2AClient(InputRequiredA2AClient): + """A2A transport that requests caller input twice for the same task.""" + + async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: + self.messages.append(request.message) + if len(self.messages) <= 2: + prompt_number = len(self.messages) + yield StreamResponse( + task=Task( + id="task-input", + context_id="group-chat-context", + status=TaskStatus( + state=TaskState.TASK_STATE_INPUT_REQUIRED, + message=A2AMessage( + message_id=f"input-request-{prompt_number}", + role=A2ARole.ROLE_AGENT, + parts=[Part(text=f"Question {prompt_number}?")], + ), + ), + ) + ) + return + yield StreamResponse( + task=Task( + id="task-input", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED), + artifacts=[Artifact(artifact_id="answer", parts=[Part(text="Complete")])], + ) + ) + + class TextlessAgent(BaseAgent): """Participant whose response projects to no Group Chat messages.""" @@ -255,6 +287,69 @@ async def test_message_less_input_required_pauses_and_resumes_group_chat() -> No assert len(peer.invocations) == 1 +async def test_repeated_input_requests_for_same_remote_task_reject_stale_responses() -> None: + """Each remote prompt occurrence has independent workflow correlation.""" + client = RepeatedInputRequiredA2AClient() + remote = A2AAgent(name="remote", client=cast(Any, client), http_client=None) + peer = SessionBackedAgent() + workflow = GroupChatBuilder( + participants=[remote, peer], + selection_func=lambda state: ["remote", "session-backed"][state.current_round], + max_rounds=2, + ).build() + + first_result = await workflow.run("Start") + [first_request] = first_result.get_request_info_events() + + second_result = await workflow.run( + responses={first_request.request_id: Content.from_text(text="First answer")}, + ) + [second_request] = second_result.get_request_info_events() + + assert second_request.request_id != first_request.request_id + with pytest.raises(ValueError, match="unknown request ID"): + await workflow.run( + responses={first_request.request_id: Content.from_text(text="Stale first answer")}, + ) + assert len(client.messages) == 2 + + await workflow.run( + responses={second_request.request_id: Content.from_text(text="Second answer")}, + ) + assert len(client.messages) == 3 + assert client.messages[2].task_id == "task-input" + assert client.messages[2].parts[0].text == "Second answer" + assert len(peer.invocations) == 1 + + +async def test_input_required_accepts_structured_content_response() -> None: + """Caller responses preserve structured content supported by A2A.""" + client = InputRequiredA2AClient() + remote = A2AAgent(name="remote", client=cast(Any, client), http_client=None) + peer = SessionBackedAgent() + workflow = GroupChatBuilder( + participants=[remote, peer], + selection_func=lambda state: ["remote", "session-backed"][state.current_round], + max_rounds=2, + ).build() + + initial_result = await workflow.run("Start") + [request] = initial_result.get_request_info_events() + + await workflow.run( + responses={ + request.request_id: Content.from_uri( + "https://example.com/answer.pdf", + media_type="application/pdf", + ) + }, + ) + + assert len(client.messages) == 2 + assert client.messages[1].task_id == "task-input" + assert client.messages[1].parts[0].url == "https://example.com/answer.pdf" + + @pytest.mark.parametrize("stream", [False, True]) async def test_consecutive_a2a_selection_rejects_empty_invocation_without_remote_call(stream: bool) -> None: """A consecutive A2A turn fails instead of inventing continuation input.""" @@ -374,7 +469,7 @@ def select_in_sequence(state: GroupChatState) -> str: assert requests[0].data.text == "What is your name?" assert peer.invocations == [] - caller_response = "Alice" + caller_response = Content.from_text(text="Alice") if stream: resumed_stream = workflow.run( stream=True, diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index e3aebb403bc..b7787736fa5 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -316,16 +316,6 @@ async def handle_user_input_response( self._pending_responses_to_agent.clear() await self._run_agent_and_emit(ctx) - @response_handler - async def handle_text_user_input_response( - self, - original_request: Content, - response: str, - ctx: WorkflowContext[AgentExecutorResponse, AgentResponse | AgentResponseUpdate], - ) -> None: - """Normalize a text response before resuming agent execution.""" - await self.handle_user_input_response(original_request, Content.from_text(text=response), ctx) - @override async def on_checkpoint_save(self) -> dict[str, Any]: """Capture current executor state for checkpointing. @@ -456,8 +446,7 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR ) for user_input_request in response.user_input_requests: self._pending_agent_requests[user_input_request.id] = user_input_request # type: ignore[index] - response_type = str if user_input_request.type == "text" else Content - await ctx.request_info(user_input_request, response_type, request_id=user_input_request.id) + await ctx.request_info(user_input_request, Content, request_id=user_input_request.id) return None # Only yield output if the response is complete and not waiting for user input. @@ -551,8 +540,7 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp if user_input_requests: for user_input_request in user_input_requests: self._pending_agent_requests[user_input_request.id] = user_input_request # type: ignore[index] - response_type = str if user_input_request.type == "text" else Content - await ctx.request_info(user_input_request, response_type, request_id=user_input_request.id) + await ctx.request_info(user_input_request, Content, request_id=user_input_request.id) return None return response diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 77060b93a91..04f3f9aa875 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -17,7 +17,7 @@ from typing import TYPE_CHECKING, Any, Literal, overload from .._sessions import ContextProvider -from .._types import ResponseStream +from .._types import Content, ResponseStream from ..exceptions import WorkflowException from ..observability import OtelAttr, capture_exception, create_workflow_span from ._checkpoint import CheckpointStorage @@ -1020,6 +1020,8 @@ async def _send_responses_internal(self, responses: Mapping[str, Any]) -> None: if request_id not in pending_requests: raise ValueError(f"Response provided for unknown request ID: {request_id}") pending_request = pending_requests[request_id] + if pending_request.response_type is Content and isinstance(response, str): + response = Content.from_text(text=response) # Try to coerce raw values (e.g., dicts from JSON) to the expected type response = try_coerce_to_type(response, pending_request.response_type) if not is_instance_of(response, pending_request.response_type): From 3e770c0b1e8dcc4d905dad6c1c2627a5c555ba45 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Wed, 19 Aug 2026 20:13:19 +0900 Subject: [PATCH 08/10] fix(a2a): deduplicate message-less input requests --- .../a2a/agent_framework_a2a/_agent.py | 60 ++++++++++++-- .../packages/a2a/tests/test_a2a_group_chat.py | 78 ++++++++++++++++++- 2 files changed, 132 insertions(+), 6 deletions(-) diff --git a/python/packages/a2a/agent_framework_a2a/_agent.py b/python/packages/a2a/agent_framework_a2a/_agent.py index 5d0abc44527..3d5ac41e86b 100644 --- a/python/packages/a2a/agent_framework_a2a/_agent.py +++ b/python/packages/a2a/agent_framework_a2a/_agent.py @@ -485,6 +485,7 @@ def run( a2a_stream: AsyncIterable[A2AStreamItem] = self.client.subscribe( SubscribeToTaskRequest(id=continuation_token["task_id"]) ) + input_request_occurrence_id = continuation_token["task_id"] else: if not normalized_messages: context_id, task_id, task_state = self._extract_a2a_session_state(session) @@ -498,6 +499,7 @@ def run( message = f"{message} Session context: {', '.join(task_context)}." raise AgentInvalidRequestException(message) a2a_message = self._prepare_message_for_a2a(normalized_messages[-1], session=session) + input_request_occurrence_id = a2a_message.message_id request = SendMessageRequest(message=a2a_message) if background and not stream: # return_immediately only applies to non-streaming (message/send) @@ -520,6 +522,7 @@ def run( a2a_stream, background=background, emit_intermediate=stream, + input_request_occurrence_id=input_request_occurrence_id, session=provider_session, session_context=session_context, ), @@ -535,6 +538,7 @@ async def _map_a2a_stream( *, background: bool = False, emit_intermediate: bool = False, + input_request_occurrence_id: str, session: AgentSession | None = None, session_context: SessionContext | None = None, ) -> AsyncIterable[AgentResponseUpdate]: @@ -551,6 +555,7 @@ async def _map_a2a_stream( carry message content are yielded to the caller. Typically set for streaming callers so non-streaming consumers only receive terminal task outputs. + input_request_occurrence_id: Stable identity for message-less input requests observed in this run. session: The agent session for context providers. session_context: The session context for context providers. """ @@ -573,6 +578,7 @@ async def _map_a2a_stream( ) all_updates: list[AgentResponseUpdate] = [] + seen_user_input_request_ids: set[str] = set() streamed_artifact_ids_by_task: dict[str, set[str]] = {} last_task_id: str | None = None last_context_id: str | None = None @@ -610,8 +616,10 @@ async def _map_a2a_stream( task, background=background, emit_intermediate=emit_intermediate, + input_request_occurrence_id=input_request_occurrence_id, streamed_artifact_ids=streamed_artifact_ids_by_task.get(task.id), ) + updates = self._deduplicate_user_input_request_updates(updates, seen_user_input_request_ids) if task.status.state in TERMINAL_TASK_STATES: streamed_artifact_ids_by_task.pop(task.id, None) # If the terminal Task has no content, flush accumulated updates @@ -631,7 +639,12 @@ async def _map_a2a_stream( if status_event.context_id: last_context_id = status_event.context_id last_task_state = status_event.status.state - updates = self._updates_from_task_update_event(status_event, background=background) + updates = self._updates_from_task_update_event( + status_event, + background=background, + input_request_occurrence_id=input_request_occurrence_id, + ) + updates = self._deduplicate_user_input_request_updates(updates, seen_user_input_request_ids) is_terminal = status_event.status.state in TERMINAL_TASK_STATES is_input_required = status_event.status.state == TaskState.TASK_STATE_INPUT_REQUIRED if emit_intermediate: @@ -714,7 +727,13 @@ def _user_input_request_id(self, task_id: str, occurrence_id: str) -> str: scoped_occurrence = f"{len(self.id)}:{self.id}{len(task_id)}:{task_id}{occurrence_id}" return f"a2a-input-{uuid.uuid5(uuid.NAMESPACE_URL, scoped_occurrence)}" - def _input_required_request(self, task_id: str, message: A2AMessage | None) -> Content: + def _input_required_request( + self, + task_id: str, + message: A2AMessage | None, + *, + fallback_occurrence_id: str, + ) -> Content: """Normalize an A2A input requirement into one durable caller request.""" contents = self._parse_contents_from_a2a(message.parts) if message is not None else [] prompt_parts = [ @@ -726,17 +745,33 @@ def _input_required_request(self, task_id: str, message: A2AMessage | None) -> C {"a2a_input_required_message": MessageToDict(message)} if message is not None else None ), ) - occurrence_id = message.message_id if message is not None and message.message_id else str(uuid.uuid4()) + occurrence_id = message.message_id if message is not None and message.message_id else fallback_occurrence_id request.id = self._user_input_request_id(task_id, occurrence_id) request.user_input_request = True return request + @staticmethod + def _deduplicate_user_input_request_updates( + updates: list[AgentResponseUpdate], + seen_request_ids: set[str], + ) -> list[AgentResponseUpdate]: + """Drop duplicate representations of a user-input request within one agent run.""" + deduplicated: list[AgentResponseUpdate] = [] + for update in updates: + request_ids = {request.id for request in update.user_input_requests if request.id} + if request_ids and request_ids.issubset(seen_request_ids): + continue + seen_request_ids.update(request_ids) + deduplicated.append(update) + return deduplicated + def _updates_from_task( self, task: Task, *, background: bool = False, emit_intermediate: bool = False, + input_request_occurrence_id: str | None = None, streamed_artifact_ids: set[str] | None = None, ) -> list[AgentResponseUpdate]: """Convert an A2A Task into AgentResponseUpdate(s). @@ -754,9 +789,16 @@ def _updates_from_task( if status.state == TaskState.TASK_STATE_INPUT_REQUIRED: message = status.message if status.HasField("message") and status.message.parts else None + occurrence_id = input_request_occurrence_id or task.id return [ AgentResponseUpdate( - contents=[self._input_required_request(task.id, message)], + contents=[ + self._input_required_request( + task.id, + message, + fallback_occurrence_id=occurrence_id, + ) + ], role="assistant" if message is None or message.role == A2ARole.ROLE_AGENT else "user", response_id=task.id, continuation_token=self._build_continuation_token(task) if background else None, @@ -838,6 +880,7 @@ def _updates_from_task_update_event( update_event: TaskStatusUpdateEvent | TaskArtifactUpdateEvent, *, background: bool = False, + input_request_occurrence_id: str | None = None, ) -> list[AgentResponseUpdate]: """Convert A2A task update events into streaming AgentResponseUpdates.""" if isinstance(update_event, TaskArtifactUpdateEvent): @@ -876,9 +919,16 @@ def _updates_from_task_update_event( message_meta = MessageToDict(message.metadata) if message is not None and message.metadata else {} event_meta = MessageToDict(update_event.metadata) if update_event.metadata else {} merged_metadata = {**message_meta, **event_meta} or None + occurrence_id = input_request_occurrence_id or update_event.task_id return [ AgentResponseUpdate( - contents=[self._input_required_request(update_event.task_id, message)], + contents=[ + self._input_required_request( + update_event.task_id, + message, + fallback_occurrence_id=occurrence_id, + ) + ], role="assistant" if message is None or message.role == A2ARole.ROLE_AGENT else "user", response_id=update_event.task_id, message_id=message.message_id if message is not None else None, diff --git a/python/packages/a2a/tests/test_a2a_group_chat.py b/python/packages/a2a/tests/test_a2a_group_chat.py index 25038490126..f11eeab7bab 100644 --- a/python/packages/a2a/tests/test_a2a_group_chat.py +++ b/python/packages/a2a/tests/test_a2a_group_chat.py @@ -5,7 +5,7 @@ from typing import Any, cast import pytest -from a2a.types import Artifact, Part, StreamResponse, Task, TaskState, TaskStatus +from a2a.types import Artifact, Part, StreamResponse, Task, TaskState, TaskStatus, TaskStatusUpdateEvent from a2a.types import Message as A2AMessage from a2a.types import Role as A2ARole from agent_framework import ( @@ -100,6 +100,46 @@ async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: ) +class DuplicateMessageLessInputRequiredA2AClient(InputRequiredA2AClient): + """A2A transport that repeats one message-less prompt in two protocol shapes.""" + + async def send_message(self, request: Any) -> AsyncIterator[StreamResponse]: + self.messages.append(request.message) + if len(self.messages) == 1: + yield StreamResponse( + status_update=TaskStatusUpdateEvent( + task_id="task-input-no-message", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_INPUT_REQUIRED), + ) + ) + yield StreamResponse( + task=Task( + id="task-input-no-message", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_INPUT_REQUIRED), + ) + ) + return + if len(self.messages) == 2: + yield StreamResponse( + task=Task( + id="task-input-no-message", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_INPUT_REQUIRED), + ) + ) + return + yield StreamResponse( + task=Task( + id="task-input-no-message", + context_id="group-chat-context", + status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED), + artifacts=[Artifact(artifact_id="answer", parts=[Part(text="Complete")])], + ) + ) + + class RepeatedInputRequiredA2AClient(InputRequiredA2AClient): """A2A transport that requests caller input twice for the same task.""" @@ -287,6 +327,42 @@ async def test_message_less_input_required_pauses_and_resumes_group_chat() -> No assert len(peer.invocations) == 1 +@pytest.mark.parametrize("stream", [False, True]) +async def test_duplicate_message_less_input_events_share_one_request_per_caller_turn(stream: bool) -> None: + """Duplicate representations share identity until caller input starts a new prompt.""" + client = DuplicateMessageLessInputRequiredA2AClient() + remote = A2AAgent(name="remote", client=cast(Any, client), http_client=None) + peer = SessionBackedAgent() + workflow = GroupChatBuilder( + participants=[remote, peer], + selection_func=lambda state: ["remote", "session-backed"][state.current_round], + max_rounds=2, + ).build() + + if stream: + first_stream = workflow.run("Start", stream=True) + async for _ in first_stream: + pass + first_result = await first_stream.get_final_response() + else: + first_result = await workflow.run("Start") + [first_request] = first_result.get_request_info_events() + + second_result = await workflow.run( + responses={first_request.request_id: Content.from_text(text="First answer")}, + ) + [second_request] = second_result.get_request_info_events() + + assert second_request.request_id != first_request.request_id + assert len(client.messages) == 2 + + await workflow.run( + responses={second_request.request_id: Content.from_text(text="Second answer")}, + ) + assert len(client.messages) == 3 + assert len(peer.invocations) == 1 + + async def test_repeated_input_requests_for_same_remote_task_reject_stale_responses() -> None: """Each remote prompt occurrence has independent workflow correlation.""" client = RepeatedInputRequiredA2AClient() From 9ae2ba94e0b234cce15aa8298d38fcc93bef52c2 Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Thu, 20 Aug 2026 07:47:47 +0900 Subject: [PATCH 09/10] fix(workflows): preserve specialized input requests --- .../core/agent_framework/_workflows/_agent.py | 11 +++---- .../tests/workflow/test_workflow_agent.py | 32 +++++++++++++++++++ .../orchestrations/tests/test_handoff.py | 18 +++++------ 3 files changed, 46 insertions(+), 15 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py index a41c32dd28a..09bd592e922 100644 --- a/python/packages/core/agent_framework/_workflows/_agent.py +++ b/python/packages/core/agent_framework/_workflows/_agent.py @@ -702,24 +702,23 @@ def _process_request_info_event( self, event: WorkflowEvent[Any], ) -> Content: - """Convert a request_info event to FunctionApprovalRequestContent. + """Convert a request_info event to caller-facing content. Args: event: A WorkflowEvent with type='request_info'. Returns: - A content object representing the request info. The content can be a `function_approval_request` - or a `function_call` depending on the structure of the event data. + Specialized user-input request content unchanged, or a `function_call` envelope for generic requests. Note: - If the event data is already a FunctionApprovalRequestContent, it will be returned as-is. + Text requests use the function-call envelope so callers can reply with a matching function result. """ if ( isinstance(event.data, Content) and event.data.user_input_request - and event.data.type in {"function_approval_request", "function_call"} + and event.data.type != "text" ): - # Preserve request contents that already have a matching response envelope. + # Preserve specialized requests that callers already understand how to present. return event.data request_id = event.request_id diff --git a/python/packages/core/tests/workflow/test_workflow_agent.py b/python/packages/core/tests/workflow/test_workflow_agent.py index d5f0c1a6861..70fd94468ca 100644 --- a/python/packages/core/tests/workflow/test_workflow_agent.py +++ b/python/packages/core/tests/workflow/test_workflow_agent.py @@ -123,6 +123,27 @@ async def handle_request_response( ) +class OAuthConsentRequestingExecutor(Executor): + """Executor that pauses for OAuth consent through a specialized request content.""" + + @handler + async def handle_message(self, _: list[Message], ctx: WorkflowContext) -> None: + await ctx.request_info( + Content.from_oauth_consent_request(consent_link="https://example.com/consent"), + Content, + request_id="oauth-consent", + ) + + @response_handler + async def handle_response( + self, + original_request: Content, + response: Content, + ctx: WorkflowContext, + ) -> None: + del original_request, response, ctx + + class ConversationHistoryCapturingExecutor(Executor): """Executor that captures the received conversation history for verification.""" @@ -304,6 +325,17 @@ async def test_end_to_end_request_info_handling(self): pending_requests = await workflow._runner_context.get_pending_request_info_events() assert len(pending_requests) == 0 + async def test_oauth_consent_request_remains_specialized_content(self) -> None: + """Workflow agents expose OAuth consent directly instead of wrapping it as a function call.""" + workflow = WorkflowBuilder(start_executor=OAuthConsentRequestingExecutor(id="oauth")).build() + agent = workflow.as_agent(name="OAuth Workflow Agent") + + response = await agent.run("Connect my account") + + [request] = response.user_input_requests + assert request.type == "oauth_consent_request" + assert request.consent_link == "https://example.com/consent" + def test_request_info_dataclass_arguments_are_serialized_when_content_is_created(self) -> None: """Test WorkflowAgent prepares request_info arguments before observability captures messages.""" executor = SimpleExecutor(id="executor1", response_text="Response") diff --git a/python/packages/orchestrations/tests/test_handoff.py b/python/packages/orchestrations/tests/test_handoff.py index ee4b62c5708..2cc044e651d 100644 --- a/python/packages/orchestrations/tests/test_handoff.py +++ b/python/packages/orchestrations/tests/test_handoff.py @@ -669,9 +669,15 @@ def terminate_after_second_real_user_turn(conversation: list[Message]) -> bool: events = list(result) final_state = result.get_final_state() - assert len(target_client.received_messages) == 1 - assert [message.text for message in target_client.received_messages[0] if message.role == "user"] == [initial_task] - assert len(source_client.received_messages) == 2 + received_messages = [ + [(message.role, message.text) for message in invocation] + for invocation in [*source_client.received_messages, *target_client.received_messages] + ] + assert received_messages == [ + [("user", initial_task)], + [("user", initial_task), ("assistant", ""), ("tool", "")], + [("user", initial_task)], + ] revisited_source_messages = source_client.received_messages[1] source_call_ids = { @@ -689,12 +695,6 @@ def terminate_after_second_real_user_turn(conversation: list[Message]) -> bool: assert "source-handoff-0" in source_call_ids assert "source-handoff-0" in source_result_ids - all_received_messages = [ - message - for invocation in [*source_client.received_messages, *target_client.received_messages] - for message in invocation - ] - assert all("continue the conversation" not in (message.text or "").lower() for message in all_received_messages) assert observed_user_turns assert all(user_turns == [initial_task] for user_turns in observed_user_turns) From 5aa041620f0a50b7ad92dd249271131232fd913c Mon Sep 17 00:00:00 2001 From: Evan Mattson Date: Thu, 20 Aug 2026 08:47:22 +0900 Subject: [PATCH 10/10] test(openai): use current web search model --- .../openai/tests/openai/test_openai_chat_completion_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/packages/openai/tests/openai/test_openai_chat_completion_client.py b/python/packages/openai/tests/openai/test_openai_chat_completion_client.py index 43026fad42c..d63f3a3e100 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_completion_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_completion_client.py @@ -2117,7 +2117,7 @@ async def test_integration_options( @pytest.mark.integration @skip_if_openai_integration_tests_disabled async def test_integration_web_search() -> None: - client = OpenAIChatCompletionClient(model="gpt-4o-search-preview") + client = OpenAIChatCompletionClient(model="gpt-5-search-api") for streaming in [False, True]: # Use static method for web search tool