"""
pipeline-service: platform_ability — 平台内部 agent 能力包（pipeline_id=platform_general）

产线平台自身的运维/管理 agent：拥有平台应用和各模块的技能，
核心能力是「通读模型 API 文档 → 自动生成模型治理配置（供应商/适配模板/模型/定价）」。

权限模型（代码层硬门禁，不依赖 prompt）：
  所有工具 handler 入口校验调用者持有 owner 组织的角色（orgtypeid='owner'，含通配 'owner.*'）。
  内部 agent 只服务 owner 组织的角色——其他组织的用户即使打开页面也调不动任何工具。

配置生成链路（用户给定 API 文档 → 完成配置）：
  ① fetch_model_doc：抓取文档页面（SSRF 防护：仅 http(s) 公网域名）
  ② extract_llm_api_spec：LLM 从文档提取端点/协议/请求响应格式/定价
  ③ apply_llm_config：写库——复用 pipeline-llm 模块的 CRUD 函数
     （llm_vendor.endpoints + llm_api_profile 模板 + llm_model 含四价）
"""

import asyncio
import json
import logging
import re
import time

import yaml

from appPublic.uniqueID import getID

from pipeline_core import (
    ToolDefinition,
    PipelineAbility,
    register_ability,
)

logger = logging.getLogger("pipeline.platform_ability")

PLATFORM_PIPELINE_ID = "platform_general"

# 组织门禁：内部 agent 仅 owner 组织的角色可操作（任一 owner.* 角色，含通配）
OWNER_ORG = "owner"

# 当前运行时调用链支持的协议（模式定义在模型的适配模板/执行器，非供应商）：
#   openai_compat —— 同步 chat/completions 一次往返
#   dashscope_async —— 异步执行器：提交任务→轮询 query_profile_ids→取结果（2026-09-05）
# 其他协议的适配模板会存入 llm_api_profile 备查，但运行时暂不渲染。
RUNTIME_PROTOCOLS = ("openai_compat", "dashscope_async")

# ────────────────────── 会话级规格缓存（extract 锚定，2026-09-05）──────────────────────
# 根因：apply_llm_config 要求 LLM 把 extract 返回的大 JSON 规格逐字复制进参数，
# 模型「转述」大 JSON 必然编造结构（实测三轮全编造 vendor{}/api_profile{} 等契约外字段）。
# 对策：extract 成功后把规格锚定到「会话级缓存」，apply/apply_model_pricing 用
# use_last_extract=true 直接取用 + 少量覆盖项（vendor_name 等），LLM 永不搬运大 JSON。
#
# 会话级隔离（用户硬要求）：缓存键 = pipeline_id:user_id:session_id，三段缺一不可，
# 防跨会话/跨用户/跨产线串数据。存 Redis db4（治理命名空间，gateway 已在用），
# 跨 worker 进程不丢；TTL 兜底过期。Redis 不可用时降级进程内字典（仍按同键隔离）。
_SPEC_CACHE_TTL = 7200          # 规格缓存 2 小时（一次配置会话足够）
_SPEC_CACHE_PREFIX = "llm_spec"
_spec_cache_local = {}          # Redis 不可用时的进程内降级（同样按隔离键）


def _spec_cache_key(ctx):
    """会话级隔离键：pipeline_id:user_id:session_id。任一段缺失返回 ''（拒绝缓存）。"""
    pl = (ctx.get("pipeline_id") or "").strip()
    uid = (ctx.get("user_id") or "").strip()
    sid = (ctx.get("session_id") or "").strip()
    if not (pl and uid and sid):
        return ''
    return "%s:%s:%s:%s" % (_SPEC_CACHE_PREFIX, pl, uid, sid)


def _spec_redis():
    """复用治理命名空间 Redis db4（与 gateway 限流同库，会话态用 db3 不冲突）。"""
    try:
        import redis as _r
        try:
            from appPublic.jsonConfig import getConfig
            url = (getConfig().website or {}).get('session_redis', {}).get('url', '')
        except Exception:
            url = ''
        if not url:
            url = 'redis://127.0.0.1:6379/3'
        base = url.rsplit('/', 1)[0]
        return _r.Redis.from_url(base + '/4', socket_timeout=2)
    except Exception:
        return None


def _spec_save(ctx, spec):
    """锚定规格到会话缓存。返回 True/False。无隔离键则不缓存（apply 会要求显式传 spec）。"""
    key = _spec_cache_key(ctx)
    if not key:
        return False
    try:
        payload = json.dumps(spec, ensure_ascii=False)
    except Exception:
        return False
    r = _spec_redis()
    if r is not None:
        try:
            r.set(key, payload, ex=_SPEC_CACHE_TTL)
            return True
        except Exception:
            pass
    _spec_cache_local[key] = (payload, time.time() + _SPEC_CACHE_TTL)
    return True


def _spec_load(ctx):
    """取本会话锚定的规格。无则返回 None。严格会话级隔离（键含 session_id）。"""
    key = _spec_cache_key(ctx)
    if not key:
        return None
    r = _spec_redis()
    raw = None
    if r is not None:
        try:
            v = r.get(key)
            raw = v.decode('utf-8') if isinstance(v, (bytes, bytearray)) else v
        except Exception:
            raw = None
    if raw is None:
        ent = _spec_cache_local.get(key)
        if ent and ent[1] > time.time():
            raw = ent[0]
        elif ent:
            _spec_cache_local.pop(key, None)
    if not raw:
        return None
    try:
        return json.loads(raw)
    except Exception:
        return None


def _spec_overlay(base, overrides):
    """覆盖项合并：仅允许少量标量覆盖（vendor_name 等），不做大 JSON 搬运。

    overrides 只认白名单键，防止 LLM 借覆盖项塞回编造结构。
    """
    allowed = {'vendor_name', 'base_url', 'protocol', 'doc_url', 'doc_notes'}
    out = dict(base or {})
    for k in allowed:
        if k in (overrides or {}) and str(overrides.get(k) or '').strip():
            out[k] = str(overrides[k]).strip()
    return out


# ────────────────────── 权限门禁（代码层） ──────────────────────

async def _require_owner(sor, ctx) -> str:
    """校验当前用户持有 owner 组织的角色。返回 ''=通过，否则错误信息。

    判定：userrole→role JOIN 后，存在任一 orgtypeid='owner' 的角色（含通配 'owner.*'）。
    """
    uid = ctx.get("user_id", "") or ""
    if not uid:
        return "无法识别当前用户身份（未登录），内部 agent 仅 owner 组织角色可用"
    recs = await sor.sqlExe(
        "SELECT r.orgtypeid, r.name FROM userrole ur JOIN role r ON ur.roleid=r.id "
        "WHERE ur.userid=${u}$", {"u": uid})
    await sor.sqlExe("COMMIT", {})
    roles = []
    for r in (recs or []):
        o = getattr(r, "orgtypeid", "") or ""
        n = getattr(r, "name", "") or ""
        if o and n:
            roles.append("%s.%s" % (o, n))
    if any(fn.split(".", 1)[0] == OWNER_ORG for fn in roles):
        return ""
    return "权限不足：内部 agent 工具仅 %s 组织的角色可操作，当前角色 %s" % (
        OWNER_ORG, sorted(set(roles)) or "无")


def _row(r):
    """sqlor row → dict（容忍属性访问）。"""
    if hasattr(r, "__dict__"):
        return {k: v for k, v in r.__dict__.items() if not k.startswith("_")}
    return {}


# ────────────────────── 工具 1：模型治理状态 ──────────────────────

async def _h_platform_llm_status(sor, params, ctx):
    """供应商/账号/模型/用量概览（owner 组织诊断配置用）。"""
    err = await _require_owner(sor, ctx)
    if err:
        return err
    out = {}
    for tbl in ("llm_vendor", "llm_account", "llm_model"):
        recs = await sor.sqlExe(
            "SELECT status, COUNT(*) AS c FROM " + tbl + " GROUP BY status", {})
        await sor.sqlExe("COMMIT", {})
        out[tbl] = {getattr(r, "status", ""): int(getattr(r, "c", 0)) for r in (recs or [])}
    accs = await sor.sqlExe(
        "SELECT name, balance, status FROM llm_account ORDER BY balance DESC LIMIT 20", {})
    await sor.sqlExe("COMMIT", {})
    out["accounts"] = [
        {"name": getattr(r, "name", ""), "balance": float(getattr(r, "balance", 0) or 0),
         "status": getattr(r, "status", "")} for r in (accs or [])]
    models = await sor.sqlExe(
        "SELECT name, vendor_model_id, capability, status "
        "FROM llm_model ORDER BY created_at DESC LIMIT 30", {})
    await sor.sqlExe("COMMIT", {})
    out["models"] = [
        {"name": getattr(r, "name", ""), "vendor_model_id": getattr(r, "vendor_model_id", ""),
         "capability": getattr(r, "capability", ""),
         "status": getattr(r, "status", "")} for r in (models or [])]
    return json.dumps(out, ensure_ascii=False)


# ────────────────────── 工具 2：抓取模型 API 文档 ──────────────────────

_MAX_DOC_CHARS = 60000
_MAX_REDIRECTS = 5


def _validate_doc_url(url: str) -> str:
    """SSRF 防护：仅允许 http(s) + 公网域名。返回 ''=通过，否则错误。"""
    if not re.match(r"^https?://", url or ""):
        return "文档 URL 必须是 http/https 链接"
    host = re.sub(r"^https?://", "", url).split("/")[0].split(":")[0].lower()
    if not re.match(r"^[a-z0-9][a-z0-9.-]+\.[a-z]{2,}$", host):
        return "文档 URL 域名非法（不接受 IP/内网地址）"
    if host in ("localhost",) or host.endswith((".local", ".internal", ".localhost")):
        return "文档 URL 不允许指向内网/本地地址"
    return ""


