.NET: Python: [Bug]: Streaming create_harness_agent runs duplicate the transcript after tool calls — 400 "insufficient tool messages following tool_calls"

[naeyn]

Description

When an agent built with create_harness_agent runs with stream=True against a chat-completions endpoint, any turn where the model calls a tool fails on the second model call of that turn. DeepSeek rejects it with:

An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. (insufficient tool messages following tool_calls message)

I captured the outgoing requests: the second request contains the current turn twice — once injected from persisted history (user message, assistant text, assistant tool_calls, with no tool results after it) and once from the function loop's in-flight transcript (same messages plus the tool results). The first, dangling tool_calls copy violates the message-ordering constraint. I hit this on DeepSeek, but the history is invalid for any endpoint that enforces the OpenAI ordering rule.

The same agent run non-streaming builds correct requests, so I dug into where the two paths diverge.

Root cause

1. After the first model call, PerServiceCallHistoryPersistingMiddleware._finalize_response correctly sets response.conversation_id = LOCAL_HISTORY_CONVERSATION_ID (plus mark_internal_conversation_id()), because the response contains function calls.
2. MessageInjectionMiddleware.process wraps the streaming result in its own ResponseStream with finalizer=lambda updates: ChatResponse.from_updates(updates, ...). The rebuilt outer response is assembled from the streamed updates only — the sentinel was set on the inner final-response object, never on an update — so it comes out with conversation_id=None.
3. The function loop's _prepare_messages_for_next_iteration sees conversation_id is None, takes the no-service-managed-history branch, and extends prepared_messages with the full turn instead of sending only response.messages[-1:].
4. On the next model call the history provider injects the persisted copy of the same messages as well → duplicated transcript with a dangling tool_calls message → HTTP 400.

Tracing the sentinel through the three layers on the streaming path:

PSC._finalize_response  -> conversation_id='agent_framework_local_history_persistence'
from_updates            -> conversation_id=None
function loop sees         conversation_id=None -> resends full turn

and on the non-streaming path (where _process_non_streaming passes context.result through unchanged):

PSC._finalize_response  -> conversation_id='agent_framework_local_history_persistence'
function loop sees         conversation_id='agent_framework_local_history_persistence' -> sends tool results only

To make sure I wasn't blaming the wrong component, I ran four configurations against the same mock endpoint and validated every outgoing request ("invalid" = an assistant tool_calls message not immediately followed by matching tool messages):

Configuration	Requests
create_harness_agent, streaming	invalid — turn duplicated
create_harness_agent, non-streaming	valid
plain Agent + require_per_service_call_history_persistence=True + InMemoryHistoryProvider, streaming, no message injection middleware	valid
create_harness_agent, streaming, disable_todo=True, disable_file_memory=True	invalid

Row 3 isolates the problem to MessageInjectionMiddleware's streaming path; row 4 rules out the todo/file-memory providers as the source of the extra messages (this is not the Python analog of #6953). Since the harness installs the middleware unconditionally ("Message injection is always on … there is no opt-out"), every streaming create_harness_agent user with tool calls on a strict endpoint is affected.

Expected behavior: the streaming path should build the same requests as the non-streaming path — persisted prefix injected once, function loop sending only the new tool-result messages.

Steps to reproduce: run the script below (self-contained; mocks the endpoint, no API key). It prints every request the framework sends and flags the ordering violation. Replacing stream=True with a plain await agent.run(...) makes request 2 come out valid.

Code Sample

"""Repro: streaming + create_harness_agent duplicates the transcript in follow-up
requests, producing history that strict endpoints reject with "An assistant
message with 'tool_calls' must be followed by tool messages"."""

import asyncio
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Annotated

from agent_framework import create_harness_agent, tool
from agent_framework.openai import OpenAIChatCompletionClient

CAPTURED: list[dict] = []


@tool
def lookup(query: Annotated[str, "Query"]) -> str:
    """Dummy lookup tool."""
    return f"result for {query}"


