"""
AI Control Service — Phase 24 C (2026-04-03 ogt)

Telegram /ai 指令動態控制 AI Router 狀態
- Redis 狀態優先於 env var
- OPENCLAW_TG_USER_WHITELIST 白名單保護

Redis Keys:
  ai:control:use_router          "true"/"false" (TTL: 30天)
  ai:control:primary_provider    legacy compatibility; only "ollama"/"ollama_gcp_a"
  ai:control:disabled:<provider> "1" disabled / paid provider "0" enabled
                                  only under a run-owned expiring lease

Usage:
  /ai status       — 顯示所有 Provider 狀態 + 當前路由模式
  /ai primary <p>  — legacy compatibility; production always starts at GCP-A
  /ai enable <p>   — 啟用特定 Provider
  /ai disable <p>  — 停用特定 Provider
  /ai cost         — 顯示費用統計
  /ai router on    — 啟用 AIRouter (覆蓋 env var)
  /ai router off   — 停用 AIRouter (回滾到舊 fallback chain)
"""

import hashlib
import json
import re
from typing import Any

import structlog

logger = structlog.get_logger(__name__)

# Redis Key 常數
_AI_ROUTER_KEY = "ai:control:use_router"
_PRIMARY_PROVIDER_KEY = "ai:control:primary_provider"
_DISABLED_KEY_PREFIX = "ai:control:disabled:"
_PAID_CANARY_LOCK_KEY = "ai:control:paid_canary:activation_lock"
_PAID_CANARY_ACTIVE_RECEIPT_KEY = "ai:control:paid_canary:active_receipt"
_PAID_CANARY_EXECUTION_CLAIM_PREFIX = "ai:control:paid_canary:execution_claim:"
_PAID_CANARY_OWNER_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,191}$")
_PAID_CANARY_MIN_LEASE_SECONDS = 60
_PAID_CANARY_MAX_LEASE_SECONDS = 900

_ACQUIRE_PAID_CANARY_LEASE_LUA = r"""
if redis.call('EXISTS', KEYS[1]) == 1 then
  return {0, 'activation_lock_held'}
end
if redis.call('EXISTS', KEYS[4]) == 1 then
  return {0, 'persistent_activation_already_recorded'}
end
if redis.call('GET', KEYS[2]) == '0' or redis.call('GET', KEYS[3]) == '0' then
  return {0, 'paid_provider_already_enabled'}
end
redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2])
redis.call('SET', KEYS[2], '0', 'EX', ARGV[2])
redis.call('SET', KEYS[3], '0', 'EX', ARGV[2])
return {1, 'activation_lease_acquired'}
"""

_ROLLBACK_PAID_CANARY_LUA = r"""
local lock_owner = redis.call('GET', KEYS[1])
local active_owner = redis.call('GET', KEYS[4])
if lock_owner and lock_owner ~= ARGV[1] then
  return {0, 'activation_owner_mismatch'}
end
if active_owner and
  string.sub(active_owner, 1, string.len(ARGV[1]) + 1) ~= ARGV[1] .. '|' then
  return {0, 'active_receipt_owner_mismatch'}
end
redis.call('SET', KEYS[2], '1')
redis.call('SET', KEYS[3], '1')
redis.call('DEL', KEYS[1])
redis.call('DEL', KEYS[4])
return {1, 'paid_canary_disabled'}
"""

_READ_PAID_CANARY_CONTROL_LUA = r"""
return {
  redis.call('GET', KEYS[1]) or '',
  redis.call('TTL', KEYS[1]),
  redis.call('GET', KEYS[2]) or '',
  redis.call('TTL', KEYS[2]),
  redis.call('GET', KEYS[3]) or '',
  redis.call('TTL', KEYS[3]),
  redis.call('GET', KEYS[4]) or ''
}
"""

_ACQUIRE_PAID_CANARY_EXECUTION_CLAIM_LUA = r"""
if redis.call('EXISTS', KEYS[1]) == 1 then
  return {0, 'execution_claim_held'}
end
redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2])
return {1, 'execution_claim_acquired'}
"""

