from __future__ import annotations

import asyncio
import math
import hashlib
import json
import mimetypes
import os
import secrets
import sqlite3
import time
import uuid
from contextlib import asynccontextmanager, contextmanager
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, AsyncIterator, Iterator
from urllib.parse import urlparse

import httpx
import yaml
from cryptography.fernet import Fernet, InvalidToken
from fastapi import Depends, FastAPI, File, Header, HTTPException, Request, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field


ROOT = Path(__file__).resolve().parent
DB_PATH = Path(os.environ.get("BRIDGE_DB_PATH", ROOT / "data" / "funsclaw_bridge.sqlite3"))
UPLOAD_DIR = Path(os.environ.get("BRIDGE_UPLOAD_DIR", ROOT / "data" / "uploads"))
FUNSCLAW_URL = os.environ.get("FUNSCLAW_API_URL", "http://127.0.0.1:8080").rstrip("/")
FUNSCLAW_TOKEN = os.environ.get("FUNSCLAW_API_TOKEN", "").strip()
FUNSCLAW_CONFIG_PATH = os.environ.get("FUNSCLAW_CONFIG_PATH", "").strip()
TOKEN_TTL_MINUTES = max(5, int(os.environ.get("BRIDGE_TOKEN_TTL_MINUTES", "30")))
YUANXU_TLS_VERIFY = os.environ.get("YUANXU_TLS_VERIFY", "true").lower() == "true"
ALLOW_REMOTE_EXCHANGE = os.environ.get("BRIDGE_ALLOW_REMOTE_EXCHANGE", "false").lower() == "true"
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif"}
SECRET_KEY_PATH = Path(
    os.environ.get("BRIDGE_SECRET_KEY_PATH", DB_PATH.parent / ".provider_secret.key")
)


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


@contextmanager
def connect_db() -> Iterator[sqlite3.Connection]:
    connection = sqlite3.connect(DB_PATH)
    connection.row_factory = sqlite3.Row
    try:
        with connection:
            yield connection
    finally:
        connection.close()


def init_db() -> None:
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
    with connect_db() as db:
        db.executescript(
            """
            CREATE TABLE IF NOT EXISTS bridge_tokens (
              token_hash TEXT PRIMARY KEY,
              user_id TEXT NOT NULL,
              tenant_id TEXT NOT NULL,
              display_name TEXT NOT NULL,
              scopes_json TEXT NOT NULL,
              identity_json TEXT NOT NULL DEFAULT '{}',
              expires_at TEXT NOT NULL,
              created_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS conversations (
              id TEXT PRIMARY KEY,
              user_id TEXT NOT NULL,
              tenant_id TEXT NOT NULL,
              title TEXT NOT NULL,
              agent_id TEXT NOT NULL DEFAULT '',
              model_id TEXT NOT NULL DEFAULT '',
              created_at TEXT NOT NULL,
              updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS messages (
              id TEXT PRIMARY KEY,
              conversation_id TEXT NOT NULL,
              role TEXT NOT NULL,
              content TEXT NOT NULL,
              attachments_json TEXT NOT NULL DEFAULT '[]',
              tool_states_json TEXT NOT NULL DEFAULT '[]',
              created_at TEXT NOT NULL,
              FOREIGN KEY(conversation_id) REFERENCES conversations(id)
            );
            CREATE TABLE IF NOT EXISTS attachments (
              id TEXT PRIMARY KEY,
              conversation_id TEXT NOT NULL,
              user_id TEXT NOT NULL,
              tenant_id TEXT NOT NULL,
              name TEXT NOT NULL,
              content_type TEXT NOT NULL,
              size INTEGER NOT NULL,
              local_path TEXT NOT NULL,
              remote_json TEXT NOT NULL DEFAULT '{}',
              created_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS platform_agents (
              id TEXT PRIMARY KEY,
              name TEXT NOT NULL,
              description TEXT NOT NULL DEFAULT '',
              definition_json TEXT NOT NULL DEFAULT '{}',
              version INTEGER NOT NULL DEFAULT 1,
              status TEXT NOT NULL DEFAULT 'published',
              source TEXT NOT NULL DEFAULT 'desktop-import',
              sync_locked INTEGER NOT NULL DEFAULT 0,
              created_at TEXT NOT NULL,
              updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS agent_versions (
              id TEXT PRIMARY KEY,
              agent_id TEXT NOT NULL,
              version INTEGER NOT NULL,
              definition_json TEXT NOT NULL,
              status TEXT NOT NULL DEFAULT 'published',
              created_by TEXT NOT NULL DEFAULT 'system-import',
              published_at TEXT NOT NULL,
              UNIQUE(agent_id, version)
            );
            CREATE TABLE IF NOT EXISTS enterprise_agent_grants (
              tenant_id TEXT NOT NULL,
              agent_id TEXT NOT NULL,
              enabled INTEGER NOT NULL DEFAULT 1,
              credit_multiplier REAL NOT NULL DEFAULT 1.0,
              granted_by TEXT NOT NULL DEFAULT 'system',
              created_at TEXT NOT NULL,
              updated_at TEXT NOT NULL,
              PRIMARY KEY(tenant_id, agent_id)
            );
            CREATE TABLE IF NOT EXISTS enterprise_agent_customizations (
              tenant_id TEXT NOT NULL,
              agent_id TEXT NOT NULL,
              supplement_prompt TEXT NOT NULL DEFAULT '',
              updated_by TEXT NOT NULL DEFAULT 'system',
              updated_at TEXT NOT NULL,
              PRIMARY KEY(tenant_id, agent_id)
            );
            CREATE TABLE IF NOT EXISTS ai_model_providers (
              id TEXT PRIMARY KEY,
              provider_name TEXT NOT NULL,
              base_url TEXT NOT NULL,
              model_ids_json TEXT NOT NULL DEFAULT '[]',
              api_key_cipher TEXT NOT NULL DEFAULT '',
              api_key_mask TEXT NOT NULL DEFAULT '',
              enabled INTEGER NOT NULL DEFAULT 1,
              pricing_json TEXT NOT NULL DEFAULT '{}',
              updated_by TEXT NOT NULL DEFAULT 'system',
              updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS user_agent_grants (
              tenant_id TEXT NOT NULL,
              user_id TEXT NOT NULL,
              agent_id TEXT NOT NULL,
              enabled INTEGER NOT NULL DEFAULT 1,
              granted_by TEXT NOT NULL DEFAULT 'system',
              created_at TEXT NOT NULL,
              updated_at TEXT NOT NULL,
              PRIMARY KEY(tenant_id, user_id, agent_id)
            );
            CREATE TABLE IF NOT EXISTS subject_agent_grants (
              tenant_id TEXT NOT NULL,
              subject_type TEXT NOT NULL,
              subject_id TEXT NOT NULL,
              agent_id TEXT NOT NULL,
              enabled INTEGER NOT NULL DEFAULT 1,
              granted_by TEXT NOT NULL DEFAULT 'system',
              created_at TEXT NOT NULL,
              updated_at TEXT NOT NULL,
              PRIMARY KEY(tenant_id, subject_type, subject_id, agent_id)
            );
            CREATE TABLE IF NOT EXISTS user_ai_policies (
              tenant_id TEXT NOT NULL,
              user_id TEXT NOT NULL,
              enabled INTEGER NOT NULL DEFAULT 1,
              daily_limit INTEGER NOT NULL DEFAULT 0,
              monthly_limit INTEGER NOT NULL DEFAULT 0,
              tools_enabled INTEGER NOT NULL DEFAULT 0,
              computer_enabled INTEGER NOT NULL DEFAULT 0,
              files_enabled INTEGER NOT NULL DEFAULT 1,
              updated_by TEXT NOT NULL DEFAULT 'system',
              updated_at TEXT NOT NULL,
              PRIMARY KEY(tenant_id, user_id)
            );
            CREATE TABLE IF NOT EXISTS enterprise_wallets (
              tenant_id TEXT PRIMARY KEY,
              total_credits INTEGER NOT NULL DEFAULT 0,
              used_credits INTEGER NOT NULL DEFAULT 0,
              frozen_credits INTEGER NOT NULL DEFAULT 0,
              warning_threshold INTEGER NOT NULL DEFAULT 100,
              status TEXT NOT NULL DEFAULT 'active',
              updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS credit_ledger (
              id TEXT PRIMARY KEY,
              tenant_id TEXT NOT NULL,
              change_type TEXT NOT NULL,
              amount INTEGER NOT NULL,
              balance_after INTEGER NOT NULL,
              operator_id TEXT NOT NULL,
              request_id TEXT NOT NULL DEFAULT '',
              note TEXT NOT NULL DEFAULT '',
              created_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS credit_recharge_requests (
              id TEXT PRIMARY KEY,
              tenant_id TEXT NOT NULL,
              amount INTEGER NOT NULL,
              note TEXT NOT NULL DEFAULT '',
              status TEXT NOT NULL DEFAULT 'pending',
              requested_by TEXT NOT NULL,
              reviewed_by TEXT NOT NULL DEFAULT '',
              review_note TEXT NOT NULL DEFAULT '',
              created_at TEXT NOT NULL,
              reviewed_at TEXT NOT NULL DEFAULT ''
            );
            CREATE TABLE IF NOT EXISTS ai_usage_records (
              id TEXT PRIMARY KEY,
              tenant_id TEXT NOT NULL,
              user_id TEXT NOT NULL,
              agent_id TEXT NOT NULL DEFAULT '',
              conversation_id TEXT NOT NULL DEFAULT '',
              message_id TEXT NOT NULL DEFAULT '',
              capability_type TEXT NOT NULL,
              model_id TEXT NOT NULL,
              provider TEXT NOT NULL DEFAULT '',
              input_tokens INTEGER NOT NULL DEFAULT 0,
              output_tokens INTEGER NOT NULL DEFAULT 0,
              tool_calls INTEGER NOT NULL DEFAULT 0,
              computer_actions INTEGER NOT NULL DEFAULT 0,
              task_duration_ms INTEGER NOT NULL DEFAULT 0,
              credits_used INTEGER NOT NULL DEFAULT 0,
              status TEXT NOT NULL,
              request_id TEXT NOT NULL UNIQUE,
              routing_reason TEXT NOT NULL DEFAULT '',
              department_id TEXT NOT NULL DEFAULT '',
              position_id TEXT NOT NULL DEFAULT '',
              created_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS ai_request_reservations (
              request_id TEXT PRIMARY KEY,
              tenant_id TEXT NOT NULL,
              user_id TEXT NOT NULL,
              created_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS conversation_audit_events (
              id TEXT PRIMARY KEY,
              tenant_id TEXT NOT NULL,
              user_id TEXT NOT NULL,
              conversation_id TEXT NOT NULL,
              event_type TEXT NOT NULL,
              summary TEXT NOT NULL DEFAULT '',
              metadata_json TEXT NOT NULL DEFAULT '{}',
              created_at TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_conversations_owner
              ON conversations(tenant_id, user_id, updated_at DESC);
            CREATE INDEX IF NOT EXISTS idx_messages_conversation
              ON messages(conversation_id, created_at);
            CREATE INDEX IF NOT EXISTS idx_usage_tenant_created
              ON ai_usage_records(tenant_id, created_at DESC);
            CREATE INDEX IF NOT EXISTS idx_audit_tenant_created
              ON conversation_audit_events(tenant_id, created_at DESC);
            CREATE INDEX IF NOT EXISTS idx_agent_versions_agent
              ON agent_versions(agent_id, version DESC);
            CREATE INDEX IF NOT EXISTS idx_recharge_tenant_created
              ON credit_recharge_requests(tenant_id, created_at DESC);
            """
        )
        # SQLite installations created by earlier bridge builds are upgraded in
        # place. Agent/model metadata belongs to each message so switching an
        # agent never rewrites the historical identity of previous replies.
        message_columns = {row["name"] for row in db.execute("PRAGMA table_info(messages)")}
        for column, ddl in {
            "agent_id": "TEXT NOT NULL DEFAULT ''",
            "agent_version": "INTEGER NOT NULL DEFAULT 0",
            "model_id": "TEXT NOT NULL DEFAULT ''",
        }.items():
            if column not in message_columns:
                db.execute(f"ALTER TABLE messages ADD COLUMN {column} {ddl}")
        usage_columns = {row["name"] for row in db.execute("PRAGMA table_info(ai_usage_records)")}
        for column, ddl in {
            "agent_version": "INTEGER NOT NULL DEFAULT 0",
            "department_id": "TEXT NOT NULL DEFAULT ''",
            "position_id": "TEXT NOT NULL DEFAULT ''",
        }.items():
            if column not in usage_columns:
                db.execute(f"ALTER TABLE ai_usage_records ADD COLUMN {column} {ddl}")
        conversation_columns = {
            row["name"] for row in db.execute("PRAGMA table_info(conversations)")
        }
        for column, ddl in {
            "pinned": "INTEGER NOT NULL DEFAULT 0",
            "archived": "INTEGER NOT NULL DEFAULT 0",
        }.items():
            if column not in conversation_columns:
                db.execute(f"ALTER TABLE conversations ADD COLUMN {column} {ddl}")
        agent_columns = {
            row["name"] for row in db.execute("PRAGMA table_info(platform_agents)")
        }
        for column, ddl in {
            "source": "TEXT NOT NULL DEFAULT 'desktop-import'",
            "sync_locked": "INTEGER NOT NULL DEFAULT 0",
        }.items():
            if column not in agent_columns:
                db.execute(f"ALTER TABLE platform_agents ADD COLUMN {column} {ddl}")
        token_columns = {
            row["name"] for row in db.execute("PRAGMA table_info(bridge_tokens)")
        }
        if "identity_json" not in token_columns:
            db.execute(
                "ALTER TABLE bridge_tokens ADD COLUMN identity_json TEXT NOT NULL DEFAULT '{}'"
            )


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_db()
    yield


