Tool progress SSE events corrupting model behavior fix

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #7500 fix(api): send tool progress as custom SSE event to prevent model corruption

  merged to main

  files: gateway/platforms/api_server.py, tests/gateway/test_api_server.py

  Show the diff

  • gateway/platforms/api_server.py

    @@ -644,15 +644,35 @@ def _on_delta(delta):
                         _stream_q.put(delta)
     
                 def _on_tool_progress(event_type, name, preview, args, **kwargs):
    -                """Inject tool progress into the SSE stream for Open WebUI."""
    +                """Send tool progress as a separate SSE event.
    +
    +                Previously, progress markers like ``⏰ list`` were injected
    +                directly into ``delta.content``.  OpenAI-compatible frontends
    +                (Open WebUI, LobeChat, …) store ``delta.content`` verbatim as
    +                the assistant message and send it back on subsequent requests.
    +                After enough turns the model learns to *emit* the markers as
    +                plain text instead of issuing real tool calls — silently
    +                hallucinating tool results.  See #6972.
    +
    +                The fix: push a tagged tuple ``("__tool_progress__", payload)``
    +                onto the stream queue.  The SSE writer emits it as a custom
    +                ``event: hermes.tool.progress`` line that compliant frontends
    +                can render for UX but will *not* persist into conversation
    +                history.  Clients that don't understand the custom event type
    +                silently ignore it per the SSE specification.
    +                """
                     if event_type != "tool.started":
    -                    return  # Only show tool start events in chat stream
    +                    return
                     if name.startswith("_"):
    -                    return  # Skip internal events (_thinking)
    +                    return
                     from agent.display import get_tool_emoji
                     emoji = get_tool_emoji(name)
                     label = preview or name
    -                _stream_q.put(f"\n`{emoji} {label}`\n")
    +                _stream_q.put(("__tool_progress__", {
    +                    "tool": name,
    +                    "emoji": emoji,
    +                    "label": label,
    +                }))
     
                 # Start agent in background.  agent_ref is a mutable container
                 # so the SSE writer can interrupt the agent on client disconnect.
    @@ -763,6 +783,29 @@ async def _write_sse_chat_completion(
                 }
                 await response.write(f"data: {json.dumps(role_chunk)}\n\n".encode())
     
    +            # Helper — route a queue item to the correct SSE event.
    +            async def _emit(item):
    +                """Write a single queue item to the SSE stream.
    +
    +                Plain strings are sent as normal ``delta.content`` chunks.
    +                Tagged tuples ``("__tool_progress__", payload)`` are sent
    +                as a custom ``event: hermes.tool.progress`` SSE event so
    +                frontends can display them without storing the markers in
    +                conversation history.  See #6972.
    +                """
    +                if isinstance(item, tuple) and len(item) == 2 and item[0] == "__tool_progress__":
    +                    event_data = json.dumps(item[1])
    +                    await response.write(
    +                        f"event: hermes.tool.progress\ndata: {event_data}\n\n".encode()
    +                    )
    +                else:
    +                    content_chunk = {
    +                        "id": completion_id, "object": "chat.completion.chunk",
    +                        "created": created, "model": model,
    +                        "choices": [{"index": 0, "delta": {"content": item}, "finish_reason": None}],
    +                    }
    +                    await response.write(f"data: {json.dumps(content_chunk)}\n\n".encode())
    +
                 # Stream content chunks as they arrive from the agent
                 loop = asyncio.get_event_loop()
                 while True:
    @@ -776,12 +819,7 @@ async def _write_sse_chat_completion(
                                     delta = stream_q.get_nowait()
                                     if delta is None:
                                         break
    -                                content_chunk = {
    -                                    "id": completion_id, "object": "chat.completion.chunk",
    -                                    "created": created, "model": model,
    -                                    "choices": [{"index": 0, "delta": {"content": delta}, "finish_reason": None}],
    -                                }
    -                                await response.write(f"data: {json.dumps(content_chunk)}\n\n".encode())
    +                                await _emit(delta)
                                 except _q.Empty:
                                     break
                             break
    @@ -790,12 +828,7 @@ async def _write_sse_chat_completion(
                     if delta is None:  # End of stream sentinel
                         break
     
    -                content_chunk = {
    -                    "id": completion_id, "object": "chat.completion.chunk",
    -                    "created": created, "model": model,
    -                    "choices": [{"index": 0, "delta": {"content": delta}, "finish_reason": None}],
    -                }
    -                await response.write(f"data: {json.dumps(content_chunk)}\n\n".encode())
    +                await _emit(delta)
     
                 # Get usage from completed agent
                 usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}

  • tests/gateway/test_api_server.py

    @@ -464,7 +464,7 @@ async def _mock_run_agent(**kwargs):
     
         @pytest.mark.asyncio
         async def test_stream_includes_tool_progress(self, adapter):
    -        """tool_progress_callback fires → progress appears in the SSE stream."""
    +        """tool_progress_callback fires → progress appears as custom SSE event, not in delta.content."""
             import asyncio
     
             app = _create_app(adapter)
    @@ -495,8 +495,26 @@ async def _mock_run_agent(**kwargs):
                     assert resp.status == 200
                     body = await resp.text()
                     assert "[DONE]" in body
    -                # Tool progress message must appear in the stream
    -                assert "ls -la" in body
    +                # Tool progress must appear as a custom SSE event, not in
    +                # delta.content — prevents model from learning to imitate
    +                # markers instead of calling tools (#6972).
    +                assert "event: hermes.tool.progress" in body
    +                assert '"tool": "terminal"' in body
    +                assert '"label": "ls -la"' in body
    +                # The progress marker must NOT appear inside any
    +                # chat.completion.chunk delta.content field.
    +                import json as _json
    +                for line in body.splitlines():
    +                    if line.startswith("data: ") and line.strip() != "data: [DONE]":
    +                        try:
    +                            chunk = _json.loads(line[len("data: "):])
    +                        except _json.JSONDecodeError:
    +                            continue
    +                        if chunk.get("object") == "chat.completion.chunk":
    +                            for choice in chunk.get("choices", []):
    +                                content = choice.get("delta", {}).get("content", "")
    +                                # Tool emoji markers must never leak into content
    +                                assert "ls -la" not in content or content == "Here are the files."
                     # Final content must also be present
                     assert "Here are the files." in body
     
    @@ -532,10 +550,12 @@ async def _mock_run_agent(**kwargs):
                     )
                     assert resp.status == 200
                     body = await resp.text()
    -                # Internal _thinking event should NOT appear
    +                # Internal _thinking event should NOT appear anywhere
                     assert "some internal state" not in body
    -                # Real tool progress should appear
    -                assert "Python docs" in body
    +                # Real tool progress should appear as custom SSE event
    +                assert "event: hermes.tool.progress" in body
    +                assert '"tool": "web_search"' in body
    +                assert '"label": "Python docs"' in body
     
         @pytest.mark.asyncio
         async def test_no_user_message_returns_400(self, adapter):