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

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` |

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),
         )
 
 

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