_RELEASE_PAID_CANARY_EXECUTION_CLAIM_LUA = r"""
if redis.call('GET', KEYS[1]) ~= ARGV[1] then
  return {0, 'execution_claim_owner_mismatch'}
end
redis.call('DEL', KEYS[1])
return {1, 'execution_claim_released'}
"""

_SET_PAID_PROVIDER_STATE_LUA = r"""
redis.call('SET', KEYS[2], '1')
redis.call('SET', KEYS[3], '1')
redis.call('DEL', KEYS[1])
redis.call('DEL', KEYS[4])
return {1, 'paired_paid_providers_disabled'}
"""

_PAID_PROVIDER_ADMISSION_LUA = r"""
local claude_state = redis.call('GET', KEYS[2])
local gemini_state = redis.call('GET', KEYS[3])
if claude_state ~= '0' or gemini_state ~= '0' then
  return {0, 'paired_paid_state_not_enabled', ''}
end

local lock_owner = redis.call('GET', KEYS[1])
local active_owner = redis.call('GET', KEYS[4])
if lock_owner then
  local expected_prefix = ARGV[1] .. '.attempt-'
  local run_matches = ARGV[1] ~= '' and
    string.sub(lock_owner, 1, string.len(expected_prefix)) == expected_prefix
  local lease_live = redis.call('TTL', KEYS[1]) > 0 and
    redis.call('TTL', KEYS[2]) > 0 and redis.call('TTL', KEYS[3]) > 0
  local promoted_pending_release = active_owner and
    string.sub(active_owner, 1, string.len(lock_owner) + 1) == lock_owner .. '|' and
    redis.call('TTL', KEYS[2]) == -1 and redis.call('TTL', KEYS[3]) == -1
  if run_matches and (lease_live or promoted_pending_release) then
    return {1, 'run_owned_canary_admitted', lock_owner}
  end
  return {0, 'run_owned_canary_not_admitted', lock_owner}
end

if active_owner then
  return {0, 'legacy_persistent_activation_blocked', active_owner}
end
return {0, 'paid_activation_receipt_missing', ''}
"""

# Free-provider/router compatibility controls retain the historical 30-day
# expiry. Paid providers are executable only during a run-owned expiring lease.
_CONTROL_KEY_TTL = 30 * 24 * 3600  # 30 days
PAID_PROVIDERS = {"claude", "gemini"}

VALID_PROVIDERS = {
    "ollama",
    "ollama_gcp_a",
    "ollama_gcp_b",
    "ollama_local",
    "claude",
    "gemini",
}
VALID_PRIMARY_PROVIDERS = {"ollama", "ollama_gcp_a"}
SHADOW_METADATA_ONLY_PROVIDERS = {
    "nvidia",
    "nemotron",
    "openclaw",
    "openclaw_nemo",
    "ollama_tool",
}


async def _get_redis():
    from src.core.redis_client import get_redis
    return get_redis()


async def get_ai_router_enabled() -> bool | None:
    """
    從 Redis 取得 AIRouter 啟用狀態。
    回傳 None 表示未設定，使用 env var (USE_AI_ROUTER)。
    """
    try:
        r = await _get_redis()
        val = await r.get(_AI_ROUTER_KEY)
        if val is None:
            return None
        return val.decode() == "true" if isinstance(val, bytes) else val == "true"
    except Exception as e:
        logger.warning("ai_control_redis_read_failed", key=_AI_ROUTER_KEY, error=str(e))
        return None


async def set_ai_router_enabled(enabled: bool) -> bool:
    """設定 AIRouter 啟用狀態到 Redis"""
    try:
        r = await _get_redis()
        await r.set(_AI_ROUTER_KEY, "true" if enabled else "false", ex=_CONTROL_KEY_TTL)
        logger.info("ai_control_router_set", enabled=enabled)
        return True
    except Exception as e:
        logger.error("ai_control_router_set_failed", error=str(e))
        return False


async def get_primary_provider() -> str | None:
    """從 Redis 取得主要 Provider，None 表示使用 Router 智慧路由"""
    try:
        r = await _get_redis()
        val = await r.get(_PRIMARY_PROVIDER_KEY)
        if val is None:
            return None
        return val.decode() if isinstance(val, bytes) else val
    except Exception:
        return None