app = FastAPI(title="Yuanxu FunsClaw Windows Bridge", version="1.0", lifespan=lifespan)


class TokenExchangeRequest(BaseModel):
    client: str = "yuanxu-windows"
    yuanxuBaseUrl: str


class ConversationCreate(BaseModel):
    title: str = "新会话"
    agentId: str = ""
    modelId: str = ""


class ConversationUpdate(BaseModel):
    title: str = Field(min_length=1, max_length=120)


class ConversationStateUpdate(BaseModel):
    pinned: bool | None = None
    archived: bool | None = None


class MessageRequest(BaseModel):
    content: str = ""
    agentId: str = ""
    modelId: str = ""
    attachmentIds: list[str] = []
    toolsEnabled: bool = True
    requestId: str = ""


class WalletTopUpRequest(BaseModel):
    tenantId: str = ""
    amount: int = Field(gt=0, le=1000000000)
    note: str = Field(default="", max_length=300)


class RechargeRequestCreate(BaseModel):
    amount: int = Field(gt=0, le=1000000000)
    note: str = Field(default="", max_length=300)


class RechargeRequestReview(BaseModel):
    approved: bool
    note: str = Field(default="", max_length=300)


class EnterpriseAgentGrantRequest(BaseModel):
    tenantId: str = ""
    agentId: str
    enabled: bool = True
    creditMultiplier: float = Field(default=1.0, ge=0.1, le=100)


class EnterpriseAgentCustomizationRequest(BaseModel):
    agentId: str = Field(min_length=1, max_length=200)
    supplementPrompt: str = Field(default="", max_length=8000)


class UserAgentGrantRequest(BaseModel):
    userId: str
    agentId: str
    enabled: bool = True


class SubjectAgentGrantRequest(BaseModel):
    subjectType: str = Field(pattern="^(department|position)$")
    subjectId: str = Field(min_length=1, max_length=200)
    agentId: str
    enabled: bool = True


class UserAiPolicyRequest(BaseModel):
    userId: str = Field(min_length=1, max_length=200)
    enabled: bool = True
    dailyLimit: int = Field(default=0, ge=0, le=1000000000)
    monthlyLimit: int = Field(default=0, ge=0, le=1000000000)
    toolsEnabled: bool = False
    computerEnabled: bool = False
    filesEnabled: bool = True


class PlatformAgentRequest(BaseModel):
    id: str = Field(default="", max_length=120)
    name: str = Field(min_length=1, max_length=120)
    description: str = Field(default="", max_length=1000)
    rolePrompt: str = Field(default="", max_length=50000)
    knowledge: list[str] = Field(default_factory=list)
    tools: list[str] = Field(default_factory=list)
    modelStrategy: dict[str, Any] = Field(default_factory=dict)
    category: str = Field(default="", max_length=120)
    icon: str = Field(default="", max_length=120)
    usage: str = Field(default="", max_length=2000)


class PlatformAgentRollbackRequest(BaseModel):
    version: int = Field(ge=1)


class PlatformAgentTestRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=8000)


class PlatformProviderRequest(BaseModel):
    id: str = Field(min_length=1, max_length=80, pattern="^[A-Za-z0-9_-]+$")
    providerName: str = Field(min_length=1, max_length=120)
    baseUrl: str = Field(min_length=8, max_length=500)
    modelIds: list[str] = Field(default_factory=list, max_length=100)
    apiKey: str = Field(default="", max_length=1000)
    enabled: bool = True
    pricing: dict[str, Any] = Field(default_factory=dict)


def _text(value: Any, fallback: str = "") -> str:
    result = str(value or "").strip()
    return result or fallback


def _pick(mapping: dict[str, Any], *paths: tuple[str, ...]) -> Any:
    for path in paths:
        value: Any = mapping
        for key in path:
            if not isinstance(value, dict):
                value = None
                break
            value = value.get(key)
        if value not in (None, ""):
            return value
    return None


def _token_hash(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def _string_set(value: Any) -> set[str]:
    if isinstance(value, str):
        return {item.strip() for item in value.replace(";", ",").split(",") if item.strip()}
    if isinstance(value, (list, tuple, set)):
        return {_text(item) for item in value if _text(item)}
    return set()


def _identity_ids(value: Any) -> set[str]:
    if isinstance(value, dict):
        candidate = _text(value.get("id") or value.get("code") or value.get("value"))
        return {candidate} if candidate else set()
    if isinstance(value, (list, tuple, set)):
        result: set[str] = set()
        for item in value:
            result.update(_identity_ids(item))
        return result
    return _string_set(value)


def _has_scope(principal: dict[str, Any], *required: str) -> bool:
    scopes = set(principal.get("scopes") or [])
    return bool(scopes.intersection(required))


def _require_scope(principal: dict[str, Any], *required: str) -> None:
    if not _has_scope(principal, *required):
        raise HTTPException(403, "当前账号没有此项AI管理权限")


def _funs_headers() -> dict[str, str]:
    return {"Authorization": f"Bearer {FUNSCLAW_TOKEN}"} if FUNSCLAW_TOKEN else {}


def _is_loopback(host: str | None) -> bool:
    return host in {"127.0.0.1", "::1", "localhost", None}


def _funs_trust_env() -> bool:
    """Local desktop FunsClaw must not be routed through a Windows HTTP proxy."""
    return not _is_loopback(urlparse(FUNSCLAW_URL).hostname)


async def current_principal(authorization: str | None = Header(default=None)) -> dict[str, Any]:
    if not authorization or not authorization.lower().startswith("bearer "):
        raise HTTPException(401, "小元AI登录已失效")
    token = authorization.split(" ", 1)[1].strip()
    with connect_db() as db:
        row = db.execute(
            "SELECT * FROM bridge_tokens WHERE token_hash = ?", (_token_hash(token),)
        ).fetchone()
    if row is None or datetime.fromisoformat(row["expires_at"]) <= datetime.now(timezone.utc):
        raise HTTPException(401, "小元AI登录已过期")
    identity = json.loads(row["identity_json"] or "{}")
    return {
        "user_id": row["user_id"],
        "tenant_id": row["tenant_id"],
        "display_name": row["display_name"],
        "scopes": json.loads(row["scopes_json"]),
        "department_ids": sorted(_identity_ids(identity.get("departmentIds"))),
        "position_ids": sorted(_identity_ids(identity.get("positionIds"))),
    }


async def _verify_yuanxu_session(base_url: str, cookie: str) -> dict[str, Any]:
    parsed = urlparse(base_url)
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise HTTPException(400, "元序服务器地址无效")
    async with httpx.AsyncClient(verify=YUANXU_TLS_VERIFY, timeout=15) as client:
        response = await client.get(
            f"{base_url.rstrip('/')}/api/session", headers={"Cookie": cookie}
        )
    if response.status_code != 200:
        raise HTTPException(401, "元序登录已失效")
    try:
        payload = response.json()
    except ValueError as error:
        raise HTTPException(502, "元序会话服务返回异常") from error
    user = payload.get("user") or payload.get("account") or payload.get("session") or payload
    if not isinstance(user, dict):
        raise HTTPException(401, "未识别到元序用户")
    user_id = _text(_pick(user, ("id",), ("userId",), ("accountId",), ("username",)))
    if not user_id:
        raise HTTPException(401, "元序用户身份不完整")
    tenant_id = _text(
        _pick(user, ("tenantId",), ("enterpriseId",), ("tenant", "id"), ("enterprise", "id")),
        "yuanxu",
    )
    roles = _string_set(
        _pick(user, ("roles",), ("roleCodes",), ("role",), ("positionCodes",))
    )
    permissions = _string_set(
        _pick(user, ("permissions",), ("permissionCodes",), ("scopes",))
    )
    department_ids = _identity_ids(
        _pick(user, ("departmentIds",), ("departmentId",), ("departments",), ("department",))
    )
    position_ids = _identity_ids(
        _pick(user, ("positionIds",), ("positionId",), ("positions",), ("position",))
    )
    return {
        "user_id": user_id,
        "tenant_id": tenant_id,
        "display_name": _text(_pick(user, ("name",), ("displayName",), ("username",)), user_id),
        "roles": sorted(roles),
        "permissions": sorted(permissions),
        "department_ids": sorted(department_ids),
        "position_ids": sorted(position_ids),
    }


def _workspace_scopes(principal: dict[str, Any]) -> list[str]:
    roles = {item.lower() for item in principal.get("roles") or []}
    permissions = {item.lower() for item in principal.get("permissions") or []}
    scopes = {"chat", "files", "history"}
    if not roles and not permissions:
        # Older Yuanxu sessions do not expose role codes yet. Keep the existing
        # workbench usable while all management endpoints remain protected.
        scopes.add("tools")
    if roles.intersection({"platform_admin", "super_admin", "enterprise_admin", "admin"}):
        scopes.update({"tools", "ai_usage_view", "ai_agent_manage", "ai_credit_request"})
    if roles.intersection({"platform_admin", "super_admin"}):
        scopes.update({
            "ai_credit_manage", "ai_conversation_audit", "ai_conversation_export",
            "ai_platform_manage",
        })
    permission_map = {
        "ai_tools": "tools",
        "ai_usage_view": "ai_usage_view",
        "ai_agent_manage": "ai_agent_manage",
        "ai_credit_manage": "ai_credit_manage",
        "ai_credit_request": "ai_credit_request",
        "ai_conversation_audit": "ai_conversation_audit",
        "ai_conversation_export": "ai_conversation_export",
        "ai_platform_manage": "ai_platform_manage",
        "ai_sensitive_content_view": "ai_sensitive_content_view",
    }
    for source, target in permission_map.items():
        if source in permissions:
            scopes.add(target)
    return sorted(scopes)


@app.get("/health")
async def health() -> dict[str, Any]:
    return {"ok": True, "funsclawConfigured": bool(FUNSCLAW_TOKEN), "desktop": True}


@app.post("/v1/token-exchange")
async def token_exchange(
    body: TokenExchangeRequest,
    request: Request,
    x_yuanxu_cookie: str | None = Header(default=None),
) -> dict[str, Any]:
    if not ALLOW_REMOTE_EXCHANGE and not _is_loopback(request.client.host if request.client else None):
        raise HTTPException(403, "本地桥接服务拒绝远程换取凭证")
    if body.client != "yuanxu-windows":
        raise HTTPException(400, "仅支持元序 Windows 客户端")
    if not x_yuanxu_cookie:
        raise HTTPException(401, "缺少元序登录会话")
    principal = await _verify_yuanxu_session(body.yuanxuBaseUrl, x_yuanxu_cookie)
    token = secrets.token_urlsafe(32)
    expires_at = datetime.now(timezone.utc) + timedelta(minutes=TOKEN_TTL_MINUTES)
    scopes = _workspace_scopes(principal)
    with connect_db() as db:
        db.execute("DELETE FROM bridge_tokens WHERE expires_at <= ?", (now_iso(),))
        db.execute(
            """INSERT INTO bridge_tokens
               (token_hash,user_id,tenant_id,display_name,scopes_json,identity_json,expires_at,created_at)
               VALUES (?,?,?,?,?,?,?,?)""",
            (
                _token_hash(token), principal["user_id"], principal["tenant_id"],
                principal["display_name"], json.dumps(scopes),
                json.dumps({
                    "departmentIds": principal.get("department_ids") or [],
                    "positionIds": principal.get("position_ids") or [],
                }),
                expires_at.isoformat(), now_iso(),
            ),
        )
    return {
        "accessToken": token,
        "expiresAt": expires_at.isoformat(),
        "gatewayBaseUrl": str(request.base_url).rstrip("/"),
        "scopes": scopes,
    }


async def _funs_get(path: str) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=20, trust_env=_funs_trust_env()) as client:
        response = await client.get(f"{FUNSCLAW_URL}{path}", headers=_funs_headers())
    if response.status_code >= 400:
        raise HTTPException(502, f"FunsClaw 服务异常（{response.status_code}）")
    return response.json()


def _provider_fernet() -> Fernet:
    configured = os.environ.get("BRIDGE_SECRET_KEY", "").strip().encode("ascii")
    if configured:
        try:
            return Fernet(configured)
        except (ValueError, TypeError) as error:
            raise RuntimeError("BRIDGE_SECRET_KEY 必须是有效的 Fernet 密钥") from error
    SECRET_KEY_PATH.parent.mkdir(parents=True, exist_ok=True)
    if not SECRET_KEY_PATH.exists():
        SECRET_KEY_PATH.write_bytes(Fernet.generate_key())
        try:
            os.chmod(SECRET_KEY_PATH, 0o600)
        except OSError:
            pass
    return Fernet(SECRET_KEY_PATH.read_bytes().strip())


