muse_shim.py

#!/usr/bin/env python3
"""Loopback OpenAI Responses proxy for Muse Code.

Modes:

* ``auto`` advertises both Codex and Claude models to Muse and routes each
  request according to the model selected in Muse's ``/model`` menu.
* ``generic`` / ``http`` (the historical mode) forwards Responses requests
  to any Responses-compatible endpoint.
* ``codex`` uses the native ChatGPT Codex Responses endpoint and Codex CLI
  OAuth credentials.
* ``claude`` uses Claude Code print mode by default and adapts its structured
  result back to Responses for Muse. Direct Messages API mode remains optional.

The implementation is intentionally dependency-free. Codex uses its ordinary
provider transport; Claude CLI mode deliberately delegates inference to the
locally installed official Claude Code client.
"""

from __future__ import annotations

import argparse
import io
import ipaddress
import json
import os
import secrets
import shutil
import signal
import ssl
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Iterable

from muse_providers import (
    AuthError,
    CLAUDE_OAUTH_BETAS,
    CLAUDE_OAUTH_CLIENT_ID,
    DEFAULT_CLAUDE_BASE_URL,
    DEFAULT_CODEX_BASE_URL,
    CODEX_OAUTH_CLIENT_ID,
    DEFAULT_CLAUDE_ENDPOINT,
    DEFAULT_CODEX_ENDPOINT,
    ProviderError,
    anthropic_json_to_responses,
    claude_headers,
    convert_anthropic_sse,
    convert_anthropic_stream,
    codex_headers,
    discover_claude_credentials,
    discover_codex_credentials,
    endpoint_for,
    prepare_codex_request,
    refresh_claude_oauth,
    refresh_codex_oauth,
    resolve_claude_credentials,
    resolve_codex_credentials,
    responses_to_anthropic,
    translate_anthropic_sse,
    validate_claude_endpoint,
    validate_claude_token_url,
    validate_codex_endpoint,
    validate_codex_token_url,
)


DEFAULT_UPSTREAM = "https://openrouter.ai/api/v1"
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8787
DEFAULT_MAX_BODY_BYTES = 32 * 1024 * 1024
DEFAULT_CODEX_MODELS = (
    "gpt-5.6-sol",
    "gpt-5.6-terra",
    "gpt-5.5",
    "gpt-5.4",
)
DEFAULT_CLAUDE_MODELS = (
    "claude-fable-5",
    "claude-opus-5",
    "claude-sonnet-5",
    "claude-opus-4-8",
    "claude-haiku-4-5-20251001",
)
CLAUDE_MODEL_LIMITS = {
    "claude-fable-5": (1_000_000, 128_000),
    "claude-opus-5": (1_000_000, 128_000),
    "claude-sonnet-5": (1_000_000, 128_000),
    "claude-haiku-4-5-20251001": (200_000, 64_000),
}
CODEX_MODELS_ENDPOINT = (
    "https://chatgpt.com/backend-api/codex/models?client_version=1.0.0"
)
MODEL_CATALOG_TTL_SECONDS = 3600


class ShimError(Exception):
    """An error that should be returned to Muse as a client error."""


