diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index a56be55736a0..1c4cb6df728b 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -1605,6 +1605,7 @@ class APIServerAdapter(BasePlatformAdapter): ) final_response = result.get("final_response", "") if isinstance(result, dict) else "" effective_session_id = result.get("session_id", session_id) if isinstance(result, dict) else session_id + turn_messages = self._turn_transcript_messages(history, user_message, result) if isinstance(result, dict) else [] await queue.put(_event_payload("assistant.completed", { "session_id": effective_session_id, "message_id": message_id, @@ -1617,6 +1618,7 @@ class APIServerAdapter(BasePlatformAdapter): "session_id": effective_session_id, "message_id": message_id, "completed": True, + "messages": turn_messages, "usage": usage, })) except Exception as exc: @@ -3329,6 +3331,44 @@ class APIServerAdapter(BasePlatformAdapter): return len(prior) return 0 + @classmethod + def _turn_transcript_messages( + cls, + conversation_history: List[Dict[str, Any]], + user_message: Any, + result: Dict[str, Any], + ) -> List[Dict[str, Any]]: + """Return this turn's assistant/tool messages in client-safe shape. + + The streaming SSE contract delivers all assistant text as + ``assistant.delta`` events under one ``message_id`` interleaved with + ``tool.*`` events, and a single ``assistant.completed`` carrying only + the final reply. A client that accumulates deltas into one buffer + cannot reconstruct *intermediate* assistant text segments that preceded + tool calls — so when the page is re-opened mid/post-stream those + segments appear lost, even though state.db persisted them correctly. + + Emitting the authoritative per-turn transcript on ``run.completed`` lets + any SSE consumer reconcile its live view against ground truth without a + separate ``GET /messages`` round-trip. Purely additive: clients that + ignore the field are unaffected. Refs #34703. + """ + agent_messages = result.get("messages") if isinstance(result, dict) else None + if not isinstance(agent_messages, list) or not agent_messages: + return [] + start = cls._response_messages_turn_start_index( + conversation_history, user_message, result + ) + turn = agent_messages[start:] + out: List[Dict[str, Any]] = [] + for msg in turn: + if not isinstance(msg, dict): + continue + if msg.get("role") not in {"assistant", "tool"}: + continue + out.append(cls._message_response(msg)) + return out + @staticmethod def _extract_output_items(result: Dict[str, Any], start_index: int = 0) -> List[Dict[str, Any]]: """ diff --git a/tests/gateway/test_session_api.py b/tests/gateway/test_session_api.py index 5b06ffd55ce6..d5262e9aecb0 100644 --- a/tests/gateway/test_session_api.py +++ b/tests/gateway/test_session_api.py @@ -268,6 +268,75 @@ async def test_session_chat_stream_emits_lifecycle_events_and_keepalive_safe_sha assert "event: done" in body +@pytest.mark.asyncio +async def test_session_chat_stream_run_completed_carries_turn_transcript(adapter, session_db): + """run.completed must include the full interleaved turn transcript so a + client that lost intermediate (pre-tool-call) assistant text from the live + delta stream can reconcile without a separate /messages fetch. Refs #34703. + """ + import json as _json + + session_id = session_db.create_session("transcript-session", "api_server") + + async def fake_run(**kwargs): + # Stream the intermediate planning text the way a real turn would. + kwargs["stream_delta_callback"]("Let me search for that:") + kwargs["stream_delta_callback"]("Here is the summary.") + result = { + "final_response": "Here is the summary.", + "session_id": session_id, + "messages": [ + {"role": "user", "content": "search then summarize"}, + { + "role": "assistant", + "content": "Let me search for that:", + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "web_search", "arguments": "{}"}, + } + ], + }, + {"role": "tool", "content": "results", "tool_call_id": "call_1", "tool_name": "web_search"}, + {"role": "assistant", "content": "Here is the summary."}, + ], + } + return result, {"total_tokens": 6} + + app = _create_session_app(adapter) + with patch.object(adapter, "_run_agent", side_effect=fake_run): + async with TestClient(TestServer(app)) as cli: + resp = await cli.post( + f"/api/sessions/{session_id}/chat/stream", + json={"message": "search then summarize"}, + ) + assert resp.status == 200 + body = await resp.text() + + # Pull the run.completed event payload out of the SSE body. + run_completed_payload = None + for block in body.split("\n\n"): + if "event: run.completed" in block: + for line in block.splitlines(): + if line.startswith("data: "): + run_completed_payload = _json.loads(line[len("data: "):]) + break + assert run_completed_payload is not None, body + messages = run_completed_payload.get("messages") + assert isinstance(messages, list) and messages, run_completed_payload + + # The colon-ended intermediate text that preceded the tool call must be present. + contents = [m.get("content") for m in messages] + assert "Let me search for that:" in contents + assert "Here is the summary." in contents + # No prior-turn user message should leak into the per-turn slice. + assert all(m.get("role") in ("assistant", "tool") for m in messages) + # The tool call is preserved alongside the intermediate text. + assert any(m.get("tool_calls") for m in messages) + + + @pytest.mark.asyncio async def test_session_endpoints_require_auth_when_key_configured(auth_adapter): app = _create_session_app(auth_adapter)