group_fqcd_jr/tools/publish_customer_service_config.py

"""发布客服 Agent 的运行期配置：意图工具白名单（走发布状态机）。

两个必须讲清的点：

1. **为什么必须发这一步**：工具白名单是失败关闭的——`ToolExecutor` 拿发布配置里
   `agent_tools` / `customer_service:<intent>` 的 `allowed_tools` 与代码声明的
   `AgentDefinition.allowed_tools` 取交集，缺配置时交集为空、任何工具调用都被拒。
   「Agent 写好了但没发配置」的表现是"客服什么都答不了、一直在引导人工"。

2. **为什么必须继承现有配置项**：`config_release` 是**整版本替换**语义——激活新版本后，
   旧版本的所有配置项都不再生效。若只发布客服自己的白名单，示例 Agent 的
   `fund_query_demo:fund_quote` 会被静默清空。所以发布前先把当前 effective 版本里的
   配置项原样搬进新版本，再追加本次新增项。

用法：python tools/publish_customer_service_config.py
"""

import asyncio
import datetime as dt
import json
import sys
import uuid
from pathlib import Path
from typing import Any

import httpx
import jwt
from sqlalchemy import func, select

from app.core.config import get_settings
from app.infrastructure.db import SessionFactory
from app.main import create_app
from app.model.configuration import PromptTemplateVersion
from app.service.config_release_service import ConfigReleaseService

ADMIN = "9003"
AGENT_TYPE = "customer_service"
TOOL_NAME = "search_knowledge"
# 访客检索工具。与 `TOOL_NAME` 是**同一个只读检索处理器**的别名（见 `bootstrap.py`），
# 但权限码与角色集不同：
#   search_knowledge -> `knowledge:reference:read`，角色 customer/advisor/operator/admin
#   query_knowledge  -> `knowledge:query`，角色 visitor/customer
# 访客令牌（`VisitorTokenIssuer`）只带 `agent:run` + `knowledge:query`，角色是 `visitor`，
# 所以公开浮窗**只能**走 `query_knowledge` —— 客服 Agent 也确实是这么分支的
# （`_answer_from_knowledge` 里按 `"visitor" in context.roles` 选工具名）。
# 漏发它的表现：访客在公开页面一问就抛 `ForbiddenAgentError`（工具不在意图白名单），
# 而登录客户侧完全正常 —— 因为客户走的是 `search_knowledge`。
# 所以知识类意图必须**同时**发两个：访客走前者、登录客户走后者。
VISITOR_TOOL = "query_knowledge"
SUITABILITY_TOOL = "check_suitability"
# 画像只读工具：客服的"出口零"（本人风险等级/投资偏好/测评是否过期）走它取权威字段。
# 那个出口复用的是 `faq` 意图 key（见 `customer_service.PROFILE_WHITELIST_INTENT`），
# 所以**必须**把它加进 `faq` 的白名单里；漏了的表现是"问画像一律转人工"，
# 而画像出口本身是对的——这是纯配置缺口（实测复现过）。
PROFILE_TOOL = "query_customer_profile"
# 只有会调用工具的意图才需要白名单；chitchat（模型生成）与 transfer_human（引导人工）
# 都不查知识库。给它们配空白名单反而会掩盖"配置漏配"，因此不发布这两条。
INTENT_TOOLS: dict[str, tuple[str, ...]] = {
    # 三个知识意图都要**同时**发两个检索工具，原因见 VISITOR_TOOL 的注释：
    # 访客与登录客户落在同一个意图里，却走不同的工具名。
    "faq": (TOOL_NAME, VISITOR_TOOL, PROFILE_TOOL),
    "product_inquiry": (TOOL_NAME, VISITOR_TOOL),
    "policy_explain": (TOOL_NAME, VISITOR_TOOL),
    # 适当性裁决要两步：先从知识库拿到产品的风险等级，再由底座按档案里的客户等级裁决。
    # 这里**不发** VISITOR_TOOL：访客意图白名单（`VISITOR_INTENTS`）不含 suitability_check，
    # 访客根本到不了这条出口，发了只是噪音。
    "suitability_check": (TOOL_NAME, SUITABILITY_TOOL),
}


def token(subject: str) -> str:
    settings = get_settings()
    private_key = Path(settings.jwt_private_key_path).read_text(encoding="utf-8")
    now = dt.datetime.now(dt.UTC)
    return jwt.encode(
        {
            "sub": subject, "iss": settings.jwt_issuer, "aud": settings.jwt_audience,
            "exp": now + dt.timedelta(minutes=30), "nbf": now - dt.timedelta(seconds=5),
            "jti": str(uuid.uuid4()),
        },
        private_key,
        algorithm="RS256",
    )


