fix(llm): cap the subscription backend's responses too

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

pyproject.toml

@@ -236,6 +236,8 @@ 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,9 +4,10 @@ from __future__ import annotations
 
 import contextlib
 import inspect
+import logging
 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 +25,19 @@ 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
+from strix.config.tool_call_limits import TurnToolCallLimiter
 
 
 if TYPE_CHECKING:
@@ -48,6 +56,9 @@ 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:
@@ -229,6 +240,130 @@ 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:
@@ -298,15 +433,16 @@ 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.
-            return _CodexResponsesModel(
+            model: Model = _CodexResponsesModel(
                 slug,
                 codex.get_subscription_client(),
                 reasoning_effort=llm.reasoning_effort,
             )
-        model = super().get_model(model_name)
-        if llm.disable_streaming:
-            return _NonStreamingModel(model)
-        return model
+        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)
 
 
 DEFAULT_MODEL_RETRY = ModelRetrySettings(

strix/config/settings.py

@@ -57,6 +57,11 @@ 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

@@ -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]

strix/config/tool_call_limits.py

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

strix/interface/tui/internal/app/model_test.go

@@ -1169,3 +1169,120 @@ func TestChatContentRerendersOnWidthAndExpansionChange(t *testing.T) {
 		}
 	}
 }
+
+// A model or backend failure can be a wrapped exception hundreds of columns
+// wide and several lines long. The status row is one line of the chat column, so
+// an oversized one widens the whole column - JoinHorizontal pads every row to the
+// widest - which pushed the sidebar off screen and wrapped the frame.
+func TestLongErrorDoesNotBreakTheFrame(t *testing.T) {
+	model := New(nil)
+	model.width, model.height = 120, 24
+	model.showSplash = false
+	model.handleEnvelope(stateEnvelope(t, 1, protocol.Snapshot{ScanState: "running"}))
+	bootstrap := protocol.CollectionBootstrap{
+		Collection: "agents", Revision: 1, Cursor: 0, NextCursor: 1, Done: true,
+		Items: []json.RawMessage{rawJSON(t, protocol.Agent{ID: "a0", Name: "Strix", Status: "running"})},
+	}
+	model.handleEnvelope(protocol.Envelope{
+		Version: protocol.Version, Type: "collection_bootstrap", Payload: rawJSON(t, bootstrap),
+	})
+	model.errorText = "litellm.APIConnectionError: OpenrouterException - Connection error " +
+		"while calling https://openrouter.ai/api/v1/chat/completions: HTTPSConnectionPool" +
+		"(host='openrouter.ai', port=443): Max retries exceeded\nTraceback (most recent " +
+		"call last):\n  File \"/x/y.py\", line 42, in send\n    raise err"
+	model.resizeViewport()
+
+	lines := strings.Split(model.View(), "\n")
+	if len(lines) > model.height {
+		t.Fatalf("frame is %d rows in a %d-row terminal", len(lines), model.height)
+	}
+	for i, line := range lines {
+		if width := ansi.StringWidth(line); width > model.width {
+			t.Fatalf("row %d is %d columns in a %d-column terminal", i, width, model.width)
+		}
+	}
+	// The sidebar has to survive: its panels are the right edge of the frame.
+	if !strings.Contains(ansi.Strip(model.View()), "Strix") {
+		t.Fatal("the agent tree was pushed out of the frame")
+	}
+}
+
+func TestStatusMessageFlattensAndKeepsItsHint(t *testing.T) {
+	row := ansi.Strip(statusMessage("boom\nsecond line\twith tabs", red, " · Send message to resume", 60))
+
+	if strings.Contains(row, "\n") || strings.Contains(row, "\t") {
+		t.Fatalf("status row is not a single line: %q", row)
+	}
+	if !strings.HasSuffix(row, " · Send message to resume") {
+		t.Fatalf("the hint was lost: %q", row)
+	}
+	if !strings.Contains(row, "boom second line with tabs") {
+		t.Fatalf("the message was mangled: %q", row)
+	}
+	// A message far too long for the row keeps the hint readable.
+	long := ansi.Strip(statusMessage(strings.Repeat("x", 500), red, " · Send message to resume", 60))
+	if width := ansi.StringWidth(long); width > 60 {
+		t.Fatalf("status message is %d columns, want at most 60", width)
+	}
+	if !strings.HasSuffix(long, " · Send message to resume") {
+		t.Fatalf("the hint was clipped away: %q", long)
+	}
+}
+
+// The status row must be exactly as wide as the column it sits in, at every
+// terminal size. A narrow terminal cannot fit the quit hint alongside any status
+// text, and keeping it anyway made the row wider than the terminal.
+func TestStatusRowIsExactlyItsWidth(t *testing.T) {
+	quitHint := lipgloss.NewStyle().Foreground(white).Render("ctrl-q") +
+		lipgloss.NewStyle().Foreground(dim).Render(" quit")
+	longMessage := lipgloss.NewStyle().Foreground(red).Render(strings.Repeat("boom ", 40))
+
+	for width := 1; width <= 60; width++ {
+		for _, testCase := range []struct {
+			name        string
+			left, right string
+		}{
+			{"empty", "", ""},
+			{"hint only", "", quitHint},
+			{"long message and hint", longMessage, quitHint},
+			{"long message alone", longMessage, ""},
+		} {
+			row := composeStatusRow(testCase.left, testCase.right, width)
+			if got := ansi.StringWidth(row); got != width {
+				t.Fatalf("%s at width %d rendered %d columns: %q",
+					testCase.name, width, got, ansi.Strip(row))
+			}
+			if strings.Contains(row, "\n") {
+				t.Fatalf("%s at width %d spans rows", testCase.name, width)
+			}
+		}
+	}
+	if row := composeStatusRow("x", "y", 0); row != "" {
+		t.Fatalf("a zero-width row should be empty, got %q", row)
+	}
+}
+
+// A running scan in a narrow terminal must not wrap the frame.
+func TestNarrowTerminalKeepsTheFrameIntact(t *testing.T) {
+	for _, width := range []int{8, 10, 13, 14, 20, 40} {
+		model := New(nil)
+		model.width, model.height = width, 20
+		model.showSplash = false
+		model.handleEnvelope(stateEnvelope(t, 1, protocol.Snapshot{ScanState: "running"}))
+		bootstrap := protocol.CollectionBootstrap{
+			Collection: "agents", Revision: 1, Cursor: 0, NextCursor: 1, Done: true,
+			Items: []json.RawMessage{rawJSON(t, protocol.Agent{ID: "a0", Name: "Strix", Status: "running"})},
+		}
+		model.handleEnvelope(protocol.Envelope{
+			Version: protocol.Version, Type: "collection_bootstrap", Payload: rawJSON(t, bootstrap),
+		})
+		model.errorText = strings.Repeat("connection failed ", 20)
+		model.resizeViewport()
+
+		for i, line := range strings.Split(model.View(), "\n") {
+			if got := ansi.StringWidth(line); got > width {
+				t.Fatalf("at width %d row %d is %d columns", width, i, got)
+			}
+		}
+	}
+}

