Compare commits

...
27 changed files with 855 additions and 116 deletions
+12
View File
@@ -315,6 +315,18 @@ strix auth status # show the active sign-in
strix auth logout # forget the sign-in strix auth logout # forget the sign-in
``` ```
#### Sign in with an OpenCode subscription
You can also run Strix on [OpenCode Zen](https://opencode.ai/docs/zen/) credits or an [OpenCode Go](https://opencode.ai/docs/go/) subscription:
```bash
strix auth login opencode # paste your API key from opencode.ai/auth
export STRIX_LLM="opencode/claude-sonnet-5" # opencode/<model> runs on Zen credits
export STRIX_LLM="opencode-go/kimi-k3" # opencode-go/<model> runs on the Go subscription
strix --target ./app-directory
```
**Recommended models for best results:** **Recommended models for best results:**
- [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4` - [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4`
+1
View File
@@ -280,6 +280,7 @@ ignore = [
# a runtime ``Callable`` annotation on ``vulnerability_found_callback``. # a runtime ``Callable`` annotation on ``vulnerability_found_callback``.
"strix/report/state.py" = ["TC003", "PLR0912", "PLR0915", "E501", "PERF401", "PLC0415"] "strix/report/state.py" = ["TC003", "PLR0912", "PLR0915", "E501", "PERF401", "PLC0415"]
"strix/report/usage.py" = ["PLC0415"] "strix/report/usage.py" = ["PLC0415"]
"strix/report/pricing.py" = ["PLC0415"]
# Lazy import of strix.config.models avoids a circular dependency between the # Lazy import of strix.config.models avoids a circular dependency between the
# report pipeline and the config layer. # report pipeline and the config layer.
"strix/report/dedupe.py" = ["PLC0415"] "strix/report/dedupe.py" = ["PLC0415"]
+27 -13
View File
@@ -72,8 +72,32 @@ def _write_store(data: dict[str, Any]) -> None:
write_secret_text(AUTH_PATH, json.dumps(data, indent=2)) write_secret_text(AUTH_PATH, json.dumps(data, indent=2))
def read_provider_record(provider: str) -> dict[str, Any] | None:
"""Raw record for *provider* from the shared subscription-auth store."""
record = _read_store().get(provider)
return record if isinstance(record, dict) else None
def save_provider_record(provider: str, record: dict[str, Any]) -> None:
data = _read_store()
data[provider] = record
_write_store(data)
def remove_provider_record(provider: str) -> None:
data = _read_store()
if provider not in data:
return
del data[provider]
if data:
_write_store(data)
return
with contextlib.suppress(OSError):
AUTH_PATH.unlink()
def read_record() -> dict[str, Any] | None: def read_record() -> dict[str, Any] | None:
record = _read_store().get(PROVIDER) record = read_provider_record(PROVIDER)
if not isinstance(record, dict) or record.get("type") != "oauth": if not isinstance(record, dict) or record.get("type") != "oauth":
return None return None
if not (record.get("access") and record.get("refresh") and record.get("account_id")): if not (record.get("access") and record.get("refresh") and record.get("account_id")):
@@ -86,21 +110,11 @@ def is_authenticated() -> bool:
def save_record(record: dict[str, Any]) -> None: def save_record(record: dict[str, Any]) -> None:
data = _read_store() save_provider_record(PROVIDER, record)
data[PROVIDER] = record
_write_store(data)
def logout() -> None: def logout() -> None:
data = _read_store() remove_provider_record(PROVIDER)
if PROVIDER not in data:
return
del data[PROVIDER]
if data:
_write_store(data)
return
with contextlib.suppress(OSError):
AUTH_PATH.unlink()
@contextlib.contextmanager @contextlib.contextmanager
+41 -9
View File
@@ -20,6 +20,7 @@ from agents.model_settings import ModelSettings
from agents.models.fake_id import FAKE_RESPONSES_ID from agents.models.fake_id import FAKE_RESPONSES_ID
from agents.models.interface import Model from agents.models.interface import Model
from agents.models.multi_provider import MultiProvider from agents.models.multi_provider import MultiProvider
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from agents.models.openai_responses import OpenAIResponsesModel from agents.models.openai_responses import OpenAIResponsesModel
from agents.retry import ( from agents.retry import (
ModelRetryBackoffSettings, ModelRetryBackoffSettings,
@@ -36,7 +37,7 @@ from openai.types.responses import (
from openai.types.responses.response_usage import ResponseUsage from openai.types.responses.response_usage import ResponseUsage
from openai.types.shared import Reasoning from openai.types.shared import Reasoning
from strix.config import codex from strix.config import codex, opencode
from strix.config.loader import load_settings from strix.config.loader import load_settings
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
from strix.config.tool_call_limits import TurnToolCallLimiter from strix.config.tool_call_limits import TurnToolCallLimiter
@@ -79,7 +80,12 @@ def _retry_statusless_provider_errors(context: RetryPolicyContext) -> bool:
class _CodexResponsesModel(OpenAIResponsesModel): class _CodexResponsesModel(OpenAIResponsesModel):
"""Responses model for the ChatGPT subscription backend (always streamed, stateless).""" """Responses model for stateless subscription gateways (always streamed).
Used for the ChatGPT subscription backend and for Responses-served models on
the OpenCode gateway: neither stores responses server-side, so reasoning is
carried inline via ``reasoning.encrypted_content``.
"""
def __init__( def __init__(
self, self,
@@ -471,6 +477,7 @@ class StrixProvider(MultiProvider):
def get_model(self, model_name: str | None) -> Model: def get_model(self, model_name: str | None) -> Model:
llm = load_settings().llm llm = load_settings().llm
slug = codex.subscription_model(model_name) slug = codex.subscription_model(model_name)
oc = opencode.subscription_model(model_name)
idle_timeout = float(llm.stream_idle_timeout) idle_timeout = float(llm.stream_idle_timeout)
if slug: if slug:
# The ChatGPT subscription backend is always streamed; it has no # The ChatGPT subscription backend is always streamed; it has no
@@ -481,6 +488,19 @@ class StrixProvider(MultiProvider):
codex.get_subscription_client(), codex.get_subscription_client(),
reasoning_effort=llm.reasoning_effort, reasoning_effort=llm.reasoning_effort,
) )
elif oc and oc.uses_responses:
model = _CodexResponsesModel(
oc.slug,
opencode.get_subscription_client(oc.base_url),
reasoning_effort=llm.reasoning_effort,
)
elif oc:
model = OpenAIChatCompletionsModel(
oc.slug, opencode.get_subscription_client(oc.base_url)
)
if llm.disable_streaming:
model = _NonStreamingModel(model)
idle_timeout = 0.0
else: else:
model = super().get_model(model_name) model = super().get_model(model_name)
if llm.disable_streaming: if llm.disable_streaming:
@@ -540,15 +560,24 @@ RECOMMENDED_MODEL_NAMES = (
_RECOMMENDED_MODEL_NAME_SET = frozenset(name.lower() for name in RECOMMENDED_MODEL_NAMES) _RECOMMENDED_MODEL_NAME_SET = frozenset(name.lower() for name in RECOMMENDED_MODEL_NAMES)
FRONTIER_MODEL_FAMILIES = ( FRONTIER_MODEL_FAMILIES = (
(("azure", "azure_ai", "bedrock_mantle", "chatgpt", "openai"), ("gpt-5",)), (("azure", "azure_ai", "bedrock_mantle", "chatgpt", "openai", "opencode"), ("gpt-5",)),
( (
("anthropic", "azure_ai", "bedrock", "claude", "databricks", "snowflake", "vertex_ai"), (
"anthropic",
"azure_ai",
"bedrock",
"claude",
"databricks",
"opencode",
"snowflake",
"vertex_ai",
),
("claude-fable-5", "claude-opus-5", "claude-opus-4", "claude-sonnet-5", "claude-sonnet-4"), ("claude-fable-5", "claude-opus-5", "claude-opus-4", "claude-sonnet-5", "claude-sonnet-4"),
), ),
(("google", "gemini", "vertex_ai"), ("gemini-3",)), (("google", "gemini", "opencode", "vertex_ai"), ("gemini-3",)),
(("deepseek",), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")), (("deepseek", "opencode"), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")),
(("alibaba", "dashscope", "qwen"), ("qwen3.8", "qwen3.7", "qwen3-max")), (("alibaba", "dashscope", "opencode", "qwen"), ("qwen3.8", "qwen3.7", "qwen3-max")),
(("moonshot", "moonshotai", "kimi"), ("kimi-k3", "kimi-k2.7", "kimi-k2.6")), (("kimi", "moonshot", "moonshotai", "opencode"), ("kimi-k3", "kimi-k2.7", "kimi-k2.6")),
) )
@@ -556,7 +585,7 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
"""Apply Strix config to SDK-native defaults.""" """Apply Strix config to SDK-native defaults."""
llm = settings.llm llm = settings.llm
set_tracing_disabled(True) set_tracing_disabled(True)
if codex.subscription_model(llm.model): if codex.subscription_model(llm.model) or opencode.subscription_model(llm.model):
return return
_configure_litellm_compatibility() _configure_litellm_compatibility()
_configure_openrouter_attribution(llm.model) _configure_openrouter_attribution(llm.model)
@@ -741,6 +770,9 @@ def uses_chat_completions_tool_schema(model_name: str, settings: Settings) -> bo
"""Return whether the resolved SDK route can only receive JSON function tools.""" """Return whether the resolved SDK route can only receive JSON function tools."""
if codex.subscription_model(model_name): if codex.subscription_model(model_name):
return False return False
oc = opencode.subscription_model(model_name)
if oc:
return not oc.uses_responses
model = model_name.strip().lower() model = model_name.strip().lower()
if "/" in model and not model.startswith("openai/"): if "/" in model and not model.startswith("openai/"):
return True return True
+158
View File
@@ -0,0 +1,158 @@
"""OpenCode subscription auth: API-key sign-in and the OpenAI clients that
route inference through the OpenCode gateway.
Covers both OpenCode offerings — Zen (pay-as-you-go credits) and Go (the
monthly subscription) — which share one account and API key but live behind
different gateway base URLs. Unlike the ChatGPT subscription there is no
OAuth: the user copies a plain API key from https://opencode.ai/auth, and
using the gateway from other agents is officially supported.
Model routing follows the endpoint each model is served on (see
https://opencode.ai/docs/zen/): GPT models use the Responses API, everything
else the OpenAI-compatible Chat Completions API.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import httpx
import requests
from openai import AsyncOpenAI
from strix.config import codex
PROVIDER = "opencode"
ZEN_BASE_URL = "https://opencode.ai/zen/v1"
GO_BASE_URL = "https://opencode.ai/zen/go/v1"
# ``opencode/<model>`` runs on Zen credits; ``opencode-go/<model>`` on the Go
# subscription (matching OpenCode's own ``opencode-go/`` model ids).
ZEN_PREFIX = "opencode/"
GO_PREFIX = "opencode-go/"
AUTH_CONSOLE_URL = "https://opencode.ai/auth"
_KEY_CHECK_TIMEOUT = 30
class OpencodeAuthError(Exception):
def __init__(self, code: str, message: str | None = None) -> None:
self.code = code
super().__init__(message or code)
@dataclass(frozen=True)
class SubscriptionModel:
slug: str
base_url: str
uses_responses: bool
def _uses_responses(slug: str, base_url: str) -> bool:
lowered = slug.lower()
if lowered.startswith("gpt-"):
return True
# Grok is served via Responses on Zen but Chat Completions on Go.
return lowered.startswith("grok") and base_url == ZEN_BASE_URL
def subscription_model(model_name: str | None) -> SubscriptionModel | None:
"""The gateway model behind an ``opencode/`` or ``opencode-go/`` STRIX_LLM."""
name = (model_name or "").strip()
lowered = name.lower()
for prefix, base_url in ((GO_PREFIX, GO_BASE_URL), (ZEN_PREFIX, ZEN_BASE_URL)):
if lowered.startswith(prefix):
slug = name[len(prefix) :]
if not slug:
return None
return SubscriptionModel(slug, base_url, _uses_responses(slug, base_url))
return None
def read_record() -> dict[str, Any] | None:
record = codex.read_provider_record(PROVIDER)
if not isinstance(record, dict) or record.get("type") != "api_key":
return None
key = record.get("key")
if not isinstance(key, str) or not key:
return None
return record
def is_authenticated() -> bool:
return read_record() is not None
def save_api_key(key: str) -> None:
codex.save_provider_record(PROVIDER, {"type": "api_key", "provider": PROVIDER, "key": key})
def logout() -> None:
codex.remove_provider_record(PROVIDER)
def get_api_key() -> str:
record = read_record()
if record is None:
raise OpencodeAuthError(
"not_authenticated", "not signed in; run: strix auth login opencode"
)
return str(record["key"])
def validate_api_key(key: str) -> None:
"""Check the key against the gateway's models endpoint; raise if rejected."""
try:
response = requests.get(
f"{ZEN_BASE_URL}/models",
headers={"Authorization": f"Bearer {key}"},
timeout=_KEY_CHECK_TIMEOUT,
)
except requests.RequestException as exc:
raise OpencodeAuthError("unavailable", str(exc)) from exc
if response.status_code in (401, 403):
raise OpencodeAuthError(
"invalid_key", f"OpenCode rejected the API key (HTTP {response.status_code})"
)
if response.status_code >= 400:
raise OpencodeAuthError("http_error", f"HTTP {response.status_code}: {response.text[:300]}")
def build_openai_client(base_url: str) -> AsyncOpenAI:
return AsyncOpenAI(
api_key=get_api_key(),
base_url=base_url,
http_client=httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=30.0)),
)
_subscription_clients: dict[str, AsyncOpenAI] = {}
def get_subscription_client(base_url: str) -> AsyncOpenAI:
client = _subscription_clients.get(base_url)
if client is None:
client = build_openai_client(base_url)
_subscription_clients[base_url] = client
return client
def auth_mode(model_name: str | None) -> str:
"""Return "subscription" when STRIX_LLM runs on any subscription
(OpenCode or ChatGPT), else "api_key"."""
if subscription_model(model_name) or codex.subscription_model(model_name):
return "subscription"
return "api_key"
def subscription_provider(model_name: str | None) -> str | None:
"""The subscription behind STRIX_LLM: "opencode", "chatgpt", or None."""
if subscription_model(model_name):
return PROVIDER
if codex.subscription_model(model_name):
return "chatgpt"
return None
+5
View File
@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, Any
from agents.model_settings import ModelSettings from agents.model_settings import ModelSettings
from openai.types.shared import Reasoning from openai.types.shared import Reasoning
from strix.config import opencode
from strix.config.models import ( from strix.config.models import (
DEFAULT_MODEL_RETRY, DEFAULT_MODEL_RETRY,
OPENROUTER_ATTRIBUTION_HEADERS, OPENROUTER_ATTRIBUTION_HEADERS,
@@ -272,6 +273,10 @@ def _prompt_cache_extra_args(model_name: str) -> dict[str, Any] | None:
""" """
if not is_claude_model(model_name): if not is_claude_model(model_name):
return None return None
# OpenCode routes use the raw OpenAI SDK, which rejects this LiteLLM-only
# argument; the gateway applies Anthropic prompt caching itself.
if opencode.subscription_model(model_name):
return None
if is_bedrock_route(model_name) and not bedrock_route_supports_prompt_caching(model_name): if is_bedrock_route(model_name) and not bedrock_route_supports_prompt_caching(model_name):
return None return None
+105 -19
View File
@@ -1,8 +1,9 @@
"""`strix auth` — ChatGPT subscription sign-in (login / status / logout). """`strix auth` — subscription sign-in (login / status / logout).
Signing in only stores OAuth tokens (``~/.strix/subscription-auth.json``); model Signing in only stores credentials (``~/.strix/subscription-auth.json``); model
selection stays with ``STRIX_LLM``. A ``chatgpt/<model>`` STRIX_LLM runs on the selection stays with ``STRIX_LLM``. A ``chatgpt/<model>`` STRIX_LLM runs on the
subscription. ChatGPT subscription; ``opencode/<model>`` (Zen credits) or
``opencode-go/<model>`` (Go subscription) run on OpenCode.
""" """
from __future__ import annotations from __future__ import annotations
@@ -21,7 +22,7 @@ from rich.console import Console
from rich.panel import Panel from rich.panel import Panel
from rich.text import Text from rich.text import Text
from strix.config import codex, load_settings from strix.config import codex, load_settings, opencode
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -32,13 +33,20 @@ logger = logging.getLogger(__name__)
_CALLBACK_TIMEOUT_S = 300 _CALLBACK_TIMEOUT_S = 300
# CLI-facing name for the login provider. Internally this is the Codex OAuth # CLI-facing name for the default login provider. Internally this is the Codex
# flow (``codex.PROVIDER``), but users know it as ChatGPT, so that's what the # OAuth flow (``codex.PROVIDER``), but users know it as ChatGPT, so that's what
# command and messaging say. ``codex`` is accepted as an alias. # the command and messaging say. ``codex`` is accepted as an alias.
LOGIN_PROVIDER = "chatgpt" LOGIN_PROVIDER = "chatgpt"
_ACCEPTED_PROVIDERS = frozenset({LOGIN_PROVIDER, codex.PROVIDER}) _ACCEPTED_PROVIDERS = frozenset({LOGIN_PROVIDER, codex.PROVIDER})
_OPENCODE_PROVIDERS = frozenset({opencode.PROVIDER, "opencode-go", "zen"})
_USAGE = "Usage:\n strix auth login chatgpt [--manual]\n strix auth status\n strix auth logout" _USAGE = (
"Usage:\n"
" strix auth login chatgpt [--manual]\n"
" strix auth login opencode\n"
" strix auth status\n"
" strix auth logout [chatgpt|opencode]"
)
def run_auth(argv: list[str]) -> int: def run_auth(argv: list[str]) -> int:
@@ -55,7 +63,7 @@ def run_auth(argv: list[str]) -> int:
handlers: dict[str, Callable[[], int]] = { handlers: dict[str, Callable[[], int]] = {
"login": lambda: _login(console, rest), "login": lambda: _login(console, rest),
"status": lambda: _status(console), "status": lambda: _status(console),
"logout": lambda: _logout(console), "logout": lambda: _logout(console, rest),
} }
handler = handlers.get(subcommand) handler = handlers.get(subcommand)
if handler is not None: if handler is not None:
@@ -84,10 +92,14 @@ def _login(console: Console, argv: list[str]) -> int:
except SystemExit as exc: # argparse already printed the message except SystemExit as exc: # argparse already printed the message
return int(exc.code or 2) return int(exc.code or 2)
if args.provider.lower() in _OPENCODE_PROVIDERS:
return _login_opencode(console)
if args.provider.lower() not in _ACCEPTED_PROVIDERS: if args.provider.lower() not in _ACCEPTED_PROVIDERS:
console.print( console.print(
f"[red]Unsupported provider:[/] {args.provider}. " f"[red]Unsupported provider:[/] {args.provider}. "
f"Only '{LOGIN_PROVIDER}' (ChatGPT subscription) is supported." f"Supported: '{LOGIN_PROVIDER}' (ChatGPT subscription) and "
f"'{opencode.PROVIDER}' (OpenCode Zen/Go)."
) )
return 2 return 2
@@ -115,6 +127,63 @@ def _login(console: Console, argv: list[str]) -> int:
return 0 return 0
def _login_opencode(console: Console) -> int:
console.print()
console.print("[bold]Signing in with OpenCode[/] [dim](provider: opencode)[/]")
console.print(
"[dim]This uses your OpenCode Zen credits or Go subscription for inference.\n"
f"Get your API key at {opencode.AUTH_CONSOLE_URL}[/]"
)
console.print()
try:
key = console.input("Paste your OpenCode API key: ", password=True).strip()
except (EOFError, KeyboardInterrupt):
console.print("\n[yellow]Sign-in cancelled.[/]")
return 130
if not key:
console.print("[red]No API key provided.[/]")
return 2
try:
opencode.validate_api_key(key)
except opencode.OpencodeAuthError as exc:
console.print(f"[red]SIGN-IN FAILED:[/] {exc}")
return 1
opencode.save_api_key(key)
_print_opencode_success(console)
return 0
def _print_opencode_success(console: Console) -> None:
text = Text()
text.append("Signed in with your OpenCode account", style="bold #22c55e")
text.append("\n\n", style="white")
text.append("Set ", style="white")
text.append("STRIX_LLM", style="bold white")
text.append(" to an ", style="white")
text.append("opencode/", style="bold cyan")
text.append(" model (e.g. ", style="white")
text.append("opencode/claude-sonnet-5", style="bold cyan")
text.append(") to run on Zen credits, or ", style="white")
text.append("opencode-go/", style="bold cyan")
text.append(" (e.g. ", style="white")
text.append("opencode-go/kimi-k3", style="bold cyan")
text.append(") to run on the Go subscription.", style="white")
text.append("\n\n", style="white")
text.append("Run a scan as usual, e.g. ", style="white")
text.append("strix --target https://example.com", style="bold cyan")
console.print()
console.print(
Panel(
text,
title="[bold white]STRIX",
title_align="left",
border_style="#22c55e",
padding=(1, 2),
)
)
console.print()
def _run_oauth_flow( def _run_oauth_flow(
console: Console, console: Console,
authorize_url: str, authorize_url: str,
@@ -244,24 +313,41 @@ def _first(query: dict[str, list[str]], key: str) -> str | None:
def _status(console: Console) -> int: def _status(console: Console) -> int:
record = codex.read_record() record = codex.read_record()
if record is None: opencode_signed_in = opencode.is_authenticated()
console.print("[yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] to sign in.") if record is None and not opencode_signed_in:
console.print(
"[yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] or "
"[cyan]strix auth login opencode[/] to sign in."
)
return 1 return 1
settings = load_settings() settings = load_settings()
console.print("[green]Signed in[/] with a ChatGPT subscription.") if record is not None:
console.print(f" Account: [bold]{record.get('account_id')}[/]") console.print("[green]Signed in[/] with a ChatGPT subscription.")
if codex.subscription_model(settings.llm.model): console.print(f" Account: [bold]{record.get('account_id')}[/]")
if opencode_signed_in:
console.print("[green]Signed in[/] with an OpenCode account.")
if codex.subscription_model(settings.llm.model) or opencode.subscription_model(
settings.llm.model
):
console.print(f" Runs use the subscription (STRIX_LLM=[bold]{settings.llm.model}[/]).") console.print(f" Runs use the subscription (STRIX_LLM=[bold]{settings.llm.model}[/]).")
else: else:
console.print( console.print(
" [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. [cyan]chatgpt/gpt-5.4[/] " " [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. [cyan]chatgpt/gpt-5.4[/] or "
"to run on the subscription." "[cyan]opencode/claude-sonnet-5[/] to run on a subscription."
) )
return 0 return 0
def _logout(console: Console) -> int: def _logout(console: Console, argv: list[str] | None = None) -> int:
codex.logout() target = (argv[0].lower() if argv else "") or "all"
if target in _ACCEPTED_PROVIDERS or target == "all":
codex.logout()
if target in _OPENCODE_PROVIDERS or target == "all":
opencode.logout()
if target != "all" and target not in _ACCEPTED_PROVIDERS | _OPENCODE_PROVIDERS:
console.print(f"[red]Unknown provider:[/] {target}\n")
console.print(_USAGE)
return 2
console.print("[green]Signed out.[/] Stored subscription credentials removed.") console.print("[green]Signed out.[/] Stored subscription credentials removed.")
return 0 return 0
+11 -1
View File
@@ -8,7 +8,7 @@ from rich.console import Console
from rich.panel import Panel from rich.panel import Panel
from rich.text import Text from rich.text import Text
from strix.config import codex, load_settings from strix.config import codex, load_settings, opencode
from strix.interface.utils import ( from strix.interface.utils import (
check_docker_connection, check_docker_connection,
image_exists, image_exists,
@@ -37,6 +37,16 @@ def validate_environment() -> None:
logger.info("Environment OK (ChatGPT subscription)") logger.info("Environment OK (ChatGPT subscription)")
return return
if opencode.subscription_model(settings.llm.model):
if not opencode.is_authenticated():
console.print(
f"[red]STRIX_LLM={settings.llm.model} uses your OpenCode subscription, "
"but you're not signed in.[/] Run [cyan]strix auth login opencode[/] first."
)
sys.exit(1)
logger.info("Environment OK (OpenCode subscription)")
return
if not settings.llm.model: if not settings.llm.model:
missing_required_vars.append("STRIX_LLM") missing_required_vars.append("STRIX_LLM")
+9 -3
View File
@@ -14,7 +14,7 @@ from rich.console import Console
from rich.panel import Panel from rich.panel import Panel
from rich.text import Text from rich.text import Text
from strix.config import codex, load_settings, persist_current from strix.config import codex, load_settings, opencode, persist_current
from strix.core.paths import run_dir_for from strix.core.paths import run_dir_for
from strix.interface.cli_args import parse_arguments from strix.interface.cli_args import parse_arguments
from strix.interface.environment import ( from strix.interface.environment import (
@@ -104,8 +104,14 @@ def _provider_import_hint(exc: BaseException, model: str) -> str | None:
def _subscription_error_hint(exc: BaseException) -> str | None: def _subscription_error_hint(exc: BaseException) -> str | None:
"""Return an actionable hint for a known ChatGPT-subscription error, or None.""" """Return an actionable hint for a known subscription error, or None."""
if not codex.subscription_model(load_settings().llm.model): model = load_settings().llm.model
if opencode.subscription_model(model):
joined = " ".join(_exception_messages(exc)).lower()
if "error code: 401" in joined or "http 401" in joined or "unauthorized" in joined:
return "Your OpenCode API key was rejected. Sign in again:\n strix auth login opencode"
return None
if not codex.subscription_model(model):
return None return None
joined = " ".join(_exception_messages(exc)).lower() joined = " ".join(_exception_messages(exc)).lower()
if "not supported when using codex with a chatgpt account" in joined: if "not supported when using codex with a chatgpt account" in joined:
+4 -3
View File
@@ -14,7 +14,7 @@ import logging
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from strix.config import Settings, codex, load_settings from strix.config import Settings, load_settings, opencode
from strix.core.paths import run_dir_for from strix.core.paths import run_dir_for
from strix.interface.utils import ( from strix.interface.utils import (
assign_workspace_subdirs, assign_workspace_subdirs,
@@ -226,7 +226,7 @@ def telemetry_start(args: argparse.Namespace) -> None:
model = load_settings().llm.model model = load_settings().llm.model
kwargs = { kwargs = {
"model": model, "model": model,
"auth_mode": codex.auth_mode(model), "auth_mode": opencode.auth_mode(model),
"scan_mode": args.scan_mode, "scan_mode": args.scan_mode,
"is_whitebox": is_whitebox_scan(args.targets_info), "is_whitebox": is_whitebox_scan(args.targets_info),
"interactive": not args.non_interactive, "interactive": not args.non_interactive,
@@ -247,7 +247,8 @@ def _persist_run_record(args: argparse.Namespace) -> None:
"status": "running", "status": "running",
"start_time": datetime.now(UTC).isoformat(), "start_time": datetime.now(UTC).isoformat(),
"end_time": None, "end_time": None,
"auth_mode": codex.auth_mode(load_settings().llm.model), "auth_mode": opencode.auth_mode(load_settings().llm.model),
"subscription_provider": opencode.subscription_provider(load_settings().llm.model),
"targets_info": args.targets_info, "targets_info": args.targets_info,
"scan_mode": args.scan_mode, "scan_mode": args.scan_mode,
"instruction": args.instruction, "instruction": args.instruction,
+6 -1
View File
@@ -24,7 +24,7 @@ from strix.interface.tui.backend.projection import (
sanitize_terminal_text, sanitize_terminal_text,
terminal_projection, terminal_projection,
) )
from strix.interface.utils import is_subscription_run from strix.interface.utils import is_subscription_run, subscription_label
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -164,6 +164,10 @@ class TuiController:
subscription = False subscription = False
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
subscription = is_subscription_run(self.report_state) subscription = is_subscription_run(self.report_state)
label = ""
if subscription:
with contextlib.suppress(Exception):
label = subscription_label()
model_warning = "" model_warning = ""
if model and not is_recommended_or_frontier_model(model): if model and not is_recommended_or_frontier_model(model):
model_warning = ( model_warning = (
@@ -200,6 +204,7 @@ class TuiController:
], ],
"usage": terminal_projection(usage, max_string=256, max_items=20), "usage": terminal_projection(usage, max_string=256, max_items=20),
"subscription": subscription, "subscription": subscription,
"subscription_label": label,
"viewer_status": self.viewer_status, "viewer_status": self.viewer_status,
"viewer_url": terminal_projection(self.viewer_url, max_string=1024), "viewer_url": terminal_projection(self.viewer_url, max_string=1024),
"error": terminal_projection(self.error, max_string=2 * 1024), "error": terminal_projection(self.error, max_string=2 * 1024),
+5 -1
View File
@@ -596,7 +596,11 @@ func (m Model) statsView() string {
if b.Len() > 0 { if b.Len() > 0 {
b.WriteString("\n") b.WriteString("\n")
} }
b.WriteString(lipgloss.NewStyle().Foreground(green).Render("ChatGPT subscription")) label := m.snapshot.SubscriptionLabel
if label == "" {
label = "ChatGPT subscription"
}
b.WriteString(lipgloss.NewStyle().Foreground(green).Render(label))
} }
total := numberValue(m.snapshot.Usage["total_tokens"]) total := numberValue(m.snapshot.Usage["total_tokens"])
if total > 0 { if total > 0 {
@@ -68,6 +68,7 @@ type Snapshot struct {
Vulnerabilities []map[string]any `json:"-"` Vulnerabilities []map[string]any `json:"-"`
Usage map[string]any `json:"usage"` Usage map[string]any `json:"usage"`
Subscription bool `json:"subscription"` Subscription bool `json:"subscription"`
SubscriptionLabel string `json:"subscription_label"`
ViewerStatus string `json:"viewer_status"` ViewerStatus string `json:"viewer_status"`
ViewerURL *string `json:"viewer_url"` ViewerURL *string `json:"viewer_url"`
Error *string `json:"error"` Error *string `json:"error"`
+14 -4
View File
@@ -262,9 +262,19 @@ def is_subscription_run(report_state: Any) -> bool:
record = getattr(report_state, "run_record", None) record = getattr(report_state, "run_record", None)
if isinstance(record, dict) and record.get("auth_mode"): if isinstance(record, dict) and record.get("auth_mode"):
return record.get("auth_mode") == "subscription" return record.get("auth_mode") == "subscription"
from strix.config import codex from strix.config import opencode
return codex.auth_mode(load_settings().llm.model) == "subscription" return opencode.auth_mode(load_settings().llm.model) == "subscription"
def subscription_label() -> str:
"""Display name of the subscription behind the configured model."""
from strix.config import opencode
model = load_settings().llm.model
if opencode.subscription_model(model):
return "OpenCode subscription"
return "ChatGPT subscription"
def _int_stat(usage: dict[str, Any], key: str) -> int: def _int_stat(usage: dict[str, Any], key: str) -> int:
@@ -368,7 +378,7 @@ def build_live_stats_text(report_state: Any) -> Text:
stats_text.append(str(model), style="white") stats_text.append(str(model), style="white")
if is_subscription_run(report_state): if is_subscription_run(report_state):
stats_text.append(" · ", style="dim white") stats_text.append(" · ", style="dim white")
stats_text.append("ChatGPT subscription", style="#22c55e") stats_text.append(subscription_label(), style="#22c55e")
stats_text.append("\n") stats_text.append("\n")
vuln_count = len(report_state.vulnerability_reports) vuln_count = len(report_state.vulnerability_reports)
@@ -414,7 +424,7 @@ def build_tui_stats_text(report_state: Any) -> Text:
subscription = is_subscription_run(report_state) subscription = is_subscription_run(report_state)
if subscription: if subscription:
stats_text.append("\n") stats_text.append("\n")
stats_text.append("ChatGPT subscription", style="#22c55e") stats_text.append(subscription_label(), style="#22c55e")
usage = _llm_usage(report_state) usage = _llm_usage(report_state)
if usage and _int_stat(usage, "total_tokens") > 0: if usage and _int_stat(usage, "total_tokens") > 0:
@@ -101,6 +101,11 @@ export function RunDetails({
const totalTokens = num(usage.total_tokens); const totalTokens = num(usage.total_tokens);
const cost = num(usage.cost); const cost = num(usage.cost);
const subscription = str(raw.auth_mode) === "subscription"; const subscription = str(raw.auth_mode) === "subscription";
const subscriptionProvider =
str(raw.subscription_provider) ??
(models.some((m) => m.toLowerCase().startsWith("opencode")) ? "opencode" : "chatgpt");
const subscriptionLabel =
subscriptionProvider === "opencode" ? "OpenCode subscription" : "ChatGPT subscription";
const sub = (n: number, word: string) => ( const sub = (n: number, word: string) => (
<span className="text-[#666]"> ({formatNumber(n)} {word})</span> <span className="text-[#666]"> ({formatNumber(n)} {word})</span>
@@ -180,7 +185,7 @@ export function RunDetails({
<Field label="Provider"> <Field label="Provider">
<span className="inline-flex items-center gap-1.5"> <span className="inline-flex items-center gap-1.5">
<span className="rounded-full border border-[#22c55e]/40 bg-[#22c55e]/10 px-2 py-0.5 text-[11px] text-[#22c55e]"> <span className="rounded-full border border-[#22c55e]/40 bg-[#22c55e]/10 px-2 py-0.5 text-[11px] text-[#22c55e]">
ChatGPT subscription {subscriptionLabel}
</span> </span>
</span> </span>
</Field> </Field>
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -6,7 +6,7 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0" /> <meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="color-scheme" content="dark" /> <meta name="color-scheme" content="dark" />
<title>Strix Results</title> <title>Strix Results</title>
<script type="module" crossorigin src="./assets/index-DBJ-RJqo.js"></script> <script type="module" crossorigin src="./assets/index-1LIW3rcB.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-DKbLYAbP.css"> <link rel="stylesheet" crossorigin href="./assets/index-DKbLYAbP.css">
</head> </head>
<body> <body>
+7 -1
View File
@@ -20,6 +20,8 @@ logger = logging.getLogger(__name__)
_STRIPPABLE_PREFIXES = ( _STRIPPABLE_PREFIXES = (
"openai/", "openai/",
"chatgpt/", "chatgpt/",
"opencode-go/",
"opencode/",
"litellm/", "litellm/",
"any-llm/", "any-llm/",
"ollama/", "ollama/",
@@ -48,7 +50,11 @@ def _model_info(model: str) -> dict[str, int]:
lookup_key = _lookup_key(model) lookup_key = _lookup_key(model)
# Provider-qualified ChatGPT lookups may start a synchronous device-login # Provider-qualified ChatGPT lookups may start a synchronous device-login
# poll. LiteLLM keys the metadata by the underlying model slug. # poll. LiteLLM keys the metadata by the underlying model slug.
candidates = (lookup_key,) if model.startswith("chatgpt/") else (model, lookup_key) candidates = (
(lookup_key,)
if model.startswith(("chatgpt/", "opencode/", "opencode-go/"))
else (model, lookup_key)
)
for candidate in candidates: for candidate in candidates:
info = _safe_get_model_info(candidate) info = _safe_get_model_info(candidate)
if info is not None: if info is not None:
+54
View File
@@ -0,0 +1,54 @@
"""LiteLLM model-name resolution for local cost estimates."""
from __future__ import annotations
from functools import lru_cache
from typing import Any, cast
@lru_cache(maxsize=512)
def resolve_litellm_model(model: str) -> str | None:
"""Return a provider-qualified model name that LiteLLM can price."""
try:
import litellm
normalized = model.strip()
for prefix in ("litellm/", "any-llm/", "openai/"):
if normalized.startswith(prefix):
normalized = normalized.removeprefix(prefix)
break
if not normalized:
return None
model_cost = cast(
"dict[str, dict[str, Any]]",
getattr(litellm, "model_cost"), # noqa: B009
)
bare_entry = model_cost.get(normalized)
if "/" not in normalized and isinstance(bare_entry, dict):
provider = bare_entry.get("litellm_provider")
if isinstance(provider, str) and provider:
return f"{provider}/{normalized}"
if "/" in normalized and isinstance(bare_entry, dict):
return normalized
names = [normalized]
if "/" in normalized:
names.append(normalized.rsplit("/", 1)[-1])
for name in names:
matches = sorted(key for key in model_cost if key.endswith(f"/{name}"))
if not matches:
continue
prices = {
(
model_cost[key].get("input_cost_per_token"),
model_cost[key].get("output_cost_per_token"),
)
for key in matches
if isinstance(model_cost.get(key), dict)
}
if len(matches) == 1 or len(prices) == 1:
return matches[0]
return None # noqa: TRY300
except Exception: # noqa: BLE001
return None
+9 -4
View File
@@ -11,9 +11,10 @@ from uuid import uuid4
from agents.usage import Usage from agents.usage import Usage
from strix.config import codex from strix.config import opencode
from strix.config.loader import load_settings from strix.config.loader import load_settings
from strix.core.paths import run_dir_for from strix.core.paths import run_dir_for
from strix.report.pricing import resolve_litellm_model
from strix.report.sarif import write_sarif from strix.report.sarif import write_sarif
from strix.report.usage import LLMUsageLedger from strix.report.usage import LLMUsageLedger
from strix.report.writer import ( from strix.report.writer import (
@@ -122,7 +123,7 @@ class ReportState:
self.scan_results: dict[str, Any] | None = None self.scan_results: dict[str, Any] | None = None
self.scan_config: dict[str, Any] | None = None self.scan_config: dict[str, Any] | None = None
self._llm_usage = LLMUsageLedger() self._llm_usage = LLMUsageLedger()
auth_mode = codex.auth_mode(load_settings().llm.model) auth_mode = opencode.auth_mode(load_settings().llm.model)
self._llm_usage.zero_cost = auth_mode == "subscription" self._llm_usage.zero_cost = auth_mode == "subscription"
self.run_record: dict[str, Any] = { self.run_record: dict[str, Any] = {
"run_id": self.run_id, "run_id": self.run_id,
@@ -131,6 +132,7 @@ class ReportState:
"end_time": None, "end_time": None,
"status": "running", "status": "running",
"auth_mode": auth_mode, "auth_mode": auth_mode,
"subscription_provider": opencode.subscription_provider(load_settings().llm.model),
"targets_info": [], "targets_info": [],
"llm_usage": self._build_llm_usage_record(), "llm_usage": self._build_llm_usage_record(),
} }
@@ -696,10 +698,13 @@ def _estimate_response_cost(kwargs: Any, completion_response: Any) -> float | No
candidates.append(model.rsplit("/", 1)[-1]) candidates.append(model.rsplit("/", 1)[-1])
for candidate in candidates: for candidate in candidates:
resolved = resolve_litellm_model(candidate)
if not resolved:
continue
try: try:
value = completion_cost( value = completion_cost(
completion_response={"model": candidate, "usage": usage_payload}, completion_response={"model": resolved, "usage": usage_payload},
model=candidate, model=resolved,
) )
except Exception: # nosec B112 # noqa: BLE001, S112 except Exception: # nosec B112 # noqa: BLE001, S112
continue continue
+30 -29
View File
@@ -7,6 +7,8 @@ from typing import Any
from agents.usage import Usage, deserialize_usage, serialize_usage from agents.usage import Usage, deserialize_usage, serialize_usage
from strix.report.pricing import resolve_litellm_model
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -18,7 +20,9 @@ class LLMUsageLedger:
self._total_usage = Usage() self._total_usage = Usage()
self._agent_usage: dict[str, Usage] = {} self._agent_usage: dict[str, Usage] = {}
self._agent_metadata: dict[str, dict[str, str]] = {} self._agent_metadata: dict[str, dict[str, str]] = {}
self._total_cost = 0.0 self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
# When True, tokens are still tracked but cost stays $0 — the run is on a # When True, tokens are still tracked but cost stays $0 — the run is on a
# model subscription, so there is no metered per-token charge to report. # model subscription, so there is no metered per-token charge to report.
self.zero_cost = False self.zero_cost = False
@@ -44,10 +48,10 @@ class LLMUsageLedger:
if model: if model:
metadata["model"] = model metadata["model"] = model
if not self.zero_cost and not _is_litellm_routed(model): if not self.zero_cost:
estimated = _estimate_litellm_cost(usage, model) estimated = _estimate_litellm_cost(usage, model)
if estimated: if estimated:
self._total_cost += estimated self._estimated_cost += estimated
return True return True
@@ -55,15 +59,18 @@ class LLMUsageLedger:
if self.zero_cost: if self.zero_cost:
return return
if isinstance(cost, int | float) and cost > 0: if isinstance(cost, int | float) and cost > 0:
self._total_cost += float(cost) self._observed_cost += float(cost)
self._has_observed_cost = True
@property @property
def total_cost(self) -> float: def total_cost(self) -> float:
return _round_cost(self._total_cost) if self.zero_cost:
return 0.0
return _round_cost(self._observed_cost if self._has_observed_cost else self._estimated_cost)
def to_record(self) -> dict[str, Any]: def to_record(self) -> dict[str, Any]:
record = serialize_usage(self._total_usage) record = serialize_usage(self._total_usage)
record["cost"] = _round_cost(self._total_cost) record["cost"] = self.total_cost
record["agents"] = [] record["agents"] = []
agent_tokens = {aid: _resolve_total_tokens(u) for aid, u in self._agent_usage.items()} agent_tokens = {aid: _resolve_total_tokens(u) for aid, u in self._agent_usage.items()}
@@ -72,7 +79,7 @@ class LLMUsageLedger:
usage = self._agent_usage[agent_id] usage = self._agent_usage[agent_id]
metadata = self._agent_metadata.get(agent_id, {}) metadata = self._agent_metadata.get(agent_id, {})
agent_cost = ( agent_cost = (
self._total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0 self.total_cost * (agent_tokens[agent_id] / total_tokens) if total_tokens else 0.0
) )
agent_record = serialize_usage(usage) agent_record = serialize_usage(usage)
@@ -92,7 +99,9 @@ class LLMUsageLedger:
self._total_usage = Usage() self._total_usage = Usage()
self._agent_usage.clear() self._agent_usage.clear()
self._agent_metadata.clear() self._agent_metadata.clear()
self._total_cost = 0.0 self._observed_cost = 0.0
self._estimated_cost = 0.0
self._has_observed_cost = False
if not isinstance(raw_usage, dict): if not isinstance(raw_usage, dict):
return return
@@ -103,7 +112,9 @@ class LLMUsageLedger:
logger.exception("Failed to hydrate aggregate llm_usage from run.json") logger.exception("Failed to hydrate aggregate llm_usage from run.json")
self._total_usage = Usage() self._total_usage = Usage()
self._total_cost = _float_or_zero(raw_usage.get("cost")) persisted_cost = _float_or_zero(raw_usage.get("cost"))
self._observed_cost = persisted_cost
self._estimated_cost = persisted_cost
for raw_agent in raw_usage.get("agents") or []: for raw_agent in raw_usage.get("agents") or []:
if not isinstance(raw_agent, dict): if not isinstance(raw_agent, dict):
@@ -136,15 +147,6 @@ def _resolve_total_tokens(usage: Usage) -> int:
return prompt + completion return prompt + completion
def _is_litellm_routed(model: str | None) -> bool:
if not model:
return False
name = model.strip().lower()
if "/" not in name:
return False
return not name.startswith("openai/")
def _usage_has_activity(usage: Usage) -> bool: def _usage_has_activity(usage: Usage) -> bool:
return bool( return bool(
usage.requests usage.requests
@@ -201,24 +203,23 @@ def _estimate_litellm_entry_cost(entry: Any, model: str) -> float | None:
candidates = [model] candidates = [model]
if "/" in model: if "/" in model:
candidates.append(model.split("/", 1)[-1]) candidates.append(model.rsplit("/", 1)[-1])
cost: Any = None
for candidate in candidates: for candidate in candidates:
resolved = resolve_litellm_model(candidate)
if not resolved:
continue
try: try:
cost = completion_cost( cost = completion_cost(
completion_response={"model": candidate, "usage": usage_payload}, completion_response={"model": resolved, "usage": usage_payload},
model=model, model=resolved,
) )
break
except Exception: # nosec B112 # noqa: BLE001, S112 except Exception: # nosec B112 # noqa: BLE001, S112
continue continue
if cost > 0:
if cost is None: return float(cost)
logger.debug("LiteLLM cost estimate unavailable for model %s", model) logger.debug("LiteLLM cost estimate unavailable for model %s", model)
return None return None
return cost if isinstance(cost, int | float) and cost >= 0 else None
def _litellm_model_name(model: str | None) -> str | None: def _litellm_model_name(model: str | None) -> str | None:
+59 -1
View File
@@ -2,11 +2,12 @@
from __future__ import annotations from __future__ import annotations
import time
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
import pytest import pytest
from strix.config import codex from strix.config import codex, opencode
from strix.interface import auth_cli from strix.interface import auth_cli
@@ -104,3 +105,60 @@ def test_login_accepts_provider_aliases(provider: str, monkeypatch: pytest.Monke
assert auth_cli.run_auth(["login", provider]) == 0 assert auth_cli.run_auth(["login", provider]) == 0
assert reached["flow"] is True assert reached["flow"] is True
@pytest.mark.parametrize("provider", ["opencode", "OpenCode", "opencode-go", "zen"])
def test_login_accepts_opencode_aliases(provider: str, monkeypatch: pytest.MonkeyPatch) -> None:
reached = {"login": False}
def _fake_login(_console: Any) -> int:
reached["login"] = True
return 0
monkeypatch.setattr(auth_cli, "_login_opencode", _fake_login)
assert auth_cli.run_auth(["login", provider]) == 0
assert reached["login"] is True
def test_login_opencode_validates_and_saves(monkeypatch: pytest.MonkeyPatch) -> None:
saved: dict[str, str] = {}
monkeypatch.setattr("rich.console.Console.input", lambda _self, *_a, **_k: " sk-oc-test ")
monkeypatch.setattr(opencode, "validate_api_key", lambda key: saved.setdefault("checked", key))
monkeypatch.setattr(opencode, "save_api_key", lambda key: saved.setdefault("key", key))
assert auth_cli.run_auth(["login", "opencode"]) == 0
assert saved == {"checked": "sk-oc-test", "key": "sk-oc-test"}
def test_login_opencode_rejects_bad_key(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("rich.console.Console.input", lambda _self, *_a, **_k: "bad")
def _reject(_key: str) -> None:
raise opencode.OpencodeAuthError("invalid_key")
monkeypatch.setattr(opencode, "validate_api_key", _reject)
assert auth_cli.run_auth(["login", "opencode"]) == 1
assert opencode.is_authenticated() is False
def test_logout_provider_scoped() -> None:
codex.save_record(
{
"type": "oauth",
"provider": "codex",
"access": "a",
"refresh": "r",
"account_id": "acct",
"expires_at": time.time() + 3600,
}
)
opencode.save_api_key("sk-oc-test")
assert auth_cli.run_auth(["logout", "opencode"]) == 0
assert opencode.is_authenticated() is False
assert codex.is_authenticated() is True
assert auth_cli.run_auth(["logout"]) == 0
assert codex.is_authenticated() is False
assert auth_cli.run_auth(["logout", "bogus"]) == 2
+1 -1
View File
@@ -143,7 +143,7 @@ def test_cost_callback_estimates_cost_with_bare_model_fallback() -> None:
} }
def fake_completion_cost(**kwargs: object) -> float: def fake_completion_cost(**kwargs: object) -> float:
if kwargs["model"] == "gpt-4o-mini": if kwargs["model"] == "openai/gpt-4o-mini":
return 0.025 return 0.025
raise ValueError(kwargs["model"]) raise ValueError(kwargs["model"])
+7
View File
@@ -111,6 +111,13 @@ def test_make_model_settings_no_prompt_cache_for_non_claude(model_name: str) ->
assert make_model_settings(None, model_name=model_name).extra_args is None assert make_model_settings(None, model_name=model_name).extra_args is None
@pytest.mark.parametrize("model_name", ["opencode/claude-sonnet-5", "opencode-go/claude-sonnet-5"])
def test_no_prompt_cache_for_opencode_claude(model_name: str) -> None:
# The OpenCode route uses the raw OpenAI SDK, whose create() rejects the
# LiteLLM-only cache_control_injection_points argument.
assert _cache_points(model_name) is None
def test_no_prompt_cache_for_unmapped_bedrock_claude_model(monkeypatch: Any) -> None: def test_no_prompt_cache_for_unmapped_bedrock_claude_model(monkeypatch: Any) -> None:
# A Bedrock Claude model LiteLLM hasn't mapped must run uncached, not crash. # A Bedrock Claude model LiteLLM hasn't mapped must run uncached, not crash.
unmapped = "bedrock/global.anthropic.claude-brand-new-9" unmapped = "bedrock/global.anthropic.claude-brand-new-9"
+5
View File
@@ -66,6 +66,11 @@ def test_recommended_models_are_matched_case_insensitively() -> None:
"moonshot/kimi-k2.6", "moonshot/kimi-k2.6",
"kimi-k2.7-code", "kimi-k2.7-code",
"moonshot/kimi-k3", "moonshot/kimi-k3",
"opencode/gpt-5.4",
"opencode/claude-sonnet-5",
"opencode-go/kimi-k3",
"opencode-go/deepseek-v4-flash",
"opencode-go/qwen3.8-max",
], ],
) )
def test_frontier_model_families_are_accepted(model_name: str) -> None: def test_frontier_model_families_are_accepted(model_name: str) -> None:
+133
View File
@@ -0,0 +1,133 @@
"""Tests for OpenCode (Zen/Go) subscription auth: prefix parsing and key store."""
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest import mock
import pytest
import requests
from strix.config import codex, opencode
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _tmp_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
path = tmp_path / "home" / ".strix" / "subscription-auth.json"
monkeypatch.setattr(codex, "AUTH_PATH", path)
return path
@pytest.mark.parametrize(
("model", "slug", "base_url", "uses_responses"),
[
("opencode/claude-sonnet-5", "claude-sonnet-5", opencode.ZEN_BASE_URL, False),
("opencode/gpt-5.4", "gpt-5.4", opencode.ZEN_BASE_URL, True),
("opencode/grok-4.5", "grok-4.5", opencode.ZEN_BASE_URL, True),
("OpenCode/Kimi-K3", "Kimi-K3", opencode.ZEN_BASE_URL, False),
("opencode-go/kimi-k3", "kimi-k3", opencode.GO_BASE_URL, False),
("opencode-go/gpt-5.6-luna", "gpt-5.6-luna", opencode.GO_BASE_URL, True),
("opencode-go/grok-4.5", "grok-4.5", opencode.GO_BASE_URL, False),
],
)
def test_subscription_model_parses_prefixes(
model: str, slug: str, base_url: str, uses_responses: bool
) -> None:
parsed = opencode.subscription_model(model)
assert parsed is not None
assert parsed.slug == slug
assert parsed.base_url == base_url
assert parsed.uses_responses == uses_responses
@pytest.mark.parametrize(
"model",
["openai/gpt-5.4", "chatgpt/gpt-5.4", "opencode/", "opencode-go/", "opencode", "", None],
)
def test_subscription_model_rejects_non_opencode(model: str | None) -> None:
assert opencode.subscription_model(model) is None
def test_store_roundtrip_and_logout() -> None:
assert opencode.read_record() is None
assert opencode.is_authenticated() is False
opencode.save_api_key("sk-oc-test")
record = opencode.read_record()
assert record is not None
assert record["key"] == "sk-oc-test"
assert opencode.is_authenticated() is True
assert opencode.get_api_key() == "sk-oc-test"
opencode.logout()
assert opencode.read_record() is None
opencode.logout() # no-op when already gone
def test_store_coexists_with_chatgpt_record() -> None:
codex.save_record({"type": "oauth", "access": "a", "refresh": "r", "account_id": "acct"})
opencode.save_api_key("sk-oc-test")
assert codex.read_record() is not None
assert opencode.get_api_key() == "sk-oc-test"
opencode.logout()
assert codex.read_record() is not None
assert opencode.read_record() is None
def test_get_api_key_raises_when_not_signed_in() -> None:
with pytest.raises(opencode.OpencodeAuthError) as exc:
opencode.get_api_key()
assert exc.value.code == "not_authenticated"
def test_auth_mode_covers_both_subscriptions() -> None:
assert opencode.auth_mode("opencode/claude-sonnet-5") == "subscription"
assert opencode.auth_mode("opencode-go/kimi-k3") == "subscription"
assert opencode.auth_mode("chatgpt/gpt-5.4") == "subscription"
assert opencode.auth_mode("openai/gpt-5.4") == "api_key"
assert opencode.auth_mode(None) == "api_key"
def test_subscription_provider() -> None:
assert opencode.subscription_provider("opencode/claude-sonnet-5") == "opencode"
assert opencode.subscription_provider("opencode-go/kimi-k3") == "opencode"
assert opencode.subscription_provider("chatgpt/gpt-5.4") == "chatgpt"
assert opencode.subscription_provider("openai/gpt-5.4") is None
assert opencode.subscription_provider(None) is None
def _response(status_code: int, text: str = "") -> mock.MagicMock:
response = mock.MagicMock()
response.status_code = status_code
response.text = text
return response
def test_validate_api_key_accepts_ok() -> None:
with mock.patch.object(requests, "get", return_value=_response(200)) as get:
opencode.validate_api_key("sk-oc-test")
assert get.call_args.kwargs["headers"]["Authorization"] == "Bearer sk-oc-test"
def test_validate_api_key_rejects_unauthorized() -> None:
with (
mock.patch.object(requests, "get", return_value=_response(401)),
pytest.raises(opencode.OpencodeAuthError) as exc,
):
opencode.validate_api_key("bad-key")
assert exc.value.code == "invalid_key"
def test_validate_api_key_maps_network_errors() -> None:
with (
mock.patch.object(requests, "get", side_effect=requests.ConnectionError("boom")),
pytest.raises(opencode.OpencodeAuthError) as exc,
):
opencode.validate_api_key("sk-oc-test")
assert exc.value.code == "unavailable"
+120
View File
@@ -0,0 +1,120 @@
from __future__ import annotations
from unittest.mock import patch
import litellm
from agents.usage import Usage
from strix.report.pricing import resolve_litellm_model
from strix.report.usage import LLMUsageLedger
def test_resolves_common_bare_model_names() -> None:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
assert resolve_litellm_model("openai/deepseek-v4-flash") == "deepseek/deepseek-v4-flash"
assert resolve_litellm_model("grok-4.5") == "xai/grok-4.5"
assert resolve_litellm_model("MiniMax-M3") == "minimax/MiniMax-M3"
def test_resolver_returns_none_for_unresolvable_model() -> None:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("provider/not-a-real-model") is None
def test_ledger_uses_estimate_when_routed_provider_reports_no_cost() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
with patch("litellm.completion_cost", return_value=0.42):
ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
assert ledger.total_cost == 0.42
def test_ledger_prefers_observed_cost_over_estimate() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
with patch("litellm.completion_cost", return_value=0.42):
ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
ledger.record_observed_cost(0.17)
assert ledger.total_cost == 0.17
def test_hydrated_estimate_continues_accumulating_new_estimates() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
ledger.hydrate({"cost": 0.42})
with patch("litellm.completion_cost", return_value=0.17):
ledger.record(agent_id="a", usage=usage, model="openai/deepseek-v4-flash")
assert ledger.total_cost == 0.59
def test_zero_cost_disables_both_observed_and_estimated_costs() -> None:
usage = Usage()
usage.requests = 1
usage.input_tokens = 1000
usage.output_tokens = 200
usage.total_tokens = 1200
ledger = LLMUsageLedger()
ledger.zero_cost = True
with patch("litellm.completion_cost", return_value=0.42) as estimate:
ledger.record(agent_id="a", usage=usage, model="deepseek-v4-flash")
ledger.record_observed_cost(1.0)
estimate.assert_not_called()
assert ledger.total_cost == 0.0
def test_resolver_uses_provider_when_bare_entry_has_one() -> None:
original = litellm.model_cost
litellm.model_cost = {
"example": {
"litellm_provider": "example-provider",
"input_cost_per_token": 1.0,
"output_cost_per_token": 2.0,
}
}
try:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("example") == "example-provider/example"
finally:
litellm.model_cost = original
resolve_litellm_model.cache_clear()
def test_resolver_does_not_guess_between_differently_priced_providers() -> None:
original = litellm.model_cost
litellm.model_cost = {
"provider-a/example": {
"input_cost_per_token": 1.0,
"output_cost_per_token": 2.0,
},
"provider-b/example": {
"input_cost_per_token": 3.0,
"output_cost_per_token": 4.0,
},
}
try:
resolve_litellm_model.cache_clear()
assert resolve_litellm_model("example") is None
finally:
litellm.model_cost = original
resolve_litellm_model.cache_clear()