def _encrypt_provider_key(value: str) -> str:
    return _provider_fernet().encrypt(value.encode("utf-8")).decode("ascii")


def _decrypt_provider_key(value: str) -> str:
    if not value:
        return ""
    try:
        return _provider_fernet().decrypt(value.encode("ascii")).decode("utf-8")
    except (InvalidToken, ValueError, UnicodeError) as error:
        raise HTTPException(500, "模型供应商密钥无法解密，请由平台管理员重新配置") from error


def _mask_provider_key(value: str) -> str:
    if len(value) <= 8:
        return "********"
    return f"{value[:3]}***{value[-4:]}"


def _provider_public_json(row: sqlite3.Row) -> dict[str, Any]:
    return {
        "id": row["id"],
        "providerName": row["provider_name"],
        "baseUrl": row["base_url"],
        "modelIds": json.loads(row["model_ids_json"] or "[]"),
        "apiKeyConfigured": bool(row["api_key_cipher"]),
        "apiKeyMask": row["api_key_mask"],
        "enabled": bool(row["enabled"]),
        "pricing": json.loads(row["pricing_json"] or "{}"),
        "updatedAt": row["updated_at"],
    }


def _cloud_configured_models() -> list[dict[str, str]]:
    with connect_db() as db:
        rows = db.execute(
            "SELECT id,model_ids_json FROM ai_model_providers WHERE enabled=1 ORDER BY id"
        ).fetchall()
    result: list[dict[str, str]] = []
    for row in rows:
        for model_id in json.loads(row["model_ids_json"] or "[]"):
            normalized = _text(model_id)
            if normalized:
                result.append({"id": normalized, "owned_by": row["id"]})
    return result


def _cloud_provider_for_model(model_id: str) -> sqlite3.Row | None:
    if not model_id:
        return None
    with connect_db() as db:
        rows = db.execute(
            "SELECT * FROM ai_model_providers WHERE enabled=1 ORDER BY id"
        ).fetchall()
    for row in rows:
        if model_id in json.loads(row["model_ids_json"] or "[]"):
            return row
    return None


def _provider_chat_payload(
    *,
    model_id: str,
    history: list[dict[str, str]],
    agent_override: dict[str, Any] | None,
) -> dict[str, Any]:
    messages = list(history)
    if agent_override:
        prompt = _text(
            agent_override.get("role_prompt")
            or agent_override.get("system_prompt")
        )
        if prompt:
            messages = [{"role": "system", "content": prompt}, *messages]
    return {
        "model": model_id,
        "messages": messages,
        "stream": True,
        "stream_options": {"include_usage": True},
    }


async def _test_provider_connection(row: sqlite3.Row) -> dict[str, Any]:
    key = _decrypt_provider_key(row["api_key_cipher"])
    models = json.loads(row["model_ids_json"] or "[]")
    if not key or not models:
        raise HTTPException(409, "请先配置 API Key 和至少一个模型")
    url = f"{row['base_url'].rstrip('/')}/chat/completions"
    started = time.perf_counter()
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.post(
            url,
            headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
            json={
                "model": models[0],
                "messages": [{"role": "user", "content": "请只回复：连接成功"}],
                "max_tokens": 16,
                "stream": False,
            },
        )
    elapsed = int((time.perf_counter() - started) * 1000)
    if response.status_code >= 400:
        detail = response.text[:300]
        raise HTTPException(502, f"供应商测试失败（{response.status_code}）：{detail}")
    return {"ok": True, "latencyMs": elapsed, "modelId": models[0]}


async def _test_platform_agent_definition(
    row: sqlite3.Row,
    prompt: str,
) -> dict[str, Any]:
    definition = json.loads(row["definition_json"] or "{}")
    strategy = definition.get("model_strategy") or {}
    preferred = _text(
        strategy.get("default")
        or strategy.get("default_model")
        or strategy.get("model")
    )
    models = _configured_models()
    selected = next(
        (item for item in models if item["id"] == preferred),
        models[0] if models else None,
    )
    if selected is None:
        raise HTTPException(409, "请先启用至少一个模型供应商")
    provider = _cloud_provider_for_model(selected["id"])
    if provider is None:
        raise HTTPException(409, "智能体测试需要平台模型供应商配置")
    payload = _provider_chat_payload(
        model_id=selected["id"],
        history=[{"role": "user", "content": prompt.strip()}],
        agent_override=definition,
    )
    payload["stream"] = False
    payload.pop("stream_options", None)
    started = time.perf_counter()
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(
            f"{provider['base_url'].rstrip('/')}/chat/completions",
            headers={
                "Authorization": (
                    f"Bearer {_decrypt_provider_key(provider['api_key_cipher'])}"
                ),
                "Content-Type": "application/json",
            },
            json=payload,
        )
    elapsed = int((time.perf_counter() - started) * 1000)
    if response.status_code >= 400:
        raise HTTPException(
            502,
            f"智能体测试失败（{response.status_code}）：{response.text[:300]}",
        )
    data = response.json()
    choices = data.get("choices") or []
    answer = ""
    if choices:
        answer = _text((choices[0].get("message") or {}).get("content"))
    usage = data.get("usage") if isinstance(data.get("usage"), dict) else {}
    return {
        "ok": True,
        "answer": answer,
        "modelId": selected["id"],
        "latencyMs": elapsed,
        "usage": usage,
    }


def _configured_models() -> list[dict[str, str]]:
    cloud = _cloud_configured_models()
    if cloud:
        return cloud
    if not FUNSCLAW_CONFIG_PATH:
        return []
    path = Path(FUNSCLAW_CONFIG_PATH)
    if not path.is_file():
        return []
    try:
        config = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError):
        return []
    models = config.get("models") or {}
    aliases = models.get("aliases") or {}
    providers = models.get("providers") or {}
    enabled = {
        str(name): value
        for name, value in providers.items()
        if isinstance(value, dict) and value.get("enabled") is True and _text(value.get("api_key"))
    }
    result: list[dict[str, str]] = []
    seen: set[str] = set()
    for alias, target in aliases.items():
        provider = _text(target).split("/", 1)[0]
        if provider not in enabled or _text(alias) in seen:
            continue
        seen.add(_text(alias))
        result.append({"id": _text(alias), "owned_by": provider})
    if not result:
        for provider, settings in enabled.items():
            for model in settings.get("models") or []:
                model_id = _text(model)
                if model_id and model_id not in seen:
                    seen.add(model_id)
                    result.append({"id": model_id, "owned_by": provider})
    return result


def _yaml_provider_for_model(model_id: str) -> dict[str, str] | None:
    """Resolve an enabled desktop provider without exposing its key to Flutter."""
    if not FUNSCLAW_CONFIG_PATH or not model_id:
        return None
    path = Path(FUNSCLAW_CONFIG_PATH)
    if not path.is_file():
        return None
    try:
        config = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError):
        return None
    models = config.get("models") or {}
    aliases = models.get("aliases") or {}
    target = _text(aliases.get(model_id), model_id)
    if "/" in target:
        provider_id, provider_model = target.split("/", 1)
    else:
        provider_id, provider_model = "", target
    providers = models.get("providers") or {}
    candidates = (
        [(provider_id, providers.get(provider_id))]
        if provider_id
        else list(providers.items())
    )
    for name, settings in candidates:
        if not isinstance(settings, dict) or settings.get("enabled") is not True:
            continue
        api_key = _text(settings.get("api_key"))
        base_url = _text(settings.get("base_url") or settings.get("baseUrl"))
        available = {_text(item) for item in settings.get("models") or []}
        if not api_key or not base_url:
            continue
        if provider_id or provider_model in available:
            return {
                "id": _text(name),
                "base_url": base_url.rstrip("/"),
                "api_key": api_key,
                "model_id": provider_model,
            }
    return None


def _model_capabilities(model: dict[str, str]) -> set[str]:
    value = f"{model.get('id', '')} {model.get('owned_by', '')}".lower()
    capabilities = {"general", "file_analysis"}
    if any(marker in value for marker in ("doubao", "volc", "vision", "seed")):
        capabilities.update({"multimodal", "image_generation"})
    if any(marker in value for marker in ("deepseek", "coder", "code", "v4")):
        capabilities.update({"code", "long_task", "tool"})
    return capabilities


def _model_json(model: dict[str, str]) -> dict[str, Any]:
    return {
        **model,
        "capabilities": sorted(_model_capabilities(model)),
        "available": True,
    }


async def _desktop_agents() -> list[dict[str, Any]]:
    try:
        data = await _funs_get("/v1/agents")
        discovered = [
            item
            for item in data.get("agents", data.get("data", []))
            if isinstance(item, dict)
        ]
        if discovered:
            return discovered
    except (HTTPException, httpx.HTTPError):
        pass
    # A real model-backed assistant remains usable when the optional desktop
    # tool daemon is offline. Computer-operation tasks still require that
    # daemon and are never silently reported as completed.
    return [
        {
            "id": "xiaoyuan-assistant",
            "name": "小元助手",
            "description": "企业智能问答、文档分析与工作内容生成",
            "category": "general",
            "capabilities": [
                "general",
                "code",
                "file_analysis",
                "multimodal",
            ],
            "source": "yuanxu-bridge",
        }
    ]


def _sync_platform_agents(db: sqlite3.Connection, agents: list[dict[str, Any]]) -> None:
    timestamp = now_iso()
    for agent in agents:
        agent_id = _text(agent.get("id") or agent.get("code"))
        if not agent_id:
            continue
        definition_json = json.dumps(agent, ensure_ascii=False, sort_keys=True)
        existing = db.execute(
            "SELECT version,definition_json,sync_locked FROM platform_agents WHERE id=?",
            (agent_id,),
        ).fetchone()
        # Once a platform administrator edits an imported definition, the
        # desktop discovery job becomes read-only for that agent. This avoids
        # silently overwriting cloud prompt/tool/version management.
        if existing is not None and bool(existing["sync_locked"]):
            continue
        changed = existing is None or existing["definition_json"] != definition_json
        version = (int(existing["version"]) + 1) if existing and changed else (int(existing["version"]) if existing else 1)
        db.execute(
            """INSERT INTO platform_agents
               (id,name,description,definition_json,version,status,source,sync_locked,created_at,updated_at)
               VALUES (?,?,?,?,?,'published','desktop-import',0,?,?)
               ON CONFLICT(id) DO UPDATE SET name=excluded.name,
                  description=excluded.description,definition_json=excluded.definition_json,
                  version=excluded.version,status=excluded.status,source=excluded.source,
                  updated_at=excluded.updated_at""",
            (
                agent_id,
                _text(agent.get("name"), agent_id),
                _text(agent.get("description")),
                definition_json,
                version,
                timestamp,
                timestamp,
            ),
        )
        version_snapshot = db.execute(
            "SELECT 1 FROM agent_versions WHERE agent_id=? AND version=?",
            (agent_id, version),
        ).fetchone()
        if changed or version_snapshot is None:
            db.execute(
                """INSERT OR IGNORE INTO agent_versions
                   (id,agent_id,version,definition_json,status,created_by,published_at)
                   VALUES (?,?,?,?,?,?,?)""",
                (uuid.uuid4().hex, agent_id, version, definition_json, "published", "desktop-import", timestamp),
            )


