feat(reporting): surface dependency chain in markdown, SARIF, and TUI outputs

pyproject.toml

@@ -236,8 +236,6 @@ 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_tool_call_limits.py" = ["N802", "SLF001"]
 "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

strix/config/models.py

@@ -4,10 +4,9 @@ from __future__ import annotations
 
 import contextlib
 import inspect
-import logging
 import os
 import time
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any
 
 from agents import (
     set_default_openai_api,
@@ -25,19 +24,12 @@ from agents.retry import (
     RetryPolicyContext,
     retry_policies,
 )
-from openai.types.responses import (
-    Response,
-    ResponseCompletedEvent,
-    ResponseOutputItemAddedEvent,
-    ResponseOutputItemDoneEvent,
-)
+from openai.types.responses import Response, ResponseCompletedEvent
 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
-from strix.config.tool_call_limits import TurnToolCallLimiter
 
 
 if TYPE_CHECKING:
@@ -56,9 +48,6 @@ if TYPE_CHECKING:
     from strix.config.settings import LlmSettings, ReasoningEffort, Settings
 
 
-logger = logging.getLogger(__name__)
-
-
 def request_timeout_extra_args(timeout_s: float | None) -> dict[str, float] | None:
     """Per-request model timeout; a plain float so ``ModelSettings.to_json_dict()`` stays serializable."""  # noqa: E501
     if not timeout_s or timeout_s <= 0:
@@ -240,130 +229,6 @@ class _NonStreamingModel(Model):
         yield _completed_stream_event(response, getattr(self._inner, "model", None))
 
 
-class _TurnGuardModel(Model):
-    """Keep one turn from corrupting the conversation or running away.
-
-    Tool-call ids: providers that number 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.
-
-    Tool-call volume: a degenerate response can queue hundreds of calls that
-    the run loop then honours one by one. Only the first
-    ``LLM_MAX_TOOL_CALLS_PER_TURN`` calls of a response are kept.
-    """
-
-    def __init__(self, inner: Model, *, max_tool_calls_per_turn: int = 0) -> None:
-        self._inner = inner
-        self._max_tool_calls_per_turn = max_tool_calls_per_turn
-
-    def _limiter(self) -> TurnToolCallLimiter:
-        return TurnToolCallLimiter(self._max_tool_calls_per_turn)
-
-    def _log_dropped(self, limiter: TurnToolCallLimiter) -> None:
-        if limiter.dropped:
-            logger.warning(
-                "dropped %d tool call(s) past the per-response limit of %d",
-                limiter.dropped,
-                self._max_tool_calls_per_turn,
-            )
-
-    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,
-        )
-        limiter = self._limiter()
-        response.output = limiter.filter_items(rewriter.rewrite_items(list(response.output)))
-        self._log_dropped(limiter)
-        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)
-        limiter = self._limiter()
-        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:
-            guarded = _guard_event(event, rewriter, limiter)
-            if guarded is not None:
-                yield guarded
-        self._log_dropped(limiter)
-
-
-def _guard_event(
-    event: TResponseStreamEvent, rewriter: TurnCallIdRewriter, limiter: TurnToolCallLimiter
-) -> TResponseStreamEvent | None:
-    if isinstance(event, ResponseOutputItemAddedEvent | ResponseOutputItemDoneEvent):
-        rewritten = rewriter.rewrite_item(event.item)
-        if not limiter.allow(rewritten):
-            return None
-        if rewritten is not event.item:
-            return event.model_copy(update={"item": rewritten})
-        return event
-    if isinstance(event, ResponseCompletedEvent):
-        original = list(event.response.output)
-        output = limiter.filter_items(rewriter.rewrite_items(original))
-        if output != original:
-            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:
@@ -433,16 +298,15 @@ class StrixProvider(MultiProvider):
             # The ChatGPT subscription backend is always streamed; it has no
             # non-streaming mode to fall back to, so LLM_DISABLE_STREAMING
             # does not apply here.