async def set_primary_provider(provider: str) -> bool:
    """Retain only the fixed GCP-A-compatible legacy override."""
    if provider not in VALID_PRIMARY_PROVIDERS:
        return False
    try:
        r = await _get_redis()
        await r.set(_PRIMARY_PROVIDER_KEY, provider, ex=_CONTROL_KEY_TTL)
        logger.info("ai_control_primary_set", provider=provider)
        return True
    except Exception as e:
        logger.error("ai_control_primary_set_failed", error=str(e))
        return False


async def clear_primary_provider() -> bool:
    """清除主要 Provider 設定（回歸智慧路由）"""
    try:
        r = await _get_redis()
        await r.delete(_PRIMARY_PROVIDER_KEY)
        return True
    except Exception:
        return False


async def is_provider_disabled(
    provider: str,
    *,
    run_id: str | None = None,
) -> bool:
    """檢查 Provider 是否被停用"""
    try:
        r = await _get_redis()
        if provider in PAID_PROVIDERS:
            result = await r.eval(
                _PAID_PROVIDER_ADMISSION_LUA,
                4,
                *_paid_canary_keys(),
                str(run_id or ""),
            )
            admitted = int(result[0]) == 1
            admission_status = _decode_redis_scalar(result[1])
            admission_receipt = _decode_redis_scalar(result[2])
            if not admitted:
                return True
            if admission_status == "run_owned_canary_admitted":
                return not _valid_paid_canary_owner(admission_receipt)
            # Unknown admission states are not forward-compatible for paid
            # execution. A server-side script change must be paired with an
            # explicit client policy update.
            return True
        val = await r.get(f"{_DISABLED_KEY_PREFIX}{provider}")
        # Free Ollama lanes retain their historical missing-key=enabled
        # behaviour. Paid providers are admitted only by the paired,
        # receipt-backed transaction above.
        return val is not None
    except Exception:
        # Paid-provider disable truth is fail-closed. Callers can catch this to
        # publish an unavailable readback, but must never re-add a paid lane.
        if provider in PAID_PROVIDERS:
            raise
        return False


async def set_provider_disabled(provider: str, disabled: bool) -> bool:
    """設定 Provider 啟用/停用"""
    if provider not in VALID_PROVIDERS:
        return False
    if provider in PAID_PROVIDERS and not disabled:
        logger.warning(
            "ai_control_paid_provider_generic_enable_blocked",
            provider=provider,
            reason="run_owned_paid_canary_required",
        )
        return False
    try:
        r = await _get_redis()
        key = f"{_DISABLED_KEY_PREFIX}{provider}"
        if provider in PAID_PROVIDERS:
            result = await r.eval(
                _SET_PAID_PROVIDER_STATE_LUA,
                4,
                *_paid_canary_keys(),
            )
            if int(result[0]) != 1:
                logger.warning(
                    "ai_control_paid_provider_set_blocked",
                    provider=provider,
                    disabled=disabled,
                    status=_decode_redis_scalar(result[1]),
                )
                return False
        elif disabled:
            await r.set(key, "1", ex=_CONTROL_KEY_TTL)
        else:
            await r.delete(key)
        logger.info("ai_control_provider_set", provider=provider, disabled=disabled)
        return True
    except Exception as e:
        logger.error("ai_control_provider_set_failed", error=str(e))
        return False


def _paid_canary_keys() -> tuple[str, str, str, str]:
    return (
        _PAID_CANARY_LOCK_KEY,
        f"{_DISABLED_KEY_PREFIX}claude",
        f"{_DISABLED_KEY_PREFIX}gemini",
        _PAID_CANARY_ACTIVE_RECEIPT_KEY,
    )


def _valid_paid_canary_owner(owner: str) -> bool:
    return bool(_PAID_CANARY_OWNER_PATTERN.fullmatch(str(owner or "")))


def _paid_canary_execution_claim_key(run_id: str) -> str:
    digest = hashlib.sha256(str(run_id).encode("utf-8")).hexdigest()
    return f"{_PAID_CANARY_EXECUTION_CLAIM_PREFIX}{digest}"


