group_fqcd_jr/tools/publish_customer_service_tool_config.py

"""Publish the customer-service tool allowlist for the ``E2c-my`` eligible-product exit.

Why a release instead of an UPDATE: the active ``config_release`` is environment data, and
``call_tool`` resolves the per-intent allowlist from it.  ``E2c-my`` needs
``query_eligible_products`` under ``customer_service:suitability_check``; without this item the
tool call fails closed with ``AGENT_PERMISSION_DENIED`` and the exit degrades to ``E5b``
("can't get your risk level"), which looks like a data problem instead of a config problem.

The script carries forward every item of the active release and replaces only the
``agent_tools/customer_service:suitability_check`` entry, so nothing else changes.
"""

from __future__ import annotations

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

import asyncmy
import httpx
import jwt

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from app.core.config import get_settings  # noqa: E402
from app.main import create_app  # noqa: E402

ADMIN_ID = "9003"
AGENT_TYPE = "customer_service"

#: 只改这一个键。`faq` / `policy_explain` / `product_inquiry` 原样继承，不在本脚本里重写
#: —— 少写一个键就少一处"顺手改宽"的机会。
INTENT_TOOLS: dict[str, tuple[str, ...]] = {
    "suitability_check": (
        "search_knowledge",
        "check_suitability",
        "query_eligible_products",
    ),
}

#: ✅ `W27` / `DEC-W27-6`：`E6` 行情出口要用的工具，加进 `customer_service:faq`。
#:
#: **为什么加在 `faq` 键**：`query_customer_profile` 已经在那里，而发布配置里
#: `customer_service:faq` 是唯一一个"跨意图工具都挂上"的键；新开一个键会让
#: `customer_service:<intent>` 的工具交集在不同意图下不一致，调用会 fail-closed
#: —— 那会表现成"行情功能坏了"，而不是"配置少了"。
#:
#: ⚠️ 这个键**不从硬编码列表重建**，而是从**当前生效版本读出原列表再追加**（见
#: `_inherit_tools`）。理由：硬编码一份"当时恰好有哪些工具"的快照，会在别处增删工具后
#: 被下一次跑本脚本**静默回退** —— 那是比漏加一个工具危险得多的失败形态。
TREND_TOOL_ADDITIONS: dict[str, tuple[str, ...]] = {
    "faq": ("query_fund_trend",),
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Publish customer-service demo tool allowlist")
    parser.add_argument("--apply", action="store_true", help="Create, review, and activate release")
    return parser.parse_args()


def issue_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",
    )


async def active_items() -> list[dict[str, Any]]:
    settings = get_settings()
    dsn = settings.mysql_dsn.split("://", 1)[1]
    credentials, location = dsn.split("@", 1)
    user, password = credentials.split(":", 1)
    host_port, database = location.split("/", 1)
    host, _, port = host_port.partition(":")
    connection = await asyncmy.connect(
        host=host, port=int(port or 3306), user=user, password=password, db=database
    )
    try:
        cursor = connection.cursor()
        await cursor.execute(
            """
            SELECT i.namespace, i.config_key, i.value_json, i.schema_version
            FROM platform_config_item i
            JOIN config_release r ON r.id = i.release_id
            WHERE r.status = 'active'
            """
        )
        rows = await cursor.fetchall()
    finally:
        connection.close()
    return [
        {
            "namespace": namespace,
            "item_key": key,
            "value_json": json.loads(value) if isinstance(value, str) else value,
            "schema_version": schema_version,
        }
        for namespace, key, value, schema_version in rows
    ]


def _json_or_none(value: Any) -> Any:
    """库里的 JSON 列取出来可能是字符串 ``"null"`` / ``"{}"``，统一还原成对象。"""
    if value is None:
        return None
    if isinstance(value, str):
        parsed = json.loads(value)
        return None if parsed is None else parsed
    return value


async def active_prompt_templates() -> list[dict[str, Any]]:
    """取要继承的提示词模板行（`prompt_template_version`），版本号 +1。

    为什么必须有这一步：提示词模板与工具白名单**不在同一张表**。只继承
    `platform_config_item` 建新版本，会把提示词**静默丢掉** —— 激活接口只给一句告警，
    点下去那条提示词就真的失效了。实测 2026-09-20：`customer_service_chitchat`
    因此丢过一次（激活版里 0 行），闲聊出口当场退化成兜底话术。
    所以这里两级回退：**活跃版本有就继承活跃的；活跃版本没有，就从最近一个有该表的版本继承**
    （并把这件事打在屏幕上，避免"看起来一切正常"）。
    """
    settings = get_settings()
    dsn = settings.mysql_dsn.split("://", 1)[1]
    credentials, location = dsn.split("@", 1)
    user, password = credentials.split(":", 1)
    host_port, database = location.split("/", 1)
    host, _, port = host_port.partition(":")
    connection = await asyncmy.connect(
        host=host, port=int(port or 3306), user=user, password=password, db=database
    )
    columns = (
        "release_id, prompt_code, task_type, agent_type, version, "
        "system_prompt, user_prompt_template, input_schema, output_schema"
    )
    try:
        cursor = connection.cursor()
        await cursor.execute(
            f"SELECT {columns} FROM prompt_template_version WHERE release_id = "
            "(SELECT id FROM config_release WHERE status = 'active')"
        )
        rows = await cursor.fetchall()
        if not rows:
            print(
                "WARNING: active release has no prompt_template_version rows; "
                "inheriting from the newest release that does"
            )
            await cursor.execute(
                f"SELECT {columns} FROM prompt_template_version "
                "WHERE release_id = (SELECT MAX(release_id) FROM prompt_template_version)"
            )
            rows = await cursor.fetchall()
    finally:
        connection.close()
    return [
        {
            "prompt_code": prompt_code,
            "task_type": task_type,
            "agent_type": agent_type,
            "version": int(version) + 1,
            "system_prompt": system_prompt,
            "user_prompt_template": user_prompt_template,
            "input_schema": _json_or_none(input_schema),
            "output_schema": _json_or_none(output_schema),
        }
        for (
            _release_id,
            prompt_code,
            task_type,
            agent_type,
            version,
            system_prompt,
            user_prompt_template,
            input_schema,
            output_schema,
        ) in rows
    ]


