"""
Local HTTP bridge for Web UIs.

This module intentionally uses only the Python standard library so the Web
bridge can run without adding a framework dependency.
"""

import json
import io
import hmac
import ipaddress
import os
import shutil
import tempfile
import threading
import time
import urllib.error
import urllib.request
import webbrowser
import zipfile
from decimal import Decimal, InvalidOperation
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path, PurePosixPath
from zipfile import ZipInfo
from urllib.parse import parse_qs, quote, unquote, urlencode, urlparse

import yaml

from .agent import Agent
from .config import AgentConfig
from .runtime import (
    active_user_model,
    auth_config,
    clear_auth_session,
    global_api_key,
    load_or_create_bridge_token,
    load_auth_session,
    load_user_models,
    load_user_profile,
    migrate_legacy_agent_memory,
    normalize_api_base,
    safe_account_id,
    save_auth_session,
    save_user_models,
    save_user_profile,
)


MAX_AGENT_PACKAGE_BYTES = 50 * 1024 * 1024
MAX_AGENT_PACKAGE_FILES = 2000
EXCLUDED_PACKAGE_PARTS = {"__pycache__", "workspace", "memory"}
SECRET_PACKAGE_FILENAMES = {".env", "secrets.yaml", "secret.yaml", "secrets.json", "secret.json"}
TEMPLATE_AGENT_ID = "default"
STT_CREDIT_UNIT = Decimal("0.01")
HARDWARE_HUB_URL = "http://127.0.0.1:8790"