#: `PromptPayload` 接受的字段（`app/api/schemas/admin.py`）。表里还有
#: `checksum` / `created_by` / `created_at`，那三个由服务端生成，不能搬。
PROMPT_API_FIELDS = (
    "prompt_code", "task_type", "agent_type",
    "system_prompt", "user_prompt_template", "input_schema", "output_schema",
)


def as_json(value: Any) -> Any:
    """JSON 列从驱动读出来可能是字符串、也可能已解析，统一归一化。

    不归一化的后果很隐蔽：`value_json` 保持字符串时搬过去会被判成"不是对象"，
    而 `input_schema` / `output_schema` 是 `dict | None` 字段，传字符串直接 422。
    """
    return json.loads(value) if isinstance(value, str) else value


async def active_snapshot() -> dict[str, list[dict[str, Any]]]:
    """当前生效版本在**全部三张受管表**里的内容。

    为什么不能只查 `platform_config_item`（旧写法就是这么写的，代价见下）：
    `config_release` 是**整版本替换**语义，新版本没带上的行**等于被删除**。
    本项目为此丢过两次配置，两次都是"功能看着正常、零告警"：

      * release 174 → 181：提示词被漏搬，`load_active_prompt` 读不到，Agent 静默回落
        到代码里的默认话术；
      * 本次（254 → 303）：本脚本只搬配置项，把 `customer_service_chitchat` 提示词漏在
        了旧版本里 —— admin 端只在激活时打一句 stderr 警告，很容易被刷过去。

    `ConfigReleaseService.effective_snapshot()` 一次读全三张表，是唯一正确的来源。
    """
    async with SessionFactory() as session:
        return await ConfigReleaseService(session).effective_snapshot()


def snapshot_items(snapshot: dict[str, list[dict[str, Any]]]) -> list[dict[str, Any]]:
    """把快照里的配置项转成 API 载荷形状。

    库是 `config_key`、API 是 `item_key` —— 字段名不同，快照行不能直接 POST。
    """
    return [
        {
            "namespace": row["namespace"],
            "item_key": row["config_key"],
            "value_json": as_json(row["value_json"]),
            "schema_version": row["schema_version"],
        }
        for row in snapshot["platform_config_item"]
    ]


async def next_prompt_versions(rows: list[dict[str, Any]]) -> list[int]:
    """给每条待搬运的提示词分配新 version。

    `prompt_template_version` 的唯一键含 `version`，照搬旧行会主键冲突；
    逐个 `prompt_code` 取现有最大值 +1。
    """
    assigned: list[int] = []
    async with SessionFactory() as session:
        for row in rows:
            latest = await session.scalar(
                select(func.max(PromptTemplateVersion.version)).where(
                    PromptTemplateVersion.prompt_code == row["prompt_code"]
                )
            )
            assigned.append(int(latest or 0) + 1)
    return assigned


def prompt_payloads(
    rows: list[dict[str, Any]], versions: list[int], release_id: int
) -> list[dict[str, Any]]:
    """提示词的 API 载荷（`input_schema` / `output_schema` **必须带上**，漏了即静默丢失）。"""
    payloads: list[dict[str, Any]] = []
    for row, version in zip(rows, versions, strict=True):
        payload: dict[str, Any] = {"release_id": release_id, "version": version}
        for field in PROMPT_API_FIELDS:
            value = row.get(field)
            payload[field] = as_json(value) if field.endswith("_schema") else value
        payloads.append(payload)
    return payloads


async def post(
    client: httpx.AsyncClient, path: str, *, auth: dict[str, str],
    payload: dict[str, object] | None = None, if_match: str | None = None,
) -> httpx.Response:
    headers = {**auth, "Idempotency-Key": uuid.uuid4().hex}
    if if_match:
        headers["If-Match"] = if_match
    return await client.post(path, json=payload, headers=headers)