def sse(obj) -> bytes:
    return b"data: " + json.dumps(obj).encode() + b"\n\n"


def chunk(delta, finish=None):
    return {
        "id": "x", "object": "chat.completion.chunk", "created": 1, "model": "mock",
        "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
    }


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        CAPTURED.append(json.loads(self.rfile.read(int(self.headers["Content-Length"]))))
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.end_headers()
        if len(CAPTURED) == 1:
            # First model turn: text + one tool call, OpenAI streaming style.
            self.wfile.write(sse(chunk({"role": "assistant", "content": "Looking it up."})))
            self.wfile.write(sse(chunk({"tool_calls": [{"index": 0, "id": "call_1", "type": "function",
                "function": {"name": "lookup", "arguments": '{"query": "test"}'}}]})))
            self.wfile.write(sse(chunk({}, finish="tool_calls")))
        else:
            self.wfile.write(sse(chunk({"role": "assistant", "content": "Done."})))
            self.wfile.write(sse(chunk({}, finish="stop")))
        self.wfile.write(b"data: [DONE]\n\n")


def validate(messages) -> list[str]:
    """OpenAI constraint: assistant tool_calls must be immediately followed by
    one tool message per tool_call_id."""
    problems, i = [], 0
    while i < len(messages):
        m = messages[i]
        if m.get("role") == "assistant" and m.get("tool_calls"):
            ids = [tc["id"] for tc in m["tool_calls"]]
            j, answered = i + 1, []
            while j < len(messages) and messages[j].get("role") == "tool":
                answered.append(messages[j].get("tool_call_id"))
                j += 1
            if sorted(ids) != sorted(answered):
                problems.append(f"messages[{i}] has tool_calls {ids} but is followed by tool messages {answered}")
            i = j
        else:
            i += 1
    return problems