def _decode_redis_scalar(value: Any) -> str:
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    return str(value or "")


async def acquire_paid_provider_canary_lease(
    owner: str,
    *,
    ttl_seconds: int = 300,
) -> bool:
    """Acquire one run-owned, crash-safe paid-provider activation lease."""

    if not _valid_paid_canary_owner(owner):
        return False
    ttl = int(ttl_seconds)
    if not _PAID_CANARY_MIN_LEASE_SECONDS <= ttl <= _PAID_CANARY_MAX_LEASE_SECONDS:
        return False
    try:
        redis = await _get_redis()
        result = await redis.eval(
            _ACQUIRE_PAID_CANARY_LEASE_LUA,
            4,
            *_paid_canary_keys(),
            owner,
            ttl,
        )
        acquired = int(result[0]) == 1
        logger.info(
            "ai_control_paid_canary_lease_acquire",
            acquired=acquired,
            status=_decode_redis_scalar(result[1]),
            ttl_seconds=ttl,
        )
        return acquired
    except Exception as exc:
        logger.error(
            "ai_control_paid_canary_lease_acquire_failed",
            error_type=type(exc).__name__,
        )
        return False


async def acquire_paid_provider_canary_execution_claim(
    run_id: str,
    owner: str,
    *,
    ttl_seconds: int = 120,
) -> bool:
    """Serialize one Gate 5 authorization before any provider preflight."""

    if not _valid_paid_canary_owner(run_id) or not _valid_paid_canary_owner(owner):
        return False
    ttl = int(ttl_seconds)
    if not _PAID_CANARY_MIN_LEASE_SECONDS <= ttl <= _PAID_CANARY_MAX_LEASE_SECONDS:
        return False
    try:
        redis = await _get_redis()
        result = await redis.eval(
            _ACQUIRE_PAID_CANARY_EXECUTION_CLAIM_LUA,
            1,
            _paid_canary_execution_claim_key(run_id),
            owner,
            ttl,
        )
        acquired = int(result[0]) == 1
        logger.info(
            "ai_control_paid_canary_execution_claim_acquire",
            acquired=acquired,
            status=_decode_redis_scalar(result[1]),
            ttl_seconds=ttl,
        )
        return acquired
    except Exception as exc:
        logger.error(
            "ai_control_paid_canary_execution_claim_acquire_failed",
            error_type=type(exc).__name__,
        )
        return False


async def release_paid_provider_canary_execution_claim(
    run_id: str,
    owner: str,
) -> bool:
    """CAS-release only the caller's pre-execution claim."""

    if not _valid_paid_canary_owner(run_id) or not _valid_paid_canary_owner(owner):
        return False
    try:
        redis = await _get_redis()
        result = await redis.eval(
            _RELEASE_PAID_CANARY_EXECUTION_CLAIM_LUA,
            1,
            _paid_canary_execution_claim_key(run_id),
            owner,
        )
        released = int(result[0]) == 1
        logger.info(
            "ai_control_paid_canary_execution_claim_release",
            released=released,
            status=_decode_redis_scalar(result[1]),
        )
        return released
    except Exception as exc:
        logger.error(
            "ai_control_paid_canary_execution_claim_release_failed",
            error_type=type(exc).__name__,
        )
        return False


async def promote_paid_provider_canary_activation(
    owner: str,
    *,
    completion_record_id: str,
) -> bool:
    """Reject persistent enablement without a separate bounded authorization.

    A Gate 5 receipt authorizes one synthetic comparison only. Keeping this
    compatibility function fail closed prevents an internal caller from
    turning that receipt into future paid-alert traffic.
    """

    _ = (owner, completion_record_id)
    logger.warning(
        "ai_control_paid_canary_persistent_promotion_blocked",
        reason="separate_time_and_cost_bounded_authorization_required",
    )
    return False


