games-development-ai/cheap-worker/tests/test_kb_externalize.py

"""test_kb_externalize.py —— 便宜档知识件外置(W-CFG-KB 切片一)单测。

覆盖交付要求的三条 + 关键安全语义:
  ① 知识包 save → 物化 → read_file 读到物化内容(用本地 FS store,不依赖真 MinIO);
  ② 缺文件显式失败(manifest 列了但缺 → 物化抛 KnowledgeMaterializeError,不静默物化半拉子;
     且 kb_root 保留上一份已物化包 = 回落,agent 仍读完整旧包);
  ③ TIER2_KB_ROOT 未配时 read_file 行为与现状一致(退回只读仓根);
另加:in-code DDL↔.sql 规范化对账(机器门)、幂等、逐文件指纹校验、路径越界、200KB 截断、物化 no-op。

跑:cd cheap-worker && .venv/bin/python -m pytest tests/test_kb_externalize.py -q
"""

import sys
from pathlib import Path

# 跨包 import:cheap-worker/(cheap_run)+ tier2/gen-worker/(worker.kb_store)都进 sys.path。
_CW = Path(__file__).resolve().parents[1]            # cheap-worker/
_REPO = _CW.parent                                    # 仓库根
_GW = _REPO / "tier2" / "gen-worker"                  # tier2/gen-worker/(worker / service 顶层包所在)
for _p in (str(_CW), str(_GW)):
    if _p not in sys.path:
        sys.path.insert(0, _p)

import cheap_run  # noqa: E402
from worker import kb_store  # noqa: E402
from worker.kb_store import (  # noqa: E402
    KnowledgeMaterializeError,
    LocalFsKnowledgePackageStore,
    materialize_active,
    resolve_active_version,
)

_NOW = 1_730_000_000.0  # 固定时钟(可复现 versionId)


def _store(tmp_path) -> LocalFsKnowledgePackageStore:
    return LocalFsKnowledgePackageStore(tmp_path / "_kb-store")


# ─────────────────────────── ① save → 物化 → read_file 读到物化内容 ───────────────────────────