async def main():
    server = ThreadingHTTPServer(("127.0.0.1", 8399), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()

    client = OpenAIChatCompletionClient(model="mock", base_url="http://127.0.0.1:8399/v1", api_key="dummy")
    agent = create_harness_agent(client, tools=[lookup], disable_web_search=True)
    session = agent.create_session()

    async for _update in agent.run("look up test", session=session, stream=True):
        pass
    server.shutdown()

    for n, req in enumerate(CAPTURED, 1):
        print(f"--- request {n} ({len(req['messages'])} messages)")
        for m in req["messages"]:
            tc = [t["id"] for t in m.get("tool_calls", [])] if m.get("tool_calls") else ""
            print(f"    {m['role']:9} {tc} {str(m.get('content'))[:60]!r}")
        for p in validate(req["messages"]):
            print(f"    !! INVALID HISTORY: {p}")


if __name__ == "__main__":
    asyncio.run(main())

Output — note that in request 2 the first tool_calls message is followed by a user message, not tool results:

--- request 1 (3 messages)
    system     'You are a helpful AI assistant that uses tools to complete t'
    user       '### Current todo list\n- none yet'
    user       'look up test'
--- request 2 (9 messages)
    system     'You are a helpful AI assistant that uses tools to complete t'
    user       '### Current todo list\n- none yet'
    user       'look up test'
    assistant  'Looking it up.'
    assistant ['call_1'] 'None'
    user       'look up test'
    assistant  'Looking it up.'
    assistant ['call_1'] 'None'
    tool       'result for test'
    !! INVALID HISTORY: messages[4] has tool_calls ['call_1'] but is followed by tool messages []

Error Messages / Stack Traces

Against a real endpoint (DeepSeek deepseek-chat via OpenAIChatCompletionClient):

agent_framework.exceptions.ChatClientException: <class 'agent_framework_openai._chat_completion_client.OpenAIChatCompletionClient'>
service failed to complete the prompt: Error code: 400 -
{'error': {'message': "An assistant message with 'tool_calls' must be followed by
tool messages responding to each 'tool_call_id'. (insufficient tool messages
following tool_calls message)", 'type': 'invalid_request_error',
'param': None, 'code': 'invalid_request_error'}}

Raised from agent_framework_openai/_chat_completion_client.py:594 on the second model call of the turn (trace passes through _harness/_tool_approval.py:456, _tools.py:3071, _sessions.py:1178).

Package Versions

agent-framework: 1.13.0, agent-framework-core: 1.13.0, agent-framework-openai: 1.12.0

Python Version

Python 3.12 (Windows 11, x64)

Additional Context

• Still present on main: _stream_injected_messages and the from_updates finalizer in process() are unchanged from 1.13.0 as of 7302d0b (2026-08-07).
• This looks like the same bug class as Python: Fix streamed workflow agent continuation context by finalizing AgentExecutor streams #3882, which fixed an AgentExecutor that "rebuilt the final response from collected updates, instead of finalizing the ResponseStream via get_final_response()", bypassing hooks that persist continuation state. The same remedy seems applicable here: _stream_injected_messages already awaits stream.get_final_response(), so the sentinel is in hand — propagating the inner final response (or at least its conversation_id + internal marker) into the outer stream's finalizer instead of rebuilding from updates should fix it.
• Searched for duplicates: .NET: [Bug]: TodoProvider injects a synthetic user message between assistant tool_calls and tool results, causing OpenAI 400 invalid_request_error #6953 (.NET TodoProvider, same 400 text, different mechanism, fixed), Python: Fix per-service-call history persistence with server-storing clients #6310 (per-service-call persistence with server-storing clients), Python: [Bug]: Magentic manager duplicates conversation history on every model call #6298 (Magentic history duplication), .NET: AG-UI: Multi-turn tool calls replay produces invalid OpenAI tool_call history #2699/.NET: Python: [Bug]: AG-UI appends approval-resolved tool results out of order — invalid history for strict chat providers #6909/.NET: [Bug]: FilterServerToolsFromMixedToolInvocationsAsync causes HTTP 400 when LLM returns mixed frontend + server-side tool calls in AGUI hosted mode #6967 (AG-UI/.NET tool-message ordering). None cover this path.
• Workaround: run the agent non-streaming.
• Not bisected against earlier releases, so I can't say whether it's a regression; both interacting pieces (message-injection middleware Python: Add message injection middleware #6998/Python: Integrate message injection into harness agent #7027, local-history sentinel) are recent.

[github-actions]

🤖 Automated triage reproduction notes (agent-authored — trust but verify)

Agent analysis

Repro: MessageInjectionMiddleware.process() in packages/core/agent_framework/_sessions.py:1427-1429 wraps the inner streaming result in an outer ResponseStream with finalizer=ChatResponse.from_updates(...), which discards the conversation_id sentinel set by PerServiceCallHistoryPersistingMiddleware._finalize_response (line 1565-1567). The function loop at _tools.py:2616 then sees conversation_id=None and extends prepared_messages with the full turn, causing duplication when PerServiceCallHistoryPersistingMiddleware loads persisted history on the next call. Minimal repro: create_harness_agent with a FunctionInvocationLayer+ChatMiddlewareLayer+BaseChatClient mock that streams a tool call, then run with stream=True — the second model request contains duplicated messages with a dangling tool_calls message.

• Failing test: python/packages/core/tests/core/test_streaming_tool_call_history_bug.py
• Files examined: python/packages/core/agent_framework/_sessions.py, python/packages/core/agent_framework/_tools.py, python/packages/core/agent_framework/_agents.py, python/packages/core/agent_framework/_clients.py, python/packages/core/agent_framework/_middleware.py, python/packages/core/agent_framework/_harness/_agent.py
• Tests run: packages/core/tests/core/test_streaming_tool_call_history_bug.py::test_streaming_harness_agent_does_not_duplicate_transcript (FAILED), packages/core/tests/core/test_streaming_tool_call_history_bug.py::test_non_streaming_harness_agent_valid_transcript (PASSED)
• Reported version: 1.13.0
• Current version: 1.13.0