fix(llm): keep tool-call ids unique so a recycled id can't erase history (#976)

Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>

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(

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]