def env_flag(name: str, default: bool = False) -> bool:
    value = os.environ.get(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _parse_model_list(value: str | None, fallback: tuple[str, ...]) -> list[str]:
    models = [item.strip() for item in str(value or "").split(",") if item.strip()]
    return list(dict.fromkeys(models or fallback))


def _is_loopback_host(host: str) -> bool:
    value = str(host or "").strip().lower()
    if value == "localhost":
        return True
    try:
        return ipaddress.ip_address(value).is_loopback
    except ValueError:
        return False


def _claude_endpoint_from_upstream(upstream: str) -> str:
    value = str(upstream or "").rstrip("/")
    if value.endswith("/messages"):
        return value
    if value.endswith("/v1"):
        return value + "/messages"
    return value + "/v1/messages"


def _validate_native_config(
    *,
    provider: str,
    upstream: str,
    codex_token_url: str | None,
    claude_token_url: str | None,
    insecure_upstream: bool,
    allow_oauth_host_override: bool,
) -> None:
    if provider not in {"codex", "claude"}:
        return
    if insecure_upstream:
        raise ProviderError(
            f"--insecure-upstream is not permitted in {provider} OAuth/native mode.",
            code="insecure_native_upstream_rejected",
        )
    if provider == "codex":
        validate_codex_endpoint(
            endpoint_for(upstream, "responses"),
            allow_host_override=allow_oauth_host_override,
        )
        if codex_token_url:
            validate_codex_token_url(
                codex_token_url,
                allow_host_override=allow_oauth_host_override,
            )
    else:
        validate_claude_endpoint(
            _claude_endpoint_from_upstream(upstream),
            allow_host_override=allow_oauth_host_override,
        )
        if claude_token_url:
            validate_claude_token_url(
                claude_token_url,
                allow_host_override=allow_oauth_host_override,
            )


def flatten_namespace_tools(tools: list[Any]) -> list[Any]:
    """Flatten Muse's namespace tool groups into Responses function tools."""
    flattened: list[Any] = []
    names: set[str] = set()

    for tool in tools:
        if not isinstance(tool, dict) or tool.get("type") != "namespace":
            flattened.append(tool)
            if isinstance(tool, dict) and isinstance(tool.get("name"), str):
                if tool["name"] in names:
                    raise ShimError(f"duplicate tool name {tool['name']!r}")
                names.add(tool["name"])
            continue

        namespace = tool.get("name", "<unnamed>")
        children = tool.get("tools")
        if not isinstance(children, list):
            raise ShimError(f"Muse namespace {namespace!r} has no tools list")

        for child in children:
            if not isinstance(child, dict):
                raise ShimError(f"Muse namespace {namespace!r} contains a non-object tool")
            name = child.get("name")
            if not isinstance(name, str) or not name:
                raise ShimError(f"Muse namespace {namespace!r} contains an unnamed tool")
            if name in names:
                raise ShimError(
                    f"duplicate tool name {name!r} after flattening namespace {namespace!r}"
                )
            names.add(name)
            flattened.append(child)

    return flattened


def transform_request(body: dict[str, Any], config: "Config") -> dict[str, Any]:
    transformed = dict(body)

    tools = transformed.get("tools")
    if isinstance(tools, list):
        transformed["tools"] = flatten_namespace_tools(tools)

    if config.model:
        transformed["model"] = config.model

    if config.max_output_tokens is not None:
        requested = transformed.get("max_output_tokens")
        if isinstance(requested, (int, float)) and not isinstance(requested, bool):
            transformed["max_output_tokens"] = min(int(requested), config.max_output_tokens)
        else:
            transformed["max_output_tokens"] = config.max_output_tokens

    if config.portable:
        transformed.pop("include", None)
        transformed.pop("prompt_cache_key", None)
        transformed.pop("reasoning", None)

    return transformed


class Config:
    """Runtime settings with backwards-compatible historical attributes."""

    def __init__(self, args: argparse.Namespace) -> None:
        requested_provider = getattr(args, "provider", None) or getattr(args, "mode", None)
        # Namespaces built by embedders/tests historically omitted provider;
        # keep that programmatic fallback generic. The CLI explicitly supplies
        # its consumer-facing default of ``auto``.
        provider = str(requested_provider or "generic").lower()
        if provider == "http":
            provider = "generic"
        self.provider: str = provider
        self.host: str = str(getattr(args, "host", DEFAULT_HOST))
        self.port: int = int(getattr(args, "port", DEFAULT_PORT))
        raw_upstream = getattr(args, "upstream", None)
        if raw_upstream:
            upstream = str(raw_upstream)
        else:
            upstream = _default_upstream_for(provider)
        self.upstream: str = upstream.rstrip("/")
        self.codex_upstream: str = (
            self.upstream if provider == "codex" else _default_upstream_for("codex")
        ).rstrip("/")
        self.claude_upstream: str = (
            self.upstream if provider == "claude" else _default_upstream_for("claude")
        ).rstrip("/")
        self.api_key: str | None = getattr(args, "api_key", None)
        self.access_token: str | None = getattr(args, "access_token", None)
        self.model: str | None = getattr(args, "model", None)
        self.portable: bool = bool(getattr(args, "portable", False))
        self.max_output_tokens: int | None = getattr(args, "max_output_tokens", None)
        self.insecure_upstream: bool = bool(getattr(args, "insecure_upstream", False))
        self.allow_oauth_host_override: bool = bool(
            getattr(args, "allow_oauth_host_override", False)
        )
        self.allow_non_loopback: bool = bool(getattr(args, "allow_non_loopback", False))
        self.verbose: bool = bool(getattr(args, "verbose", False))
        self.timeout: float = float(getattr(args, "timeout", 600.0) or 600.0)
        self.max_body_bytes: int = int(
            getattr(args, "max_body_bytes", DEFAULT_MAX_BODY_BYTES)
        )
        if self.max_body_bytes <= 0:
            raise ValueError("max incoming body size must be a positive integer")
        self.refresh_skew_seconds: int = int(getattr(args, "refresh_skew", 120) or 120)
        self.codex_home: str | None = getattr(args, "codex_home", None)
        self.codex_auth_path: str | None = getattr(args, "codex_auth_path", None)
        self.codex_token_url: str | None = getattr(args, "codex_token_url", None)
        self.claude_credentials: str | None = getattr(args, "claude_credentials", None)
        self.claude_token_url: str | None = getattr(args, "claude_token_url", None)
        self.claude_transport: str = str(
            getattr(args, "claude_transport", "cli") or "cli"
        ).lower()
        if self.claude_transport not in {"cli", "http"}:
            raise ValueError("Claude transport must be 'cli' or 'http'.")
        self.claude_cli: str = str(getattr(args, "claude_cli", "claude") or "claude")
        self.doctor_provider: str | None = getattr(args, "doctor_provider", None)
        self.codex_models: list[str] = _parse_model_list(
            getattr(args, "codex_models", None), DEFAULT_CODEX_MODELS
        )
        self.claude_models: list[str] = _parse_model_list(
            getattr(args, "claude_models", None), DEFAULT_CLAUDE_MODELS
        )
        self.codex_models_explicit: bool = bool(getattr(args, "codex_models", None))
        self.claude_models_explicit: bool = bool(getattr(args, "claude_models", None))
        self.model_discovery: bool = bool(getattr(args, "model_discovery", True))
        self._model_catalog_cache: dict[str, tuple[float, list[dict[str, Any]]]] = {}
        if not _is_loopback_host(self.host) and not self.allow_non_loopback:
            raise ProviderError(
                f"refusing non-loopback bind address {self.host!r}; "
                "use --allow-non-loopback only when remote access is intentional.",
                code="non_loopback_bind_rejected",
            )
        native_providers = (
            ("codex", "claude") if self.provider == "auto" else (self.provider,)
        )
        for native_provider in native_providers:
            native_upstream = (
                self.codex_upstream
                if native_provider == "codex"
                else self.claude_upstream
            )
            _validate_native_config(
                provider=native_provider,
                upstream=native_upstream,
                codex_token_url=self.codex_token_url,
                claude_token_url=self.claude_token_url,
                insecure_upstream=self.insecure_upstream,
                allow_oauth_host_override=self.allow_oauth_host_override,
            )

    def upstream_url(self, path: str) -> str:
        return f"{self.upstream}/{path.lstrip('/')}"

    def provider_url(self, provider: str | None = None) -> str:
        selected = provider or self.provider
        if selected == "codex":
            return endpoint_for(self.codex_upstream, "responses")
        if selected == "claude":
            return _claude_endpoint_from_upstream(self.claude_upstream)
        return self.upstream_url("responses")

    def resolve_model(self, model: Any) -> tuple[str, str]:
        value = str(model or "").strip()
        if not value:
            raise ProviderError(
                "request is missing a model; open Muse's /model menu and choose one.",
                code="model_missing",
            )

        for prefix in ("codex/", "codex:"):
            if value.startswith(prefix):
                if self.provider not in {"auto", "codex"}:
                    raise ProviderError(
                        "selected model belongs to Codex.",
                        code="model_provider_mismatch",
                    )
                return "codex", value[len(prefix) :]
        for prefix in ("claude/", "claude:"):
            if value.startswith(prefix):
                if self.provider not in {"auto", "claude"}:
                    raise ProviderError(
                        "selected model belongs to Claude.",
                        code="model_provider_mismatch",
                    )
                return "claude", value[len(prefix) :]

        if self.provider in {"generic", "codex", "claude"}:
            return self.provider, value
        if value in self.claude_models or value.lower().startswith("claude-"):
            return "claude", value
        if value in self.codex_models or value.lower().startswith(
            ("gpt-", "o1", "o3", "o4")
        ):
            return "codex", value
        raise ProviderError(
            f"cannot infer a provider for model {value!r}; select a prefixed model from Muse's /model menu.",
            code="model_provider_unknown",
        )

    def secret_values(self) -> list[str]:
        values: list[str] = []
        for value in (self.api_key, self.access_token):
            if isinstance(value, str) and value:
                values.append(value)
        return values


def _catalog_entry(
    provider: str,
    model: str,
    *,
    context_limit: int,
    output_limit: int,
    is_default: bool,
) -> dict[str, Any]:
    routed_id = f"{provider}/{model}"
    provider_label = "Codex" if provider == "codex" else "Claude"
    return {
        "model_id": routed_id,
        "display_label": f"{provider_label} · {model}",
        "release_date": "2026-08-06",
        "context_limit": context_limit,
        "output_limit": output_limit,
        "is_current": is_default,
        "is_default": is_default,
        "description": f"{provider_label} via local muse-shim",
        "cost": {
            "input": "0",
            "output": "0",
            "cached": "0",
            "currency": "USD",
        },
    }


def _wire_catalog_model(row: dict[str, Any]) -> dict[str, Any]:
    model_id = str(row["model_id"])
    family = model_id.split("/", 1)[0] if "/" in model_id else "generic"
    reasoning = family == "codex"
    effort = "high"
    variants = {
        level: {"reasoningEffort": level}
        for level in ("minimal", "low", "medium", "high", "xhigh")
    }
    return {
        "id": model_id,
        "object": "model",
        "created": 1785974400,
        "owned_by": family,
        "metadata": {
            "muse-code": {
                "name": row["display_label"],
                "family": family,
                "release_date": row["release_date"],
                "is_hidden": False,
                "attachment": True,
                "reasoning": reasoning,
                "temperature": False,
                "tool_call": True,
                "modalities": {
                    "input": ["text", "image"],
                    "output": ["text"],
                },
                "limit": {
                    "context": row["context_limit"],
                    "output": row["output_limit"],
                },
                "options": {
                    "reasoningEffort": effort,
                    "forceReasoning": reasoning,
                    "include": [],
                    "temperature": 0.9,
                    "top_p": 0.9,
                },
                "variants": variants,
                "description": row["description"],
                "cost": row["cost"],
            }
        },
    }


def muse_catalog_response(rows: list[dict[str, Any]]) -> dict[str, Any]:
    """Return the official network catalog schema consumed by Muse Code."""
    return {
        "object": "list",
        "data": [_wire_catalog_model(row) for row in rows],
    }


def _discover_provider_models(config: Config, provider: str) -> list[dict[str, Any]]:
    """Fetch the account-visible catalog; return an empty list on any failure."""
    if not config.model_discovery:
        return []
    # Claude Code owns model selection/auth in CLI mode. Do not inspect direct
    # Anthropic credentials or make a separate HTTP catalog request here.
    if provider == "claude" and config.claude_transport == "cli":
        return []
    try:
        if provider == "codex":
            state = discover_codex_credentials(config)
            if state is None:
                return []
            request = urllib.request.Request(
                CODEX_MODELS_ENDPOINT,
                method="GET",
                headers=codex_headers(state.access_token),
            )
            with urllib.request.urlopen(
                request, timeout=min(config.timeout, 8.0)
            ) as response:
                payload = json.loads(response.read().decode("utf-8"))
            raw_models = payload.get("models", []) if isinstance(payload, dict) else []
            result = []
            for item in raw_models:
                if not isinstance(item, dict) or not isinstance(item.get("slug"), str):
                    continue
                result.append(
                    {
                        "model": item["slug"].strip(),
                        "context_limit": int(item.get("context_window") or 272000),
                        "output_limit": int(item.get("max_output_tokens") or 32768),
                    }
                )
            return [item for item in result if item["model"]]

        state = discover_claude_credentials(config)
        if state is None or state.auth_mode != "api_key":
            return []
        endpoint = config.provider_url("claude").removesuffix("/messages") + "/models"
        headers = claude_headers(state)
        headers.pop("Content-Type", None)
        request = urllib.request.Request(endpoint, method="GET", headers=headers)
        with urllib.request.urlopen(
            request, timeout=min(config.timeout, 8.0)
        ) as response:
            payload = json.loads(response.read().decode("utf-8"))
        raw_models = payload.get("data", []) if isinstance(payload, dict) else []
        return [
            {
                "model": item["id"].strip(),
                "context_limit": int(item.get("max_input_tokens") or 200000),
                "output_limit": int(item.get("max_tokens") or 32768),
            }
            for item in raw_models
            if isinstance(item, dict)
            and isinstance(item.get("id"), str)
            and item["id"].strip()
        ]
    except (
        AuthError,
        ProviderError,
        OSError,
        ValueError,
        TypeError,
        json.JSONDecodeError,
    ):
        return []


def provider_model_catalog(config: Config, provider: str) -> list[dict[str, Any]]:
    cached = config._model_catalog_cache.get(provider)
    if cached and time.time() - cached[0] < MODEL_CATALOG_TTL_SECONDS:
        return cached[1]

    discovered = _discover_provider_models(config, provider)
    configured = config.codex_models if provider == "codex" else config.claude_models
    configured_explicit = (
        config.codex_models_explicit
        if provider == "codex"
        else config.claude_models_explicit
    )
    fallbacks = {}
    for model in configured:
        if provider == "claude":
            context_limit, output_limit = CLAUDE_MODEL_LIMITS.get(
                model, (200_000, 32_768)
            )
        else:
            context_limit, output_limit = 272_000, 32_768
        fallbacks[model] = {
            "model": model,
            "context_limit": context_limit,
            "output_limit": output_limit,
        }
    defaults = fallbacks if not discovered or configured_explicit else {}
    for item in discovered:
        defaults[item["model"]] = item

    entries = [
        _catalog_entry(
            provider,
            item["model"],
            context_limit=item["context_limit"],
            output_limit=min(item["output_limit"], config.max_output_tokens)
            if config.max_output_tokens
            else item["output_limit"],
            is_default=index == 0,
        )
        for index, item in enumerate(defaults.values())
    ]
    config._model_catalog_cache[provider] = (time.time(), entries)
    return entries


def _claude_cli_output_schema(tool_names: list[str]) -> dict[str, Any]:
    name_schema: dict[str, Any] = {"type": "string"}
    if tool_names:
        name_schema["enum"] = tool_names
    return {
        "type": "object",
        "properties": {
            "text": {"type": "string"},
            "tool_calls": {
                "type": "array",
                "maxItems": 8 if tool_names else 0,
                "items": {
                    "type": "object",
                    "properties": {
                        "name": name_schema,
                        "arguments_json": {"type": "string"},
                    },
                    "required": ["name", "arguments_json"],
                    "additionalProperties": False,
                },
            },
        },
        "required": ["text", "tool_calls"],
        "additionalProperties": False,
    }


def _claude_cli_prompt(payload: dict[str, Any]) -> str:
    request_document = {
        "system": payload.get("system", []),
        "messages": payload.get("messages", []),
        "tools": payload.get("tools", []),
        "tool_choice": payload.get("tool_choice", {"type": "auto"}),
    }
    return (
        "Continue the conversation represented by the JSON document below. "
        "Honor its system instructions and message order. If a listed tool is "
        "needed, return it in tool_calls without pretending it was executed; "
        "arguments_json must be a JSON object encoded as a string and must match "
        "that tool's input_schema. If no tool is needed, return the complete "
        "assistant answer in text and an empty tool_calls array. Do not mention "
        "this transport envelope.\n\nREQUEST_JSON\n"
        + json.dumps(request_document, ensure_ascii=False, separators=(",", ":"))
    )


def run_claude_cli(
    body: dict[str, Any], config: Config
) -> tuple[dict[str, Any], Any]:
    """Execute one Muse turn through Claude Code print mode."""
    model = str(body.get("model") or "").strip()
    if not model:
        raise ProviderError("Claude CLI request is missing a model.", code="model_missing")
    payload, tool_names = responses_to_anthropic(
        body,
        oauth=False,
        max_output_tokens=config.max_output_tokens,
    )
    available_tools = payload.get("tools") if isinstance(payload.get("tools"), list) else []
    available_names = [
        str(tool.get("name"))
        for tool in available_tools
        if isinstance(tool, dict) and isinstance(tool.get("name"), str)
    ]
    schema = _claude_cli_output_schema(available_names)
    command = [
        config.claude_cli,
        "-p",
        "--safe-mode",
        "--model",
        model,
        "--tools",
        "",
        "--disable-slash-commands",
        "--no-session-persistence",
        "--output-format",
        "json",
        "--json-schema",
        json.dumps(schema, ensure_ascii=False, separators=(",", ":")),
        "--system-prompt",
        "You are a model backend for Muse Code. Return only the requested structured result.",
    ]
    reasoning = body.get("reasoning")
    effort = reasoning.get("effort") if isinstance(reasoning, dict) else None
    if isinstance(effort, str) and "haiku" not in model.lower():
        normalized_effort = {"minimal": "low", "ultra": "max"}.get(
            effort.lower(), effort.lower()
        )
        if normalized_effort in {"low", "medium", "high", "xhigh", "max"}:
            command.extend(["--effort", normalized_effort])

    environment = os.environ.copy()
    # Claude Code refuses accidental recursive launches. The shim is an
    # intentional transport boundary, so do not inherit parent-agent markers.
    environment.pop("CLAUDECODE", None)
    environment.pop("CLAUDE_CODE_ENTRYPOINT", None)
    try:
        completed = subprocess.run(
            command,
            input=_claude_cli_prompt(payload),
            text=True,
            capture_output=True,
            timeout=config.timeout,
            env=environment,
            check=False,
        )
    except FileNotFoundError as error:
        raise ProviderError(
            f"Claude CLI executable {config.claude_cli!r} was not found. Install Claude Code or set MUSE_SHIM_CLAUDE_CLI.",
            status=502,
            code="claude_cli_missing",
        ) from error
    except subprocess.TimeoutExpired as error:
        raise ProviderError(
            f"Claude CLI exceeded the {config.timeout:g}s timeout.",
            status=504,
            code="claude_cli_timeout",
        ) from error

    result: dict[str, Any] | None = None
    try:
        decoded = json.loads(str(completed.stdout or ""))
        if isinstance(decoded, dict):
            result = decoded
    except json.JSONDecodeError:
        pass
    if result is not None and result.get("is_error") is True:
        message = str(result.get("result") or "Claude CLI reported an error.")[:500]
        raw_status = result.get("api_error_status")
        status = int(raw_status) if isinstance(raw_status, int) else 502
        raise ProviderError(
            message,
            status=status,
            code="claude_cli_rate_limited" if status == 429 else "claude_cli_failed",
        )
    if completed.returncode != 0:
        detail = " ".join(str(completed.stderr or "").strip().splitlines())[-500:]
        suffix = f": {detail}" if detail else ""
        raise ProviderError(
            f"Claude CLI exited with status {completed.returncode}{suffix}",
            status=502,
            code="claude_cli_failed",
        )
    if result is None:
        raise ProviderError(
            "Claude CLI returned invalid JSON.",
            status=502,
            code="claude_cli_invalid_response",
        )
    structured = result.get("structured_output")
    if not isinstance(structured, dict):
        try:
            structured = json.loads(str(result.get("result") or ""))
        except json.JSONDecodeError as error:
            raise ProviderError(
                "Claude CLI returned no structured output.",
                status=502,
                code="claude_cli_invalid_response",
            ) from error
    text = structured.get("text")
    calls = structured.get("tool_calls")
    if not isinstance(text, str) or not isinstance(calls, list):
        raise ProviderError(
            "Claude CLI structured output has an invalid shape.",
            status=502,
            code="claude_cli_invalid_response",
        )

    content: list[dict[str, Any]] = []
    if text:
        content.append({"type": "text", "text": text})
    for call in calls:
        if not isinstance(call, dict) or call.get("name") not in available_names:
            raise ProviderError(
                "Claude CLI returned an unknown tool call.",
                status=502,
                code="claude_cli_invalid_tool",
            )
        try:
            arguments = json.loads(str(call.get("arguments_json") or "{}"))
        except json.JSONDecodeError as error:
            raise ProviderError(
                "Claude CLI returned invalid tool arguments.",
                status=502,
                code="claude_cli_invalid_tool",
            ) from error
        if not isinstance(arguments, dict):
            raise ProviderError(
                "Claude CLI tool arguments must be a JSON object.",
                status=502,
                code="claude_cli_invalid_tool",
            )
        content.append(
            {
                "type": "tool_use",
                "id": "toolu_" + secrets.token_hex(12),
                "name": str(call["name"]),
                "input": arguments,
            }
        )
    if not content:
        content.append({"type": "text", "text": ""})
    usage = result.get("usage") if isinstance(result.get("usage"), dict) else {}
    return (
        {
            "id": "msg_" + secrets.token_hex(12),
            "type": "message",
            "role": "assistant",
            "model": model,
            "content": content,
            "stop_reason": "tool_use" if calls else "end_turn",
            "usage": {
                "input_tokens": int(usage.get("input_tokens") or 0),
                "output_tokens": int(usage.get("output_tokens") or 0),
            },
        },
        tool_names,
    )


def _anthropic_message_sse(payload: dict[str, Any]) -> io.BytesIO:
    events: list[tuple[str, dict[str, Any]]] = [
        ("message_start", {"type": "message_start", "message": payload})
    ]
    for index, block in enumerate(payload.get("content", [])):
        if block.get("type") == "text":
            events.append(
                (
                    "content_block_start",
                    {
                        "type": "content_block_start",
                        "index": index,
                        "content_block": {"type": "text", "text": ""},
                    },
                )
            )
            events.append(
                (
                    "content_block_delta",
                    {
                        "type": "content_block_delta",
                        "index": index,
                        "delta": {"type": "text_delta", "text": block.get("text", "")},
                    },
                )
            )
        else:
            events.append(
                (
                    "content_block_start",
                    {
                        "type": "content_block_start",
                        "index": index,
                        "content_block": {
                            "type": "tool_use",
                            "id": block["id"],
                            "name": block["name"],
                            "input": {},
                        },
                    },
                )
            )
            events.append(
                (
                    "content_block_delta",
                    {
                        "type": "content_block_delta",
                        "index": index,
                        "delta": {
                            "type": "input_json_delta",
                            "partial_json": json.dumps(
                                block.get("input", {}),
                                ensure_ascii=False,
                                separators=(",", ":"),
                            ),
                        },
                    },
                )
            )
        events.append(
            (
                "content_block_stop",
                {"type": "content_block_stop", "index": index},
            )
        )
    events.extend(
        [
            (
                "message_delta",
                {
                    "type": "message_delta",
                    "delta": {"stop_reason": payload.get("stop_reason")},
                    "usage": payload.get("usage", {}),
                },
            ),
            ("message_stop", {"type": "message_stop"}),
        ]
    )
    raw = b"".join(
        (
            f"event: {event_name}\n"
            f"data: {json.dumps(event, ensure_ascii=False, separators=(',', ':'))}\n\n"
        ).encode("utf-8")
        for event_name, event in events
    )
    return io.BytesIO(raw)


def _redact_bytes(payload: bytes, secrets: Iterable[str]) -> bytes:
    result = payload
    for secret in secrets:
        if secret:
            result = result.replace(secret.encode("utf-8"), b"[REDACTED]")
    return result


def _ssl_context(config: Config) -> ssl.SSLContext | None:
    if config.insecure_upstream:
        return ssl._create_unverified_context()  # noqa: SLF001 - explicit opt-in
    return None


def _iter_upstream_chunks(upstream: Any, *, content_type: str) -> Iterable[bytes]:
    """Yield promptly from an upstream stream without a large read buffer."""
    if "text/event-stream" in content_type.lower():
        reader = getattr(upstream, "readline", None)
        if callable(reader):
            while True:
                chunk = reader()
                if not chunk:
                    return
                yield chunk.encode("utf-8") if isinstance(chunk, str) else chunk

    reader = getattr(upstream, "read1", None)
    if not callable(reader):
        reader = getattr(upstream, "read", None)
    if not callable(reader):
        return
    while True:
        chunk = reader(8192)
        if not chunk:
            return
        yield chunk.encode("utf-8") if isinstance(chunk, str) else chunk


def _write_upstream_body(handler: BaseHTTPRequestHandler, upstream: Any, *, stream: bool, content_type: str) -> None:
    if stream:
        for chunk in _iter_upstream_chunks(upstream, content_type=content_type):
            handler.wfile.write(chunk)
            handler.wfile.flush()
        return
    payload = upstream.read()
    if payload:
        handler.wfile.write(payload.encode("utf-8") if isinstance(payload, str) else payload)
        handler.wfile.flush()


class MuseShimHandler(BaseHTTPRequestHandler):
    server_version = "muse-shim/0.4"
    protocol_version = "HTTP/1.1"

    @property
    def config(self) -> Config:
        return self.server.config  # type: ignore[attr-defined]

    def log_message(self, fmt: str, *args: Any) -> None:
        # Request paths/status are useful; request bodies and headers are not.
        if self.config.verbose:
            super().log_message(fmt, *args)

    def send_json(self, status: int, value: Any) -> None:
        payload = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(payload)
        self.close_connection = True

    def send_provider_error(self, error: ProviderError | ShimError, *, status: int | None = None) -> None:
        error_status = status or getattr(error, "status", 400)
        code = getattr(error, "code", "shim_error")
        if self.config.verbose:
            print(
                f"muse-shim provider error: status={error_status} code={code} message={error}",
                file=sys.stderr,
                flush=True,
            )
        self.send_json(
            int(error_status),
            {"error": {"message": str(error), "type": code}},
        )

    def do_GET(self) -> None:
        path = self.path.split("?", 1)[0].rstrip("/")
        if path == "/health":
            self.send_json(
                200,
                {
                    "ok": True,
                    "service": "muse-shim",
                    "provider": self.config.provider,
                    "claude_transport": self.config.claude_transport,
                },
            )
            return

        if path in {"/muse-code/models", "/v1/muse-code/models"}:
            if self.config.provider == "auto":
                models = [
                    dict(item) for item in provider_model_catalog(self.config, "codex")
                ]
                models.extend(
                    dict(item) for item in provider_model_catalog(self.config, "claude")
                )
                self.send_json(200, muse_catalog_response(models))
                return
            if self.config.provider in {"codex", "claude"} and not self.config.model:
                self.send_json(
                    200,
                    muse_catalog_response(
                        provider_model_catalog(self.config, self.config.provider)
                    ),
                )
                return
            model = self.config.model or {
                "codex": "codex-model-from-request",
                "claude": "claude-model-from-request",
            }.get(self.config.provider, "upstream-model")
            provider = (
                self.config.provider
                if self.config.provider in {"codex", "claude"}
                else "generic"
            )
            provider_label = provider.title()
            self.send_json(
                200,
                muse_catalog_response(
                    [
                        {
                            "model_id": model,
                            "display_label": model,
                            "release_date": "2026-08-06",
                            "context_limit": 200000,
                            "output_limit": self.config.max_output_tokens or 32768,
                            "is_current": True,
                            "is_default": True,
                            "description": f"{provider_label} via local muse-shim",
                            "cost": {
                                "input": "0",
                                "output": "0",
                                "cached": "0",
                                "currency": "USD",
                            },
                        }
                    ]
                ),
            )
            return

        self.send_json(404, {"error": {"message": "not found", "type": "shim_error"}})

    def _read_request_body(self) -> dict[str, Any]:
        try:
            content_length = int(self.headers.get("Content-Length", "0"))
        except ValueError as exc:
            raise ShimError("Content-Length must be an integer") from exc
        if content_length <= 0:
            raise ShimError("request body is empty")
        if content_length > self.config.max_body_bytes:
            raise ShimError(
                f"request body exceeds the configured limit of {self.config.max_body_bytes} bytes"
            )
        raw = self.rfile.read(content_length)
        try:
            body = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise ShimError("request body must be valid JSON") from exc
        if not isinstance(body, dict):
            raise ShimError("request body must be a JSON object")
        return body

    def do_POST(self) -> None:
        path = self.path.split("?", 1)[0].rstrip("/")
        if path not in {"/responses", "/v1/responses"}:
            self.send_json(404, {"error": {"message": "not found", "type": "shim_error"}})
            return

        try:
            body = self._read_request_body()
            transformed = transform_request(body, self.config)
            provider, upstream_model = self.config.resolve_model(
                transformed.get("model")
            )
            transformed["model"] = upstream_model
        except (ProviderError, ShimError, ValueError) as error:
            local_error = (
                error
                if isinstance(error, (ProviderError, ShimError))
                else ShimError(str(error))
            )
            self.send_provider_error(local_error)
            return

        try:
            if provider == "codex":
                self.forward_codex(transformed)
            elif provider == "claude":
                if self.config.claude_transport == "cli":
                    self.forward_claude_cli(transformed)
                else:
                    self.forward_claude(transformed)
            else:
                self.forward_generic(transformed)
        except (ProviderError, ShimError) as error:
            self.send_provider_error(error)
        except urllib.error.HTTPError as error:
            # Generic mode historically passed the upstream JSON through. It
            # still does, but known local credentials are redacted first.
            incoming_auth = self.headers.get("Authorization")
            known_secrets = self.config.secret_values()
            if incoming_auth:
                known_secrets.append(incoming_auth)
            payload = _redact_bytes(error.read(), known_secrets)
            self.send_response(error.code)
            self.send_header("Content-Type", error.headers.get("Content-Type", "application/json"))
            self.send_header("Content-Length", str(len(payload)))
            self.send_header("Connection", "close")
            self.end_headers()
            self.wfile.write(payload)
            self.close_connection = True
        except (urllib.error.URLError, TimeoutError, OSError) as error:
            self.send_provider_error(
                ProviderError(
                    f"upstream request failed: {error}",
                    status=502,
                    code="upstream_unreachable",
                )
            )

    def _send_upstream_headers(self, upstream: Any, *, content_type: str, stream: bool) -> None:
        self.send_response(int(getattr(upstream, "status", 200)))
        self.send_header("Content-Type", content_type)
        self.send_header("Cache-Control", "no-cache" if stream else "no-store")
        if not stream:
            content_length = upstream.headers.get("Content-Length")
            if content_length:
                self.send_header("Content-Length", content_length)
        else:
            self.send_header("Connection", "close")
            self.close_connection = True
        self.end_headers()

    def forward_generic(self, body: dict[str, Any]) -> None:
        incoming_auth = self.headers.get("Authorization")
        if self.config.api_key:
            authorization = f"Bearer {self.config.api_key}"
        elif incoming_auth:
            authorization = incoming_auth
        else:
            raise ShimError(
                "no upstream API key; set MUSE_SHIM_API_KEY or launch Muse with META_API_KEY"
            )

        request = urllib.request.Request(
            self.config.upstream_url("responses"),
            data=json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8"),
            method="POST",
            headers={
                "Authorization": authorization,
                "Content-Type": "application/json",
                "Accept": "text/event-stream, application/json",
                "User-Agent": self.server_version,
            },
        )
        with urllib.request.urlopen(
            request,
            timeout=self.config.timeout,
            context=_ssl_context(self.config),
        ) as upstream:
            content_type = upstream.headers.get("Content-Type", "application/octet-stream")
            stream = "text/event-stream" in content_type.lower() or body.get("stream") is True
            self._send_upstream_headers(upstream, content_type=content_type, stream=stream)
            _write_upstream_body(
                self,
                upstream,
                stream=stream,
                content_type=content_type,
            )

    def forward_codex(self, body: dict[str, Any]) -> None:
        state = resolve_codex_credentials(self.config)
        if not body.get("model"):
            raise ProviderError(
                "Codex request is missing model; set MUSE_SHIM_MODEL or send model.",
                code="model_missing",
            )
        endpoint = self.config.provider_url("codex")
        validate_codex_endpoint(
            endpoint,
            allow_host_override=self.config.allow_oauth_host_override,
        )
        codex_body = prepare_codex_request(body)
        encoded_body = json.dumps(
            codex_body,
            ensure_ascii=False,
            separators=(",", ":"),
        ).encode("utf-8")
        for attempt in range(2):
            request = urllib.request.Request(
                endpoint,
                data=encoded_body,
                method="POST",
                headers=codex_headers(state.access_token),
            )
            try:
                with urllib.request.urlopen(
                    request,
                    timeout=self.config.timeout,
                    context=_ssl_context(self.config),
                ) as upstream:
                    content_type = upstream.headers.get("Content-Type", "application/json")
                    stream = "text/event-stream" in content_type.lower() or codex_body.get("stream") is True
                    self._send_upstream_headers(upstream, content_type=content_type, stream=stream)
                    _write_upstream_body(
                        self,
                        upstream,
                        stream=stream,
                        content_type=content_type,
                    )
                    return
            except urllib.error.HTTPError as error:
                if error.code in {401, 403} and attempt == 0:
                    try:
                        state = resolve_codex_credentials(self.config, force_refresh=True)
                    except ProviderError as refresh_error:
                        raise AuthError(
                            "Codex upstream rejected the OAuth credentials and refresh failed. "
                            "Run `codex login` and try again.",
                            code="codex_upstream_unauthorized",
                        ) from refresh_error
                    continue
                if error.code in {401, 403}:
                    raise AuthError(
                        "Codex upstream rejected the OAuth credentials. Run `codex login` and try again.",
                        code="codex_upstream_unauthorized",
                    ) from error
                raise ProviderError(
                    f"Codex upstream returned HTTP {error.code}.",
                    status=error.code,
                    code="codex_upstream_error",
                ) from error

    def forward_claude_cli(self, body: dict[str, Any]) -> None:
        payload, tool_names = run_claude_cli(body, self.config)
        model = str(payload.get("model") or body.get("model") or "")
        if body.get("stream") is True:
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Connection", "close")
            self.end_headers()
            self.close_connection = True
            for chunk in convert_anthropic_stream(
                _anthropic_message_sse(payload),
                model=model,
                tool_names=tool_names,
            ):
                self.wfile.write(chunk)
                self.wfile.flush()
            return
        self.send_json(
            200,
            anthropic_json_to_responses(
                payload,
                model=model,
                tool_names=tool_names,
            ),
        )

    def forward_claude(self, body: dict[str, Any]) -> None:
        if "previous_response_id" in body:
            raise ProviderError(
                "Claude mode does not support previous_response_id because the shim has no durable response-state store.",
                code="claude_previous_response_unsupported",
            )
        state = resolve_claude_credentials(self.config)
        payload, tool_names = responses_to_anthropic(
            body,
            oauth=state.auth_mode != "api_key",
            model_override=(
                self.config.model if self.config.provider == "claude" else None
            ),
            max_output_tokens=self.config.max_output_tokens,
        )
        endpoint = self.config.provider_url("claude")
        validate_claude_endpoint(
            endpoint,
            allow_host_override=self.config.allow_oauth_host_override,
        )
        encoded_body = json.dumps(
            payload,
            ensure_ascii=False,
            separators=(",", ":"),
        ).encode("utf-8")
        for attempt in range(2):
            request = urllib.request.Request(
                endpoint,
                data=encoded_body,
                method="POST",
                headers=claude_headers(state, stream=bool(body.get("stream"))),
            )
            try:
                with urllib.request.urlopen(
                    request,
                    timeout=self.config.timeout,
                    context=_ssl_context(self.config),
                ) as upstream:
                    content_type = upstream.headers.get("Content-Type", "application/json")
                    stream = "text/event-stream" in content_type.lower() or body.get("stream") is True
                    if stream:
                        self._send_upstream_headers(
                            upstream,
                            content_type="text/event-stream",
                            stream=True,
                        )
                        for chunk in convert_anthropic_stream(
                            upstream,
                            model=str(payload.get("model") or ""),
                            tool_names=tool_names,
                        ):
                            self.wfile.write(chunk)
                            self.wfile.flush()
                        return
                    try:
                        response_payload = json.loads(upstream.read().decode("utf-8"))
                    except (UnicodeError, json.JSONDecodeError) as exc:
                        raise ProviderError(
                            "Claude returned invalid JSON.",
                            status=502,
                            code="claude_invalid_response",
                        ) from exc
                    if not isinstance(response_payload, dict):
                        raise ProviderError(
                            "Claude returned an invalid response.",
                            status=502,
                            code="claude_invalid_response",
                        )
                    converted = anthropic_json_to_responses(
                        response_payload,
                        model=str(payload.get("model") or ""),
                        tool_names=tool_names,
                    )
                    encoded = json.dumps(converted, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
                    self.send_response(200)
                    self.send_header("Content-Type", "application/json")
                    self.send_header("Content-Length", str(len(encoded)))
                    self.send_header("Cache-Control", "no-store")
                    self.end_headers()
                    self.wfile.write(encoded)
                    return
            except urllib.error.HTTPError as error:
                if error.code in {401, 403} and attempt == 0 and state.auth_mode != "api_key":
                    try:
                        state = resolve_claude_credentials(self.config, force_refresh=True)
                    except ProviderError as refresh_error:
                        raise AuthError(
                            "Claude upstream rejected the OAuth credentials and refresh failed. "
                            "Run `claude setup-token` or `claude login` and try again.",
                            code="claude_upstream_unauthorized",
                        ) from refresh_error
                    continue
                if error.code in {401, 403}:
                    raise AuthError(
                        "Claude upstream rejected the credentials. Run `claude setup-token` or `claude login` and try again.",
                        code="claude_upstream_unauthorized",
                    ) from error
                raise ProviderError(
                    f"Claude upstream returned HTTP {error.code}.",
                    status=error.code,
                    code="claude_upstream_error",
                ) from error


class MuseShimServer(ThreadingHTTPServer):
    daemon_threads = True

    def __init__(self, address: tuple[str, int], config: Config) -> None:
        self.config = config
        super().__init__(address, MuseShimHandler)


def _default_upstream_for(provider: str) -> str:
    if provider == "codex":
        return os.environ.get(
            "MUSE_SHIM_CODEX_UPSTREAM",
            os.environ.get("MUSE_SHIM_CODEX_BASE_URL", DEFAULT_CODEX_ENDPOINT),
        )
    if provider == "claude":
        return os.environ.get(
            "MUSE_SHIM_CLAUDE_UPSTREAM",
            os.environ.get("MUSE_SHIM_CLAUDE_BASE_URL", DEFAULT_CLAUDE_ENDPOINT),
        )
    return os.environ.get("MUSE_SHIM_UPSTREAM", DEFAULT_UPSTREAM)


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Loopback Muse proxy with a unified Codex + Claude model catalog",
        epilog=(
            "Use `muse-shim run [MUSE_ARGS...]` to start the proxy and Muse "
            "together. Shim options go before `run`."
        ),
    )
    parser.add_argument(
        "mode",
        nargs="?",
        choices=("auto", "generic", "http", "codex", "claude", "doctor"),
        default=os.environ.get("MUSE_SHIM_PROVIDER", "auto"),
        help="provider mode or doctor",
    )
    parser.add_argument(
        "--provider", choices=("auto", "generic", "http", "codex", "claude")
    )
    parser.add_argument("--host", default=os.environ.get("MUSE_SHIM_HOST", DEFAULT_HOST))
    parser.add_argument("--port", type=int, default=int(os.environ.get("MUSE_SHIM_PORT", str(DEFAULT_PORT))))
    parser.add_argument("--upstream", default=None)
    parser.add_argument("--api-key", default=os.environ.get("MUSE_SHIM_API_KEY"))
    parser.add_argument("--access-token", default=None)
    parser.add_argument("--model", default=os.environ.get("MUSE_SHIM_MODEL"))
    parser.add_argument(
        "--codex-models",
        default=os.environ.get("MUSE_SHIM_CODEX_MODELS"),
        help="comma-separated Codex fallback models shown in Muse's /model menu",
    )
    parser.add_argument(
        "--claude-models",
        default=os.environ.get("MUSE_SHIM_CLAUDE_MODELS"),
        help="comma-separated Claude fallback models shown in Muse's /model menu",
    )
    parser.add_argument(
        "--no-model-discovery",
        action="store_false",
        dest="model_discovery",
        default=not env_flag("MUSE_SHIM_NO_MODEL_DISCOVERY"),
        help="do not query provider model catalogs; advertise configured fallbacks only",
    )
    parser.add_argument(
        "--portable",
        action="store_true",
        default=env_flag("MUSE_SHIM_PORTABLE"),
        help="strip reasoning and cache fields for stricter Responses servers",
    )
    parser.add_argument(
        "--max-output-tokens",
        type=int,
        default=(
            int(os.environ["MUSE_SHIM_MAX_OUTPUT_TOKENS"])
            if os.environ.get("MUSE_SHIM_MAX_OUTPUT_TOKENS")
            else None
        ),
    )
    parser.add_argument(
        "--insecure-upstream",
        action="store_true",
        default=env_flag("MUSE_SHIM_INSECURE_UPSTREAM"),
        help="disable upstream TLS verification (development only)",
    )
    parser.add_argument(
        "--allow-oauth-host-override",
        action="store_true",
        default=False,
        help="DANGEROUS development-only opt-in for non-native OAuth hosts; never enable from an environment variable",
    )
    parser.add_argument(
        "--allow-non-loopback",
        action="store_true",
        default=False,
        help="allow binding beyond loopback (remote clients can reach this proxy)",
    )
    parser.add_argument(
        "--max-body-bytes",
        type=int,
        default=int(os.environ.get("MUSE_SHIM_MAX_BODY_BYTES", str(DEFAULT_MAX_BODY_BYTES))),
        help=f"maximum incoming JSON body size (default: {DEFAULT_MAX_BODY_BYTES})",
    )
    parser.add_argument("--timeout", type=float, default=float(os.environ.get("MUSE_SHIM_TIMEOUT", "600")))
    parser.add_argument("--refresh-skew", type=int, default=int(os.environ.get("MUSE_SHIM_REFRESH_SKEW", "120")))
    parser.add_argument("--codex-home", default=os.environ.get("CODEX_HOME"))
    parser.add_argument("--codex-auth-path", default=os.environ.get("MUSE_SHIM_CODEX_AUTH"))
    parser.add_argument("--codex-token-url", default=os.environ.get("MUSE_SHIM_CODEX_TOKEN_URL"))
    parser.add_argument("--claude-credentials", default=os.environ.get("MUSE_SHIM_CLAUDE_CREDENTIALS"))
    parser.add_argument("--claude-token-url", default=os.environ.get("MUSE_SHIM_CLAUDE_TOKEN_URL"))
    parser.add_argument(
        "--claude-transport",
        choices=("cli", "http"),
        default=os.environ.get("MUSE_SHIM_CLAUDE_TRANSPORT", "cli"),
        help="Claude inference transport (default: cli, using claude -p)",
    )
    parser.add_argument(
        "--claude-cli",
        default=os.environ.get("MUSE_SHIM_CLAUDE_CLI", "claude"),
        help="Claude Code executable used by CLI transport",
    )
    parser.add_argument("--doctor-provider", choices=("generic", "codex", "claude"), default=None)
    parser.add_argument("--verbose", action="store_true", default=env_flag("MUSE_SHIM_VERBOSE"))
    args = parser.parse_args(argv)
    provider = args.provider or args.mode
    if provider == "http":
        provider = "generic"
    args.provider = provider
    if not args.upstream:
        args.upstream = _default_upstream_for(provider)
    try:
        if provider == "auto" and args.model:
            raise ProviderError(
                "--model / MUSE_SHIM_MODEL cannot pin auto mode; choose a model "
                "in Muse with /model, or start a provider-specific mode.",
                code="auto_model_override_rejected",
            )
        if args.max_body_bytes <= 0:
            raise ProviderError(
                "--max-body-bytes must be a positive integer.",
                code="invalid_body_limit",
            )
        if not _is_loopback_host(args.host) and not args.allow_non_loopback:
            raise ProviderError(
                f"refusing non-loopback bind address {args.host!r}; use --allow-non-loopback when intentional.",
                code="non_loopback_bind_rejected",
            )
        native_providers = ("codex", "claude") if provider == "auto" else (provider,)
        for native_provider in native_providers:
            native_upstream = (
                args.upstream
                if provider == native_provider
                else _default_upstream_for(native_provider)
            )
            _validate_native_config(
                provider=native_provider,
                upstream=native_upstream,
                codex_token_url=args.codex_token_url,
                claude_token_url=args.claude_token_url,
                insecure_upstream=args.insecure_upstream,
                allow_oauth_host_override=args.allow_oauth_host_override,
            )
    except (ProviderError, ValueError) as error:
        parser.error(str(error))
    return args


def _credential_status(config: Config, provider: str) -> tuple[bool, str]:
    if provider == "generic":
        if config.api_key:
            return True, "API key configured"
        return False, "API key not configured"
    if provider == "claude" and config.claude_transport == "cli":
        executable = shutil.which(config.claude_cli)
        if executable:
            return True, f"Claude CLI print transport; executable={executable}"
        return False, f"Claude CLI executable {config.claude_cli!r} not found"
    try:
        state = discover_codex_credentials(config) if provider == "codex" else discover_claude_credentials(config)
    except Exception:
        state = None
    if state is None:
        return False, "credentials not found"
    expiry = "expiry unknown"
    if state.expires_at is not None:
        expiry = "expired" if state.expires_at <= time.time() else "not expired"
    refresh = "refresh available" if state.refresh_token else "no refresh token"
    mode = "API key" if state.auth_mode == "api_key" else "OAuth"
    return True, f"{mode}; {refresh}; {expiry}; source={state.source}"


def run_doctor(config: Config, *, output: Any = None) -> int:
    output = output or sys.stdout
    print("muse-shim doctor", file=output)
    print(f"listen: {config.host}:{config.port}", file=output)
    print(f"generic endpoint configured: {'yes' if bool(config.upstream) else 'no'}", file=output)
    selected = getattr(config, "doctor_provider", None)
    if selected is None and config.provider in {"codex", "claude"}:
        selected = config.provider
    providers = (selected,) if selected else ("codex", "claude")
    ready = bool(config.upstream)
    for provider in providers:
        ok, detail = _credential_status(config, provider)
        ready = ready and ok
        print(f"{provider}: {'ready' if ok else 'not ready'} ({detail})", file=output)
        if provider == "claude":
            if config.claude_transport == "cli":
                print(
                    "claude capability: inference uses Claude Code print mode; "
                    "Muse tools are bridged through structured output",
                    file=output,
                )
            else:
                print(
                    "claude capability: direct Messages API mode; Muse reasoning "
                    "is not sent as Anthropic thinking",
                    file=output,
                )
    print("secrets: not displayed", file=output)
    return 0 if ready else 1


def _server_destination(config: Config) -> str:
    if config.provider == "auto":
        return "Codex API + Claude CLI (select with Muse /model)"
    if config.provider == "claude" and config.claude_transport == "cli":
        return f"Claude CLI ({config.claude_cli} -p)"
    return config.provider_url()


def _existing_shim_is_compatible(
    base_url: str, provider: str, claude_transport: str = "cli"
) -> bool:
    try:
        with urllib.request.urlopen(f"{base_url}/health", timeout=1.0) as response:
            payload = json.loads(response.read().decode("utf-8"))
        return (
            response.status == 200
            and isinstance(payload, dict)
            and payload.get("ok") is True
            and payload.get("service") == "muse-shim"
            and payload.get("provider") == provider
            and payload.get("claude_transport") == claude_transport
        )
    except (OSError, ValueError, json.JSONDecodeError):
        return False


class _LauncherSignal(BaseException):
    def __init__(self, signum: int) -> None:
        self.signum = signum


def _stop_child(process: subprocess.Popen[Any]) -> None:
    if process.poll() is not None:
        return
    process.terminate()
    try:
        process.wait(timeout=3.0)
    except subprocess.TimeoutExpired:
        process.kill()
        process.wait()


def _run_command_index(argv: list[str]) -> int | None:
    """Find the run subcommand without mistaking an option value for it."""
    flag_options = {
        "-h",
        "--help",
        "--no-model-discovery",
        "--portable",
        "--insecure-upstream",
        "--allow-oauth-host-override",
        "--allow-non-loopback",
        "--verbose",
    }
    index = 0
    while index < len(argv):
        token = argv[index]
        if token == "run":
            return index
        if token == "--":
            return None
        if token in flag_options or not token.startswith("-") or "=" in token:
            index += 1
        else:
            # Every non-flag shim option consumes exactly one value.
            index += 2
    return None


def run_muse(config: Config, muse_args: list[str], *, output: Any = None) -> int:
    """Run the local shim and the Muse client as one foreground command."""
    output = output or sys.stderr
    server: MuseShimServer | None = None
    server_thread: threading.Thread | None = None
    child: subprocess.Popen[Any] | None = None
    previous_signal_handlers: dict[int, Any] = {}
    base_url = f"http://{config.host}:{config.port}"

    try:
        server = MuseShimServer((config.host, config.port), config)
    except OSError as error:
        if not _existing_shim_is_compatible(
            base_url, config.provider, config.claude_transport
        ):
            print(
                f"muse-shim could not listen on {base_url}: {error}",
                file=output,
                flush=True,
            )
            return 1
        print(f"muse-shim reusing {base_url}", file=output, flush=True)
    else:
        base_url = f"http://{config.host}:{server.server_port}"
        server_thread = threading.Thread(target=server.serve_forever, daemon=True)
        server_thread.start()
        print(
            f"muse-shim {config.provider} listening on {base_url} "
            f"-> {_server_destination(config)}",
            file=output,
            flush=True,
        )

    environment = os.environ.copy()
    environment["META_API_KEY"] = "local-shim-placeholder"
    command = [
        "muse",
        "--provider",
        "meta",
        "--base-url",
        base_url,
        *muse_args,
    ]

    if threading.current_thread() is threading.main_thread():
        def handle_launcher_signal(signum: int, _frame: Any) -> None:
            raise _LauncherSignal(signum)

        for signum in (signal.SIGHUP, signal.SIGTERM):
            previous_signal_handlers[signum] = signal.getsignal(signum)
            signal.signal(signum, handle_launcher_signal)

    try:
        child = subprocess.Popen(command, env=environment)
        return int(child.wait())
    except FileNotFoundError:
        print("muse-shim could not find `muse` on PATH.", file=output, flush=True)
        return 127
    except KeyboardInterrupt:
        return 130
    except _LauncherSignal as received:
        return 128 + received.signum
    finally:
        if child is not None:
            _stop_child(child)
        for signum, previous in previous_signal_handlers.items():
            signal.signal(signum, previous)
        if server is not None:
            server.shutdown()
            server.server_close()
        if server_thread is not None:
            server_thread.join(timeout=2.0)


def main(argv: list[str] | None = None) -> int:
    raw_argv = list(sys.argv[1:] if argv is None else argv)
    run_index = _run_command_index(raw_argv)
    if run_index is not None:
        args = parse_args(raw_argv[:run_index])
        return run_muse(Config(args), raw_argv[run_index + 1:])
    args = parse_args(raw_argv)
    if args.mode == "doctor":
        config = Config(args)
        config.doctor_provider = args.doctor_provider
        return run_doctor(config)
    config = Config(args)
    if not _is_loopback_host(config.host):
        print(
            "WARNING: muse-shim is listening on a non-loopback address; "
            "remote clients can reach the proxy.",
            file=sys.stderr,
            flush=True,
        )
    server = MuseShimServer((config.host, config.port), config)
    print(
        f"muse-shim {config.provider} listening on http://{config.host}:{server.server_port} "
        f"-> {_server_destination(config)}",
        file=sys.stderr,
        flush=True,
    )
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())