lzf_0626
6516ccb385
相对第一版 46fc976 的完整变更。组员迁移对照表见 docs/20。
一、对外契约对齐 docs/05（破坏性，共 4 处，组员需按 docs/20 调整）
1) 配置发布端点改为文档规定的复数资源名：submit→validations、
approve→reviews（需 body decision）、activate→activations、
rollback→rollbacks；第一版这 4 个动词式路径 docs/05 从未定义过。
2) 错误码由 8 个笼统码改为 15 个具体语义码（FORBIDDEN→AGENT_PERMISSION_DENIED、
UNAUTHORIZED→AUTHENTICATION_REQUIRED、CONFLICT→RESOURCE_VERSION_CONFLICT、
RESOURCE_NOT_FOUND→RUN_NOT_FOUND/SESSION_NOT_FOUND 等），
输入类错误状态码 400→422。
3) POST /api/v1/agent-runs 与 GET /api/v1/agent-runs/{run_id} 统一为
{data, meta} 信封（data 内字段名与语义未变）。
4) 错误响应体统一为 {error:{code,message,retryable,field_errors}, meta:{trace_id}}，
不再返回 FastAPI 默认的 {"detail": ...}。
二、数据库基线与约束
新增 39 张表的基线迁移（链根）与联合唯一键纠偏（4 张表、删 8 增 4，幂等收敛）；
撤下 config_release 的双人复核 CHECK（应用层已允许自审，审核节点保留，
自审如实写入 reviewer_id）；记忆 active key 生成列与唯一键；
activate 开始记录 supersedes_release_id 使版本链可追溯。
docs/00 基线未修改，未重命名或删除任何表与字段。
三、修复会静默出错或无报错的缺陷
- 跑完集成测试后平台会静默失去生效配置：清理只删自己创建的版本，却没有恢复被它
顶成 superseded 的原生效版本，且审计一并删除因而完全无痕，表现为所有工具被拒
但没有任何报错。已修清理逻辑并加恢复。
- Worker 单轮异常导致进程退出；记忆抽取调用方的“事务已开始”异常；
召回缓存丢失 degraded 标记；连接时区未生效导致 created_at/updated_at 差 8 小时；
.env 与 os.getenv 密钥来源分裂导致“没有可用的已批准模型端点”。
- 记忆信号识别漏判与跨键误命中；SSE 未带 Accept 的协商行为。
四、功能补齐
记忆链路 P1/P2/P3（抽取、受控词表、召回与缓存、生命周期级联及投影事件）、
fin_* 场内交易只读 ORM 层、agent_intent_config 状态流转并在运行期真正生效、
限流（Redis 固定窗口、故障一律放行）、游标校验、trace_id 中间件、
示例业务 Agent fund_query_demo 与一键端到端验证脚本，以及审计/指纹/迁移状态工具。
五、文档与验证
新增 docs/19（业务 Agent 接入实操）、docs/20（第一版迁移指南）与 docs/evidence 证据；
docs/01/02/06/08/09/17 同步实现现状。
验证结果：ruff 通过、mypy 103 文件无错、unit+contract 447 passed、
integration 29 passed、acceptance_check --production 7 PASS、
demo_agent_e2e 9/9 PASS（含失败关闭反证）。
299 lines
14 KiB
Python
299 lines
14 KiB
Python
"""示例业务 Agent（fund_query_demo）的端到端接入验证脚本。
|
||
|
||
用途：机器化复现 docs/19 的接入步骤——发布配置 → 真实 JWT 受理 → Worker 执行 →
|
||
核对工具调用与审计 → **反证**（撤掉工具白名单后工具必须被拒）→ 恢复配置 → 清理测试数据。
|
||
其中意图配置会走完 `draft → approved → active`，因为运行期只读 `status='active'`。
|
||
|
||
用法（在仓库根目录、用项目解释器执行）：
|
||
|
||
python tools/demo_agent_e2e.py # 全流程并清理 run 数据
|
||
python tools/demo_agent_e2e.py --keep-run # 保留 run/审计数据便于排查
|
||
|
||
前置：数据库与配置中心已就绪；`config/jwt/jwt-private.pem` 存在；
|
||
**运行前请停掉常驻 Worker**，否则它会抢走本次 run 并使用同一队列的执行路径。
|
||
|
||
复核方式：配置发布保留"提交审核 → 审核 → 激活"状态机，但**不再要求审核人不是创建人**，
|
||
因此 9003 一个 admin 身份即可完成发布，无需再临时创建第二个复核身份。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
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 delete, select, text
|
||
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.main import create_app
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.conversation import ConversationMessage
|
||
from app.model.platform import AgentRun, DomainEventOutbox, OutboxDelivery, RequestIdempotency
|
||
from app.worker.runtime import WorkerRuntime
|
||
|
||
PRIVATE_KEY = Path("config/jwt/jwt-private.pem").read_text(encoding="utf-8")
|
||
ADMIN = "9003"
|
||
CUSTOMER = "9001"
|
||
AGENT_TYPE = "fund_query_demo"
|
||
INTENT = "fund_quote"
|
||
TOOL = "query_fund_quote"
|
||
MESSAGE = "帮我看下 159382 这只场内基金的行情"
|
||
|
||
_results: list[tuple[str, str, str]] = []
|
||
|
||
|
||
def record(name: str, expected: str, actual: str) -> None:
|
||
_results.append(("PASS" if expected == actual else "FAIL", name, f"期望 {expected} / 实际 {actual}"))
|
||
|
||
|
||
def token(sub: str) -> str:
|
||
now = dt.datetime.now(dt.UTC)
|
||
return jwt.encode(
|
||
{
|
||
"sub": sub, "iss": "jr-local", "aud": "jr-agent-platform",
|
||
"exp": now + dt.timedelta(minutes=30), "nbf": now - dt.timedelta(seconds=5),
|
||
"jti": str(uuid.uuid4()),
|
||
},
|
||
PRIVATE_KEY, algorithm="RS256",
|
||
)
|
||
|
||
|
||
def idempotency_key() -> str:
|
||
return uuid.uuid4().hex
|
||
|
||
|
||
async def etag_of(client: httpx.AsyncClient, release_id: int, auth: dict[str, str]) -> str:
|
||
response = await client.get(f"/api/v1/admin/config-releases/{release_id}", headers=auth)
|
||
if response.status_code != 200:
|
||
raise SystemExit(f"读取发布版本失败：{response.status_code} {response.text}")
|
||
etag = response.headers.get("ETag")
|
||
if not etag:
|
||
raise SystemExit("发布版本缺少 ETag")
|
||
return etag
|
||
|
||
|
||
async def post(client: httpx.AsyncClient, path: str, *, auth: dict[str, str],
|
||
payload: dict[str, Any] | None = None, if_match: str | None = None) -> httpx.Response:
|
||
headers = {**auth, "Idempotency-Key": idempotency_key()}
|
||
if if_match is not None:
|
||
headers["If-Match"] = if_match
|
||
return await client.post(path, json=payload if payload is not None else {}, headers=headers)
|
||
|
||
|
||
async def activate_release(
|
||
client: httpx.AsyncClient, admin_auth: dict[str, str], *, with_whitelist: bool,
|
||
) -> int:
|
||
"""走管理 API 完成一次完整发布：建版本 → 写配置 → 提交复核 → 自审 → 激活。
|
||
|
||
审核仍走 `reviews` 状态机节点，但审核人就是创建人（已取消双人复核约束）。
|
||
"""
|
||
created = await post(client, "/api/v1/admin/config-releases", auth=admin_auth, payload={
|
||
"release_no": f"demo-fund-{uuid.uuid4().hex[:12]}",
|
||
"title": f"示例 Agent {AGENT_TYPE} 接入配置",
|
||
"change_summary": "为示例 Agent 发布意图工具白名单",
|
||
})
|
||
if created.status_code != 201:
|
||
raise SystemExit(f"创建发布版本失败：{created.status_code} {created.text}")
|
||
release_id = int(created.json()["data"]["id"])
|
||
|
||
if with_whitelist:
|
||
item = await post(client, f"/api/v1/admin/config-releases/{release_id}/platform-config-items",
|
||
auth=admin_auth, payload={
|
||
"namespace": "agent_tools",
|
||
"item_key": f"{AGENT_TYPE}:{INTENT}",
|
||
"value_json": {"allowed_tools": [TOOL]},
|
||
"schema_version": "1",
|
||
})
|
||
if item.status_code != 201:
|
||
raise SystemExit(f"写入工具白名单失败：{item.status_code} {item.text}")
|
||
await ensure_intent_config(client, admin_auth)
|
||
|
||
submitted = await post(client, f"/api/v1/admin/config-releases/{release_id}/validations",
|
||
auth=admin_auth, if_match=await etag_of(client, release_id, admin_auth))
|
||
reviewed = await post(client, f"/api/v1/admin/config-releases/{release_id}/reviews",
|
||
auth=admin_auth, payload={"decision": "approved", "comment": "创建人自审"},
|
||
if_match=await etag_of(client, release_id, admin_auth))
|
||
activated = await post(client, f"/api/v1/admin/config-releases/{release_id}/activations",
|
||
auth=admin_auth, if_match=await etag_of(client, release_id, admin_auth))
|
||
print(f"[发布配置] release={release_id} 白名单={'有' if with_whitelist else '无'} "
|
||
f"submit={submitted.status_code} review={reviewed.status_code} "
|
||
f"activate={activated.status_code} status={activated.json()['data']['status']}")
|
||
return release_id
|
||
|
||
|
||
async def etag_of_row(
|
||
client: httpx.AsyncClient, path: str, row_id: int, auth: dict[str, str]
|
||
) -> str:
|
||
response = await client.get(f"{path}/{row_id}", headers=auth)
|
||
if response.status_code != 200:
|
||
raise SystemExit(f"读取资源失败：{response.status_code} {response.text}")
|
||
etag = response.headers.get("ETag")
|
||
if not etag:
|
||
raise SystemExit("资源缺少 ETag")
|
||
return etag
|
||
|
||
|
||
async def ensure_intent_config(client: httpx.AsyncClient, admin_auth: dict[str, str]) -> None:
|
||
"""意图配置属于交付物的一部分：不存在则创建，然后走"审核 → 生效"状态流转。
|
||
|
||
运行期只读 `status='active'` 的意图配置（`RuntimeConfigService.active_intents`），
|
||
因此这里必须把 `draft` 推成 `active`，否则配置只是存了一张表、分类链路看不到。
|
||
"""
|
||
path = "/api/v1/admin/agent-intent-configs"
|
||
listed = await client.get(f"{path}?limit=100", headers=admin_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") == INTENT),
|
||
None,
|
||
)
|
||
status = str(existing.get("status")) if existing else ""
|
||
if status == "active":
|
||
print(f"[意图配置] id={existing['id']} 已生效，跳过")
|
||
return
|
||
if existing is not None and 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=admin_auth, payload={
|
||
"agent_type": AGENT_TYPE, "intent_code": INTENT, "intent_name": "场内基金行情查询",
|
||
"description": "查询场内基金最新净值与涨跌信息",
|
||
"examples": ["帮我看看 159382 的行情", "查一下这只基金净值"],
|
||
"confidence_threshold": "0.6000", "allowed_tools": [TOOL], "version": version,
|
||
})
|
||
if created.status_code != 201:
|
||
raise SystemExit(f"创建意图配置失败：{created.status_code} {created.text}")
|
||
config_id = int(created.json()["data"]["id"])
|
||
for action, payload in (
|
||
("reviews", {"decision": "approved", "comment": "创建人自审"}),
|
||
("activations", None),
|
||
):
|
||
response = await post(
|
||
client, f"{path}/{config_id}/{action}", auth=admin_auth, payload=payload,
|
||
if_match=await etag_of_row(client, path, config_id, admin_auth),
|
||
)
|
||
if response.status_code != 200:
|
||
raise SystemExit(f"意图配置 {action} 失败：{response.status_code} {response.text}")
|
||
status = str(response.json()["data"]["status"])
|
||
print(f"[意图配置] id={config_id} 已生效（status={status}，运行期按 status='active' 读取）")
|
||
|
||
|
||
async def audit_rows(trace_id: str) -> list[dict[str, Any]]:
|
||
async with SessionFactory() as session:
|
||
rows = (await session.execute(text(
|
||
"SELECT id, action_type, detail FROM interaction_audit"
|
||
" WHERE JSON_UNQUOTE(JSON_EXTRACT(detail,'$.trace_id'))=:trace ORDER BY id"),
|
||
{"trace": trace_id})).mappings().all()
|
||
audits: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
detail = row["detail"]
|
||
if isinstance(detail, str):
|
||
detail = json.loads(detail)
|
||
audits.append({"id": row["id"], "action_type": row["action_type"], "detail": detail})
|
||
return audits
|
||
|
||
|
||
async def run_case(client: httpx.AsyncClient, *, label: str) -> tuple[str, str, dict[str, Any]]:
|
||
"""受理一次真实 run 并手动执行，返回（run_id, session_id, 证据字典）。"""
|
||
session_id = f"demo-fund-{uuid.uuid4()}"
|
||
auth = {"Authorization": f"Bearer {token(CUSTOMER)}"}
|
||
accepted = await client.post("/api/v1/agent-runs", json={
|
||
"agent_type": AGENT_TYPE, "message": MESSAGE, "session_id": session_id,
|
||
"idempotency_key": idempotency_key(),
|
||
}, headers=auth)
|
||
record(f"{label}：受理返回 202", "202", str(accepted.status_code))
|
||
if accepted.status_code != 202:
|
||
raise SystemExit(f"{label} 受理失败：{accepted.text}")
|
||
# 文档 §3.3/§6.2：受理响应也是 {data, meta} 信封。
|
||
run_id = str(accepted.json()["data"]["run_id"])
|
||
trace_id = str(accepted.json()["data"]["trace_id"])
|
||
|
||
await WorkerRuntime().execute(run_id)
|
||
# 文档 §3.3/§6.3：查询运行是 {data, meta} 信封，业务字段在 data 里。
|
||
body = (await client.get(f"/api/v1/agent-runs/{run_id}", headers=auth)).json()
|
||
detail: dict[str, Any] = body.get("data") or {}
|
||
core: dict[str, Any] = detail.get("result") or {}
|
||
rows = await audit_rows(trace_id)
|
||
tool_rows = [row for row in rows if row["action_type"] == "agent.tool_executed"]
|
||
print(f"[{label}] run={run_id} status={detail.get('status')} "
|
||
f"error_code={detail.get('error_code')}")
|
||
print(f" 工具调用={core.get('tool_calls')} 来源引用={core.get('source_references')}")
|
||
print(f" 审计 {len(rows)} 条：{[(row['action_type'], row['detail'].get('status')) for row in rows]}")
|
||
return run_id, session_id, {"detail": detail, "tool_rows": tool_rows, "core": core}
|
||
|
||
|
||
async def cleanup_run(session_id: str, run_id: str) -> None:
|
||
async with SessionFactory() as session, session.begin():
|
||
event_ids = select(DomainEventOutbox.event_id).where(
|
||
DomainEventOutbox.aggregate_id == run_id)
|
||
await session.execute(delete(OutboxDelivery).where(
|
||
OutboxDelivery.event_id.in_(event_ids)))
|
||
await session.execute(delete(DomainEventOutbox).where(
|
||
DomainEventOutbox.aggregate_id == run_id))
|
||
await session.execute(delete(InteractionAudit).where(
|
||
InteractionAudit.session_id == session_id))
|
||
await session.execute(delete(AgentRun).where(AgentRun.session_id == session_id))
|
||
await session.execute(delete(RequestIdempotency).where(
|
||
RequestIdempotency.session_id == session_id))
|
||
await session.execute(delete(ConversationMessage).where(
|
||
ConversationMessage.session_id == session_id))
|
||
print(f"[清理] 已删除 session={session_id} run={run_id} 及其消息/事件/审计/幂等回执")
|
||
|
||
|
||
async def main() -> None:
|
||
keep_run = "--keep-run" in sys.argv
|
||
app = create_app()
|
||
admin_auth = {"Authorization": f"Bearer {token(ADMIN)}"}
|
||
async with httpx.AsyncClient(
|
||
transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=60
|
||
) as client:
|
||
print("== 1. 发布并激活带工具白名单的版本 ==")
|
||
await activate_release(client, admin_auth, with_whitelist=True)
|
||
|
||
print("\n== 2. 正常路径：工具白名单生效 ==")
|
||
run_id, session_id, evidence = await run_case(client, label="正常路径")
|
||
detail, core = evidence["detail"], evidence["core"]
|
||
record("正常路径：run 终态", "succeeded", str(detail.get("status")))
|
||
calls = (core.get("tool_calls") or {}).get("calls") or []
|
||
record("正常路径：工具调用记录", "query_fund_quote/succeeded",
|
||
f"{calls[0].get('tool_name')}/{calls[0].get('status')}" if calls else "无")
|
||
references = core.get("source_references") or []
|
||
record("正常路径：来源引用", "tool", str(references[0].get("source_type")) if references else "无")
|
||
record("正常路径：工具审计", "succeeded",
|
||
str(evidence["tool_rows"][0]["detail"].get("status")) if evidence["tool_rows"] else "无")
|
||
if not keep_run:
|
||
await cleanup_run(session_id, run_id)
|
||
|
||
print("\n== 3. 反证：撤掉工具白名单后必须失败关闭 ==")
|
||
await activate_release(client, admin_auth, with_whitelist=False)
|
||
denied_run, denied_session, denied = await run_case(client, label="反证空白名单")
|
||
record("反证：run 终态", "failed", str(denied["detail"].get("status")))
|
||
record("反证：错误码", "AGENT_PERMISSION_DENIED", str(denied["detail"].get("error_code")))
|
||
record("反证：工具审计为 denied", "denied",
|
||
str(denied["tool_rows"][0]["detail"].get("status")) if denied["tool_rows"] else "无")
|
||
if not keep_run:
|
||
await cleanup_run(denied_session, denied_run)
|
||
|
||
print("\n== 4. 恢复带白名单的配置 ==")
|
||
await activate_release(client, admin_auth, with_whitelist=True)
|
||
|
||
width = max(len(name) for _, name, _ in _results)
|
||
print()
|
||
for verdict, name, detail_text in _results:
|
||
print(f"[{verdict}] {name.ljust(width)} {detail_text}")
|
||
failed = sum(1 for verdict, _, _ in _results if verdict == "FAIL")
|
||
print(f"\n合计 {len(_results)} 项，失败 {failed} 项")
|
||
if failed:
|
||
raise SystemExit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|