games-development-ai/contracts/play-loop/test_reference_asset_and_acceptance_v3.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""W-GOLD-LIVE 检查点 1 离线验证：参照资产契约 + acceptance v3 开口向后兼容。

零依赖确定性测试，不烧真模型、不触 live 注入：复用 validate.py 内置的 stdlib 子集校验器
（与 `python3 validate.py --suite` 同一机制），验证四件事：
  ① 四个新 schema（acceptance-request-v3 / acceptance-provenance-v3 /
     reference-asset-record / reference-asset-registry）自洽：正样本过、负样本拦；
  ② 迁移清单初始数据 reference-asset-registry.initial.json 符合 registry schema
     （结构层 + recordId 唯一/migration-list 禁 active 语义层）；
  ③ acceptance-request/3 与 acceptance-provenance/3 带/不带新字段都 valid（向后兼容：
     新字段全部 optional，v2 字段集原样可过 v3）；
  ④ 旧 v2 acceptance 样本在 v2 schema 下仍 valid、负样本仍被拦（v2 契约未被破坏）。
另加金标 SoT §7 对账断言：迁移清单编码的条目数/角色/生命周期与 SoT 表一致，占位 hash
全部在 migrationNotes 标注，正式 game_content_gold 当前 0 款。

运行：python3 contracts/play-loop/test_reference_asset_and_acceptance_v3.py
全绿 exit 0；任一失败 exit 1 并打印失败证据。
"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import validate as V  # noqa: E402  —— 同目录 stdlib 校验器，唯一依赖

_PASS: list = []
_FAIL: list = []


def check(name: str, ok: bool, detail: str = "") -> None:
    """记录并打印单条断言；失败时附原始证据，便于复核而非只看红绿。"""
    (_PASS if ok else _FAIL).append((name, detail))
    line = ("PASS  " if ok else "FAIL  ") + name
    if not ok and detail:
        line += f"\n      证据：{detail}"
    print(line)


def errors_of(schema_file: str, instance) -> list:
    """结构层 + 语义层完整校验，返回错误列表（空=合规）。与 _validate_file 同口径。"""
    schema = V._load((HERE / schema_file).resolve())
    errs = V.validate(schema, instance, schema)
    if not errs:
        errs += V._semantic_validate(schema, instance)
    return errs


def sample(rel_path: str):
    """按套件口径读取样本（支持 _base/_set/_delete 声明式派生）。"""
    return V._load_sample_instance((HERE / rel_path).resolve())