def _html_to_text(html: str) -> str:
    """粗暴去标签提取文本（文档页面用，不追求完美排版）。"""
    txt = re.sub(r"(?is)<(script|style|noscript)[^>]*>.*?</\1>", " ", html or "")
    txt = re.sub(r"(?is)<[^>]+>", " ", txt)
    txt = re.sub(r"&nbsp;", " ", txt)
    txt = re.sub(r"&lt;", "<", txt)
    txt = re.sub(r"&gt;", ">", txt)
    txt = re.sub(r"&amp;", "&", txt)
    txt = re.sub(r"&quot;", '"', txt)
    txt = re.sub(r"[ \t]+", " ", txt)
    txt = re.sub(r"\n\s*\n+", "\n", txt)
    return txt.strip()


def _is_private_ip(ip: str) -> bool:
    """DNS 解析后的二次防线：解析到私有/环回/链路本地地址一律拒绝（DNS rebinding 防护）。"""
    import ipaddress
    try:
        a = ipaddress.ip_address(ip)
    except ValueError:
        return True  # 解析不出就当私有（拒绝）
    return (a.is_private or a.is_loopback or a.is_link_local
            or a.is_reserved or a.is_multicast or a.is_unspecified)


async def _fetch_url_safe(url: str, max_redirects: int = _MAX_REDIRECTS):
    """带重定向逐跳校验的抓取（每一跳都过域名+DNS双重校验）。

    返回 (text, content_type)；失败抛 ValueError。
    """
    import aiohttp
    current = url
    for _ in range(max_redirects + 1):
        verr = _validate_doc_url(current)
        if verr:
            raise ValueError(verr)
        host = re.sub(r"^https?://", "", current).split("/")[0].split(":")[0].lower()
        # DNS 解析校验（同步阻塞短调用，可接受；防 DNS rebinding 指向内网）
        try:
            infos = await asyncio.get_event_loop().getaddrinfo(host, None)
        except Exception as e:
            raise ValueError("文档域名无法解析：%s" % str(e)[:120])
        for info in infos:
            ip = str(info[4][0])
            if _is_private_ip(ip):
                raise ValueError("文档域名解析到内网地址（%s），拒绝访问" % ip)
        timeout = aiohttp.ClientTimeout(total=30)
        async with aiohttp.ClientSession(timeout=timeout) as sess:
            async with sess.get(current, headers={"User-Agent": "Mozilla/5.0"},
                                allow_redirects=False, ssl=False) as resp:
                if resp.status in (301, 302, 303, 307, 308):
                    loc = resp.headers.get("Location", "")
                    if not loc:
                        raise ValueError("重定向缺少 Location")
                    if loc.startswith("/"):
                        scheme = "https" if current.startswith("https") else "http"
                        loc = "%s://%s%s" % (scheme, host, loc)
                    current = loc
                    continue
                if resp.status != 200:
                    raise ValueError("抓取失败：HTTP %d" % resp.status)
                ctype = resp.headers.get("Content-Type", "")
                raw = await resp.text(errors="replace")
                return raw, ctype
    raise ValueError("重定向次数超限（>%d）" % max_redirects)


async def _h_fetch_model_doc(sor, params, ctx):
    """抓取模型 API 文档页面 → 纯文本（供 LLM 提取配置规格）。"""
    err = await _require_owner(sor, ctx)
    if err:
        return err
    url = (params.get("url") or "").strip()
    try:
        raw, ctype = await _fetch_url_safe(url)
    except ValueError as e:
        return str(e)
    except Exception as e:
        return "抓取失败：%s" % str(e)[:200]
    if "html" in ctype.lower() or raw.lstrip()[:15].lower().startswith(("<!doctype", "<html")):
        text = _html_to_text(raw)
    else:
        text = raw
    if len(text) > _MAX_DOC_CHARS:
        text = text[:_MAX_DOC_CHARS] + "\n[文档过长已截断，共 %d 字符]" % len(text)
    # 出处行：提取规格照抄进 doc_url，落库写进模型/定价描述（2026-09-05 用户规则）
    text = text + "\n\n[出处URL] " + url
    return text


# ────────────────────── 工具 3：提取配置规格（LLM） ──────────────────────

_EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型 API 文档，提取配置规格。

只输出一个 JSON 对象（不要 markdown 代码块），字段：
{
  "vendor_name": "供应商名称",
  "base_url": "API 基础地址（如 https://dashscope.aliyuncs.com/compatible-mode/v1）",
  "protocol": "openai_compat | dashscope_async | custom（能走 /chat/completions 的填 openai_compat）",
  "endpoints": [{"base_url": "...", "region": "domestic|international", "timeout": 60}],
  "chat_path": "对话接口路径（如 /chat/completions）",
  "auth_header": "认证头格式说明（如 Bearer API_KEY）",
  "request_headers": {"说明": "文档调用示例里除认证外的必需请求头，逐字照抄（如 DashScope 异步的 X-DashScope-Async: enable）；无则 null"},
  "request_fields": ["请求体字段名列表"],
  "request_example": {"说明": "文档调用示例(cURL/代码)中的完整请求体 JSON，逐字照抄结构与各字段示例值，不要修改；文档无示例填 null"},
  "response_format": "响应格式说明（content 字段路径 + usage 字段路径）",
  "response_example": {"说明": "文档中的响应示例 JSON（或结果字段说明），逐字照抄；无则 null"},
  "async_steps": [{"purpose": "query|download", "path": "该步骤接口路径", "method": "GET|POST", "request_fields": ["请求字段名列表"], "response_format": "该步骤响应格式说明"}],
  "models": [{"vendor_model_id": "供应商侧模型ID", "capability": "t2t|t2i|i2t|t2v|i2v|embedding|rerank|tts|asr",
              "sync_mode": "sync|async（提交后需轮询查询结果的填 async）",
              "description": "一句话说明"}],
  "pricing": {
    "currency": "CNY|USD（文档标注的币种）",
    "items": [{"vendor_model_id": "对应模型ID",
               "factor": "duration|flat|prompt_tokens|completion_tokens（计价因子：视频按秒=duration，按次=flat，token计价=对应token因子）",
               "unit_price": 0.45,
               "unit": "秒|次|百万（文档标注的计价单位名）",
               "dimensions": {"resolution": "480P（影响价格档位的维度，按文档原文值；禁止放 model）"},
               "doc_quote": "文档原文定价句（逐字照抄，标明出处用）"}]
  },
  "doc_url": "从文档文本末尾的 [出处URL] 行逐字照抄（配置出处，必填）",
  "doc_notes": "文档中影响配置的关键注意点"
}

异步模型规则：
- 文档描述「先提交任务、再轮询查询结果」的模型，sync_mode 填 async，并提取 async_steps：
  至少一条 purpose=query（查询任务状态/结果）；若文档另有独立下载/取文件接口，再加一条 purpose=download。
- 任务查询接口通常是**供应商级共用**的（如 DashScope 全系生成模型都是 GET /tasks/{task_id}），
  path 照文档逐字抄；同供应商多个模型会复用同一份查询模板，不要因模型而异。
- 同步模型 async_steps 填空数组 []。

定价规则：
- pricing.items 每条必须带 doc_quote（文档原文定价句逐字照抄）——定价只允许来自文档原文
- 价格/单位按文档原文记录，禁止自行换算、推导或统一单位
- 同一模型多档价格（如不同分辨率）拆成多条 item，各带自己的 dimensions 与 doc_quote
- dimensions 禁止放 model——一个定价方案只服务一个模型；定价完全相同的多个模型
  共享同一个定价方案（同一 ppid），不是往定价里加 model 过滤
- 文档没写价格：items 填空数组 []（禁止编造价格）