async def request(
    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(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> str | None:
    return (await client.get(path, headers=auth)).headers.get("ETag")


def _inherit_tools(
    inherited: list[dict[str, Any]], intent: str, additions: tuple[str, ...]
) -> tuple[str, ...] | None:
    """读出生效版本里 `customer_service:<intent>` 的工具列表，再追加 `additions`。

    取不到生效条目时返回 `None`（**不猜**）：调用方会跳过该键并在日志里说明。
    这里刻意不"取不到就用 additions 新建一个" —— 那会把 `faq` 键原有的十来个
    工具一次性砍掉，是最坏的一种"顺手改窄"。
    """
    key = f"{AGENT_TYPE}:{intent}"
    for item in inherited:
        if str(item.get("namespace")) != "agent_tools" or str(item.get("item_key")) != key:
            continue
        value = item.get("value_json")
        if isinstance(value, str):
            # `active_items()` 已经解析过一次，这里是防御性分支。
            # 解析失败**不抛**：抛出去会让整个发布失败，而"发布失败"与
            # "跳过该键、让新版本原样继承旧条目"相比，后者才是"最接近没改动"的结果
            # （失败时 `customer_service:faq` 仍是旧值，安全，但操作者会以为脚本坏了）。
            try:
                value = json.loads(value)
            except json.JSONDecodeError:
                return None
        if not isinstance(value, dict):
            return None
        tools = value.get("allowed_tools")
        if not isinstance(tools, list):
            return None
        merged = [str(tool) for tool in tools]
        for tool in additions:
            if tool not in merged:
                merged.append(tool)
        return tuple(merged)
    return None


def cs_items() -> list[dict[str, object]]:
    return [
        {
            "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()
    ]


async def publish() -> int:
    app = create_app()
    auth = {"Authorization": f"Bearer {issue_token(ADMIN_ID)}"}
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        inherited = await active_items()
        prompt_templates = await active_prompt_templates()
        items = cs_items()
        for intent, additions in TREND_TOOL_ADDITIONS.items():
            merged = _inherit_tools(inherited, intent, additions)
            if merged is None:
                print(f"skip: 生效版本里没有 agent_tools/{AGENT_TYPE}:{intent}，不新建以免改窄")
                continue
            items.append({
                "namespace": "agent_tools",
                "item_key": f"{AGENT_TYPE}:{intent}",
                "value_json": {"allowed_tools": list(merged)},
                "schema_version": "1",
            })
            print(f"{AGENT_TYPE}:{intent} -> {', '.join(merged)}")
        replacement_keys = {("agent_tools", str(item["item_key"])) for item in items}
        carried = [
            item
            for item in inherited
            if (str(item["namespace"]), str(item["item_key"])) not in replacement_keys
        ]
        created = await request(
            client,
            "/api/v1/admin/config-releases",
            auth=auth,
            payload={
                "release_no": f"cs-tools-{uuid.uuid4().hex[:12]}",
                "title": "Customer service tool allowlist (E2c-my eligible + E6 fund trend)",
                "change_summary": (
                    "Add query_eligible_products to suitability_check; add query_fund_trend to faq"
                ),
            },
        )
        if created.status_code != 201:
            print(f"create release failed: {created.status_code} {created.text[:240]}")
            return 1
        release_id = int(created.json()["data"]["id"])
        base = f"/api/v1/admin/config-releases/{release_id}"
        for item in [*carried, *items]:
            response = await request(
                client, f"{base}/platform-config-items", auth=auth, payload=item
            )
            if response.status_code != 201:
                print(
                    f"write {item['namespace']}/{item['item_key']} failed: "
                    f"{response.status_code} {response.text[:400]}"
                )
                return 1
        for template in prompt_templates:
            response = await request(
                client,
                # 提示词模板**不走** release 作用域前缀：`register_resource` 对它用的
                # 是 `scoped=False`，路由挂在 `/api/v1/admin/prompt-templates`，
                # release 由 body 里的 `release_id` 指定（实测 2026-09-20 踩过 404）。
                "/api/v1/admin/prompt-templates",
                auth=auth,
                payload={**template, "release_id": release_id},
            )
            if response.status_code != 201:
                print(
                    f"write prompt {template['prompt_code']} failed: "
                    f"{response.status_code} {response.text[:240]}"
                )
                return 1
        for suffix, payload in (
            ("validations", {}),
            ("reviews", {"decision": "approved", "comment": "add eligible-product + fund-trend tools"}),
            ("activations", {}),
        ):
            response = await request(
                client,
                f"{base}/{suffix}",
                auth=auth,
                payload=payload,
                if_match=await etag(client, base, auth),
            )
            if response.status_code not in (200, 201):
                print(f"{suffix} failed: {response.status_code} {response.text[:240]}")
                return 1
        print(f"published customer-service tool release id={release_id}")
    return 0


def main() -> int:
    args = parse_args()
    for intent, tools in INTENT_TOOLS.items():
        print(f"customer_service:{intent} -> {', '.join(tools)}")
    if not args.apply:
        print("dry run only; pass --apply to publish a new config release")
        return 0
    return asyncio.run(publish())


if __name__ == "__main__":
    raise SystemExit(main())