strix/interface/tui/internal/app/setup.go

@@ -214,8 +214,11 @@ func (m *Model) setupLogAppend(line string) {
 }
 
 // setupMsg appends a styled feedback line (success green, error red, notice dim).
+// The log budgets rows by entry, so a message is flattened to one line first: a
+// wrapped exception would otherwise render as several rows and push the launch
+// column past the bottom of the terminal.
 func (m *Model) setupMsg(text string, style lipgloss.Style) {
-	m.setupLogAppend(style.Render(text))
+	m.setupLogAppend(style.Render(flattenStatus(text)))
 }
 
 // setupLogRows is how many feedback lines the launch column shows before the

strix/interface/tui/internal/app/setup_log_test.go

@@ -93,3 +93,25 @@ func TestFocusedPanelsCarryTheGreenBorder(t *testing.T) {
 		}
 	}
 }
+
+// A wrapped exception is several lines. The log budgets rows by entry, so it has
+// to become one row or the launch column grows past the terminal.
+func TestSetupLogKeepsMultiLineErrorsToOneRow(t *testing.T) {
+	model := New(nil)
+	model.width, model.height = 100, 26
+	model.showSplash = false
+	model.handleEnvelope(stateEnvelope(t, 1, protocol.Snapshot{SetupMode: true, ScanState: "setup"}))
+	model.setupMsg("boom\nTraceback (most recent call last):\n  File \"x.py\", line 1\n    raise", render.Col(red))
+	model.resizeViewport()
+
+	if entries := len(model.setupLog); entries != 1 {
+		t.Fatalf("one message became %d log entries", entries)
+	}
+	if strings.Contains(model.setupLog[0], "\n") {
+		t.Fatalf("log entry spans rows: %q", model.setupLog[0])
+	}
+	lines := strings.Split(model.View(), "\n")
+	if len(lines) > model.height {
+		t.Fatalf("start screen is %d rows in a %d-row terminal", len(lines), model.height)
+	}
+}