文档内容：
"""


async def _h_extract_llm_api_spec(sor, params, ctx):
    """LLM 通读文档文本 → 结构化配置规格（JSON）。"""
    err = await _require_owner(sor, ctx)
    if err:
        return err
    doc_text = (params.get("doc_text") or "").strip()
    if len(doc_text) < 100:
        return "文档文本太短（<100字符），无法提取配置——先用 fetch_model_doc 抓取"
    try:
        from pipeline_service.llm_bridge import llm_call_msgs
        # purpose='utility'：辅助任务走辅助模型链（机构策略配置则优先，便宜快）
        # timeout=300：长文档提取慢，端点默认 60 秒实测跑不完（2026-09-04 根因）
        raw = await llm_call_msgs(
            [{"role": "system", "content": _EXTRACT_PROMPT},
             {"role": "user", "content": doc_text[:48000]}],
            temperature=0.1, org_id="0", user_id=ctx.get("user_id", ""),
            purpose="utility", timeout=300)
    except Exception as e:
        return "LLM 提取失败：%s" % str(e)[:200]
    txt = (raw or "").strip()
    if txt.startswith("```"):
        txt = txt.split("\n", 1)[1] if "\n" in txt else txt
        txt = txt.rsplit("```", 1)[0]
    try:
        spec = json.loads(txt)
    except Exception:
        m = re.search(r"\{.*\}", txt, re.S)
        if not m:
            return "LLM 输出不是合法 JSON：%s" % txt[:300]
        try:
            spec = json.loads(m.group(0))
        except Exception:
            return "LLM 输出不是合法 JSON：%s" % txt[:300]
    # 会话级锚定（2026-09-05）：规格存会话缓存（键含 session_id，跨会话隔离），
    # 后续 apply_llm_config / apply_model_pricing 用 use_last_extract=true 取用——
    # LLM 不再搬运/转述大 JSON（转述必编造结构，实测三轮全编造）。
    anchored = _spec_save(ctx, spec)
    out = dict(spec)
    if anchored:
        out["__anchored__"] = ("规格已锚定到本会话缓存。下一步调用 apply_llm_config 与 "
                               "apply_model_pricing 时传 {\"use_last_extract\": true}，"
                               "需改动只传覆盖项（overrides，仅 vendor_name/base_url/"
                               "protocol/doc_url/doc_notes 白名单键）。"
                               "禁止把本规格复制进 spec 参数——复制即编造。")
    return json.dumps(out, ensure_ascii=False)


# ────────────────────── 工具 4：写入配置（幂等） ──────────────────────

def _f(v, default=0.0):
    try:
        return float(v)
    except (TypeError, ValueError):
        return default


async def _h_apply_llm_config(sor, params, ctx):
    """按提取的规格写库：供应商（复用/新建）+ 适配模板 + 模型。

    幂等：供应商按名称复用；模型按 name/vendor_model_id 复用（name 全局唯一键）——
    已存在则更新元数据；供应商归属不一致时不改挂、如实报 vendor_conflicts。

    规格来源（2026-09-05 防编造改造）：
      use_last_extract=true（推荐）→ 取本会话 extract 锚定的规格，
        overrides 白名单覆盖（vendor_name 等少量标量）；
      否则用 params.spec（LLM 手传，历史路径）——结构不符时报错列出期望键。
    """
    err = await _require_owner(sor, ctx)
    if err:
        return err
    use_last = str(params.get("use_last_extract") or "").lower() in ("true", "1", "yes")
    overrides = params.get("overrides") or {}
    if isinstance(overrides, str):
        try:
            overrides = json.loads(overrides)
        except Exception:
            overrides = {}
    spec = None
    if use_last:
        spec = _spec_load(ctx)
        if spec is None:
            return ("本会话没有锚定的提取规格——先调 extract_llm_api_spec（成功后规格自动"
                    "锚定本会话），再传 {\"use_last_extract\": true}。"
                    "注意会话级隔离：其他会话提取的规格本会话不可见，需在本会话重新提取。")
        spec = _spec_overlay(spec, overrides)
    else:
        try:
            spec = json.loads(params.get("spec") or "{}")
        except Exception:
            return "spec 不是合法 JSON"
        # 结构校验（修B：报错给出路，不让模型盲改重试）
        if not isinstance(spec, dict) or "vendor_name" not in spec or "models" not in spec:
            got = sorted(spec.keys())[:12] if isinstance(spec, dict) else "(非对象)"
            return ("spec 结构不符——不要自己编写/改写规格结构。期望顶层键："
                    "vendor_name / base_url / endpoints / protocol / chat_path / models / "
                    "pricing / doc_url / doc_notes（extract_llm_api_spec 的原样输出）。"
                    "你传入的顶层键：%s。正确做法：调 extract_llm_api_spec 后传 "
                    "{\"use_last_extract\": true}（规格已锚定本会话，禁止复制转述），"
                    "需改动加 overrides（仅 vendor_name/base_url/protocol/doc_url/doc_notes）。"
                    % got)
    vendor_name = (spec.get("vendor_name") or "").strip()
    if not vendor_name:
        return "spec 缺少 vendor_name（可用 overrides.vendor_name 指定）"
    endpoints = spec.get("endpoints") or []
    if not endpoints and spec.get("base_url"):
        endpoints = [{"base_url": spec["base_url"], "region": "domestic", "timeout": 60}]
    if not endpoints:
        return "spec 缺少 endpoints/base_url"
    protocol = (spec.get("protocol") or "openai_compat").strip() or "openai_compat"
    models = spec.get("models") or []
    if not models:
        return "spec.models 为空——文档里没有可配置的模型？"

    # 能力分类校验（2026-09-05 用户定夺的规则）：模型能力必须是字典已登记的
    # 分类；不存在则拒绝落库——先加能力分类（appcodes_kv llm_capability，
    # 含种子/提示词/端点注释四处同步），再配模型。防 LLM 静默塞进近似分类。
    recs = await sor.sqlExe(
        "SELECT k FROM appcodes_kv WHERE parentid='llm_capability'", {})
    await sor.sqlExe("COMMIT", {})
    known_caps = set(getattr(r, "k", "") for r in (recs or []))
    unknown = sorted(set(
        (m.get("capability") or "t2t").strip() for m in models) - known_caps)
    if unknown:
        return ("能力分类未登记，拒绝落库：%s。请先在能力分类字典"
                "（appcodes_kv parentid=llm_capability，含种子数据/提取提示词/"
                "模型列表端点注释同步）中添加该分类，再重新执行本工具。"
                "现有分类：%s" % ("、".join(unknown), "、".join(sorted(known_caps))))

    from appPublic.uniqueID import getID

    # 1. 供应商：按名称复用
    recs = await sor.sqlExe(
        "SELECT id, endpoints FROM llm_vendor WHERE name=${n}$", {"n": vendor_name})
    await sor.sqlExe("COMMIT", {})
    if recs:
        vendor_id = getattr(recs[0], "id", "")
        eps_old = []
        try:
            eps_old = json.loads(getattr(recs[0], "endpoints", "") or "[]")
        except Exception:
            eps_old = []
        merged = list(eps_old)
        added_eps = []
        for ep in endpoints:
            bu = (ep.get("base_url") or "").rstrip("/")
            if bu and bu not in [(e.get("base_url") or "").rstrip("/") for e in merged]:
                merged.append({"base_url": bu, "region": ep.get("region") or "domestic",
                               "timeout": int(_f(ep.get("timeout"), 60))})
                added_eps.append(bu)
        # 2026-09-05：供应商表不再有 protocol——请求形态由模型挂的适配模板决定
        await sor.sqlExe(
            "UPDATE llm_vendor SET endpoints=${e}$, updated_at=NOW() WHERE id=${i}$",
            {"e": json.dumps(merged, ensure_ascii=False), "i": vendor_id})
        await sor.sqlExe("COMMIT", {})
        vendor_action = "复用供应商 %s（新增端点 %d 个）" % (vendor_id, len(added_eps))
    else:
        vendor_id = getID()
        eps_norm = [{"base_url": (ep.get("base_url") or "").rstrip("/"),
                     "region": ep.get("region") or "domestic",
                     "timeout": int(_f(ep.get("timeout"), 60))} for ep in endpoints]
        await sor.C("llm_vendor", {
            "id": vendor_id, "name": vendor_name,
            "endpoints": json.dumps(eps_norm, ensure_ascii=False),
            "description": spec.get("doc_notes", "") or "",
            "status": "active", "org_id": ctx.get("org_id", "") or "0"})
        vendor_action = "新建供应商 %s（%s）" % (vendor_id, vendor_name)

    # 2. 适配模板：按 (协议×能力) 复用；旧骨架模板（硬编码 xxx_file/__from_doc__，
    #    2026-09-05 实测 t2v 因它渲染崩）视为不可用，按文档示例自愈重建
    profile_ids = {}
    tpl_errors = []
    tpl_notes = []
    caps = sorted(set((m.get("capability") or "t2t") for m in models))
    for cap in caps:
        recs = await sor.sqlExe(
            "SELECT id, request_template, response_template FROM llm_api_profile "
            "WHERE protocol=${p}$ AND capability=${c}$ AND status='active' "
            "ORDER BY created_at DESC LIMIT 1",
            {"p": protocol, "c": cap})
        await sor.sqlExe("COMMIT", {})
        if recs:
            pid_old = getattr(recs[0], "id", "")
            old_tpl = (getattr(recs[0], "request_template", "") or "") + \
                      (getattr(recs[0], "response_template", "") or "")
            if not any(mk in old_tpl for mk in _SKELETON_MARKERS):
                profile_ids[cap] = pid_old
                continue
            await sor.sqlExe("UPDATE llm_api_profile SET status='deprecated' "
                             "WHERE id=${i}$", {"i": pid_old})
            await sor.sqlExe("COMMIT", {})
            tpl_notes.append("旧骨架模板 %s（%s）含硬编码占位符，已弃用并按文档示例重建"
                             % (pid_old, cap))
        tpl = _gen_templates(protocol, cap, spec)
        dry_err = _dry_render_check(tpl["req"], cap, tpl["biz_params"], tpl["media_params"])
        if dry_err:
            # 宁可不建，不埋雷：渲染不了的模板落库=运行时必崩，如实报错由助手修正规格重跑
            tpl_errors.append("%s: %s" % (cap, dry_err))
            continue
        pid = getID()
        await sor.C("llm_api_profile", {
            "id": pid, "name": "%s-%s-自动配置" % (protocol, cap),
            "protocol": protocol, "capability": cap,
            "path": tpl["path"],
            "headers": json.dumps(tpl["headers"], ensure_ascii=False),
            "request_template": tpl["req"], "response_template": tpl["resp"],
            "param_schema": json.dumps(tpl["param_schema"], ensure_ascii=False),
            "status": "active"})
        profile_ids[cap] = pid
        for nt in tpl["notes"]:
            tpl_notes.append("%s: %s" % (cap, nt))

    # 3. 模型：按 name/vendor_model_id 幂等（name 全局唯一键）
    doc_url = (spec.get("doc_url") or "").strip()
    created, updated, skipped = [], [], []
    conflicts = []  # 供应商归属冲突（模型已挂别的供应商）——如实报告不静默迁移
    for m in models:
        vmid = (m.get("vendor_model_id") or "").strip()
        if not vmid:
            skipped.append("(缺 vendor_model_id)")
            continue
        cap = m.get("capability") or "t2t"
        sync_mode = "async" if str(m.get("sync_mode") or "").strip() == "async" else "sync"
        desc = (m.get("description", "") or "").strip()
        # 出处标注（2026-09-05 用户规则）：模型注册表描述字段必须保存文档 URL
        if doc_url and ("[出处:" + doc_url + "]") not in desc:
            desc = (desc + " " if desc else "") + "[出处:%s]" % doc_url
        # 异步模型：后续步骤模板链（提交后顺序执行 query→[download]）
        query_ids = []
        if sync_mode == "async":
            steps = spec.get("async_steps") or []
            if not any((s.get("purpose") or "") == "query" for s in steps):
                desc = "⚠️异步模型但文档未提取到查询步骤（query_profile_ids 需人工补录）。" + desc
            for s in steps:
                qid = await _ensure_step_profile(sor, protocol, cap, vmid, s, spec)
                if qid:
                    query_ids.append(qid)
        query_ids_json = json.dumps(query_ids) if query_ids else ""
        # 幂等键=name（uk_llm_model_name 全局唯一）。不能按 (vendor_id, vmid) 查：
        # 模型换挂供应商后（如迁到阿里云百炼），旧查询找不到→INSERT→撞唯一键崩（实测）。
        recs = await sor.sqlExe(
            "SELECT id, vendor_id FROM llm_model WHERE name=${m}$ OR vendor_model_id=${m}$",
            {"m": vmid})
        await sor.sqlExe("COMMIT", {})
        if recs:
            mid = getattr(recs[0], "id", "")
            cur_vendor = getattr(recs[0], "vendor_id", "") or ""
            if cur_vendor and cur_vendor != vendor_id:
                # 供应商归属不一致：治理决策，不静默迁移——只更新元数据并如实报告
                conflicts.append(
                    "%s: 模型已存在且挂供应商 id=%s，与本次规格供应商（%s）不一致——"
                    "未改供应商归属，仅更新描述/同步模式；如需迁移请明确指示"
                    % (vmid, cur_vendor, vendor_name))
            await sor.sqlExe(
                "UPDATE llm_model SET description=${d}$, "
                "sync_mode=${sm}$, query_profile_ids=${q}$, updated_at=NOW() "
                "WHERE id=${i}$",
                {"d": desc, "sm": sync_mode, "q": query_ids_json, "i": mid})
            await sor.sqlExe("COMMIT", {})
            updated.append(vmid)
            continue
        mid = getID()
        await sor.C("llm_model", {
            "id": mid, "vendor_id": vendor_id, "account_id": "",
            "name": vmid, "vendor_model_id": vmid, "capability": cap,
            "sync_mode": sync_mode, "profile_id": profile_ids.get(cap, ""),
            "query_profile_ids": query_ids_json,
            "ppid": "",
            "default_params": "{}", "status": "active",
            "description": desc, "org_id": "0"})
        created.append(vmid)

    rt_note = ""
    if protocol not in RUNTIME_PROTOCOLS:
        rt_note = ("⚠️ 协议「%s」的适配模板已存档，但当前运行时调用链仅支持 %s——"
                   "该供应商模型暂不可被产线直接调用" % (protocol, "/".join(RUNTIME_PROTOCOLS)))
    audit_note = await _audit_media_convention(sor, list(profile_ids.values()))
    return json.dumps({
        "vendor": vendor_action, "profiles": profile_ids,
        "models_created": created, "models_updated": updated, "models_skipped": skipped,
        "vendor_conflicts": conflicts,
        "template_errors": tpl_errors,
        "template_notes": tpl_notes,
        "runtime_note": rt_note,
        "media_audit": audit_note,
    }, ensure_ascii=False)


# 生成类能力：出参是文件（图/视频/音频/3D），response 必须 downloadfile2url 落地
_MEDIA_CAPS = ('t2i', 'i2v', 't2v', 't2a', 'tts', 'i2i', 'v2v', '3d')


async def _audit_media_convention(sor, profile_ids):
    """强制检查（2026-09-05 用户规则：每个 llm 配置都要检查媒体转换约定）：

    生成类能力的适配模板——
      request_template 含上传媒体参数时，必须用 {{b64media2url(request, xxx_file)}}
        转本地公网 URL 再传上游；
      response_template 必须用 {{downloadfile2url(request, <产物url>)}}
        把生成物落地为本地持久 URL（上游 URL 有效期短，视频仅 24 小时）。
    返回检查结论（无问题为空串），随 apply 结果返回给内部 agent 转告。
    """
    if not profile_ids:
        return ""
    issues = []
    for pid in profile_ids:
        recs = await sor.sqlExe(
            "SELECT name, capability, request_template, response_template "
            "FROM llm_api_profile WHERE id=${i}$", {"i": pid})
        await sor.sqlExe("COMMIT", {})
        if not recs:
            continue
        r = recs[0]
        cap = getattr(r, 'capability', '') or ''
        name = getattr(r, 'name', '') or pid
        if cap not in _MEDIA_CAPS:
            continue  # 非生成类（t2t/embedding/rerank）无生成物
        rt = getattr(r, 'request_template', '') or ''
        st = getattr(r, 'response_template', '') or ''
        if '__note__' in st and 'downloadfile2url' not in st:
            issues.append("「%s」response 模板是骨架——生成物必须用 "
                          "downloadfile2url(request, <产物url>) 落地后再返回" % name)
        elif st and 'downloadfile2url' not in st:
            issues.append("「%s」response 模板缺 downloadfile2url——生成类能力"
                          "（%s）的产物 URL 必须落地为本地持久 URL" % (name, cap))
        # 上传参数线索：模板里出现 image/video/audio 文件参数但没有 b64media2url
        has_upload_ref = any(k in rt for k in ('image_file', 'video_file', 'audio_file',
                                               'first_frame', 'image_url', 'media'))
        if has_upload_ref and 'b64media2url' not in rt:
            issues.append("「%s」request 模板含媒体上传参数但缺 b64media2url——"
                          "上传文件必须转本地公网 URL 再传上游" % name)
    if issues:
        return "⚠️ 媒体转换约定检查未通过：" + "；".join(issues)
    return ""


# 能力 → 统一出参键（同类能力对外契约一致：视频出 video、图出 image）
_MEDIA_OUT_KEY = {
    't2v': 'video', 'i2v': 'video', 'v2v': 'video',
    't2i': 'image', 'i2i': 'image',
    'tts': 'audio', 't2a': 'audio',
    '3d': 'glb',
}
# 能力 → 上游产物字段名兜底（文档没给响应示例时用供应商惯例）
_MEDIA_OUT_FIELD = {
    'video': 'video_url', 'image': 'image_url', 'audio': 'audio_url', 'glb': 'model_url',
}
# 文档示例里表示「上传媒体」的结构线索
_MEDIA_TYPES = ('first_frame', 'last_frame', 'ref_image', 'ref_images', 'image', 'img')
_MEDIA_KEY_HINTS = ('img_url', 'image_url', 'first_frame_url', 'last_frame_url',
                    'ref_image_url', 'video_url', 'audio_url')
# 需要「输入媒体」的能力（这些能力里出现 URL 叶子才判定为上传媒体）
_MEDIA_INPUT_CAPS = ('i2v', 'i2i', 'v2v', 'i2t', '2i2v')


def _media_param_name(typeval, keyname):
    """按文档媒体结构推断运行时统一媒体参数名（xxx_file 契约）。"""
    t = ('%s %s' % (typeval or '', keyname or '')).lower()
    if 'video' in t:
        return 'video_file'
    if 'audio' in t:
        return 'audio_file'
    return 'image_file'


def _is_url_example(v):
    return isinstance(v, str) and v.strip().lower().startswith(('http://', 'https://'))


def _example_to_nested(fields):
    """request_fields 点号路径 → 嵌套 dict（文档无请求示例时的兜底结构来源）。"""
    root = {}
    for f in fields or []:
        parts = [p for p in str(f).split('.') if p]
        if not parts:
            continue
        cur = root
        for p in parts[:-1]:
            nxt = cur.get(p)
            if not isinstance(nxt, dict):
                nxt = {}
                cur[p] = nxt
            cur = nxt
        cur[parts[-1]] = ''
    return root


def _tpl_from_example(example, capability):
    """文档请求示例 → Jinja2 请求体模板（结构逐字保留，值换成运行时变量）。

    规则（2026-09-05，替代旧的硬编码骨架）：
      model 键            → {{ model|tojson }}
      prompt/text 键      → {{ prompt|tojson }}
      上传媒体（示例里是 URL 叶子，且结构/键名带媒体线索或能力属输入媒体类）
                          → {{ b64media2url(request, params.<media>_file)|tojson }}
      其他标量            → {{ params.<key>|default(<文档示例值>)|tojson }}
    **示例里没有上传媒体就不生成媒体参数**——纯文生视频(t2v)曾因硬编码
    params.xxx_file 骨架在运行时崩（'dict object' has no attribute 'xxx_file'）。

    返回 (模板串, media_params, biz_params, used_example)。
    """
    tokens = {}
    media_params, biz_params = [], []
    used_example = isinstance(example, dict) and bool(example)
    tree = example if used_example else _example_to_nested(None)

    def _put(expr):
        k = '__TPL%d__' % len(tokens)
        tokens[k] = expr
        return k

    def walk(node, parent_key='', sib_type=None):
        if isinstance(node, list):
            return [walk(v, parent_key, sib_type) for v in node]
        if not isinstance(node, dict):
            return node
        tval = node.get('type') if isinstance(node.get('type'), str) else sib_type
        out = {}
        for k, v in node.items():
            if isinstance(v, (dict, list)):
                out[k] = walk(v, k, tval)
                continue
            kl = str(k).lower()
            if kl == 'model' and isinstance(v, str):
                out[k] = _put('{{ model|tojson }}')
                continue
            if kl in ('prompt', 'text') and isinstance(v, str):
                out[k] = _put('{{ prompt|tojson }}')
                continue
            is_media = _is_url_example(v) and (
                str(tval or '').lower() in _MEDIA_TYPES
                or kl in _MEDIA_KEY_HINTS
                or capability in _MEDIA_INPUT_CAPS)
            if is_media:
                mp = _media_param_name(tval, k)
                if mp not in media_params:
                    media_params.append(mp)
                out[k] = _put('{{ b64media2url(request, params.%s)|tojson }}' % mp)
                continue
            biz_params.append({'name': k, 'example': v})
            if v == '' or v is None:
                out[k] = _put('{{ params.%s|tojson }}' % k)   # 无示例值：调用必传
            else:
                out[k] = _put('{{ params.%s|default(%s)|tojson }}'
                              % (k, json.dumps(v, ensure_ascii=False)))
        return out

    body = walk(tree)
    s = json.dumps(body, ensure_ascii=False)
    for k, expr in tokens.items():
        s = s.replace('"%s"' % k, expr)
    return s, media_params, biz_params, used_example


def _collect_url_fields(node, prefix=''):
    """递归收集响应示例里的 URL 字段点号路径（如 output.video_url）。"""
    hits = []
    if isinstance(node, dict):
        for k, v in node.items():
            p = (prefix + '.' + str(k)) if prefix else str(k)
            if isinstance(v, (dict, list)):
                hits.extend(_collect_url_fields(v, p))
            elif _is_url_example(v):
                hits.append(p)
    elif isinstance(node, list):
        for i, v in enumerate(node[:3]):
            hits.extend(_collect_url_fields(v, prefix))
    return hits


def _resp_tpl_for(capability, spec):
    """生成类响应模板：产物 URL 经 downloadfile2url 落地（上游 URL 有效期短）。

    产物字段路径优先取文档响应示例里的 URL 字段（递归，支持 output.video_url
    这类嵌套——运行时 ns 有 output/usage/task_id），缺则用供应商惯例兜底，
    并在 notes 里如实说明来源（不假装是文档确证）。
    """
    outkey = _MEDIA_OUT_KEY.get(capability, 'video')
    field = ''
    rex = spec.get('response_example')
    if isinstance(rex, dict):
        hits = _collect_url_fields(rex)
        # 优先 *_url 结尾且与产物类型匹配的字段
        want = ('video', 'image', 'audio', 'glb', 'model')
        for h in hits:
            leaf = h.split('.')[-1].lower()
            if leaf.endswith('_url') and any(w in leaf for w in want):
                field = h
                break
        if not field and hits:
            field = hits[0]
    note = ''
    if not field:
        field = _MEDIA_OUT_FIELD.get(outkey, 'result_url')
        note = ('产物字段名「%s」按供应商惯例兜底（文档未提供响应示例或示例中无 URL 字段），'
                '如上游字段不同需修正' % field)
    resp = json.dumps({
        'status': 'SUCCEEDED',
        outkey: '{{ downloadfile2url(request, %s) }}' % field,
        'usage': '{{ json.dumps(usage) }}',
        'task_id': '{{ task_id }}',
    }, ensure_ascii=False)
    return resp, note


def _headers_from_spec(spec):
    """请求头：认证头统一 api_key 变量，其余按文档示例逐字带上
    （如 DashScope 异步必需的 X-DashScope-Async: enable）。"""
    headers = {'Authorization': 'Bea' + 'rer {{api_key}}',
               'Content-Type': 'application/json'}
    rh = spec.get('request_headers')
    if isinstance(rh, dict):
        for k, v in rh.items():
            if str(k).lower() in ('authorization', 'content-type'):
                continue
            if isinstance(v, str) and v.strip():
                headers[k] = v.strip()
    # 协议兜底：dashscope_async 提交必须带异步开关头（文档示例遗漏也不至于提交即失败）
    if str(spec.get('protocol') or '').strip() == 'dashscope_async':
        headers.setdefault('X-DashScope-Async', 'enable')
    return headers


def _gen_templates(protocol: str, capability: str, spec: dict) -> dict:
    """按文档示例动态生成适配模板（path/headers/request/response/param_schema）。

    返回 {path, headers, req, resp, param_schema, media_params, biz_params, notes}。
    """
    chat_path = (spec.get("chat_path") or "/chat/completions").strip()
    headers = _headers_from_spec(spec)
    notes = []
    if protocol == "openai_compat":
        req = json.dumps({
            "model": "{{model}}", "messages": "{{messages}}",
            "temperature": "{{temperature}}", "stream": False,
        }, ensure_ascii=False)
        resp = json.dumps({
            "content": "choices[0].message.content",
            "usage": {"prompt_tokens": "usage.prompt_tokens",
                      "completion_tokens": "usage.completion_tokens"},
        }, ensure_ascii=False)
        return {"path": chat_path, "headers": headers, "req": req, "resp": resp,
                "param_schema": [{"name": "prompt", "label": "提示词",
                                  "uitype": "textarea", "required": True}],
                "media_params": [], "biz_params": [], "notes": notes}

    example = spec.get("request_example")
    if not (isinstance(example, dict) and example):
        example = _example_to_nested(spec.get("request_fields"))
        notes.append("文档未提供完整请求示例，模板按 request_fields 字段路径生成，"
                     "业务参数无默认值（调用时必须显式传入），建议人工核对")
    req, media_params, biz_params, _used = _tpl_from_example(example, capability)
    if capability in _MEDIA_CAPS:
        resp, rnote = _resp_tpl_for(capability, spec)
        if rnote:
            notes.append(rnote)
        if not media_params and capability in _MEDIA_INPUT_CAPS:
            notes.append("能力 %s 通常需要输入媒体，但文档示例未见上传字段——"
                         "模板未生成媒体参数，请核对文档" % capability)
    else:
        resp = json.dumps({"content": "{{ text }}"}, ensure_ascii=False)
    schema = [{"name": "prompt", "label": "提示词", "uitype": "textarea",
               "required": capability not in ('embedding', 'rerank')}]
    for mp in media_params:
        schema.append({"name": mp, "label": "上传媒体（%s）" % mp,
                       "uitype": "file", "required": True})
    for b in biz_params:
        schema.append({"name": b['name'], "label": b['name'],
                       "uitype": "number" if isinstance(b['example'], (int, float)) else "text",
                       "required": False, "default": b['example']})
    return {"path": chat_path, "headers": headers, "req": req, "resp": resp,
            "param_schema": schema, "media_params": media_params,
            "biz_params": biz_params, "notes": notes}


# 旧骨架模板标记：含这些标记的 profile 视为不可用，apply 时按文档自愈重建
_SKELETON_MARKERS = ('__from_doc__', '__note__', 'xxx_file')


def _dry_render_check(req_tpl, capability, biz_params, media_params):
    """落库前干跑渲染（StrictUndefined）：模板引用了运行时拿不到的变量当场报错。

    运行时命名空间见 pipeline-llm inference._build_async_body：
      model / prompt / messages / params(业务参数) / api_key / org_id
      + request / json / b64media2url / downloadfile2url。
    旧缺陷正是漏了这步：模板硬编码 params.xxx_file，直到 test_model_call 才崩。
    """
    try:
        from jinja2 import Environment, StrictUndefined
    except Exception as e:
        return "干跑校验跳过（jinja2 不可用：%s）" % str(e)[:60]

    def _stub(request, value, *a, **k):
        return str(value)     # StrictUndefined 传入未定义值时在此抛错

    params = {}
    for b in biz_params or []:
        # 干跑模拟「业务参数全部提供」场景：无示例值的参数给占位串
        # （模板 {{ params.x|tojson }} 无默认——运行时必须显式传，属预期契约）
        params[b['name']] = b['example'] if b.get('example') not in ('', None) else 'dry'
    for mp in media_params or []:
        params[mp] = 'https://dry-run.invalid/sample.bin'
    ns = {'model': 'dry-run-model', 'prompt': '干跑校验', 'messages': [],
          'params': params, 'api_key': 'sk-dry-run', 'org_id': '0',
          'request': None, 'json': json,
          'b64media2url': _stub, 'downloadfile2url': _stub}
    try:
        env = Environment(undefined=StrictUndefined)
        out = env.from_string(req_tpl).render(**ns)
        json.loads(out)
    except Exception as e:
        return ("请求模板干跑渲染失败（落库前拦截）：%s: %s——模板引用了运行时不存在的"
                "变量（业务参数可用：%s）" % (type(e).__name__, str(e)[:160],
                                              sorted(params.keys()) or '无'))
    return ""


async def _ensure_step_profile(sor, protocol, cap, vmid, step, spec):
    """异步后续步骤的适配模板（query/download）。

    2026-09-05 改造（用户：同类任务查询接口供应商级相同，可复用）：
      幂等键=名称「{protocol}-query-{path}」（去掉模型维度）——同供应商同协议
      下多个模型共享同一份查询模板；旧的按 {vmid}-{purpose} 命名继续兼容复用。
      模板按文档真实生成：运行时查询只用 path/headers/method，task_id 经
      path 占位（如 /tasks/{{task_id}}）；不再产 __from_doc__ 骨架。
    返回 profile id。
    """
    purpose = (step.get("purpose") or "query").strip() or "query"
    path = (step.get("path") or "").strip()
    method = (step.get("method") or ("GET" if purpose == "query" else "POST")).strip().upper()
    shared_name = "%s-%s-%s" % (protocol, purpose, path or "default")
    legacy_name = "%s-%s" % (vmid, purpose)
    recs = await sor.sqlExe(
        "SELECT id FROM llm_api_profile WHERE name=${n}$ AND status='active' LIMIT 1",
        {"n": shared_name})
    await sor.sqlExe("COMMIT", {})
    if recs:
        return getattr(recs[0], "id", "")
    recs = await sor.sqlExe(
        "SELECT id FROM llm_api_profile WHERE name=${n}$ AND status='active' LIMIT 1",
        {"n": legacy_name})
    await sor.sqlExe("COMMIT", {})
    if recs:
        return getattr(recs[0], "id", "")
    headers = _headers_from_spec(spec)
    if method == "GET":
        req = json.dumps({"method": method}, ensure_ascii=False)
        if "{{task_id}}" not in path:
            # path 没带任务号占位时按 DashScope 惯例拼 /tasks/{id}；notes 如实标注
            path = (path.rstrip("/") + "/{{task_id}}") if path else "/tasks/{{task_id}}"
    else:
        req = json.dumps({"method": method,
                          "data": {"task_id": "{{task_id}}"}}, ensure_ascii=False)
    # 查询步骤的响应渲染在运行时由提交模板的 response_template 接管
    # （inference 轮询直接调 _render_response(submit_profile,...)），这里存档即可
    resp = json.dumps({
        "__note__": "查询步骤模板：运行时响应解析走模型提交模板的 response_template",
        "response_format": step.get("response_format", "") or "",
    }, ensure_ascii=False)
    pid = getID()
    await sor.C("llm_api_profile", {
        "id": pid, "name": shared_name, "protocol": protocol, "capability": cap,
        "path": path,
        "headers": json.dumps(headers, ensure_ascii=False),
        "request_template": req, "response_template": resp,
        "param_schema": "", "status": "active"})
    return pid


# ────────────────────── 工具 5：定价自动导入 ──────────────────────

_FACTOR_LABELS = {
    'duration': '时长', 'flat': '按次', 'prompt_tokens': '输入tokens',
    'completion_tokens': '输出tokens',
}


def _build_pricing_yaml(currency_items, dimensions_all, factor):
    """生成定价 YAML（2026-09-05 用户定夺的新模式）：

    - 不放 model 过滤：一个定价方案只服务一个模型；定价相同的多个模型共享
      同一 ppid（不是往定价里加 model 维度）
    - 不用 filters 子结构：维度直接平铺在定价项（引擎对非保留键做 AND 匹配）
    - fields 需定义每个维度（role: filter），否则引擎报「在fields中没有定义」
    """
    fields = {
        'price_factors': {'type': 'string', 'role': 'factor', 'label': '计价因子'},
        'unit_prices': {'type': 'float', 'role': 'factor', 'label': '单位定价'},
        'unit': {'type': 'string', 'role': 'factor', 'label': '计价单位'},
    }
    if factor not in fields:
        fields[factor] = {'type': 'float', 'role': 'factor',
                          'label': _FACTOR_LABELS.get(factor, factor)}
    for dim in sorted(dimensions_all):
        fields[dim] = {'type': 'string', 'role': 'filter', 'label': dim}
    pricings = []
    for it in currency_items:
        item = {'price_factors': factor, 'unit_prices': it['unit_price'],
                'unit': it['unit']}
        for dk, dv in sorted((it.get('dimensions') or {}).items()):
            item[dk] = dv
        pricings.append(item)
    units = sorted(set(it['unit'] for it in currency_items))
    unit_values = {u: 1 for u in units}
    if '百万' in unit_values:
        unit_values['百万'] = 1000000
    if '千' in unit_values:
        unit_values['千'] = 1000
    doc = {'unit_values': unit_values, 'fields': fields, 'pricings': pricings}
    return yaml.dump(doc, allow_unicode=True, sort_keys=False)


async def _h_apply_model_pricing(sor, params, ctx):
    """定价自动导入：按提取规格建定价方案并挂模型 ppid（幂等）。

    幂等键：模型已有 ppid 且方案描述含同一出处 URL → 更新时序（拉链）；
    否则新建方案。价格只来自文档原文（提取层强制 doc_quote）。

    规格来源（2026-09-05 防编造改造）：use_last_extract=true → 取本会话
    extract 锚定规格（含 pricing/doc_url），禁止 LLM 复制转述大 JSON。
    """
    err = await _require_owner(sor, ctx)
    if err:
        return err
    use_last = str(params.get("use_last_extract") or "").lower() in ("true", "1", "yes")
    if use_last:
        spec = _spec_load(ctx)
        if spec is None:
            return ("本会话没有锚定的提取规格——先调 extract_llm_api_spec（成功后规格自动"
                    "锚定本会话，含 pricing 与 doc_url），再传 {\"use_last_extract\": true}。"
                    "会话级隔离：其他会话的规格本会话不可见。")
    else:
        try:
            spec = json.loads(params.get("spec") or "{}")
        except Exception:
            return "spec 不是合法 JSON"
        if not isinstance(spec, dict) or "pricing" not in spec:
            return ("spec 结构不符——不要自己编写规格。正确做法：extract_llm_api_spec 后传 "
                    "{\"use_last_extract\": true}（pricing 与 doc_url 已锚定本会话）。")
    pricing = spec.get("pricing") or {}
    items = pricing.get("items") or []
    if not items:
        return "spec.pricing.items 为空——文档没提取到价格（禁止编造），如需定价请先核对文档"
    doc_url = (spec.get("doc_url") or "").strip()
    currency = (pricing.get("currency") or "CNY").strip() or "CNY"
    # 校验：每条必须有 doc_quote（出处）；dimensions 禁止 model
    for it in items:
        if not (it.get("doc_quote") or "").strip():
            return "定价条目缺 doc_quote（文档原文定价句）——定价只允许来自文档原文，拒绝落库"
        if "model" in (it.get("dimensions") or {}):
            return "定价 dimensions 含 model——一个定价方案只服务一个模型，禁止 model 维度；定价相同的模型共享同一 ppid"
    # 按 vendor_model_id 分组（一个模型一个定价方案）
    by_model = {}
    for it in items:
        vmid = (it.get("vendor_model_id") or "").strip()
        if not vmid:
            return "定价条目缺 vendor_model_id"
        by_model.setdefault(vmid, []).append(it)

    results = []
    for vmid, mitems in by_model.items():
        # 找模型
        recs = await sor.sqlExe(
            "SELECT id, name, ppid FROM llm_model WHERE vendor_model_id=${m}$ "
            "AND status='active' LIMIT 1", {"m": vmid})
        await sor.sqlExe("COMMIT", {})
        if not recs:
            results.append({"model": vmid, "ok": False,
                            "error": "模型未注册（先 apply_llm_config）"})
            continue
        model_id = getattr(recs[0], "id", "")
        old_ppid = getattr(recs[0], "ppid", "") or ""
        factors = sorted(set((it.get("factor") or "flat") for it in mitems))
        if len(factors) != 1:
            results.append({"model": vmid, "ok": False,
                            "error": "同一模型混用多个计价因子 %s——请拆分为多个定价方案" % factors})
            continue
        factor = factors[0]
        dims_all = set()
        for it in mitems:
            dims_all.update((it.get("dimensions") or {}).keys())
        yaml_str = _build_pricing_yaml(mitems, dims_all, factor)
        # 引擎试算护栏：渲染一遍确认合法 YAML + fields 完整（不落库先验证）
        try:
            parsed = yaml.safe_load(yaml_str)
            assert parsed.get('pricings') and parsed.get('fields')
        except Exception as e:
            results.append({"model": vmid, "ok": False,
                            "error": "定价 YAML 生成校验失败：%s" % str(e)[:150]})
            continue
        desc_quotes = "；".join((it.get("doc_quote") or "")[:80] for it in mitems)[:500]
        pp_desc = "定价出处：%s | 文档原文：%s" % (doc_url or "(未提供)", desc_quotes)
        now = time.strftime('%Y-%m-%d')
        if old_ppid:
            ppid = old_ppid
            # 幂等核心（2026-09-05 用户纠正：时序应只有一条有效行，重复执行不得堆历史）：
            # 先取当前生效行比对内容——
            #   内容相同 → 跳过（不产生新时序行）
            #   内容不同且生效行今天才启用 → 原地 UPDATE（同日替换无历史区间可保留，
            #     拉链会产生 enabled==expired 的零宽死行，纯垃圾）
            #   内容不同且生效行是历史日期 → 正常拉链（关旧行+插新行）
            recs2 = await sor.sqlExe(
                "SELECT id, pricing_data, enabled_date FROM pricing_program_timing "
                "WHERE ppid=${p}$ AND expired_date='9999-12-31' "
                "ORDER BY enabled_date DESC LIMIT 1", {"p": ppid})
            await sor.sqlExe("COMMIT", {})
            cur = recs2[0] if recs2 else None
            same = False
            if cur is not None:
                try:
                    same = (yaml.safe_load(getattr(cur, 'pricing_data', '') or '{}')
                            == yaml.safe_load(yaml_str))
                except Exception:
                    same = (getattr(cur, 'pricing_data', '') or '') == yaml_str
            await sor.sqlExe(
                "UPDATE pricing_program SET description=${d}$ WHERE id=${i}$",
                {"d": pp_desc[:1000], "i": ppid})
            await sor.sqlExe("COMMIT", {})
            if same:
                results.append({"model": vmid, "ok": True, "ppid": ppid,
                                "action": "无变化（幂等跳过，未产生新时序行）",
                                "items": len(mitems), "factor": factor,
                                "dimensions": sorted(dims_all)})
                continue
            if cur is not None and str(getattr(cur, 'enabled_date', ''))[:10] == now:
                await sor.sqlExe(
                    "UPDATE pricing_program_timing SET pricing_data=${y}$, name=${n}$ "
                    "WHERE id=${i}$",
                    {"y": yaml_str, "n": "%s %s计价" % (vmid, factor),
                     "i": getattr(cur, 'id', '')})
                await sor.sqlExe("COMMIT", {})
                action = "原地更新（同日替换，不产生历史行）"
                results.append({"model": vmid, "ok": True, "ppid": ppid,
                                "action": action, "items": len(mitems),
                                "factor": factor, "dimensions": sorted(dims_all)})
                continue
            # 历史生效行 → 拉链
            await sor.sqlExe(
                "UPDATE pricing_program_timing SET expired_date=${d}$ "
                "WHERE ppid=${p}$ AND expired_date='9999-12-31'",
                {"d": now, "p": ppid})
            action = "更新（拉链：旧行截至今日，新时序生效 %s）" % now
        else:
            ppid = getID()
            await sor.C("pricing_program", {
                "id": ppid, "name": "%s 定价" % vmid, "ownerid": "0",
                "providerid": "", "pricing_belong": "",
                "description": pp_desc[:1000], "currency": currency})
            action = "新建"
        await sor.C("pricing_program_timing", {
            "id": getID(), "ppid": ppid, "name": "%s %s计价" % (vmid, factor),
            "pricing_data": yaml_str,
            "enabled_date": now, "expired_date": "9999-12-31"})
        await sor.sqlExe("COMMIT", {})
        # 挂模型
        await sor.sqlExe("UPDATE llm_model SET ppid=${p}$ WHERE id=${i}$",
                         {"p": ppid, "i": model_id})
        await sor.sqlExe("COMMIT", {})
        results.append({"model": vmid, "ok": True, "ppid": ppid,
                        "action": action, "items": len(mitems),
                        "factor": factor,
                        "dimensions": sorted(dims_all)})
    ok_n = sum(1 for r in results if r.get("ok"))
    return json.dumps({
        "summary": "定价导入：%d/%d 个模型成功" % (ok_n, len(results)),
        "results": results,
        "convention": "一个定价方案只服务一个模型（无 model 维度/无 filters，维度平铺）；"
                      "定价完全相同的模型共享同一 ppid",
    }, ensure_ascii=False)


# ────────────────────── 工具 6：模型测试（真实调用） ──────────────────────

async def _h_test_model_call(sor, params, ctx):
    """真实调用测试：走完整治理链（门禁→上游→结算），返回调用结果+流水ID。

    参数：model_name 必填；prompt 文本提示（t2t）；业务参数放 params JSON
    （如 image_file/resolution/duration）；timeout 等待秒数（异步模型默认 600）。
    前置：供应商账号已配 api_key（无 key 会报可行动错误，不静默）。
    """
    err = await _require_owner(sor, ctx)
    if err:
        return err
    model_name = (params.get("model_name") or "").strip()
    if not model_name:
        return "缺 model_name"
    prompt = (params.get("prompt") or "回复两个字：正常").strip()
    try:
        extra = json.loads(params.get("params") or "{}")
    except Exception:
        return "params 不是合法 JSON"
    timeout_s = int(_f(params.get("timeout"), 0)) or 600
    try:
        from pipeline_llm.inference import chat_inference
    except Exception as e:
        return "推理模块未加载：%s" % str(e)[:100]
    payload = {"model": model_name,
               "messages": [{"role": "user", "content": prompt}]}
    payload.update(extra)
    if timeout_s:
        payload["_timeout"] = timeout_s
    task_ref = "agent-test:%s:%d" % (model_name[:20], int(time.time()))
    t0 = time.time()
    try:
        data = await chat_inference(ctx.get("org_id", "") or "0",
                                    ctx.get("user_id", ""), payload,
                                    model_name=model_name, task_ref=task_ref)
    except Exception as e:
        return json.dumps({
            "ok": False, "model": model_name, "task_ref": task_ref,
            "elapsed_sec": round(time.time() - t0, 1),
            "error": str(e)[:400],
            "hint": "常见原因：账号无api_key/余额不足/端点超时/模板配置缺失——按错误消息处置",
        }, ensure_ascii=False)
    elapsed = round(time.time() - t0, 1)
    content = ""
    try:
        content = data["choices"][0]["message"]["content"]
    except Exception:
        content = str(data)[:200]
    out = {
        "ok": True, "model": model_name, "task_ref": task_ref,
        "elapsed_sec": elapsed,
        "content_head": str(content)[:200],
        "usage": data.get("usage") or {},
    }
    if data.get("media"):
        out["media"] = data["media"]   # 生成物本地持久 URL（已落地）
    if data.get("task_id"):
        out["task_id"] = data["task_id"]
    return json.dumps(out, ensure_ascii=False)


# ────────────────────── 工具 7：记账检查 ──────────────────────

async def _h_check_model_accounting(sor, params, ctx):
    """记账正确性检查（三态：created/accounted/failed）。

    按 task_ref（测试调用返回的）或 model_name 查最近流水：
    1. 流水存在性 + accounting_status
    2. usages 计价因子完整性
    3. accounted → charge 金额与定价引擎重算对比（独立复算，不信记账侧）
    4. failed → 给出 note 原因与处置建议
    5. created → 出账循环 60 秒一轮，提示等待或查 worker 进程
    """
    err = await _require_owner(sor, ctx)
    if err:
        return err
    task_ref = (params.get("task_ref") or "").strip()
    model_name = (params.get("model_name") or "").strip()
    if not task_ref and not model_name:
        return "需要 task_ref（测试调用返回值）或 model_name 之一"
    if task_ref:
        recs = await sor.sqlExe(
            "SELECT id, model_id, status, accounting_status, charge, cost, usages, "
            "ppid, note, created_at FROM llm_usage WHERE task_ref=${t}$ "
            "ORDER BY created_at DESC LIMIT 3", {"t": task_ref})
    else:
        recs = await sor.sqlExe(
            "SELECT u.id, u.model_id, u.status, u.accounting_status, u.charge, u.cost, "
            "u.usages, u.ppid, u.note, u.created_at FROM llm_usage u "
            "JOIN llm_model m ON m.id=u.model_id "
            "WHERE m.name=${n}$ OR m.vendor_model_id=${n}$ "
            "ORDER BY u.created_at DESC LIMIT 3", {"n": model_name})
    await sor.sqlExe("COMMIT", {})
    if not recs:
        return "未找到流水（task_ref=%s model=%s）——调用可能没发生或没结算" % (
            task_ref, model_name)
    out_rows = []
    for r in recs:
        row = dict(r)
        st = row.get("accounting_status", "")
        item = {
            "usage_id": row.get("id", ""),
            "call_status": row.get("status", ""),
            "accounting_status": st,
            "charge": float(row.get("charge") or 0),
            "usages": row.get("usages", "") or "",
            "ppid": row.get("ppid", "") or "",
            "created_at": str(row.get("created_at", "")),
        }
        if st == "failed":
            item["verdict"] = "记账失败"
            item["reason"] = (row.get("note") or "")[:200]
            item["hint"] = ("常见处置：模型未挂ppid→apply_model_pricing；"
                            "模型未映射产品→产品管理导入产线模型；"
                            "定价无匹配档位→核对usages维度值与定价YAML是否一致")
        elif st == "created":
            item["verdict"] = "待记账（出账循环60秒一轮，稍后复查；持续不变查记账worker进程）"
        elif st == "accounted":
            # 独立复算：用 usages 因子过定价引擎，对比 charge
            item["verdict"] = "已记账"
            try:
                usage_data = json.loads(row.get("usages") or "{}")
            except Exception:
                usage_data = {}
            ppid = row.get("ppid") or ""
            if ppid and usage_data:
                try:
                    from ahserver.serverenv import ServerEnv
                    env = ServerEnv()
                    fn = getattr(env, "buffered_charging", None)
                    if fn:
                        prices = await fn(ppid, usage_data)
                        expect = round(sum(float(getattr(p, "amount", 0) or 0)
                                           for p in (prices or [])), 6)
                        item["recomputed_amount"] = expect
                        item["amount_match"] = abs(expect - float(row.get("charge") or 0)) < 0.01
                        if not item["amount_match"]:
                            item["verdict"] = "已记账但金额不符（复算 %.4f ≠ charge %.4f）" % (
                                expect, float(row.get("charge") or 0))
                except Exception as e:
                    item["recompute_error"] = str(e)[:150]
            else:
                item["verdict"] = "已记账（缺ppid或usages，无法独立复算）"
        out_rows.append(item)
    return json.dumps({"rows": out_rows}, ensure_ascii=False)


# ────────────────────── 工具 8：模块/应用信息查询 ──────────────────────

async def _h_platform_modules(sor, params, ctx):
    """列出平台已装载的业务模块（内部 agent 了解平台构成用）。"""
    err = await _require_owner(sor, ctx)
    if err:
        return err
    mods = []
    for name, title in [
        ("pipeline_core", "产线核心（会话/技能/项目）"),
        ("pipeline_service", "执行引擎（任务链/角色/治理）"),
        ("pipeline-llm", "模型治理（供应商/账号/模型/定价/记账）"),
        ("pipeline-bidding", "投标产线"),
        ("pipeline-opportunity", "商机产线"),
        ("pipeline-sdlc", "开发产线前端"),
        ("pipeline-ops", "运维"),
        ("pipeline-dist", "分销"),
        ("pipeline-task", "任务中心"),
    ]:
        try:
            __import__(name.replace("-", "_"))
            loaded = True
        except Exception:
            loaded = False
        mods.append({"module": name, "title": title, "loaded": loaded})
    return json.dumps(mods, ensure_ascii=False)


# ────────────────────── 工具定义 ──────────────────────

PLATFORM_TOOLS = [
    ToolDefinition(
        name="platform_llm_status",
        description="查看模型治理状态：供应商/账号余额/模型定价/各状态分布。用户问「模型配置状态/账号余额」时调用。仅owner组织角色可用。",
        parameters={},
        category="platform",
    ),
    ToolDefinition(
        name="fetch_model_doc",
        description="抓取大模型供应商的 API 文档页面（返回纯文本）。配置新模型前先用此抓取官方文档。仅owner组织角色可用。",
        parameters={"url": "文档页面 URL（必须是公网 http/https）"},
        category="platform",
    ),
    ToolDefinition(
        name="extract_llm_api_spec",
        description="通读文档文本，LLM 提取 API 配置规格（端点/协议/请求响应格式/定价）。配合 fetch_model_doc 使用。仅owner组织角色可用。",
        parameters={"doc_text": "fetch_model_doc 返回的文档文本"},
        category="platform",
    ),
    ToolDefinition(
        name="apply_llm_config",
        description="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型。幂等。推荐 use_last_extract=true 取本会话锚定规格（禁止把规格复制进 spec——复制即编造）；需改动加 overrides（仅 vendor_name/base_url/protocol/doc_url/doc_notes）。仅owner组织角色可用。",
        parameters={
            "use_last_extract": "true=用本会话 extract_llm_api_spec 锚定的规格（推荐，会话级隔离）",
            "overrides": "覆盖项 JSON（白名单：vendor_name/base_url/protocol/doc_url/doc_notes，如用户要求改供应商名）",
            "spec": "（仅无锚定时的兜底）extract 返回的 JSON 规格原文",
        },
        category="platform",
        # 交互原则（2026-09-05 用户定）：配置链工具不设确认门——缺信息才问，
        # 否则一口气做完给测试结果，用户发现问题再按说明修改。
    ),
    ToolDefinition(
        name="apply_model_pricing",
        description="定价自动导入：建定价方案（pricing_program+时序YAML）并挂模型 ppid。幂等（内容无变化跳过；同日原地更新；历史行才拉链——时序表始终只一条有效行）。一个定价方案只服务一个模型；定价相同的模型共享同一 ppid。推荐 use_last_extract=true 取本会话锚定规格（含 pricing/doc_url），禁止复制转述。仅owner组织角色可用。",
        parameters={
            "use_last_extract": "true=用本会话锚定规格的 pricing+doc_url（推荐）",
            "spec": "（仅无锚定时的兜底）extract 返回的 JSON 规格原文",
        },
        category="platform",
    ),
    ToolDefinition(
        name="test_model_call",
        description="模型真实调用测试：走完整治理链（门禁→上游→结算）。返回 ok/content/usage/media（生成物本地URL）/task_ref。前置：供应商账号已配 api_key。异步模型（视频等）等待至任务完成（默认600秒）。仅owner组织角色可用。",
        parameters={
            "model_name": "模型名（llm_model.name 或 vendor_model_id）",
            "prompt": "文本提示词（t2t 默认「回复两个字：正常」；生成类填生成描述）",
            "params": "业务参数 JSON 字符串（生成类模型用，如 {\"image_file\": \"https://...\", \"resolution\": \"480P\", \"duration\": 5}）",
            "timeout": "等待秒数（默认600，上限900）",
        },
        category="platform",
    ),
    ToolDefinition(
        name="check_model_accounting",
        description="记账正确性检查（三态 created/accounted/failed）：查最近流水的记账状态/usages因子/charge金额，accounted 时用定价引擎独立复算金额对比。failed 给原因与处置建议。仅owner组织角色可用。",
        parameters={
            "task_ref": "test_model_call 返回的 task_ref（优先）",
            "model_name": "或按模型名查最近流水",
        },
        category="platform",
    ),
    ToolDefinition(
        name="platform_modules",
        description="列出平台已装载的业务模块清单。用户问「平台有哪些模块」时调用。仅owner组织角色可用。",
        parameters={},
        category="platform",
    ),
]

PLATFORM_PROMPT = """
你是产线平台的内部运维 agent，服务对象是 owner 组织的角色。