class AuthSession:
    def __init__(self):
        self.on_change = None
        self._lock = threading.RLock()
        self._credit_lock = threading.RLock()
        self._credit_server_balance: Decimal | None = None
        self._credit_pending = Decimal("0")
        self._credit_pending_items: list[dict] = []

    def account_id(self) -> str:
        return load_auth_session().get("account_id") or "local"

    def public_session(self) -> dict:
        self._sync_authenticated_session()

        session = load_auth_session()
        payload = {
            "authenticated": bool(session.get("authenticated")),
            "account_id": session.get("account_id") or "local",
            "email": session.get("email", ""),
            "display_name": session.get("display_name", ""),
            "expires_at": session.get("expires_at") or 0,
            "has_token": bool(session.get("token")),
        }
        with self._credit_lock:
            if self._credit_server_balance is not None:
                payload.update({
                    "credits": _credit_payload(self._available_credit_locked()),
                    "server_credits": _credit_payload(self._credit_server_balance),
                    "pending_credits": _credit_payload(self._credit_pending),
                    "pending_count": len(self._credit_pending_items),
                })
        return payload

    def login(self, account: str, password: str) -> dict:
        account = str(account or "").strip()
        password = str(password or "")
        if not account or not password:
            raise BadRequest("请输入账号和密码")
        self.settle_credit()
        data = self._request("login_path", {"account": account, "password": password})
        return self._save_server_session(data, account, reset_credit=True)

    def register(self, email: str, password: str) -> dict:
        email = str(email or "").strip()
        password = str(password or "")
        if not email or not password:
            raise BadRequest("请输入邮箱和密码")
        self.settle_credit()
        data = self._request("register_path", {"email": email, "password": password})
        return self._save_server_session(data, email, reset_credit=True)

    def refresh(self) -> dict:
        with self._lock:
            current = load_auth_session()
            refresh_token = current.get("refresh_token", "")
            if not refresh_token:
                raise BadRequest("未登录")
            data = self._request("refresh_path", {"refresh_token": refresh_token})
            merged = {
                "user": {
                    "id": current.get("account_id", "local"),
                    "email": current.get("email", ""),
                    "nickname": current.get("display_name", ""),
                },
                "access_token": data.get("access_token") or data.get("token") or current.get("token", ""),
                "refresh_token": data.get("refresh_token") or refresh_token,
                "expires_in": data.get("expires_in") or 7200,
                "message": data.get("message") or "令牌已刷新",
            }
            return self._save_server_session(merged, current.get("email", ""))

    def logout(self) -> dict:
        self.settle_credit()
        with self._lock:
            self._clear_session()
            return {"ok": True, "session": self.public_session()}

    def ensure_token(self) -> str:
        session = load_auth_session()
        token = session.get("token", "")
        if not token:
            return ""
        expires_at = int(session.get("expires_at") or 0)
        expires_in = expires_at - int(time.time()) if expires_at else None
        if expires_in is not None and expires_in < 60 and session.get("refresh_token"):
            try:
                self.refresh()
            except BridgeHTTPError as exc:
                if exc.status in {401, 403}:
                    self._clear_session()
                    return ""
                raise
            token = load_auth_session().get("token", "")
        elif expires_in is not None and expires_in <= 0:
            self._clear_session()
            return ""
        return token

    def auth_headers(self) -> dict:
        token = self.ensure_token()
        return {"Authorization": f"Bearer {token}"} if token else {}

    def credit_balance(self) -> dict:
        self._require_login()
        with self._credit_lock:
            data = self._server_request("GET", "credit_balance_path")
            self._credit_server_balance = _decimal_value(data.get("credits") or 0)
            return self._credit_payload()

    def pay_create(self, amount) -> dict:
        self._require_login()
        amount_value = _positive_decimal(amount, "充值金额必须是正数")
        if amount_value != amount_value.to_integral_value():
            raise BadRequest("充值金额必须是整数")
        data = self._server_request("POST", "pay_create_path", {
            "amount": _number_payload(amount_value),
        })
        if isinstance(data, dict) and data.get("order_no"):
            data["recharge_url"] = self._recharge_url(
                str(data.get("order_no")),
                data.get("amount") or _number_payload(amount_value),
            )
        return data

    def pay_status(self, order_no: str) -> dict:
        self._require_login()
        order_no = str(order_no or "").strip()
        if not order_no:
            raise BadRequest("order_no 不能为空")
        data = self._server_request("GET", "pay_status_path", {
            "order_no": order_no,
        })
        if str(data.get("status") or "").lower() == "paid":
            with self._credit_lock:
                self._credit_server_balance = None
        return data

    def credit_consume(self, amount, action: str, description: str = "") -> dict:
        self._require_login()
        amount_value = _positive_decimal(amount, "积分数量必须是正数")
        action = str(action or "").strip()[:32]
        if not action:
            raise BadRequest("action 不能为空")
        payload = {
            "amount": _number_payload(amount_value),
            "action": action,
            "description": str(description or "").strip(),
        }
        data = self._server_request("POST", "credit_consume_path", payload)
        with self._credit_lock:
            self._credit_server_balance = _decimal_value(data.get("credits") or 0)
            balance_payload = self._credit_payload()
        return {
            **balance_payload,
            "transaction": data.get("transaction") if isinstance(data.get("transaction"), dict) else None,
        }

    def ensure_credit_available(self, amount) -> dict:
        self._require_login()
        amount_value = _positive_decimal(amount, "credit_unit 必须是正数")
        with self._credit_lock:
            self._ensure_credit_balance_locked()
            available = self._available_credit_locked()
            if amount_value > available:
                raise BridgeHTTPError(402, f"积分不足，当前可用余额 {_credit_payload(available)}")
            return self._credit_payload()

    def consume_conversation(
        self,
        unit_amount,
        user_text: str,
        assistant_text: str,
        action: str = "chat",
        description: str = "对话消耗",
    ) -> dict:
        unit = _positive_decimal(unit_amount, "credit_unit 必须是正数")
        units = _conversation_billing_units(user_text, assistant_text)
        amount = unit * units
        billing = {
            "unit_amount": _number_payload(unit),
            "units": units,
            "amount": _number_payload(amount),
            "user_chars": _billable_chars(user_text),
            "assistant_chars": _billable_chars(assistant_text),
        }
        return self.queue_credit(amount, action, description, billing)

    def queue_credit(self, amount, action: str, description: str = "", billing: dict | None = None) -> dict:
        self._require_login()
        amount_value = _positive_decimal(amount, "积分数量必须是正数")
        action = str(action or "").strip()[:32]
        if not action:
            raise BadRequest("action 不能为空")

        with self._credit_lock:
            self._ensure_credit_balance_locked()
            projected = self._credit_pending + amount_value
            if self._credit_server_balance is not None and projected > self._credit_server_balance:
                available = self._credit_server_balance - self._credit_pending
                raise BridgeHTTPError(402, f"积分不足，当前可用余额 {_credit_payload(available)}")

            item = {
                "amount": _credit_payload(amount_value),
                "action": action,
                "description": str(description or "").strip(),
                "billing": billing or {},
            }
            self._credit_pending += amount_value
            self._credit_pending_items.append(item)
            payload = self._credit_payload()
            payload["queued"] = item
            return payload

    def settle_credit(self) -> dict:
        with self._credit_lock:
            amount = self._credit_pending
            items = list(self._credit_pending_items)
            if amount <= 0:
                payload = self._credit_payload() if self._credit_server_balance is not None else {"ok": True}
                return {**payload, "settled": False, "transaction": None}
            self._require_login()
            data = self.credit_consume(
                amount,
                "conversation",
                f"合并对话消耗（{len(items)}轮）",
            )
            self._credit_pending = Decimal("0")
            self._credit_pending_items = []
            payload = self._credit_payload()
        return {
            **payload,
            "settled": True,
            "transaction": data.get("transaction"),
            "items": items,
        }

    def _ensure_credit_balance_locked(self) -> None:
        if self._credit_server_balance is None:
            data = self._server_request("GET", "credit_balance_path")
            self._credit_server_balance = _decimal_value(data.get("credits") or 0)

    def _credit_payload(self) -> dict:
        server = self._credit_server_balance if self._credit_server_balance is not None else Decimal("0")
        available = self._available_credit_locked()
        return {
            "ok": True,
            "credits": _credit_payload(available),
            "server_credits": _credit_payload(server),
            "pending_credits": _credit_payload(self._credit_pending),
            "pending_count": len(self._credit_pending_items),
        }

    def _available_credit_locked(self) -> Decimal:
        server = self._credit_server_balance if self._credit_server_balance is not None else Decimal("0")
        available = server - self._credit_pending
        return available if available > 0 else Decimal("0")

    def _reset_credit_state(self, server_balance=None) -> None:
        with self._credit_lock:
            self._credit_server_balance = _decimal_value(server_balance) if server_balance is not None else None
            self._credit_pending = Decimal("0")
            self._credit_pending_items = []

    def _set_credit_server_balance(self, server_balance) -> None:
        with self._credit_lock:
            self._credit_server_balance = _decimal_value(server_balance)

    def _clear_session(self) -> None:
        clear_auth_session()
        self._reset_credit_state()
        self._notify_change()

    def _sync_authenticated_session(self) -> None:
        session = load_auth_session()
        if not session.get("authenticated") or not session.get("token"):
            return
        if not self.ensure_token():
            return

        try:
            data = self._server_request("GET", "me_path")
        except BridgeHTTPError as exc:
            if exc.status in {401, 403}:
                self._clear_session()
            return
        except Exception:
            return

        user = data.get("user") if isinstance(data, dict) and isinstance(data.get("user"), dict) else {}
        if not user:
            return
        self._sync_local_session_user(user)
        if "credits" in user:
            self._set_credit_server_balance(user.get("credits"))

    def _sync_local_session_user(self, user: dict) -> dict:
        current = load_auth_session()
        email = str(user.get("email") or user.get("username") or current.get("email") or "").strip()
        display_name = str(
            user.get("nickname")
            or user.get("display_name")
            or user.get("username")
            or current.get("display_name")
            or email
            or ""
        ).strip()
        account_id = safe_account_id(user.get("id") or user.get("uid") or current.get("account_id") or email or "local")
        return save_auth_session({
            "account_id": account_id,
            "email": email,
            "display_name": display_name,
            "token": current.get("token", ""),
            "refresh_token": current.get("refresh_token", ""),
            "expires_at": current.get("expires_at") or 0,
            "authenticated": True,
        })

    def _save_server_session(self, data: dict, account: str = "", reset_credit: bool = False) -> dict:
        if not isinstance(data, dict):
            raise BadRequest("认证服务器响应无效")
        if data.get("ok") is False:
            raise BadRequest(str(data.get("error") or "认证失败"))

        user = data.get("user") if isinstance(data.get("user"), dict) else {}
        access_token = str(data.get("access_token") or data.get("token") or "").strip()
        refresh_token = str(data.get("refresh_token") or "").strip()
        if not access_token:
            raise BadRequest("认证服务器未返回 access_token")

        email = str(user.get("email") or user.get("username") or account or "").strip()
        display_name = str(user.get("nickname") or user.get("username") or email or "").strip()
        account_id = safe_account_id(user.get("id") or user.get("uid") or email or account or "local")
        try:
            expires_in = int(data.get("expires_in") or 7200)
        except (TypeError, ValueError):
            expires_in = 7200

        session = save_auth_session({
            "account_id": account_id,
            "email": email,
            "display_name": display_name,
            "access_token": access_token,
            "refresh_token": refresh_token,
            "expires_at": int(time.time()) + max(60, expires_in),
            "authenticated": True,
        })

        profile = load_user_profile(account_id)
        if display_name and not profile.get("display_name"):
            save_user_profile({
                "display_name": display_name,
                "notes": profile.get("notes", ""),
                "preferences": profile.get("preferences", {}),
            }, account_id=account_id)

        if reset_credit:
            server_balance = user.get("credits") if "credits" in user else None
            self._reset_credit_state(server_balance)
        self._notify_change()
        return {
            "ok": True,
            "session": self.public_session(),
            "user": self._public_user(user, session),
            "message": str(data.get("message") or "认证成功"),
        }

    def _public_user(self, user: dict, session: dict) -> dict:
        allowed = {"id", "uid", "username", "email", "nickname", "avatar", "bio", "credits", "is_admin", "status"}
        public = {key: user.get(key) for key in allowed if key in user}
        public.setdefault("id", session.get("account_id"))
        public.setdefault("email", session.get("email", ""))
        public.setdefault("nickname", session.get("display_name", ""))
        return public

    def _request(self, path_key: str, payload: dict) -> dict:
        return self._server_request("POST", path_key, payload)

    def _recharge_url(self, order_no: str, amount) -> str:
        config = auth_config()
        server_url = config.get("server_url", "")
        if not server_url:
            raise BadRequest("未配置认证服务器")
        path = config.get("pay_recharge_path", "") or "/recharge.php"
        url = f"{server_url}{path if path.startswith('/') else '/' + path}"
        query = urlencode({
            "order_no": order_no,
            "amount": amount,
        })
        return f"{url}{'&' if '?' in url else '?'}{query}"

    def _server_request(
        self,
        method: str,
        path_key: str,
        payload: dict | None = None,
        require_dict: bool = True,
    ):
        config = auth_config()
        server_url = config.get("server_url", "")
        if not server_url:
            raise BadRequest("未配置认证服务器")

        path = config.get(path_key, "")
        if path_key == "refresh_path":
            path = "/api/auth/refresh"
        url = f"{server_url}{path if path.startswith('/') else '/' + path}"
        if method == "GET" and payload:
            query = urlencode({
                key: value
                for key, value in payload.items()
                if value is not None
            })
            if query:
                url = f"{url}{'&' if '?' in url else '?'}{query}"
        body = json.dumps(payload or {}, ensure_ascii=False).encode("utf-8") if method != "GET" else None
        headers = {"Content-Type": "application/json"}
        if path_key not in {"login_path", "register_path", "refresh_path"}:
            headers.update(self.auth_headers())
        request = urllib.request.Request(
            url,
            data=body,
            headers=headers,
            method=method,
        )

        try:
            with urllib.request.urlopen(request, timeout=20) as response:
                raw = response.read().decode("utf-8")
        except urllib.error.HTTPError as exc:
            raw = exc.read().decode("utf-8", errors="replace")
            try:
                data = _parse_json_text(raw)
                message = str(data.get("error") or raw or f"服务器错误: {exc.code}")
            except BadRequest:
                message = raw or f"服务器错误: {exc.code}"
            raise BridgeHTTPError(exc.code, message) from exc
        except urllib.error.URLError as exc:
            raise BadRequest(f"无法连接服务器: {exc.reason}") from exc

        data = _parse_json_text(raw) if require_dict else _parse_json_value(raw)
        if isinstance(data, dict) and data.get("ok") is False:
            raise BadRequest(str(data.get("error") or "认证失败"))
        return data

    def _require_login(self) -> None:
        if not load_auth_session().get("authenticated") or not self.ensure_token():
            raise BridgeHTTPError(401, "请先登录")

    def _notify_change(self) -> None:
        callback = self.on_change
        if callable(callback):
            callback()