def test_save_materialize_read(tmp_path, monkeypatch):
    """存一版知识包 → 物化到知识根 → read_file 按原相对路径读到物化内容;未外置的路径回落仓根。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"
    files = [
        {"path": ".agents/skills/fake-skill.md", "content": "KB-EXTERNALIZED-V1\n设计范式改这里。"},
        {"path": "game-runtime/games/_fewshot-x/src/game-logic.js", "content": "// 外置起点范例 V1"},
    ]
    saved = store.save(files, now_ts=_NOW, package_id="cheap-kb", note="切片一首版")
    assert saved["versionId"].startswith("v")

    res = materialize_active(store, kb_root, package_id="cheap-kb", active_version=saved["versionId"])
    assert res["materialized"] is True
    assert res["files"] == 2

    # 开启知识根 shadow,read_file 读到物化内容。
    monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root))
    r1 = cheap_run.read_file(".agents/skills/fake-skill.md")
    assert r1["ok"] is True, r1
    assert r1["content"] == "KB-EXTERNALIZED-V1\n设计范式改这里。"
    r2 = cheap_run.read_file("game-runtime/games/_fewshot-x/src/game-logic.js")
    assert r2["ok"] is True and r2["content"] == "// 外置起点范例 V1"

    # 未外置进知识包的路径(如仓内 AGENTS.md)→ 知识根未命中 → 回落仓根,读到仓里的真内容。
    r3 = cheap_run.read_file("AGENTS.md")
    assert r3["ok"] is True
    assert r3["content"] == (_REPO / "AGENTS.md").read_text(encoding="utf-8")


def test_materialize_resolves_active_via_env(tmp_path, monkeypatch):
    """当前激活版解析:env TIER2_KB_ACTIVE_VERSION 钉版优先,否则最新 committed。"""
    store = _store(tmp_path)
    v1 = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb")
    v2 = store.save([{"path": "a.md", "content": "A2"}], now_ts=_NOW + 10, package_id="cheap-kb")
    # 无 env → 最新 committed = v2。
    monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False)
    assert resolve_active_version(store, package_id="cheap-kb") == v2["versionId"]
    # 钉版 env → v1。
    monkeypatch.setenv("TIER2_KB_ACTIVE_VERSION", v1["versionId"])
    assert resolve_active_version(store, package_id="cheap-kb") == v1["versionId"]


# ─────────────────────────── ② 缺文件显式失败 + 回落上一份 ───────────────────────────

def test_missing_file_explicit_failure_and_fallback(tmp_path, monkeypatch):
    """先物化好的 v1;v2 缺文件(manifest 列了但 MinIO/FS 取不到)→ 物化显式失败,kb_root 保留 v1。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"

    # v1:完整,物化成功。
    v1 = store.save([{"path": "skill.md", "content": "GOOD-V1"}], now_ts=_NOW, package_id="cheap-kb")
    materialize_active(store, kb_root, package_id="cheap-kb", active_version=v1["versionId"])
    monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root))
    assert cheap_run.read_file("skill.md")["content"] == "GOOD-V1"

    # v2:两文件,落库后把其中一个文件从 store 的 files/ 删掉,模拟「manifest 列了但取不到」。
    v2 = store.save(
        [{"path": "skill.md", "content": "V2"}, {"path": "extra.md", "content": "E2"}],
        now_ts=_NOW + 10, package_id="cheap-kb")
    victim = tmp_path / "_kb-store" / "cheap-kb" / v2["versionId"] / "files" / "extra.md"
    assert victim.exists()
    victim.unlink()  # 缺文件

    # fetch 应把缺文件标 missingContent(不静默)。
    fetched = store.fetch(package_id="cheap-kb", version_id=v2["versionId"])
    missing = [f for f in fetched["files"] if "missingContent" in f]
    assert len(missing) == 1 and missing[0]["path"] == "extra.md"

    # 物化 v2 → 显式失败(拒绝物化半拉子)。
    try:
        materialize_active(store, kb_root, package_id="cheap-kb", active_version=v2["versionId"])
        raise AssertionError("缺文件物化本应抛 KnowledgeMaterializeError,却成功了")
    except KnowledgeMaterializeError as e:
        assert "缺文件" in str(e)

    # 回落:kb_root 未被触碰,仍是 v1,read_file 读到完整旧包。
    assert cheap_run.read_file("skill.md")["content"] == "GOOD-V1"


def test_materialize_hash_mismatch_rejected(tmp_path):
    """逐文件指纹校验(K2 寻址断言):物化内容 hash ≠ manifest → 显式失败,kb_root 不动。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"
    v1 = store.save([{"path": "skill.md", "content": "ORIGINAL"}], now_ts=_NOW, package_id="cheap-kb")
    # 篡改 store files/ 里的文件内容,但不更新 manifest → fetch 返篡改内容、manifest sha256 是旧的。
    tampered = tmp_path / "_kb-store" / "cheap-kb" / v1["versionId"] / "files" / "skill.md"
    tampered.write_text("TAMPERED", encoding="utf-8")
    try:
        materialize_active(store, kb_root, package_id="cheap-kb", active_version=v1["versionId"])
        raise AssertionError("指纹不匹配本应抛 KnowledgeMaterializeError")
    except KnowledgeMaterializeError as e:
        assert "指纹不匹配" in str(e)
    assert not kb_root.exists()  # 从未物化成功 → kb_root 根本没建


def test_missing_active_version_raises(tmp_path):
    """当前激活版整个取不到(fetch 返 None)→ 显式失败,不静默清空 kb_root。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"
    try:
        materialize_active(store, kb_root, package_id="cheap-kb", active_version="v0-nonexistent")
        raise AssertionError("取不到激活版本本应抛 KnowledgeMaterializeError")
    except KnowledgeMaterializeError as e:
        assert "取不到" in str(e)


