feat(models): STRIX_STREAM_MODE opt-in non-streaming for custom endpoints

Replace the api_base-based non-streaming heuristic with an explicit
STRIX_STREAM_MODE=auto|always|never setting. auto/always stream (unchanged
default); never routes through the non-streaming wrapper for endpoints whose
streamed responses drop tool calls.

strix/config/models.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import contextlib
 import inspect
 import os
+import time
 from typing import TYPE_CHECKING, Any
 
 from agents import (
@@ -13,6 +14,8 @@ from agents import (
     set_tracing_disabled,
 )
 from agents.model_settings import ModelSettings
+from agents.models.fake_id import FAKE_RESPONSES_ID
+from agents.models.interface import Model
 from agents.models.multi_provider import MultiProvider
 from agents.models.openai_responses import OpenAIResponsesModel
 from agents.retry import (
@@ -21,6 +24,12 @@ from agents.retry import (
     RetryPolicyContext,
     retry_policies,
 )
+from openai.types.responses import (
+    Response,
+    ResponseCompletedEvent,
+    ResponseOutputItemDoneEvent,
+    ResponseUsage,
+)
 from openai.types.shared import Reasoning
 
 from strix.config import codex
@@ -30,8 +39,14 @@ from strix.config.loader import load_settings
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
 
-    from agents.models.interface import Model, ModelProvider
+    from agents.agent_output import AgentOutputSchemaBase
+    from agents.handoffs import Handoff
+    from agents.items import ModelResponse, TResponseInputItem, TResponseStreamEvent
+    from agents.models.interface import ModelProvider, ModelTracing
+    from agents.tool import Tool
+    from agents.usage import Usage
     from openai import AsyncOpenAI
+    from openai.types.responses import ResponsePromptParam
 
     from strix.config.settings import ReasoningEffort, Settings
 
@@ -135,6 +150,99 @@ class _CodexResponsesModel(OpenAIResponsesModel):
                     await result
 
 
+def _to_response_usage(usage: Usage | None) -> ResponseUsage | None:
+    if usage is None:
+        return None
+    return ResponseUsage(
+        input_tokens=usage.input_tokens,
+        output_tokens=usage.output_tokens,
+        total_tokens=usage.total_tokens,
+        input_tokens_details=usage.input_tokens_details,
+        output_tokens_details=usage.output_tokens_details,
+    )
+
+
+class _NonStreamingModel(Model):
+    """Run a model non-streamed but expose the streaming interface the runner uses.
+
+    Some OpenAI-compatible endpoints (notably gateways serving reasoning models)
+    return valid ``tool_calls`` for a non-streamed completion but, when streamed,
+    emit the tool call as plain text or drop it and close the stream — leaving
+    Strix's tool-driven loop with nothing to execute. Selecting
+    ``STRIX_STREAM_MODE=never`` routes through this wrapper, which makes the real
+    request non-streamed (where tool calling works) and synthesizes the minimal
+    event sequence the runner consumes from a stream, so the rest of the pipeline
+    is unchanged. The only user-visible difference is no token-by-token output.
+    """
+
+    def __init__(self, inner: Model) -> None:
+        self._inner = inner
+
+    async def get_response(self, *args: Any, **kwargs: Any) -> ModelResponse:
+        return await self._inner.get_response(*args, **kwargs)
+
+    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 = None,
+        conversation_id: str | None = None,
+        prompt: ResponsePromptParam | None = None,
+    ) -> AsyncIterator[TResponseStreamEvent]:
+        model_response = await self._inner.get_response(
+            system_instructions,
+            input,
+            model_settings,
+            tools,
+            output_schema,
+            handoffs,
+            tracing,
+            previous_response_id=previous_response_id,
+            conversation_id=conversation_id,
+            prompt=prompt,
+        )
+
+        sequence = 0
+        for index, item in enumerate(model_response.output):
+            yield ResponseOutputItemDoneEvent(
+                item=item,
+                output_index=index,
+                type="response.output_item.done",
+                sequence_number=sequence,
+            )
+            sequence += 1
+
+        response = Response(
+            id=model_response.response_id or FAKE_RESPONSES_ID,
+            created_at=time.time(),
+            model=str(getattr(self._inner, "model", "")),
+            object="response",
+            output=model_response.output,
+            tool_choice="auto",
+            tools=[],
+            parallel_tool_calls=False,
+            usage=_to_response_usage(model_response.usage),
+        )
+        yield ResponseCompletedEvent(
+            response=response,
+            type="response.completed",
+            sequence_number=sequence,
+        )
+
+    def get_retry_advice(self, request: Any) -> Any:
+        return self._inner.get_retry_advice(request)
+
+
+def _should_run_non_streamed(settings: Settings) -> bool:
+    return settings.llm.stream_mode == "never"
+
+
 class StrixProvider(MultiProvider):
     """Route any non-OpenAI prefix through LiteLLM with the prefix preserved,
     so users type ``deepseek/deepseek-chat`` rather than
@@ -159,14 +267,18 @@ class StrixProvider(MultiProvider):
         return self._get_fallback_provider("litellm"), original_model_name
 
     def get_model(self, model_name: str | None) -> Model:
+        settings = load_settings()
         slug = codex.subscription_model(model_name)
         if slug:
             return _CodexResponsesModel(
                 slug,
                 codex.get_subscription_client(),
-                reasoning_effort=load_settings().llm.reasoning_effort,
+                reasoning_effort=settings.llm.reasoning_effort,
             )
-        return super().get_model(model_name)
+        model = super().get_model(model_name)
+        if _should_run_non_streamed(settings):
+            return _NonStreamingModel(model)
+        return model
 
 
 DEFAULT_MODEL_RETRY = ModelRetrySettings(
@@ -277,6 +389,51 @@ def _configure_litellm_compatibility() -> None:
     litellm.suppress_debug_info = True
 
     _register_litellm_cost_callback()
+    _install_openrouter_stream_cost_capture()
+
+
+def _install_openrouter_stream_cost_capture() -> None:
+    """Preserve OpenRouter's per-stream cost, which LiteLLM drops when streaming.
+
+    OpenRouter reports the real charge in ``usage.cost`` of the final stream
+    chunk, but LiteLLM rebuilds streamed responses from token-only fields and
+    discards it (its non-streamed path stashes the cost in hidden params; the
+    streaming path does not). Every scan streams, so without this the cost is
+    lost and Strix falls back to a cost-map estimate that is missing entirely
+    for new models (e.g. kimi-k3), reporting $0. Subclass the OpenRouter
+    streaming handler to record the cost keyed by response id so the cost
+    callback can recover the exact charge for the matching rebuilt response.
+    """
+    import litellm
+    from litellm.llms.openrouter.chat.transformation import (
+        OpenRouterChatCompletionStreamingHandler,
+        OpenrouterConfig,
+    )
+
+    from strix.report.state import streamed_openrouter_costs
+
+    class _StrixOpenRouterStreamingHandler(OpenRouterChatCompletionStreamingHandler):
+        def chunk_parser(self, chunk: dict[str, Any]) -> Any:
+            stream = super().chunk_parser(chunk)
+            streamed_openrouter_costs.remember(
+                chunk.get("id") or getattr(stream, "id", None), chunk.get("usage")
+            )
+            return stream
+
+    class _StrixOpenrouterConfig(OpenrouterConfig):
+        def get_model_response_iterator(
+            self, streaming_response: Any, sync_stream: bool, json_mode: bool | None = False
+        ) -> Any:
+            return _StrixOpenRouterStreamingHandler(
+                streaming_response=streaming_response,
+                sync_stream=sync_stream,
+                json_mode=json_mode,
+            )
+
+    # LiteLLM's provider-config factory reads litellm.OpenrouterConfig at call
+    # time, so overriding the attribute is enough for the subclass to take
+    # effect. (type: ignore — mypy rejects reassigning a class attribute.)
+    litellm.OpenrouterConfig = _StrixOpenrouterConfig  # type: ignore[misc]
 
 
 _OPENROUTER_ATTRIBUTION_HEADERS = {

strix/config/settings.py

@@ -9,6 +9,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
 ReasoningEffort = Literal["none", "minimal", "low", "medium", "high", "xhigh"]
+StreamMode = Literal["auto", "always", "never"]
 
 _BASE_CONFIG = SettingsConfigDict(
     case_sensitive=False,
@@ -40,6 +41,12 @@ class LlmSettings(BaseSettings):
         default=False,
         alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE",
     )
+    # auto/always stream; never runs non-streamed, for endpoints that stream tool
+    # calls incorrectly (some OpenAI-compatible gateways serving reasoning models).
+    stream_mode: StreamMode = Field(
+        default="auto",
+        alias="STRIX_STREAM_MODE",
+    )
     prompt_cache: bool = Field(
         default=True,
         alias="STRIX_PROMPT_CACHE",

strix/report/state.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import subprocess
+import threading
 from collections.abc import Callable
 from datetime import UTC, datetime
 from importlib.metadata import PackageNotFoundError, version
@@ -95,6 +96,8 @@ def get_global_report_state() -> Optional["ReportState"]:
 def set_global_report_state(report_state: "ReportState") -> None:
     global _global_report_state  # noqa: PLW0603
     _global_report_state = report_state
+    # New run: drop any streamed-cost entries a prior run left unconsumed.
+    streamed_openrouter_costs.clear()
 
 
 class ReportState:
@@ -507,6 +510,72 @@ class ReportState:
         self._sync_llm_usage_record()
 
 
+def openrouter_stream_cost(usage: Any) -> float | None:
+    """Total OpenRouter-reported cost from a raw stream ``usage`` block, or None.
+
+    Non-BYOK responses bill everything to ``usage.cost``. BYOK responses put the
+    OpenRouter fee in ``usage.cost`` (often 0) and the provider charge in
+    ``usage.cost_details.upstream_inference_cost``, so BYOK totals sum the two.
+    """
+    if not isinstance(usage, dict):
+        return None
+    total = 0.0
+    cost = usage.get("cost")
+    if isinstance(cost, int | float) and cost > 0:
+        total += float(cost)
+    if bool(usage.get("is_byok")):
+        details = usage.get("cost_details")
+        upstream = details.get("upstream_inference_cost") if isinstance(details, dict) else None
+        if isinstance(upstream, int | float) and upstream > 0:
+            total += float(upstream)
+    return total if total > 0 else None
+
+
+def _response_id(completion_response: Any) -> str | None:
+    response_id = getattr(completion_response, "id", None)
+    if response_id is None and isinstance(completion_response, dict):
+        response_id = cast("dict[str, Any]", completion_response).get("id")
+    return response_id if isinstance(response_id, str) and response_id else None
+
+
+class StreamedOpenRouterCosts:
+    """Correlates OpenRouter's per-stream cost from the parser to the cost callback.
+
+    LiteLLM rebuilds streamed responses from token-only chunks and drops the
+    ``usage.cost`` OpenRouter reports in its final stream chunk (its non-streamed
+    path preserves it; streaming snapshots hidden params at stream start). Every
+    scan streams, so the OpenRouter streaming handler (see strix.config.models)
+    records the cost here keyed by response id, and the callback takes it back out
+    for the matching rebuilt response. Entries are removed on read; ``clear()``
+    runs per scan so nothing accumulates across runs.
+    """
+
+    def __init__(self) -> None:
+        self._costs: dict[str, float] = {}
+        self._lock = threading.Lock()
+
+    def remember(self, response_id: Any, usage: Any) -> None:
+        cost = openrouter_stream_cost(usage)
+        if cost is None or not (isinstance(response_id, str) and response_id):
+            return
+        with self._lock:
+            self._costs[response_id] = cost
+
+    def take(self, completion_response: Any) -> float | None:
+        response_id = _response_id(completion_response)
+        if response_id is None:
+            return None
+        with self._lock:
+            return self._costs.pop(response_id, None)
+
+    def clear(self) -> None:
+        with self._lock:
+            self._costs.clear()
+
+
+streamed_openrouter_costs = StreamedOpenRouterCosts()
+
+
 def litellm_cost_callback(
     kwargs: Any,
     completion_response: Any,
@@ -541,6 +610,11 @@ def litellm_cost_callback(
     if cost is None:
         cost = _usage_reported_cost(completion_response)
 
+    # Recover the exact OpenRouter cost the streaming handler stashed for this
+    # response — LiteLLM drops it from streamed usage, so nothing above sees it.
+    if cost is None:
+        cost = streamed_openrouter_costs.take(completion_response)
+
     if cost is None:
         cost = _estimate_response_cost(kwargs, completion_response)
 

tests/test_cost_tracking.py

@@ -7,9 +7,25 @@ from unittest.mock import MagicMock, patch
 
 import litellm
 import pytest
+from litellm.types.utils import LlmProviders
+from litellm.utils import ProviderConfigManager
 
-from strix.config.models import _configure_litellm_compatibility
-from strix.report.state import litellm_cost_callback
+from strix.config.models import (
+    _configure_litellm_compatibility,
+    _install_openrouter_stream_cost_capture,
+)
+from strix.report.state import (
+    ReportState,
+    litellm_cost_callback,
+    openrouter_stream_cost,
+    set_global_report_state,
+    streamed_openrouter_costs,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clear_streamed_costs() -> None:
+    streamed_openrouter_costs.clear()
 
 
 def test_streaming_logging_stays_enabled_for_cost_callback() -> None:
@@ -151,3 +167,90 @@ def test_cost_callback_records_nothing_when_no_cost_available() -> None:
         litellm_cost_callback({"response_cost": None, "model": "x/y"}, response)
 
     report_state.record_observed_llm_cost.assert_not_called()
+
+
+def test_openrouter_stream_cost_extracts_plain_and_byok_totals() -> None:
+    assert openrouter_stream_cost({"cost": 0.003168}) == pytest.approx(0.003168)
+    assert openrouter_stream_cost(
+        {"cost": 0.01, "is_byok": True, "cost_details": {"upstream_inference_cost": 0.2}}
+    ) == pytest.approx(0.21)
+    # Upstream cost is only added for BYOK responses.
+    assert openrouter_stream_cost(
+        {"cost": 0.05, "is_byok": False, "cost_details": {"upstream_inference_cost": 0.04}}
+    ) == pytest.approx(0.05)
+    assert openrouter_stream_cost({"prompt_tokens": 10}) is None
+    assert openrouter_stream_cost(None) is None
+
+
+def test_cost_callback_recovers_streamed_openrouter_cost_by_response_id() -> None:
+    report_state = MagicMock()
+    streamed_openrouter_costs.remember("gen-abc", {"cost": 0.42})
+    # LiteLLM strips cost from the rebuilt streamed usage; only the id survives.
+    response = SimpleNamespace(id="gen-abc", usage=SimpleNamespace(cost=None), _hidden_params={})
+
+    with (
+        patch("strix.report.state.get_global_report_state", return_value=report_state),
+        patch("litellm.completion_cost", side_effect=ValueError("unknown model")),
+    ):
+        litellm_cost_callback({"response_cost": None, "model": "moonshotai/kimi-k3"}, response)
+
+    report_state.record_observed_llm_cost.assert_called_once_with(0.42)
+    # The entry is consumed so a later response cannot double-count it.
+    assert streamed_openrouter_costs.take(response) is None
+
+
+def test_streamed_openrouter_cost_prefers_provider_report_over_estimate() -> None:
+    report_state = MagicMock()
+    streamed_openrouter_costs.remember("gen-xyz", {"cost": 0.9})
+    response = SimpleNamespace(
+        id="gen-xyz",
+        usage=SimpleNamespace(prompt_tokens=10, completion_tokens=5, total_tokens=15),
+        _hidden_params={},
+    )
+
+    with (
+        patch("strix.report.state.get_global_report_state", return_value=report_state),
+        patch("litellm.completion_cost", return_value=0.1) as estimate,
+    ):
+        litellm_cost_callback({"response_cost": None, "model": "moonshotai/kimi-k3"}, response)
+
+    report_state.record_observed_llm_cost.assert_called_once_with(0.9)
+    estimate.assert_not_called()
+
+
+def test_streamed_openrouter_costs_ignores_entries_without_cost() -> None:
+    streamed_openrouter_costs.remember("gen-none", {"prompt_tokens": 10})
+    streamed_openrouter_costs.remember("", {"cost": 0.5})
+    assert streamed_openrouter_costs.take(SimpleNamespace(id="gen-none")) is None
+
+
+def test_streamed_openrouter_costs_cleared_on_new_run() -> None:
+    streamed_openrouter_costs.remember("gen-stale", {"cost": 0.7})
+    set_global_report_state(ReportState.__new__(ReportState))
+    assert streamed_openrouter_costs.take(SimpleNamespace(id="gen-stale")) is None
+
+
+def test_openrouter_stream_handler_records_cost() -> None:
+    _install_openrouter_stream_cost_capture()
+    # Resolve the config the way LiteLLM does in production so we prove the
+    # override is actually reachable through provider resolution, not just as a
+    # directly-constructed class.
+    config = ProviderConfigManager.get_provider_chat_config(
+        model="moonshotai/kimi-k3", provider=LlmProviders.OPENROUTER
+    )
+    assert config is not None
+    assert type(config).__name__ == "_StrixOpenrouterConfig"
+    handler = config.get_model_response_iterator(streaming_response=iter([]), sync_stream=True)
+
+    chunk = {
+        "id": "gen-stream",
+        "created": 1,
+        "model": "moonshotai/kimi-k3",
+        "choices": [{"index": 0, "delta": {"content": None}}],
+        "usage": {"prompt_tokens": 89, "completion_tokens": 138, "cost": 0.0035055},
+    }
+    handler.chunk_parser(chunk)
+
+    assert streamed_openrouter_costs.take(SimpleNamespace(id="gen-stream")) == pytest.approx(
+        0.0035055
+    )

tests/test_nonstreaming_model.py

@@ -0,0 +1,173 @@
+"""Tests for the non-streaming wrapper used on custom OpenAI-compatible endpoints."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, cast
+from unittest.mock import patch
+
+import pytest
+from agents.items import ModelResponse
+from agents.model_settings import ModelSettings
+from agents.models.interface import Model, ModelTracing
+from agents.usage import Usage
+from openai.types.responses import (
+    ResponseCompletedEvent,
+    ResponseFunctionToolCall,
+    ResponseOutputItemDoneEvent,
+    ResponseStreamEvent,
+)
+
+from strix.config.models import StrixProvider, _NonStreamingModel, _to_response_usage
+
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+
+def _tool_call() -> ResponseFunctionToolCall:
+    return ResponseFunctionToolCall(
+        arguments='{"command": "ls"}',
+        call_id="call_1",
+        name="terminal_execute",
+        type="function_call",
+    )
+
+
+class _FakeModel(Model):
+    def __init__(self, response: ModelResponse) -> None:
+        self.model = "fake-model"
+        self._response = response
+        self.get_response_calls = 0
+
+    async def get_response(self, *_args: object, **_kwargs: object) -> ModelResponse:
+        self.get_response_calls += 1
+        return self._response
+
+    async def stream_response(  # pragma: no cover
+        self, *_args: object, **_kwargs: object
+    ) -> AsyncIterator[ResponseStreamEvent]:
+        for _ in range(0):
+            yield cast("ResponseStreamEvent", None)
+        raise AssertionError("inner stream_response must never be called")
+
+
+def _settings(*, api_base: str | None, stream_mode: str = "auto") -> SimpleNamespace:
+    return SimpleNamespace(
+        llm=SimpleNamespace(
+            model="openai/glm",
+            api_base=api_base,
+            stream_mode=stream_mode,
+            reasoning_effort="high",
+        )
+    )
+
+
+def test_to_response_usage_maps_token_details() -> None:
+    usage = Usage(requests=1, input_tokens=10, output_tokens=5, total_tokens=15)
+    usage.input_tokens_details.cached_tokens = 4
+    usage.output_tokens_details.reasoning_tokens = 3
+    mapped = _to_response_usage(usage)
+    assert mapped is not None
+    assert (mapped.input_tokens, mapped.output_tokens, mapped.total_tokens) == (10, 5, 15)
+    assert mapped.input_tokens_details.cached_tokens == 4
+    assert mapped.output_tokens_details.reasoning_tokens == 3
+
+
+def test_to_response_usage_none() -> None:
+    assert _to_response_usage(None) is None
+
+
+@pytest.mark.asyncio
+async def test_stream_response_synthesizes_tool_call_from_non_streamed() -> None:
+    tool_call = _tool_call()
+    inner = _FakeModel(
+        ModelResponse(
+            output=[tool_call],
+            usage=Usage(requests=1, input_tokens=10, output_tokens=5, total_tokens=15),
+            response_id="resp_123",
+        )
+    )
+    wrapper = _NonStreamingModel(inner)
+
+    events = [
+        event
+        async for event in wrapper.stream_response(
+            "sys",
+            "hi",
+            ModelSettings(),
+            [],
+            None,
+            [],
+            ModelTracing.DISABLED,
+        )
+    ]
+
+    assert inner.get_response_calls == 1
+    item_done = [e for e in events if isinstance(e, ResponseOutputItemDoneEvent)]
+    completed = [e for e in events if isinstance(e, ResponseCompletedEvent)]
+    assert len(item_done) == 1
+    assert item_done[0].item == tool_call
+    assert len(completed) == 1
+    final = completed[0].response
+    assert final.output == [tool_call]
+    assert final.id == "resp_123"
+    assert final.usage is not None
+    assert final.usage.total_tokens == 15
+    # sequence numbers are strictly increasing
+    assert [e.sequence_number for e in events] == list(range(len(events)))
+
+
+@pytest.mark.asyncio
+async def test_stream_response_delegates_get_response() -> None:
+    inner = _FakeModel(
+        ModelResponse(output=[], usage=Usage(), response_id=None),
+    )
+    wrapper = _NonStreamingModel(inner)
+    result = await wrapper.get_response(
+        "sys", "hi", ModelSettings(), [], None, [], ModelTracing.DISABLED
+    )
+    assert result is inner._response
+    assert inner.get_response_calls == 1
+
+
+def test_get_model_auto_streams_custom_endpoint() -> None:
+    sentinel = _FakeModel(ModelResponse(output=[], usage=Usage(), response_id=None))
+    with (
+        patch("strix.config.models.load_settings", return_value=_settings(api_base="http://x/v1")),
+        patch(
+            "agents.models.multi_provider.MultiProvider.get_model",
+            return_value=sentinel,
+        ),
+    ):
+        model = StrixProvider().get_model("openai/glm")
+    assert model is sentinel
+
+
+def test_get_model_auto_streams_hosted() -> None:
+    sentinel = _FakeModel(ModelResponse(output=[], usage=Usage(), response_id=None))
+    with (
+        patch("strix.config.models.load_settings", return_value=_settings(api_base="")),
+        patch(
+            "agents.models.multi_provider.MultiProvider.get_model",
+            return_value=sentinel,
+        ),
+    ):
+        model = StrixProvider().get_model("openai/gpt-4o")
+    assert model is sentinel
+
+
+def test_get_model_stream_mode_never_wraps() -> None:
+    sentinel = _FakeModel(ModelResponse(output=[], usage=Usage(), response_id=None))
+    with (
+        patch(
+            "strix.config.models.load_settings",
+            return_value=_settings(api_base="http://x/v1", stream_mode="never"),
+        ),
+        patch(
+            "agents.models.multi_provider.MultiProvider.get_model",
+            return_value=sentinel,
+        ),
+    ):
+        model = StrixProvider().get_model("openai/glm")
+    assert isinstance(model, _NonStreamingModel)