def main() -> int:
    # ── ① 新 schema 自洽：正样本过、负样本拦 ──────────────────────────────
    for schema_file, contract in [
        ("acceptance-request-v3.schema.json", "acceptance-request-v3"),
        ("acceptance-provenance-v3.schema.json", "acceptance-provenance-v3"),
        ("reference-asset-record.schema.json", "reference-asset-record"),
    ]:
        valid_dir = HERE / "samples" / contract / "valid"
        invalid_dir = HERE / "samples" / contract / "invalid"
        for f in sorted(valid_dir.glob("*.json")):
            errs = errors_of(schema_file, sample(f"samples/{contract}/valid/{f.name}"))
            check(f"① {contract} 正样本通过：{f.name}", not errs, "; ".join(errs))
        for f in sorted(invalid_dir.glob("*.json")):
            errs = errors_of(schema_file, sample(f"samples/{contract}/invalid/{f.name}"))
            check(f"① {contract} 负样本被拦：{f.name}", bool(errs), "负样本竟然通过")

    # registry 正样本 = 迁移清单初始数据（canonical，套件同口径）
    registry = json.loads((HERE / "reference-asset-registry.initial.json").read_text(encoding="utf-8"))
    registry_errs = errors_of("reference-asset-registry.schema.json", registry)
    check("② 迁移清单初始数据符合 registry schema（结构+语义）", not registry_errs,
          "; ".join(registry_errs))
    for f in sorted((HERE / "samples/reference-asset-registry/invalid").glob("*.json")):
        errs = errors_of("reference-asset-registry.schema.json",
                         sample(f"samples/reference-asset-registry/invalid/{f.name}"))
        check(f"② registry 负样本被拦：{f.name}", bool(errs), "负样本竟然通过")

    # ── ③ v3 向后兼容：同身份数据不带新字段也过 v3；v2 原样数据不过 v3（版本线独立）──
    v2_base = json.loads(
        (HERE / "samples/acceptance-request-v2/valid/01-absent-puzzle.json").read_text(encoding="utf-8"))
    as_v3 = {**v2_base, "schemaVersion": "acceptance-request/3"}
    errs = errors_of("acceptance-request-v3.schema.json", as_v3)
    check("③ v2 身份数据仅改 schemaVersion=3 即过 v3（新字段全 optional）", not errs,
          "; ".join(errs))
    errs = errors_of("acceptance-request-v3.schema.json", dict(v2_base))
    check("③ v2 原样数据（schemaVersion/2）被 v3 const 拒绝（版本线独立、非就地放松）",
          bool(errs), "v2 原样数据竟通过 v3")

    v2_prov = json.loads(
        (HERE / "samples/acceptance-provenance-v2/valid/01-absent-puzzle.json").read_text(encoding="utf-8"))
    errs = errors_of("acceptance-provenance-v3.schema.json",
                     {**v2_prov, "schemaVersion": "acceptance-provenance/3"})
    check("③ v2 provenance 身份数据仅改 schemaVersion=3 即过 v3", not errs, "; ".join(errs))

    # ── ④ 旧 v2 契约不被破坏：v2 样本在 v2 schema 下仍过/仍拦 ──────────────
    for f in sorted((HERE / "samples/acceptance-request-v2/valid").glob("*.json")):
        errs = errors_of("acceptance-request-v2.schema.json",
                         sample(f"samples/acceptance-request-v2/valid/{f.name}"))
        check(f"④ 旧 v2 request 正样本仍 valid：{f.name}", not errs, "; ".join(errs))
    for f in sorted((HERE / "samples/acceptance-request-v2/invalid").glob("*.json")):
        errs = errors_of("acceptance-request-v2.schema.json",
                         sample(f"samples/acceptance-request-v2/invalid/{f.name}"))
        check(f"④ 旧 v2 request 负样本仍被拦：{f.name}", bool(errs), "负样本竟然通过")
    for f in sorted((HERE / "samples/acceptance-provenance-v2/valid").glob("*.json")):
        errs = errors_of("acceptance-provenance-v2.schema.json",
                         sample(f"samples/acceptance-provenance-v2/valid/{f.name}"))
        check(f"④ 旧 v2 provenance 正样本仍 valid：{f.name}", not errs, "; ".join(errs))

    # ── ⑤ 迁移清单编码对账金标 SoT §7 ────────────────────────────────────
    records = registry["records"]
    by_id = {r["recordId"]: r for r in records}
    check("⑤ 迁移清单共 15 条且 recordId 唯一", len(records) == 15 and len(by_id) == 15,
          f"records={len(records)} unique={len(by_id)}")

    gold_m3 = [f"gold-m3-{kind}-r3" for kind in ("gem", "candy", "fruit", "porcelain", "rune")]
    ok = all(by_id[rid]["role"] == "harness_fixture"
             and by_id[rid]["lifecycleStatus"] == "migration_pending" for rid in gold_m3)
    check("⑤ 五款 gold-m3-*-r3 → harness_fixture/migration_pending", ok)

    exemplars = [f"_fewshot-{k}" for k in ("feiyi", "puzzle")] + \
                [f"_template-{k}" for k in ("feiyi", "puzzle", "shop", "story", "trpg")]
    ok = all(by_id[rid]["role"] == "generation_exemplar"
             and by_id[rid]["lifecycleStatus"] == "migration_pending" for rid in exemplars)
    check("⑤ 2 款 _fewshot + 5 款 _template → generation_exemplar/migration_pending", ok)

    fable = ["gac-shanhai-xingji", "gac-shanhai-xunyi-lu", "gac-yeshi-yitiaojie"]
    ok = all(by_id[rid]["role"] == "game_content_gold"
             and by_id[rid]["lifecycleStatus"] == "candidate" for rid in fable)
    check("⑤ Fable 三款 → game_content_gold/candidate", ok)

    xingji_refs = by_id["gac-shanhai-xingji"]["designRef"] or []
    ok = len(xingji_refs) == 2 and all((HERE.parent.parent / ref).is_file() for ref in xingji_refs)
    check("⑤ 《山海行纪》designRef 指向 2026-07-23 追认的两份 designIntent 且文件真实存在", ok,
          f"designRef={xingji_refs}")

    active_gold = [r for r in records
                   if r["role"] == "game_content_gold" and r["lifecycleStatus"] == "active"]
    check("⑤ 正式签认 game_content_gold 当前 0 款（SoT §7 表）", not active_gold,
          f"active={active_gold}")

    # ── ⑥ 占位 hash 红线：格式合法 + 逐条注释标注，不伪装 active ────────────
    hash_pattern = re.compile(r"^[0-9a-f]{64}$")
    notes = registry.get("migrationNotes") or {}
    ok = all(hash_pattern.match(r["artifactHash"]) for r in records)
    check("⑥ 全部 artifactHash 符合 64 位小写十六进制格式", ok)
    missing_note = [r["recordId"] for r in records
                    if "占位" not in notes.get(r["recordId"], "")]
    check("⑥ 每条记录的 artifactHash 占位均在 migrationNotes 显式标注", not missing_note,
          f"缺标注={missing_note}")
    ok = all(r["lifecycleStatus"] in ("candidate", "migration_pending") for r in records)
    check("⑥ 迁移清单无任何 active 记录（不伪装 active）", ok)

    # ── 汇总 ─────────────────────────────────────────────────────────────
    print(f"\n══ 汇总：{len(_PASS)} 过 / {len(_FAIL)} 败 ══")
    if _FAIL:
        for name, detail in _FAIL:
            print(f"  ✗ {name}")
        return 1
    print("全部通过：新 schema 自洽 + 迁移清单符合 schema + v3 向后兼容 + 旧 v2 未破坏。")
    return 0


if __name__ == "__main__":
    sys.exit(main())