# ─────────────────────────── ③ TIER2_KB_ROOT 未配 → read_file 行为与现状一致 ───────────────────────────

def test_read_file_unset_falls_back_to_repo(monkeypatch):
    """TIER2_KB_ROOT 未配置时,read_file 退回只读仓根,行为与外置前一致。"""
    monkeypatch.delenv("TIER2_KB_ROOT", raising=False)
    # 读一个真存在于仓根的知识件,内容 == 直接读盘。
    rel = ".agents/skills/littlejs-game-dev.md"
    r = cheap_run.read_file(rel)
    assert r["ok"] is True
    raw = (_REPO / rel).read_bytes()
    expect = raw[: cheap_run._MAX_READ_BYTES].decode("utf-8", "ignore") if len(raw) > cheap_run._MAX_READ_BYTES \
        else raw.decode("utf-8", "ignore")
    assert r["content"] == expect
    assert r.get("truncated") == (len(raw) > cheap_run._MAX_READ_BYTES)


def test_read_file_unset_missing_and_dir(monkeypatch):
    """未配知识根:缺文件 / 读目录的软失败行为不变(与现状一致)。"""
    monkeypatch.delenv("TIER2_KB_ROOT", raising=False)
    assert cheap_run.read_file("does/not/exist.md")["ok"] is False
    dir_res = cheap_run.read_file(".agents")  # 目录
    assert dir_res["ok"] is False and "是目录" in dir_res["error"]


def test_kb_root_set_but_file_absent_falls_back(tmp_path, monkeypatch):
    """知识根已配但没物化该文件 → 回落仓根读到仓内内容(shadow 只在命中时接管)。"""
    kb_root = tmp_path / "kb-active"
    kb_root.mkdir()  # 空知识根(未物化任何知识件)
    monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root))
    rel = ".agents/skills/littlejs-game-dev.md"
    r = cheap_run.read_file(rel)
    assert r["ok"] is True
    assert r["content"][:64] == (_REPO / rel).read_text(encoding="utf-8")[:64]


# ─────────────────────────── 安全边界:路径越界 / 截断 ───────────────────────────

def test_read_file_path_escape_rejected(tmp_path, monkeypatch):
    """两根都守边界:含 '..' 逃出根的路径被拒(软失败 ok=False),无论知识根配没配。"""
    # 未配知识根:走仓根边界。
    monkeypatch.delenv("TIER2_KB_ROOT", raising=False)
    assert cheap_run.read_file("../../../etc/passwd")["ok"] is False
    # 配了知识根:知识根边界同样拦。
    kb_root = tmp_path / "kb-active"
    kb_root.mkdir()
    monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root))
    esc = cheap_run.read_file("../../../etc/passwd")
    assert esc["ok"] is False and "越界" in esc["error"]