class AgentRegistry:
    def __init__(self, agents_dir: Path, auth_session: AuthSession):
        self.agents_dir = Path(agents_dir)
        self.auth_session = auth_session
        self._agents: dict[tuple[str, str], Agent] = {}
        self._lock = threading.RLock()

    def account_id(self) -> str:
        return self.auth_session.account_id()

    def list_agents(self) -> list[dict]:
        if not self.agents_dir.exists():
            return []

        selected = self.selected_agent()
        result = []
        for path in sorted(self.agents_dir.iterdir()):
            config_path = path / "config.yaml"
            if path.name == TEMPLATE_AGENT_ID:
                continue
            if not path.is_dir() or not config_path.exists():
                continue
            config = AgentConfig.from_yaml(config_path)
            result.append({
                "id": path.name,
                "name": path.name,
                "display_name": config.name or path.name,
                "description": config.description,
                **_agent_state_payload(),
                "selected": path.name == selected,
                "voice": {
                    "enabled": config.voice.enabled,
                    "language": config.voice.language,
                    "stt_provider": config.voice.stt_provider,
                    "stt_engine_model_type": config.voice.stt_engine_model_type,
                    "tts_voice": config.voice.tts_voice,
                    "tts_rate": config.voice.tts_rate,
                },
            })
        return result

    def selected_agent(self) -> str:
        profile = load_user_profile(self.account_id())
        prefs = profile.get("preferences") or {}
        selected = self._safe_agent_name(prefs.get("active_agent", ""))
        if selected and self._agent_dir(selected).exists():
            return selected

        for path in sorted(self.agents_dir.iterdir() if self.agents_dir.exists() else []):
            if path.name == TEMPLATE_AGENT_ID:
                continue
            if path.is_dir() and (path / "config.yaml").exists():
                return path.name
        return ""

    def select_agent(self, name: str) -> dict:
        agent_dir = self._agent_dir(name)
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能选择")
        if not (agent_dir / "config.yaml").exists():
            raise FileNotFoundError(f"智能体不存在: {name}")

        profile = load_user_profile(self.account_id())
        prefs = profile.get("preferences") or {}
        prefs["active_agent"] = agent_dir.name
        save_user_profile({
            "display_name": profile.get("display_name", ""),
            "notes": profile.get("notes", ""),
            "preferences": prefs,
        }, account_id=self.account_id())
        return self.agent_payload(agent_dir.name, selected=True)

    def get(self, name: str) -> Agent:
        agent_dir = self._agent_dir(name)
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能直接使用")
        if not agent_dir.exists():
            raise FileNotFoundError(f"智能体不存在: {name}")

        with self._lock:
            agent_id = agent_dir.name
            key = (self.account_id(), agent_id)
            if key not in self._agents:
                self._agents[key] = Agent(str(agent_dir), account_id=key[0])
            return self._agents[key]

    def agent_payload(self, name: str, selected: bool | None = None) -> dict:
        agent_dir = self._agent_dir(name)
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能直接使用")
        config_path = agent_dir / "config.yaml"
        if not config_path.exists():
            raise FileNotFoundError(f"智能体不存在: {name}")

        config = AgentConfig.from_yaml(config_path)
        identity_path = agent_dir / "IDENTITY.md"
        if selected is None:
            selected = agent_dir.name == self.selected_agent()

        return {
            "id": agent_dir.name,
            "name": config.name or agent_dir.name,
            "description": config.description,
            "owner_name": config.owner_name,
            "relationship": config.relationship,
            "max_response_tokens": config.max_response_tokens,
            "identity": identity_path.read_text(encoding="utf-8") if identity_path.exists() else "",
            **_agent_state_payload(),
            "selected": selected,
            "voice": {
                "enabled": config.voice.enabled,
                "language": config.voice.language,
                "stt_provider": config.voice.stt_provider,
                "stt_engine_model_type": config.voice.stt_engine_model_type,
                "tts_voice": config.voice.tts_voice,
                "tts_rate": config.voice.tts_rate,
            },
        }

    def history_index(self, name: str) -> dict:
        agent_dir = self._agent_dir(name)
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能直接使用")
        if not (agent_dir / "config.yaml").exists():
            raise FileNotFoundError(f"智能体不存在: {name}")

        manifest = self._history_manifest(agent_dir)
        modules = manifest.get("modules") if isinstance(manifest.get("modules"), dict) else {}
        days = []

        memory_dir = migrate_legacy_agent_memory(agent_dir.name, agent_dir / "memory", account_id=self.account_id())
        episodic_dir = memory_dir / "episodic"
        for path in sorted(episodic_dir.glob("*.json")):
            date = path.stem
            if not _valid_history_date(date):
                continue
            entries = _read_json_file(path, [])
            if not isinstance(entries, list):
                entries = []

            module_id = f"episodic_{date}"
            meta = modules.get(module_id) if isinstance(modules.get(module_id), dict) else {}
            timestamps = [str(item.get("timestamp") or "") for item in entries if isinstance(item, dict)]
            days.append({
                "date": date,
                "month": date[:7],
                "day": date[-2:],
                "module_id": module_id,
                "entry_count": len(entries),
                "summary": str(meta.get("summary") or ""),
                "created_at": str(meta.get("created_at") or (timestamps[0] if timestamps else "")),
                "updated_at": str(meta.get("updated_at") or (timestamps[-1] if timestamps else "")),
            })

        months = []
        for day in days:
            if not months or months[-1]["month"] != day["month"]:
                months.append({"month": day["month"], "days": []})
            months[-1]["days"].append(day)

        return {
            "months": months,
            "days": days,
            "selected": "",
        }

    def history_day(self, name: str, date: str) -> dict:
        agent_dir = self._agent_dir(name)
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能直接使用")
        if not (agent_dir / "config.yaml").exists():
            raise FileNotFoundError(f"智能体不存在: {name}")

        date = _safe_history_date(date)
        memory_dir = migrate_legacy_agent_memory(agent_dir.name, agent_dir / "memory", account_id=self.account_id())
        path = memory_dir / "episodic" / f"{date}.json"
        if not path.exists():
            raise FileNotFoundError(f"历史不存在: {date}")

        raw_entries = _read_json_file(path, [])
        if not isinstance(raw_entries, list):
            raw_entries = []

        entries = []
        for item in raw_entries:
            if not isinstance(item, dict):
                continue
            role = str(item.get("role") or "")
            if role not in {"user", "assistant"}:
                continue
            entries.append({
                "role": role,
                "content": str(item.get("content") or ""),
                "timestamp": str(item.get("timestamp") or ""),
            })

        module_id = f"episodic_{date}"
        manifest = self._history_manifest(agent_dir)
        modules = manifest.get("modules") if isinstance(manifest.get("modules"), dict) else {}
        meta = modules.get(module_id) if isinstance(modules.get(module_id), dict) else {}

        return {
            "date": date,
            "module_id": module_id,
            "summary": str(meta.get("summary") or ""),
            "entry_count": len(entries),
            "entries": entries,
        }

    def update_agent(self, name: str, data: dict) -> dict:
        agent_dir = self._agent_dir(name)
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能编辑")
        config_path = agent_dir / "config.yaml"
        if not config_path.exists():
            raise FileNotFoundError(f"智能体不存在: {name}")

        # Close the live instance before writing the edited config. Agent.close()
        # persists its in-memory config, so doing this after writing would restore
        # the stale pre-edit settings.
        self.close_agent(agent_dir.name)
        config = AgentConfig.from_yaml(config_path)
        if "name" in data:
            config.name = str(data.get("name") or agent_dir.name).strip() or agent_dir.name
        if "description" in data:
            config.description = str(data.get("description") or "").strip()
        if "owner_name" in data:
            config.owner_name = str(data.get("owner_name") or "").strip() or "公子"
        if "relationship" in data:
            config.relationship = str(data.get("relationship") or "").strip()
        if "max_response_tokens" in data:
            config.max_response_tokens = max(64, min(32768, int(data.get("max_response_tokens") or 2048)))
        voice = data.get("voice") if isinstance(data.get("voice"), dict) else {}
        if "tts_voice" in data:
            config.voice.tts_voice = str(data.get("tts_voice") or config.voice.tts_voice).strip()
        if "tts_voice" in voice:
            config.voice.tts_voice = str(voice.get("tts_voice") or config.voice.tts_voice).strip()
        if "tts_rate" in data:
            config.voice.tts_rate = _normalize_tts_rate(data.get("tts_rate"))
        if "tts_rate" in voice:
            config.voice.tts_rate = _normalize_tts_rate(voice.get("tts_rate"))

        config.to_yaml(config_path)

        if "identity" in data:
            (agent_dir / "IDENTITY.md").write_text(str(data.get("identity") or ""), encoding="utf-8")

        return self.agent_payload(agent_dir.name)

    def apply_model(self, name: str, model: dict) -> dict:
        agent_dir = self._agent_dir(name)
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能直接使用")
        config_path = agent_dir / "config.yaml"
        if not config_path.exists():
            raise FileNotFoundError(f"智能体不存在: {name}")

        return self.agent_payload(agent_dir.name)

    def delete_agent(self, name: str, force: bool = False) -> dict:
        agent_dir = self._agent_dir(name)
        if not agent_dir.exists():
            raise FileNotFoundError(f"智能体不存在: {name}")
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能删除")

        agents = [item["id"] for item in self.list_agents()]
        if len(agents) <= 1:
            raise BadRequest("至少需要保留一个智能体")

        was_selected = agent_dir.name == self.selected_agent()
        if was_selected and not force:
            raise BadRequest("删除当前智能体需要确认")

        self.close_agent(agent_dir.name)
        shutil.rmtree(agent_dir)

        selected = self.selected_agent()
        if was_selected or not selected or selected == agent_dir.name:
            remaining = [item["id"] for item in self.list_agents()]
            if remaining:
                selected = remaining[0]
                self.select_agent(selected)

        return {"deleted": agent_dir.name, "selected": self.selected_agent()}

    def export_agent(self, name: str) -> tuple[bytes, str]:
        agent_dir = self._agent_dir(name)
        if not (agent_dir / "config.yaml").exists():
            raise FileNotFoundError(f"智能体不存在: {name}")
        if agent_dir.name == TEMPLATE_AGENT_ID:
            raise BadRequest("default 是模板，不能导出")

        buffer = io.BytesIO()
        written = 0
        with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as zf:
            manifest = {
                "type": "coco-agent",
                "version": 1,
                "id": agent_dir.name,
                "name": AgentConfig.from_yaml(agent_dir / "config.yaml").name or agent_dir.name,
            }
            zf.writestr("coco-agent.json", json.dumps(manifest, ensure_ascii=False, indent=2))

            config = AgentConfig.from_yaml(agent_dir / "config.yaml")
            zf.writestr("config.yaml", yaml.safe_dump(_agent_config_dict(config), allow_unicode=True, sort_keys=False))

            for path in sorted(agent_dir.rglob("*")):
                if not path.is_file():
                    continue
                rel = path.relative_to(agent_dir)
                rel_text = rel.as_posix()
                if rel_text == "config.yaml" or not _exportable_agent_path(rel):
                    continue
                written += path.stat().st_size
                if written > MAX_AGENT_PACKAGE_BYTES:
                    raise BadRequest("智能体包过大，无法导出")
                zf.write(path, rel_text)

        return buffer.getvalue(), f"{agent_dir.name}.coco-agent.zip"

    def create_agent(self, data: dict) -> dict:
        display_name = str(data.get("name") or "").strip()
        if not display_name:
            raise BadRequest("请填写智能体名称")

        template_dir = self._agent_dir(TEMPLATE_AGENT_ID)
        if not (template_dir / "config.yaml").exists():
            raise FileNotFoundError("default 模板不存在")

        agent_id = self._unique_agent_name(_agent_name_to_id(display_name))
        target_dir = self._agent_dir(agent_id)
        shutil.copytree(template_dir, target_dir, ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))

        config_path = target_dir / "config.yaml"
        config = AgentConfig.from_yaml(config_path)
        config.name = display_name
        if "description" in data:
            config.description = str(data.get("description") or "").strip()
        if "owner_name" in data:
            config.owner_name = str(data.get("owner_name") or "").strip() or "公子"
        if "relationship" in data:
            config.relationship = str(data.get("relationship") or "").strip()
        config.to_yaml(config_path)

        selected = self.select_agent(agent_id)
        return selected

    def import_agent(self, body: bytes) -> dict:
        if not body:
            raise BadRequest("zip body is empty")
        if len(body) > MAX_AGENT_PACKAGE_BYTES:
            raise BadRequest("智能体包过大")

        try:
            zf_context = zipfile.ZipFile(io.BytesIO(body))
        except zipfile.BadZipFile as exc:
            raise BadRequest("智能体包不是有效 zip") from exc

        with zf_context as zf:
            infos = [info for info in zf.infolist() if not info.is_dir()]
            if len(infos) > MAX_AGENT_PACKAGE_FILES:
                raise BadRequest("智能体包文件数量过多")
            if sum(info.file_size for info in infos) > MAX_AGENT_PACKAGE_BYTES:
                raise BadRequest("智能体包解压后过大")

            for info in infos:
                _safe_zip_name(info)

            root = _zip_agent_root(zf, infos)
            with tempfile.TemporaryDirectory(prefix="coco-agent-import-") as tmp:
                tmp_root = Path(tmp)
                for info in infos:
                    rel = _safe_zip_name(info)
                    if root:
                        root_prefix = f"{root}/"
                        if not rel.startswith(root_prefix):
                            continue
                        rel = rel[len(root_prefix):]
                    if not rel or rel == "coco-agent.json":
                        continue
                    if not _exportable_agent_path(Path(rel)):
                        raise BadRequest(f"智能体包包含不允许的文件: {rel}")

                    target = (tmp_root / rel).resolve()
                    if tmp_root.resolve() not in target.parents and target != tmp_root.resolve():
                        raise BadRequest(f"非法路径: {info.filename}")
                    target.parent.mkdir(parents=True, exist_ok=True)
                    with zf.open(info) as src, open(target, "wb") as dst:
                        shutil.copyfileobj(src, dst)

                config_path = tmp_root / "config.yaml"
                identity_path = tmp_root / "IDENTITY.md"
                if not config_path.exists() or not identity_path.exists():
                    raise BadRequest("智能体包必须包含 config.yaml 和 IDENTITY.md")

                config = AgentConfig.from_yaml(config_path)
                agent_id = self._unique_agent_name(config.name or "agent")
                config.name = config.name or agent_id
                config.to_yaml(config_path)

                (tmp_root / "public_memory").mkdir(parents=True, exist_ok=True)

                target_dir = self._agent_dir(agent_id)
                shutil.copytree(tmp_root, target_dir)

        selected = self.select_agent(agent_id)
        return selected

    def close_agent(self, name: str):
        with self._lock:
            closing = [
                self._agents.pop(key)
                for key in list(self._agents)
                if key[1] == name
            ]
        for agent in closing:
            agent.close()

    def close_all(self):
        with self._lock:
            agents = list(self._agents.values())
            self._agents.clear()

        for agent in agents:
            agent.close()

    def _safe_agent_name(self, name: str) -> str:
        name = str(name or "").strip()
        return "".join(ch for ch in name if ch.isalnum() or ch in ("-", "_"))[:80]

    def _unique_agent_name(self, name: str) -> str:
        base = self._safe_agent_name(name) or "agent"
        candidate = base
        i = 2
        while self._agent_dir(candidate).exists():
            candidate = f"{base}-{i}"
            i += 1
        return candidate

    def _agent_dir(self, name: str) -> Path:
        safe = self._safe_agent_name(name)
        if not safe:
            raise BadRequest("智能体名称无效")

        root = self.agents_dir.resolve()
        target = (root / safe).resolve()
        if root != target and root not in target.parents:
            raise BadRequest("智能体路径无效")
        return target

    def _history_manifest(self, agent_dir: Path) -> dict:
        memory_dir = migrate_legacy_agent_memory(agent_dir.name, agent_dir / "memory", account_id=self.account_id())
        manifest = _read_json_file(memory_dir / "manifest.json", {})
        return manifest if isinstance(manifest, dict) else {}


