diff --git a/agent-langgraph-agui/src/parallel_tools.py b/agent-langgraph-agui/src/parallel_tools.py index 7c737b4b6..49aad2e0a 100644 --- a/agent-langgraph-agui/src/parallel_tools.py +++ b/agent-langgraph-agui/src/parallel_tools.py @@ -45,6 +45,7 @@ async def run(self, input): # for explicit reuse and when an HTTP consumer cancels/closes its run. streams = {} self._tool_streams = streams + self._response_message_ids = {} self._upstream_stream = None self._graph_stream = None try: @@ -66,6 +67,7 @@ async def run(self, input): await self._graph_stream.aclose() finally: streams.clear() + self._response_message_ids.clear() def _handle_stream_events(self, input): # Upstream run uses a bare async-for. Keep its generator so closing @@ -82,7 +84,25 @@ async def _handle_single_event(self, event, state): active_run = self.active_run kind = event.get("event") model_key = (self._current_lane(), event.get("run_id")) + if kind == "on_chat_model_stream": + chunk = event.get("data", {}).get("chunk") + # Responses API announces its durable ID in a metadata-only chunk. + # Later chunks get lc_run IDs from LangChain; emitting those creates + # a second copy when the final snapshot restores the provider ID. + metadata = _get(chunk, "response_metadata", {}) or {} + response_id = metadata.get("id") + if response_id and _get(chunk, "id") == response_id: + self._response_message_ids.setdefault(model_key, response_id) + stable_id = self._response_message_ids.get(model_key) + if stable_id and _get(chunk, "id") != stable_id: + clean = ( + {**chunk, "id": stable_id} + if isinstance(chunk, dict) + else chunk.model_copy(update={"id": stable_id}) + ) + event = {**event, "data": {**event["data"], "chunk": clean}} if kind == "on_chat_model_end": + self._response_message_ids.pop(model_key, None) saved = self.get_message_in_progress(self.active_run["id"]) try: for slot in self._tool_streams.pop(model_key, {}).values(): diff --git a/agent-langgraph-agui/tests/test_tool_protocol.py b/agent-langgraph-agui/tests/test_tool_protocol.py index 70ffb00bd..b061e3bd3 100644 --- a/agent-langgraph-agui/tests/test_tool_protocol.py +++ b/agent-langgraph-agui/tests/test_tool_protocol.py @@ -603,3 +603,12 @@ async def send(_client, request, **_kwargs): assert all("public marker 43" in part["output"] for part in results) assert "Both client results: public marker 43" in json.dumps(snapshot(second)) assert "synthetic-access" not in json.dumps(first + second) + # Streamed owners and final history must describe the same messages. A + # Responses API metadata-only chunk has the provider ID before text/tools. + for events in (first, second): + final_ids = {message["id"] for message in snapshot(events)} + for event in events: + if event["type"] == "TEXT_MESSAGE_START": + assert event["messageId"] in final_ids + elif event["type"] == "TOOL_CALL_START": + assert event["parentMessageId"] in final_ids