async def rollback_paid_provider_canary_activation(owner: str) -> bool:
    """Fail closed without overwriting another run's activation state."""

    if not _valid_paid_canary_owner(owner):
        return False
    try:
        redis = await _get_redis()
        result = await redis.eval(
            _ROLLBACK_PAID_CANARY_LUA,
            4,
            *_paid_canary_keys(),
            owner,
        )
        rolled_back = int(result[0]) == 1
        logger.info(
            "ai_control_paid_canary_activation_rollback",
            rolled_back=rolled_back,
            status=_decode_redis_scalar(result[1]),
        )
        return rolled_back
    except Exception as exc:
        logger.error(
            "ai_control_paid_canary_activation_rollback_failed",
            error_type=type(exc).__name__,
        )
        return False


async def get_paid_provider_canary_control_readback(
    *,
    expected_owner: str | None = None,
    expected_completion_record_id: str | None = None,
) -> dict[str, Any]:
    """Return a content-free, atomic readback of paid canary control state."""

    if expected_owner is not None and not _valid_paid_canary_owner(expected_owner):
        raise ValueError("paid_canary_owner_invalid")
    if expected_completion_record_id is not None and not _valid_paid_canary_owner(
        expected_completion_record_id
    ):
        raise ValueError("paid_canary_completion_record_id_invalid")
    redis = await _get_redis()
    result = await redis.eval(
        _READ_PAID_CANARY_CONTROL_LUA,
        4,
        *_paid_canary_keys(),
    )
    lock_owner = _decode_redis_scalar(result[0])
    lock_ttl = int(result[1])
    claude_value = _decode_redis_scalar(result[2])
    claude_ttl = int(result[3])
    gemini_value = _decode_redis_scalar(result[4])
    gemini_ttl = int(result[5])
    active_receipt = _decode_redis_scalar(result[6])
    active_owner, separator, completion_record_id = active_receipt.partition("|")
    active_completion_receipt_present = bool(
        separator and _valid_paid_canary_owner(completion_record_id)
    )
    active_completion_receipt_matches = bool(
        expected_completion_record_id
        and completion_record_id == expected_completion_record_id
    )
    claude_disabled = claude_value != "0"
    gemini_disabled = gemini_value != "0"
    lock_matches = bool(expected_owner and lock_owner == expected_owner)
    active_matches = bool(expected_owner and active_owner == expected_owner)
    leased_enabled = bool(
        lock_matches
        and not active_receipt
        and not claude_disabled
        and not gemini_disabled
        and lock_ttl > 0
        and claude_ttl > 0
        and gemini_ttl > 0
    )
    legacy_persistent_state_detected = bool(
        active_receipt
        or (
            not claude_disabled
            and not gemini_disabled
            and claude_ttl == -1
            and gemini_ttl == -1
        )
    )
    disabled_verified = bool(
        claude_disabled
        and gemini_disabled
        and not lock_owner
        and not active_receipt
    )
    return {
        "claude_disabled": claude_disabled,
        "gemini_disabled": gemini_disabled,
        "lock_present": bool(lock_owner),
        "lock_owner_matches": lock_matches,
        "lock_ttl_seconds": lock_ttl,
        "active_receipt_present": bool(active_receipt),
        "active_receipt_owner_matches": active_matches,
        "active_completion_receipt_present": active_completion_receipt_present,
        "active_completion_receipt_matches": active_completion_receipt_matches,
        "leased_enabled": leased_enabled,
        "persistent_promoted": False,
        "persistent_enabled": False,
        "legacy_persistent_state_detected": legacy_persistent_state_detected,
        "disabled_verified": disabled_verified,
    }


