180 lines
7.4 KiB
Python
180 lines
7.4 KiB
Python
# ============================================================
|
||
# 黄金测试：会话引用（plan.md §4.7 / 矩阵 63 行）
|
||
# 解析稳定性 / 按时点快照取值 / 来源删除显式失败 / 进入证据链
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime
|
||
|
||
import pytest
|
||
|
||
from server.agent_core.audit import write_audit
|
||
from server.agent_core.session_ref import (
|
||
RefResolutionError,
|
||
parse_refs,
|
||
resolve_all,
|
||
resolve_ref,
|
||
)
|
||
|
||
FIXED_NOW = datetime(2026, 8, 2, 10, 30, tzinfo=UTC)
|
||
|
||
|
||
def _store_data() -> dict:
|
||
"""工作区快照（与 ProjectStore.snapshot 结构一致 + 方案/版本/报告表）。"""
|
||
return {
|
||
"projects": [
|
||
{"id": "p1", "name": "青岛Q3排产", "scopeLabel": "工厂/车间", "archived": False},
|
||
],
|
||
"sessions": [
|
||
{"id": "sess_a", "projectId": "p1", "title": "比亚迪插单评估", "status": "done"},
|
||
{"id": "sess_personal", "projectId": "__personal__", "title": "草稿", "status": "running"},
|
||
],
|
||
"messages": {
|
||
"sess_a": [
|
||
{"role": "user", "text": "评估比亚迪插单"},
|
||
{"role": "agent", "text": "结论：可插单，交期优先"},
|
||
],
|
||
},
|
||
"scenarios": [
|
||
{"scenarioId": "scen_1", "strategy": "交期优先", "kpi": {"avgUtilization": 0.92}},
|
||
],
|
||
"scheduleVersions": [
|
||
{"id": "V20260716-003", "name": "V20260716-003", "summary": "基准版本", "planCount": 12},
|
||
],
|
||
"reports": [
|
||
{"reportId": "rep_1", "title": "上周复盘", "markdown": "插单流程需优化"},
|
||
],
|
||
}
|
||
|
||
|
||
def test_parse_refs_stable_kinds_and_dedup():
|
||
text = (
|
||
"按 @会话:比亚迪插单评估 的结论重排本周计划，"
|
||
"参考 @方案:交期优先-v2 与 @版本:V20260716-003、@报告:上周复盘；"
|
||
"再提一次 @会话:比亚迪插单评估。"
|
||
)
|
||
refs = parse_refs(text)
|
||
assert [(r.kind, r.target_id) for r in refs] == [
|
||
("session", "比亚迪插单评估"),
|
||
("scenario", "交期优先-v2"),
|
||
("version", "V20260716-003"),
|
||
("report", "上周复盘"),
|
||
]
|
||
# 稳定：相同输入 -> 相同输出（顺序、去重、ref_id 均确定）
|
||
again = parse_refs(text)
|
||
assert [(r.ref_id, r.raw) for r in refs] == [(r.ref_id, r.raw) for r in again]
|
||
assert refs[0].ref_id == "session:比亚迪插单评估"
|
||
# 无引用/非引用 @ 不产生误解析
|
||
assert parse_refs("直接重排，没有引用") == []
|
||
assert parse_refs("联系邮箱 a@b.com 和 @xxx:空") == []
|
||
|
||
|
||
def test_resolve_ref_takes_time_point_snapshot():
|
||
sd = _store_data()
|
||
resolved = resolve_ref(sd, "session", "sess_a", now=FIXED_NOW)
|
||
assert resolved.kind == "session"
|
||
assert resolved.snapshot_at == "2026-08-02 10:30"
|
||
assert resolved.value["session"]["status"] == "done"
|
||
assert len(resolved.value["messages"]) == 2
|
||
# 被引会话后续变化不影响已取快照（按时点取值硬规则）
|
||
sd["sessions"][0]["status"] = "archived"
|
||
sd["messages"]["sess_a"].append({"role": "agent", "text": "后续新结论"})
|
||
assert resolved.value["session"]["status"] == "done"
|
||
assert len(resolved.value["messages"]) == 2
|
||
|
||
# 版本引用同样按时点快照
|
||
version = resolve_ref(sd, "version", "V20260716-003", now=FIXED_NOW)
|
||
assert version.value["planCount"] == 12
|
||
sd["scheduleVersions"][0]["planCount"] = 999
|
||
assert version.value["planCount"] == 12
|
||
|
||
|
||
def test_resolve_kind_value_shapes():
|
||
sd = _store_data()
|
||
scenario = resolve_ref(sd, "scenario", "scen_1", now=FIXED_NOW)
|
||
assert scenario.value["strategy"] == "交期优先"
|
||
assert scenario.value["kpi"]["avgUtilization"] == 0.92
|
||
report = resolve_ref(sd, "report", "rep_1", now=FIXED_NOW)
|
||
assert report.value["title"] == "上周复盘"
|
||
# 来源信息随解析结果携带（出处可查）
|
||
assert resolve_ref(sd, "session", "sess_a", now=FIXED_NOW).source["messageCount"] == 2
|
||
|
||
|
||
def test_resolve_ref_explicit_failure_when_source_deleted_or_inaccessible():
|
||
sd = _store_data()
|
||
# 来源删除：目标不在可见会话列表
|
||
with pytest.raises(RefResolutionError) as exc:
|
||
resolve_ref(sd, "session", "sess_deleted", now=FIXED_NOW)
|
||
assert exc.value.kind == "session"
|
||
assert "deleted" in exc.value.reason
|
||
|
||
# 无权访问：会话归属项目已归档
|
||
sd["projects"][0]["archived"] = True
|
||
with pytest.raises(RefResolutionError) as exc:
|
||
resolve_ref(sd, "session", "sess_a", now=FIXED_NOW)
|
||
assert "no-access" in exc.value.reason
|
||
sd["projects"][0]["archived"] = False
|
||
|
||
# 各 kind 目标缺失均显式失败，不泛化掩盖
|
||
for kind, target in (("scenario", "scen_missing"), ("version", "V0000"), ("report", "rep_missing")):
|
||
with pytest.raises(RefResolutionError) as exc:
|
||
resolve_ref(sd, kind, target, now=FIXED_NOW)
|
||
assert exc.value.kind == kind
|
||
assert exc.value.target_id == target
|
||
|
||
# @知识 已是合法 kind（矩阵 63 行）：未注入知识查找回调必须显式报错（不静默、不编造）
|
||
with pytest.raises(RefResolutionError) as exc:
|
||
resolve_ref(sd, "knowledge", "sop-1", now=FIXED_NOW)
|
||
assert exc.value.kind == "knowledge"
|
||
assert "not-wired" in exc.value.reason
|
||
# 真正的未知 kind 仍是参数级拒绝
|
||
with pytest.raises(ValueError):
|
||
resolve_ref(sd, "bogus", "x", now=FIXED_NOW)
|
||
|
||
|
||
def test_resolved_refs_enter_evidence_chain_via_write_audit():
|
||
sd = _store_data()
|
||
refs = parse_refs("按 @会话:sess_a 结论重排，以 @版本:V20260716-003 为基准")
|
||
resolved, evidence_refs = resolve_all(sd, refs, now=FIXED_NOW)
|
||
assert len(resolved) == 2
|
||
assert evidence_refs == ["session:sess_a", "version:V20260716-003"]
|
||
|
||
# 证据链：evidenceRefs 进入审计事件（write_audit 原样落库）
|
||
world: dict = {"auditEvents": []}
|
||
counter = iter(range(1, 100))
|
||
|
||
def next_id(_kind: str) -> int:
|
||
return next(counter)
|
||
|
||
event = write_audit(
|
||
world,
|
||
next_id,
|
||
actor="planner",
|
||
category="SESSION",
|
||
action="session.ref.resolve",
|
||
target={"type": "session", "id": "sess_a"},
|
||
power="P0",
|
||
rationale={"intent": "引用方案重排", "refCount": len(evidence_refs)},
|
||
evidence_refs=evidence_refs,
|
||
)
|
||
assert event["evidenceRefs"] == ["session:sess_a", "version:V20260716-003"]
|
||
assert world["auditEvents"][-1]["hash"] == event["hash"]
|
||
# 引用可追溯：kind:target 可直接回查解析结果
|
||
assert [r.ref for r in resolved] == event["evidenceRefs"]
|
||
|
||
|
||
def test_four_kinds_remain_compatible_with_knowledge_lookup_param():
|
||
"""新增 knowledge_lookup 参数不影响 4 类既有引用（向后兼容）。"""
|
||
sd = _store_data()
|
||
refs = parse_refs(
|
||
"按 @会话:sess_a 结论，以 @方案:scen_1、@版本:V20260716-003、@报告:rep_1 为输入")
|
||
resolved, evidence_refs = resolve_all(sd, refs, now=FIXED_NOW,
|
||
knowledge_lookup=lambda q: None)
|
||
assert [r.kind for r in resolved] == ["session", "scenario", "version", "report"]
|
||
assert evidence_refs == [
|
||
"session:sess_a", "scenario:scen_1", "version:V20260716-003", "report:rep_1",
|
||
]
|
||
# 不含 @知识 时，不传 knowledge_lookup 与传入等价（无副作用）
|
||
resolved2, evidence2 = resolve_all(sd, refs, now=FIXED_NOW)
|
||
assert evidence2 == evidence_refs
|