## 模型自动配置全链工作流（用户给文档 URL 要求配置模型时，一口气独立完成）
1. **先 `load_skill` 加载 `model-auto-config`**（含媒体转换铁律/能力分类先行/定价新模式），严格执行
2. `fetch_model_doc` 抓取官方文档（返回文本末尾带 [出处URL] 行）
3. `extract_llm_api_spec` 通读文档提取规格——**成功后规格自动锚定到本会话缓存**
   （会话级隔离），返回里有 __anchored__ 提示
4. **不要停下来找用户确认**——直接进入下一步落库。只有文档里确实缺失、
   且无法从上下文合理推断的信息（如 api_key、能力分类需新增）才向用户提问。
5. `apply_llm_config` 传 **{"use_last_extract": true}**（用户要求改名等只加
   overrides 白名单覆盖项，如 {"overrides": {"vendor_name": "阿里云百炼"}}）。
   **绝对禁止把规格 JSON 复制/转述进 spec 参数——转述必编造结构（历史教训：
   三轮全编造 vendor{}/api_profile{} 等不存在的字段）。**
6. `apply_model_pricing` 同样传 **{"use_last_extract": true}**（定价+出处已锚定）
7. `test_model_call` 真实调用测试（前置：账号已配 api_key，没配则如实告知用户去补，
   不要伪造结果）——生成类模型传 params（image_file/resolution/duration 等）