class ModelRegistry:
    def __init__(self, agent_registry: AgentRegistry, auth_session: AuthSession):
        self.agent_registry = agent_registry
        self.auth_session = auth_session
        self._lock = threading.RLock()

    def list_models(self) -> dict:
        with self._lock:
            cloud_models = self._cloud_models()
            config = self._load_config(cloud_models)
            if self._refresh_active_cloud_model(config, cloud_models):
                self._save_config(config)
            return self._payload(config, cloud_models)

    def upsert_model(self, data: dict, model_id: str = "") -> dict:
        with self._lock:
            cloud_models = self._cloud_models()
            config = self._load_config(cloud_models)
            models = config["models"]
            model_id = self._safe_model_id(model_id or data.get("id", ""))
            index = next((i for i, item in enumerate(models) if item["id"] == model_id), -1)

            existing = models[index] if index >= 0 else {}
            if existing.get("can_edit") is False:
                raise BadRequest("云端大模型不能编辑")
            model = self._normalize_model(data, existing)
            if index < 0:
                model["id"] = self._unique_model_id(model["id"], models)
                models.append(model)
            else:
                model["id"] = existing["id"]
                models[index] = model

            if not config.get("active_model"):
                config["active_model"] = model["id"]
            self._save_config(config)
            return {
                "ok": True,
                "model": self._public_model(model, model["id"] == config["active_model"]),
                **self._payload(config, cloud_models),
            }

    def select_model(self, model_id: str) -> dict:
        with self._lock:
            cloud_models = self._cloud_models()
            config = self._load_config(cloud_models)
            model = self._find_model(config, model_id, cloud_models)
            if model.get("source") == "cloud":
                config["active_model_config"] = model
            else:
                config.pop("active_model_config", None)
            config["active_model"] = model["id"]
            self._save_config(config)

        agent_payload = None
        selected_agent = self.agent_registry.selected_agent()
        if selected_agent:
            agent_payload = self.agent_registry.apply_model(selected_agent, model)

        return {
            "ok": True,
            "model": self._public_model(model, True),
            "selected": model["id"],
            "agent": agent_payload,
            **self._payload(config, cloud_models),
        }

    def delete_model(self, model_id: str) -> dict:
        with self._lock:
            cloud_models = self._cloud_models()
            config = self._load_config(cloud_models)
            models = config["models"]
            model = self._find_model(config, model_id, cloud_models)
            if model.get("can_delete") is False:
                raise BadRequest("云端大模型不能删除")

            models[:] = [item for item in models if item["id"] != model["id"]]
            selected_model = None
            if config.get("active_model") == model["id"]:
                if cloud_models:
                    selected_model = cloud_models[0]
                    config["active_model"] = selected_model["id"]
                    config["active_model_config"] = selected_model
                elif models:
                    selected_model = models[0]
                    config["active_model"] = selected_model["id"]
                    config.pop("active_model_config", None)
                else:
                    config["active_model"] = ""
                    config.pop("active_model_config", None)
            self._save_config(config)

        agent_payload = None
        if selected_model:
            selected_agent = self.agent_registry.selected_agent()
            if selected_agent:
                agent_payload = self.agent_registry.apply_model(selected_agent, selected_model)

        return {
            "ok": True,
            "deleted": model["id"],
            "selected": config.get("active_model"),
            "agent": agent_payload,
            **self._payload(config, cloud_models),
        }

    def current_model_credit_unit(self) -> Decimal:
        with self._lock:
            cloud_models = self._cloud_models()
            config = self._load_config(cloud_models)
            if self._refresh_active_cloud_model(config, cloud_models):
                self._save_config(config)
            active = config.get("active_model", "")
            active_model_config = config.get("active_model_config")
            candidates = [
                *cloud_models,
                *config.get("models", []),
            ]
            if isinstance(active_model_config, dict) and active_model_config:
                candidates.append(self._normalize_model(active_model_config))

            for model in candidates:
                if model.get("id") != active:
                    continue
                if model.get("source") != "cloud":
                    return Decimal("0")
                return self._model_credit_unit(model)
        return Decimal("0")

    def _load_config(self, cloud_models: list[dict] | None = None) -> dict:
        if cloud_models is None:
            cloud_models = self._cloud_models()

        config = load_user_models(self.agent_registry.account_id())
        models = [self._normalize_model(item) for item in config.get("models", [])]
        raw_active_model_config = config.get("active_model_config")
        active_model_config = (
            self._normalize_model(raw_active_model_config)
            if isinstance(raw_active_model_config, dict) and raw_active_model_config
            else {}
        )
        active = self._safe_model_id(config.get("active_model", ""))
        active_is_known = (
            any(item["id"] == active for item in models)
            or any(item["id"] == active for item in cloud_models)
            or active_model_config.get("id") == active
        )
        if not active or not active_is_known:
            if cloud_models:
                active_model_config = cloud_models[0]
                active = active_model_config["id"]
            elif models:
                active = models[0]["id"]
                active_model_config = {}
            else:
                active = ""
                active_model_config = {}
        else:
            active_cloud_model = next(
                (item for item in cloud_models if item["id"] == active),
                None,
            )
            if active_cloud_model is not None:
                active_model_config = active_cloud_model

        config["active_model"] = active
        config["active_model_config"] = active_model_config
        config["models"] = models
        self._save_config(config)
        return config

    def _save_config(self, config: dict) -> dict:
        return save_user_models({
            "active_model": config.get("active_model", ""),
            "active_model_config": config.get("active_model_config", {}),
            "models": config.get("models", []),
        }, account_id=self.agent_registry.account_id())

    def _payload(self, config: dict, cloud_models: list[dict] | None = None) -> dict:
        selected = config.get("active_model", "")
        models = [*(cloud_models if cloud_models is not None else self._cloud_models()), *config.get("models", [])]
        return {
            "models": [self._public_model(item, item["id"] == selected) for item in models],
            "selected": selected,
        }

    def _refresh_active_cloud_model(self, config: dict, cloud_models: list[dict]) -> bool:
        active = config.get("active_model", "")
        if not active:
            return False
        active_model_config = config.get("active_model_config")
        active_is_cloud = any(item["id"] == active for item in cloud_models)
        if not active_is_cloud and isinstance(active_model_config, dict):
            active_is_cloud = (
                str(active_model_config.get("source") or "").strip() == "cloud"
                and str(active_model_config.get("id") or "").strip() == active
            )
        if not active_is_cloud:
            return False
        model = next((item for item in cloud_models if item["id"] == active), None)
        if model:
            if config.get("active_model_config") == model:
                return False
            config["active_model_config"] = model
            return True
        if cloud_models:
            model = cloud_models[0]
            if config.get("active_model") == model["id"] and config.get("active_model_config") == model:
                return False
            config["active_model"] = model["id"]
            config["active_model_config"] = model
            return True
        local_models = config.get("models", [])
        next_active = local_models[0]["id"] if local_models else ""
        if config.get("active_model") == next_active and not config.get("active_model_config"):
            return False
        config["active_model"] = next_active
        config.pop("active_model_config", None)
        return True

    def _cloud_models(self) -> list[dict]:
        data = self._cloud_model_response()
        if data is None:
            return []

        items = self._cloud_model_items(data)
        models = []
        for item in items:
            if not isinstance(item, dict):
                continue
            model = self._normalize_model({
                **item,
                "source": "cloud",
                "can_edit": False,
                "can_delete": False,
            })
            if not model["id"]:
                continue
            models.append(model)
        return models

    def _cloud_model_response(self):
        for path_key, clear_auth_errors in (
            ("cloud_models_path", True),
            ("legacy_cloud_models_path", False),
        ):
            try:
                return self.auth_session._server_request(
                    "GET",
                    path_key,
                    require_dict=False,
                )
            except BridgeHTTPError as exc:
                if path_key == "cloud_models_path" and exc.status == 404:
                    continue
                if clear_auth_errors and exc.status in {401, 403}:
                    self.auth_session._clear_session()
                return None
            except Exception:
                return None
        return None

    def _cloud_model_items(self, data) -> list:
        if isinstance(data, list):
            return data
        if not isinstance(data, dict):
            return []
        for key in ("models", "items", "list", "rows"):
            value = data.get(key)
            if isinstance(value, list):
                return value
        nested = data.get("data")
        if isinstance(nested, list):
            return nested
        if isinstance(nested, dict):
            for key in ("models", "items", "list", "rows"):
                value = nested.get(key)
                if isinstance(value, list):
                    return value
        return []

    def _normalize_model(self, data: dict, existing: dict | None = None) -> dict:
        existing = existing or {}
        name = str(data.get("name") or data.get("display_name") or existing.get("name") or data.get("model") or existing.get("model") or "model").strip()
        model_name = str(data.get("model") or data.get("model_name") or existing.get("model") or name).strip()
        if "api_base" in data:
            api_base = normalize_api_base(data.get("api_base") or "")
        elif "base_url" in data:
            api_base = normalize_api_base(data.get("base_url") or "")
        elif "api_url" in data:
            api_base = normalize_api_base(data.get("api_url") or "")
        elif "url" in data:
            api_base = normalize_api_base(data.get("url") or "")
        elif "endpoint" in data:
            api_base = normalize_api_base(data.get("endpoint") or "")
        else:
            api_base = normalize_api_base(existing.get("api_base") or "")
        if "api_key" in data:
            api_key = str(data.get("api_key") or "").strip()
        elif "key" in data:
            api_key = str(data.get("key") or "").strip()
        else:
            api_key = str(existing.get("api_key", "")).strip()
        model_id = self._safe_model_id(data.get("id") or existing.get("id") or model_name or name)
        source = str(data.get("source") or existing.get("source") or "local").strip() or "local"
        can_edit = bool(data.get("can_edit", existing.get("can_edit", True)))
        can_delete = bool(data.get("can_delete", existing.get("can_delete", True)))
        credit_unit = self._model_credit_unit(data, existing)
        if source == "cloud":
            can_edit = False
            can_delete = False

        return {
            "id": model_id or "model",
            "name": name or model_name,
            "model": model_name or name,
            "api_base": api_base,
            "api_key": api_key,
            "source": source,
            "can_edit": can_edit,
            "can_delete": can_delete,
            "credit_unit": _credit_payload(credit_unit) if credit_unit > 0 else "",
        }

    def _public_model(self, model: dict, selected: bool = False) -> dict:
        api_key = model.get("api_key") or global_api_key()
        is_cloud = model.get("source") == "cloud"
        return {
            "id": model["id"],
            "name": model.get("name") or model["id"],
            "display_name": model.get("name") or model["id"],
            "model": model.get("model") or model["id"],
            "api_base": model.get("api_base", ""),
            "api_key": "" if is_cloud else api_key,
            "selected": selected,
            "has_api_key": bool(api_key),
            "source": model.get("source") or "local",
            "can_edit": model.get("can_edit") is not False,
            "can_delete": model.get("can_delete") is not False,
            "credit_unit": model.get("credit_unit") or "",
        }

    def _find_model(self, config: dict, model_id: str, cloud_models: list[dict] | None = None) -> dict:
        safe = self._safe_model_id(model_id)
        for model in [*(cloud_models if cloud_models is not None else self._cloud_models()), *config.get("models", [])]:
            if model["id"] == safe:
                return model
        raise FileNotFoundError(f"大模型不存在: {model_id}")

    def _safe_model_id(self, value: str) -> str:
        value = str(value or "").strip()
        safe = "".join(ch for ch in value if ch.isalnum() or ch in ("-", "_", "."))
        return safe[:100]

    def _model_credit_unit(self, data: dict, existing: dict | None = None) -> Decimal:
        existing = existing or {}
        for key in ("credit_unit", "model_credit_unit", "price", "credits"):
            if key in data:
                return _decimal_value(data.get(key))
        return _decimal_value(existing.get("credit_unit") or 0)

    def _unique_model_id(self, preferred: str, models: list[dict]) -> str:
        base = self._safe_model_id(preferred) or "model"
        used = {item["id"] for item in models}
        candidate = base
        i = 2
        while candidate in used:
            candidate = f"{base}-{i}"
            i += 1
        return candidate


