qyqy
a7e2d3eac0
两个都是"代码里存在但没接对"的缺陷，都不是新功能。
## A1 客服热线在代码里有两个值（一个出口给假号码）
- `app/core/customer_service_rules.py:35` `CONTACT_PHONE = "15936583816"` ← 真号码，安全路由 6 处在用
- `app/service/agent/implementations/customer_service.py:155` `HOTLINE = "400-XXX-XXXX"` ← 占位符，兜底出口在用
后果：**同一个客服给客户两个不同的电话号码**。问"风险等级怎么划分"被安全路由处理时给真号码；
问一个知识库答不了的问题走兜底时给 `400-XXX-XXXX` —— 客户按这个号码永远打不通。
修法：`HOTLINE` / `SERVICE_HOURS` 改为**转发** `customer_service_rules` 的两个常量
（不是"改成相同的值"，而是引用同一对象，避免日后再次漂移）；工作时间也随之从
"每日 7:00-22:00" 统一为 "工作日 09:00-18:00"（与安全路由出口一致）。
新增守卫测试用 `is` 断言对象同一性 —— 值相等挡不住"两边各写一份恰好相同"的漂移。
## A2 `transfer_required` 既没落库也没出参
`docs/05` §6.3 一直规定 `GET /agent-runs/{run_id}` 的 `result` 里有
`transfer_required` / `transfer_reason`，但实现里两个都没有：前端判断"这轮要不要转人工"
只能靠**猜正文里有没有兜底话术的开头**（`docs/24` 自己把这称为权宜之计）。
- 写入侧：`conversation_message` **没有** `transfer_required` 列，加列要迁移且规则 4 禁止改既有
字段定义 ⇒ 放进 `tool_calls` 这个现成 JSON 列，作为 `calls` 的兄弟键
（`{"calls": [...], "transfer_required": bool, "transfer_reason": str|None}`）
- 读取侧：`RunQueryService.get` 取出来放进 `result`；**兼容历史行**（`calls` 裸列表 / None →
按 False/None 处理，不抛异常、也不凭正文猜）
刻意**没做**的一半：`docs/05` §6.3 的 `result` 里还有 `degraded` / `degradation_reason`，
但 `CoreResult` 里根本没有这两个字段（降级信息目前只在工具出参里）—— 补它要改
`CoreResult` 并让各 Agent 传递降级状态，属另一个改动范围。**已在交付说明里注明这一半仍缺。**
## 真机验证
| 问题 | transfer_required | transfer_reason | 正文电话 |
|---|---|---|---|
「请介绍一下量子纠缠在基金估值中的应用」 | **True** | 置信度不足：score=0.571 gap=0.004 | 15936583816 ✅ |
「请帮我计算一下三体问题的数值解」 | **True** | 置信度不足：score=0.499 gap=0.011 | 15936583816 ✅ |
「基金申购后多久确认」（正常知识直返） | False | — | 无（正确） |
「你们公司明天会下雪吗」（闲聊出口） | False | — | 无（正确） |
（第一次我用"下雪"当兜底用例，结果它被闲聊出口正确接住了 —— 是我的期望值写错，不是代码问题。）
门禁：测试 1223 passed（新增 4 个用例）/ 3 failed（均为已知非代码缺陷）/ mypy 0 错。
242 lines
9.0 KiB
Python
242 lines
9.0 KiB
Python
"""`RunQueryService` 契约测试（全部使用替身，不连数据库）。
|
||
|
||
关注三条对客户端与安全都重要的语义：
|
||
1. 运行不存在与越权访问**返回同一个 404**——不能靠错误码区分"存在但无权"；
|
||
2. 未成功的运行**不返回任何结果内容**（`result` 为 None），避免未提交结果外泄；
|
||
3. 快照里的时间统一以 `...Z` 结尾（UTC），以及 `watch` 的心跳与终态退出行为。
|
||
"""
|
||
|
||
from dataclasses import dataclass, replace
|
||
from datetime import datetime
|
||
from decimal import Decimal
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.core.errors import RunNotFoundError
|
||
from app.service.run_query_service import RunQueryService, RunSnapshot
|
||
|
||
CONTEXT = RequestContext(user_id="9001", trace_id="trace-1")
|
||
NOW = datetime(2026, 9, 10, 12, 0, 0)
|
||
|
||
|
||
@dataclass
|
||
class FakeRun:
|
||
run_id: str = "run-1"
|
||
trace_id: str = "trace-1"
|
||
status: str = "running"
|
||
agent_type: str = "customer_service"
|
||
session_id: str = "session-1"
|
||
error_code: str | None = None
|
||
created_at: datetime = NOW
|
||
completed_at: datetime | None = None
|
||
|
||
|
||
@dataclass
|
||
class FakeMessage:
|
||
content: str = "稳健型"
|
||
tool_calls: Any = None
|
||
intent: str | None = "general"
|
||
confidence: Any = None
|
||
source_references: Any = None
|
||
|
||
|
||
class FakeRepository:
|
||
def __init__(self, rows: tuple[Any, Any] | None) -> None:
|
||
self.rows = rows
|
||
|
||
async def run_result(self, run_id: str, user_id: int) -> tuple[Any, Any] | None:
|
||
del run_id, user_id
|
||
return self.rows
|
||
|
||
|
||
class FakeSession:
|
||
async def __aenter__(self) -> "FakeSession":
|
||
return self
|
||
|
||
async def __aexit__(self, *exc: object) -> bool:
|
||
return False
|
||
|
||
|
||
def patch_repository(monkeypatch: pytest.MonkeyPatch, rows: tuple[Any, Any] | None) -> None:
|
||
monkeypatch.setattr("app.service.run_query_service.SessionFactory", FakeSession)
|
||
monkeypatch.setattr(
|
||
"app.service.run_query_service.ConversationRepository",
|
||
lambda _session: FakeRepository(rows),
|
||
)
|
||
|
||
|
||
async def test_unknown_run_raises_not_found(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""不存在与越权必须是同一个错误，否则可以据此探测资源是否存在。"""
|
||
patch_repository(monkeypatch, None)
|
||
|
||
with pytest.raises(RunNotFoundError) as excinfo:
|
||
await RunQueryService().get("run-x", CONTEXT)
|
||
|
||
assert excinfo.value.code == "RUN_NOT_FOUND"
|
||
assert excinfo.value.status_code == 404
|
||
|
||
|
||
async def test_running_run_does_not_expose_result(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
patch_repository(monkeypatch, (FakeRun(status="running"), None))
|
||
|
||
snapshot = await RunQueryService().get("run-1", CONTEXT)
|
||
|
||
assert snapshot.status == "running"
|
||
assert snapshot.result is None
|
||
assert snapshot.completed_at is None
|
||
assert snapshot.created_at.endswith("Z")
|
||
|
||
|
||
async def test_succeeded_run_exposes_result_with_string_confidence(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
message = FakeMessage(
|
||
content="稳健型",
|
||
tool_calls={"calls": []},
|
||
confidence=Decimal("0.9000"),
|
||
source_references=[{"source_type": "memory"}],
|
||
)
|
||
patch_repository(monkeypatch, (FakeRun(status="succeeded", completed_at=NOW), message))
|
||
|
||
snapshot = await RunQueryService().get("run-1", CONTEXT)
|
||
|
||
assert snapshot.result is not None
|
||
assert snapshot.result["content"] == "稳健型"
|
||
# DECIMAL(5,4) 经驱动回来是 Decimal，接口层统一转字符串避免浮点精度歧义。
|
||
assert snapshot.result["confidence"] == "0.9000"
|
||
assert snapshot.result["source_references"] == [{"source_type": "memory"}]
|
||
assert snapshot.completed_at is not None and snapshot.completed_at.endswith("Z")
|
||
|
||
|
||
async def test_failed_run_does_not_expose_result(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""即使库里有结果消息，失败运行也不得把它当作成功结果返回。"""
|
||
patch_repository(
|
||
monkeypatch, (FakeRun(status="failed", error_code="AGENT_INTERNAL_ERROR"), FakeMessage())
|
||
)
|
||
|
||
snapshot = await RunQueryService().get("run-1", CONTEXT)
|
||
|
||
assert snapshot.result is None
|
||
assert snapshot.error_code == "AGENT_INTERNAL_ERROR"
|
||
|
||
|
||
# --- `docs/05` §6.3 要求的 `transfer_required` / `transfer_reason` 出参 -----------------
|
||
|
||
|
||
async def test_result_exposes_transfer_marker_when_transferred(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""兜底/转人工分支必须把标记**出参**给客户端。
|
||
|
||
为什么这条重要：前端判断"这轮要不要转人工"此前只能靠**猜正文里有没有兜底话术开头**
|
||
（`docs/24` 自称权宜之计）。`docs/05` §6.3 一直规定 `result` 里有这两个字段，
|
||
但此前既没落库也没出参 —— 这个用例把"契约已兑现"钉住。
|
||
"""
|
||
message = FakeMessage(
|
||
content="抱歉，这个问题我暂时无法给出准确答复…",
|
||
tool_calls={"calls": [], "transfer_required": True,
|
||
"transfer_reason": "置信度不足：score=0.571 gap=0.004"},
|
||
)
|
||
patch_repository(monkeypatch, (FakeRun(status="succeeded", completed_at=NOW), message))
|
||
|
||
snapshot = await RunQueryService().get("run-1", CONTEXT)
|
||
|
||
assert snapshot.result is not None
|
||
assert snapshot.result["transfer_required"] is True
|
||
assert snapshot.result["transfer_reason"] == "置信度不足：score=0.571 gap=0.004"
|
||
|
||
|
||
async def test_result_transfer_marker_defaults_to_false(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""正常回答：标记为 False、原因为 None（不能因为缺键就返回 None 让客户端混淆）。"""
|
||
message = FakeMessage(content="交易日 15:00 前提交…",
|
||
tool_calls={"calls": [], "transfer_required": False,
|
||
"transfer_reason": None})
|
||
patch_repository(monkeypatch, (FakeRun(status="succeeded", completed_at=NOW), message))
|
||
|
||
snapshot = await RunQueryService().get("run-1", CONTEXT)
|
||
|
||
assert snapshot.result is not None
|
||
assert snapshot.result["transfer_required"] is False
|
||
assert snapshot.result["transfer_reason"] is None
|
||
|
||
|
||
async def test_result_tolerates_legacy_tool_calls_shape(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""**兼容历史行**：本字段上线前落库的 `tool_calls` 里没有这两个键。
|
||
|
||
那种行的 `tool_calls` 可能就是裸列表，甚至为 None。读取时必须按 False/None 处理，
|
||
**不得抛异常、也不得凭正文内容去猜**——猜错方向会让"不需要转人工"的答复被标成转人工。
|
||
"""
|
||
for legacy in ({"calls": []}, [], None):
|
||
message = FakeMessage(content="稳健型", tool_calls=legacy)
|
||
patch_repository(monkeypatch, (FakeRun(status="succeeded", completed_at=NOW), message))
|
||
|
||
snapshot = await RunQueryService().get("run-1", CONTEXT)
|
||
|
||
assert snapshot.result is not None, legacy
|
||
assert snapshot.result["transfer_required"] is False, legacy
|
||
assert snapshot.result["transfer_reason"] is None, legacy
|
||
|
||
|
||
def terminal_snapshot(status: str = "succeeded") -> RunSnapshot:
|
||
return RunSnapshot(
|
||
run_id="run-1",
|
||
trace_id="trace-1",
|
||
status=status,
|
||
agent_type="customer_service",
|
||
session_id="session-1",
|
||
result={"content": "ok"},
|
||
error_code=None,
|
||
created_at="2026-09-10T12:00:00Z",
|
||
completed_at="2026-09-10T12:00:01Z",
|
||
)
|
||
|
||
|
||
async def test_watch_stops_immediately_on_terminal_snapshot(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""终态快照只发一次且不再轮询——断流重连也走这条路径。"""
|
||
calls = {"count": 0}
|
||
|
||
async def never_called(self: RunQueryService, run_id: str, context: RequestContext) -> Any:
|
||
calls["count"] += 1
|
||
raise AssertionError("终态不应继续轮询数据库")
|
||
|
||
monkeypatch.setattr(RunQueryService, "get", never_called)
|
||
initial = terminal_snapshot()
|
||
|
||
events = [snapshot async for snapshot in RunQueryService().watch(initial, CONTEXT)]
|
||
|
||
assert events == [initial]
|
||
assert calls["count"] == 0
|
||
|
||
|
||
async def test_watch_emits_heartbeat_between_polls(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setattr(
|
||
"app.service.run_query_service.get_settings",
|
||
lambda: SimpleNamespace(sse_max_connection_seconds=30, sse_heartbeat_seconds=0),
|
||
)
|
||
polls = {"count": 0}
|
||
|
||
async def fake_get(self: RunQueryService, run_id: str, context: RequestContext) -> RunSnapshot:
|
||
del self, run_id, context
|
||
polls["count"] += 1
|
||
return replace(terminal_snapshot(), status="succeeded")
|
||
|
||
monkeypatch.setattr(RunQueryService, "get", fake_get)
|
||
initial = replace(terminal_snapshot(), status="running", result=None, completed_at=None)
|
||
|
||
events = [snapshot async for snapshot in RunQueryService().watch(initial, CONTEXT)]
|
||
|
||
# 序列：非终态快照 → 心跳(None) → 轮询得到的终态快照
|
||
assert [event is None for event in events] == [False, True, False]
|
||
assert events[0] is initial
|
||
assert events[2] is not None and events[2].status == "succeeded"
|
||
assert polls["count"] == 1
|