8. `check_model_accounting` 检查记账（传上一步的 task_ref）：
   - accounted 且 amount_match=true → 全链通过，汇报金额
   - created → 等 60~120 秒再查一次（出账循环 60 秒一轮）
   - failed → 按 reason/hint 处置（未映射产品→告知用户在产品管理导入产线模型）
9. 汇报：配置/定价/测试/记账四段结果 + 出处 URL + 遗留事项（如模板需人工核对项）。
   用户发现问题后按其说明修改，再重新走对应环节（全链幂等，重跑安全）。

## 交互原则（用户定，2026-09-05）
- **只有缺失信息需要用户补充时才交互**（api_key、业务参数、需新增能力分类等）
- 否则从抓取到记账检查**一口气做完**，最后给出完整测试结果
- **不要每步找用户确认**——落库/定价/测试都直接执行（全部幂等，改起来安全）
- 用户发现问题 → 按用户说明修改重跑，这才是正确的纠错循环

## 工具报错自救规则
- 报「spec 结构不符/缺少 xxx」→ **不要改写结构重试**，改用 use_last_extract=true；
  本会话没锚定就重新走 fetch→extract（锚定是会话级的，跨会话不可见）
- 报「能力分类未登记」→ 如实转告用户需先加能力分类（这是人工评审门，不要绕过）
- 报「media_audit/runtime_note/需人工核对」警告 → 原样转告用户，不假装完成

