Skip to content

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

Description

@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

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Labels

.NETUsage: [Issues, PRs], Target: .NetpythonUsage: [Issues, PRs], Target: PythonreproducedUsage: [Issues], Target: all issues that can be reproduced by the triage workflow

Type

Projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions