"""
Runtime / provider management for the web UI.

Owns the globals for chat + exec provider selection, runtime creation,
provider detection, and runtime switching. Broadcasts use a late import
to avoid a circular dep with server.py.
"""

from __future__ import annotations

import json
import threading
from typing import Optional


# ---------------------------------------------------------------------------
# Globals — live here so server.py doesn't own provider state directly.
# ---------------------------------------------------------------------------

_CLI_PROVIDERS = {"openai-codex", "gemini-cli"}

_runtime_lock = threading.Lock()

_chat_provider: Optional[str] = None
_chat_model: Optional[str] = None
_chat_runtime = None

_exec_provider: Optional[str] = None
_exec_model: Optional[str] = None

_default_provider: Optional[str] = None
_default_runtime = None
_providers_initialized = False

_available_providers: dict[str, dict] = {}


def _log(text: str) -> None:
    """Log line — surfaces to webui clients and writes to a log file.

    Earlier versions just ``print(text)`` to stdout as a fallback when
    the webui ``_log`` import failed. That polluted CLI chat with
    "[probe] xxx unavailable" lines on every startup. Now: broadcast
    via webui when available, always write to
    ``~/.openprogram/logs/runtime.log``, only mirror to stderr when
    ``OPENPROGRAM_DEBUG_RUNTIME=1``. Per-call sites that DO want to
    surface a one-liner to the user (e.g. "no provider available")
    use ``_user_log`` instead.
    """
    try:
        from openprogram.webui.server import _log as _srv_log
        _srv_log(text)
    except Exception:
        pass
    try:
        from pathlib import Path as _Path
        log_dir = _Path.home() / ".openprogram" / "logs"
        log_dir.mkdir(parents=True, exist_ok=True)
        with open(log_dir / "runtime.log", "a", encoding="utf-8") as f:
            import time as _time
            f.write(f"{_time.strftime('%Y-%m-%d %H:%M:%S')} {text}\n")
    except OSError:
        pass
    import os as _os
    if _os.environ.get("OPENPROGRAM_DEBUG_RUNTIME", "").strip() in ("1", "true", "yes"):
        import sys as _sys
        print(text, file=_sys.stderr, flush=True)


def _any_agent_has_pinned_provider() -> bool:
    """True if any saved agent has a non-empty ``model.provider``.

    Used by ``_init_providers`` to suppress the "no provider available"
    warning when auto-detection fails but an agent has a pin that the
    CLI's ``_get_chat_runtime`` will honour next. Best-effort: any
    error means "we don't know" which we treat conservatively as False
    so a genuinely-broken install still gets the warning.
    """
    try:
        from openprogram.agent.management import manager as _A
        for spec in _A.list_all():
            if spec.model and (spec.model.provider or "").strip():
                return True
    except Exception:
        pass
    return False


def _user_log(text: str) -> None:
    """Important event that should surface to the user.

    Used sparingly — e.g. "no provider available, run `openprogram
    providers setup`". Goes through the same broadcast + file-log path
    as ``_log`` plus a stderr print so the user sees it even when
    running bare ``openprogram`` without other diagnostic output.
    """
    _log(text)
    try:
        import sys as _sys
        print(text, file=_sys.stderr, flush=True)
    except Exception:
        pass


def _broadcast(msg: str) -> None:
    try:
        from openprogram.webui.server import _broadcast as _srv_bc
        _srv_bc(msg)
    except Exception:
        pass


def _broadcast_chat_response(session_id: str, msg_id: str, response: dict) -> None:
    try:
        from openprogram.webui.server import _broadcast_chat_response as _srv_bcr
        _srv_bcr(session_id, msg_id, response)
    except Exception:
        pass


def _get_sessions():
    """Return (conversations dict, lock). Late import to avoid cycle."""
    from openprogram.webui.server import _sessions, _sessions_lock
    return _sessions, _sessions_lock


# ---------------------------------------------------------------------------
# Runtime inspection
# ---------------------------------------------------------------------------

def _prev_rt_closed(rt) -> bool:
    """Check if a Claude Code runtime's process has exited."""
    proc = getattr(rt, "_proc", None)
    return proc is None or proc.poll() is not None


# ---------------------------------------------------------------------------
# Runtime creation — provider-specific setup
# ---------------------------------------------------------------------------

def _preferred_default_model(provider: str) -> str | None:
    """Pick a non-hardcoded default model for a provider based on user config.

    Priority:
      1. Top-level ``default_model`` in ``~/.openprogram/config.json``
         (only when ``default_provider`` matches or is unset).
      2. First id in ``providers.<provider>.enabled_models`` if the user
         has any models enabled for this provider.
      3. ``None`` — caller uses its hardcoded fallback.
    """
    try:
        from openprogram.webui._model_catalog import _read_providers_cfg
        from openprogram.paths import get_config_path
        import json
        try:
            with open(get_config_path(), "r", encoding="utf-8") as f:
                root_cfg = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            root_cfg = {}
        default_provider = root_cfg.get("default_provider")
        default_model = root_cfg.get("default_model")
        if default_model and (not default_provider or default_provider == provider):
            return default_model
        providers_cfg = _read_providers_cfg()
        enabled = (providers_cfg.get(provider, {}).get("enabled_models") or [])
        if enabled:
            return enabled[0]
    except Exception:
        pass
    return None


def _create_runtime_for_visualizer(provider: str, model: str | None = None):
    """Create a runtime appropriate for the web UI.

    Two shapes:
      - Provider listed in ``legacy_providers.PROVIDERS`` (CLI runtimes + the
        classic HTTP subclasses like AnthropicRuntime/OpenAIRuntime/...):
        route through ``legacy_providers.create_runtime`` so their per-provider
        conventions (Codex search=True, Claude Code has_session, ...) apply.
      - Any other provider id present in the HTTP model registry (openrouter,
        groq, cerebras, minimax, mistral, ...): build a plain ``Runtime`` with
        ``model="<provider>:<id>"``. These go through AgentSession end-to-end.
    """
    from openprogram.providers.registry import create_runtime, PROVIDERS

    # If caller didn't pin a model, prefer user config (default_model or
    # the first entry in enabled_models) over the hardcoded PROVIDERS
    # default. Keeps fresh installs on the hardcoded fallback while
    # letting users promote e.g. Opus to default by enabling it in
    # Settings — no manual switch needed every session.
    if model is None:
        model = _preferred_default_model(provider)

    if provider in PROVIDERS:
        kwargs = {"provider": provider}
        if provider == "openai-codex":
            kwargs["search"] = True
        if model:
            kwargs["model"] = model
        return create_runtime(**kwargs)

    # Registry-based provider. Pick a default model if caller didn't specify.
    from openprogram.agentic_programming.runtime import Runtime
    from openprogram.providers import get_model as _get_model, get_models as _get_models
    if model is None:
        models = _get_models(provider)
        # Static registry empty? Fall back to the user's custom_models
        # (post-Fetch rows or hand-pinned ones) before giving up — same
        # rule that drives the Settings model table.
        if not models:
            from openprogram.webui._model_catalog import _read_providers_cfg
            customs = (_read_providers_cfg().get(provider, {}).get("custom_models") or [])
            if not customs:
                raise RuntimeError(f"Provider {provider!r} has no models registered")
            model = customs[0].get("id")
            if not model:
                raise RuntimeError(f"Provider {provider!r} has no models registered")
        else:
            model = models[0].id
    # ``custom_models`` (fetched + manual) are valid runtime targets
    # too. ``get_model`` only looks at the static MODELS dict baked
    # into ``providers/models_generated.py``; without this side-effect
    # registration, every model the user pulled via Fetch
    # (DeepSeek V4, claude-sonnet-4-6, …) trips the
    # ``raise ValueError("Unknown model ...")`` inside Runtime.__init__
    # and the picker switch 500s — even though Settings → Models
    # happily toggles those rows enabled.
    if _get_model(provider, model) is None:
        if not _register_custom_model_in_registry(provider, model):
            raise RuntimeError(f"Unknown model {provider}:{model}")
    return Runtime(model=f"{provider}:{model}")


def _register_custom_model_in_registry(provider: str, model_id: str) -> bool:
    """Look ``model_id`` up in the user's ``custom_models`` for
    ``provider`` and, if present, insert a ``Model`` row into the
    global ``MODELS`` dict so ``providers.get_model`` finds it.

    Side-effect on the module-level registry — deliberate. The
    alternative is plumbing custom-model metadata through every
    callsite that goes "look up Model row → read context_window /
    cost / modalities". Registering once at runtime-construction
    time is the smallest change that makes the registry the single
    source of truth.

    Returns ``True`` when a registration happened, ``False`` when the
    model wasn't in custom_models either (genuine unknown id → caller
    re-raises).
    """
    try:
        from openprogram.webui._model_catalog import (
            _read_providers_cfg,
            _default_api_for,
        )
        from openprogram.providers.models_generated import MODELS
        from openprogram.providers.types import Model, ModelCost
    except Exception:
        return False

    cfg_pcfg = _read_providers_cfg().get(provider, {})
    raw = next(
        (c for c in (cfg_pcfg.get("custom_models") or []) if c.get("id") == model_id),
        None,
    )
    if not raw:
        return False

    api = raw.get("api") or _default_api_for(provider) or "openai-completions"
    inputs: list[str] = list(raw.get("input_modalities") or ["text"])
    # Cost is optional — only stamp the fields the row actually has,
    # default 0.0 for missing keys so ModelCost validates.
    cost = ModelCost(
        input=float(raw.get("input_cost", 0) or 0),
        output=float(raw.get("output_cost", 0) or 0),
        cache_read=float(raw.get("cache_read_cost", 0) or 0),
        cache_write=float(raw.get("cache_write_cost", 0) or 0),
    )
    # Resolve through the catalog's _resolve_base_url (user config →
    # static registry → models.dev) so the row gets the SAME normalised
    # base the rest of the pipeline uses — crucially the Anthropic-wire
    # /v1 strip. Reading models.dev raw here gave Anthropic providers a
    # ``…/anthropic/v1`` base, which the anthropic-messages layer then
    # doubled into ``…/v1/v1/messages`` → 404, so the model registered but
    # was unusable.
    from ._model_catalog import _resolve_base_url
    base_url = _resolve_base_url(provider) or ""

    try:
        m = Model(
            id=model_id,
            name=str(raw.get("name") or model_id),
            api=api,
            provider=provider,
            base_url=base_url,
            reasoning=bool(raw.get("reasoning", False)),
            input=inputs,
            cost=cost,
            context_window=int(raw.get("context_window", 0) or 0),
            max_tokens=int(raw.get("max_tokens", 0) or 0),
        )
    except Exception:
        return False
    MODELS[f"{provider}/{model_id}"] = m
    return True


_PROVIDER_PRIORITY = ("openai-codex", "gemini-cli", "anthropic", "gemini", "openai", "claude-code")
_CLI_BINS = {"openai-codex": "codex", "gemini-cli": "gemini"}


def _build_model_caps(provider_name: str, model_ids: list[str]) -> dict[str, dict]:
    """Return {model_id: {vision, video, audio, reasoning, tools}} for each model."""
    try:
        from openprogram.providers.models_generated import MODELS
        caps: dict[str, dict] = {}
        for mid in model_ids:
            # Try provider-qualified key first, then bare id
            key = f"{provider_name}/{mid}"
            m = MODELS.get(key) or MODELS.get(mid)
            if m is None:
                # Fallback: scan for any key whose model.id matches
                for k, v in MODELS.items():
                    if v.id == mid and (v.provider == provider_name or v.api == provider_name):
                        m = v
                        break
            if m:
                inputs = list(getattr(m, "input", []) or [])
                caps[mid] = {
                    "vision": "image" in inputs,
                    "video": "video" in inputs,
                    "audio": "audio" in inputs,
                    "reasoning": bool(getattr(m, "reasoning", False)),
                    "tools": True,
                }
        return caps
    except Exception:
        return {}


def _probe_one_provider(p_name: str):
    """Probe a single provider. Returns (name, runtime, models) or None on failure.

    `runtime` is kept open so a successful probe can be reused as the default
    runtime without rebuilding. Caller is responsible for closing the ones it
    doesn't keep.

    Constructibility ≠ usability. ``Runtime.__init__`` for HTTP-based
    providers like ``claude-code`` only stamps a base_url + model list
    onto the runtime; nothing actually hits the network. So the
    structural probe alone is happy to pick a provider whose backing
    daemon isn't running, or whose credential is stale-but-present.
    That gets persisted into the new session's ``provider_name`` field
    and then explodes at send time. Defer to ``_is_configured`` — the
    same predicate the Settings page uses — so the auto-detect default
    only picks providers that pass the catalog's own usability check
    (API key present / CLI binary on PATH / local proxy actually
    answering on its port).
    """
    try:
        from openprogram.webui._model_catalog import _is_configured
        if not _is_configured(p_name):
            raise RuntimeError(f"{p_name} not configured")
        if p_name in _CLI_PROVIDERS:
            import shutil as _shutil
            if not _shutil.which(_CLI_BINS.get(p_name, p_name)):
                raise RuntimeError(f"{p_name} not installed")
        rt = _create_runtime_for_visualizer(p_name)
        models = rt.list_models() if hasattr(rt, "list_models") else []
        if rt.model and rt.model not in models:
            models = [rt.model] + models
        return p_name, rt, models
    except Exception as e:
        _log(f"[probe] {p_name} unavailable: {e}")
        return None


def _detect_default_provider() -> tuple:
    """Kept as a back-compat shim; new code goes through `_init_providers`."""
    for p in _PROVIDER_PRIORITY:
        result = _probe_one_provider(p)
        if result is not None:
            _, rt, _models = result
            return p, rt
    return None, None


_rest_probe_started = False


def _probe_rest_async(skip: str | None) -> None:
    """Probe the non-default providers in a background thread.

    The default is already in `_available_providers` from the foreground
    `_init_providers` call. This fills in the rest so the Settings UI has a
    complete provider/model list without blocking startup on it.
    """
    global _rest_probe_started
    if _rest_probe_started:
        return
    _rest_probe_started = True

    def _run():
        from concurrent.futures import ThreadPoolExecutor
        targets = [p for p in _PROVIDER_PRIORITY if p != skip]
        with ThreadPoolExecutor(max_workers=len(targets) or 1) as ex:
            for r in ex.map(_probe_one_provider, targets):
                if r is None:
                    continue
                name, rt, models = r
                _available_providers[name] = {"models": models, "default_model": rt.model, "model_caps": _build_model_caps(name, models)}
                if hasattr(rt, "close"):
                    try:
                        rt.close()
                    except Exception:
                        pass

    threading.Thread(target=_run, name="probe-rest-providers", daemon=True).start()


def _init_providers():
    """Initialize chat and exec provider defaults.

    Foreground only probes providers in priority order until one succeeds;
    the rest are probed asynchronously to populate `_available_providers`
    for the Settings UI without blocking startup.
    """
    global _chat_provider, _chat_model, _chat_runtime
    global _exec_provider, _exec_model
    global _default_provider, _default_runtime
    global _providers_initialized

    with _runtime_lock:
        if _providers_initialized:
            return
        _providers_initialized = True

        provider_name = None
        rt = None
        for p in _PROVIDER_PRIORITY:
            result = _probe_one_provider(p)
            if result is None:
                continue
            name, probe_rt, models = result
            _available_providers[name] = {"models": models, "default_model": probe_rt.model}
            provider_name = name
            rt = probe_rt
            break

        if provider_name:
            # Quiet — banner will show the active provider; users don't
            # need a separate "detect: OK" line above it. Diagnostic
            # path still logs to file via ``_log``.
            _log(f"[detect] {provider_name} OK")
        else:
            # Auto-detect failed, but an agent may still have a pinned
            # provider that bypasses ``_is_configured`` (e.g.
            # ``_is_configured("openai-codex")`` returns False under
            # some config-file layouts even when an OAuth token is in
            # AuthManager). Suppress the user-visible "no provider"
            # warning when any agent has a pinned model.provider —
            # ``_cli_chat.setup._get_chat_runtime`` will try it next
            # and the banner will say so.
            _log("[detect] no provider via auto-detect")
            if not _any_agent_has_pinned_provider():
                _user_log(
                    "openprogram: no LLM provider configured.\n"
                    "  Run `openprogram providers setup` to connect one,\n"
                    "  or `openprogram providers login <name>` for a specific provider."
                )

        _chat_provider = provider_name
        _chat_model = rt.model if rt else None
        _chat_runtime = rt

        _exec_provider = provider_name
        _exec_model = rt.model if rt else None

        _default_provider = provider_name
        _default_runtime = rt

    _probe_rest_async(skip=provider_name)


def _get_session_runtime(session_id: str, msg_id: str = None):
    """Get chat runtime for a conversation, creating if needed.

    Resolution order for the runtime's provider/model:
      1. The conversation's already-attached runtime (sticky once created).
      2. Explicit user choice (``provider_override`` set by ``/model``).
      3. The conversation's agent's configured ``model.provider`` /
         ``model.id`` (agent.json). Honouring this is what makes the
         agent setting load-bearing instead of cosmetic — the previous
         version always used the global auto-detected default and
         silently ignored what the user picked in the agents UI.
      4. Global ``_chat_provider`` fallback.
    """
    _init_providers()

    _sessions, _ = _get_sessions()
    conv = _sessions.get(session_id)
    if conv and conv.get("runtime"):
        return conv["runtime"]

    provider, model = _resolve_session_provider_model(conv)

    if not provider:
        raise RuntimeError(
            "No model available. Enable a model in Settings → Providers "
            "(or install a CLI such as codex/claude/gemini, or set an API key)."
        )

    rt = _create_runtime_for_visualizer(provider, model=model)
    if conv:
        conv["runtime"] = rt
        conv["provider_name"] = provider
    return rt


def _resolve_session_provider_model(conv: dict | None) -> tuple[str | None, str | None]:
    """Pick (provider, model) for a conversation.

    Resolution order:
      1. Explicit ``provider_override`` on the conv (set by ``/model``
         switch). The older ``provider_name`` field is deliberately NOT
         consulted as a user choice — past versions of
         ``_get_session_runtime`` polluted it with the global default, so
         persisted ``provider_name`` is treated as a runtime-cache only.
      2. The conversation's agent's configured model (agent.json
         ``model.provider`` / ``model.id``).
      3. Global ``_chat_provider`` fallback when no agent / no model.

    All three are short-circuited when the user has NOTHING enabled in
    Settings: disabling every provider must mean "no chat runs", even for
    a conversation whose agent pins a model whose credentials still
    happen to be present. Otherwise a send after "disable all" silently
    executes on the pinned default — the exact surprise the user hit.
    """
    if not _enabled_model_keys():
        return None, None

    if conv:
        if conv.get("provider_override"):
            return conv["provider_override"], conv.get("model_override") or _chat_model

        agent_id = conv.get("agent_id")
        if agent_id:
            try:
                from openprogram.agent.management.manager import get as _get_agent
                spec = _get_agent(agent_id)
            except Exception:
                spec = None
            if spec is not None:
                ap = (spec.model.provider or "").strip()
                am = (spec.model.id or "").strip()
                if ap:
                    return ap, am or None

    # Final fallback = the global auto-detected default. Gate it on the
    # user-enabled set: the globals are stamped ONCE at startup, so after
    # the user disables every provider in Settings they'd otherwise keep
    # the conversation pinned to the old (e.g. gpt) default and silently
    # send through it. When nothing is enabled, return no default so the
    # send path raises a clear "enable a model" error instead.
    if _default_is_enabled(_chat_provider, _chat_model):
        return _chat_provider, _chat_model
    return None, None


def _enabled_model_keys() -> set[tuple[str, str]]:
    """Set of ``(provider_id, model_id)`` for every model the user has
    enabled in Settings. Empty when all providers are disabled.

    This is the SAME source the chat model picker reads
    (``list_enabled_models``), so "the picker is empty" and "the top bar
    shows no default" stay in lockstep.
    """
    try:
        from openprogram.webui._model_catalog.listing import list_enabled_models
        keys: set[tuple[str, str]] = set()
        for m in list_enabled_models():
            pid, mid = m.get("provider"), m.get("id")
            if pid and mid:
                keys.add((pid, mid))
        return keys
    except Exception:
        return set()


def _default_is_enabled(provider: str | None, model: str | None) -> bool:
    """True when ``(provider, model)`` is among the user-enabled models.

    Used to suppress a STALE global default: the chat/exec provider+model
    globals are set once at startup from auto-detect, so after the user
    disables every provider the top bar must show no model and the send
    path must not fall back to it.
    """
    if not provider:
        return False
    keys = _enabled_model_keys()
    if not keys:
        return False
    if model:
        bare = model.split(":", 1)[1] if ":" in model else model
        return (provider, model) in keys or (provider, bare) in keys
    # Provider set but no specific model → enabled iff the provider has
    # at least one enabled model.
    return any(p == provider for (p, _m) in keys)


def _get_exec_runtime(no_tools: bool = False):
    """Create a fresh runtime for function execution."""
    _init_providers()
    if not _exec_provider:
        raise RuntimeError(
            "No provider available. Install a CLI (codex/gemini) or set an API key."
        )
    if no_tools and _exec_provider == "openai-codex":
        from openprogram.providers.registry import create_runtime
        rt = create_runtime(
            provider="openai-codex", session_id=None, search=False,
            full_auto=False, sandbox="read-only",
        )
    else:
        rt = _create_runtime_for_visualizer(_exec_provider)
    if _exec_model:
        rt.model = _exec_model
    return rt


