diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index 0b4251b9b4..a5f88c0368 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/docs/specs/004-python-function-calling-loop.md @@ -376,6 +376,11 @@ that manually replay messages own the equivalent rule: do not resend an approval - Model-bound history contains one function call/result pair per completed logical occurrence. - Append-only history must not replay stale approval request/response wrappers to the model. - Framework-managed and service-managed continuation must preserve the same logical call/result transcript. +- A streaming response rebuilt from updates by an intermediate middleware must carry over the inner response's + conversation id and its internal-conversation-id marker, so framework-managed continuation appends only the latest + message instead of replaying a transcript the provider already holds. The rebuilt response mirrors the inner + conversation id exactly, including clearing it, and never retains an id emitted by an earlier service call in the + same turn. - A trusted terminal result consumes the corresponding approval authority in explicit stateless replay; a result in a server-registered pending occurrence cannot consume that authority before local execution. @@ -479,6 +484,7 @@ that manually replay messages own the equivalent rule: do not resend an approval | Pending hosted history replay | Stateless hosted approval requests remain replayable until a response is recorded, then both controls become inert. | `packages/openai/tests/openai/test_openai_chat_client.py::test_stateless_history_preserves_pending_hosted_approval_request_until_response` | | Non-history provider plus session | Local history is still auto-injected for approval resume. | `packages/core/tests/core/test_agents.py::test_non_history_context_provider_still_injects_inmemory` | | Hosted per-service-call persistence | A host-managed transcript remains available throughout a local function-call loop without being persisted into the framework session and replayed on the next hosted request. | `packages/foundry_hosting/tests/test_responses.py::TestAgentSessionPersistence::test_per_service_call_persistence_preserves_function_loop_history` | +| Streaming message injection with per-service-call persistence | A streaming response rebuilt from updates mirrors the inner conversation id exactly, including clearing it, and keeps its internal marker, so the next iteration appends only the latest message rather than replaying the whole turn on top of provider-held history, and never persists a conversation id from an earlier injected service call. | `packages/core/tests/core/test_middleware_with_chat.py::TestChatMiddleware::test_message_injection_middleware_streaming_preserves_inner_continuation_state`, `test_message_injection_middleware_streaming_keeps_service_conversation_id_external`, `test_message_injection_middleware_streaming_clears_conversation_id_when_final_call_has_none`, `test_message_injection_middleware_conversation_id_matches_across_streaming_modes`, `packages/core/tests/core/test_harness_agent.py::test_streaming_harness_tool_call_does_not_duplicate_transcript` | | Service-side approval decision | Stored hosted request is skipped; the current approved or rejected hosted response is sent, while local approval controls are omitted from provider input. | `packages/openai/tests/openai/test_openai_chat_client.py::test_prepare_messages_strips_approval_request_but_keeps_response_under_storage`, `test_prepare_messages_drops_local_approval_controls` | | OpenAI approval serialization | Hosted approval id and decision serialize to `mcp_approval_response`; local approvals remain in-process. | `test_prepare_message_for_openai_with_function_approval_response`, `test_prepare_content_for_opentool_approval_response`, `test_function_approval_response_with_mcp_tool_call` | | OpenAI end-to-end hosted approval | Hosted request parses, response sends, and continuation completes. | `test_end_to_end_mcp_approval_flow` | diff --git a/python/packages/core/agent_framework/_sessions.py b/python/packages/core/agent_framework/_sessions.py index 893bb96e52..5d936118cb 100644 --- a/python/packages/core/agent_framework/_sessions.py +++ b/python/packages/core/agent_framework/_sessions.py @@ -1260,6 +1260,27 @@ def _response_contains_follow_up_request(response: ChatResponse) -> bool: ) +def _carry_over_stream_control_state(response: ChatResponse, inner_response: ChatResponse) -> None: + """Copy control-flow state from an inner final response onto a rebuilt outer response. + + A response rebuilt from streamed updates can only see state that was emitted on an update. + Middleware that runs closer to the leaf client (for example + :class:`PerServiceCallHistoryPersistingMiddleware`) applies its continuation state through a + result hook, i.e. on the inner final response *after* the stream has been consumed, so that + state has to be carried over explicitly. Dropping it makes the function-invocation loop treat + the turn as if no history were managed and resend the whole turn, duplicating the transcript. + + Args: + response: The outer response rebuilt from the streamed updates. + inner_response: The final response of the innermost stream that produced those updates. + """ + response.conversation_id = inner_response.conversation_id + if inner_response.has_internal_conversation_id(): + response.mark_internal_conversation_id() + else: + response.clear_internal_conversation_id() + + def _split_service_call_messages(messages: Sequence[Message]) -> tuple[list[Message], dict[str, list[Message]]]: """Split service-call messages into input messages and attributed context messages.""" input_messages: list[Message] = [] @@ -1383,6 +1404,7 @@ async def _stream_injected_messages( context: ChatContext, call_next: Callable[[], Awaitable[None]], session: AgentSession, + inner_responses: list[ChatResponse], ) -> AsyncIterable[ChatResponseUpdate]: while True: context.messages = self._drain_pending_messages(session, context.messages) @@ -1396,12 +1418,27 @@ async def _stream_injected_messages( async for update in stream: yield update response = await stream.get_final_response() + inner_responses.append(response) if _response_contains_follow_up_request(response) or not self._has_pending_messages(session): return self._update_context_conversation_id(context, response.conversation_id) empty_messages: list[Message] = [] context.messages = empty_messages + @staticmethod + def _finalize_injected_stream( + updates: Sequence[ChatResponseUpdate], + inner_responses: Sequence[ChatResponse], + response_format: Any | None, + ) -> ChatResponse: + """Rebuild the outer response from the streamed updates, keeping inner continuation state.""" + response = ChatResponse.from_updates(updates, output_format_type=response_format) + if inner_responses: + # The last inner response is the one the non-streaming path would return, so it also + # owns the continuation state for the next function-loop iteration. + _carry_over_stream_control_state(response, inner_responses[-1]) + return response + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: """Inject pending session messages into chat model calls. @@ -1424,9 +1461,10 @@ async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[ return response_format = context.options.get("response_format") if context.options is not None else None + inner_responses: list[ChatResponse] = [] context.result = ResponseStream( - self._stream_injected_messages(context, call_next, session), - finalizer=lambda updates: ChatResponse.from_updates(updates, output_format_type=response_format), + self._stream_injected_messages(context, call_next, session, inner_responses), + finalizer=lambda updates: self._finalize_injected_stream(updates, inner_responses, response_format), ) diff --git a/python/packages/core/tests/core/test_harness_agent.py b/python/packages/core/tests/core/test_harness_agent.py index 36a1b5f99c..b74902116e 100644 --- a/python/packages/core/tests/core/test_harness_agent.py +++ b/python/packages/core/tests/core/test_harness_agent.py @@ -1434,3 +1434,123 @@ def test_create_harness_agent_shell_dedup_does_not_suppress_harness_warning() -> disable_file_memory=True, background_agents=[bg_agent], # type: ignore[list-item] # pyrefly: ignore[bad-argument-type] # ty: ignore[invalid-argument-type] ) + + +def _tool_call_ordering_problems(messages: Sequence[Message]) -> list[str]: + """Report assistant function calls that are not immediately followed by their results.""" + problems: list[str] = [] + index = 0 + while index < len(messages): + call_ids = [ + str(content.call_id) + for content in messages[index].contents + if content.type == "function_call" and not content.informational_only + ] + if call_ids: + result_ids: list[str] = [] + next_index = index + 1 + while next_index < len(messages) and any( + content.type == "function_result" for content in messages[next_index].contents + ): + result_ids.extend( + str(content.call_id) + for content in messages[next_index].contents + if content.type == "function_result" + ) + next_index += 1 + if sorted(call_ids) != sorted(result_ids): + problems.append(f"messages[{index}] requests {call_ids} but is followed by results {result_ids}") + index = next_index + else: + index += 1 + return problems + + +async def _run_harness_tool_call_turn( + chat_client_base: Any, + *, + stream: bool, +) -> tuple[list[list[Message]], AgentSession]: + """Run one harness turn where the model calls a tool, capturing the messages sent per model call.""" + from agent_framework import tool + + @tool + def lookup(query: str) -> str: + """Look up a fact.""" + return f"result for {query}" + + captured: list[list[Message]] = [] + + def build_updates(call_index: int) -> list[ChatResponseUpdate]: + if call_index == 0: + return [ + ChatResponseUpdate(contents=[Content.from_text("Looking it up.")], role="assistant"), + ChatResponseUpdate( + contents=[ + Content.from_function_call(call_id="call_1", name="lookup", arguments={"query": "widgets"}) + ], + role="assistant", + ), + ] + return [ChatResponseUpdate(contents=[Content.from_text("Done.")], role="assistant", finish_reason="stop")] + + def fake_streaming_response( + *, messages: Sequence[Message], options: dict[str, Any], **kwargs: Any + ) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + call_index = len(captured) + captured.append(list(messages)) + + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + for update in build_updates(call_index): + yield update + + return ResponseStream(_stream(), finalizer=ChatResponse.from_updates) + + async def fake_get_response(*, messages: Sequence[Message], options: dict[str, Any], **kwargs: Any) -> ChatResponse: + call_index = len(captured) + captured.append(list(messages)) + return ChatResponse.from_updates(build_updates(call_index)) + + agent = create_harness_agent( + client=chat_client_base, + tools=[lookup], + disable_web_search=True, + disable_todo=True, + disable_mode=True, + disable_file_memory=True, + ) + session = agent.create_session() + + with ( + patch.object(chat_client_base, "_get_streaming_response", side_effect=fake_streaming_response), + patch.object(chat_client_base, "_get_non_streaming_response", side_effect=fake_get_response), + ): + if stream: + async for _update in agent.run("look up widgets", session=session, stream=True): + pass + else: + await agent.run("look up widgets", session=session) + + return captured, session + + +async def test_streaming_harness_tool_call_does_not_duplicate_transcript(chat_client_base: Any) -> None: + """Regression for #7591: a streaming tool-call turn must not resend the turn twice. + + The streaming path rebuilt the response from updates in ``MessageInjectionMiddleware``, losing the + local-history conversation-id sentinel set by ``PerServiceCallHistoryPersistingMiddleware``. The + function loop then resent the whole turn on top of the injected history, leaving an assistant + function call with no results after it — which strict endpoints reject. + """ + streaming_calls, session = await _run_harness_tool_call_turn(chat_client_base, stream=True) + non_streaming_calls, _ = await _run_harness_tool_call_turn(chat_client_base, stream=False) + + assert len(streaming_calls) == 2 + second_call = streaming_calls[1] + assert _tool_call_ordering_problems(second_call) == [] + assert sum(1 for message in second_call if message.text == "look up widgets") == 1 + assert sum(1 for message in second_call for content in message.contents if content.type == "function_call") == 1 + # The streaming path must build the same transcript as the non-streaming path. + assert [message.text for message in second_call] == [message.text for message in non_streaming_calls[1]] + # The local sentinel is control-flow state and must never become a service session id. + assert session.service_session_id is None diff --git a/python/packages/core/tests/core/test_middleware_with_chat.py b/python/packages/core/tests/core/test_middleware_with_chat.py index c2b794f4c5..bb5f0c80b3 100644 --- a/python/packages/core/tests/core/test_middleware_with_chat.py +++ b/python/packages/core/tests/core/test_middleware_with_chat.py @@ -27,6 +27,7 @@ function_middleware, tool, ) +from agent_framework._sessions import LOCAL_HISTORY_CONVERSATION_ID from agent_framework.exceptions import ChatClientInvalidRequestException from .conftest import MockBaseChatClient @@ -624,6 +625,276 @@ async def stream() -> AsyncIterable[ChatResponseUpdate]: assert [update.text for update in updates] == ["", "done"] assert captured_messages == [["user message"], ["queued while streaming hosted tool"]] + async def test_message_injection_middleware_streaming_preserves_inner_continuation_state( + self, chat_client_base: "MockBaseChatClient" + ) -> None: + """Regression for #7591: result-hook state on the inner response survives the outer rebuild. + + ``PerServiceCallHistoryPersistingMiddleware`` marks the local-history sentinel on the inner + final response through a result hook, so it is never emitted on an update. Rebuilding the + outer response from updates dropped it, and the function loop then resent the whole turn. + """ + session = AgentSession() + injection = MessageInjectionMiddleware() + observed: list[ChatResponse] = [] + + class _ObservingMiddleware(ChatMiddleware): + """Capture the response the function-invocation loop sees for each model call.""" + + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + await call_next() + stream = cast(ResponseStream[ChatResponseUpdate, ChatResponse], context.result) + + def record(response: ChatResponse) -> ChatResponse: + observed.append(response) + return response + + context.result = stream.with_result_hook(record) + + class _SentinelMiddleware(ChatMiddleware): + """Stand-in for the per-service-call history middleware's result hook.""" + + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + await call_next() + stream = cast(ResponseStream[ChatResponseUpdate, ChatResponse], context.result) + + def mark(response: ChatResponse) -> ChatResponse: + response.conversation_id = LOCAL_HISTORY_CONVERSATION_ID + response.mark_internal_conversation_id() + return response + + context.result = stream.with_result_hook(mark) + + stream = chat_client_base.get_response( + [Message(role="user", contents=["user message"])], + stream=True, + client_kwargs={ + "middleware": [_ObservingMiddleware(), injection, _SentinelMiddleware()], + "session": session, + }, + ) + async for _update in stream: + pass + await stream.get_final_response() + + assert [response.conversation_id for response in observed] == [LOCAL_HISTORY_CONVERSATION_ID] + assert observed[0].has_internal_conversation_id() + + async def test_message_injection_middleware_streaming_keeps_service_conversation_id_external( + self, chat_client_base: "MockBaseChatClient" + ) -> None: + """Test that a real service conversation id is preserved and not marked internal.""" + session = AgentSession() + injection = MessageInjectionMiddleware() + observed: list[ChatResponse] = [] + + class _ObservingMiddleware(ChatMiddleware): + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + await call_next() + stream = cast(ResponseStream[ChatResponseUpdate, ChatResponse], context.result) + + def record(response: ChatResponse) -> ChatResponse: + observed.append(response) + return response + + context.result = stream.with_result_hook(record) + + def fake_streaming_response( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + async def stream() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate( + contents=[Content.from_text("done")], + role="assistant", + conversation_id="service-conversation", + ) + + return ResponseStream(stream(), finalizer=ChatResponse.from_updates) + + with patch.object(chat_client_base, "_get_streaming_response", side_effect=fake_streaming_response): + stream = chat_client_base.get_response( + [Message(role="user", contents=["user message"])], + stream=True, + client_kwargs={ + "middleware": [_ObservingMiddleware(), injection], + "session": session, + }, + ) + async for _update in stream: + pass + await stream.get_final_response() + + assert [response.conversation_id for response in observed] == ["service-conversation"] + assert not observed[0].has_internal_conversation_id() + + async def test_message_injection_middleware_streaming_clears_conversation_id_when_final_call_has_none( + self, chat_client_base: "MockBaseChatClient" + ) -> None: + """The rebuilt response must not keep a conversation id from an earlier injected service call. + + ``ChatResponse.from_updates`` keeps the last non-``None`` conversation id and never clears + one, and the outer stream spans every injected service call. Without an unconditional + mirror, an id emitted on the first call survives onto the rebuilt response even though the + final inner response has none, so the function loop would persist a stale service + conversation that the non-streaming path never produces. + """ + session = AgentSession() + injection = MessageInjectionMiddleware() + observed: list[ChatResponse] = [] + + class _ObservingMiddleware(ChatMiddleware): + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + await call_next() + stream = cast(ResponseStream[ChatResponseUpdate, ChatResponse], context.result) + + def record(response: ChatResponse) -> ChatResponse: + observed.append(response) + return response + + context.result = stream.with_result_hook(record) + + call_count = 0 + + def fake_streaming_response( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + nonlocal call_count + call_count += 1 + first_call = call_count == 1 + + async def stream() -> AsyncIterable[ChatResponseUpdate]: + if first_call: + # The service returns an id for this call only, then stops returning one. + yield ChatResponseUpdate( + contents=[Content.from_text("first")], + role="assistant", + conversation_id="service-conversation", + ) + enqueue_messages(session, "queued while streaming") + return + yield ChatResponseUpdate(contents=[Content.from_text("second")], role="assistant") + + return ResponseStream(stream(), finalizer=ChatResponse.from_updates) + + with patch.object(chat_client_base, "_get_streaming_response", side_effect=fake_streaming_response): + stream = chat_client_base.get_response( + [Message(role="user", contents=["user message"])], + stream=True, + client_kwargs={ + "middleware": [_ObservingMiddleware(), injection], + "session": session, + }, + ) + async for _update in stream: + pass + await stream.get_final_response() + + assert call_count == 2 + assert [response.conversation_id for response in observed] == [None] + assert session.service_session_id is None + + async def _injected_conversation_id( + self, chat_client_base: "MockBaseChatClient", *, stream_mode: bool + ) -> str | None: + """Run two injected service calls and return the final conversation id. + + The first call reports a conversation id and queues a message, the second call reports + none. Defined as a helper rather than a loop body so the closures below do not capture a + loop variable. + """ + session = AgentSession() + injection = MessageInjectionMiddleware() + observed: list[ChatResponse] = [] + call_count = 0 + + class _ObservingMiddleware(ChatMiddleware): + async def process(self, context: ChatContext, call_next: Callable[[], Awaitable[None]]) -> None: + await call_next() + if not isinstance(context.result, ResponseStream): + observed.append(cast(ChatResponse, context.result)) + return + stream = cast(ResponseStream[ChatResponseUpdate, ChatResponse], context.result) + + def record(response: ChatResponse) -> ChatResponse: + observed.append(response) + return response + + context.result = stream.with_result_hook(record) + + def fake_streaming_response( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + nonlocal call_count + call_count += 1 + first_call = call_count == 1 + + async def updates() -> AsyncIterable[ChatResponseUpdate]: + if first_call: + yield ChatResponseUpdate( + contents=[Content.from_text("first")], + role="assistant", + conversation_id="service-conversation", + ) + enqueue_messages(session, "queued after first call") + return + yield ChatResponseUpdate(contents=[Content.from_text("second")], role="assistant") + + return ResponseStream(updates(), finalizer=ChatResponse.from_updates) + + async def fake_response( + *, + messages: Sequence[Message], + options: dict[str, Any], + **kwargs: Any, + ) -> ChatResponse: + nonlocal call_count + call_count += 1 + if call_count == 1: + enqueue_messages(session, "queued after first call") + return ChatResponse( + messages=[Message(role="assistant", contents=["first"])], + conversation_id="service-conversation", + ) + return ChatResponse(messages=[Message(role="assistant", contents=["second"])]) + + messages = [Message(role="user", contents=["user message"])] + client_kwargs: dict[str, Any] = { + "middleware": [_ObservingMiddleware(), injection], + "session": session, + } + with ( + patch.object(chat_client_base, "_get_streaming_response", side_effect=fake_streaming_response), + patch.object(chat_client_base, "_get_non_streaming_response", side_effect=fake_response), + ): + if stream_mode: + stream = chat_client_base.get_response(messages, stream=True, client_kwargs=client_kwargs) + async for _update in stream: + pass + await stream.get_final_response() + else: + await chat_client_base.get_response(messages, stream=False, client_kwargs=client_kwargs) + + assert call_count == 2 + return observed[-1].conversation_id + + async def test_message_injection_middleware_conversation_id_matches_across_streaming_modes( + self, chat_client_base: "MockBaseChatClient" + ) -> None: + """Streaming and non-streaming injection must agree on the final conversation id.""" + streaming = await self._injected_conversation_id(chat_client_base, stream_mode=True) + non_streaming = await self._injected_conversation_id(chat_client_base, stream_mode=False) + + assert streaming == non_streaming + def test_enqueue_messages_uses_session_state_queue(self) -> None: """Test that standalone message injection enqueueing stores messages in session state.""" session = AgentSession()