test(qa): AI 对话(聊天记录)E2E + #1/#6 真源围栏守卫(AC-4)

计划 §6/§8。收口 AI-chat-history 功能(AC-1..AC-4 全交付)。

- tests/test_ai_chat_history_e2e.py:真 pg、零 LLM/无网关(纯 CRUD 侧记录),
  覆盖 §6 全 6 用例——缓冲多轮线程一批落库 + GET newest-first/meta 保真、
  作用域分区(本章 ∪ 项目级 NULL)、跨批分页 + kind 过滤、
  append 只写 ai_messages(其它业务表零变 + usage_ledger 恒 0)/GET 只读、
  404 + 五类 422 失败零落库、clarify-abandon(服务端无部分写路径)。
- tests/test_ai_messages_not_in_generation_path.py:AST 静态围栏——
  memory/orchestrator 两目录无一 import/引用 ai_messages,
  MemoryRepos 捆绑与 assemble 形参都不含它;植入违规自证探测器非空,能变红。
- memory/gotchas.md:登记 ai_messages 真源围栏(append-only 侧记录,
  绝不喂 assemble/prompt;接入生成链 = code-review BLOCKER)。

门禁全绿:ruff/format · mypy 238 · alembic 无漂移 · pytest 1002 passed(+10)。
This commit is contained in:
Yaojia Wang
2026-07-09 17:42:45 +02:00
parent 5ba3eabda3
commit f6e88d7438
4 changed files with 679 additions and 2 deletions

View File