def _switch_runtime(provider: str, session_id: str = None, msg_id: str = None):
    """Switch provider. Updates current conversation + global default."""
    global _default_provider, _default_runtime

    with _runtime_lock:
        if session_id and msg_id:
            _broadcast_chat_response(session_id, msg_id, {
                "type": "status",
                "content": f"Switching to {provider}...",
            })

        try:
            if provider == "auto":
                name, rt = _detect_default_provider()
                if name is None:
                    raise RuntimeError("No provider available")
            else:
                name, rt = provider, _create_runtime_for_visualizer(provider)
        except Exception as e:
            if session_id and msg_id:
                _broadcast_chat_response(session_id, msg_id, {
                    "type": "error",
                    "content": f"Failed to set up {provider}: {e}",
                })
            raise

        _default_provider = name
        _default_runtime = rt

        if session_id:
            _sessions, _sessions_lock = _get_sessions()
            with _sessions_lock:
                conv = _sessions.get(session_id)
            if conv:
                conv["runtime"] = _create_runtime_for_visualizer(name)
                conv["provider_name"] = name
                # User-explicit switch — record as override so future
                # resolutions don't fall back to agent config.
                conv["provider_override"] = name
                conv["model_override"] = getattr(rt, "model", None)

        if session_id and msg_id:
            _broadcast_chat_response(session_id, msg_id, {
                "type": "status",
                "content": f"Using {name} ({rt.model})",
            })

        _broadcast(json.dumps({
            "type": "provider_changed",
            "data": _get_provider_info(session_id),
        }))

        return rt


def _get_provider_info(session_id: str = None) -> dict:
    """Get provider info. If session_id given, return that conversation's provider.

    When the conversation has no live runtime yet (lazy restore — runtime
    is only built on the first turn after a server restart), resolve
    provider/model from agent config instead of falling through to the
    global default. Otherwise pre-chat displays show the auto-detected
    fallback even after the user fixed the agent's model.
    """
    if session_id:
        _sessions, _sessions_lock = _get_sessions()
        with _sessions_lock:
            conv = _sessions.get(session_id)
        if conv:
            runtime = conv.get("runtime")
            if runtime is not None:
                provider_name = conv.get("provider_name") or _default_provider
                provider_type = "CLI" if provider_name in _CLI_PROVIDERS else "API"
                return {
                    "provider": provider_name,
                    "type": provider_type,
                    "model": runtime.model,
                    "runtime": type(runtime).__name__,
                    "session_id": getattr(runtime, "_session_id", None),
                }
            provider_name, model = _resolve_session_provider_model(conv)
            if provider_name:
                provider_type = "CLI" if provider_name in _CLI_PROVIDERS else "API"
                return {
                    "provider": provider_name,
                    "type": provider_type,
                    "model": model,
                    "runtime": None,
                    "session_id": None,
                }

    runtime = _default_runtime
    provider_name = _default_provider
    if runtime is None:
        return {"provider": None, "type": None, "model": None,
                "runtime": None, "session_id": None}

    provider_type = "CLI" if provider_name in _CLI_PROVIDERS else "API"
    session_id = getattr(runtime, "_session_id", None)
    return {
        "provider": provider_name,
        "type": provider_type,
        "model": runtime.model,
        "runtime": type(runtime).__name__,
        "session_id": session_id,
    }


# ───────────────────────────────────────────────────────────────────
# Usage sync — dispatcher 直接走 stream_simple, 不通过 runtime.exec,
# 所以 runtime.last_usage 永远是 None / 0. 把 TurnResult.usage 拷过去,
# 让 _broadcast_context_stats 这种读 runtime.last_usage 的下游能拿到
# 真实 token 数. server.py 调一行: sync_turn_usage_to_runtime(runtime, turn_result.usage).
# ───────────────────────────────────────────────────────────────────


def sync_turn_usage_to_runtime(runtime, turn_usage: Optional[dict]) -> None:
    """Mirror TurnResult.usage onto runtime.last_usage for downstream readers."""
    if not runtime or not turn_usage:
        return
    try:
        runtime.last_usage = {
            "input_tokens":  int(turn_usage.get("input_tokens") or 0),
            "output_tokens": int(turn_usage.get("output_tokens") or 0),
            "cache_read":    int(turn_usage.get("cache_read_tokens") or 0),
            "cache_create":  int(turn_usage.get("cache_write_tokens") or 0),
            "total_tokens":  int(turn_usage.get("total_tokens") or 0),
        }
    except Exception:
        pass
