feat: 第二版——接口契约对齐 docs/05，修复静默故障与数据库基线
相对第一版 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（含失败关闭反证）。
This commit is contained in:
@@ -0,0 +1,264 @@
|
||||
"""记忆抽取集成测试（真实 MySQL）。
|
||||
|
||||
P2 之后记忆链路的语义变化：
|
||||
|
||||
- 记忆键来自**受控词表**（如 `preference:risk_level`），不再是 `conversation.{session_id}`；
|
||||
- 记忆内容是模型抽取出的**结构化值**，而不是用户原文整句。
|
||||
|
||||
抽取依赖模型端点，而本地 `model_endpoint_config` 为 0 行，端点缺失时按设计失败关闭。
|
||||
因此本文件注入确定性替身抽取器，只验证与模型供应商无关的契约：正文回查、幂等边界、
|
||||
客户隔离、证据落库与清理。
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import delete, select
|
||||
|
||||
from app.infrastructure.db import SessionFactory
|
||||
from app.model.conversation import ConversationMessage
|
||||
from app.model.memory import MemoryConflict, MemoryEvidence, MemoryUnit
|
||||
from app.model.platform import AgentRun, DomainEventOutbox, OutboxDelivery, RequestIdempotency
|
||||
from app.service.memory_extraction_service import ExtractedMemory
|
||||
from app.service.memory_service import MemoryService
|
||||
from app.worker.memory_extraction_worker import MemoryExtractionWorker
|
||||
from app.worker.outbox_worker import OutboxWorker
|
||||
|
||||
KEY_PREFIX = "it-memory-"
|
||||
MEMORY_KEY = "preference:risk_level"
|
||||
EXTRACTED_VALUE = "稳健型"
|
||||
USER_FACT = f"{KEY_PREFIX}我偏好低风险稳健型基金"
|
||||
OTHER_FACT = f"{KEY_PREFIX}本人可承受中等风险"
|
||||
THIRD_FACT = f"{KEY_PREFIX}本人只做货币基金"
|
||||
PLAN_FACT = f"{KEY_PREFIX}我计划两年内买房"
|
||||
ASSISTANT_REPLY = "已为您记录该偏好。"
|
||||
|
||||
|
||||
class StubExtractor:
|
||||
"""确定性替身抽取器：记录收到的正文，返回固定抽取结果，不发网络请求。"""
|
||||
|
||||
def __init__(self, value: str = EXTRACTED_VALUE) -> None:
|
||||
self.value = value
|
||||
self.messages: list[str] = []
|
||||
|
||||
async def extract(
|
||||
self, *, message: str, agent_type: str = "customer_service"
|
||||
) -> ExtractedMemory:
|
||||
del agent_type
|
||||
self.messages.append(message)
|
||||
return ExtractedMemory(
|
||||
memory_key=MEMORY_KEY,
|
||||
value=self.value,
|
||||
memory_type="preference",
|
||||
confidence=0.9,
|
||||
)
|
||||
|
||||
|
||||
class Case:
|
||||
def __init__(self, customer_id: int, run_id: str, session_id: str) -> None:
|
||||
self.customer_id = customer_id
|
||||
self.run_id = run_id
|
||||
self.session_id = session_id
|
||||
|
||||
@property
|
||||
def memory_key(self) -> str:
|
||||
"""抽取结果对应的受控键；同一客户的记忆键与客户一一对应，便于隔离断言。"""
|
||||
return MEMORY_KEY
|
||||
|
||||
|
||||
def new_case(offset: int) -> Case:
|
||||
return Case(
|
||||
customer_id=uuid4().int % 10**14 + offset * 10**14,
|
||||
run_id=str(uuid4()),
|
||||
session_id=f"{KEY_PREFIX}{uuid4().hex[:24]}",
|
||||
)
|
||||
|
||||
|
||||
async def seed(case: Case, *, user_fact: str) -> int:
|
||||
"""建立 run 到用户消息与助手消息的真实链路，事件只携带 message_id。"""
|
||||
now = datetime.now(UTC).replace(tzinfo=None)
|
||||
async with SessionFactory() as session, session.begin():
|
||||
request_message = ConversationMessage(
|
||||
session_id=case.session_id, customer_id=case.customer_id, portal="api", role="user",
|
||||
content=user_fact, trace_id=str(uuid4()), created_at=now,
|
||||
)
|
||||
session.add(request_message)
|
||||
await session.flush()
|
||||
result_message = ConversationMessage(
|
||||
session_id=case.session_id, customer_id=case.customer_id, portal="agent",
|
||||
role="assistant", content=ASSISTANT_REPLY, trace_id=str(uuid4()), created_at=now,
|
||||
)
|
||||
session.add(result_message)
|
||||
await session.flush()
|
||||
idempotency = RequestIdempotency(
|
||||
user_id=case.customer_id, session_id=case.session_id, agent_type="customer_service",
|
||||
idempotency_key=str(uuid4()), request_hash=uuid4().hex * 2, trace_id=str(uuid4()),
|
||||
status="completed", expire_at=now, created_at=now, updated_at=now,
|
||||
)
|
||||
session.add(idempotency)
|
||||
await session.flush()
|
||||
session.add(AgentRun(
|
||||
run_id=case.run_id, idempotency_id=idempotency.id, session_id=case.session_id,
|
||||
user_id=case.customer_id, agent_type="customer_service", trace_id=str(uuid4()),
|
||||
request_message_id=request_message.id, result_message_id=result_message.id,
|
||||
status="succeeded", attempt_count=1, created_at=now, updated_at=now,
|
||||
))
|
||||
return result_message.id
|
||||
|
||||
|
||||
async def enqueue(case: Case, message_id: int, *, event_id: str) -> None:
|
||||
now = datetime.now(UTC).replace(tzinfo=None)
|
||||
async with SessionFactory() as session, session.begin():
|
||||
session.add(DomainEventOutbox(
|
||||
event_id=event_id, event_type="memory.extraction_requested",
|
||||
aggregate_type="agent_run", aggregate_id=case.run_id, trace_id=str(uuid4()),
|
||||
payload={"run_id": case.run_id, "message_id": message_id,
|
||||
"customer_id": case.customer_id},
|
||||
status="pending", retry_count=0, occurred_at=now, created_at=now, updated_at=now,
|
||||
))
|
||||
|
||||
|
||||
async def consume_once(case: Case, extractor: StubExtractor) -> bool:
|
||||
async with SessionFactory() as session:
|
||||
worker = OutboxWorker(session, {
|
||||
"memory.extraction_requested": MemoryExtractionWorker(
|
||||
session, extractor=extractor
|
||||
).handle,
|
||||
})
|
||||
return await worker.publish_one(aggregate_id=case.run_id)
|
||||
|
||||
|
||||
async def realtime_seen_count(extractor: StubExtractor, fact: str) -> int:
|
||||
return sum(1 for message in extractor.messages if message == fact)
|
||||
|
||||
|
||||
async def cleanup(cases: list[Case]) -> None:
|
||||
run_ids = [case.run_id for case in cases]
|
||||
session_ids = [case.session_id for case in cases]
|
||||
customer_ids = [case.customer_id for case in cases]
|
||||
async with SessionFactory() as session, session.begin():
|
||||
message_ids = list(await session.scalars(select(ConversationMessage.id).where(
|
||||
ConversationMessage.session_id.in_(session_ids))))
|
||||
memory_ids = list(await session.scalars(select(MemoryUnit.id).where(
|
||||
MemoryUnit.customer_id.in_(customer_ids))))
|
||||
# 测试证据只可能挂在本次测试的会话消息或本次测试客户的记忆上。
|
||||
await session.execute(delete(MemoryEvidence).where(
|
||||
(MemoryEvidence.source_record_id.in_([str(item) for item in message_ids]))
|
||||
| (MemoryEvidence.memory_id.in_(memory_ids))))
|
||||
await session.execute(delete(MemoryConflict).where(
|
||||
(MemoryConflict.left_memory_id.in_(memory_ids))
|
||||
| (MemoryConflict.right_memory_id.in_(memory_ids))))
|
||||
await session.execute(delete(MemoryUnit).where(
|
||||
MemoryUnit.customer_id.in_(customer_ids)))
|
||||
event_ids = select(DomainEventOutbox.event_id).where(
|
||||
DomainEventOutbox.aggregate_id.in_(run_ids))
|
||||
await session.execute(delete(OutboxDelivery).where(OutboxDelivery.event_id.in_(event_ids)))
|
||||
await session.execute(delete(DomainEventOutbox).where(
|
||||
DomainEventOutbox.aggregate_id.in_(run_ids)))
|
||||
await session.execute(delete(AgentRun).where(AgentRun.run_id.in_(run_ids)))
|
||||
await session.execute(delete(RequestIdempotency).where(
|
||||
RequestIdempotency.session_id.in_(session_ids)))
|
||||
await session.execute(delete(ConversationMessage).where(
|
||||
ConversationMessage.session_id.in_(session_ids)))
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_duplicate_consumption_produces_single_memory() -> None:
|
||||
case = new_case(9)
|
||||
extractor = StubExtractor()
|
||||
event_id = str(uuid4())
|
||||
try:
|
||||
message_id = await seed(case, user_fact=USER_FACT)
|
||||
await enqueue(case, message_id, event_id=event_id)
|
||||
|
||||
assert await consume_once(case, extractor)
|
||||
# 重复投递同一事件：事件回到 pending 后再次消费。
|
||||
async with SessionFactory() as session, session.begin():
|
||||
event = await session.scalar(select(DomainEventOutbox).where(
|
||||
DomainEventOutbox.event_id == event_id))
|
||||
assert event is not None
|
||||
event.status = "pending"
|
||||
event.published_at = None
|
||||
assert await consume_once(case, extractor)
|
||||
|
||||
async with SessionFactory() as session:
|
||||
memories = list(await session.scalars(select(MemoryUnit).where(
|
||||
MemoryUnit.customer_id == case.customer_id,
|
||||
MemoryUnit.memory_key == case.memory_key)))
|
||||
assert len(memories) == 1
|
||||
# 落库的是抽取结果，不是用户原文整句。
|
||||
assert memories[0].content == EXTRACTED_VALUE
|
||||
assert memories[0].content != USER_FACT
|
||||
assert memories[0].memory_type == "preference"
|
||||
assert memories[0].evidence_count <= 1
|
||||
evidence = list(await session.scalars(select(MemoryEvidence).where(
|
||||
MemoryEvidence.idempotency_key == f"memory.extraction_requested:{event_id}")))
|
||||
assert len(evidence) == 1
|
||||
assert evidence[0].memory_id == memories[0].id
|
||||
assert evidence[0].source_record_id is not None
|
||||
finally:
|
||||
await cleanup([case])
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_memory_is_not_recallable_across_customers() -> None:
|
||||
owner = new_case(8)
|
||||
other = new_case(7)
|
||||
# 两个客户用不同的抽取值，才能用内容区分"谁记住了什么"——受控键相同，
|
||||
# 单靠 memory_key 无法判定归属。
|
||||
owner_extractor = StubExtractor("稳健型")
|
||||
other_extractor = StubExtractor("激进型")
|
||||
try:
|
||||
owner_message = await seed(owner, user_fact=OTHER_FACT)
|
||||
other_message = await seed(other, user_fact=THIRD_FACT)
|
||||
await enqueue(owner, owner_message, event_id=str(uuid4()))
|
||||
await enqueue(other, other_message, event_id=str(uuid4()))
|
||||
assert await consume_once(owner, owner_extractor)
|
||||
assert await consume_once(other, other_extractor)
|
||||
|
||||
async with SessionFactory() as session:
|
||||
service = MemoryService(session)
|
||||
owner_memories = await service.recall(owner.customer_id)
|
||||
other_memories = await service.recall(other.customer_id)
|
||||
assert [item.memory_key for item in owner_memories] == [owner.memory_key]
|
||||
assert [item.memory_key for item in other_memories] == [other.memory_key]
|
||||
assert all(item.customer_id == owner.customer_id for item in owner_memories)
|
||||
assert all(item.customer_id == other.customer_id for item in other_memories)
|
||||
assert owner_memories[0].content == "稳健型"
|
||||
assert other_memories[0].content == "激进型"
|
||||
# 客户范围过滤：即使知道他人的内容，也查不到挂在他人名下的行。
|
||||
assert await session.scalar(select(MemoryUnit.id).where(
|
||||
MemoryUnit.customer_id == owner.customer_id,
|
||||
MemoryUnit.content == "激进型")) is None
|
||||
assert await session.scalar(select(MemoryUnit.id).where(
|
||||
MemoryUnit.customer_id == other.customer_id,
|
||||
MemoryUnit.content == "稳健型")) is None
|
||||
finally:
|
||||
await cleanup([owner, other])
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_worker_reads_user_content_not_assistant_reply() -> None:
|
||||
case = new_case(6)
|
||||
extractor = StubExtractor()
|
||||
try:
|
||||
message_id = await seed(case, user_fact=PLAN_FACT)
|
||||
await enqueue(case, message_id, event_id=str(uuid4()))
|
||||
assert await consume_once(case, extractor)
|
||||
|
||||
# 抽取器收到的必须是库中的用户消息正文，而不是助手回复。
|
||||
assert extractor.messages == [PLAN_FACT]
|
||||
assert ASSISTANT_REPLY not in extractor.messages[0]
|
||||
|
||||
async with SessionFactory() as session:
|
||||
memory = await session.scalar(select(MemoryUnit).where(
|
||||
MemoryUnit.customer_id == case.customer_id,
|
||||
MemoryUnit.memory_key == case.memory_key))
|
||||
assert memory is not None
|
||||
assert memory.content == EXTRACTED_VALUE
|
||||
finally:
|
||||
await cleanup([case])
|
||||
Reference in New Issue
Block a user