## 硬规则
- **定价只允许来自文档原文**（doc_quote 强制），禁止编造/换算/推导
- **出处必留**：模型注册表描述字段与定价方案描述字段都必须保存文档 URL
- 能力分类不存在 → 先加分类（字典/种子/提示词/端点注释四处同步）再配模型，禁止塞近似分类
- 生成类模型媒体铁律：上行 b64media2url(request, xxx_file)、下行 downloadfile2url(request, url)；
  apply 返回的 media_audit 警告必须如实转告
- 非标协议骨架模板需人工核对的项（__note__）必须如实转告用户，不假装完成
- 文档抓取失败/内容不足时如实说明，不要凭记忆编造 API 格式
- 测试没有 api_key 就停在配置阶段如实汇报，禁止跳过测试谎称完成
- 所有工具仅 owner 组织角色可用，权限报错时如实转告用户

## 平台知识
平台模块清单用 `platform_modules` 查询；技能库中「平台」相关技能（skills_library/all/ 与
pipelines/platform_general/）包含各模块的开发规范与踩坑记录，需要时用 load_skill 加载。
"""

PLATFORM_HANDLERS = {
    "platform_llm_status": _h_platform_llm_status,
    "fetch_model_doc": _h_fetch_model_doc,
    "extract_llm_api_spec": _h_extract_llm_api_spec,
    "apply_llm_config": _h_apply_llm_config,
    "apply_model_pricing": _h_apply_model_pricing,
    "test_model_call": _h_test_model_call,
    "check_model_accounting": _h_check_model_accounting,
    "platform_modules": _h_platform_modules,
}


def register_platform_ability():
    """注册平台内部 agent 能力包（幂等）。"""
    ability = PipelineAbility(
        pipeline_id=PLATFORM_PIPELINE_ID,
        name="平台内部 agent",
        tools=PLATFORM_TOOLS,
        system_prompt=PLATFORM_PROMPT,
        handlers=PLATFORM_HANDLERS,
        roles=[],
        menus=[
            {"label": "📊 模型治理", "icon": "", "url": "/pipeline-llm",
             "type": "tab"},
        ],
    )
    register_ability(ability)
    return ability


register_platform_ability()