def _ensure_tenant_defaults(db: sqlite3.Connection, tenant_id: str, agents: list[dict[str, Any]]) -> None:
    timestamp = now_iso()
    wallet = db.execute("SELECT tenant_id FROM enterprise_wallets WHERE tenant_id=?", (tenant_id,)).fetchone()
    if wallet is None:
        initial = max(0, int(os.environ.get("BRIDGE_INITIAL_GIFT_CREDITS", "50000")))
        db.execute(
            "INSERT INTO enterprise_wallets VALUES (?,?,?,?,?,?,?)",
            (tenant_id, initial, 0, 0, min(1000, max(100, initial // 10)), "active", timestamp),
        )
        if initial:
            db.execute(
                "INSERT INTO credit_ledger VALUES (?,?,?,?,?,?,?,?,?)",
                (uuid.uuid4().hex, tenant_id, "gift", initial, initial, "system", "", "企业AI初始体验算力", timestamp),
            )
    # Imported desktop definitions become platform assets first. Production
    # tenants must receive an explicit platform grant before an enterprise can
    # distribute an agent to departments, positions or employees.
    if os.environ.get("BRIDGE_AUTO_GRANT_IMPORTED_AGENTS", "false").lower() == "true":
        for agent in agents:
            agent_id = _text(agent.get("id") or agent.get("code"))
            if not agent_id:
                continue
            db.execute(
                """INSERT OR IGNORE INTO enterprise_agent_grants
                   (tenant_id,agent_id,enabled,credit_multiplier,granted_by,created_at,updated_at)
                   VALUES (?,?,1,1.0,'initial-import',?,?)""",
                (tenant_id, agent_id, timestamp, timestamp),
            )


def _wallet_json(db: sqlite3.Connection, tenant_id: str) -> dict[str, Any]:
    row = db.execute("SELECT * FROM enterprise_wallets WHERE tenant_id=?", (tenant_id,)).fetchone()
    if row is None:
        return {"total": 0, "used": 0, "remaining": 0, "frozen": 0, "status": "inactive"}
    remaining = max(0, int(row["total_credits"]) - int(row["used_credits"]) - int(row["frozen_credits"]))
    return {
        "total": int(row["total_credits"]),
        "used": int(row["used_credits"]),
        "remaining": remaining,
        "frozen": int(row["frozen_credits"]),
        "warningThreshold": int(row["warning_threshold"]),
        "status": row["status"],
    }


def _user_policy_json(db: sqlite3.Connection, principal: dict[str, Any]) -> dict[str, Any]:
    row = db.execute(
        "SELECT * FROM user_ai_policies WHERE tenant_id=? AND user_id=?",
        (principal["tenant_id"], principal["user_id"]),
    ).fetchone()
    if row is None:
        # Existing Yuanxu accounts keep their current workspace capabilities
        # until an enterprise administrator explicitly applies a policy.
        return {
            "userId": principal["user_id"],
            "enabled": True,
            "dailyLimit": 0,
            "monthlyLimit": 0,
            "toolsEnabled": _has_scope(principal, "tools"),
            "computerEnabled": _has_scope(principal, "tools"),
            "filesEnabled": _has_scope(principal, "files"),
            "managed": False,
        }
    return {
        "userId": row["user_id"],
        "enabled": bool(row["enabled"]),
        "dailyLimit": int(row["daily_limit"]),
        "monthlyLimit": int(row["monthly_limit"]),
        "toolsEnabled": bool(row["tools_enabled"]),
        "computerEnabled": bool(row["computer_enabled"]),
        "filesEnabled": bool(row["files_enabled"]),
        "managed": True,
        "updatedBy": row["updated_by"],
        "updatedAt": row["updated_at"],
    }


def _user_usage_totals(
    db: sqlite3.Connection, tenant_id: str, user_id: str
) -> dict[str, int]:
    today = datetime.now(timezone.utc).date().isoformat()
    month = today[:7]
    row = db.execute(
        """SELECT
             COALESCE(SUM(CASE WHEN substr(created_at,1,10)=? THEN credits_used ELSE 0 END),0) AS daily,
             COALESCE(SUM(CASE WHEN substr(created_at,1,7)=? THEN credits_used ELSE 0 END),0) AS monthly
           FROM ai_usage_records
           WHERE tenant_id=? AND user_id=? AND status='success'""",
        (today, month, tenant_id, user_id),
    ).fetchone()
    return {"dailyUsed": int(row["daily"]), "monthlyUsed": int(row["monthly"])}


def _usage_summary(db: sqlite3.Connection, tenant_id: str) -> dict[str, Any]:
    """Return one consistent aggregation source for every enterprise AI view."""

    def grouped(column: str, *, where: str = "") -> list[dict[str, Any]]:
        safe_columns = {
            "user_id",
            "department_id",
            "position_id",
            "agent_id",
            "model_id",
            "capability_type",
        }
        if column not in safe_columns:
            raise ValueError("unsupported usage grouping")
        rows = db.execute(
            f"""SELECT {column} AS id,COUNT(*) AS request_count,
                COALESCE(SUM(input_tokens),0) AS input_tokens,
                COALESCE(SUM(output_tokens),0) AS output_tokens,
                COALESCE(SUM(credits_used),0) AS credits_used
                FROM ai_usage_records
                WHERE tenant_id=? AND status='success' {where}
                GROUP BY {column} ORDER BY credits_used DESC,{column}""",
            (tenant_id,),
        ).fetchall()
        return [dict(row) for row in rows]

    total = db.execute(
        """SELECT COUNT(*) AS request_count,
           COALESCE(SUM(input_tokens),0) AS input_tokens,
           COALESCE(SUM(output_tokens),0) AS output_tokens,
           COALESCE(SUM(tool_calls),0) AS tool_calls,
           COALESCE(SUM(computer_actions),0) AS computer_actions,
           COALESCE(SUM(task_duration_ms),0) AS task_duration_ms,
           COALESCE(SUM(credits_used),0) AS credits_used
           FROM ai_usage_records WHERE tenant_id=? AND status='success'""",
        (tenant_id,),
    ).fetchone()
    today = datetime.now(timezone.utc).date().isoformat()
    month = today[:7]
    periods = db.execute(
        """SELECT
           COALESCE(SUM(CASE WHEN substr(created_at,1,10)=? THEN credits_used ELSE 0 END),0) AS today,
           COALESCE(SUM(CASE WHEN substr(created_at,1,7)=? THEN credits_used ELSE 0 END),0) AS month
           FROM ai_usage_records WHERE tenant_id=? AND status='success'""",
        (today, month, tenant_id),
    ).fetchone()
    return {
        "total": dict(total),
        "periods": {"today": int(periods["today"]), "month": int(periods["month"])},
        "byUser": grouped("user_id"),
        "byDepartment": grouped("department_id", where="AND department_id<>''"),
        "byPosition": grouped("position_id", where="AND position_id<>''"),
        "byAgent": grouped("agent_id", where="AND agent_id<>''"),
        "byModel": grouped("model_id", where="AND model_id<>''"),
        "byCapability": grouped("capability_type"),
    }


def _enforce_user_policy(
    db: sqlite3.Connection,
    principal: dict[str, Any],
    *,
    tools_enabled: bool,
    has_files: bool,
    capability: str,
) -> dict[str, Any]:
    policy = _user_policy_json(db, principal)
    usage = _user_usage_totals(db, principal["tenant_id"], principal["user_id"])
    if not policy["enabled"]:
        raise HTTPException(403, "企业已暂停当前账号的小元AI使用权限")
    if tools_enabled and not policy["toolsEnabled"]:
        raise HTTPException(403, "企业尚未给当前账号开通AI工具权限")
    if has_files and not policy["filesEnabled"]:
        raise HTTPException(403, "企业尚未给当前账号开通AI文件权限")
    if capability == "computer_action" and not policy["computerEnabled"]:
        raise HTTPException(403, "企业尚未给当前账号开通电脑操作权限")
    if policy["dailyLimit"] and usage["dailyUsed"] >= policy["dailyLimit"]:
        raise HTTPException(402, "当前账号今日AI算力额度已用完")
    if policy["monthlyLimit"] and usage["monthlyUsed"] >= policy["monthlyLimit"]:
        raise HTTPException(402, "当前账号本月AI算力额度已用完")
    return {**policy, **usage}


def _allowed_agent_rows(db: sqlite3.Connection, principal: dict[str, Any]) -> list[sqlite3.Row]:
    user_overrides = db.execute(
        "SELECT agent_id,enabled FROM user_agent_grants WHERE tenant_id=? AND user_id=?",
        (principal["tenant_id"], principal["user_id"]),
    ).fetchall()
    override_map = {row["agent_id"]: bool(row["enabled"]) for row in user_overrides}
    departments = set(principal.get("department_ids") or [])
    positions = set(principal.get("position_ids") or [])
    subject_overrides: dict[str, list[bool]] = {}
    if departments or positions:
        grants = db.execute(
            """SELECT subject_type,subject_id,agent_id,enabled
               FROM subject_agent_grants WHERE tenant_id=?""",
            (principal["tenant_id"],),
        ).fetchall()
        for grant in grants:
            matched = (
                grant["subject_type"] == "department" and grant["subject_id"] in departments
            ) or (
                grant["subject_type"] == "position" and grant["subject_id"] in positions
            )
            if matched:
                subject_overrides.setdefault(grant["agent_id"], []).append(
                    bool(grant["enabled"])
                )
    rows = db.execute(
        """SELECT p.*,g.credit_multiplier FROM platform_agents p
           JOIN enterprise_agent_grants g ON g.agent_id=p.id
           WHERE g.tenant_id=? AND g.enabled=1 AND p.status='published'
           ORDER BY p.name""",
        (principal["tenant_id"],),
    ).fetchall()
    result: list[sqlite3.Row] = []
    for row in rows:
        agent_id = row["id"]
        if agent_id in override_map:
            enabled = override_map[agent_id]
        elif agent_id in subject_overrides:
            # A deny on any of the employee's matching subjects wins. A
            # concrete employee grant can still override this decision.
            enabled = all(subject_overrides[agent_id])
        else:
            enabled = True
        if enabled:
            result.append(row)
    return result


async def _prepare_workspace(principal: dict[str, Any]) -> tuple[list[dict[str, Any]], list[sqlite3.Row]]:
    desktop_agents = await _desktop_agents()
    with connect_db() as db:
        _sync_platform_agents(db, desktop_agents)
        _ensure_tenant_defaults(db, principal["tenant_id"], desktop_agents)
        allowed = _allowed_agent_rows(db, principal)
    return desktop_agents, allowed


def _agent_public_json(row: sqlite3.Row) -> dict[str, Any]:
    definition = json.loads(row["definition_json"] or "{}")
    return {
        "id": row["id"],
        "name": row["name"],
        "description": row["description"],
        "icon": _text(definition.get("icon")),
        "category": _text(definition.get("category")),
        "capabilities": sorted(_string_set(definition.get("capabilities") or definition.get("tags"))),
        "usage": _text(definition.get("usage") or definition.get("usage_notes")),
        "version": int(row["version"]),
        "source": "yuanxu-cloud",
    }


def _platform_definition(body: PlatformAgentRequest) -> dict[str, Any]:
    """Build the private cloud definition. Never use this in workspace APIs."""
    return {
        "name": body.name.strip(),
        "description": body.description.strip(),
        "role_prompt": body.rolePrompt.strip(),
        "knowledge": [item.strip() for item in body.knowledge if item.strip()],
        "tools": sorted({item.strip() for item in body.tools if item.strip()}),
        "model_strategy": body.modelStrategy,
        "category": body.category.strip(),
        "icon": body.icon.strip(),
        "usage": body.usage.strip(),
    }


def _platform_agent_json(row: sqlite3.Row, versions: list[sqlite3.Row] | None = None) -> dict[str, Any]:
    definition = json.loads(row["definition_json"] or "{}")
    result = {
        "id": row["id"],
        "name": row["name"],
        "description": row["description"],
        "definition": definition,
        "version": int(row["version"]),
        "status": row["status"],
        "source": row["source"],
        "syncLocked": bool(row["sync_locked"]),
        "createdAt": row["created_at"],
        "updatedAt": row["updated_at"],
    }
    if versions is not None:
        result["versions"] = [
            {
                "version": int(item["version"]),
                "status": item["status"],
                "createdBy": item["created_by"],
                "publishedAt": item["published_at"],
            }
            for item in versions
        ]
    return result


def _detect_capability(content: str, attachments: list[sqlite3.Row], tools_enabled: bool) -> tuple[str, str]:
    lowered = content.lower()
    has_image = any(Path(row["name"]).suffix.lower() in IMAGE_EXTENSIONS for row in attachments)
    if has_image or any(word in lowered for word in ("图片", "截图", "看图", "识图", "image")):
        return "multimodal", "包含图片或视觉理解任务"
    if any(word in lowered for word in ("代码", "编程", "bug", "报错", "函数", "sql", "python", "flutter", "dart", "javascript")):
        return "code", "识别为代码与工程任务"
    if tools_enabled and any(word in lowered for word in ("打开", "点击", "执行", "操作电脑", "浏览器", "文件", "自动完成")):
        return "computer_action", "任务需要获准的工具或电脑操作"
    if attachments:
        return "file_analysis", "包含待分析文件"
    return "funsclaw_chat", "通用智能对话"


def _route_model(models: list[dict[str, str]], capability: str) -> tuple[dict[str, str], str]:
    if not models:
        raise HTTPException(503, "平台暂未配置可用AI模型")
    preferred = {
        "multimodal": ("multimodal", "doubao", "volc", "vision"),
        "image_generation": ("image_generation", "doubao", "volc"),
        "code": ("code", "deepseek", "coder", "v4"),
        "computer_action": ("tool", "deepseek", "doubao"),
        "long_task": ("long_task", "deepseek"),
        "file_analysis": ("file_analysis", "doubao", "deepseek"),
    }.get(capability, ("doubao", "volc", "general"))
    for model in models:
        haystack = f"{model.get('id', '')} {model.get('owned_by', '')}".lower()
        capabilities = _model_capabilities(model)
        if any(marker in capabilities or marker in haystack for marker in preferred):
            return model, f"按{capability}能力自动选择"
    return models[0], "使用企业当前可用的默认模型"


@app.get("/v1/agents")
async def agents(principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _, allowed = await _prepare_workspace(principal)
    return {"agents": [_agent_public_json(row) for row in allowed], "scopes": principal["scopes"]}


@app.get("/v1/models")
async def models(_: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    configured = _configured_models()
    if configured:
        return {"models": [_model_json(item) for item in configured]}
    data = await _funs_get("/v1/models")
    return {"models": data.get("data", data.get("models", []))}


@app.get("/v1/tools")
async def tools(_: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    if not _has_scope(_, "tools"):
        return {"tools": []}
    try:
        data = await _funs_get("/auth/tools")
        return {"tools": data.get("tools", data.get("data", []))}
    except (HTTPException, httpx.HTTPError):
        return {"tools": []}


@app.get("/v1/workspace/bootstrap")
async def workspace_bootstrap(principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _, allowed = await _prepare_workspace(principal)
    models = [_model_json(item) for item in _configured_models()]
    with connect_db() as db:
        wallet = _wallet_json(db, principal["tenant_id"])
        user_policy = _user_policy_json(db, principal)
        user_usage = _user_usage_totals(db, principal["tenant_id"], principal["user_id"])
    tools_data: list[Any] = []
    if _has_scope(principal, "tools") and user_policy["toolsEnabled"]:
        try:
            raw = await _funs_get("/auth/tools")
            tools_data = raw.get("tools", raw.get("data", []))
        except (HTTPException, httpx.HTTPError):
            tools_data = []
    return {
        "agents": [_agent_public_json(row) for row in allowed],
        "models": models,
        "tools": tools_data,
        "wallet": wallet,
        "userPolicy": user_policy,
        "userUsage": user_usage,
        "scopes": principal["scopes"],
        "policy": {
            "automaticModelRouting": True,
            "modelSelectionVisible": False,
            "toolsToggleVisible": _has_scope(principal, "tools") and user_policy["toolsEnabled"],
            "conversationAuditNotice": "企业AI会话可由获得审计权限的管理员依规查看。",
            "providerKeysExposed": False,
        },
    }


@app.get("/v1/platform/agents")
async def platform_agents(
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    with connect_db() as db:
        rows = db.execute("SELECT * FROM platform_agents ORDER BY updated_at DESC").fetchall()
        result = []
        for row in rows:
            versions = db.execute(
                "SELECT * FROM agent_versions WHERE agent_id=? ORDER BY version DESC",
                (row["id"],),
            ).fetchall()
            result.append(_platform_agent_json(row, versions))
    return {"agents": result}


@app.get("/v1/platform/providers")
async def platform_providers(
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    with connect_db() as db:
        rows = db.execute(
            "SELECT * FROM ai_model_providers ORDER BY provider_name,id"
        ).fetchall()
    return {"providers": [_provider_public_json(row) for row in rows]}


@app.post("/v1/platform/providers")
async def save_platform_provider(
    body: PlatformProviderRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    parsed = urlparse(body.baseUrl.strip())
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise HTTPException(422, "Base URL 必须是有效的 HTTP 或 HTTPS 地址")
    if parsed.scheme != "https" and not _is_loopback(parsed.hostname):
        raise HTTPException(422, "非本机模型供应商必须使用 HTTPS")
    model_ids = list(dict.fromkeys(_text(item) for item in body.modelIds if _text(item)))
    timestamp = now_iso()
    with connect_db() as db:
        existing = db.execute(
            "SELECT * FROM ai_model_providers WHERE id=?", (body.id,)
        ).fetchone()
        cipher = existing["api_key_cipher"] if existing is not None else ""
        mask = existing["api_key_mask"] if existing is not None else ""
        if body.apiKey.strip():
            cipher = _encrypt_provider_key(body.apiKey.strip())
            mask = _mask_provider_key(body.apiKey.strip())
        if body.enabled and (not cipher or not model_ids):
            raise HTTPException(422, "启用供应商前必须配置 API Key 和至少一个模型")
        db.execute(
            """INSERT INTO ai_model_providers
               (id,provider_name,base_url,model_ids_json,api_key_cipher,api_key_mask,
                enabled,pricing_json,updated_by,updated_at)
               VALUES (?,?,?,?,?,?,?,?,?,?)
               ON CONFLICT(id) DO UPDATE SET provider_name=excluded.provider_name,
               base_url=excluded.base_url,model_ids_json=excluded.model_ids_json,
               api_key_cipher=excluded.api_key_cipher,api_key_mask=excluded.api_key_mask,
               enabled=excluded.enabled,pricing_json=excluded.pricing_json,
               updated_by=excluded.updated_by,updated_at=excluded.updated_at""",
            (
                body.id, body.providerName.strip(), body.baseUrl.strip().rstrip("/"),
                json.dumps(model_ids, ensure_ascii=False), cipher, mask,
                int(body.enabled), json.dumps(body.pricing, ensure_ascii=False),
                principal["user_id"], timestamp,
            ),
        )
        stored = db.execute(
            "SELECT * FROM ai_model_providers WHERE id=?", (body.id,)
        ).fetchone()
    return {"provider": _provider_public_json(stored)}


@app.post("/v1/platform/providers/{provider_id}/test")
async def test_platform_provider(
    provider_id: str,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    with connect_db() as db:
        row = db.execute(
            "SELECT * FROM ai_model_providers WHERE id=?", (provider_id,)
        ).fetchone()
    if row is None:
        raise HTTPException(404, "模型供应商不存在")
    return await _test_provider_connection(row)


@app.post("/v1/platform/agents")
async def create_platform_agent(
    body: PlatformAgentRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    agent_id = body.id.strip() or f"agent-{uuid.uuid4().hex[:16]}"
    if len(agent_id) > 120 or any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for char in agent_id):
        raise HTTPException(422, "智能体 ID 只能使用字母、数字、横线和下划线")
    timestamp = now_iso()
    definition_json = json.dumps(_platform_definition(body), ensure_ascii=False, sort_keys=True)
    with connect_db() as db:
        if db.execute("SELECT 1 FROM platform_agents WHERE id=?", (agent_id,)).fetchone():
            raise HTTPException(409, "智能体 ID 已存在")
        db.execute(
            """INSERT INTO platform_agents
               (id,name,description,definition_json,version,status,source,sync_locked,created_at,updated_at)
               VALUES (?,?,?,?,1,'draft','yuanxu-cloud',1,?,?)""",
            (agent_id, body.name.strip(), body.description.strip(), definition_json, timestamp, timestamp),
        )
        db.execute(
            """INSERT INTO agent_versions
               (id,agent_id,version,definition_json,status,created_by,published_at)
               VALUES (?,?,1,?,'draft',?,?)""",
            (uuid.uuid4().hex, agent_id, definition_json, principal["user_id"], timestamp),
        )
        row = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
    return {"agent": _platform_agent_json(row)}


@app.put("/v1/platform/agents/{agent_id}")
async def update_platform_agent(
    agent_id: str,
    body: PlatformAgentRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    timestamp = now_iso()
    definition_json = json.dumps(_platform_definition(body), ensure_ascii=False, sort_keys=True)
    with connect_db() as db:
        existing = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
        if existing is None:
            raise HTTPException(404, "智能体不存在")
        version = int(existing["version"])
        if existing["status"] != "draft":
            version += 1
            db.execute(
                """INSERT INTO agent_versions
                   (id,agent_id,version,definition_json,status,created_by,published_at)
                   VALUES (?,?,?,?, 'draft',?,?)""",
                (uuid.uuid4().hex, agent_id, version, definition_json, principal["user_id"], timestamp),
            )
        else:
            db.execute(
                """UPDATE agent_versions SET definition_json=?,created_by=?,published_at=?
                   WHERE agent_id=? AND version=?""",
                (definition_json, principal["user_id"], timestamp, agent_id, version),
            )
        db.execute(
            """UPDATE platform_agents SET name=?,description=?,definition_json=?,version=?,
               status='draft',source='yuanxu-cloud',sync_locked=1,updated_at=? WHERE id=?""",
            (body.name.strip(), body.description.strip(), definition_json, version, timestamp, agent_id),
        )
        row = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
    return {"agent": _platform_agent_json(row)}


@app.post("/v1/platform/agents/{agent_id}/test")
async def test_platform_agent(
    agent_id: str,
    body: PlatformAgentTestRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    with connect_db() as db:
        row = db.execute(
            "SELECT * FROM platform_agents WHERE id=?", (agent_id,)
        ).fetchone()
    if row is None:
        raise HTTPException(404, "智能体不存在")
    return await _test_platform_agent_definition(row, body.prompt)


@app.post("/v1/platform/agents/{agent_id}/publish")
async def publish_platform_agent(
    agent_id: str,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    timestamp = now_iso()
    with connect_db() as db:
        row = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
        if row is None:
            raise HTTPException(404, "智能体不存在")
        db.execute(
            "UPDATE platform_agents SET status='published',sync_locked=1,updated_at=? WHERE id=?",
            (timestamp, agent_id),
        )
        db.execute(
            """UPDATE agent_versions SET status='published',created_by=?,published_at=?
               WHERE agent_id=? AND version=?""",
            (principal["user_id"], timestamp, agent_id, int(row["version"])),
        )
        published = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
    return {"agent": _platform_agent_json(published)}


@app.post("/v1/platform/agents/{agent_id}/rollback")
async def rollback_platform_agent(
    agent_id: str,
    body: PlatformAgentRollbackRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    timestamp = now_iso()
    with connect_db() as db:
        current = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
        source = db.execute(
            "SELECT * FROM agent_versions WHERE agent_id=? AND version=?",
            (agent_id, body.version),
        ).fetchone()
        if current is None or source is None:
            raise HTTPException(404, "找不到指定智能体版本")
        definition = json.loads(source["definition_json"] or "{}")
        new_version = int(current["version"]) + 1
        definition_json = source["definition_json"]
        db.execute(
            """INSERT INTO agent_versions
               (id,agent_id,version,definition_json,status,created_by,published_at)
               VALUES (?,?,?,?,'draft',?,?)""",
            (uuid.uuid4().hex, agent_id, new_version, definition_json, principal["user_id"], timestamp),
        )
        db.execute(
            """UPDATE platform_agents SET name=?,description=?,definition_json=?,version=?,
               status='draft',source='yuanxu-cloud',sync_locked=1,updated_at=? WHERE id=?""",
            (
                _text(definition.get("name"), current["name"]),
                _text(definition.get("description"), current["description"]),
                definition_json, new_version, timestamp, agent_id,
            ),
        )
        rolled_back = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
    return {"agent": _platform_agent_json(rolled_back), "rolledBackFrom": body.version}


@app.post("/v1/platform/agents/{agent_id}/disable")
async def disable_platform_agent(
    agent_id: str,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage")
    with connect_db() as db:
        row = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
        if row is None:
            raise HTTPException(404, "智能体不存在")
        db.execute(
            "UPDATE platform_agents SET status='disabled',sync_locked=1,updated_at=? WHERE id=?",
            (now_iso(), agent_id),
        )
        disabled = db.execute("SELECT * FROM platform_agents WHERE id=?", (agent_id,)).fetchone()
    return {"agent": _platform_agent_json(disabled)}


@app.post("/v1/admin/wallet/top-up")
async def wallet_top_up(body: WalletTopUpRequest, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _require_scope(principal, "ai_credit_manage")
    tenant_id = body.tenantId.strip() or principal["tenant_id"]
    if tenant_id != principal["tenant_id"] and not _has_scope(principal, "ai_platform_manage"):
        raise HTTPException(403, "企业管理员只能维护本企业算力")
    with connect_db() as db:
        _ensure_tenant_defaults(db, tenant_id, [])
        db.execute(
            "UPDATE enterprise_wallets SET total_credits=total_credits+?,updated_at=? WHERE tenant_id=?",
            (body.amount, now_iso(), tenant_id),
        )
        wallet = _wallet_json(db, tenant_id)
        db.execute(
            "INSERT INTO credit_ledger VALUES (?,?,?,?,?,?,?,?,?)",
            (
                uuid.uuid4().hex, tenant_id, "purchase", body.amount,
                wallet["remaining"], principal["user_id"], "", body.note.strip(), now_iso(),
            ),
        )
    return {"wallet": wallet}


@app.post("/v1/admin/recharge-requests")
async def create_recharge_request(
    body: RechargeRequestCreate,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_credit_request", "ai_credit_manage")
    timestamp = now_iso()
    request_id = uuid.uuid4().hex
    with connect_db() as db:
        _ensure_tenant_defaults(db, principal["tenant_id"], [])
        db.execute(
            """INSERT INTO credit_recharge_requests
               (id,tenant_id,amount,note,status,requested_by,created_at)
               VALUES (?,?,?,?,?,?,?)""",
            (
                request_id, principal["tenant_id"], body.amount,
                body.note.strip(), "pending", principal["user_id"], timestamp,
            ),
        )
        stored = db.execute(
            "SELECT * FROM credit_recharge_requests WHERE id=?", (request_id,)
        ).fetchone()
    return {"request": dict(stored)}


@app.get("/v1/admin/recharge-requests")
async def tenant_recharge_requests(
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_credit_request", "ai_credit_manage")
    with connect_db() as db:
        rows = db.execute(
            """SELECT * FROM credit_recharge_requests WHERE tenant_id=?
               ORDER BY created_at DESC""",
            (principal["tenant_id"],),
        ).fetchall()
    return {"requests": [dict(row) for row in rows]}


@app.get("/v1/platform/recharge-requests")
async def platform_recharge_requests(
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage", "ai_credit_manage")
    with connect_db() as db:
        rows = db.execute(
            "SELECT * FROM credit_recharge_requests ORDER BY created_at DESC"
        ).fetchall()
    return {"requests": [dict(row) for row in rows]}


@app.post("/v1/platform/recharge-requests/{request_id}/review")
async def review_recharge_request(
    request_id: str,
    body: RechargeRequestReview,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_platform_manage", "ai_credit_manage")
    timestamp = now_iso()
    with connect_db() as db:
        stored = db.execute(
            "SELECT * FROM credit_recharge_requests WHERE id=?", (request_id,)
        ).fetchone()
        if stored is None:
            raise HTTPException(404, "充值申请不存在")
        if stored["status"] != "pending":
            raise HTTPException(409, "充值申请已经处理，请勿重复审批")
        next_status = "approved" if body.approved else "rejected"
        if body.approved:
            _ensure_tenant_defaults(db, stored["tenant_id"], [])
            db.execute(
                """UPDATE enterprise_wallets SET
                   total_credits=total_credits+?,updated_at=? WHERE tenant_id=?""",
                (stored["amount"], timestamp, stored["tenant_id"]),
            )
            wallet = _wallet_json(db, stored["tenant_id"])
            db.execute(
                "INSERT INTO credit_ledger VALUES (?,?,?,?,?,?,?,?,?)",
                (
                    uuid.uuid4().hex, stored["tenant_id"], "purchase",
                    stored["amount"], wallet["remaining"], principal["user_id"],
                    request_id, body.note.strip() or stored["note"], timestamp,
                ),
            )
        else:
            wallet = _wallet_json(db, stored["tenant_id"])
        db.execute(
            """UPDATE credit_recharge_requests SET status=?,reviewed_by=?,
               review_note=?,reviewed_at=? WHERE id=?""",
            (
                next_status, principal["user_id"], body.note.strip(),
                timestamp, request_id,
            ),
        )
        reviewed = db.execute(
            "SELECT * FROM credit_recharge_requests WHERE id=?", (request_id,)
        ).fetchone()
    return {"request": dict(reviewed), "wallet": wallet}


@app.post("/v1/admin/enterprise-agent-grants")
async def enterprise_agent_grant(body: EnterpriseAgentGrantRequest, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _require_scope(principal, "ai_agent_manage")
    tenant_id = body.tenantId.strip() or principal["tenant_id"]
    if tenant_id != principal["tenant_id"] and not _has_scope(principal, "ai_platform_manage"):
        raise HTTPException(403, "企业管理员只能维护本企业智能体授权")
    timestamp = now_iso()
    with connect_db() as db:
        agent = db.execute(
            "SELECT status FROM platform_agents WHERE id=?", (body.agentId,)
        ).fetchone()
        if agent is None or (body.enabled and agent["status"] != "published"):
            raise HTTPException(404, "智能体不存在或尚未发布")
        db.execute(
            """INSERT INTO enterprise_agent_grants
               (tenant_id,agent_id,enabled,credit_multiplier,granted_by,created_at,updated_at)
               VALUES (?,?,?,?,?,?,?) ON CONFLICT(tenant_id,agent_id) DO UPDATE SET
               enabled=excluded.enabled,credit_multiplier=excluded.credit_multiplier,
               granted_by=excluded.granted_by,updated_at=excluded.updated_at""",
            (tenant_id, body.agentId, int(body.enabled), body.creditMultiplier, principal["user_id"], timestamp, timestamp),
        )
    return {"ok": True}


@app.post("/v1/admin/enterprise-agent-customizations")
async def enterprise_agent_customization(
    body: EnterpriseAgentCustomizationRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_agent_manage")
    timestamp = now_iso()
    with connect_db() as db:
        grant = db.execute(
            """SELECT 1 FROM enterprise_agent_grants
               WHERE tenant_id=? AND agent_id=? AND enabled=1""",
            (principal["tenant_id"], body.agentId),
        ).fetchone()
        if grant is None:
            raise HTTPException(409, "企业尚未获授此智能体")
        db.execute(
            """INSERT INTO enterprise_agent_customizations
               (tenant_id,agent_id,supplement_prompt,updated_by,updated_at)
               VALUES (?,?,?,?,?) ON CONFLICT(tenant_id,agent_id) DO UPDATE SET
               supplement_prompt=excluded.supplement_prompt,
               updated_by=excluded.updated_by,updated_at=excluded.updated_at""",
            (
                principal["tenant_id"], body.agentId,
                body.supplementPrompt.strip(), principal["user_id"], timestamp,
            ),
        )
    return {"ok": True, "agentId": body.agentId}


@app.post("/v1/admin/user-agent-grants")
async def user_agent_grant(body: UserAgentGrantRequest, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _require_scope(principal, "ai_agent_manage")
    await _prepare_workspace(principal)
    timestamp = now_iso()
    with connect_db() as db:
        enterprise = db.execute(
            "SELECT enabled FROM enterprise_agent_grants WHERE tenant_id=? AND agent_id=?",
            (principal["tenant_id"], body.agentId),
        ).fetchone()
        if enterprise is None or not enterprise["enabled"]:
            raise HTTPException(409, "企业尚未获授此智能体")
        db.execute(
            """INSERT INTO user_agent_grants
               (tenant_id,user_id,agent_id,enabled,granted_by,created_at,updated_at)
               VALUES (?,?,?,?,?,?,?) ON CONFLICT(tenant_id,user_id,agent_id) DO UPDATE SET
               enabled=excluded.enabled,granted_by=excluded.granted_by,updated_at=excluded.updated_at""",
            (principal["tenant_id"], body.userId, body.agentId, int(body.enabled), principal["user_id"], timestamp, timestamp),
        )
    return {"ok": True}


@app.post("/v1/admin/subject-agent-grants")
async def subject_agent_grant(
    body: SubjectAgentGrantRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    """Grant an enterprise agent to a department or position.

    A concrete employee grant remains the highest-priority override. This
    keeps workforce allocation under human enterprise-admin control while
    avoiding duplicated per-employee setup for normal cases.
    """
    _require_scope(principal, "ai_agent_manage")
    await _prepare_workspace(principal)
    subject_id = body.subjectId.strip()
    if not subject_id:
        raise HTTPException(400, "请选择部门或岗位")
    timestamp = now_iso()
    with connect_db() as db:
        enterprise = db.execute(
            "SELECT enabled FROM enterprise_agent_grants WHERE tenant_id=? AND agent_id=?",
            (principal["tenant_id"], body.agentId),
        ).fetchone()
        if enterprise is None or not enterprise["enabled"]:
            raise HTTPException(409, "企业尚未获授此智能体")
        db.execute(
            """INSERT INTO subject_agent_grants
               (tenant_id,subject_type,subject_id,agent_id,enabled,granted_by,created_at,updated_at)
               VALUES (?,?,?,?,?,?,?,?)
               ON CONFLICT(tenant_id,subject_type,subject_id,agent_id) DO UPDATE SET
               enabled=excluded.enabled,granted_by=excluded.granted_by,
               updated_at=excluded.updated_at""",
            (
                principal["tenant_id"], body.subjectType, subject_id,
                body.agentId, int(body.enabled), principal["user_id"],
                timestamp, timestamp,
            ),
        )
    return {"ok": True}


@app.post("/v1/admin/user-ai-policies")
async def user_ai_policy(
    body: UserAiPolicyRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_agent_manage", "ai_credit_manage")
    timestamp = now_iso()
    with connect_db() as db:
        db.execute(
            """INSERT INTO user_ai_policies
               (tenant_id,user_id,enabled,daily_limit,monthly_limit,tools_enabled,
                computer_enabled,files_enabled,updated_by,updated_at)
               VALUES (?,?,?,?,?,?,?,?,?,?)
               ON CONFLICT(tenant_id,user_id) DO UPDATE SET
               enabled=excluded.enabled,daily_limit=excluded.daily_limit,
               monthly_limit=excluded.monthly_limit,tools_enabled=excluded.tools_enabled,
               computer_enabled=excluded.computer_enabled,files_enabled=excluded.files_enabled,
               updated_by=excluded.updated_by,updated_at=excluded.updated_at""",
            (
                principal["tenant_id"], body.userId, int(body.enabled),
                body.dailyLimit, body.monthlyLimit, int(body.toolsEnabled),
                int(body.computerEnabled), int(body.filesEnabled),
                principal["user_id"], timestamp,
            ),
        )
        stored = _user_policy_json(
            db,
            {
                "tenant_id": principal["tenant_id"],
                "user_id": body.userId,
                "scopes": [],
            },
        )
    return {"policy": stored}


@app.get("/v1/admin/usage")
async def usage_records(limit: int = 100, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _require_scope(principal, "ai_usage_view", "ai_credit_manage")
    safe_limit = max(1, min(limit, 500))
    with connect_db() as db:
        rows = db.execute(
            "SELECT * FROM ai_usage_records WHERE tenant_id=? ORDER BY created_at DESC LIMIT ?",
            (principal["tenant_id"], safe_limit),
        ).fetchall()
        wallet = _wallet_json(db, principal["tenant_id"])
        summary = _usage_summary(db, principal["tenant_id"])
    return {"usage": [dict(row) for row in rows], "wallet": wallet, "summary": summary}


@app.get("/v1/admin/conversation-audit")
async def conversation_audit(limit: int = 100, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _require_scope(principal, "ai_conversation_audit")
    safe_limit = max(1, min(limit, 500))
    with connect_db() as db:
        rows = db.execute(
            "SELECT * FROM conversation_audit_events WHERE tenant_id=? ORDER BY created_at DESC LIMIT ?",
            (principal["tenant_id"], safe_limit),
        ).fetchall()
    return {"events": [dict(row) for row in rows]}


@app.get("/v1/admin/bootstrap")
async def ai_admin_bootstrap(principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    _require_scope(
        principal, "ai_usage_view", "ai_agent_manage", "ai_credit_manage",
        "ai_conversation_audit",
    )
    await _prepare_workspace(principal)
    with connect_db() as db:
        agents = db.execute(
            """SELECT p.id,p.name,p.description,p.version,p.status,
               COALESCE(g.enabled,0) AS enterprise_enabled,
               COALESCE(g.credit_multiplier,1.0) AS credit_multiplier,
               COALESCE(c.supplement_prompt,'') AS supplement_prompt
               FROM platform_agents p LEFT JOIN enterprise_agent_grants g
               ON g.agent_id=p.id AND g.tenant_id=?
               LEFT JOIN enterprise_agent_customizations c
               ON c.agent_id=p.id AND c.tenant_id=?
               WHERE p.status='published' ORDER BY p.name""",
            (principal["tenant_id"], principal["tenant_id"]),
        ).fetchall()
        usage = []
        if _has_scope(principal, "ai_usage_view", "ai_credit_manage"):
            usage = db.execute(
                "SELECT * FROM ai_usage_records WHERE tenant_id=? ORDER BY created_at DESC LIMIT 100",
                (principal["tenant_id"],),
            ).fetchall()
        events = []
        if _has_scope(principal, "ai_conversation_audit"):
            events = db.execute(
                "SELECT * FROM conversation_audit_events WHERE tenant_id=? ORDER BY created_at DESC LIMIT 100",
                (principal["tenant_id"],),
            ).fetchall()
        ledger = db.execute(
            "SELECT * FROM credit_ledger WHERE tenant_id=? ORDER BY created_at DESC LIMIT 100",
            (principal["tenant_id"],),
        ).fetchall()
        user_grants = db.execute(
            """SELECT user_id,agent_id,enabled,granted_by,updated_at
               FROM user_agent_grants WHERE tenant_id=? ORDER BY user_id,agent_id""",
            (principal["tenant_id"],),
        ).fetchall()
        subject_grants = db.execute(
            """SELECT subject_type,subject_id,agent_id,enabled,granted_by,updated_at
               FROM subject_agent_grants WHERE tenant_id=?
               ORDER BY subject_type,subject_id,agent_id""",
            (principal["tenant_id"],),
        ).fetchall()
        user_policies = db.execute(
            """SELECT tenant_id,user_id,enabled,daily_limit,monthly_limit,
               tools_enabled,computer_enabled,files_enabled,updated_by,updated_at
               FROM user_ai_policies WHERE tenant_id=? ORDER BY user_id""",
            (principal["tenant_id"],),
        ).fetchall()
        recharge_requests = db.execute(
            """SELECT * FROM credit_recharge_requests WHERE tenant_id=?
               ORDER BY created_at DESC LIMIT 100""",
            (principal["tenant_id"],),
        ).fetchall()
        wallet = _wallet_json(db, principal["tenant_id"])
        usage_summary = _usage_summary(db, principal["tenant_id"])
    return {
        "wallet": wallet,
        "agents": [dict(row) for row in agents],
        "usage": [dict(row) for row in usage],
        "usageSummary": usage_summary,
        "ledger": [dict(row) for row in ledger],
        "userGrants": [dict(row) for row in user_grants],
        "subjectGrants": [dict(row) for row in subject_grants],
        "userPolicies": [dict(row) for row in user_policies],
        "rechargeRequests": [dict(row) for row in recharge_requests],
        "auditEvents": [dict(row) for row in events],
        "scopes": principal["scopes"],
    }


@app.get("/v1/admin/audited-conversations")
async def audited_conversations(
    limit: int = 100,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_conversation_audit")
    safe_limit = max(1, min(limit, 500))
    with connect_db() as db:
        rows = db.execute(
            """SELECT c.id,c.user_id,c.title,c.agent_id,c.model_id,c.created_at,c.updated_at,
               COUNT(m.id) AS message_count
               FROM conversations c LEFT JOIN messages m ON m.conversation_id=c.id
               WHERE c.tenant_id=? GROUP BY c.id
               ORDER BY c.updated_at DESC LIMIT ?""",
            (principal["tenant_id"], safe_limit),
        ).fetchall()
    return {"conversations": [dict(row) for row in rows]}


@app.get("/v1/admin/audited-conversations/{conversation_id}")
async def audited_conversation_detail(
    conversation_id: str,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    _require_scope(principal, "ai_conversation_audit")
    _require_scope(principal, "ai_sensitive_content_view")
    with connect_db() as db:
        conversation = db.execute(
            "SELECT * FROM conversations WHERE id=? AND tenant_id=?",
            (conversation_id, principal["tenant_id"]),
        ).fetchone()
        if conversation is None:
            raise HTTPException(404, "会话不存在")
        messages = db.execute(
            "SELECT * FROM messages WHERE conversation_id=? ORDER BY created_at,id",
            (conversation_id,),
        ).fetchall()
    _audit_event(
        principal,
        conversation_id,
        "conversation_content_viewed",
        f"审计查看员工 {conversation['user_id']} 的企业AI会话",
        {"conversationOwner": conversation["user_id"], "messageCount": len(messages)},
    )
    return {
        "conversation": {
            **dict(conversation),
            "messages": [_message_json(row) for row in messages],
        }
    }


def _owner_where(principal: dict[str, Any]) -> tuple[str, str]:
    return principal["tenant_id"], principal["user_id"]


def _message_json(row: sqlite3.Row) -> dict[str, Any]:
    return {
        "id": row["id"], "role": row["role"], "content": row["content"],
        "attachments": json.loads(row["attachments_json"]),
        "toolStates": json.loads(row["tool_states_json"]), "createdAt": row["created_at"],
        "agentId": row["agent_id"], "agentVersion": int(row["agent_version"]),
        "modelId": row["model_id"],
    }


def _conversation_json(db: sqlite3.Connection, row: sqlite3.Row, include_messages: bool = True) -> dict[str, Any]:
    messages: list[dict[str, Any]] = []
    if include_messages:
        messages = [
            _message_json(item) for item in db.execute(
                "SELECT * FROM messages WHERE conversation_id=? ORDER BY created_at,id", (row["id"],)
            ).fetchall()
        ]
    return {
        "id": row["id"], "title": row["title"], "agentId": row["agent_id"],
        "modelId": row["model_id"], "createdAt": row["created_at"],
        "updatedAt": row["updated_at"], "messages": messages,
        "pinned": bool(row["pinned"]), "archived": bool(row["archived"]),
    }


def _owned_conversation(db: sqlite3.Connection, conversation_id: str, principal: dict[str, Any]) -> sqlite3.Row:
    row = db.execute(
        "SELECT * FROM conversations WHERE id=? AND tenant_id=? AND user_id=?",
        (conversation_id, *_owner_where(principal)),
    ).fetchone()
    if row is None:
        raise HTTPException(404, "会话不存在")
    return row


@app.get("/v1/conversations")
async def list_conversations(principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    with connect_db() as db:
        rows = db.execute(
            """SELECT * FROM conversations WHERE tenant_id=? AND user_id=?
               ORDER BY pinned DESC,updated_at DESC""",
            _owner_where(principal),
        ).fetchall()
        return {"conversations": [_conversation_json(db, row, False) for row in rows]}


@app.post("/v1/conversations")
async def create_conversation(body: ConversationCreate, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    conversation_id = uuid.uuid4().hex
    created_at = now_iso()
    with connect_db() as db:
        db.execute(
            """INSERT INTO conversations
               (id,user_id,tenant_id,title,agent_id,model_id,created_at,updated_at,pinned,archived)
               VALUES (?,?,?,?,?,?,?,?,0,0)""",
            (conversation_id, principal["user_id"], principal["tenant_id"], _text(body.title, "新会话"), body.agentId, body.modelId, created_at, created_at),
        )
        row = _owned_conversation(db, conversation_id, principal)
        return {"conversation": _conversation_json(db, row)}


@app.get("/v1/conversations/{conversation_id}")
async def get_conversation(conversation_id: str, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    with connect_db() as db:
        row = _owned_conversation(db, conversation_id, principal)
        return {"conversation": _conversation_json(db, row)}


@app.put("/v1/conversations/{conversation_id}")
async def rename_conversation(conversation_id: str, body: ConversationUpdate, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    with connect_db() as db:
        _owned_conversation(db, conversation_id, principal)
        db.execute("UPDATE conversations SET title=?,updated_at=? WHERE id=?", (body.title.strip(), now_iso(), conversation_id))
        row = _owned_conversation(db, conversation_id, principal)
        return {"conversation": _conversation_json(db, row)}


@app.put("/v1/conversations/{conversation_id}/state")
async def update_conversation_state(
    conversation_id: str,
    body: ConversationStateUpdate,
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    if body.pinned is None and body.archived is None:
        raise HTTPException(400, "没有需要更新的会话状态")
    with connect_db() as db:
        current = _owned_conversation(db, conversation_id, principal)
        pinned = int(body.pinned) if body.pinned is not None else int(current["pinned"])
        archived = int(body.archived) if body.archived is not None else int(current["archived"])
        db.execute(
            "UPDATE conversations SET pinned=?,archived=?,updated_at=? WHERE id=?",
            (pinned, archived, now_iso(), conversation_id),
        )
        row = _owned_conversation(db, conversation_id, principal)
        return {"conversation": _conversation_json(db, row)}


@app.delete("/v1/conversations/{conversation_id}")
async def delete_conversation(conversation_id: str, principal: dict[str, Any] = Depends(current_principal)) -> dict[str, Any]:
    with connect_db() as db:
        _owned_conversation(db, conversation_id, principal)
        db.execute("DELETE FROM messages WHERE conversation_id=?", (conversation_id,))
        db.execute("DELETE FROM attachments WHERE conversation_id=?", (conversation_id,))
        db.execute("DELETE FROM conversations WHERE id=?", (conversation_id,))
    return {"ok": True}


async def _proxy_upload(file: UploadFile, content: bytes) -> dict[str, Any]:
    suffix = Path(file.filename or "file.bin").suffix.lower()
    endpoint = "/v1/uploads/image" if suffix in IMAGE_EXTENSIONS else "/v1/uploads/document"
    async with httpx.AsyncClient(timeout=120, trust_env=_funs_trust_env()) as client:
        response = await client.post(
            f"{FUNSCLAW_URL}{endpoint}", headers=_funs_headers(),
            files={"file": (file.filename, content, file.content_type or "application/octet-stream")},
        )
    if response.status_code >= 400:
        raise HTTPException(502, f"FunsClaw 文件处理失败（{response.status_code}）")
    return response.json()


@app.post("/v1/conversations/{conversation_id}/files")
async def upload_files(
    conversation_id: str,
    files: list[UploadFile] = File(...),
    principal: dict[str, Any] = Depends(current_principal),
) -> dict[str, Any]:
    uploaded: list[dict[str, Any]] = []
    with connect_db() as db:
        _owned_conversation(db, conversation_id, principal)
        policy = _user_policy_json(db, principal)
        if not policy["enabled"]:
            raise HTTPException(403, "企业已暂停当前账号的小元AI使用权限")
        if not policy["filesEnabled"]:
            raise HTTPException(403, "企业尚未给当前账号开通AI文件权限")
    for file in files:
        content = await file.read()
        if len(content) > 50 * 1024 * 1024:
            raise HTTPException(413, "单个文件不能超过50MB")
        attachment_id = uuid.uuid4().hex
        safe_name = Path(file.filename or "附件").name
        local_path = UPLOAD_DIR / f"{attachment_id}{Path(safe_name).suffix.lower()}"
        local_path.write_bytes(content)
        remote = await _proxy_upload(file, content)
        item = {
            "id": attachment_id, "name": safe_name, "size": len(content),
            "contentType": file.content_type or mimetypes.guess_type(safe_name)[0] or "application/octet-stream",
            "status": "ready",
        }
        with connect_db() as db:
            db.execute(
                "INSERT INTO attachments VALUES (?,?,?,?,?,?,?,?,?,?)",
                (attachment_id, conversation_id, principal["user_id"], principal["tenant_id"], safe_name, item["contentType"], len(content), str(local_path), json.dumps(remote, ensure_ascii=False), now_iso()),
            )
        uploaded.append(item)
    return {"files": uploaded}


def _sse(data: dict[str, Any] | str) -> str:
    value = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False)
    return f"data: {value}\n\n"


def _extract_openai_delta(data: dict[str, Any]) -> tuple[str, list[str]]:
    content = ""
    tools: list[str] = []
    for choice in data.get("choices") or []:
        delta = choice.get("delta") or {}
        content += _text(delta.get("content"))
        for tool in delta.get("tool_calls") or choice.get("tool_calls") or []:
            name = _text((tool.get("function") or {}).get("name"), "工具任务")
            if name not in tools:
                tools.append(name)
    return content, tools


def _remote_attachment_text(row: sqlite3.Row) -> str:
    """Return parsed file text without leaking bridge-local file paths."""
    try:
        remote = json.loads(row["remote_json"] or "{}")
    except ValueError:
        return ""
    candidates: list[Any] = [remote]
    if isinstance(remote, dict):
        candidates.extend(
            remote.get(key) for key in ("data", "result", "document", "file")
        )
    for candidate in candidates:
        if not isinstance(candidate, dict):
            continue
        for key in ("text", "content", "extractedText", "extracted_text", "markdown"):
            value = _text(candidate.get(key))
            if value:
                return value[:12000]
    return ""


def _agent_override(agent_id: str, tenant_id: str = "") -> dict[str, Any] | None:
    if not agent_id:
        return None
    with connect_db() as db:
        row = db.execute(
            "SELECT definition_json FROM platform_agents WHERE id=? AND status='published'",
            (agent_id,),
        ).fetchone()
        customization = db.execute(
            """SELECT supplement_prompt FROM enterprise_agent_customizations
               WHERE tenant_id=? AND agent_id=?""",
            (tenant_id, agent_id),
        ).fetchone() if tenant_id else None
    if row is None:
        return None
    try:
        definition = json.loads(row["definition_json"] or "{}")
        supplement = _text(customization["supplement_prompt"] if customization else "")
        if supplement:
            core_prompt = _text(
                definition.get("role_prompt") or definition.get("system_prompt")
            )
            definition["role_prompt"] = (
                f"{core_prompt}\n\n[企业补充要求]\n{supplement}"
            ).strip()
        return definition
    except ValueError:
        return None


def _reserve_request(db: sqlite3.Connection, principal: dict[str, Any], request_id: str) -> None:
    duplicate = db.execute(
        """SELECT request_id FROM ai_request_reservations WHERE request_id=?
           UNION SELECT request_id FROM ai_usage_records WHERE request_id=?""",
        (request_id, request_id),
    ).fetchone()
    if duplicate is not None:
        raise HTTPException(409, "该AI请求已提交，请勿重复发送")
    wallet = _wallet_json(db, principal["tenant_id"])
    if wallet["status"] != "active":
        raise HTTPException(402, "企业AI算力账户已停用，请联系管理员")
    if wallet["remaining"] < 1:
        raise HTTPException(402, "企业AI算力不足，请联系管理员充值")
    db.execute(
        "UPDATE enterprise_wallets SET frozen_credits=frozen_credits+1,updated_at=? WHERE tenant_id=?",
        (now_iso(), principal["tenant_id"]),
    )
    db.execute(
        "INSERT INTO ai_request_reservations VALUES (?,?,?,?)",
        (request_id, principal["tenant_id"], principal["user_id"], now_iso()),
    )


def _usage_credits(input_tokens: int, output_tokens: int, tool_calls: int, multiplier: float) -> int:
    token_credits = max(1, math.ceil(max(1, input_tokens + output_tokens) / 1000))
    return max(1, math.ceil((token_credits + tool_calls * 2) * multiplier))


def _finish_usage(
    *,
    principal: dict[str, Any],
    request_id: str,
    agent_id: str,
    agent_version: int,
    conversation_id: str,
    message_id: str,
    capability: str,
    model: dict[str, str],
    routing_reason: str,
    input_tokens: int,
    output_tokens: int,
    tool_calls: int,
    duration_ms: int,
    status: str,
    multiplier: float,
) -> int:
    credits = _usage_credits(input_tokens, output_tokens, tool_calls, multiplier) if status == "success" else 0
    with connect_db() as db:
        wallet = _wallet_json(db, principal["tenant_id"])
        charge = min(credits, wallet["remaining"] + 1)
        db.execute(
            """UPDATE enterprise_wallets SET
               frozen_credits=MAX(0,frozen_credits-1),
               used_credits=used_credits+?,updated_at=? WHERE tenant_id=?""",
            (charge, now_iso(), principal["tenant_id"]),
        )
        db.execute(
            """INSERT INTO ai_usage_records
                (id,tenant_id,user_id,agent_id,conversation_id,message_id,capability_type,
                 model_id,provider,input_tokens,output_tokens,tool_calls,computer_actions,
                 task_duration_ms,credits_used,status,request_id,routing_reason,department_id,
                 position_id,created_at,agent_version)
                VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
            (
                uuid.uuid4().hex, principal["tenant_id"], principal["user_id"], agent_id,
                conversation_id, message_id, capability, model.get("id", ""),
                model.get("owned_by", ""), input_tokens, output_tokens, tool_calls,
                tool_calls if capability == "computer_action" else 0, duration_ms, charge,
                status, request_id, routing_reason,
                _text((principal.get("department_ids") or [""])[0]),
                _text((principal.get("position_ids") or [""])[0]),
                now_iso(), agent_version,
            ),
        )
        db.execute("DELETE FROM ai_request_reservations WHERE request_id=?", (request_id,))
        current = _wallet_json(db, principal["tenant_id"])
        db.execute(
            "INSERT INTO credit_ledger VALUES (?,?,?,?,?,?,?,?,?)",
            (
                uuid.uuid4().hex, principal["tenant_id"], "usage" if charge else "release",
                -charge, current["remaining"], principal["user_id"], request_id,
                f"{capability} · {model.get('id', '')} · {status}", now_iso(),
            ),
        )
    return charge


def _audit_event(
    principal: dict[str, Any], conversation_id: str, event_type: str,
    summary: str, metadata: dict[str, Any],
) -> None:
    with connect_db() as db:
        db.execute(
            "INSERT INTO conversation_audit_events VALUES (?,?,?,?,?,?,?,?)",
            (
                uuid.uuid4().hex, principal["tenant_id"], principal["user_id"],
                conversation_id, event_type, summary[:300],
                json.dumps(metadata, ensure_ascii=False), now_iso(),
            ),
        )


@app.post("/v1/conversations/{conversation_id}/messages:stream")
async def stream_message(
    conversation_id: str,
    body: MessageRequest,
    principal: dict[str, Any] = Depends(current_principal),
) -> StreamingResponse:
    request_id = body.requestId.strip() or uuid.uuid4().hex
    configured_models = _configured_models()
    _, allowed_rows = await _prepare_workspace(principal)
    with connect_db() as db:
        conversation = _owned_conversation(db, conversation_id, principal)
        attachments = []
        attachment_rows: list[sqlite3.Row] = []
        if body.attachmentIds:
            placeholders = ",".join("?" for _ in body.attachmentIds)
            attachment_rows = db.execute(
                f"SELECT * FROM attachments WHERE id IN ({placeholders}) AND conversation_id=? AND tenant_id=? AND user_id=?",
                (*body.attachmentIds, conversation_id, principal["tenant_id"], principal["user_id"]),
            ).fetchall()
            attachments = [
                {"id": row["id"], "name": row["name"], "size": row["size"], "contentType": row["content_type"]}
                for row in attachment_rows
            ]
        user_content = body.content.strip()
        if attachments:
            details = []
            for row in attachment_rows:
                parsed_text = _remote_attachment_text(row)
                details.append(
                    f"- {row['name']}（已解析）"
                    + (f"\n{parsed_text}" if parsed_text else "")
                )
            user_content = (
                f"{user_content}\n\n[用户附加了以下文件]\n"
                + "\n".join(details)
            ).strip()
        if not user_content:
            raise HTTPException(400, "请输入内容或添加附件")
        allowed_ids = {row["id"] for row in allowed_rows}
        agent_id = body.agentId or conversation["agent_id"]
        if not agent_id and allowed_rows:
            agent_id = allowed_rows[0]["id"]
        if not agent_id or agent_id not in allowed_ids:
            raise HTTPException(403, "当前账号未获授所选智能体")
        if body.toolsEnabled and not _has_scope(principal, "tools"):
            raise HTTPException(403, "当前账号未获授AI工具权限")
        capability, capability_reason = _detect_capability(
            body.content, attachment_rows, body.toolsEnabled
        )
        _enforce_user_policy(
            db,
            principal,
            tools_enabled=body.toolsEnabled,
            has_files=bool(attachment_rows),
            capability=capability,
        )
        selected_model, model_reason = _route_model(configured_models, capability)
        routing_reason = f"{capability_reason}；{model_reason}"
        grant = db.execute(
            "SELECT credit_multiplier FROM enterprise_agent_grants WHERE tenant_id=? AND agent_id=? AND enabled=1",
            (principal["tenant_id"], agent_id),
        ).fetchone()
        multiplier = float(grant["credit_multiplier"] if grant else 1.0)
        agent_row = next(row for row in allowed_rows if row["id"] == agent_id)
        agent_version = int(agent_row["version"])
        model_id = selected_model["id"]
        _reserve_request(db, principal, request_id)
        user_message_id = uuid.uuid4().hex
        db.execute(
            """INSERT INTO messages
               (id,conversation_id,role,content,attachments_json,tool_states_json,created_at,
                agent_id,agent_version,model_id) VALUES (?,?,?,?,?,?,?,?,?,?)""",
            (user_message_id, conversation_id, "user", user_content,
             json.dumps(attachments, ensure_ascii=False), "[]", now_iso(),
             agent_id, agent_version, model_id),
        )
        current_title = _text(conversation["title"], "新会话")
        next_title = current_title
        if current_title == "新会话":
            title_source = body.content.strip() or (attachments[0]["name"] if attachments else "新会话")
            next_title = title_source.replace("\n", " ")[:30]
        db.execute(
            "UPDATE conversations SET title=?,agent_id=?,model_id=?,updated_at=? WHERE id=?",
            (next_title, agent_id, model_id, now_iso(), conversation_id),
        )
        history = [
            {"role": row["role"], "content": row["content"]}
            for row in db.execute("SELECT role,content FROM messages WHERE conversation_id=? ORDER BY created_at,id", (conversation_id,)).fetchall()
        ]
    _audit_event(
        principal, conversation_id, "message_submitted", body.content,
        {
            "requestId": request_id, "agentId": agent_id, "capabilityType": capability,
            "model": model_id, "provider": selected_model.get("owned_by", ""),
            "toolsEnabled": body.toolsEnabled, "attachmentCount": len(attachment_rows),
        },
    )

    async def generate() -> AsyncIterator[str]:
        assistant_text = ""
        tool_states: list[str] = []
        assistant_id = uuid.uuid4().hex
        input_tokens = max(1, len(json.dumps(history, ensure_ascii=False)) // 4)
        output_tokens = 0
        tool_calls = 0
        started_at = time.monotonic()
        settled = False
        yield _sse({"event": "started"})
        yield _sse({
            "event": "route", "model": model_id,
            "provider": selected_model.get("owned_by", ""),
            "capabilityType": capability, "reason": routing_reason,
        })
        try:
            override = _agent_override(agent_id, principal["tenant_id"])
            payload: dict[str, Any] = {
                "model": model_id or "deepseek-chat", "messages": history,
                "stream": True, "user": f"yuanxu-{principal['tenant_id']}-{principal['user_id']}-{conversation_id}",
            }
            usage_model = next(
                (item for item in _configured_models() if item["id"] == payload["model"]),
                {"id": payload["model"], "owned_by": ""},
            )
            if usage_model.get("owned_by"):
                payload["provider"] = usage_model["owned_by"]
            if override:
                payload["agent_override"] = override
            if not body.toolsEnabled:
                payload["tool_choice"] = "none"

            async def consume_lines(lines: AsyncIterator[str]) -> AsyncIterator[str]:
                nonlocal assistant_text, input_tokens, output_tokens
                nonlocal tool_calls, tool_states
                async for line in lines:
                    if not line.startswith("data:"):
                        continue
                    raw = line[5:].strip()
                    if not raw or raw == "[DONE]":
                        continue
                    try:
                        data = json.loads(raw)
                    except ValueError:
                        continue
                    event_name = _text(data.get("event") or data.get("type"))
                    usage = data.get("usage") if isinstance(data.get("usage"), dict) else {}
                    input_tokens = max(input_tokens, int(usage.get("prompt_tokens", usage.get("input_tokens", 0)) or 0))
                    output_tokens = max(output_tokens, int(usage.get("completion_tokens", usage.get("output_tokens", 0)) or 0))
                    if event_name in {"assistant_delta", "content"}:
                        delta = _text(data.get("delta") or data.get("content"))
                        assistant_text += delta
                        yield _sse({"event": "content", "delta": delta})
                        continue
                    if event_name in {"tool_call", "tool_started"}:
                        name = _text(data.get("name"), "工具任务")
                        tool_states.append(f"{name} · 执行中")
                        tool_calls += 1
                        yield _sse({"event": "tool_started", "name": name})
                        continue
                    if event_name in {"tool_result", "tool_finished"}:
                        name = _text(data.get("name"), "工具任务")
                        tool_states = [item for item in tool_states if not item.startswith(f"{name} ·")]
                        tool_states.append(f"{name} · 已完成")
                        yield _sse({"event": "tool_finished", "name": name})
                        continue
                    delta, tool_names = _extract_openai_delta(data)
                    if delta:
                        assistant_text += delta
                        yield _sse({"event": "content", "delta": delta})
                    for name in tool_names:
                        if not any(item.startswith(f"{name} ·") for item in tool_states):
                            tool_states.append(f"{name} · 执行中")
                            yield _sse({"event": "tool_started", "name": name})

            cloud_provider = _cloud_provider_for_model(model_id)
            yaml_provider = (
                None if cloud_provider is not None
                else _yaml_provider_for_model(model_id)
            )
            direct_provider = cloud_provider or yaml_provider
            if direct_provider is not None and capability != "computer_action":
                provider_model_id = (
                    model_id
                    if cloud_provider is not None
                    else yaml_provider["model_id"]
                )
                direct_payload = _provider_chat_payload(
                    model_id=provider_model_id,
                    history=history,
                    agent_override=override,
                )
                provider_url = (
                    f"{direct_provider['base_url'].rstrip('/')}/chat/completions"
                )
                provider_key = (
                    _decrypt_provider_key(cloud_provider["api_key_cipher"])
                    if cloud_provider is not None
                    else yaml_provider["api_key"]
                )
                async with httpx.AsyncClient(
                    timeout=httpx.Timeout(180, read=180),
                ) as client:
                    async with client.stream(
                        "POST",
                        provider_url,
                        headers={
                            "Authorization": f"Bearer {provider_key}",
                            "Content-Type": "application/json",
                            "Accept": "text/event-stream",
                        },
                        json=direct_payload,
                    ) as response:
                        if response.status_code >= 400:
                            detail = (await response.aread()).decode(
                                "utf-8", "replace"
                            )[:500]
                            raise RuntimeError(
                                f"模型供应商 {response.status_code}: {detail}"
                            )
                        async for event in consume_lines(response.aiter_lines()):
                            yield event
            else:
                async with httpx.AsyncClient(
                    timeout=httpx.Timeout(180, read=180),
                    trust_env=_funs_trust_env(),
                ) as client:
                    async with client.stream(
                        "POST", f"{FUNSCLAW_URL}/v1/chat/completions",
                        headers={**_funs_headers(), "Content-Type": "application/json", "Accept": "text/event-stream"},
                        json=payload,
                    ) as response:
                        if response.status_code >= 400:
                            detail = (await response.aread()).decode("utf-8", "replace")[:500]
                            raise RuntimeError(f"FunsClaw {response.status_code}: {detail}")
                        async for event in consume_lines(response.aiter_lines()):
                            yield event
            if assistant_text or tool_states:
                with connect_db() as db:
                    db.execute(
                        """INSERT INTO messages
                           (id,conversation_id,role,content,attachments_json,tool_states_json,created_at,
                            agent_id,agent_version,model_id) VALUES (?,?,?,?,?,?,?,?,?,?)""",
                        (assistant_id, conversation_id, "assistant", assistant_text, "[]",
                         json.dumps(tool_states, ensure_ascii=False), now_iso(),
                         agent_id, agent_version, model_id),
                    )
                    db.execute("UPDATE conversations SET updated_at=? WHERE id=?", (now_iso(), conversation_id))
            output_tokens = max(output_tokens, len(assistant_text) // 4)
            credits = _finish_usage(
                principal=principal, request_id=request_id, agent_id=agent_id,
                agent_version=agent_version,
                conversation_id=conversation_id, message_id=assistant_id,
                capability=capability, model=usage_model, routing_reason=routing_reason,
                input_tokens=input_tokens, output_tokens=output_tokens, tool_calls=tool_calls,
                duration_ms=int((time.monotonic() - started_at) * 1000), status="success",
                multiplier=multiplier,
            )
            settled = True
            yield _sse({
                "event": "usage", "inputTokens": input_tokens,
                "outputTokens": output_tokens, "creditsUsed": credits,
            })
            yield _sse({"event": "done"})
            yield _sse("[DONE]")
        except (httpx.HTTPError, RuntimeError) as error:
            if not settled:
                _finish_usage(
                    principal=principal, request_id=request_id, agent_id=agent_id,
                    agent_version=agent_version,
                    conversation_id=conversation_id, message_id=assistant_id,
                    capability=capability, model=usage_model, routing_reason=routing_reason,
                    input_tokens=input_tokens, output_tokens=output_tokens, tool_calls=tool_calls,
                    duration_ms=int((time.monotonic() - started_at) * 1000), status="error",
                    multiplier=multiplier,
                )
            yield _sse({"event": "error", "message": f"FunsClaw 暂时无法完成请求：{error}"})
            yield _sse("[DONE]")

    return StreamingResponse(generate(), media_type="text/event-stream")
