Python: fix streaming transcript duplication with message injection and per-service-call persistence

docs/specs/004-python-function-calling-loop.md

@@ -372,6 +372,9 @@ 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.
 - 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.
 
@@ -475,6 +478,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 retains the inner conversation id and its internal marker, so the next iteration appends only the latest message rather than replaying the whole turn on top of provider-held history. | `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`, `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` |

python/packages/core/agent_framework/_sessions.py

@@ -1260,6 +1260,28 @@ 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.
+    """
+    if inner_response.conversation_id is not None:
+        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 +1405,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 +1419,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 +1462,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),
         )
 
 

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

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,111 @@ 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()
+
     def test_enqueue_messages_uses_session_state_queue(self) -> None:
         """Test that standalone message injection enqueueing stores messages in session state."""
         session = AgentSession()