def test_read_file_kb_truncation(tmp_path, monkeypatch):
    """知识根命中的大文件同样走 200KB 截断(边界与仓根一致)。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"
    big = "A" * (cheap_run._MAX_READ_BYTES + 50_000)  # > 200KB
    v = store.save([{"path": "big.md", "content": big}], now_ts=_NOW, package_id="cheap-kb")
    materialize_active(store, kb_root, package_id="cheap-kb", active_version=v["versionId"])
    monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root))
    r = cheap_run.read_file("big.md")
    assert r["ok"] is True and r["truncated"] is True
    assert len(r["content"]) == cheap_run._MAX_READ_BYTES


# ─────────────────────────── 存储范式:幂等 / no-op ───────────────────────────

def test_save_idempotent(tmp_path):
    """整包内容未变 → 二次 save 幂等命中同一 versionId,不重复落。"""
    store = _store(tmp_path)
    files = [{"path": "a.md", "content": "SAME"}]
    v1 = store.save(files, now_ts=_NOW, package_id="cheap-kb")
    v2 = store.save(files, now_ts=_NOW + 999, package_id="cheap-kb")  # 时钟变,但内容没变
    assert v1["versionId"] == v2["versionId"]
    assert v1["contentHash"] == v2["contentHash"]


def test_materialize_noop_when_unset(tmp_path, monkeypatch):
    """未配 TIER2_KB_ROOT / 无激活版 → 物化 no-op(不抛、不建目录),知识件外置整体等价关闭。"""
    store = _store(tmp_path)
    monkeypatch.delenv("TIER2_KB_ROOT", raising=False)
    monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False)
    # kb_root 未配 → no-op。
    res_unset = materialize_active(store, None, package_id="cheap-kb")
    assert res_unset["materialized"] is False
    # kb_root 配了但包里无任何版本 → 无激活版 → no-op。
    res_no_ver = materialize_active(store, tmp_path / "kb-active", package_id="cheap-kb")
    assert res_no_ver["materialized"] is False
    assert not (tmp_path / "kb-active").exists()


# ─────────────────────────── doc↔code:in-code DDL ↔ .sql 规范化对账机器门 ───────────────────────────

_SQL_PATH = _REPO / "tier2" / "config" / "schema" / "cheap_knowledge_package_version.sql"


def _norm_sql(text: str) -> str:
    """规范化 SQL:去整行 -- 注释、中性化分号、压空白(与 test_store_ddl_reconcile 同口径)。"""
    kept = [ln for ln in text.splitlines() if not ln.strip().startswith("--")]
    joined = " ".join(kept).replace(";", " ")
    return " ".join(joined.split()).strip()


def test_incode_ddl_matches_sql_file():
    """in-code DDL(kb_store._DDL_KB_PACKAGE_VERSION)与 .sql 留档规范化后逐 token 相等(双源同步机器门)。"""
    assert _SQL_PATH.exists(), f"缺 .sql 权威声明: {_SQL_PATH}"
    sql_norm = _norm_sql(_SQL_PATH.read_text(encoding="utf-8"))
    incode_norm = _norm_sql(kb_store._DDL_KB_PACKAGE_VERSION)
    assert sql_norm == incode_norm, (
        "in-code DDL 与 .sql 漂移(改了一处没同步另一处)。\n"
        f"--- in-code ---\n{incode_norm}\n--- .sql ---\n{sql_norm}"
    )


def test_kb_addressing_anchors():
    """知识包寻址三点(另立表名 / 双唯一键)立为 doc↔code 断言:改名即红。且绝不撞 ASSET-SRC 的表 / 桶。"""
    assert kb_store._KB_MYSQL_TABLE == "cheap_knowledge_package_version"
    assert kb_store._KB_BUCKET_DEFAULT == "cheap-kb"
    for anchor in ("uk_pkg_content", "uk_pkg_version"):
        assert anchor in kb_store._DDL_KB_PACKAGE_VERSION, f"in-code DDL 缺唯一键 {anchor}"
    # 与 ASSET-SRC 不合表:表名 / 桶都不同。
    from worker import store as _src_store
    assert kb_store._KB_MYSQL_TABLE != _src_store._MYSQL_TABLE
    assert kb_store._KB_BUCKET_DEFAULT != "tier2-src"


# ─────────────────────────── K3:治理激活版本源接线(resolve 读 Nacos 激活版 · env 回落 · 变更即物化)───────────────────────────

from worker import kb_nacos  # noqa: E402


class _FakeSource:
    """满足 kb_store 激活版本源 get(package_id)->version|None 契约的假源(单测注入,不连 Nacos)。"""

    def __init__(self, version, *, package_id="cheap-kb", raise_exc=None):
        self._version = version
        self._package_id = package_id
        self._raise = raise_exc

    def get(self, package_id):
        if self._raise is not None:
            raise self._raise
        if self._package_id is not None and package_id != self._package_id:
            return None
        return self._version


def test_resolve_prefers_governance_source(tmp_path, monkeypatch):
    """K3 取值链:治理激活版本源(Nacos 下发)> env 钉版 > 最新 committed;detach 后回落。"""
    store = _store(tmp_path)
    latest = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb")
    monkeypatch.setenv("TIER2_KB_ACTIVE_VERSION", "v-env-pinned")  # env 钉了别的版
    kb_store.attach_active_version_source(_FakeSource("v-gov-active"))
    try:
        # 源优先于 env 与最新 committed。
        assert resolve_active_version(store, package_id="cheap-kb") == "v-gov-active"
    finally:
        kb_store.detach_active_version_source()
    # detach 后回落 env 钉版。
    assert resolve_active_version(store, package_id="cheap-kb") == "v-env-pinned"
    monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False)
    # 无源无 env → 最新 committed(切片一行为、字节不变)。
    assert resolve_active_version(store, package_id="cheap-kb") == latest["versionId"]


def test_resolve_source_none_falls_back(tmp_path, monkeypatch):
    """源返回 None(dataId 尚无激活版)→ 回落 env,再回落最新 committed。"""
    store = _store(tmp_path)
    v = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb")
    monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False)
    kb_store.attach_active_version_source(_FakeSource(None))
    try:
        assert resolve_active_version(store, package_id="cheap-kb") == v["versionId"]
    finally:
        kb_store.detach_active_version_source()


def test_resolve_source_error_is_best_effort(tmp_path, monkeypatch):
    """源取值抛异常 → best-effort 静默回落 env/最新 committed,绝不因治理源拖垮解析(→拖垮生成)。"""
    store = _store(tmp_path)
    v = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb")
    monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False)
    kb_store.attach_active_version_source(_FakeSource(None, raise_exc=RuntimeError("nacos down")))
    try:
        assert resolve_active_version(store, package_id="cheap-kb") == v["versionId"]
    finally:
        kb_store.detach_active_version_source()


def test_resolve_source_package_mismatch_falls_back(tmp_path, monkeypatch):
    """源里 dataId 标的是别的包 → get 返 None → 回落(不冒充本包激活版)。"""
    store = _store(tmp_path)
    v = store.save([{"path": "a.md", "content": "A1"}], now_ts=_NOW, package_id="cheap-kb")
    monkeypatch.delenv("TIER2_KB_ACTIVE_VERSION", raising=False)
    kb_store.attach_active_version_source(_FakeSource("v-other", package_id="some-other-kb"))
    try:
        assert resolve_active_version(store, package_id="cheap-kb") == v["versionId"]
    finally:
        kb_store.detach_active_version_source()


class _FakeNacosClient:
    """假 v1 NacosClient:get_config 返预置内容,add_config_watcher 记回调供手动触发(不连真 Nacos)。"""

    def __init__(self, initial=None):
        self._content = initial
        self._cb = None

    def get_config(self, data_id, group):
        return self._content

    def add_config_watcher(self, data_id, group, cb):
        self._cb = cb

    def push(self, content):
        """模拟 Nacos 长轮询推送:更新内容 + 以 v1 回调 dict 形态触发回调。"""
        self._content = content
        if self._cb is not None:
            self._cb({"data_id": "gen-hot-params-knowledge", "group": "DEFAULT_GROUP", "content": content})


def _kb_dataid_content(package_id, version):
    """组一份知识包 dataId 内容(与 game-cloud codec 产的复合键载荷逐字对齐)。"""
    import json
    return json.dumps({"knowledge.packageId": package_id, "knowledge.activeVersion": version})


def test_kb_source_parse_and_get(tmp_path):
    """KbActiveVersionSource 解析 dataId 复合键 → get 返激活版;包 id 不符返 None(初读不物化)。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"
    client = _FakeNacosClient(initial=_kb_dataid_content("cheap-kb", "v0001-aaa"))
    src = kb_nacos.KbActiveVersionSource(
        client=client, data_id="gen-hot-params-knowledge", group="DEFAULT_GROUP",
        store=store, kb_root=kb_root, package_id="cheap-kb")
    src.start()  # 初读 populate 版本,不物化
    assert src.get("cheap-kb") == "v0001-aaa"
    assert src.get("some-other-kb") is None  # 包 id 不符不冒充
    assert not kb_root.exists()  # 初读不物化(启动物化由 setup 显式做)


