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（含失败关闭反证）。
447 lines
21 KiB
Python
447 lines
21 KiB
Python
from datetime import UTC, date, datetime
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy import delete
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.core.cursor import parse_cursor
|
||
from app.core.errors import (
|
||
GenericResourceNotFoundError,
|
||
InvalidStateError,
|
||
ResourceAlreadyExistsError,
|
||
ResourceVersionConflictError,
|
||
ValidationAgentError,
|
||
)
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.platform import DomainEventOutbox
|
||
from app.repository.platform_repository import PlatformRepository
|
||
from app.service.agent.bootstrap import get_agent_factory
|
||
from app.service.api_transaction_service import ApiTransactionService, digest
|
||
from app.service.authorization_service import AuthorizationService
|
||
from app.service.config_release_service import ConfigReleaseService
|
||
|
||
RESOURCES = {
|
||
"config-releases": "config_release",
|
||
"platform-config-items": "platform_config_item",
|
||
"model-endpoints": "model_endpoint_config",
|
||
"model-routing-rules": "model_routing_rule",
|
||
"prompt-templates": "prompt_template_version",
|
||
"agent-intent-configs": "agent_intent_config",
|
||
"reply-templates": "agent_reply_template",
|
||
"negative-word-rules": "agent_negative_word",
|
||
"audit-records": "interaction_audit",
|
||
}
|
||
|
||
# 行情 namespace 的允许字段。必须与 `FundQuoteRuntimeConfig.from_mapping` 的
|
||
# 接受范围保持一致，否则会出现"配置能发布、运行期却被静默忽略"的假成功。
|
||
# 该 namespace 由 `RuntimeConfigService.fund_quote` 以
|
||
# namespace=fund_market / config_key=default 读取。
|
||
FUND_MARKET_FIELDS = frozenset({"allowed_codes", "intraday_cache_ttl", "closing_cache_ttl"})
|
||
|
||
|
||
def validate_fund_market(values: dict[str, Any]) -> None:
|
||
"""校验行情配置取值。
|
||
|
||
行情配置不是 1-100 的整数表：`allowed_codes` 是六位代码字符串列表，
|
||
TTL 上限 86400 与运行期接受范围一致（`FundQuoteRuntimeConfig`）。
|
||
"""
|
||
codes = values.get("allowed_codes", [])
|
||
if not isinstance(codes, list) or not all(
|
||
isinstance(code, str) and len(code) == 6 and code.isdigit() for code in codes
|
||
):
|
||
raise ValidationAgentError("行情基金代码必须为六位数字字符串列表")
|
||
for name in ("intraday_cache_ttl", "closing_cache_ttl"):
|
||
if name not in values:
|
||
continue
|
||
ttl = values[name]
|
||
if type(ttl) is not int or not 1 <= ttl <= 86400:
|
||
raise ValidationAgentError(f"{name} 必须为 1-86400 之间的整数")
|
||
|
||
|
||
def public(value: Any, key: str = "") -> Any:
|
||
if isinstance(value, dict):
|
||
return {
|
||
("item_key" if k == "config_key" else k): public(v, k)
|
||
for k, v in value.items()
|
||
if k not in {"secret_ref", "fallback_endpoint_ids", "active_slot", "active_key"}
|
||
}
|
||
if isinstance(value, list):
|
||
return [public(item) for item in value]
|
||
if isinstance(value, datetime):
|
||
return value.isoformat() + ("Z" if value.tzinfo is None else "")
|
||
if isinstance(value, (Decimal, date)):
|
||
return str(value)
|
||
if isinstance(value, int) and (key == "id" or key.endswith("_id") or key == "created_by"):
|
||
return str(value)
|
||
return value
|
||
|
||
|
||
class AdminService:
|
||
async def query(
|
||
self,
|
||
resource: str,
|
||
context: RequestContext,
|
||
row_id: int | None = None,
|
||
release_id: int | None = None,
|
||
limit: int = 20,
|
||
cursor: str | None = None,
|
||
) -> dict[str, Any]:
|
||
permission = "audit:read" if resource == "audit-records" else "config:read"
|
||
await AuthorizationService.require(context, permission, admin=True)
|
||
# 游标校验必须在权限闸门**之后**：未授权调用一律 403，不能因为参数格式先漏出
|
||
# 一个 400——响应差异本身就是一条越权探测信号（文档 §4.2）。校验同时早于
|
||
# 任何数据库访问，非法游标不会变成一次静默的全量查询。
|
||
before = parse_cursor(cursor)
|
||
async with SessionFactory() as session:
|
||
repo = PlatformRepository(session)
|
||
filters = {"release_id": release_id} if release_id is not None else {}
|
||
if row_id is not None:
|
||
row = await repo.get(RESOURCES[resource], row_id)
|
||
if row is None or any(row[k] != v for k, v in filters.items()):
|
||
raise GenericResourceNotFoundError("资源不存在")
|
||
return {
|
||
"data": public(row),
|
||
"meta": {"trace_id": context.trace_id, "etag": digest(row)},
|
||
}
|
||
rows = await repo.rows(RESOURCES[resource], filters, before=before, limit=limit)
|
||
if resource == "audit-records" and "audit:read-sensitive" not in context.permissions:
|
||
rows = [{**row, "detail": {"redacted": True}} for row in rows]
|
||
return {"data": public(rows), "meta": {"trace_id": context.trace_id}}
|
||
|
||
async def mutate(
|
||
self,
|
||
resource: str,
|
||
context: RequestContext,
|
||
values: dict[str, Any],
|
||
key: str | None,
|
||
if_match: str | None,
|
||
*,
|
||
row_id: int | None = None,
|
||
release_id: int | None = None,
|
||
action: str = "write",
|
||
) -> dict[str, Any]:
|
||
permission = "model-endpoint:manage" if resource == "model-endpoints" else "config:write"
|
||
if action == "reviews":
|
||
permission = "config:review"
|
||
elif action in {"activations", "rollbacks"} and resource == "config-releases":
|
||
permission = "config:activate"
|
||
await AuthorizationService.require(context, permission, admin=True)
|
||
|
||
async def operation(session: AsyncSession) -> dict[str, Any]:
|
||
repo = PlatformRepository(session)
|
||
table = RESOURCES[resource]
|
||
existing = await repo.get(table, row_id, lock=True) if row_id is not None else None
|
||
if row_id is not None:
|
||
if existing is None or (
|
||
release_id is not None and existing.get("release_id") != release_id
|
||
):
|
||
raise GenericResourceNotFoundError("资源不存在")
|
||
if if_match is None or if_match.strip('"') != digest(existing):
|
||
raise ResourceVersionConflictError("If-Match 版本不一致")
|
||
if action != "write":
|
||
assert existing is not None
|
||
row = await self._transition(repo, resource, existing, action, values, context)
|
||
else:
|
||
row = await self._write(repo, resource, existing, values, release_id, context)
|
||
session.add(
|
||
InteractionAudit(
|
||
actor_type="user",
|
||
actor_id=int(context.user_id),
|
||
portal="admin",
|
||
action_type=f"platform.{resource}.{action}",
|
||
detail={
|
||
"resource_id": row["id"],
|
||
"trace_id": context.trace_id,
|
||
"before_hash": digest(existing) if existing else None,
|
||
"after_hash": digest(row),
|
||
},
|
||
created_at=datetime.now(UTC).replace(tzinfo=None),
|
||
)
|
||
)
|
||
return {
|
||
"data": public(row),
|
||
"meta": {"trace_id": context.trace_id, "etag": digest(row)},
|
||
}
|
||
|
||
try:
|
||
return await ApiTransactionService().execute(
|
||
context,
|
||
f"admin:{resource}:{row_id}:{release_id}:{action}",
|
||
key,
|
||
{"body": values, "if_match": if_match},
|
||
operation,
|
||
)
|
||
except IntegrityError as exc:
|
||
raise ResourceAlreadyExistsError("资源唯一性、引用或状态约束冲突") from exc
|
||
|
||
async def _write(
|
||
self,
|
||
repo: PlatformRepository,
|
||
resource: str,
|
||
existing: dict[str, Any] | None,
|
||
values: dict[str, Any],
|
||
release_id: int | None,
|
||
context: RequestContext,
|
||
) -> dict[str, Any]:
|
||
data = dict(values)
|
||
if existing is not None and existing.get("status", "draft") not in {"draft", "disabled"}:
|
||
raise InvalidStateError("只能编辑草稿或停用版本")
|
||
if release_id is not None:
|
||
data["release_id"] = release_id
|
||
if "release_id" in data:
|
||
parent = await repo.get("config_release", int(data["release_id"]), lock=True)
|
||
if parent is None or parent["status"] != "draft":
|
||
raise InvalidStateError("只能修改草稿发布批次")
|
||
if resource == "platform-config-items":
|
||
data["config_key"] = data.pop("item_key")
|
||
allowed = {
|
||
"memory": {"recall_limit", "decay_days"},
|
||
"relationship": {"max_hops", "limit"},
|
||
"runtime": {"timeout_seconds"},
|
||
"agent_tools": {"allowed_tools"},
|
||
"fund_market": set(FUND_MARKET_FIELDS),
|
||
}
|
||
if set(data["value_json"]) - allowed[data["namespace"]]:
|
||
raise ValidationAgentError("配置包含未声明字段")
|
||
if data["namespace"] == "agent_tools":
|
||
raw = data["value_json"].get("allowed_tools", [])
|
||
if not isinstance(raw, list) or not all(isinstance(v, str) for v in raw):
|
||
raise ValidationAgentError("工具白名单必须为字符串列表")
|
||
agent_type, separator, intent = data["config_key"].partition(":")
|
||
definition = get_agent_factory().definition(agent_type)
|
||
if not separator or intent not in definition.supported_intents:
|
||
raise ValidationAgentError("意图不在 Agent 声明内")
|
||
if not set(raw) <= set(definition.allowed_tools):
|
||
raise ValidationAgentError("配置超出 Agent 工具上限")
|
||
else:
|
||
for value in data["value_json"].values():
|
||
if type(value) is not int or not 1 <= value <= 100:
|
||
raise ValidationAgentError("配置值须为 1-100 整数")
|
||
data["checksum"] = digest(data["value_json"])
|
||
if resource == "model-endpoints":
|
||
if "@" in data["base_url"] or "?" in data["base_url"]:
|
||
raise ValidationAgentError("端点地址不得包含凭证或查询参数")
|
||
if resource == "agent-intent-configs":
|
||
definition = get_agent_factory().definition(data["agent_type"])
|
||
if data["intent_code"] not in definition.supported_intents:
|
||
raise ValidationAgentError("意图未注册")
|
||
if not set(data["allowed_tools"]) <= set(definition.allowed_tools):
|
||
raise ValidationAgentError("工具超出代码上限")
|
||
fallbacks = data.pop("fallbacks", [])
|
||
if resource == "model-routing-rules":
|
||
await self._validate_routing(repo, data, fallbacks)
|
||
if resource == "prompt-templates":
|
||
data["checksum"] = digest(data)
|
||
table = await repo.table(RESOURCES[resource])
|
||
if "created_by" in table.c and existing is None:
|
||
data["created_by"] = int(context.user_id)
|
||
if "status" in table.c:
|
||
data["status"] = "draft"
|
||
if "reviewer_id" in table.c:
|
||
data.update(reviewer_id=None, reviewed_at=None)
|
||
if "updated_at" in table.c:
|
||
data["updated_at"] = datetime.now(UTC).replace(tzinfo=None)
|
||
row = (
|
||
await repo.update(RESOURCES[resource], existing["id"], data)
|
||
if existing
|
||
else await repo.create(RESOURCES[resource], data)
|
||
)
|
||
if resource == "model-routing-rules":
|
||
fallback_table = await repo.table("model_routing_fallback")
|
||
# Draft child relationships may be replaced; no historical column/schema mutation.
|
||
await repo.session.execute(
|
||
delete(fallback_table).where(fallback_table.c.routing_rule_id == row["id"])
|
||
)
|
||
for fallback in fallbacks:
|
||
await repo.create(
|
||
"model_routing_fallback", {**fallback, "routing_rule_id": row["id"]}
|
||
)
|
||
row["fallbacks"] = fallbacks
|
||
return row
|
||
|
||
async def _validate_routing(
|
||
self, repo: PlatformRepository, data: dict[str, Any], fallbacks: list[dict[str, Any]]
|
||
) -> None:
|
||
get_agent_factory().definition(data["agent_type"])
|
||
ids = [data["primary_endpoint_id"], *(item["endpoint_id"] for item in fallbacks)]
|
||
orders = [item["fallback_order"] for item in fallbacks]
|
||
if len(ids) != len(set(ids)) or sorted(orders) != list(range(1, len(orders) + 1)):
|
||
raise ValidationAgentError("备用端点重复或顺序不连续")
|
||
if data["max_attempts"] > len(ids):
|
||
raise ValidationAgentError("重试次数超过端点数量")
|
||
for endpoint_id in ids:
|
||
endpoint = await repo.get("model_endpoint_config", endpoint_id, lock=True)
|
||
if endpoint is None or endpoint["status"] != "active":
|
||
raise ValidationAgentError("模型端点不存在或未激活")
|
||
|
||
async def _transition(
|
||
self,
|
||
repo: PlatformRepository,
|
||
resource: str,
|
||
row: dict[str, Any],
|
||
action: str,
|
||
values: dict[str, Any],
|
||
context: RequestContext,
|
||
) -> dict[str, Any]:
|
||
actor = int(context.user_id)
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
if resource == "config-releases":
|
||
service = ConfigReleaseService(repo.session)
|
||
if action == "validations":
|
||
await self._validate_release(repo, row["id"])
|
||
result = await service.submit_for_review(row["id"], actor)
|
||
elif action == "reviews":
|
||
if values["decision"] == "rejected":
|
||
# 允许创建人自审（不再要求 reviewer 与 created_by 不同）；仍然强制审核节点。
|
||
# 自审时 reviewer_id 留空：库约束 chk_config_release_separation 是基线的一部分，
|
||
# 自审人身份由 interaction_audit 的 reviews 审计（actor_id）承担。
|
||
if row["status"] != "pending_review":
|
||
raise InvalidStateError("只能驳回待审核版本")
|
||
return await repo.update(
|
||
"config_release",
|
||
row["id"],
|
||
{
|
||
"status": "rejected",
|
||
"reviewer_id": None if row["created_by"] == actor else actor,
|
||
"reviewed_at": now,
|
||
},
|
||
)
|
||
result = await service.approve(row["id"], actor)
|
||
elif action == "activations":
|
||
await self._validate_release(repo, row["id"])
|
||
result = await service.activate(row["id"], actor)
|
||
elif action == "rollbacks":
|
||
result = await service.rollback(row["id"], actor)
|
||
await self._copy_release(repo, row["id"], result.id)
|
||
else:
|
||
raise ValidationAgentError("未知发布操作")
|
||
refreshed = await repo.get("config_release", result.id)
|
||
assert refreshed is not None
|
||
return refreshed
|
||
if resource == "agent-intent-configs":
|
||
return await self._transition_intent_config(repo, row, action, values, actor, now)
|
||
if action == "reviews":
|
||
# 审核节点不可跳过（必须先是 draft）；但允许创建人自审。
|
||
if row["status"] != "draft":
|
||
raise InvalidStateError("只能审核草稿版本")
|
||
status = "approved" if values["decision"] == "approved" else "draft"
|
||
update_values = {"status": status, "reviewer_id": actor, "reviewed_at": now}
|
||
elif action == "activations":
|
||
if row["status"] != "approved" or row["reviewer_id"] is None:
|
||
raise InvalidStateError("版本未审核")
|
||
update_values = {"status": "active"}
|
||
elif action == "disablements":
|
||
if row["status"] != "active":
|
||
raise InvalidStateError("只能停用激活版本")
|
||
update_values = {"status": "disabled"}
|
||
else:
|
||
raise ValidationAgentError("未知状态操作")
|
||
update_values["updated_at"] = now
|
||
return await repo.update(RESOURCES[resource], row["id"], update_values)
|
||
|
||
async def _transition_intent_config(
|
||
self,
|
||
repo: PlatformRepository,
|
||
row: dict[str, Any],
|
||
action: str,
|
||
values: dict[str, Any],
|
||
actor: int,
|
||
now: datetime,
|
||
) -> dict[str, Any]:
|
||
"""意图配置自己的状态机：`draft -> approved -> active -> archived`。
|
||
|
||
`agent_intent_config.status` 有 CHECK 约束，只允许
|
||
`draft/approved/active/archived`（**没有 `disabled`**），所以这里不能套用
|
||
`model-endpoints` 的 `disablements` 语义，归档落到 `archived`。运行期仍以
|
||
`status='active'` 判定生效（`RuntimeConfigService.active_intents`），口径一致。
|
||
|
||
同一 `agent_type:intent_code` 只能有一个 active 版本（生成列唯一键
|
||
`uk_intent_config_active_one`），激活前先把旧版本归档，语义与
|
||
`ConfigReleaseService.activate` 挂起上一个生效版本一致。
|
||
"""
|
||
if action == "reviews":
|
||
if row["status"] != "draft":
|
||
raise InvalidStateError("只能审核草稿意图配置")
|
||
update_values: dict[str, Any] = {
|
||
"status": "approved" if values["decision"] == "approved" else "draft",
|
||
"reviewer_id": actor,
|
||
"reviewed_at": now,
|
||
}
|
||
elif action == "activations":
|
||
if row["status"] != "approved" or row["reviewer_id"] is None:
|
||
raise InvalidStateError("意图配置未审核")
|
||
expire_at = row.get("expire_at")
|
||
if isinstance(expire_at, datetime) and expire_at <= now:
|
||
raise InvalidStateError("意图配置已过期，不能激活")
|
||
for previous in await repo.rows(
|
||
"agent_intent_config",
|
||
{"agent_type": row["agent_type"], "status": "active"},
|
||
limit=10,
|
||
):
|
||
if previous["id"] != row["id"]:
|
||
await repo.update(
|
||
"agent_intent_config",
|
||
previous["id"],
|
||
{"status": "archived", "updated_at": now},
|
||
)
|
||
update_values = {"status": "active"}
|
||
if row.get("effective_at") is None:
|
||
update_values["effective_at"] = now
|
||
elif action == "archivals":
|
||
if row["status"] != "active":
|
||
raise InvalidStateError("只能归档生效版本")
|
||
update_values = {"status": "archived"}
|
||
else:
|
||
raise ValidationAgentError("未知状态操作")
|
||
update_values["updated_at"] = now
|
||
return await repo.update("agent_intent_config", row["id"], update_values)
|
||
|
||
async def _validate_release(self, repo: PlatformRepository, release_id: int) -> None:
|
||
rules = await repo.rows("model_routing_rule", {"release_id": release_id}, limit=10000)
|
||
for rule in rules:
|
||
fallbacks = await repo.rows(
|
||
"model_routing_fallback", {"routing_rule_id": rule["id"]}, limit=3
|
||
)
|
||
await self._validate_routing(repo, rule, fallbacks)
|
||
# Checksums catch out-of-band draft edits before approval/activation.
|
||
for item in await repo.rows(
|
||
"platform_config_item", {"release_id": release_id}, limit=10000
|
||
):
|
||
if item["checksum"] != digest(item["value_json"]):
|
||
raise ValidationAgentError("配置校验和不一致")
|
||
|
||
async def _copy_release(self, repo: PlatformRepository, source: int, target: int) -> None:
|
||
for name in ("platform_config_item", "prompt_template_version", "model_routing_rule"):
|
||
for row in await repo.rows(name, {"release_id": source}, limit=10000):
|
||
old_id = row.pop("id")
|
||
row.pop("created_at", None)
|
||
row["release_id"] = target
|
||
copied = await repo.create(name, row)
|
||
if name == "model_routing_rule":
|
||
for fallback in await repo.rows(
|
||
"model_routing_fallback", {"routing_rule_id": old_id}, limit=3
|
||
):
|
||
fallback.pop("id")
|
||
fallback.pop("created_at", None)
|
||
fallback["routing_rule_id"] = copied["id"]
|
||
await repo.create("model_routing_fallback", fallback)
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
repo.session.add(
|
||
DomainEventOutbox(
|
||
event_id=str(uuid4()),
|
||
event_type="config.cache_invalidate_requested",
|
||
aggregate_type="config_release",
|
||
aggregate_id=str(target),
|
||
trace_id=str(uuid4()),
|
||
payload={"release_id": target},
|
||
occurred_at=now,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
)
|