def _normalize_tts_rate(value) -> str:
    text = str(value or "+0%").strip().replace("%", "")
    try:
        rate = int(round(float(text)))
    except ValueError:
        rate = 0
    rate = max(-50, min(50, rate))
    return f"{rate:+d}%"


def _agent_name_to_id(name: str) -> str:
    name = str(name or "").strip()
    try:
        from pypinyin import lazy_pinyin
        raw = "".join(lazy_pinyin(name))
    except Exception:
        fallback = {
            "小": "xiao",
            "乐": "le",
            "智": "zhi",
            "助": "zhu",
            "手": "shou",
            "可": "ke",
            "爱": "ai",
            "猫": "mao",
            "主": "zhu",
            "人": "ren",
            "工": "gong",
            "作": "zuo",
            "学": "xue",
            "习": "xi",
            "生": "sheng",
            "活": "huo",
        }
        raw = "".join(fallback.get(ch, ch if ch.isascii() else "") for ch in name)
    safe = "".join(ch.lower() for ch in raw if ch.isalnum() or ch in ("-", "_"))
    return safe or "agent"


def _agent_config_dict(config: AgentConfig) -> dict:
    return {
        "name": config.name,
        "description": config.description,
        "owner_name": config.owner_name,
        "relationship": config.relationship,
        "system_prompt": config.system_prompt,
        "max_response_tokens": config.max_response_tokens,
        "memory": {
            "max_recent_episodes": config.memory.max_recent_episodes,
            "max_working_tokens": config.memory.max_working_tokens,
            "max_context_tokens": config.memory.max_context_tokens,
            "summary_threshold_days": config.memory.summary_threshold_days,
            "summary_retention_days": config.memory.summary_retention_days,
            "semantic_decay_days": config.memory.semantic_decay_days,
            "semantic_forget_days": config.memory.semantic_forget_days,
            "consolidate_interval": config.memory.consolidate_interval,
            "enable_semantic_search": config.memory.enable_semantic_search,
        },
        "voice": {
            "enabled": config.voice.enabled,
            "stt_provider": config.voice.stt_provider,
            "stt_region": config.voice.stt_region,
            "stt_engine_model_type": config.voice.stt_engine_model_type,
            "stt_hotword_list": config.voice.stt_hotword_list,
            "stt_hotword_id": config.voice.stt_hotword_id,
            "stt_customization_id": config.voice.stt_customization_id,
            "stt_filter_dirty": config.voice.stt_filter_dirty,
            "stt_filter_modal": config.voice.stt_filter_modal,
            "stt_filter_punc": config.voice.stt_filter_punc,
            "stt_convert_num_mode": config.voice.stt_convert_num_mode,
            "tts_voice": config.voice.tts_voice,
            "tts_rate": config.voice.tts_rate,
            "language": config.voice.language,
        },
    }