def test_kb_source_change_triggers_materialize(tmp_path, monkeypatch):
    """激活推送(dataId 变更)→ 源 _on_change 触发重新物化 → read_file 读到新一版知识件。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"
    v1 = store.save([{"path": "skill.md", "content": "GOV-V1"}], now_ts=_NOW, package_id="cheap-kb")
    v2 = store.save([{"path": "skill.md", "content": "GOV-V2"}], now_ts=_NOW + 10, package_id="cheap-kb")

    client = _FakeNacosClient(initial=_kb_dataid_content("cheap-kb", v1["versionId"]))
    src = kb_nacos.KbActiveVersionSource(
        client=client, data_id="gen-hot-params-knowledge", group="DEFAULT_GROUP",
        store=store, kb_root=kb_root, package_id="cheap-kb")
    src.start()  # 初读 v1(不物化)
    kb_store.attach_active_version_source(src)
    monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root))
    try:
        # 启动物化(setup 会做;这里显式模拟)→ resolve 读源 v1 → 物化 v1。
        materialize_active(store, kb_root)
        assert cheap_run.read_file("skill.md")["content"] == "GOV-V1"
        # 激活推送 v2 → 源 _on_change 触发重新物化。
        client.push(_kb_dataid_content("cheap-kb", v2["versionId"]))
        assert src.get("cheap-kb") == v2["versionId"]
        assert cheap_run.read_file("skill.md")["content"] == "GOV-V2"
    finally:
        kb_store.detach_active_version_source()


def test_kb_source_materialize_best_effort_on_missing(tmp_path, monkeypatch):
    """激活推送到一个缺文件的版本 → 物化【显式失败】被 best-effort 接住(不抛)、kb_root 保留上一份完整包。"""
    store = _store(tmp_path)
    kb_root = tmp_path / "kb-active"
    v1 = store.save([{"path": "skill.md", "content": "GOOD-V1"}], now_ts=_NOW, package_id="cheap-kb")
    v2 = store.save(
        [{"path": "skill.md", "content": "V2"}, {"path": "extra.md", "content": "E2"}],
        now_ts=_NOW + 10, package_id="cheap-kb")

    client = _FakeNacosClient(initial=_kb_dataid_content("cheap-kb", v1["versionId"]))
    src = kb_nacos.KbActiveVersionSource(
        client=client, data_id="gen-hot-params-knowledge", group="DEFAULT_GROUP",
        store=store, kb_root=kb_root, package_id="cheap-kb")
    src.start()
    kb_store.attach_active_version_source(src)
    monkeypatch.setenv("TIER2_KB_ROOT", str(kb_root))
    try:
        materialize_active(store, kb_root)  # 物化 v1
        assert cheap_run.read_file("skill.md")["content"] == "GOOD-V1"
        # 破坏 v2 的一个文件(manifest 列了但取不到)。
        victim = tmp_path / "_kb-store" / "cheap-kb" / v2["versionId"] / "files" / "extra.md"
        victim.unlink()
        # 推送 v2 → _on_change 触发物化 → 显式失败被 best-effort 接住(不抛),kb_root 保留 v1。
        client.push(_kb_dataid_content("cheap-kb", v2["versionId"]))  # 绝不抛
        assert cheap_run.read_file("skill.md")["content"] == "GOOD-V1"  # 仍读 v1 完整包
    finally:
        kb_store.detach_active_version_source()