"""AWOOOI SRE war-room conversation manager.

Conversation generation uses the same controlled provider clients and router as
incident analysis.  It never performs infrastructure remediation.  The fixed
policy order is GCP-A Ollama -> GCP-B Ollama -> host111 Ollama -> Claude ->
Gemini; paid providers remain eligible only for an explicitly sanitized,
run-owned canary with a bounded cost contract.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import re
from dataclasses import dataclass, field
from decimal import Decimal, InvalidOperation
from html import escape
from typing import Any
from uuid import uuid4

import structlog

from src.core.config import get_settings
from src.repositories.incident_repository import get_incident_repository
from src.repositories.k8s_repository import get_k8s_repository
from src.services.ai_provider_policy import (
    PAID_PROVIDER_ORDER,
    PRODUCTION_OLLAMA_ORDER,
    PRODUCTION_PROVIDER_ORDER,
)
from src.services.ai_providers.interfaces import (
    AIResult,
    is_provider_enabled_by_env,
)
from src.services.ai_router import get_ai_executor
from src.services.sanitization_service import sanitize
from src.utils.timezone import now_taipei

logger = structlog.get_logger(__name__)

OPENCLAW_PERSONA = """你是 OpenClaw，AWOOOI 平台的 SRE AI 首席顧問。
個性: 精準、果斷、專業，直接給出有證據的判讀。
語氣: 簡短有力，繁體中文，不超過 300 字。
稱呼用戶為「老闆」。
你只能分析與討論；不得執行、宣稱執行或跨 domain 指派修復。
"""

NEMOCLAW_PERSONA = """你是 NemoClaw，AWOOOI 平台的 AI 戰術參謀。
個性: 分析型、質疑假設，補充不同角度。
語氣: 建設性、繁體中文，不超過 200 字。
稱呼用戶為「老闆」。評論 OpenClaw 時說「我補充」或「我有不同看法」。
你只能分析與討論；不得執行、宣稱執行或跨 domain 指派修復。
"""

_CORRELATION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$")
_RECEIPT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$")
_PROVIDER_STATUS_TERMS = (
    "狀態",
    "可用",
    "啟用",
    "鎖定",
    "硬鎖",
    "provider status",
    "route status",
)
_PROVIDER_TERMS = (
    "ollama",
    "gcp-a",
    "gcp-b",
    "111",
    "claude",
    "gemini",
    "provider",
    "模型",
    "路由",
)
_IPV4_OCTET = r"(?:25[0-5]|2[0-4]\d|1?\d?\d)"
_PRIVATE_IPV4 = re.compile(
    rf"\b(?:10\.{_IPV4_OCTET}\.{_IPV4_OCTET}\.{_IPV4_OCTET}"
    rf"|192\.168\.{_IPV4_OCTET}\.{_IPV4_OCTET}"
    rf"|172\.(?:1[6-9]|2\d|3[01])\.{_IPV4_OCTET}\.{_IPV4_OCTET})\b"
)
_SANITIZER_PRIVATE_IP_MARKER = re.compile(r"\[PRIVATE_IP:[^\]\r\n]+\]")


def _cloud_chat_dlp(text: str) -> str:
    """Remove RFC1918 values after the legacy sanitizer's visible marker."""

    redacted = _SANITIZER_PRIVATE_IP_MARKER.sub("[PRIVATE_IP_REDACTED]", text)
    return _PRIVATE_IPV4.sub("[PRIVATE_IP_REDACTED]", redacted)


