feat: integrate customer service agent into ZSY develop

# Conflicts:
#	app/core/config.py
#	app/main.py
#	app/service/agent/bootstrap.py

tools/audit_schema.py

@@ -37,6 +37,14 @@ INCREMENTAL_TABLES = {
     "outbox_delivery",
     "svc_conversation_session",
     "api_request_receipt",
+    "offsite_fund_mail",
+    "offsite_mail_cursor",
+    "offsite_fund_attachment",
+    "offsite_fund_document",
+    "offsite_execution_plan_task",
+    "offsite_rule_result",
+    "offsite_query_record",
+    "offsite_notification",
 }
 
 

tools/dependency_health_check.py

@@ -25,7 +25,7 @@ async def main() -> None:
     finally:
         await driver.close()
     try:
-        connections.connect(alias="default", uri=settings.milvus_uri)
+        connections.connect(alias="default", uri=settings.resolved_milvus_uri)
         print("milvus", "ok")
     except Exception as exc:
         print("milvus", f"unavailable:{type(exc).__name__}")

tools/foundation_migration_preflight.py

@@ -0,0 +1,117 @@
+"""为无损底座迁移保存并校验工作区的只读 Git 状态证据。"""
+
+# 导入命令行参数解析器以支持生成和校验报告。
+import argparse
+# 导入 JSON 序列化工具以写入 UTF-8 状态证据。
+import json
+# 导入子进程工具以直接调用 Git 而不经过 shell。
+import subprocess
+# 导入可调用协议和类型别名支持。
+from collections.abc import Callable
+# 导入路径类型以约束工作区和报告位置。
+from pathlib import Path
+# 导入任意 JSON 对象的静态类型。
+from typing import Any
+
+# 定义只允许执行的 Git 只读子命令首参数集合。
+READ_ONLY_GIT_COMMANDS = frozenset({"branch", "rev-parse", "status"})
+# 定义便于测试注入的 Git 调用函数类型。
+GitRunner = Callable[..., str]
+
+
+# 执行受限的 Git 只读命令并返回标准输出文本。
+def run_git(worktree: Path, *args: str) -> str:
+    # 解析目标工作区以确保 Git 和安全目录使用同一个绝对路径。
+    resolved = worktree.resolve()
+    # 拒绝空命令，避免形成未约束的 Git 调用。
+    if not args:
+        raise ValueError("git command is required")
+    # 拒绝任何不在白名单中的 Git 子命令。
+    if args[0] not in READ_ONLY_GIT_COMMANDS:
+        raise ValueError("git command is not read-only")
+    # 将安全目录限定为当前查询工作区，避免写入全局 Git 配置。
+    safe_directory = resolved.as_posix()
+    # 以参数数组执行 Git，禁止 shell 解释路径或输入内容。
+    result = subprocess.run(
+        ["git", "-c", f"safe.directory={safe_directory}", "-C", str(resolved), *args],
+        check=True,
+        capture_output=True,
+        encoding="utf-8",
+        errors="replace",
+    )
+    # 返回 Git 的标准输出，供调用方以确定性方式解析。
+    return result.stdout
+
+
+# 收集一个工作区的分支、提交与简短状态，不读取环境变量或业务数据。
+def collect_workspace_state(worktree: Path, runner: GitRunner = run_git) -> dict[str, object]:
+    # 解析绝对路径，防止报告中出现随当前目录变化的相对路径。
+    resolved = worktree.resolve()
+    # 依次执行已白名单化的 Git 查询并构造安全状态字典。
+    return {
+        "path": str(resolved),
+        "branch": runner(resolved, "branch", "--show-current").strip(),
+        "head": runner(resolved, "rev-parse", "HEAD").strip(),
+        "status": runner(resolved, "status", "--short").splitlines(),
+    }
+
+
+# 将已收集的状态以 UTF-8 JSON 写入调用者指定的报告文件。
+def write_preflight_report(target: Path, states: list[dict[str, object]]) -> None:
+    # 确保报告父目录存在，但不创建或改动任何工作区内容。
+    target.parent.mkdir(parents=True, exist_ok=True)
+    # 使用稳定缩进和 UTF-8 编码写入仅由调用方提供的状态数据。
+    target.write_text(json.dumps(states, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+# 从磁盘读取先前报告并验证指定工作区状态完全一致。
+def verify_preflight_report(target: Path, worktrees: list[Path]) -> list[str]:
+    # 以 UTF-8 读取 JSON 报告，避免系统默认编码影响比较。
+    expected_value: Any = json.loads(target.read_text(encoding="utf-8"))
+    # 拒绝非列表报告，防止错误文件被误当作迁移证据。
+    if not isinstance(expected_value, list):
+        raise ValueError("preflight report must be a list")
+    # 为每个当前工作区重新采集只读 Git 状态。
+    current = [collect_workspace_state(worktree) for worktree in worktrees]
+    # 返回 JSON 表示不同的工作区路径，空列表代表完全一致。
+    return [
+        str(item["path"])
+        for item, expected in zip(current, expected_value, strict=True)
+        if item != expected
+    ]
+
+
+# 解析 CLI 工作区参数并执行报告生成或校验。
+def main() -> int:
+    # 创建命令行解析器并限定所有输入为显式路径。
+    parser = argparse.ArgumentParser(description=__doc__)
+    # 要求调用者指定报告文件路径。
+    parser.add_argument("--report", type=Path, required=True)
+    # 允许多次提供要采集或校验的工作区路径。
+    parser.add_argument("--worktree", type=Path, action="append", required=True)
+    # 启用校验模式时不覆盖报告。
+    parser.add_argument("--verify", action="store_true")
+    # 解析用户传入的参数。
+    arguments = parser.parse_args()
+    # 在校验模式下输出差异并返回非零状态。
+    if arguments.verify:
+        # 比较当前状态和报告状态。
+        differences = verify_preflight_report(arguments.report, arguments.worktree)
+        # 输出机器和人工都可识别的校验结果。
+        print("UNCHANGED" if not differences else f"CHANGED: {', '.join(differences)}")
+        # 有任何差异时返回失败状态。
+        return 0 if not differences else 1
+    # 收集调用方明确列出的工作区状态。
+    states = [collect_workspace_state(worktree) for worktree in arguments.worktree]
+    # 写入新的迁移前证据报告。
+    write_preflight_report(arguments.report, states)
+    # 输出报告保存位置，避免输出任何敏感运行配置。
+    print(f"WROTE: {arguments.report}")
+    # 报告创建成功时返回零状态。
+    return 0
+
+
+# 仅在脚本直接执行时运行命令行入口。
+if __name__ == "__main__":
+    # 用 main 的返回值作为进程退出码。
+    raise SystemExit(main())

tools/knowledge_import_preflight.py

@@ -0,0 +1,221 @@
+"""将一期公开 QA 候选转换为不含外部写入的待审核导入清单。"""
+
+import argparse
+import json
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Any
+
+from app.core.knowledge_contracts import ALLOWED_KNOWLEDGE_COLLECTIONS
+from app.service.knowledge_config import KnowledgeRuntimeConfig
+
+
+# 仅允许一期三类公开知识按既定路由进入后续发布流程。
+EXPECTED_COLLECTIONS = {
+    intent: collection
+    for intent, (collection, _top_k) in KnowledgeRuntimeConfig.DEFAULT_ROUTES.items()
+}
+# 公开候选必须具备的字段，缺失时不能生成不完整的发布清单。
+REQUIRED_PUBLIC_FIELDS = frozenset({
+    "qa_id",
+    "title",
+    "question",
+    "paraphrases",
+    "answer",
+    "scope",
+    "intent",
+    "collection",
+    "execution_mode",
+    "retrieval_status",
+    "audience",
+    "agent_data_access",
+    "tags",
+    "source_type",
+    "source_file",
+    "source_version",
+    "review_status",
+    "status",
+})
+
+
+def _required_string(record: Mapping[str, object], field: str, qa_id: str) -> str:
+    """读取非空字符串字段，拒绝将不完整资料带入后续发布阶段。"""
+    value = record.get(field)
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{qa_id}: {field} must be a non-empty string")
+    return value.strip()
+
+
+def _required_strings(record: Mapping[str, object], field: str, qa_id: str) -> list[str]:
+    """读取非空字符串数组，并保持来源中已经审核的条目顺序。"""
+    value = record.get(field)
+    if not isinstance(value, list) or not value:
+        raise ValueError(f"{qa_id}: {field} must be a non-empty string list")
+    strings = [item.strip() for item in value if isinstance(item, str) and item.strip()]
+    if len(strings) != len(value):
+        raise ValueError(f"{qa_id}: {field} must contain only non-empty strings")
+    return strings
+
+
+def _is_rule_only(record: Mapping[str, object]) -> bool:
+    """控制类记录必须由应用层固定路由处理，绝不能进入向量导入清单。"""
+    return (
+        record.get("collection") is None
+        and record.get("execution_mode") == "fixed_route"
+        and record.get("retrieval_status") == "rule_only"
+    )
+
+
+def _validate_public_record(record: Mapping[str, object]) -> None:
+    """验证候选是否符合一期公开知识与最小权限边界。"""
+    qa_id = _required_string(record, "qa_id", "<unknown>")
+    missing = sorted(field for field in REQUIRED_PUBLIC_FIELDS if field not in record)
+    if missing:
+        raise ValueError(f"{qa_id}: missing required fields: {', '.join(missing)}")
+    intent = _required_string(record, "intent", qa_id)
+    collection = _required_string(record, "collection", qa_id)
+    if intent not in EXPECTED_COLLECTIONS:
+        raise ValueError(f"{qa_id}: unsupported public intent: {intent}")
+    if collection not in ALLOWED_KNOWLEDGE_COLLECTIONS:
+        raise ValueError(f"{qa_id}: collection is not allowlisted: {collection}")
+    if collection != EXPECTED_COLLECTIONS[intent]:
+        raise ValueError(f"{qa_id}: collection does not match intent route")
+    if record.get("scope") != "public":
+        raise ValueError(f"{qa_id}: public vector record must have scope=public")
+    if record.get("execution_mode") != "vector_search":
+        raise ValueError(f"{qa_id}: public record must use vector_search")
+    if record.get("retrieval_status") != "approved_candidate":
+        raise ValueError(f"{qa_id}: record is not an approved candidate")
+    if record.get("review_status") != "approved_candidate":
+        raise ValueError(f"{qa_id}: review state cannot enter preflight")
+    if record.get("status") != "active":
+        raise ValueError(f"{qa_id}: inactive record cannot enter preflight")
+    if record.get("agent_data_access") != "none":
+        raise ValueError(f"{qa_id}: agent data access must remain none")
+    audience = _required_strings(record, "audience", qa_id)
+    if set(audience) != {"visitor", "authenticated_user"}:
+        raise ValueError(f"{qa_id}: audience must be visitor and authenticated_user")
+    _required_string(record, "title", qa_id)
+    _required_string(record, "question", qa_id)
+    _required_strings(record, "paraphrases", qa_id)
+    _required_string(record, "answer", qa_id)
+    _required_strings(record, "tags", qa_id)
+    _required_string(record, "source_type", qa_id)
+    _required_string(record, "source_file", qa_id)
+    _required_string(record, "source_version", qa_id)
+
+
+def _public_entry(record: Mapping[str, object]) -> dict[str, object]:
+    """构造供管理员审核的确定性条目，不生成数据库主键或向量。"""
+    qa_id = _required_string(record, "qa_id", "<unknown>")
+    question = _required_string(record, "question", qa_id)
+    paraphrases = _required_strings(record, "paraphrases", qa_id)
+    tags = _required_strings(record, "tags", qa_id)
+    answer = _required_string(record, "answer", qa_id)
+    content = {
+        "qa_id": qa_id,
+        "question": question,
+        "paraphrases": paraphrases,
+        "answer": answer,
+        "audience": _required_strings(record, "audience", qa_id),
+        "agent_data_access": "none",
+        "source_version": _required_string(record, "source_version", qa_id),
+    }
+    return {
+        "qa_id": qa_id,
+        "knowledge_type": _required_string(record, "intent", qa_id),
+        "title": _required_string(record, "title", qa_id),
+        "milvus_collection": _required_string(record, "collection", qa_id),
+        "version": _required_string(record, "source_version", qa_id),
+        "source_file": _required_string(record, "source_file", qa_id),
+        "source_type": _required_string(record, "source_type", qa_id),
+        "source_url": record.get("source_url"),
+        "effective_date": record.get("effective_date"),
+        "expire_date": record.get("expire_date"),
+        "tags": tags,
+        "content_text": json.dumps(content, ensure_ascii=False, separators=(",", ":")),
+        "retrieval_text": (
+            f"标准问题：{question}\n"
+            f"相似问法：{'；'.join(paraphrases)}\n"
+            f"标签：{'、'.join(tags)}"
+        ),
+        "snippet": question[:300],
+        # 管理员填入真实审核人并批准前，预检清单绝不伪装成已发布数据。
+        "review_status": "pending_review",
+        "status": "active",
+    }
+
+
+def build_import_manifest(
+    records: Sequence[Mapping[str, object]], *, source_name: str
+) -> dict[str, object]:
+    """构建可复查的导入清单，并拒绝任何不满足公开边界的非控制类记录。"""
+    entries: list[dict[str, object]] = []
+    seen_ids: set[str] = set()
+    excluded_rule_records = 0
+    for record in records:
+        if _is_rule_only(record):
+            excluded_rule_records += 1
+            continue
+        _validate_public_record(record)
+        entry = _public_entry(record)
+        qa_id = str(entry["qa_id"])
+        if qa_id in seen_ids:
+            raise ValueError(f"duplicate qa_id: {qa_id}")
+        seen_ids.add(qa_id)
+        entries.append(entry)
+    return {
+        "source_name": source_name,
+        "summary": {
+            "total_records": len(records),
+            "eligible_records": len(entries),
+            "excluded_rule_records": excluded_rule_records,
+            "publication_state": "pending_review",
+        },
+        "records": entries,
+    }
+
+
+def load_jsonl(source: Path) -> list[dict[str, object]]:
+    """读取 UTF-8 JSONL，并为每个无效 JSON 行返回带行号的明确错误。"""
+    records: list[dict[str, object]] = []
+    for line_number, line in enumerate(source.read_text(encoding="utf-8").splitlines(), start=1):
+        if not line.strip():
+            continue
+        try:
+            value: Any = json.loads(line)
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"line {line_number}: invalid JSON") from exc
+        if not isinstance(value, dict):
+            raise ValueError(f"line {line_number}: record must be an object")
+        records.append(value)
+    return records
+
+
+def write_manifest(target: Path, manifest: Mapping[str, object]) -> None:
+    """仅在调用者显式传入输出路径时，写入本地待审核 JSON 清单。"""
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(
+        json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+    )
+
+
+def main() -> int:
+    """执行本地预检；该入口不读取配置也不连接任何外部服务。"""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("--output", type=Path)
+    arguments = parser.parse_args()
+    manifest = build_import_manifest(
+        load_jsonl(arguments.input), source_name=arguments.input.name
+    )
+    if arguments.output is None:
+        print(json.dumps(manifest, ensure_ascii=False, indent=2))
+    else:
+        write_manifest(arguments.output, manifest)
+        print(f"WROTE: {arguments.output}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/publish_customer_service_knowledge.py

@@ -0,0 +1,283 @@
+"""管理员显式批准后发布一期客服公开知识；默认只验证清单，不写外部服务。"""
+
+import argparse
+import asyncio
+import json
+import sys
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, cast
+
+from sqlalchemy import select, text, update
+
+# 直接执行 tools 脚本时优先解析当前工作树，避免误导入相邻 worktree 的 app 包。
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from app.service.knowledge_publication_service import (  # noqa: E402
+    KnowledgePublicationRecord,
+    KnowledgePublicationService,
+)
+
+# 外部服务依赖仅在显式 --apply 时载入，dry-run 不需要本地 .env 或服务可达。
+SessionFactory: Any
+InteractionAudit: Any
+FinKnowledgeMeta: Any
+DatabaseModelGateway: Any
+get_settings: Any
+
+
+class DatabaseKnowledgeEmbedder:
+    """发布工具只走专用知识向量端点，禁止意外使用聊天端点。"""
+
+    async def embed(self, text_value: str) -> list[float]:
+        settings = get_settings()
+        endpoint_code = settings.knowledge_embedding_endpoint_code
+        if not endpoint_code:
+            raise RuntimeError("KNOWLEDGE_EMBEDDING_ENDPOINT_CODE 未配置")
+        return cast(
+            list[float],
+            await DatabaseModelGateway().embed(
+                endpoint_code=endpoint_code,
+                text=text_value,
+                timeout_ms=settings.knowledge_embedding_timeout_ms,
+            ),
+        )
+
+
+class SqlAlchemyKnowledgePublicationStore:
+    """使用现有知识表暂存、发布和停用记录，不变更数据库表结构。"""
+
+    def __init__(self, reviewer_id: int) -> None:
+        self._reviewer_id = reviewer_id
+
+    async def stage(self, records: tuple[KnowledgePublicationRecord, ...]) -> dict[str, int]:
+        now = datetime.now(UTC).replace(tzinfo=None)
+        async with SessionFactory() as session, session.begin():
+            await self._assert_reviewer(session)
+            await self._reject_existing_qa_ids(session, records)
+            rows: list[Any] = []
+            for record in records:
+                row = FinKnowledgeMeta(
+                    knowledge_type=str(record.metadata["knowledge_type"]),
+                    title=record.title,
+                    source_file=str(record.metadata["source_file"]),
+                    minio_path=None,
+                    milvus_collection=record.milvus_collection,
+                    version=record.version,
+                    effective_date=record.metadata.get("effective_date"),
+                    expire_date=record.metadata.get("expire_date"),
+                    content_text=str(record.metadata["content_text"]),
+                    tags=list(record.tags),
+                    reviewer_id=None,
+                    review_status="pending",
+                    status="disabled",
+                    created_at=now,
+                    updated_at=now,
+                )
+                session.add(row)
+                rows.append(row)
+            await session.flush()
+            session.add(InteractionAudit(
+                actor_type="admin",
+                actor_id=self._reviewer_id,
+                portal="admin",
+                action_type="knowledge.publication_staged",
+                detail={"qa_ids": [record.qa_id for record in records]},
+                created_at=now,
+            ))
+            return {record.qa_id: int(row.id) for record, row in zip(records, rows, strict=True)}
+
+    async def publish(self, knowledge_ids: tuple[int, ...], reviewer_id: int) -> None:
+        now = datetime.now(UTC).replace(tzinfo=None)
+        async with SessionFactory() as session, session.begin():
+            await session.execute(
+                update(FinKnowledgeMeta)
+                .where(FinKnowledgeMeta.id.in_(knowledge_ids))
+                .values(
+                    reviewer_id=reviewer_id,
+                    review_status="published",
+                    status="active",
+                    updated_at=now,
+                )
+            )
+            session.add(InteractionAudit(
+                actor_type="admin",
+                actor_id=reviewer_id,
+                portal="admin",
+                action_type="knowledge.publication_completed",
+                detail={"knowledge_ids": list(knowledge_ids)},
+                created_at=now,
+            ))
+
+    async def disable(self, knowledge_ids: tuple[int, ...]) -> None:
+        now = datetime.now(UTC).replace(tzinfo=None)
+        async with SessionFactory() as session, session.begin():
+            await session.execute(
+                update(FinKnowledgeMeta)
+                .where(FinKnowledgeMeta.id.in_(knowledge_ids))
+                .values(status="disabled", updated_at=now)
+            )
+            session.add(InteractionAudit(
+                actor_type="system",
+                actor_id=None,
+                portal="admin",
+                action_type="knowledge.publication_failed",
+                detail={"knowledge_ids": list(knowledge_ids)},
+                created_at=now,
+            ))
+
+    async def _assert_reviewer(self, session: Any) -> None:
+        row = await session.execute(
+            text(
+                "SELECT id FROM sys_user "
+                "WHERE id = :reviewer_id AND status IN ('正常', 'active') "
+                "AND user_type IN ('employee', 'admin')"
+            ),
+            {"reviewer_id": self._reviewer_id},
+        )
+        if row.scalar_one_or_none() is None:
+            raise RuntimeError("reviewer_id 不是有效的在职管理员或员工账号")
+
+    @staticmethod
+    async def _reject_existing_qa_ids(
+        session: Any, records: tuple[KnowledgePublicationRecord, ...]
+    ) -> None:
+        collections = tuple({record.milvus_collection for record in records})
+        rows = await session.scalars(
+            select(FinKnowledgeMeta)
+            .where(FinKnowledgeMeta.milvus_collection.in_(collections))
+            .with_for_update()
+        )
+        existing_ids: set[str] = set()
+        for row in rows:
+            try:
+                content = json.loads(row.content_text)
+            except json.JSONDecodeError:
+                continue
+            qa_id = content.get("qa_id") if isinstance(content, dict) else None
+            if isinstance(qa_id, str):
+                existing_ids.add(qa_id)
+        duplicates = sorted(existing_ids & {record.qa_id for record in records})
+        if duplicates:
+            raise RuntimeError(f"qa_id 已存在，拒绝重复发布: {', '.join(duplicates)}")
+
+
+class MilvusKnowledgePublicationStore:
+    """管理员发布期的最小 Milvus 写适配器；客服 Agent 运行期仍只能检索。"""
+
+    def __init__(self) -> None:
+        settings = get_settings()
+        self._uri = settings.resolved_milvus_uri
+        self._token = settings.milvus_token or None
+        self._client: Any | None = None
+
+    async def _client_instance(self) -> Any:
+        if self._client is None:
+            from pymilvus import AsyncMilvusClient  # type: ignore[import-untyped]
+
+            self._client = AsyncMilvusClient(uri=self._uri, token=self._token)
+        return self._client
+
+    async def upsert(self, collection: str, records: tuple[dict[str, object], ...]) -> None:
+        client = await self._client_instance()
+        payload = [
+            {**record, "tags": json.dumps(record["tags"], ensure_ascii=False)}
+            for record in records
+        ]
+        await client.upsert(collection_name=collection, data=payload)
+
+    async def delete(self, collection: str, knowledge_ids: tuple[str, ...]) -> None:
+        if not all(knowledge_id.isdecimal() for knowledge_id in knowledge_ids):
+            raise ValueError("knowledge_id 必须为十进制主键")
+        client = await self._client_instance()
+        values = ", ".join(json.dumps(knowledge_id) for knowledge_id in knowledge_ids)
+        await client.delete(collection_name=collection, filter=f"knowledge_id in [{values}]")
+
+
+def load_pending_manifest(value: Mapping[str, object]) -> tuple[KnowledgePublicationRecord, ...]:
+    """只接受预检工具输出的 pending_review 清单，拒绝手工伪造已发布状态。"""
+    summary = value.get("summary")
+    records = value.get("records")
+    if not isinstance(summary, dict) or summary.get("publication_state") != "pending_review":
+        raise ValueError("发布清单必须处于 pending_review 状态")
+    if not isinstance(records, list) or summary.get("eligible_records") != len(records):
+        raise ValueError("发布清单记录数与汇总不一致")
+    result: list[KnowledgePublicationRecord] = []
+    for raw in records:
+        if not isinstance(raw, dict):
+            raise ValueError("发布清单记录必须是对象")
+        required = ("qa_id", "milvus_collection", "retrieval_text", "title", "snippet", "version")
+        if any(not isinstance(raw.get(field), str) or not raw[field].strip() for field in required):
+            raise ValueError("发布清单缺少字符串字段")
+        tags = raw.get("tags")
+        if not isinstance(tags, list) or not all(isinstance(tag, str) and tag for tag in tags):
+            raise ValueError("发布清单标签无效")
+        if raw.get("review_status") != "pending_review" or raw.get("status") != "active":
+            raise ValueError("只有预检待审核记录可以发布")
+        result.append(KnowledgePublicationRecord(
+            qa_id=str(raw["qa_id"]),
+            milvus_collection=str(raw["milvus_collection"]),
+            retrieval_text=str(raw["retrieval_text"]),
+            title=str(raw["title"]),
+            snippet=str(raw["snippet"]),
+            tags=tuple(tags),
+            version=str(raw["version"]),
+            metadata=dict(raw),
+        ))
+    return tuple(result)
+
+
+def _read_manifest(path: Path) -> tuple[KnowledgePublicationRecord, ...]:
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError("发布清单根节点必须是对象")
+    return load_pending_manifest(value)
+
+
+async def _apply(records: tuple[KnowledgePublicationRecord, ...], reviewer_id: int) -> None:
+    global DatabaseModelGateway, FinKnowledgeMeta, InteractionAudit, SessionFactory, get_settings
+    from app.core.config import get_settings
+    from app.infrastructure.db import SessionFactory
+    from app.model.audit import InteractionAudit
+    from app.model.knowledge import FinKnowledgeMeta
+    from app.service.model_gateway import DatabaseModelGateway
+
+    service = KnowledgePublicationService(
+        DatabaseKnowledgeEmbedder(),
+        SqlAlchemyKnowledgePublicationStore(reviewer_id),
+        MilvusKnowledgePublicationStore(),
+    )
+    result = await service.publish(records, reviewer_id=reviewer_id)
+    print(
+        json.dumps(
+            {"published_records": len(result.knowledge_ids), "collections": result.collections},
+            ensure_ascii=False,
+        )
+    )
+
+
+def main() -> int:
+    """默认 dry-run，且 apply 必须三重确认，防止候选资料被误发布。"""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("--reviewer-id", type=int)
+    parser.add_argument("--apply", action="store_true")
+    parser.add_argument("--confirm-count", type=int)
+    arguments = parser.parse_args()
+    records = _read_manifest(arguments.input)
+    if not arguments.apply:
+        print(f"DRY RUN: {len(records)} records are pending administrator review")
+        return 0
+    if arguments.reviewer_id is None or arguments.reviewer_id <= 0:
+        raise SystemExit("--apply requires a positive --reviewer-id")
+    if arguments.confirm_count != len(records):
+        raise SystemExit("--apply requires --confirm-count equal to the manifest record count")
+    asyncio.run(_apply(records, arguments.reviewer_id))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/verify_customer_service_phase1.py

@@ -0,0 +1,165 @@
+"""只读核验一期客服 Agent 的数据库、知识库和向量运行环境。"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import text
+
+# 直接执行 tools 脚本时优先解析当前工作树，避免误导入相邻 worktree 的 app 包。
+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.infrastructure.db import SessionFactory  # noqa: E402
+
+COLLECTIONS = (
+    "fin_faq_collection",
+    "fin_product_collection",
+    "fin_policy_collection",
+)
+EXPECTED_KNOWLEDGE_COUNTS = {
+    "fin_faq_collection": 15,
+    "fin_product_collection": 26,
+    "fin_policy_collection": 11,
+}
+
+
+def _failures(values: Iterable[str]) -> list[str]:
+    """统一收集失败项，保证脚本最后一次性输出可操作结果。"""
+    return [value for value in values if value]
+
+
+async def _database_checks() -> list[str]:
+    """只读检查管理员、Embedding 端点、配置版本与知识发布状态。"""
+    failures: list[str] = []
+    async with SessionFactory() as session:
+        admin = await session.execute(text("""
+            SELECT id FROM sys_user
+            WHERE id = 9003 AND user_no = 'SYS-KNOWLEDGE-ADMIN'
+              AND user_type IN ('employee', 'admin')
+              AND status IN ('正常', 'active')
+        """))
+        if admin.scalar_one_or_none() is None:
+            failures.append("缺少启用的 SYS-KNOWLEDGE-ADMIN(9003)")
+
+        endpoint = await session.execute(text("""
+            SELECT endpoint_code, model_name, secret_ref, capabilities, status
+            FROM model_endpoint_config
+            WHERE endpoint_code = 'knowledge-embedding-qwen-v3'
+              AND status = 'active'
+        """))
+        endpoint_row = endpoint.mappings().first()
+        if endpoint_row is None:
+            failures.append("Qwen Embedding 端点未激活")
+        else:
+            if endpoint_row["model_name"] != "text-embedding-v3":
+                failures.append("Embedding 模型不是 text-embedding-v3")
+            if not str(endpoint_row["secret_ref"]).startswith("env:"):
+                failures.append("Embedding 密钥不是 env: 引用")
+
+        release = await session.execute(text("""
+            SELECT id FROM config_release
+            WHERE release_no = 'customer-service-phase1-public-kb-v1'
+              AND status = 'active'
+        """))
+        release_id = release.scalar_one_or_none()
+        if release_id is None:
+            failures.append("一期客服公开检索配置未激活")
+        else:
+            tools = await session.execute(text("""
+                SELECT config_key, value_json
+                FROM platform_config_item
+                WHERE release_id = :release_id AND namespace = 'agent_tools'
+            """), {"release_id": release_id})
+            configured: dict[str, Any] = {}
+            for row in tools.mappings():
+                raw_value = row["value_json"]
+                configured[str(row["config_key"])] = (
+                    json.loads(raw_value) if isinstance(raw_value, str) else raw_value
+                )
+            expected_keys = {
+                "customer_service:public_knowledge",
+                "customer_service:faq",
+                "customer_service:product_inquiry",
+                "customer_service:policy_explain",
+            }
+            if set(configured) != expected_keys:
+                failures.append("一期客服工具白名单缺失或包含额外意图")
+            if any(value != {"allowed_tools": ["query_knowledge"]} for value in configured.values()):
+                failures.append("一期客服工具白名单不是仅 query_knowledge")
+
+        knowledge = await session.execute(text("""
+            SELECT milvus_collection, review_status, status, COUNT(*) AS count
+            FROM fin_knowledge_meta
+            GROUP BY milvus_collection, review_status, status
+        """))
+        actual: dict[str, int] = {}
+        for row in knowledge.mappings():
+            if row["review_status"] == "published" and row["status"] == "active":
+                actual[str(row["milvus_collection"])] = int(row["count"])
+        if actual != EXPECTED_KNOWLEDGE_COUNTS:
+            failures.append(f"公开知识数量不符合预期: {actual}")
+    return failures
+
+
+async def _milvus_checks() -> list[str]:
+    """只读检查三类集合的存在、维度、主键和行数。"""
+    settings = get_settings()
+    failures: list[str] = []
+    try:
+        from pymilvus import AsyncMilvusClient  # type: ignore[import-untyped]
+
+        client: Any = AsyncMilvusClient(
+            uri=settings.resolved_milvus_uri, token=settings.milvus_token or None
+        )
+        for collection in COLLECTIONS:
+            if not await client.has_collection(collection_name=collection):
+                failures.append(f"集合不存在: {collection}")
+                continue
+            description = await client.describe_collection(collection_name=collection)
+            fields = {field["name"]: field for field in description.get("fields", [])}
+            embedding = fields.get("embedding", {})
+            if embedding.get("params", {}).get("dim") != 1024:
+                failures.append(f"集合 {collection} 不是 1024 维")
+            if not fields.get("knowledge_id", {}).get("is_primary"):
+                failures.append(f"集合 {collection} 缺少 knowledge_id 主键")
+            await client.load_collection(collection_name=collection)
+            stats = await client.get_collection_stats(collection_name=collection)
+            expected = EXPECTED_KNOWLEDGE_COUNTS[collection]
+            if int(stats.get("row_count", -1)) != expected:
+                failures.append(f"集合 {collection} 行数不符合预期: {stats}")
+        await client.close()
+    except Exception as exc:
+        failures.append(f"Milvus 检查失败: {type(exc).__name__}")
+    return failures
+
+
+async def verify() -> int:
+    """执行所有只读门禁并返回适合 CI 的退出码。"""
+    failures = _failures([*(await _database_checks()), *(await _milvus_checks())])
+    settings = get_settings()
+    print({
+        "milvus_uri_mode": "local" if settings.milvus_local_uri else "remote",
+        "knowledge_embedding_endpoint": settings.knowledge_embedding_endpoint_code,
+        "expected_public_records": sum(EXPECTED_KNOWLEDGE_COUNTS.values()),
+        "failures": failures,
+    })
+    return 1 if failures else 0
+
+
+def main() -> None:
+    """命令行入口；保留无参数形式，便于整合测试直接调用。"""
+    argparse.ArgumentParser(description=__doc__).parse_args()
+    raise SystemExit(asyncio.run(verify()))
+
+
+if __name__ == "__main__":
+    main()