From 68ea6fca653153677979e55fda6e4d06d281d31b Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 6 Aug 2026 00:06:46 +0300 Subject: [PATCH] fix(llm): keep tool-call ids unique so a recycled id can't erase history (#976) Co-authored-by: Ahmed Allam --- pyproject.toml | 1 + strix/config/models.py | 113 ++++++++++++++- strix/config/tool_call_ids.py | 117 +++++++++++++++ tests/test_disable_streaming.py | 10 +- tests/test_tool_call_ids.py | 249 ++++++++++++++++++++++++++++++++ 5 files changed, 482 insertions(+), 8 deletions(-) create mode 100644 strix/config/tool_call_ids.py create mode 100644 tests/test_tool_call_ids.py diff --git a/pyproject.toml b/pyproject.toml index 38d5154c..d264dcca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -236,6 +236,7 @@ ignore = [ "strix/interface/auth_cli.py" = ["N802"] "tests/test_codex_streaming.py" = ["N802"] "tests/test_disable_streaming.py" = ["N802"] +"tests/test_tool_call_ids.py" = ["N802"] "tests/test_unknown_tool_recovery.py" = ["N802"] "tests/test_report_pdf.py" = ["S105", "S106"] # Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a diff --git a/strix/config/models.py b/strix/config/models.py index a3bb481a..8685a2fb 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -6,7 +6,7 @@ import contextlib import inspect import os import time -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from agents import ( set_default_openai_api, @@ -24,12 +24,18 @@ from agents.retry import ( RetryPolicyContext, retry_policies, ) -from openai.types.responses import Response, ResponseCompletedEvent +from openai.types.responses import ( + Response, + ResponseCompletedEvent, + ResponseOutputItemAddedEvent, + ResponseOutputItemDoneEvent, +) from openai.types.responses.response_usage import ResponseUsage from openai.types.shared import Reasoning from strix.config import codex from strix.config.loader import load_settings +from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input if TYPE_CHECKING: @@ -229,6 +235,105 @@ class _NonStreamingModel(Model): yield _completed_stream_event(response, getattr(self._inner, "model", None)) +class _UniqueToolCallIdModel(Model): + """Keep tool-call ids unique so a recycled id can't invalidate the history. + + Providers that number tool calls per turn (``exec_command:0``, ...) restart + the counter each turn, so the same id eventually appears twice in one + conversation and strict providers reject every subsequent request. Ids that + collide with the history are rewritten before the turn is recorded, and + already-corrupted histories are repaired on the way out. + """ + + def __init__(self, inner: Model) -> None: + self._inner = inner + + async def close(self) -> None: + await self._inner.close() + + def get_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None: + return self._inner.get_retry_advice(request) + + async def get_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + *, + previous_response_id: str | None, + conversation_id: str | None, + prompt: ResponsePromptParam | None, + ) -> ModelResponse: + sanitized = dedupe_input(input) + rewriter = TurnCallIdRewriter(sanitized) + response = await self._inner.get_response( + system_instructions, + cast("str | list[TResponseInputItem]", sanitized), + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + prompt=prompt, + ) + response.output = rewriter.rewrite_items(list(response.output)) + return response + + async def stream_response( + self, + system_instructions: str | None, + input: str | list[TResponseInputItem], # noqa: A002 + model_settings: ModelSettings, + tools: list[Tool], + output_schema: AgentOutputSchemaBase | None, + handoffs: list[Handoff], + tracing: ModelTracing, + *, + previous_response_id: str | None, + conversation_id: str | None, + prompt: ResponsePromptParam | None, + ) -> AsyncIterator[TResponseStreamEvent]: + sanitized = dedupe_input(input) + rewriter = TurnCallIdRewriter(sanitized) + stream = self._inner.stream_response( + system_instructions, + cast("str | list[TResponseInputItem]", sanitized), + model_settings, + tools, + output_schema, + handoffs, + tracing, + previous_response_id=previous_response_id, + conversation_id=conversation_id, + prompt=prompt, + ) + async for event in stream: + yield _rewrite_event_call_ids(event, rewriter) + + +def _rewrite_event_call_ids( + event: TResponseStreamEvent, rewriter: TurnCallIdRewriter +) -> TResponseStreamEvent: + if isinstance(event, ResponseOutputItemAddedEvent | ResponseOutputItemDoneEvent): + rewritten = rewriter.rewrite_item(event.item) + if rewritten is not event.item: + return event.model_copy(update={"item": rewritten}) + return event + if isinstance(event, ResponseCompletedEvent): + output = rewriter.rewrite_items(list(event.response.output)) + if output != list(event.response.output): + return event.model_copy( + update={"response": event.response.model_copy(update={"output": output})} + ) + return event + + def _completed_stream_event( model_response: ModelResponse, model_name: object | None ) -> TResponseStreamEvent: @@ -305,8 +410,8 @@ class StrixProvider(MultiProvider): ) model = super().get_model(model_name) if llm.disable_streaming: - return _NonStreamingModel(model) - return model + model = _NonStreamingModel(model) + return _UniqueToolCallIdModel(model) DEFAULT_MODEL_RETRY = ModelRetrySettings( diff --git a/strix/config/tool_call_ids.py b/strix/config/tool_call_ids.py new file mode 100644 index 00000000..31f78850 --- /dev/null +++ b/strix/config/tool_call_ids.py @@ -0,0 +1,117 @@ +"""Keep tool-call ids unique within a conversation. + +Some providers return per-turn tool-call ids (``exec_command:0``, +``exec_command:1``, ...) whose counter restarts on every turn. Once the same +id appears twice in one conversation, the request payload has two assistant +tool calls sharing an id and strict providers reject the whole turn, which +permanently kills the agent because the malformed history is replayed on +every retry. Rewriting duplicates to fresh unique ids keeps the history +valid for any provider. +""" + +from __future__ import annotations + +from collections import defaultdict, deque +from typing import Any +from uuid import uuid4 + +from openai.types.responses import ResponseFunctionToolCall + + +def new_call_id() -> str: + return f"call_{uuid4().hex}" + + +def collect_call_ids(items: list[Any]) -> set[str]: + used: set[str] = set() + for item in items: + if isinstance(item, dict): + call_id = item.get("call_id") + if isinstance(call_id, str): + used.add(call_id) + elif isinstance(item, ResponseFunctionToolCall): + used.add(item.call_id) + return used + + +def dedupe_history_call_ids(items: list[Any]) -> tuple[list[Any], bool]: + """Rewrite duplicate call ids in a conversation history. + + Outputs are paired with their call by order, so parallel calls that share + an id keep answering the right call after the rewrite. + """ + used: set[str] = set() + pending: dict[str, deque[str]] = defaultdict(deque) + rebuilt: list[Any] = [] + changed = False + + for item in items: + if not isinstance(item, dict): + rebuilt.append(item) + continue + call_id = item.get("call_id") + if not isinstance(call_id, str): + rebuilt.append(item) + continue + + kind = item.get("type") + if kind == "function_call": + effective = call_id + if call_id in used: + effective = new_call_id() + item = {**item, "call_id": effective} # noqa: PLW2901 + changed = True + used.add(effective) + pending[call_id].append(effective) + elif kind == "function_call_output": + queue = pending.get(call_id) + if queue: + effective = queue.popleft() + if effective != call_id: + item = {**item, "call_id": effective} # noqa: PLW2901 + changed = True + rebuilt.append(item) + + return rebuilt, changed + + +def dedupe_input(model_input: str | list[Any]) -> str | list[Any]: + if isinstance(model_input, str): + return model_input + rebuilt, changed = dedupe_history_call_ids(model_input) + return rebuilt if changed else model_input + + +class TurnCallIdRewriter: + """Rewrite a single turn's tool-call ids that collide with the history. + + A turn's items surface several times (streamed item events, then the + completed response), so the same original id must always map to the same + replacement within the turn. + """ + + def __init__(self, model_input: str | list[Any]) -> None: + self._used = set() if isinstance(model_input, str) else collect_call_ids(model_input) + self._remap: dict[str, str] = {} + self._settled: set[str] = set() + + def rewrite_item(self, item: Any) -> Any: + if not isinstance(item, ResponseFunctionToolCall): + return item + original = item.call_id + if original in self._settled: + return item + replacement = self._remap.get(original) + if replacement is None: + if original not in self._used: + self._used.add(original) + self._settled.add(original) + return item + replacement = new_call_id() + self._remap[original] = replacement + self._used.add(replacement) + self._settled.add(replacement) + return item.model_copy(update={"call_id": replacement}) + + def rewrite_items(self, items: list[Any]) -> list[Any]: + return [self.rewrite_item(item) for item in items] diff --git a/tests/test_disable_streaming.py b/tests/test_disable_streaming.py index 1d667e61..d8438a54 100644 --- a/tests/test_disable_streaming.py +++ b/tests/test_disable_streaming.py @@ -31,7 +31,7 @@ from openai.types.responses import ( from strix.config import codex, loader from strix.config.loader import load_settings -from strix.config.models import StrixProvider, _NonStreamingModel +from strix.config.models import StrixProvider, _NonStreamingModel, _UniqueToolCallIdModel if TYPE_CHECKING: @@ -299,10 +299,11 @@ def test_get_model_wraps_when_disabled( load_settings() model = StrixProvider().get_model("openai/gpt-4o-mini") - assert isinstance(model, _NonStreamingModel) + assert isinstance(model, _UniqueToolCallIdModel) + assert isinstance(model._inner, _NonStreamingModel) -def test_get_model_unwrapped_by_default( +def test_get_model_keeps_streaming_by_default( monkeypatch: pytest.MonkeyPatch, _reset_settings: None ) -> None: inner = _DummyModel() @@ -310,7 +311,8 @@ def test_get_model_unwrapped_by_default( load_settings() model = StrixProvider().get_model("openai/gpt-4o-mini") - assert model is inner + assert isinstance(model, _UniqueToolCallIdModel) + assert model._inner is inner def test_get_model_does_not_wrap_subscription_model( diff --git a/tests/test_tool_call_ids.py b/tests/test_tool_call_ids.py new file mode 100644 index 00000000..a4033f4b --- /dev/null +++ b/tests/test_tool_call_ids.py @@ -0,0 +1,249 @@ +"""Tests for tool-call id uniqueness. + +Providers that number tool calls per turn (``exec_command:0``, ``:1``, ...) +restart the counter on every turn, so the same id eventually appears twice in +one conversation. Strict providers then reject the whole request, and because +the history is replayed on every retry the agent can never recover. A gateway +that validates id uniqueness the way those providers do proves both the +failure and the fix. +""" + +from __future__ import annotations + +import json +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import TYPE_CHECKING, Any + +import pytest +from agents import Agent, Runner, function_tool +from agents.models.interface import Model, ModelProvider +from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel +from agents.run import RunConfig +from openai import AsyncOpenAI +from openai.types.responses import ResponseFunctionToolCall + +from strix.config.models import _NonStreamingModel, _UniqueToolCallIdModel +from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_history_call_ids + + +if TYPE_CHECKING: + from collections.abc import Iterator + + +def _tool_call_completion(call_id: str, n: int = 1) -> dict[str, Any]: + return { + "id": "chatcmpl-1", + "object": "chat.completion", + "created": 0, + "model": "gw-model", + "choices": [ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": call_id, + "type": "function", + "function": {"name": "do_thing", "arguments": json.dumps({"n": n})}, + } + ], + }, + } + ], + "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7}, + } + + +def _text_completion(text: str) -> dict[str, Any]: + return { + "id": "chatcmpl-2", + "object": "chat.completion", + "created": 0, + "model": "gw-model", + "choices": [ + {"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": text}} + ], + "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}, + } + + +_REQUESTS: list[list[dict[str, Any]]] = [] + + +def _assistant_call_ids(messages: list[dict[str, Any]]) -> list[str]: + return [str(call.get("id")) for message in messages for call in message.get("tool_calls") or []] + + +def _tool_results(messages: list[dict[str, Any]]) -> list[str]: + return [str(m.get("content")) for m in messages if m.get("role") == "tool"] + + +class _StrictHandler(BaseHTTPRequestHandler): + """Gateway that rejects a history reusing a tool-call id, like strict providers do.""" + + def log_message(self, *args: Any) -> None: + pass + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length) or b"{}") + messages = body.get("messages", []) + _REQUESTS.append(messages) + call_ids = _assistant_call_ids(messages) + + if len(call_ids) != len(set(call_ids)): + self._respond( + 400, + { + "error": { + "message": ( + "tool messages need a resolvable tool name: carry `tool`/`name`, " + "or match a preceding assistant tool_call by order" + ) + } + }, + ) + return + + turn = len(_REQUESTS) + if turn <= 2: + # The provider restarts its per-turn counter, so both turns say ":0". + self._respond(200, _tool_call_completion("exec_command:0", n=turn)) + else: + self._respond(200, _text_completion("all done")) + + def _respond(self, status: int, payload: dict[str, Any]) -> None: + encoded = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + +@pytest.fixture +def strict_gateway() -> Iterator[str]: + _REQUESTS.clear() + server = HTTPServer(("127.0.0.1", 0), _StrictHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}/v1" + finally: + server.shutdown() + server.server_close() + + +def _model(base_url: str) -> Model: + # The gateway answers plain JSON, so the run loop's streamed turns are + # served non-streamed; the ids on the wire are the same either way. + client = AsyncOpenAI(api_key="tok", base_url=base_url, max_retries=0) + return _NonStreamingModel(OpenAIChatCompletionsModel(model="gw-model", openai_client=client)) + + +async def _run_agent(base_url: str, *, wrap: bool) -> Any: + @function_tool + def do_thing(n: int) -> str: + return f"did {n}" + + class _Provider(ModelProvider): + def get_model(self, model_name: str | None) -> Model: # noqa: ARG002 + model = _model(base_url) + return _UniqueToolCallIdModel(model) if wrap else model + + agent = Agent(name="t", instructions="use the tool", tools=[do_thing], model="gw-model") + result = Runner.run_streamed( + agent, input="please", run_config=RunConfig(model_provider=_Provider()) + ) + async for _ in result.stream_events(): + pass + return result + + +@pytest.mark.asyncio +async def test_recycled_call_id_erases_a_turn_without_the_wrapper(strict_gateway: str) -> None: + # Repro: two turns run a tool and both are labelled ``exec_command:0``, so + # the colliding call and its result are dropped as duplicates. The agent + # ends the run having silently lost a turn of its own work — and a provider + # that does not drop them instead rejects the malformed history outright. + result = await _run_agent(strict_gateway, wrap=False) + + assert result.final_output == "all done" + assert _assistant_call_ids(_REQUESTS[-1]) == ["exec_command:0"] + assert _tool_results(_REQUESTS[-1]) == ["did 2"] + + +@pytest.mark.asyncio +async def test_recycled_call_id_is_rewritten_so_no_turn_is_lost(strict_gateway: str) -> None: + result = await _run_agent(strict_gateway, wrap=True) + + assert result.final_output == "all done" + call_ids = _assistant_call_ids(_REQUESTS[-1]) + assert len(call_ids) == len(set(call_ids)) == 2 + assert call_ids[0] == "exec_command:0" + assert call_ids[1].startswith("call_") + assert _tool_results(_REQUESTS[-1]) == ["did 1", "did 2"] + + +def test_history_dedupe_keeps_outputs_paired_with_their_call() -> None: + items = [ + {"type": "function_call", "call_id": "exec_command:0", "name": "a", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "exec_command:0", "output": "first"}, + {"type": "function_call", "call_id": "exec_command:0", "name": "b", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "exec_command:0", "output": "second"}, + ] + + rebuilt, changed = dedupe_history_call_ids(items) + + assert changed + ids = [item["call_id"] for item in rebuilt] + assert ids[0] == ids[1] == "exec_command:0" + assert ids[2] == ids[3] != "exec_command:0" + assert rebuilt[3]["output"] == "second" + + +def test_history_dedupe_pairs_parallel_calls_by_order() -> None: + items = [ + {"type": "function_call", "call_id": "dup", "name": "a", "arguments": "{}"}, + {"type": "function_call", "call_id": "dup", "name": "b", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "dup", "output": "for-a"}, + {"type": "function_call_output", "call_id": "dup", "output": "for-b"}, + ] + + rebuilt, changed = dedupe_history_call_ids(items) + + assert changed + assert rebuilt[0]["call_id"] == rebuilt[2]["call_id"] == "dup" + assert rebuilt[1]["call_id"] == rebuilt[3]["call_id"] + assert rebuilt[1]["call_id"] != "dup" + + +def test_history_dedupe_leaves_unique_ids_alone() -> None: + items = [ + {"type": "function_call", "call_id": "call_a", "name": "a", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_a", "output": "x"}, + {"type": "function_call", "call_id": "call_b", "name": "b", "arguments": "{}"}, + ] + + rebuilt, changed = dedupe_history_call_ids(items) + + assert not changed + assert rebuilt == items + + +def test_turn_rewriter_is_stable_across_repeated_sightings() -> None: + history = [{"type": "function_call", "call_id": "exec_command:0", "name": "a"}] + rewriter = TurnCallIdRewriter(history) + call = ResponseFunctionToolCall( + call_id="exec_command:0", name="a", arguments="{}", type="function_call" + ) + + first = rewriter.rewrite_item(call) + second = rewriter.rewrite_item(first) + + assert first.call_id != "exec_command:0" + assert second.call_id == first.call_id