@dataclass(frozen=True)
class ChatRouteResult:
    """Content plus public-safe execution evidence for one chat generation."""

    text: str
    success: bool
    provider: str
    model: str
    fallback_reason: str
    policy_order: tuple[str, ...] = PRODUCTION_PROVIDER_ORDER
    admitted_provider_order: tuple[str, ...] = PRODUCTION_OLLAMA_ORDER
    cost_receipt_id: str | None = None
    tokens: int = 0
    cost_usd: float = 0.0
    runtime_readback: dict[str, Any] = field(default_factory=dict)
    infrastructure_remediation_write_allowed: bool = False
    cross_domain_execution_allowed: bool = False

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema_version": "sre_chat_provider_route_receipt_v1",
            "success": self.success,
            "text": self.text,
            "provider": self.provider,
            "model": self.model,
            "fallback_reason": self.fallback_reason,
            "policy_order": list(self.policy_order),
            "admitted_provider_order": list(self.admitted_provider_order),
            "cost_receipt_id": self.cost_receipt_id,
            "tokens": self.tokens,
            "cost_usd": round(self.cost_usd, 8),
            "runtime_readback": dict(self.runtime_readback),
            "infrastructure_remediation_write_allowed": False,
            "cross_domain_execution_allowed": False,
        }

    def render_html(self) -> str:
        body = escape(self.text or "AI 路由目前無可驗證回應。", quote=False)
        receipt = self.cost_receipt_id or "none"
        evidence = (
            f"provider={self.provider} | model={self.model} | "
            f"fallback={self.fallback_reason} | cost_receipt={receipt}"
        )
        return f"{body}\n\n<i>🤖 {escape(evidence, quote=False)}</i>"


