writer-work-flow/tests/test_ai_messages_not_in_generation_path.py

"""AC-4 / 计划 §8(#1/#6) 真源围栏守卫：`ai_messages` 绝不进生成链。

`ai_messages` 是 append-only 旁路侧记录（与 `usage_ledger` 同物种）——记录「作者↔AI 说了
什么」，**不是手稿真源**。守不变量 #1（DB 单一真源指手稿/审稿真源）/ #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>") == []