def _agent_state_payload() -> dict:
    return {
        "can_edit": True,
        "can_delete": True,
        "can_export": True,
    }


def _read_json_file(path: Path, default):
    if not path.exists():
        return default
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except (OSError, json.JSONDecodeError):
        return default


def _parse_json_text(text: str) -> dict:
    data = _parse_json_value(text)
    if not isinstance(data, dict):
        raise BadRequest("认证服务器响应无效")
    return data


def _parse_json_value(text: str):
    try:
        data = json.loads(text or "{}")
    except json.JSONDecodeError as exc:
        raise BadRequest(f"认证服务器响应不是 JSON: {text[:200]}") from exc
    return data


def _open_system_url(url: str) -> dict:
    url = str(url or "").strip()
    parsed = urlparse(url)
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise BadRequest("只能打开 http/https 地址")
    if not webbrowser.open(url, new=2):
        raise BadRequest("无法打开系统默认浏览器")
    return {"ok": True}


def _hardware_hub_config_path() -> Path:
    return Path(__file__).resolve().parents[1] / "panel" / "data" / "config.json"


def _hardware_panel_token() -> str:
    config = _read_json_file(_hardware_hub_config_path(), {})
    token = str(config.get("panel_token") or "").strip() if isinstance(config, dict) else ""
    if not token:
        raise BadRequest("Hardware Hub 尚未初始化")
    return token


def _hardware_hub_request(path: str, method: str = "GET", payload: dict | None = None) -> dict:
    url = f"{HARDWARE_HUB_URL}{path}"
    data = None
    headers = {"X-CoCo-Panel-Token": _hardware_panel_token()}
    if payload is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(request, timeout=3) as response:
            text = response.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as exc:
        text = exc.read().decode("utf-8", errors="replace")
        try:
            data = json.loads(text or "{}")
        except json.JSONDecodeError:
            data = {}
        raise BridgeHTTPError(exc.code, str(data.get("error") or text or "Hardware Hub 请求失败")) from exc
    except urllib.error.URLError as exc:
        raise BadRequest(f"无法连接 Hardware Hub：{exc.reason}") from exc

    try:
        data = json.loads(text or "{}")
    except json.JSONDecodeError as exc:
        raise BadRequest("Hardware Hub 返回内容不是 JSON") from exc
    if not isinstance(data, dict):
        raise BadRequest("Hardware Hub 返回内容无效")
    return data


def _number_payload(value):
    try:
        number = Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        number = Decimal("0")
    if number == number.to_integral_value():
        return int(number)
    return float(number)


def _decimal_value(value) -> Decimal:
    try:
        return Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        return Decimal("0")


def _positive_decimal(value, message: str) -> Decimal:
    number = _decimal_value(value)
    if number <= 0:
        raise BadRequest(message)
    return number


def _credit_payload(value) -> str:
    try:
        number = Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        number = Decimal("0")
    return f"{number:.2f}"


def _billable_chars(text: str) -> int:
    return len(str(text or "").strip())