class ChatManager:
    """AWOOOI SRE chat manager with a fail-closed provider boundary."""

    async def get_system_context(self) -> str:
        """Collect a bounded live context for advisory conversation only."""

        now = now_taipei()
        k8s = get_k8s_repository()
        incidents = get_incident_repository()

        try:
            k8s_status = await k8s.get_pod_status_summary(namespace="awoooi-prod")
            cluster_info = (
                f"Cluster: {k8s_status['running']}/{k8s_status['total']} Pods Running"
            )
            if k8s_status.get("problem_pods"):
                cluster_info += f", {len(k8s_status['problem_pods'])} 異常"
        except Exception:
            cluster_info = "Cluster: 無法取得狀態"

        try:
            active_incidents = await incidents.get_active()
            if active_incidents:
                lines = [
                    (
                        f"- {incident.incident_id}: {incident.status.value} "
                        f"(SEV {incident.severity.value})"
                    )
                    for incident in active_incidents[:3]
                ]
                incident_summary = "\n".join(lines)
            else:
                incident_summary = "無活躍告警"
        except Exception:
            incident_summary = "無法取得告警"

        return (
            f"## 系統狀態 ({now.strftime('%Y-%m-%d %H:%M')} 台北)\n"
            f"- {cluster_info}\n"
            f"- 活躍告警: {incident_summary}\n"
        )

    @staticmethod
    def _is_provider_status_question(message: str) -> bool:
        normalized = message.casefold()
        return any(term in normalized for term in _PROVIDER_TERMS) and any(
            term in normalized for term in _PROVIDER_STATUS_TERMS
        )

    @staticmethod
    def _safe_correlation(value: Any, fallback: str) -> str:
        candidate = str(value or "").strip()
        return candidate if _CORRELATION_ID.fullmatch(candidate) else fallback

    @staticmethod
    def _paid_canary_context_is_bounded(route_context: dict[str, Any]) -> bool:
        if route_context.get("synthetic_validation") is not True:
            return False
        if route_context.get("data_classification") != "sanitized":
            return False
        if route_context.get("infrastructure_remediation_write_allowed") is not False:
            return False
        if not all(
            _CORRELATION_ID.fullmatch(str(route_context.get(field) or ""))
            for field in ("trace_id", "run_id", "work_item_id")
        ):
            return False
        try:
            cost_cap = Decimal(str(route_context["paid_canary_max_cost_usd"]))
            output_cap = int(route_context["paid_canary_max_output_tokens"])
        except (InvalidOperation, KeyError, TypeError, ValueError):
            return False
        return bool(
            cost_cap.is_finite()
            and Decimal("0") < cost_cap <= Decimal("0.25")
            and 1 <= output_cap <= 512
        )

    async def _paid_provider_candidates(
        self,
        route_context: dict[str, Any] | None,
    ) -> tuple[list[str], dict[str, Any]]:
        """Return paid lanes admitted before their atomic reservation step."""

        payload = dict(route_context or {})
        readback: dict[str, Any] = {
            "schema_version": "sre_chat_paid_canary_admission_v1",
            "status": "not_requested",
            "run_owned_lease_verified": False,
            "cost_policy_preflight_verified": False,
            "candidates": [],
        }
        if not self._paid_canary_context_is_bounded(payload):
            readback["status"] = "bounded_canary_context_missing"
            return [], readback

        run_id = str(payload["run_id"])
        try:
            from src.services.ai_control import (
                get_paid_provider_canary_control_readback,
                is_provider_disabled,
            )

            control = await get_paid_provider_canary_control_readback(
                expected_owner=run_id,
            )
            if not control.get("leased_enabled"):
                readback["status"] = "run_owned_canary_not_admitted"
                return [], readback
            readback["run_owned_lease_verified"] = True

            from src.services.ai_rate_limiter import get_ai_rate_limiter

            limiter = get_ai_rate_limiter()
            settings = get_settings()
            admitted: list[str] = []
            provider_states: dict[str, str] = {}
            for provider in PAID_PROVIDER_ORDER:
                prefix = provider.upper()
                configured = bool(getattr(settings, f"{prefix}_API_KEY", ""))
                mode = (
                    str(getattr(settings, f"{prefix}_EXECUTION_MODE", "shadow"))
                    .strip()
                    .lower()
                )
                percent = max(
                    0,
                    min(100, int(getattr(settings, f"{prefix}_CANARY_PERCENT", 0))),
                )
                bucket = int(hashlib.sha256(run_id.encode()).hexdigest()[:8], 16) % 100
                if not (
                    configured
                    and is_provider_enabled_by_env(provider)
                    and mode == "canary"
                    and percent > 0
                    and bucket < percent
                ):
                    provider_states[provider] = "configured_canary_not_selected"
                    continue
                if await is_provider_disabled(provider, run_id=run_id):
                    provider_states[provider] = "run_owned_canary_not_admitted"
                    continue
                usage = await limiter.get_usage_stats(provider)
                pricing = usage.get("pricing_policy") or {}
                if (
                    pricing.get("supported") is not True
                    or usage.get("cost_exceeded") is True
                ):
                    provider_states[provider] = "cost_policy_not_ready"
                    continue
                admitted.append(provider)
                provider_states[provider] = "admitted_atomic_reservation_pending"

            readback.update(
                status=("admitted" if admitted else "no_paid_candidate_admitted"),
                cost_policy_preflight_verified=bool(admitted),
                candidates=list(admitted),
                provider_states=provider_states,
            )
            return admitted, readback
        except Exception as error:
            logger.warning(
                "sre_chat_paid_canary_admission_unavailable",
                error_type=type(error).__name__,
            )
            readback["status"] = "paid_admission_readback_unavailable"
            return [], readback

    async def get_provider_runtime_readback(
        self,
        route_context: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Read current provider policy/health without guessing a lock state."""

        health = {provider: "not_checked" for provider in PRODUCTION_OLLAMA_ORDER}
        selected_primary = "ollama_gcp_a"
        observed_fallback: str | None = None
        health_status = "unavailable"
        try:
            from src.services.ollama_failover_manager import (
                get_ollama_failover_manager,
            )

            route = await get_ollama_failover_manager().select_provider(
                task_type="chat_status",
                emit_side_effects=False,
            )
            selected_primary = route.primary.provider_name
            observed_fallback = (
                route.observed_fallback.provider_name
                if route.observed_fallback
                else None
            )
            health["ollama_gcp_a"] = route.health_gcp_a.status.value
            if route.health_gcp_b is not None:
                health["ollama_gcp_b"] = route.health_gcp_b.status.value
            if route.health_local is not None:
                health["ollama_local"] = route.health_local.status.value
            health_status = "verified"
        except Exception as error:
            logger.warning(
                "sre_chat_ollama_runtime_readback_unavailable",
                error_type=type(error).__name__,
            )

        paid_candidates, paid_admission = await self._paid_provider_candidates(
            route_context
        )
        bounded_context = self._paid_canary_context_is_bounded(
            dict(route_context or {})
        )
        paid_control: dict[str, Any] = {
            "readback_status": "unavailable",
            "route_lease_state": "readback_unavailable",
            "lock_present": False,
            "lock_ttl_seconds": -2,
            "disabled_verified": False,
        }
        try:
            from src.services.ai_control import (
                get_paid_provider_canary_control_readback,
            )

            expected_owner = (
                str((route_context or {}).get("run_id") or "")
                if bounded_context
                else None
            )
            control = await get_paid_provider_canary_control_readback(
                expected_owner=expected_owner,
            )
            if paid_admission.get("run_owned_lease_verified") is True:
                route_lease_state = "active_for_this_run"
            elif control.get("legacy_persistent_state_detected") is True:
                route_lease_state = "legacy_persistent_state_detected"
            elif control.get("lock_present") is True:
                route_lease_state = "active_for_another_run"
            elif control.get("disabled_verified") is True:
                route_lease_state = "inactive_disabled_after_canary"
            else:
                route_lease_state = "inconsistent_fail_closed"
            paid_control = {
                "readback_status": "verified",
                "route_lease_state": route_lease_state,
                "lock_present": bool(control.get("lock_present")),
                "lock_ttl_seconds": int(control.get("lock_ttl_seconds") or -1),
                "claude_disabled": bool(control.get("claude_disabled", True)),
                "gemini_disabled": bool(control.get("gemini_disabled", True)),
                "disabled_verified": bool(control.get("disabled_verified")),
                "persistent_enabled": bool(control.get("persistent_enabled")),
                "legacy_persistent_state_detected": bool(
                    control.get("legacy_persistent_state_detected")
                ),
            }
        except Exception as error:
            logger.warning(
                "sre_chat_paid_control_readback_unavailable",
                error_type=type(error).__name__,
            )

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

            limiter = get_ai_rate_limiter()
            raw_authentication = await asyncio.gather(
                *(
                    limiter.get_provider_authentication_status(provider)
                    for provider in PAID_PROVIDER_ORDER
                ),
                return_exceptions=True,
            )
            for provider, authentication in zip(
                PAID_PROVIDER_ORDER,
                raw_authentication,
                strict=True,
            ):
                if isinstance(authentication, dict):
                    authentication_readbacks[provider] = authentication
                else:
                    authentication_readbacks[provider] = {
                        "authenticated": None,
                        "authentication_verified": False,
                        "verification_status": "authentication_readback_unavailable",
                    }
        except Exception as error:
            logger.warning(
                "sre_chat_paid_authentication_readback_unavailable",
                error_type=type(error).__name__,
            )

        settings = get_settings()
        paid: dict[str, Any] = {}
        provider_states = paid_admission.get("provider_states") or {}
        for provider in PAID_PROVIDER_ORDER:
            prefix = provider.upper()
            configured = bool(getattr(settings, f"{prefix}_API_KEY", ""))
            execution_mode = (
                str(getattr(settings, f"{prefix}_EXECUTION_MODE", "shadow"))
                .strip()
                .lower()
            )
            canary_percent = int(getattr(settings, f"{prefix}_CANARY_PERCENT", 0))
            authentication = authentication_readbacks.get(provider) or {
                "authenticated": None,
                "authentication_verified": False,
                "verification_status": "authentication_readback_unavailable",
            }
            if provider in paid_candidates:
                runtime_state = "run_owned_canary_admitted"
            elif not configured:
                runtime_state = "not_configured"
            elif not is_provider_enabled_by_env(provider):
                runtime_state = "disabled_by_environment"
            elif execution_mode != "canary" or canary_percent <= 0:
                runtime_state = "configured_non_canary"
            elif authentication.get("authenticated") is False:
                runtime_state = "configured_canary_authentication_rejected"
            elif authentication.get("authentication_verified") is not True:
                runtime_state = "configured_canary_authentication_not_verified"
            elif paid_control["route_lease_state"] == "inactive_disabled_after_canary":
                runtime_state = (
                    "configured_authenticated_canary_waiting_for_run_owned_lease"
                )
            else:
                runtime_state = str(
                    provider_states.get(provider)
                    or paid_admission.get("status")
                    or "runtime_readback_unavailable"
                )
            paid[provider] = {
                "configured": configured,
                "environment_enabled": is_provider_enabled_by_env(provider),
                "execution_mode": execution_mode,
                "canary_percent": canary_percent,
                "authenticated": authentication.get("authenticated"),
                "authentication_verified": bool(
                    authentication.get("authentication_verified") is True
                ),
                "verification_status": str(
                    authentication.get("verification_status")
                    or "authentication_readback_unavailable"
                ),
                "route_lease_state": paid_control["route_lease_state"],
                "runtime_state": runtime_state,
                "run_owned_admitted": provider in paid_candidates,
            }

        return {
            "schema_version": "sre_chat_provider_runtime_readback_v1",
            "policy_order": list(PRODUCTION_PROVIDER_ORDER),
            "selected_primary": selected_primary,
            "observed_fallback": observed_fallback,
            "health_readback_status": health_status,
            "ollama_health": health,
            "paid": paid,
            "paid_admission": paid_admission,
            "paid_control": paid_control,
            "provider_route_switch_performed": False,
            "infrastructure_remediation_write_allowed": False,
            "cross_domain_execution_allowed": False,
        }

    @staticmethod
    def _format_provider_runtime_readback(readback: dict[str, Any]) -> str:
        health = readback.get("ollama_health") or {}
        paid = readback.get("paid") or {}

        def paid_line(label: str, provider: str) -> str:
            state = paid.get(provider) or {}
            authenticated = state.get("authenticated")
            authenticated_text = (
                "true"
                if authenticated is True
                else "false"
                if authenticated is False
                else "unknown"
            )
            return (
                f"{label}：configured={str(bool(state.get('configured'))).lower()}｜"
                f"authenticated={authenticated_text}｜"
                f"auth_verified={str(bool(state.get('authentication_verified'))).lower()}｜"
                f"verification={state.get('verification_status', 'authentication_readback_unavailable')}｜"
                f"canary={state.get('execution_mode', 'unknown')} "
                f"{int(state.get('canary_percent') or 0)}%｜"
                f"route_lease={state.get('route_lease_state', 'readback_unavailable')}"
            )

        return "\n".join(
            [
                "AI 路由即時讀回",
                "順序：GCP-A → GCP-B → 111 → Claude → Gemini",
                (
                    "Ollama："
                    f"GCP-A={health.get('ollama_gcp_a', 'not_checked')}；"
                    f"GCP-B={health.get('ollama_gcp_b', 'not_checked')}；"
                    f"111={health.get('ollama_local', 'not_checked')}"
                ),
                paid_line("Claude", "claude"),
                paid_line("Gemini", "gemini"),
                "此讀回未切換 provider，也未執行任何主機或服務修復。",
            ]
        )

    @staticmethod
    def _result_provider(provider: str) -> str:
        # The registry's GCP-A compatibility alias returns ``ollama``.  Within
        # this fixed executor order that identity can only be the first GCP-A hop.
        return "ollama_gcp_a" if provider == "ollama" else provider

    @staticmethod
    def _result_model(provider: str, requested_ollama_model: str) -> str:
        settings = get_settings()
        if provider in {"ollama_gcp_a", "ollama_gcp_b"}:
            return str(settings.OLLAMA_HEALTH_CHECK_MODEL)
        if provider == "ollama_local":
            return requested_ollama_model
        if provider == "claude":
            return str(settings.CLAUDE_MODEL)
        if provider == "gemini":
            return str(settings.GEMINI_MODEL)
        return "none"

    @staticmethod
    def _fallback_reason(result: AIResult, provider: str) -> str:
        if not result.success:
            status = str(result.audit_metadata.get("status") or "").strip()
            if status and re.fullmatch(r"[a-z0-9_:-]{1,96}", status):
                return f"terminal_{status}"
            return "canonical_route_exhausted_or_policy_blocked"
        if result.from_cache:
            return "current_policy_cache_hit"
        try:
            position = PRODUCTION_PROVIDER_ORDER.index(provider) + 1
        except ValueError:
            return "provider_identity_outside_policy"
        if position == 1:
            return "canonical_primary_succeeded"
        return f"canonical_position_{position}_after_prior_hops_no_success"

    @staticmethod
    def _cost_receipt_id(result: AIResult, provider: str) -> str | None:
        if provider not in PAID_PROVIDER_ORDER:
            return None
        candidate = str(
            result.audit_metadata.get("generation_receipt_id") or ""
        ).strip()
        return candidate if _RECEIPT_ID.fullmatch(candidate) else None

    @staticmethod
    def _plain_response(raw_response: str) -> str:
        text = re.sub(
            r"<think>.*?</think>",
            "",
            str(raw_response or ""),
            flags=re.DOTALL,
        ).strip()
        if not text:
            return ""
        try:
            payload = json.loads(text)
        except (TypeError, ValueError):
            return text
        if isinstance(payload, dict):
            for field_name in ("response", "message", "description", "reasoning"):
                value = payload.get(field_name)
                if isinstance(value, str) and value.strip():
                    return value.strip()
        return text

    async def route_chat(
        self,
        *,
        role: str,
        system_prompt: str,
        user_message: str,
        route_context: dict[str, Any] | None = None,
    ) -> ChatRouteResult:
        """Generate one advisory response through the controlled provider route."""

        if self._is_provider_status_question(user_message):
            readback = await self.get_provider_runtime_readback(route_context)
            return ChatRouteResult(
                text=self._format_provider_runtime_readback(readback),
                success=True,
                provider="runtime_readback",
                model="deterministic_policy",
                fallback_reason="provider_status_request_no_generation",
                admitted_provider_order=PRODUCTION_OLLAMA_ORDER,
                runtime_readback=readback,
            )

        persona = OPENCLAW_PERSONA if role == "openclaw" else NEMOCLAW_PERSONA
        requested_ollama_model = (
            str(get_settings().OPENCLAW_DEFAULT_MODEL)
            if role == "openclaw"
            else "deepseek-r1:14b"
        )
        prompt = _cloud_chat_dlp(
            sanitize(
                (
                    f"{persona}\n\n{system_prompt}\n\n"
                    f"使用者問題：\n{user_message}\n\n"
                    "只能回覆分析；不得呼叫 executor、不得聲稱完成修復。"
                ),
                source_label="awoooi_sre_group_chat",
            )
        )
        prompt_sha256 = hashlib.sha256(prompt.encode("utf-8")).hexdigest()
        seed = uuid4().hex
        requested_context = dict(route_context or {})
        trace_id = self._safe_correlation(
            requested_context.get("trace_id"),
            f"sre-chat-trace:{seed}",
        )
        run_id = self._safe_correlation(
            requested_context.get("run_id"),
            f"sre-chat-run:{seed}",
        )
        work_item_id = self._safe_correlation(
            requested_context.get("work_item_id"),
            "SRE-CHAT-ADVISORY",
        )
        context: dict[str, Any] = {
            "trace_id": trace_id,
            "run_id": run_id,
            "work_item_id": work_item_id,
            "intent_hint": "chat",
            "task_type": "chat",
            "data_classification": "sanitized",
            # Ordinary SRE conversation is real traffic, never synthetic.  Its
            # GCP boundary is the exact prompt-bound chat receipt below.
            "sre_chat_sanitization_receipt": {
                "schema_version": "sre_chat_sanitization_receipt_v1",
                "status": "verified",
                "sanitizer": "src.services.sanitization_service.sanitize",
                "source_label": "awoooi_sre_group_chat",
                "data_classification": "sanitized",
                "prompt_sha256": prompt_sha256,
                "trace_id": trace_id,
                "run_id": run_id,
                "work_item_id": work_item_id,
                "raw_input_forwarded": False,
                "raw_log_payload_forwarded": False,
                "secret_value_exposed": False,
                "private_network_value_exposed": False,
                "infrastructure_remediation_write_allowed": False,
            },
            "ollama_model": requested_ollama_model,
            "enforce_ollama_first": True,
            "cloud_transport_auto_preflight": True,
            "infrastructure_remediation_write_allowed": False,
            "executor_invocation_allowed": False,
            "agent99_dispatch_allowed": False,
            "cross_domain_execution_allowed": False,
            "canonical_asset_identity": {
                "canonical_id": "service:awoooi:sre-war-room-chat",
                "resolution_status": "resolved",
                "asset_domain": "conversation",
                "circuit_state": "closed",
            },
        }

        bounded_canary = self._paid_canary_context_is_bounded(requested_context)
        require_local = False
        if bounded_canary:
            context.update(
                # Synthetic markers are reserved for the explicitly bounded
                # paid-provider canary contract and bind its exact prompt.
                synthetic_validation=True,
                synthetic_prompt_sha256=prompt_sha256,
                paid_canary_max_cost_usd=str(
                    requested_context["paid_canary_max_cost_usd"]
                ),
                paid_canary_max_output_tokens=int(
                    requested_context["paid_canary_max_output_tokens"]
                ),
            )
        else:
            # Interactive chat must not wait indefinitely on one cold Ollama
            # hop. Paid validation canaries retain their separate 300s client
            # contract and therefore do not inherit this interactive timeout.
            context["provider_timeout_seconds"] = 35
            context["sre_chat_sanitization_receipt"]["provider_timeout_seconds"] = 35

        paid_candidates, admission = await self._paid_provider_candidates(
            {**requested_context, **context} if bounded_canary else None
        )
        admitted_order = [*PRODUCTION_OLLAMA_ORDER, *paid_candidates]
        result = await get_ai_executor().execute(
            prompt,
            provider_order=admitted_order,
            context=context,
            cache_ttl=300,
            require_local=require_local,
        )
        provider = self._result_provider(str(result.provider or "none"))
        receipt_id = self._cost_receipt_id(result, provider)
        paid_cost_cap = float(context.get("paid_canary_max_cost_usd") or 0.0)
        paid_receipt_invalid = bool(
            result.success
            and provider in PAID_PROVIDER_ORDER
            and (
                not receipt_id
                or float(result.cost_usd or 0.0) <= 0.0
                or float(result.cost_usd or 0.0) > paid_cost_cap
            )
        )
        reported_admitted_order = admitted_order
        if paid_receipt_invalid:
            logger.error(
                "sre_chat_paid_result_missing_cost_receipt",
                provider=provider,
            )
            return ChatRouteResult(
                text="付費 Provider 回應缺少費用 receipt，已拒絕呈現。",
                success=False,
                provider=provider,
                model=self._result_model(provider, requested_ollama_model),
                fallback_reason="paid_generation_cost_receipt_invalid",
                admitted_provider_order=tuple(reported_admitted_order),
                runtime_readback={"paid_admission": admission},
            )

        body = self._plain_response(result.raw_response)
        if not result.success or not body:
            readback = await self.get_provider_runtime_readback(
                requested_context if bounded_canary else None
            )
            body = (
                "目前沒有可驗證的模型回應；以下狀態來自 runtime readback，"
                "不是推測，也未執行修復。\n"
                + self._format_provider_runtime_readback(readback)
            )
        else:
            readback = {"paid_admission": admission}

        logger.info(
            "sre_chat_provider_route_completed",
            provider=provider,
            model=self._result_model(provider, requested_ollama_model),
            success=bool(result.success and body),
            fallback_reason=self._fallback_reason(result, provider),
            cost_receipt_present=bool(receipt_id),
            infrastructure_remediation_write_allowed=False,
            cross_domain_execution_allowed=False,
        )
        return ChatRouteResult(
            text=body,
            success=bool(result.success and body),
            provider=provider,
            model=self._result_model(provider, requested_ollama_model),
            fallback_reason=self._fallback_reason(result, provider),
            admitted_provider_order=tuple(reported_admitted_order),
            cost_receipt_id=receipt_id,
            tokens=max(0, int(result.tokens or 0)),
            cost_usd=max(0.0, float(result.cost_usd or 0.0)),
            runtime_readback=readback,
        )

    async def _call_openclaw(
        self,
        system_prompt: str,
        user_message: str,
        route_context: dict[str, Any] | None = None,
    ) -> str | None:
        result = await self.route_chat(
            role="openclaw",
            system_prompt=system_prompt,
            user_message=user_message,
            route_context=route_context,
        )
        return result.render_html()

    async def _call_nemotron(
        self,
        system_prompt: str,
        user_message: str,
        route_context: dict[str, Any] | None = None,
    ) -> str | None:
        result = await self.route_chat(
            role="nemoclaw",
            system_prompt=system_prompt,
            user_message=user_message,
            route_context=route_context,
        )
        return result.render_html()

    async def generate_response(
        self,
        user_id: int,  # noqa: ARG002
        username: str,  # noqa: ARG002
        message_text: str,
    ) -> str:
        """Generate the requested single- or dual-persona response."""

        context = await self.get_system_context()
        text = message_text.strip()

        if text.lower().startswith("@openclaw"):
            message = text[9:].strip() or text
            result = await self._call_openclaw(context, message)
            return f"🦞 <b>OpenClaw:</b>\n{result or '🔴 OpenClaw 無響應'}"

        if text.lower().startswith("@nemo"):
            message = text[5:].strip() or text
            result = await self._call_nemotron(context, message)
            return f"🤖 <b>NemoClaw:</b>\n{result or '🔴 NemoClaw 無響應'}"

        openclaw_task = asyncio.create_task(self._call_openclaw(context, text))
        nemo_task = asyncio.create_task(
            self._call_nemotron(
                context,
                f"老闆問了: {text}\n\n請從 NemoClaw 角度補充或評論。",
            )
        )
        try:
            openclaw_raw = await asyncio.wait_for(openclaw_task, timeout=40.0)
        except TimeoutError:
            openclaw_raw = None
        openclaw_block = f"🦞 <b>OpenClaw:</b>\n{openclaw_raw or '🔴 無響應'}"

        try:
            nemo_raw = await asyncio.wait_for(nemo_task, timeout=60.0)
        except TimeoutError:
            nemo_raw = None
        if nemo_raw:
            return f"{openclaw_block}\n\n🤖 <b>NemoClaw:</b>\n{nemo_raw}"
        return openclaw_block


_chat_manager: ChatManager | None = None


def get_chat_manager() -> ChatManager:
    global _chat_manager
    if _chat_manager is None:
        _chat_manager = ChatManager()
    return _chat_manager