-            model: Model = _CodexResponsesModel(
+            return _CodexResponsesModel(
                 slug,
                 codex.get_subscription_client(),
                 reasoning_effort=llm.reasoning_effort,
             )
-        else:
-            model = super().get_model(model_name)
-            if llm.disable_streaming:
-                model = _NonStreamingModel(model)
-        return _TurnGuardModel(model, max_tool_calls_per_turn=llm.max_tool_calls_per_turn)
+        model = super().get_model(model_name)
+        if llm.disable_streaming:
+            return _NonStreamingModel(model)
+        return model
 
 
 DEFAULT_MODEL_RETRY = ModelRetrySettings(

strix/config/settings.py

@@ -57,11 +57,6 @@ class LlmSettings(BaseSettings):
         alias="LLM_DISABLE_STREAMING",
     )
     timeout: int = Field(default=300, alias="LLM_TIMEOUT")
-    max_tool_calls_per_turn: int = Field(
-        default=32,
-        ge=0,
-        alias="LLM_MAX_TOOL_CALLS_PER_TURN",
-    )
 
 
 class DedupeSettings(BaseSettings):

strix/config/tool_call_ids.py

@@ -1,117 +0,0 @@
-"""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]

strix/config/tool_call_limits.py

@@ -1,46 +0,0 @@
-"""Bound how many tool calls one assistant response may queue.
-
-A degenerate generation can emit hundreds or thousands of tool calls in a
-single response — typically a poll/wait loop the model writes out ahead of
-time instead of issuing one call and yielding. The run loop honours all of
-them, so the agent stops reacting to anything for hours. Keeping only the
-first ``limit`` calls of a response bounds that blast radius; the model sees
-their results on the next turn and can reconsider.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from openai.types.responses import ResponseFunctionToolCall
-
-
-class TurnToolCallLimiter:
-    """Decide, once per call, whether a turn's tool call is within the limit."""
-
-    def __init__(self, limit: int) -> None:
-        self._limit = limit
-        self._decisions: dict[str, bool] = {}
-        self._kept = 0
-        self.dropped = 0
-
-    @property
-    def enabled(self) -> bool:
-        return self._limit > 0
-
-    def allow(self, item: Any) -> bool:
-        if not self.enabled or not isinstance(item, ResponseFunctionToolCall):
-            return True
-        decided = self._decisions.get(item.call_id)
-        if decided is not None:
-            return decided
-        allowed = self._kept < self._limit
-        if allowed:
-            self._kept += 1
-        else:
-            self.dropped += 1
-        self._decisions[item.call_id] = allowed
-        return allowed
-
-    def filter_items(self, items: list[Any]) -> list[Any]:
-        return [item for item in items if self.allow(item)]

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, _TurnGuardModel
+from strix.config.models import StrixProvider, _NonStreamingModel
 
 
 if TYPE_CHECKING:
@@ -299,11 +299,10 @@ def test_get_model_wraps_when_disabled(
     load_settings()
 
     model = StrixProvider().get_model("openai/gpt-4o-mini")
-    assert isinstance(model, _TurnGuardModel)
-    assert isinstance(model._inner, _NonStreamingModel)
+    assert isinstance(model, _NonStreamingModel)
 
 
-def test_get_model_keeps_streaming_by_default(
+def test_get_model_unwrapped_by_default(
     monkeypatch: pytest.MonkeyPatch, _reset_settings: None
 ) -> None:
     inner = _DummyModel()
@@ -311,20 +310,17 @@ def test_get_model_keeps_streaming_by_default(
     load_settings()
 
     model = StrixProvider().get_model("openai/gpt-4o-mini")
-    assert isinstance(model, _TurnGuardModel)
-    assert model._inner is inner
+    assert model is inner
 
 
-def test_get_model_guards_subscription_model_but_keeps_it_streaming(
+def test_get_model_does_not_wrap_subscription_model(
     monkeypatch: pytest.MonkeyPatch, _reset_settings: None
 ) -> None:
-    # Subscription (ChatGPT) models are always streamed, so LLM_DISABLE_STREAMING
-    # must not apply — but a runaway response needs capping there too.
+    # Subscription (ChatGPT) models are always streamed and must not be wrapped.
     monkeypatch.setattr(codex, "subscription_model", lambda *_: "gpt-5.5")
     monkeypatch.setattr(codex, "get_subscription_client", lambda: AsyncOpenAI(api_key="x"))
     monkeypatch.setenv("LLM_DISABLE_STREAMING", "true")
     load_settings()
 
     model = StrixProvider().get_model("gpt-5.5")
-    assert isinstance(model, _TurnGuardModel)
-    assert not isinstance(model._inner, _NonStreamingModel)
+    assert not isinstance(model, _NonStreamingModel)

tests/test_tool_call_ids.py

@@ -1,249 +0,0 @@
-"""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, _TurnGuardModel
-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 _TurnGuardModel(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

tests/test_tool_call_limits.py

@@ -1,189 +0,0 @@
-"""Tests for the per-response tool-call cap.
-
-A degenerate generation can emit hundreds of tool calls in one assistant
-response — a wait/poll loop the model writes out ahead of time. The run loop
-honours every one of them, so the agent stops reacting for hours. The cap
-keeps the first N calls of a response and drops the tail.
-"""
-
-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 strix.config import loader
-from strix.config.loader import load_settings
-from strix.config.models import StrixProvider, _NonStreamingModel, _TurnGuardModel
-
-
-if TYPE_CHECKING:
-    from collections.abc import Iterator
-
-
-_RUNAWAY_CALLS = 200
-_CAP = 32
-
-
-def _runaway_completion() -> 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": f"call_{i}",
-                            "type": "function",
-                            "function": {"name": "wait_for_message", "arguments": "{}"},
-                        }
-                        for i in range(_RUNAWAY_CALLS)
-                    ],
-                },
-            }
-        ],
-        "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7},
-    }
-
-
-def _text_completion() -> 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": "done"},
-            }
-        ],
-        "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8},
-    }
-
-
-_TURNS: list[int] = []
-
-
-class _RunawayHandler(BaseHTTPRequestHandler):
-    """First turn queues a huge poll loop; the next turn ends the run."""
-
-    def log_message(self, *args: Any) -> None:
-        pass
-
-    def do_POST(self) -> None:
-        length = int(self.headers.get("Content-Length", 0))
-        self.rfile.read(length)
-        _TURNS.append(1)
-        payload = _runaway_completion() if len(_TURNS) == 1 else _text_completion()
-        encoded = json.dumps(payload).encode()
-        self.send_response(200)
-        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 runaway_gateway() -> Iterator[str]:
-    _TURNS.clear()
-    server = HTTPServer(("127.0.0.1", 0), _RunawayHandler)
-    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:
-    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, *, cap: int) -> list[int]:
-    executed: list[int] = []
-
-    @function_tool
-    def wait_for_message() -> str:
-        executed.append(1)
-        return "nothing new"
-
-    class _Provider(ModelProvider):
-        def get_model(self, model_name: str | None) -> Model:  # noqa: ARG002
-            return _TurnGuardModel(_model(base_url), max_tool_calls_per_turn=cap)
-
-    agent = Agent(name="t", instructions="orchestrate", tools=[wait_for_message], model="gw-model")
-    result = Runner.run_streamed(
-        agent, input="go", run_config=RunConfig(model_provider=_Provider())
-    )
-    async for _ in result.stream_events():
-        pass
-    assert result.final_output == "done"
-    return executed
-
-
-@pytest.mark.asyncio
-async def test_runaway_response_runs_every_queued_call_when_uncapped(runaway_gateway: str) -> None:
-    # Repro: one response queues 200 calls and the run loop honours all of them.
-    executed = await _run_agent(runaway_gateway, cap=0)
-
-    assert len(executed) == _RUNAWAY_CALLS
-
-
-@pytest.mark.asyncio
-async def test_runaway_response_is_capped(runaway_gateway: str) -> None:
-    executed = await _run_agent(runaway_gateway, cap=_CAP)
-
-    assert len(executed) == _CAP
-
-
-@pytest.mark.asyncio
-async def test_response_below_the_cap_is_untouched(runaway_gateway: str) -> None:
-    executed = await _run_agent(runaway_gateway, cap=_RUNAWAY_CALLS + 1)
-
-    assert len(executed) == _RUNAWAY_CALLS
-
-
-@pytest.fixture
-def _reset_settings(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
-    for key in ("STRIX_LLM", "LLM_DISABLE_STREAMING", "LLM_MAX_TOOL_CALLS_PER_TURN"):
-        monkeypatch.delenv(key, raising=False)
-    monkeypatch.setattr(loader, "_cached", None)
-    monkeypatch.setattr(loader, "_override", None)
-    yield
-
-
-class _DummyModel(Model):
-    async def get_response(self, *args: Any, **kwargs: Any) -> Any:
-        raise NotImplementedError
-
-    def stream_response(self, *args: Any, **kwargs: Any) -> Any:
-        raise NotImplementedError
-
-
-def test_cap_is_configurable(monkeypatch: pytest.MonkeyPatch, _reset_settings: None) -> None:
-    monkeypatch.setattr("strix.config.models.MultiProvider.get_model", lambda *_: _DummyModel())
-    monkeypatch.setenv("LLM_MAX_TOOL_CALLS_PER_TURN", "7")
-    load_settings()
-
-    model = StrixProvider().get_model("openai/gpt-4o-mini")
-    assert isinstance(model, _TurnGuardModel)
-    assert model._max_tool_calls_per_turn == 7