strix/interface/tui/internal/app/view.go

@@ -654,8 +654,7 @@ func (m Model) statusView(width int) string {
 		case "waiting":
 			left = lipgloss.NewStyle().Foreground(dim).Render("Send message to resume")
 			if msg := agent.ErrorMessage; msg != "" {
-				left = lipgloss.NewStyle().Foreground(red).Render(msg) +
-					lipgloss.NewStyle().Foreground(dim).Render(" · Send message to resume")
+				left = statusMessage(msg, red, " · Send message to resume", width)
 			}
 		case "budget_paused":
 			left = lipgloss.NewStyle().Foreground(amber).Render("Budget limit reached") +
@@ -670,15 +669,54 @@ func (m Model) statusView(width int) string {
 			if msg == "" {
 				msg = "Agent failed"
 			}
-			left = lipgloss.NewStyle().Foreground(red).Render(msg) +
-				lipgloss.NewStyle().Foreground(dim).Render(" · Send message to resume")
+			left = statusMessage(msg, red, " · Send message to resume", width)
 		}
 	}
 	if m.errorText != "" {
-		left = lipgloss.NewStyle().Foreground(red).Render(m.errorText)
+		left = statusMessage(m.errorText, red, "", width-lipgloss.Width(right))
 	}
-	gap := max(1, width-lipgloss.Width(left)-lipgloss.Width(right))
-	return " " + left + strings.Repeat(" ", max(1, gap-1)) + right
+	return composeStatusRow(left, right, width)
+}
+
+// composeStatusRow lays the status text and the corner hint on one row exactly
+// width columns wide. A wider row would widen the whole chat column, because
+// JoinHorizontal pads every row of a block to its widest, which pushes the
+// sidebar off screen and wraps the frame.
+func composeStatusRow(left, right string, width int) string {
+	if width <= 0 {
+		return ""
+	}
+	const leading = 1 // the row is indented one column, like the panels above it
+	// A terminal can be narrower than the hint itself. Drop the hint rather than
+	// keep it at the cost of the status, which is the part carrying information;
+	// ctrl-q works whether or not the row has room to say so.
+	if lipgloss.Width(right) > 0 && width < lipgloss.Width(right)+leading+2 {
+		right = ""
+	}
+	separator := 0
+	if lipgloss.Width(right) > 0 {
+		separator = 1
+	}
+	left = truncate(left, max(0, width-leading-lipgloss.Width(right)-separator))
+	padding := max(0, width-leading-lipgloss.Width(left)-lipgloss.Width(right))
+	return " " + left + strings.Repeat(" ", padding) + right
+}
+
+// statusMessage fits a message and its trailing hint on the one status row. A
+// model or backend error can be a wrapped exception several lines long, so it is
+// flattened to a single line and clipped, leaving the hint readable.
+func statusMessage(message string, color lipgloss.Color, hint string, width int) string {
+	styledHint := lipgloss.NewStyle().Foreground(dim).Render(hint)
+	room := max(1, width-2-lipgloss.Width(styledHint))
+	flat := truncate(flattenStatus(message), room)
+	return lipgloss.NewStyle().Foreground(color).Render(flat) + styledHint
+}
+
+// flattenStatus turns a multi-line message into one line, collapsing the runs of
+// whitespace that joining its lines leaves behind.
+func flattenStatus(message string) string {
+	message = strings.NewReplacer("\r\n", " ", "\r", " ", "\n", " ", "\t", " ").Replace(message)
+	return strings.Join(strings.Fields(message), " ")
 }
 
 func (m Model) sweepView() string {

strix/interface/tui/internal/app/vuln_report.go

@@ -74,6 +74,8 @@ func vulnerabilityMarkdownReport(v map[string]any) string {
 		field("Ecosystem", render.StringValue(dep["package_ecosystem"]))
 		field("Installed Version", render.StringValue(dep["installed_version"]))
 		field("Fixed Version", render.StringValue(dep["fixed_version"]))
+		field("Introduced By", render.StringValue(dep["introduced_by"]))
+		field("Dependency Chain", render.StringValue(dep["dependency_path"]))
 	}
 	field("Endpoint", render.StringValue(v["endpoint"]))
 	field("Method", render.StringValue(v["method"]))

strix/interface/tui/internal/app/vulnerabilities.go

@@ -298,6 +298,8 @@ func vulnerabilityBody(v map[string]any) string {
 		field("Ecosystem", render.StringValue(dep["package_ecosystem"]))
 		field("Installed Version", render.StringValue(dep["installed_version"]))
 		field("Fixed Version", render.StringValue(dep["fixed_version"]))
+		field("Introduced By", render.StringValue(dep["introduced_by"]))
+		field("Dependency Chain", render.StringValue(dep["dependency_path"]))
 	}
 	field("Endpoint", render.StringValue(v["endpoint"]))
 	field("Method", render.StringValue(v["method"]))

strix/report/sarif.py

@@ -531,6 +531,10 @@ def _result_properties(
         if value not in (None, ""):
             strix[key] = value
 
+    dependency_metadata = report.get("dependency_metadata")
+    if isinstance(dependency_metadata, dict) and dependency_metadata:
+        strix["dependency_metadata"] = dependency_metadata
+
     # SARIF is written for external upload (code-scanning / ASPM), so it must
     # NOT carry the weaponized exploit payload — that stays a local run
     # artifact (vulnerabilities.json / the finding MD). We surface the PoC

strix/report/writer.py

@@ -205,6 +205,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
         ("Ecosystem", dep_meta.get("package_ecosystem")),
         ("Installed Version", dep_meta.get("installed_version")),
         ("Fixed Version", dep_meta.get("fixed_version")),
+        ("Introduced By", dep_meta.get("introduced_by")),
+        ("Dependency Chain", dep_meta.get("dependency_path")),
         ("Endpoint", report.get("endpoint")),
         ("Method", report.get("method")),
         ("CVE", report.get("cve")),

strix/skills/custom/dependency_cve_scanning.md

@@ -39,9 +39,11 @@ trivy version --format json 2>/dev/null | tee "$ART/trivy-version.json"
 # sandbox with egress gets the freshest CVEs; if the update fails, fall back to the
 # cached DB instead of failing the scan. --offline-scan keeps per-package advisory
 # lookups offline.
-trivy fs --scanners vuln --timeout 30m --offline-scan \
+# --list-all-pkgs includes the package graph (Relationship + DependsOn) needed
+# to attribute transitive CVEs to the direct dependency that introduces them.
+trivy fs --scanners vuln --timeout 30m --offline-scan --list-all-pkgs \
   --format json --output "$ART/trivy-sca.json" . \
-  || trivy fs --scanners vuln --timeout 30m --offline-scan --skip-db-update \
+  || trivy fs --scanners vuln --timeout 30m --offline-scan --skip-db-update --list-all-pkgs \
        --format json --output "$ART/trivy-sca.json" . \
   || true
 ```
@@ -78,6 +80,36 @@ For each entry under `.Results[].Vulnerabilities[]` in `trivy-sca.json`, collect
 Deduplicate by `(CVE, PkgName, InstalledVersion)`. File one
 `create_dependency_report` per CVE — do not batch multiple CVEs into one report.
 
+### Attribute transitive CVEs to the direct dependency
+
+With `--list-all-pkgs`, each `.Results[].Packages[]` entry carries `ID`
+(`name@version`), `Relationship` (`direct` / `indirect`) and `DependsOn` (the
+`ID`s it resolves to). For every vulnerable package that is **indirect**, walk
+the `DependsOn` graph backwards to find the `direct` package(s) whose closure
+contains it, then pass to `create_dependency_report`:
+
+- `introduced_by` — the direct dependency as `name@version` (e.g.
+  `express@4.18.1`). If several direct dependencies pull it in, pick the
+  primary one and name the rest in `technical_analysis`.
+- `dependency_path` — the shortest resolution chain from that direct
+  dependency to the vulnerable package, joined with ` > ` (e.g.
+  `express@4.18.1 > body-parser@1.20.0 > qs@6.10.2`).
+- Omit both when the vulnerable package is itself a direct dependency.
+
+If the ecosystem's lockfile gives trivy no graph (`DependsOn` absent), derive
+the chain from the package manager instead (`npm ls <pkg>`, `pnpm why <pkg>`,
+`yarn why <pkg>`, `pipdeptree --reverse -p <pkg>`, `go mod graph`,
+`mvn dependency:tree`, ...) — and if that also fails, leave the fields out
+rather than guessing.
+
+For transitive findings, `remediation_steps` must be actionable at the
+**direct-dependency level**: upgrading the vulnerable package directly is
+usually impossible from the app's own manifest. Say which direct dependency to
+bump (a version whose closure resolves the fixed version), or how to force the
+resolution (npm `overrides` / yarn `resolutions` / pnpm `pnpm.overrides` /
+Maven `dependencyManagement` / Gradle resolution strategy / `go mod edit`),
+not just "upgrade <vulnerable pkg> to <fixed>".
+
 ### Reachability is a confidence modifier, not a gate
 
 Do NOT suppress or downgrade a known CVE just because you could not prove the

strix/tools/reporting/tool.py

@@ -725,6 +725,8 @@ def _build_dependency_metadata(
     installed_version: str,
     package_ecosystem: str | None,
     fixed_version: str | None,
+    introduced_by: str | None,
+    dependency_path: str | None,
 ) -> dict[str, str]:
     metadata = {
         "package_name": package_name.strip(),
@@ -734,6 +736,10 @@ def _build_dependency_metadata(
         metadata["package_ecosystem"] = package_ecosystem.strip()
     if fixed_version and fixed_version.strip():
         metadata["fixed_version"] = fixed_version.strip()
+    if introduced_by and introduced_by.strip():
+        metadata["introduced_by"] = introduced_by.strip()
+    if dependency_path and dependency_path.strip():
+        metadata["dependency_path"] = dependency_path.strip()
     return metadata
 
 
@@ -743,6 +749,8 @@ def _build_dependency_evidence(
     package_name: str,
     installed_version: str,
     fixed_version: str | None,
+    introduced_by: str | None,
+    dependency_path: str | None,
 ) -> str:
     evidence = (
         f"**Advisory evidence:** `{cve}` applies to `{package_name}` "
@@ -750,6 +758,13 @@ def _build_dependency_evidence(
     )
     if fixed_version and fixed_version.strip():
         evidence += f" The advisory is fixed in `{fixed_version.strip()}`."
+    if introduced_by and introduced_by.strip():
+        evidence += (
+            f"\n\n**Transitive dependency:** introduced by the direct "
+            f"dependency `{introduced_by.strip()}`."
+        )
+    if dependency_path and dependency_path.strip():
+        evidence += f"\n\n**Dependency chain:** `{dependency_path.strip()}`"
     return evidence
 
 
@@ -770,6 +785,8 @@ async def _do_create_dependency(  # noqa: PLR0912
     advisory_cvss: float | None,
     technical_analysis: str | None,
     fix_effort: str,
+    introduced_by: str | None = None,
+    dependency_path: str | None = None,
     agent_id: str | None = None,
     agent_name: str | None = None,
 ) -> dict[str, Any]:
@@ -824,12 +841,16 @@ async def _do_create_dependency(  # noqa: PLR0912
         installed_version=installed_version,
         package_ecosystem=package_ecosystem,
         fixed_version=fixed_version,
+        introduced_by=introduced_by,
+        dependency_path=dependency_path,
     )
     evidence = _build_dependency_evidence(
         cve=parsed_cve,
         package_name=package_name.strip(),
         installed_version=installed_version.strip(),
         fixed_version=fixed_version,
+        introduced_by=introduced_by,
+        dependency_path=dependency_path,
     )
 
     try:
@@ -926,6 +947,8 @@ async def create_dependency_report(
     cwe: str | None = None,
     technical_analysis: str | None = None,
     fix_effort: str = "low",
+    introduced_by: str | None = None,
+    dependency_path: str | None = None,
 ) -> str:
     """File a known-CVE dependency (SCA) finding — one report per CVE x package.
 
@@ -978,6 +1001,15 @@ async def create_dependency_report(
         technical_analysis: Optional deeper mechanism/root-cause detail.
         fix_effort: One of ``trivial`` / ``low`` / ``medium`` / ``high``
             (dependency upgrades are usually ``trivial``/``low``).
+        introduced_by: For a **transitive** dependency, the direct
+            dependency (from the project's own manifest) that pulls the
+            vulnerable package in, as ``name@version`` (e.g.
+            ``express@4.18.1``). Omit when the vulnerable package is
+            itself a direct dependency.
+        dependency_path: The resolution chain from the direct dependency
+            to the vulnerable package, joined with `` > `` (e.g.
+            ``express@4.18.1 > body-parser@1.20.0 > qs@6.10.2``). Omit
+            for direct dependencies.
     """
     agent_id, agent_name = _caller_identity(ctx)
 
@@ -997,6 +1029,8 @@ async def create_dependency_report(
         advisory_cvss=advisory_cvss,
         technical_analysis=technical_analysis,
         fix_effort=fix_effort,
+        introduced_by=introduced_by,
+        dependency_path=dependency_path,
         agent_id=agent_id,
         agent_name=agent_name,
     )

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, _TurnGuardModel
 
 
 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, _TurnGuardModel)
+    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,17 +311,20 @@ def test_get_model_unwrapped_by_default(
     load_settings()
 
     model = StrixProvider().get_model("openai/gpt-4o-mini")
-    assert model is inner
+    assert isinstance(model, _TurnGuardModel)
+    assert model._inner is inner
 
 
-def test_get_model_does_not_wrap_subscription_model(
+def test_get_model_guards_subscription_model_but_keeps_it_streaming(
     monkeypatch: pytest.MonkeyPatch, _reset_settings: None
 ) -> None:
-    # Subscription (ChatGPT) models are always streamed and must not be wrapped.
+    # Subscription (ChatGPT) models are always streamed, so LLM_DISABLE_STREAMING
+    # must not apply — but a runaway response needs capping there too.
     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 not isinstance(model, _NonStreamingModel)
+    assert isinstance(model, _TurnGuardModel)
+    assert not isinstance(model._inner, _NonStreamingModel)

tests/test_reporting_fields.py

@@ -164,6 +164,69 @@ async def test_dependency_report_sets_class_and_metadata(report_state: ReportSta
     }
 
 
+async def test_dependency_report_records_transitive_chain(report_state: ReportState) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2022-24999 in qs 6.10.2",
+        description="Prototype pollution in qs parsing.",
+        target="repo/package.json",
+        cve="CVE-2022-24999",
+        package_name="qs",
+        installed_version="6.10.2",
+        impact="Denial of service via crafted query strings.",
+        remediation_steps="Upgrade express to 4.18.2, which resolves qs 6.11.0.",
+        assumptions="qs parses all incoming query strings by default.",
+        package_ecosystem="npm",
+        fixed_version="6.10.3",
+        cwe="CWE-1321",
+        advisory_cvss=7.5,
+        technical_analysis=None,
+        fix_effort="trivial",
+        introduced_by="express@4.18.1",
+        dependency_path="express@4.18.1 > body-parser@1.20.0 > qs@6.10.2",
+    )
+    assert result["success"] is True
+    report = report_state.vulnerability_reports[0]
+    assert report["dependency_metadata"]["introduced_by"] == "express@4.18.1"
+    assert (
+        report["dependency_metadata"]["dependency_path"]
+        == "express@4.18.1 > body-parser@1.20.0 > qs@6.10.2"
+    )
+    assert (
+        "**Transitive dependency:** introduced by the direct dependency `express@4.18.1`."
+        in report["evidence"]
+    )
+    assert (
+        "**Dependency chain:** `express@4.18.1 > body-parser@1.20.0 > qs@6.10.2`"
+        in report["evidence"]
+    )
+
+
+async def test_dependency_report_omits_blank_chain_fields(report_state: ReportState) -> None:
+    result = await _do_create_dependency(
+        title="CVE-2024-0001 in sample 1.0.0",
+        description="Published advisory affects the pinned version.",
+        target="repo/package.json",
+        cve="CVE-2024-0001",
+        package_name="sample",
+        installed_version="1.0.0",
+        impact="Impact.",
+        remediation_steps="Upgrade.",
+        assumptions="Assumptions.",
+        package_ecosystem="npm",
+        fixed_version=None,
+        cwe=None,
+        advisory_cvss=5.0,
+        technical_analysis=None,
+        fix_effort="trivial",
+        introduced_by="  ",
+        dependency_path=None,
+    )
+    assert result["success"] is True
+    report = report_state.vulnerability_reports[0]
+    assert "introduced_by" not in report["dependency_metadata"]
+    assert "dependency_path" not in report["dependency_metadata"]
+
+
 async def test_dependency_report_with_zero_cvss_remains_low_severity(
     report_state: ReportState,
 ) -> None:

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, _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

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