def _conversation_billing_units(user_text: str, assistant_text: str) -> int:
    user_units = max(1, (_billable_chars(user_text) + 99) // 100)
    assistant_units = max(1, (_billable_chars(assistant_text) + 99) // 100)
    return max(user_units, assistant_units)


def _valid_history_date(value: str) -> bool:
    value = str(value or "")
    return (
        len(value) == 10
        and value[4] == "-"
        and value[7] == "-"
        and value[:4].isdigit()
        and value[5:7].isdigit()
        and value[8:10].isdigit()
    )


def _safe_history_date(value: str) -> str:
    value = str(value or "").strip()
    if not _valid_history_date(value):
        raise BadRequest("历史日期无效")
    return value


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


def _exportable_agent_path(rel: Path) -> bool:
    parts = [part.lower() for part in rel.parts]
    if any(part in EXCLUDED_PACKAGE_PARTS for part in parts):
        return False
    if rel.name.lower() in SECRET_PACKAGE_FILENAMES:
        return False
    if rel.suffix.lower() in {".pyc", ".pyo"}:
        return False
    return True


def _safe_zip_name(info: ZipInfo) -> str:
    name = (info.filename or "").replace("\\", "/").strip()
    path = PurePosixPath(name)
    if (
        not name
        or ":" in name
        or path.is_absolute()
        or any(part in ("", ".", "..") for part in path.parts)
    ):
        raise BadRequest(f"非法路径: {info.filename}")
    if info.file_size < 0 or info.file_size > MAX_AGENT_PACKAGE_BYTES:
        raise BadRequest(f"文件过大: {info.filename}")
    return path.as_posix()


def _zip_agent_root(zf: zipfile.ZipFile, infos: list[ZipInfo]) -> str:
    names = {_safe_zip_name(info) for info in infos}
    if "config.yaml" in names and "IDENTITY.md" in names:
        return ""

    top_dirs = {PurePosixPath(name).parts[0] for name in names}
    if len(top_dirs) == 1:
        root = next(iter(top_dirs))
        if f"{root}/config.yaml" in names and f"{root}/IDENTITY.md" in names:
            return root

    if "coco-agent.json" in names:
        try:
            data = json.loads(zf.read("coco-agent.json").decode("utf-8"))
        except Exception as exc:
            raise BadRequest(f"coco-agent.json 无法读取: {exc}") from exc
        root = str(data.get("root") or "").strip().strip("/")
        if root and f"{root}/config.yaml" in names and f"{root}/IDENTITY.md" in names:
            return root

    raise BadRequest("智能体包必须包含 config.yaml 和 IDENTITY.md")


def run_bridge(agents_dir: Path, host: str = "127.0.0.1", port: int = 8765):
    auth_session = AuthSession()
    registry = AgentRegistry(agents_dir, auth_session)
    auth_session.on_change = registry.close_all
    models = ModelRegistry(registry, auth_session)
    bridge_token = load_or_create_bridge_token()

    class Handler(cocoBridgeHandler):
        pass

    Handler.auth_session = auth_session
    Handler.agent_registry = registry
    Handler.model_registry = models
    Handler.bridge_token = bridge_token

    server = ThreadingHTTPServer((host, port), Handler)
    print(f"coco Web Bridge: http://{host}:{port}")
    print("按 Ctrl+C 停止")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n正在停止 Web Bridge...")
    finally:
        try:
            auth_session.settle_credit()
        except Exception as exc:
            print(f"积分结算失败，待下次重试: {exc}")
        registry.close_all()
        server.server_close()


class cocoBridgeHandler(BaseHTTPRequestHandler):
    auth_session: AuthSession
    agent_registry: AgentRegistry
    model_registry: ModelRegistry
    bridge_token: str

    server_version = "cocoWebBridge/0.1"

    def do_OPTIONS(self):
        try:
            self._authorize_preflight()
            self._send_empty(204)
        except Exception as exc:
            self._send_exception(exc)

    def do_GET(self):
        try:
            path, parts, query = self._request_parts()
            self._authorize_request(path)

            if path == "/api/health":
                self._send_json({"ok": True})
                return

            if path == "/api/auth/session":
                self._send_json({"ok": True, "session": self.auth_session.public_session()})
                return

            if path == "/api/credit/balance":
                self._send_json(self.auth_session.credit_balance())
                return

            if path == "/api/pay/status":
                order_no = query.get("order_no", [""])[0]
                self._send_json(self.auth_session.pay_status(order_no))
                return

            if path == "/api/agents":
                self._send_json({
                    "agents": self.agent_registry.list_agents(),
                    "selected": self.agent_registry.selected_agent(),
                })
                return

            if path == "/api/models":
                self._send_json(self.model_registry.list_models())
                return

            if path == "/api/user/profile":
                self._send_json({
                    "profile": load_user_profile(self.agent_registry.account_id()),
                    "session": self.auth_session.public_session(),
                })
                return

            if path == "/api/hardware/state":
                self._send_json(_hardware_hub_request("/api/panel/state"))
                return

            agent_name, tail = self._agent_route(parts)

            if tail == ["config"]:
                self._send_json({"agent": self.agent_registry.agent_payload(agent_name)})
                return

            if tail == ["export"]:
                content, filename = self.agent_registry.export_agent(agent_name)
                self._send_bytes(
                    content,
                    "application/zip",
                    headers={
                        "Content-Disposition": (
                            f'attachment; filename="agent.coco-agent.zip"; '
                            f"filename*=UTF-8''{quote(filename)}"
                        ),
                    },
                )
                return

            if tail == ["history"]:
                self._send_json(self.agent_registry.history_index(agent_name))
                return

            if len(tail) == 2 and tail[0] == "history":
                self._send_json({"history": self.agent_registry.history_day(agent_name, tail[1])})
                return

            agent = self.agent_registry.get(agent_name)

            if tail == ["stats"]:
                stats = agent.memory.get_stats()
                stats["max_context_tokens"] = agent.config.memory.max_context_tokens
                self._send_json(stats)
                return

            if tail == ["memories"]:
                memory_type = query.get("type", [None])[0]
                self._send_json({"memories": agent.memory.list_modules(memory_type)})
                return

            self._send_error(404, "接口不存在")
        except Exception as exc:
            self._send_exception(exc)

    def do_POST(self):
        try:
            path, parts, query = self._request_parts()
            self._authorize_request(path)

            if path == "/api/auth/login":
                data = self._read_json()
                self._send_json(self.auth_session.login(data.get("account") or data.get("email"), data.get("password")))
                return

            if path == "/api/auth/register":
                data = self._read_json()
                self._send_json(self.auth_session.register(data.get("email") or data.get("account"), data.get("password")))
                return

            if path == "/api/auth/refresh":
                self._send_json(self.auth_session.refresh())
                return

            if path == "/api/auth/logout":
                self._send_json(self.auth_session.logout())
                return

            if path == "/api/credit/consume":
                data = self._read_json()
                self._send_json(self.auth_session.credit_consume(
                    data.get("amount"),
                    data.get("action"),
                    data.get("description", ""),
                ))
                return

            if path == "/api/credit/settle":
                self._send_json(self.auth_session.settle_credit())
                return

            if path == "/api/system/open-url":
                data = self._read_json()
                self._send_json(_open_system_url(data.get("url")))
                return

            if path == "/api/pay/create":
                data = self._read_json()
                self._send_json(self.auth_session.pay_create(data.get("amount")))
                return

            if path == "/api/user/profile":
                profile = save_user_profile(self._read_json(), account_id=self.agent_registry.account_id())
                self._send_json({"ok": True, "profile": profile})
                return

            if path == "/api/hardware/pair":
                self._send_json(_hardware_hub_request("/api/panel/device/pair", "POST", self._read_json()))
                return

            if path == "/api/hardware/delete":
                self._send_json(_hardware_hub_request("/api/panel/device/delete", "POST", self._read_json()))
                return

            if path == "/api/hardware/command":
                self._send_json(_hardware_hub_request("/api/panel/command", "POST", self._read_json()))
                return

            if path == "/api/hardware/config":
                self._send_json(_hardware_hub_request("/api/panel/config", "POST", self._read_json()))
                return

            if path == "/api/agents":
                agent = self.agent_registry.create_agent(self._read_json())
                self._send_json({"ok": True, "agent": agent, "selected": agent["id"]})
                return

            if path == "/api/agents/import":
                agent = self.agent_registry.import_agent(self._read_body())
                self._send_json({"ok": True, "agent": agent, "selected": agent["id"]})
                return

            if parts[:2] == ["api", "models"]:
                if len(parts) == 2:
                    self._send_json(self.model_registry.upsert_model(self._read_json()))
                    return
                if len(parts) == 3:
                    self._send_json(self.model_registry.upsert_model(self._read_json(), parts[2]))
                    return
                if len(parts) == 4 and parts[3] == "select":
                    self._send_json(self.model_registry.select_model(parts[2]))
                    return

            agent_name, tail = self._agent_route(parts)

            if tail == ["select"]:
                selected = self.agent_registry.select_agent(agent_name)
                self._send_json({"ok": True, "agent": selected, "selected": selected["id"]})
                return

            if tail == ["config"]:
                agent_data = self.agent_registry.update_agent(agent_name, self._read_json())
                self._send_json({"ok": True, "agent": agent_data})
                return

            agent = self.agent_registry.get(agent_name)

            if tail == ["chat"]:
                data = self._read_json()
                message = str(data.get("message", "")).strip()
                if not message:
                    self._send_error(400, "message 不能为空")
                    return
                system_overlay = str(data.get("system_overlay") or data.get("overlay") or "").strip()
                consume_credits = bool(data.get("consume_credits", data.get("save", True)))
                credit_unit = self.model_registry.current_model_credit_unit()
                should_charge_model = consume_credits and credit_unit > 0
                if should_charge_model:
                    self.auth_session.ensure_credit_available(credit_unit)
                response = agent.chat(
                    message,
                    save=bool(data.get("save", True)),
                    system_overlay=system_overlay,
                )
                credit = None
                if should_charge_model:
                    credit = self.auth_session.consume_conversation(
                        credit_unit,
                        message,
                        response,
                        action="model",
                        description="云端大模型消耗",
                    )
                self._send_json({"response": response, "credit": credit})
                return

            if tail == ["memory", "load"]:
                data = self._read_json()
                module_id = str(data.get("module_id", "")).strip()
                self._send_json({
                    "ok": agent.memory.load_module(module_id),
                    "module_id": module_id,
                })
                return

            if tail == ["memory", "unload"]:
                data = self._read_json()
                module_id = str(data.get("module_id", "")).strip()
                self._send_json({
                    "ok": agent.memory.unload_module(module_id),
                    "module_id": module_id,
                })
                return

            if tail == ["memory", "search"]:
                data = self._read_json()
                query_text = str(data.get("query", "")).strip()
                limit = int(data.get("limit", 5))
                self._send_json({"results": agent.memory.search(query_text, limit=limit)})
                return

            if tail == ["memory", "remember"]:
                data = self._read_json()
                fact = str(data.get("fact", "")).strip()
                tags = data.get("tags") or []
                if not fact:
                    self._send_error(400, "fact 不能为空")
                    return
                entry = agent.memory.semantic.add(fact, tags=tags, source="web")
                self._send_json({"ok": True, "fact": entry})
                return

            if tail == ["stt"]:
                self._handle_stt(agent, query)
                return

            if tail == ["tts", "stream"]:
                self._handle_tts_stream(agent)
                return

            if tail == ["tts"]:
                self._handle_tts(agent)
                return

            self._send_error(404, "接口不存在")
        except Exception as exc:
            self._send_exception(exc)

    def do_DELETE(self):
        try:
            path, parts, query = self._request_parts()
            self._authorize_request(path)
            if parts[:2] == ["api", "models"] and len(parts) == 3:
                self._send_json(self.model_registry.delete_model(parts[2]))
                return

            agent_name, tail = self._agent_route(parts)
            if tail:
                self._send_error(404, "接口不存在")
                return

            force = query.get("force", ["false"])[0].lower() in {"1", "true", "yes"}
            result = self.agent_registry.delete_agent(agent_name, force=force)
            self._send_json({"ok": True, **result})
        except Exception as exc:
            self._send_exception(exc)

    def _handle_stt(self, agent: Agent, query: dict):
        body = self._read_body()
        if not body:
            self._send_error(400, "音频内容为空")
            return

        ext = query.get("ext", [""])[0].strip(".").lower()
        if not ext:
            ext = self._guess_audio_ext()
        if ext not in {"wav", "pcm", "ogg", "opus", "mp3", "m4a", "aac", "amr", "silk", "speex"}:
            ext = "wav"
        self.auth_session.ensure_credit_available(STT_CREDIT_UNIT)
        agent._init_voice()
        with tempfile.NamedTemporaryFile(suffix=f".{ext}", delete=False) as fp:
            fp.write(body)
            audio_path = Path(fp.name)

        try:
            text = agent._stt.transcribe_file(
                str(audio_path),
                language=agent.config.voice.language,
            )
        finally:
            audio_path.unlink(missing_ok=True)

        payload = {"text": text}
        payload["credit"] = self.auth_session.queue_credit(
            STT_CREDIT_UNIT,
            "stt",
            "语音识别消耗",
        )
        audio_stats = getattr(agent._stt, "last_audio_stats", None)
        if audio_stats:
            payload["audio"] = audio_stats
        self._send_json(payload)

    def _handle_tts(self, agent: Agent):
        data = self._read_json()
        text = str(data.get("text", "")).strip()
        if not text:
            self._send_error(400, "text 不能为空")
            return

        agent._init_voice()
        spoken = agent._prepare_spoken_text(text)
        if not spoken:
            self._send_error(400, "没有可朗读文本")
            return

        output = Path(tempfile.mktemp(suffix=".mp3"))
        try:
            agent._tts.synthesize(spoken, str(output))
            content = output.read_bytes()
        finally:
            output.unlink(missing_ok=True)

        self._send_bytes(content, "audio/mpeg")

    def _handle_tts_stream(self, agent: Agent):
        data = self._read_json()
        text = str(data.get("text", "")).strip()
        if not text:
            self._send_error(400, "text 不能为空")
            return

        agent._init_voice()
        spoken = agent._prepare_spoken_text(text)
        if not spoken:
            self._send_error(400, "没有可朗读文本")
            return

        self.close_connection = True
        self.send_response(200)
        self._send_common_headers()
        self.send_header("Content-Type", "audio/mpeg")
        self.send_header("Connection", "close")
        self.end_headers()
        try:
            for chunk in agent._tts.iter_audio(spoken):
                self.wfile.write(chunk)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError):
            pass
        except Exception as exc:
            print(f"[web] TTS 流式输出失败: {exc}")

    def _request_parts(self) -> tuple[str, list[str], dict]:
        parsed = urlparse(self.path)
        path = parsed.path.rstrip("/") or "/"
        parts = [unquote(part) for part in path.split("/") if part]
        query = parse_qs(parsed.query)
        return path, parts, query

    def _authorize_preflight(self) -> None:
        if not self._client_allowed():
            raise BridgeHTTPError(403, "Bridge 只允许本机访问")
        if not self._origin_allowed():
            raise BridgeHTTPError(403, "Origin 不允许访问 Bridge")

    def _authorize_request(self, path: str) -> None:
        if not self._client_allowed():
            raise BridgeHTTPError(403, "Bridge 只允许本机访问")
        if not self._origin_allowed():
            raise BridgeHTTPError(403, "Origin 不允许访问 Bridge")
        if path == "/api/health":
            return

        request_token = self.headers.get("X-CoCo-Bridge-Token", "")
        if not self.bridge_token or not hmac.compare_digest(request_token, self.bridge_token):
            raise BridgeHTTPError(403, "Bridge token 无效")

    def _client_allowed(self) -> bool:
        if str(os.environ.get("coco_bridge_allow_remote", "")).strip() == "1":
            return True
        return _is_loopback_host(str(self.client_address[0] if self.client_address else ""))

    def _origin_allowed(self) -> bool:
        origin = self.headers.get("Origin", "")
        if not origin:
            return True
        parsed = urlparse(origin)
        if parsed.scheme not in {"http", "https"}:
            return False
        return _is_loopback_host(parsed.hostname or "")

    def _agent_route(self, parts: list[str]) -> tuple[str, list[str]]:
        if len(parts) < 3 or parts[0] != "api" or parts[1] != "agents":
            raise RouteError("需要 /api/agents/{name}/...")
        return parts[2], parts[3:]

    def _read_body(self) -> bytes:
        length = int(self.headers.get("Content-Length", "0") or "0")
        return self.rfile.read(length) if length else b""

    def _read_json(self) -> dict:
        raw = self._read_body()
        if not raw:
            return {}
        try:
            return json.loads(raw.decode("utf-8"))
        except json.JSONDecodeError as exc:
            raise BadRequest(f"JSON 格式错误: {exc}") from exc

    def _guess_audio_ext(self) -> str:
        content_type = (self.headers.get("Content-Type") or "").lower()
        if "wav" in content_type:
            return "wav"
        if "mpeg" in content_type or "mp3" in content_type:
            return "mp3"
        if "ogg" in content_type:
            return "ogg"
        if "mp4" in content_type or "m4a" in content_type:
            return "m4a"
        return "wav"

    def _send_json(self, data: dict, status: int = 200):
        payload = json.dumps(data, ensure_ascii=False).encode("utf-8")
        self._send_bytes(payload, "application/json; charset=utf-8", status=status)

    def _send_error(self, status: int, message: str):
        self._send_json({"ok": False, "error": message}, status=status)

    def _send_exception(self, exc: Exception):
        if isinstance(exc, FileNotFoundError):
            self._send_error(404, str(exc))
        elif isinstance(exc, BadRequest):
            self._send_error(400, str(exc))
        elif isinstance(exc, BridgeHTTPError):
            self._send_error(exc.status, str(exc))
        elif isinstance(exc, RouteError):
            self._send_error(404, str(exc))
        else:
            self._send_error(500, str(exc))

    def _send_empty(self, status: int):
        self.send_response(status)
        self._send_common_headers()
        self.end_headers()

    def _send_bytes(self, content: bytes, content_type: str, status: int = 200, headers: dict | None = None):
        self.send_response(status)
        self._send_common_headers()
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(content)))
        for key, value in (headers or {}).items():
            self.send_header(key, value)
        self.end_headers()
        self.wfile.write(content)

    def _send_common_headers(self):
        origin = self.headers.get("Origin", "")
        if origin and self._origin_allowed():
            self.send_header("Access-Control-Allow-Origin", origin)
            self.send_header("Vary", "Origin")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-CoCo-Bridge-Token")
        self.send_header("Access-Control-Expose-Headers", "Content-Disposition")
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")

    def log_message(self, fmt, *args):
        print(f"[web] {self.address_string()} - {fmt % args}")


class BadRequest(Exception):
    pass


class BridgeHTTPError(Exception):
    def __init__(self, status: int, message: str):
        super().__init__(message)
        self.status = max(400, min(599, int(status or 500)))


class RouteError(Exception):
    pass