async def get_status_summary() -> str:
    """
    取得 AI 控制狀態摘要 (Telegram 格式 HTML)
    """
    from src.core.config import get_settings
    settings = get_settings()

    # AIRouter 狀態
    redis_router = await get_ai_router_enabled()
    if redis_router is not None:
        router_status = "✅ ON (Redis 覆蓋)" if redis_router else "❌ OFF (Redis 覆蓋)"
    else:
        router_status = "✅ ON (env)" if settings.USE_AI_ROUTER else "❌ OFF (env)"

    # Primary Provider
    primary = await get_primary_provider()
    primary_str = f"🎯 {primary}" if primary else "🤖 智慧路由 (AIRouter 決定)"

    # Provider 狀態：只讀既有 health/control receipts，不呼叫模型或端點。
    ollama_specs = (
        ("ollama_gcp_a", "GCP-A", "OLLAMA_URL"),
        ("ollama_gcp_b", "GCP-B", "OLLAMA_SECONDARY_URL"),
        ("ollama_local", "111", "OLLAMA_FALLBACK_URL"),
    )
    ollama_health = {provider: "not_observed" for provider, _, _ in ollama_specs}
    try:
        from src.services.ollama_health_monitor import make_cache_key

        health_redis = await _get_redis()
        for provider, _, setting_name in ollama_specs:
            endpoint = str(getattr(settings, setting_name, "") or "").strip()
            if not endpoint:
                ollama_health[provider] = "not_configured"
                continue
            try:
                raw = await health_redis.get(make_cache_key(endpoint))
                if raw:
                    if isinstance(raw, bytes):
                        raw = raw.decode()
                    receipt = json.loads(raw)
                    ollama_health[provider] = str(
                        receipt.get("status") or "invalid_receipt"
                    )
            except Exception as exc:
                ollama_health[provider] = "readback_unavailable"
                logger.warning(
                    "ai_control_ollama_health_receipt_read_failed",
                    provider=provider,
                    error_type=type(exc).__name__,
                )
    except Exception as exc:
        logger.warning(
            "ai_control_ollama_health_readback_unavailable",
            error_type=type(exc).__name__,
        )

    provider_lines = []
    for provider, label, _ in ollama_specs:
        try:
            control = "disabled" if await is_provider_disabled(provider) else "enabled"
        except Exception:
            control = "readback_unavailable"
        health = ollama_health[provider]
        if control == "disabled":
            icon = "⏸️"
        elif health == "healthy":
            icon = "🟢"
        elif health in {"slow", "degraded"}:
            icon = "🟠"
        elif health == "offline":
            icon = "🔴"
        else:
            icon = "⚪"
        provider_lines.append(
            f"  {icon} {label} | health={health} | control={control}"
        )

    paid_control: dict[str, Any] = {}
    route_lease = "readback_unavailable"
    try:
        paid_control = await get_paid_provider_canary_control_readback()
        if paid_control.get("legacy_persistent_state_detected") is True:
            route_lease = "legacy_persistent_state_detected"
        elif (
            paid_control.get("lock_present") is True
            and paid_control.get("claude_disabled") is False
            and paid_control.get("gemini_disabled") is False
            and int(paid_control.get("lock_ttl_seconds") or 0) > 0
        ):
            route_lease = "active_run_owned_lease"
        elif paid_control.get("disabled_verified") is True:
            route_lease = "待單次 run-owned lease"
        else:
            route_lease = "inconsistent_fail_closed"
    except Exception as exc:
        logger.warning(
            "ai_control_paid_control_status_readback_unavailable",
            error_type=type(exc).__name__,
        )

    authentication: dict[str, dict[str, Any]] = {}
    try:
        from src.services.ai_rate_limiter import get_ai_rate_limiter

        limiter = get_ai_rate_limiter()
        for provider in ("claude", "gemini"):
            try:
                authentication[provider] = (
                    await limiter.get_provider_authentication_status(provider)
                )
            except Exception:
                authentication[provider] = {
                    "configured": None,
                    "authenticated": None,
                    "authentication_verified": False,
                    "verification_status": "authentication_readback_unavailable",
                }
    except Exception as exc:
        logger.warning(
            "ai_control_paid_authentication_status_readback_unavailable",
            error_type=type(exc).__name__,
        )

    def _truth_text(value: Any) -> str:
        if value is True:
            return "true"
        if value is False:
            return "false"
        return "unknown"

    for provider, label in (("claude", "Claude"), ("gemini", "Gemini")):
        auth = authentication.get(provider) or {}
        configured = auth.get("configured")
        mode = str(
            getattr(settings, f"{provider.upper()}_EXECUTION_MODE", "shadow")
        ).lower()
        percent = int(
            getattr(settings, f"{provider.upper()}_CANARY_PERCENT", 0) or 0
        )
        if configured is True and auth.get("authenticated") is False:
            icon = "⚠️"
        elif configured is True and route_lease == "待單次 run-owned lease":
            icon = "🧪"
        elif configured is True and route_lease == "active_run_owned_lease":
            icon = "🟢"
        elif configured is True:
            icon = "⚠️"
        else:
            icon = "⚪"
        provider_lines.append(
            f"  {icon} {label} | configured={_truth_text(configured)} | "
            f"authenticated={_truth_text(auth.get('authenticated'))} | "
            "verification="
            f"{auth.get('verification_status', 'authentication_readback_unavailable')} | "
            f"canary={mode} {percent}% | route_lease={route_lease}"
        )
    provider_lines.extend(
        f"  ⚪ {provider} (shadow metadata only)"
        for provider in sorted(SHADOW_METADATA_ONLY_PROVIDERS)
    )

    # 費用統計
    cost_lines = []
    try:
        from src.services.ai_rate_limiter import get_ai_rate_limiter

        limiter = get_ai_rate_limiter()
        for provider in ("claude", "gemini"):
            usage = await limiter.get_usage_stats(provider)
            cost = float(
                (usage.get("total_cost_usd") or {}).get("current") or 0.0
            )
            accounting = usage.get("accounting") or {}
            receipt_backed = bool(
                usage.get("readback_status") == "verified"
                and accounting.get("complete") is True
                and accounting.get("status") == "receipt_backed"
                and accounting.get("degraded") is not True
            )
            cost_lines.append(
                f"  💰 {provider}: ${cost:.6f} | "
                f"receipt={'verified' if receipt_backed else 'degraded'}"
            )
    except Exception as exc:
        logger.warning(
            "ai_control_status_cost_readback_unavailable",
            error_type=type(exc).__name__,
        )
        cost_lines.append("  (accounting receipt readback unavailable)")

    providers_str = "\n".join(provider_lines)
    costs_str = "\n".join(cost_lines) if cost_lines else "  (尚無費用記錄)"

    return (
        f"<b>🤖 AI Router 控制面板</b>\n\n"
        f"<b>路由模式</b>: {router_status}\n"
        f"<b>主要 Provider</b>: {primary_str}\n\n"
        f"<b>Provider 狀態</b>:\n{providers_str}\n\n"
        f"<b>費用統計</b>:\n{costs_str}\n\n"
        f"<code>/ai primary &lt;provider&gt;</code> 切換\n"
        f"<code>/ai router on/off</code> 開關 AIRouter\n"
        f"<code>/ai enable/disable &lt;ollama provider&gt;</code> 控制 Ollama\n"
        f"Claude/Gemini 僅由單次 run-owned canary lease 啟用"
    )