@@ -0,0 +1,144 @@
"""AC-4 / 计划 §8(#1/#6) 真源围栏守卫:`ai_messages` 绝不进生成链。
`ai_messages` 是 append-only 旁路侧记录(与 `usage_ledger` 同物种——记录「作者↔AI 说了
什么」,**不是手稿真源**。守不变量 #1DB 单一真源指手稿/审稿真源)/ #6记忆注入是确定性
选择,非把聊天记录喂进 prompt`assemble()` 与任何 memory/orchestrator 节点**永不读它**。
这是一道**能变红的**真守卫(非空断言):
- 若日后有人把 `AiMessageRepo`/`AiMessage` 接入 `packages/core/ww_core/{memory,orchestrator}`
下任一模块import 或引用),`test_generation_path_never_references_ai_messages` 直接失败。
- 若有人把 ai_message repo 塞进 `assemble()` 消费的 `MemoryRepos` 捆绑,
`test_assemble_repo_bundle_excludes_ai_messages` 直接失败。
- `test_detector_flags_a_planted_violation` 用植入的违规源证明探测器确会命中——守卫非空。
纯静态分析AST + 反射),无 DB / 无网络 / 无 LLM。
"""
from __future__ import annotations
import ast
import dataclasses
import inspect
from pathlib import Path
import ww_core.memory.assemble as assemble_mod
from ww_core.domain.repositories import MemoryRepos
from ww_core.memory import assemble
# 生成链禁止出现的标识符repo/ORM 模型/视图/行)。
_FORBIDDEN_NAMES = frozenset(
{
"AiMessage",
"AiMessageRepo",
"SqlAiMessageRepo",
"AiMessageView",
"AiTurnRow",
}
)
# 禁止 import 的模块子串catch `from ww_core.domain.ai_message_repo import ...`)。
_FORBIDDEN_MODULE_SUBSTR = "ai_message"
# 受守卫的生成链目录(记忆装配 + 编排器;相对仓库根)。
_GUARDED_DIRS = (
"packages/core/ww_core/memory",
"packages/core/ww_core/orchestrator",
)
_REPO_ROOT = Path(__file__).resolve().parents[1]
def _ai_message_refs(source: str, filename: str) -> list[str]:
"""返回 `source` 里对 ai_messages 生成链禁忌标识符的所有引用AST 精确匹配)。
命中面(真实的「接入生成链」向量):
- `import ...ai_message...` / `from ...ai_message... import ...`(模块子串);
- `from X import AiMessageRepo`(导入名在禁忌集);
- 代码里任意 `AiMessageRepo` 名引用 / `x.AiMessageView` 属性访问Name/Attribute
docstring/注释是 `ast.Constant`,不计入——避免「文档提到它」的假阳性。
"""
hits: list[str] = []
tree = ast.parse(source, filename=filename)
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
if node.module and _FORBIDDEN_MODULE_SUBSTR in node.module.lower():
hits.append(f"from {node.module} import ...")
for alias in node.names:
if alias.name in _FORBIDDEN_NAMES:
hits.append(f"from {node.module} import {alias.name}")
elif isinstance(node, ast.Import):
for alias in node.names:
if _FORBIDDEN_MODULE_SUBSTR in alias.name.lower():
hits.append(f"import {alias.name}")
elif isinstance(node, ast.Name) and node.id in _FORBIDDEN_NAMES:
hits.append(f"name {node.id}")
elif isinstance(node, ast.Attribute) and node.attr in _FORBIDDEN_NAMES:
hits.append(f"attr .{node.attr}")
return hits
def _guarded_py_files() -> list[Path]:
files: list[Path] = []
for rel in _GUARDED_DIRS:
d = _REPO_ROOT / rel
assert d.is_dir(), f"guarded dir missing (moved?): {d}"
files.extend(sorted(d.rglob("*.py")))
return files
def test_generation_path_never_references_ai_messages() -> None:
"""memory/orchestrator 下任一模块都不得 import/引用 ai_messages否则真源蔓延"""
files = _guarded_py_files()
# 非空守卫:确有文件被扫(否则断言形同虚设)。
assert len(files) >= 20, f"expected the generation path to have modules; scanned {len(files)}"
violations: dict[str, list[str]] = {}
for path in files:
refs = _ai_message_refs(path.read_text(encoding="utf-8"), str(path))
if refs:
violations[str(path.relative_to(_REPO_ROOT))] = refs
assert not violations, (
"ai_messages leaked into the generation path (violates invariant #1/#6 — it is an "
f"append-only side-record, never a truth-source): {violations}"
)
def test_assemble_repo_bundle_excludes_ai_messages() -> None:
"""`assemble()` 消费的 `MemoryRepos` 捆绑不得含任何 ai_message repo 字段。"""
for field in dataclasses.fields(MemoryRepos):
assert "ai_message" not in field.name.lower(), (
f"MemoryRepos.{field.name} wires ai_messages into assemble() — forbidden (#1/#6)"
)
annotation = str(field.type)
assert "AiMessage" not in annotation, (
f"MemoryRepos.{field.name} type {annotation!r} references AiMessage — forbidden (#1/#6)"
)
def test_assemble_signature_only_takes_memory_repos_bundle() -> None:
"""`assemble` 的 repo 入参恒为 `MemoryRepos`——阻止旁路加一个 ai_message repo 形参。"""
# eval_str 求值 PEP 563 字符串化注解assemble.py 有 `from __future__ import annotations`)。
sig = inspect.signature(assemble, eval_str=True)
repos_param = sig.parameters["repos"]
assert repos_param.annotation is MemoryRepos
# assemble 模块本身也不得引用禁忌标识符(双保险,直接扫其源)。
source = inspect.getsource(assemble_mod)
assert _ai_message_refs(source, "assemble.py") == []
def test_detector_flags_a_planted_violation() -> None:
"""证明守卫非空:植入的违规源确被探测器命中(否则上面的绿全是假绿)。"""
planted_import = "from ww_core.domain.ai_message_repo import AiMessageRepo\n"
assert _ai_message_refs(planted_import, "<planted>")
planted_usage = (
"from ww_core.domain import AiMessageRepo\n"
"async def assemble(repos, ai_log: AiMessageRepo):\n"
" rows = await ai_log.list_for_chapter(...)\n"
" return rows\n"
)
assert _ai_message_refs(planted_usage, "<planted>")
# 干净源不命中(避免误报——探测器有区分度)。
clean = "from ww_core.domain.repositories import MemoryRepos\nx = 1\n"
assert _ai_message_refs(clean, "<clean>") == []