async def etag_of(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> str | None:
    return (await client.get(path, headers=auth)).headers.get("ETag")


SUITABILITY_INTENT = "suitability_check"
# description 与 examples 是**给分类器看的**：意图码本身只是个名字，真正让模型分辨
# "能买吗"和"这个产品是什么"的是这几个例子。所以 examples 全部取客户真实说法。
SUITABILITY_INTENT_SPEC: dict[str, Any] = {
    "intent_name": "投资者适当性判断",
    "description": "客户询问以自己的风险承受能力能否购买某只产品，或询问自身风险等级与产品的匹配情况",
    "examples": [
        "c1客户能买它吗", "我能买这个产品吗", "那它我能买吗",
        "这只基金适合我吗", "我的风险等级能买吗",
    ],
    "confidence_threshold": "0.6000",
}


async def ensure_suitability_intent(client: httpx.AsyncClient, auth: dict[str, str]) -> int:
    """确保 `suitability_check` 意图在运行期生效（返回 0 成功、1 失败）。

    为什么必须做这一步：意图码要三处对齐（见 `customer_service.py` 的注释），而运行期
    只读 `agent_intent_config` 里 status='active' 的行。少了这一行，分类链路看不到这个
    意图，"能买吗"会被分到别的意图里去，客户拿到的就是"C1 的通用规则"而不是结论。
    """
    path = "/api/v1/admin/agent-intent-configs"
    listed = await client.get(f"{path}?limit=100", headers=auth)
    rows = listed.json().get("data", []) if listed.status_code == 200 else []
    existing = next(
        (row for row in rows
         if row.get("agent_type") == AGENT_TYPE and row.get("intent_code") == SUITABILITY_INTENT),
        None,
    )
    if existing is not None and str(existing.get("status")) == "active":
        print(f"[意图配置] id={existing['id']} 已生效，跳过")
        return 0
    if existing is not None and str(existing.get("status")) in {"draft", "approved"}:
        config_id = int(existing["id"])
    else:
        # 没有历史行或历史行已归档：新版本号（该表 agent_type+intent_code+version 唯一）
        version = int(existing.get("version", 0)) + 1 if existing else 1
        created = await post(client, path, auth=auth, payload={
            "agent_type": AGENT_TYPE,
            "intent_code": SUITABILITY_INTENT,
            **SUITABILITY_INTENT_SPEC,
            "allowed_tools": [TOOL_NAME, SUITABILITY_TOOL],
            "version": version,
        })
        if created.status_code != 201:
            print(f"[意图配置] 创建失败：{created.status_code} {created.text[:200]}")
            return 1
        config_id = int(created.json()["data"]["id"])
        print(f"[意图配置] 已创建 id={config_id}（{SUITABILITY_INTENT} v{version}）")

    base = f"{path}/{config_id}"
    # 幂等：重跑脚本时某一步可能已经推进过（"已经审过了"不该 409 让整个脚本失败）
    settled = {"reviews": {"approved", "active"}, "activations": {"active"}}
    for action, payload in (
        ("reviews", {"decision": "approved", "comment": "创建人自审"}),
        # 激活端点要求 body 是对象；传 None 时 httpx 根本不发 body，会被判 422
        ("activations", {}),
    ):
        current = (await client.get(base, headers=auth)).json().get("data", {})
        if str(current.get("status")) in settled[action]:
            print(f"[意图配置] {action} 已在目标状态（{current.get('status')}），跳过")
            continue
        response = await post(
            client, f"{base}/{action}", auth=auth, payload=payload,
            if_match=await etag_of(client, base, auth),
        )
        if response.status_code != 200:
            print(f"[意图配置] {action} 失败：{response.status_code} {response.text[:200]}")
            return 1
    print(f"[意图配置] id={config_id} 已生效（运行期按 status='active' 读取）")
    return 0


async def main() -> int:
    app = create_app()
    auth = {"Authorization": f"Bearer {token(ADMIN)}"}
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=60
    ) as client:
        if await ensure_suitability_intent(client, auth) != 0:
            return 1

        snapshot = await active_snapshot()
        inherited = snapshot_items(snapshot)
        prompts = list(snapshot["prompt_template_version"])
        rules = list(snapshot["model_routing_rule"])
        print(
            f"当前生效版本：配置项 {len(inherited)} 条、提示词 {len(prompts)} 条、"
            f"路由规则 {len(rules)} 条（将原样继承）"
        )
        for item in inherited:
            print(f"  · {item['namespace']} / {item['item_key']}")
        for prompt in prompts:
            print(f"  · 提示词 {prompt['prompt_code']}（v{prompt['version']}）")
        if rules:
            # 本环境当前为空；不为空就必须先支持搬运，否则激活即静默清空。
            print(f"\n[失败] 生效版本里有 {len(rules)} 条 model_routing_rule，本脚本尚未支持搬运。")
            print("       直接发布会把它们清空 —— 先补上搬运逻辑再跑。")
            return 1

        new_items = [
            {
                "namespace": "agent_tools",
                "item_key": f"{AGENT_TYPE}:{intent}",
                "value_json": {"allowed_tools": list(tools)},
                "schema_version": "1",
            }
            for intent, tools in INTENT_TOOLS.items()
        ]
        # **同 key 的继承项必须被本次新定义覆盖**，不能原样搬过去。
        # 血泪教训（实测）：上一版发布的是 `faq = ["query_knowledge", ...]`，而那个工具
        # 已随"客服检索改为 search_knowledge"从代码上限移除；原样继承会让 admin 服务的
        # 子集校验（白名单 ⊆ 代码限定的 allowed_tools）直接 422 拒绝整次发布，
        # 报错是"配置超出 Agent 工具上限"，看不出是继承造成的。
        inherited_only = [
            item for item in inherited
            if (str(item["namespace"]), str(item["item_key"])) not in {
                ("agent_tools", f"{AGENT_TYPE}:{intent}") for intent in INTENT_TOOLS
            }
        ]
        dropped = [item for item in inherited if item not in inherited_only]
        for item in dropped:
            print(f"  [覆盖] {item['namespace']}/{item['item_key']} 将由本次定义替换"
                  f"（原值 {item['value_json']}）")
        pending = [
            item for item in new_items
            if (str(item["namespace"]), str(item["item_key"])) not in
            {(str(i["namespace"]), str(i["item_key"])) for i in inherited_only}
        ]
        if not pending:
            print("客服白名单已存在于当前生效版本，无需发布")
            return 0

        created = await post(client, "/api/v1/admin/config-releases", auth=auth, payload={
            "release_no": f"cs-tools-{uuid.uuid4().hex[:12]}",
            "title": "客服 Agent 意图工具白名单",
            "change_summary": (
                "知识类意图补发访客检索工具 query_knowledge（访客令牌只有 knowledge:query），"
                "保留 search_knowledge 供登录客户使用，并继承既有配置项"
            ),
        })
        if created.status_code != 201:
            print(f"创建发布版本失败：{created.status_code} {created.text[:200]}")
            return 1
        release_id = int(created.json()["data"]["id"])
        print(f"\n发布版本 id={release_id}")

        base = f"/api/v1/admin/config-releases/{release_id}/platform-config-items"
        for item in [*inherited_only, *pending]:
            response = await post(client, base, auth=auth, payload=item)
            mark = "继承" if item in inherited_only else "新增"
            print(f"  [{mark}] {item['namespace']}/{item['item_key']} → {response.status_code}")
            if response.status_code != 201:
                print(f"    失败：{response.text[:200]}")
                return 1

        for payload in prompt_payloads(prompts, await next_prompt_versions(prompts), release_id):
            response = await post(
                client, "/api/v1/admin/prompt-templates", auth=auth, payload=payload
            )
            print(
                f"  [继承] 提示词 {payload['prompt_code']} → v{payload['version']}"
                f" → {response.status_code}"
            )
            if response.status_code not in (200, 201):
                print(f"    失败：{response.text[:200]}")
                return 1

        release_base = f"/api/v1/admin/config-releases/{release_id}"
        submitted = await post(
            client, f"{release_base}/validations", auth=auth, payload={},
            if_match=await etag_of(client, release_base, auth),
        )
        print(f"\n提交复核：{submitted.status_code}")
        reviewed = await post(
            client, f"{release_base}/reviews", auth=auth,
            payload={"decision": "approved", "comment": "客服工具白名单"},
            if_match=await etag_of(client, release_base, auth),
        )
        print(f"审核：{reviewed.status_code}")
        activated = await post(
            client, f"{release_base}/activations", auth=auth, payload={},
            if_match=await etag_of(client, release_base, auth),
        )
        print(f"激活：{activated.status_code}")
        if activated.status_code not in (200, 201):
            print(f"  失败：{activated.text[:200]}")
            return 1
        print(f"最终状态：{activated.json()['data']['status']}")

        after = await active_snapshot()
        remaining = snapshot_items(after)
        print(
            f"\n激活后生效版本：配置项 {len(remaining)} 条、"
            f"提示词 {len(after['prompt_template_version'])} 条"
        )
        for item in remaining:
            print(f"  · {item['namespace']} / {item['item_key']} = {item['value_json']}")
        # 条数对不上就是静默丢失 —— 这正是本脚本上一次踩的坑，必须硬校验。
        if len(remaining) != len(inherited_only) + len(pending):
            print("[失败] 配置项条数与预期不符，可能发生静默丢失，请人工核查")
            return 1
        if len(after["prompt_template_version"]) != len(prompts):
            print("[失败] 提示词条数与继承前不一致，可能发生静默丢失，请人工核查")
            return 1
        print("[OK] 完成")
        return 0


sys.exit(asyncio.run(main()))