async def handle_ai_command(text: str) -> str:
    """
    處理 /ai 指令，回傳 Telegram HTML 回應。

    Args:
        text: 完整訊息文字 (e.g., "/ai status", "/ai primary gemini")

    Returns:
        HTML 格式回應字串
    """
    parts = text.strip().split()
    if len(parts) < 2:
        return await get_status_summary()

    sub = parts[1].lower()

    if sub == "status":
        return await get_status_summary()

    elif sub == "router":
        if len(parts) < 3:
            return "用法: <code>/ai router on</code> 或 <code>/ai router off</code>"
        action = parts[2].lower()
        if action in ("on", "true", "enable"):
            ok = await set_ai_router_enabled(True)
            return "✅ AIRouter 已啟用 (Redis 覆蓋)" if ok else "❌ 寫入 Redis 失敗"
        elif action in ("off", "false", "disable"):
            ok = await set_ai_router_enabled(False)
            return "⚠️ AIRouter 已停用，回滾舊 fallback chain (Redis 覆蓋)" if ok else "❌ 寫入 Redis 失敗"
        else:
            return f"❌ 未知動作: {action}，請用 on/off"

    elif sub == "primary":
        if len(parts) < 3:
            current = await get_primary_provider()
            if current:
                return f"當前主要 Provider: <b>{current}</b>\n用法: <code>/ai primary &lt;provider&gt;</code> 或 <code>/ai primary auto</code>"
            return "當前: 🤖 智慧路由\n用法: <code>/ai primary &lt;provider&gt;</code>"
        provider = parts[2].lower()
        if provider == "auto":
            ok = await clear_primary_provider()
            return "✅ 已清除主要 Provider，恢復智慧路由" if ok else "❌ 操作失敗"
        if provider not in VALID_PRIMARY_PROVIDERS:
            return (
                f"❌ Provider 不可改寫 production 起點: {provider}\n"
                "production 固定從 ollama_gcp_a 開始"
            )
        ok = await set_primary_provider(provider)
        return f"✅ 主要 Provider 已設為 <b>{provider}</b>" if ok else "❌ 寫入 Redis 失敗"

    elif sub == "enable":
        if len(parts) < 3:
            return "用法: <code>/ai enable &lt;provider&gt;</code>"
        provider = parts[2].lower()
        if provider not in VALID_PROVIDERS:
            return f"❌ 未知 Provider: {provider}"
        if provider in PAID_PROVIDERS:
            return (
                "❌ Claude/Gemini 付費路由只能由 run-owned canary 驗證流程啟用；"
                "一般 /ai enable 已 fail closed"
            )
        ok = await set_provider_disabled(provider, False)
        return f"✅ {provider} 已啟用" if ok else "❌ 操作失敗"

    elif sub == "disable":
        if len(parts) < 3:
            return "用法: <code>/ai disable &lt;provider&gt;</code>"
        provider = parts[2].lower()
        if provider not in VALID_PROVIDERS:
            return f"❌ 未知 Provider: {provider}"
        ok = await set_provider_disabled(provider, True)
        if ok and provider in PAID_PROVIDERS:
            return "⚠️ Claude/Gemini 已成對停用，activation lease/receipt 已清除"
        return f"⚠️ {provider} 已停用" if ok else "❌ 操作失敗"

    elif sub == "cost":
        lines = ["<b>💰 AI 費用 receipt</b>\n"]
        try:
            from src.services.ai_rate_limiter import get_ai_rate_limiter

            limiter = get_ai_rate_limiter()
            total = 0.0
            all_receipt_backed = True
            for provider in ("claude", "gemini"):
                usage = await limiter.get_usage_stats(provider)
                cost = float(
                    (usage.get("total_cost_usd") or {}).get("current") or 0.0
                )
                accounting = usage.get("accounting") or {}
                receipt_backed = bool(
                    usage.get("readback_status") == "verified"
                    and accounting.get("complete") is True
                    and accounting.get("status") == "receipt_backed"
                    and accounting.get("degraded") is not True
                )
                total += cost
                all_receipt_backed = all_receipt_backed and receipt_backed
                lines.append(
                    f"  {provider}: ${cost:.6f} | "
                    f"receipt={'verified' if receipt_backed else 'degraded'}"
                )
            lines.extend(
                [
                    f"\n  <b>總計: ${total:.6f}</b>",
                    (
                        "  ✅ receipt-backed accounting"
                        if all_receipt_backed
                        else "  ⚠️ accounting receipt degraded；禁止假性估算"
                    ),
                ]
            )
        except Exception as exc:
            logger.warning(
                "ai_control_cost_readback_unavailable",
                error_type=type(exc).__name__,
            )
            lines.append(
                "  ⚠️ accounting readback unavailable；未以舊 Redis key 或估算值替代。"
            )
        return "\n".join(lines)

    elif sub == "help":
        return (
            "<b>/ai 指令說明</b>\n\n"
            "<code>/ai status</code> — 顯示所有狀態\n"
            "<code>/ai router on/off</code> — 開關 AIRouter\n"
            "<code>/ai primary &lt;p&gt;</code> — 設主要 Provider\n"
            "<code>/ai primary auto</code> — 恢復智慧路由\n"
            "<code>/ai enable &lt;p&gt;</code> — 啟用 Provider\n"
            "<code>/ai disable &lt;p&gt;</code> — 停用 Provider\n"
            "<code>/ai cost</code> — 費用統計\n\n"
            f"可控 Provider: {', '.join(sorted(VALID_PROVIDERS))}\n"
            "Legacy providers 僅保留 shadow metadata；Claude/Gemini 受 persistent paid gate 控制"
        )

    else:
        return f"❌ 未知子指令: {sub}\n輸入 <code>/ai help</code> 查看說明"