feat(planB-p2): U9 M0 core 联合门 turnkey（003-U3）——shell U8 + 三契约结构核

M0 core = 一句话真生成可玩游戏 + 三契约对齐： - u9_m0_core_gate.py：shell U8 e2e + 运行时读真契约做结构核（API 信封 code==0 / dispatcher=source-project.schema 4键 / manifest=game-package.schema 8键）；selftest 5/5 - 读契约 required 防硬编码漂移；缺 artifact 如实记 gap（不伪报） - README 补「gen-done 接线时须补」4 项（play 路 / 遥测查询 / U8→U9 artifacts / 抽样源） turnkey 三件齐（U7 rubric + U8 e2e + U9 M0 门），gen-done 落地即跑。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
feat(planB-p2): U8 feed真玩 e2e 驱动 turnkey（R4+R11）——编排全链 + 资产对位断言
2026-06-20 01:38:39 -07:00 · 2026-06-20 01:35:41 -07:00 · 2026-06-20 01:28:47 -07:00
5 changed files with 717 additions and 0 deletions
--- a/docs/agent-specs/2026-06-20-planB-phase2-harness/README.md
+++ b/docs/agent-specs/2026-06-20-planB-phase2-harness/README.md
@ -0,0 +1,31 @@
+# Plan B Phase-2 turnkey harness（U7/U8/U9 · gated on gen-done）
+
+> 本目录 = [Plan B](../../plans/2026-06-19-001-feat-coredone-验收闭环执行-plan.md) **Phase-2 game-quality 门**的可一键跑 harness。
+> **全部 gated on gen-done**（Plan A `feat/mac-gamedef-qc-concurrent` flip 到 ≥80% 后）——本目录是「**非生成半边**」：harness 先建好，gen-done 批一落地即跑、不再现搭。
+> 复用既有基底：`docs/agent-specs/2026-06-09-agent-loop-v1/orchestrator/`（player_cdp / studio_walk / backend_gw / judge / run_batch）+ `game-runtime/games/_wg1-gen/_shared/play.cdp.cjs`（九门四件套）。
+
+## 单元 → 文件 → 跑法
+
+| 单元 | 文件 | 状态 | gen-done 后跑法 |
+|---|---|---|---|
+| **U7 rubric 评分（R9）** | `rubric_score.py` + `rubric-protocol.md` | ✅ 建好（selftest 7/7） | `--template --batch <批manifest> -o sheet.json` → 人工填分 → `--score sheet.json` 判门 |
+| **U8 feed真玩 e2e（R4+R11）** | `u8_feed_play_e2e.py`（接 backend_gw 全链 + play.cdp.cjs/前端 player_cdp + 资产对位断言） | ✅ 建好（selftest 8/8 纯逻辑断言；编排步 gated on gen-done） | `--run --prompt "…" --cdp <端点>`：一句话→SAA 生成→发布→feed→试玩→断言（published/in-feed/资产对位/九门/终局遥测）。play 路二选一见文件注（产品前端 :4173 / 本地 bundle 九门） |
+| **U9 M0 core 联合门（003-U3）** | `u9_m0_core_gate.py`（shell U8 + 运行时读契约结构核） | ✅ 建好（selftest 5/5；读真契约 dispatcher=4键/manifest=8键） | `--run`：跑 U8 e2e + 三契约核（API 信封 code==0 / dispatcher=source-project.schema / manifest=game-package.schema） |
+
+## 门禁判据（plan 003 对齐）
+- **U7**：N=8 评分表 + 总均分 ≥ 1.4 + 资产对位项无 0（一票否决）。
+- **U8**：CDP 四件套（非空 canvas 启动帧 + 交互后截图 + DOM/console）+ `game_project` ≥ published + 遥测 `session_start`/终局事件 + 资产经 U1 比对命中。
+- **U9**：M0 core e2e 过（一句话端到端可玩）+ 三契约对上。
+
+## 为什么现在能建（不被 gen-done 卡）
+harness 的**逻辑/断言/抽样/聚合**与具体生成产物无关——只在**跑**时喂入 gen-done 批。故先建 + 自测（合成数据），gen-done 落地后零搭建直接跑，符合 plan「依赖排序=独立先、gen-依赖后」与 [efficiency §8.4 验证门前置 / §8.8 可复现编排]。
+
+## 串行约束
+U8/U9 的 mini-desktop CDP 真跑与 Plan A 的 e2e **串行排队**（同机端口 4320/9222、flock）——见 `.agents/skills/game-e2e-cdp-harness.md`。
+
+## gen-done 接线时须补（诚实标注·非现编）
+本目录 harness 的**逻辑/断言/抽样/三契约核**已建 + selftest；下列须**待 gen-done 产出形态明确后**接线（现无法验，故标 deferred 不伪报）：
+1. **U8 play 路二选一**：Plan A 产「staging 发布游戏」→ 走产品前端 :4173 GamePlayer（player_cdp，同 u5u2_walk）；产「本地 gameDefinition」→ 走 play.cdp.cjs 本地 bundle 九门。二者断言口径同。
+2. **U8 遥测查询接口**：`backend_gw` 现无 `get_telemetry`，终局断言暂记 gap——gen-done 接线时补遥测查询（或经 DB/日志取 `game_end`）。
+3. **U8→U9 artifacts 传递**：U9 三契约核需 U8 evidence 带 `api_envelope/source_project/package` 三件 artifact；现 U8 未输出（形态待 gen-done 实物定），U9 已对此记 gap。接线时在 `run_e2e` 补 `ev["artifacts"]`。
+4. **抽样源**：U7 `--batch` 须喂 Plan A ≥80% 批 manifest（形态：list[gameId|对象]）。
--- a/docs/agent-specs/2026-06-20-planB-phase2-harness/rubric-protocol.md
+++ b/docs/agent-specs/2026-06-20-planB-phase2-harness/rubric-protocol.md
@ -0,0 +1,34 @@
+# U7 设计完整性 rubric 评分协议（Plan B Phase-2 / R9）
+
+> turnkey 人工打分协议。配套工具 `rubric_score.py`（抽样/聚合/判门，逐项分由人工给）。
+> **gated on gen-done**：须 Plan A ≥80% 批就绪后跑；与 [Plan B 计划](../../plans/2026-06-19-001-feat-coredone-验收闭环执行-plan.md) U7 对齐。
+
+## 为什么要 rubric（与九门划界）
+九门 real-play harness（`game-runtime/games/_wg1-gen/_shared/play.cdp.cjs`：F_wiring / I_control / H_progress + latch）测的是「**能不能玩**」——接线通、能操作、有进展。
+rubric 只补**九门测不到的「设计质量」层**——「**玩得有没有意思 / 像不像那句话**」。两者**不重复打分**：九门过是 rubric 的前置，rubric 只评设计完整度。
+
+## 5 项评分锚点（各 0-2）
+
+| 项 | 0 | 1 | 2 |
+|---|---|---|---|
+| **资产对位**（`asset_match`·**一票否决**） | 素材与一句话意图无关/缺失/全占位色块 | 部分对位（主角或核心物对，余泛化） | 素材明确对位意图（主角/场景/物件可辨认且贴题） |
+| **玩法目标完整**（`goal`） | 无目标/无胜负 | 有隐含目标但无明确达成或失败提示 | 明确可达成/失败目标 + 状态可见（分数/进度/结局） |
+| **反馈充分**（`feedback`） | 操作无反馈 | 仅视觉或仅数值单一反馈 | 操作有即时视听反馈（动效 + 数值/屏提示） |
+| **难度合理**（`difficulty`） | 无难度（秒赢/不可玩）或必败 | 有难度但曲线突兀/单一 | 难度合理且有递进感 |
+| **内容非退化**（`non_degen`） | 空壳/占位/重复糊弄/与他款雷同 | 基本成型但单薄 | 内容完整、非套模板糊弄 |
+
+## 判门（plan 003-U7 Verification）
+**N=8 评分表齐** + **总均分 ≥ 1.4**（所有款×项均值）+ **资产对位项无 0**（任一款 0 分 → 整体否决）。
+
+## 执行步骤（gen-done 后）
+1. **抽样**：取 Plan A ≥80% 批 manifest（list[gameId|对象]）→
+   `python3 rubric_score.py --template --batch <批manifest.json> -n 8 -o sheet.json`
+   （确定性等距抽 8 款、可复现；不足 8 则全取并记 gap）。
+2. **人工打分**：创始人或指派对每款**亲玩或观玩**（可借 CDP 录屏/截图辅助回看），按上表逐项填 `sheet.json` 各 `scores`（0/1/2）+ `note` 写依据。
+3. **判门**：`python3 rubric_score.py --score sheet.json` → 输出逐款 + `RUBRIC_VERDICT=PASS/FAIL`。
+4. **留证**：`sheet.json` + 判门输出 = U7 证据；连同 Plan A 批 id 一并归档。
+
+## 诚实口径
+- 工具**不代打分**——逐项分必须人工出（设计质量是主观判断，机器只做抽样/算术/判门）。
+- 抽样确定性（等距）= 可复现、防「挑好的打」；如需随机抽样另议（不引入不可复现性）。
+- rubric **不替代**九门/U8 e2e：它是 Phase-1「done」里「生成内容质量」那一层的独立证据。
--- a/docs/agent-specs/2026-06-20-planB-phase2-harness/rubric_score.py
+++ b/docs/agent-specs/2026-06-20-planB-phase2-harness/rubric_score.py
@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+U7 设计完整性 rubric 打分仪（Plan B Phase-2 / R9）—— turnkey 工具。
+
+用途（gen-done 落地后跑）：
+  1. 从 Plan A ≥80% 批的 manifest 抽 N=8 款 → 生成空评分表（人工逐项打分）；
+  2. 收回填好的评分表 → 聚合 + 判门（均分 ≥ 1.4 且 每款「资产对位」≥ 1[一票否决]）。
+
+rubric 5 项（各 0-2，与九门划界：只补九门 F/I/H/control 测不到的「设计质量」层）：
+  - asset_match  资产对位     —— 生成游戏的素材与一句话意图对位（**一票否决**：任一款 0 分则整体不过）
+  - goal         玩法目标完整 —— 有明确可达成/失败的目标
+  - feedback     反馈充分     —— 操作有即时视听反馈
+  - difficulty   难度合理     —— 非过易/过难/无难度曲线
+  - non_degen    内容非退化   —— 非空壳/占位/重复糊弄
+
+判门（plan 003-U7 Verification）：N=8 评分表齐 + 总均分 ≥ 1.4 + 资产对位项无 0。
+
+CLI：
+  python rubric_score.py --template --batch <manifest.json> [-n 8] [-o sheet.json]
+  python rubric_score.py --score <sheet.json>
+  python rubric_score.py --selftest
+诚实口径：本工具只做抽样/聚合/判门；**逐项分由人工（创始人或指派）亲玩/观玩给出**，不代打分。
+"""
+import argparse
+import json
+import sys
+
+# rubric 5 项的 key 与中文标签；ASSET_MATCH 为一票否决项
+ITEMS = [
+    ("asset_match", "资产对位（一票否决）"),
+    ("goal", "玩法目标完整"),
+    ("feedback", "反馈充分"),
+    ("difficulty", "难度合理"),
+    ("non_degen", "内容非退化"),
+]
+ITEM_KEYS = [k for k, _ in ITEMS]
+VETO_KEY = "asset_match"          # 一票否决项
+N_DEFAULT = 8                      # 抽样款数（plan：抽 8 款）
+MEAN_GATE = 1.4                    # 总均分门
+VETO_MIN = 1                       # 资产对位每款下限
+
+
+def sample_n(items, n=N_DEFAULT):
+    """从批 manifest 确定性均匀抽 N 款（按 gameId 排序后等距取，可复现、避头部偏置）。"""
+    # manifest 容差：list[str|dict]；dict 取 gameId/id/versionId 之一作标识
+    def gid(x):
+        if isinstance(x, dict):
+            for k in ("gameId", "id", "versionId", "game_id"):
+                if x.get(k) is not None:
+                    return str(x[k])
+            return json.dumps(x, ensure_ascii=False, sort_keys=True)
+        return str(x)
+    ids = sorted({gid(x) for x in items})
+    if not ids:
+        raise ValueError("批 manifest 为空，无法抽样")
+    if len(ids) <= n:
+        return ids                                  # 不足 N 则全取（并由调用方记 gap）
+    # 等距取 n 个下标（含首尾）
+    step = (len(ids) - 1) / (n - 1)
+    picked = [ids[round(i * step)] for i in range(n)]
+    # round 可能撞重，去重后顺序补足
+    seen, out = set(), []
+    for g in picked:
+        if g not in seen:
+            seen.add(g); out.append(g)
+    i = 0
+    while len(out) < n and i < len(ids):
+        if ids[i] not in seen:
+            seen.add(ids[i]); out.append(ids[i])
+        i += 1
+    return out
+
+
+def make_template(game_ids):
+    """生成空评分表：每款 5 项置 null（待人工填 0-2）。"""
+    return {
+        "rubric": "U7 设计完整性 rubric（5项×0-2；资产对位一票否决；总均分≥1.4）",
+        "items": [{"key": k, "label": lbl} for k, lbl in ITEMS],
+        "n": len(game_ids),
+        "games": [
+            {"id": g, "scores": {k: None for k in ITEM_KEYS}, "note": ""}
+            for g in game_ids
+        ],
+    }
+
+
+def validate(sheet):
+    """校验评分表完整性：每款 5 项齐、各项 0/1/2。返回 (ok, errors)。"""
+    errs = []
+    games = sheet.get("games") or []
+    if not games:
+        errs.append("评分表无 games")
+        return False, errs
+    if len(games) != N_DEFAULT:
+        errs.append(f"款数={len(games)} ≠ 约定 {N_DEFAULT}（抽样口径需复核）")
+    for g in games:
+        gid = g.get("id", "?")
+        sc = g.get("scores") or {}
+        for k in ITEM_KEYS:
+            v = sc.get(k)
+            if v is None:
+                errs.append(f"[{gid}] 缺项 {k}（未打分）")
+            elif v not in (0, 1, 2):
+                errs.append(f"[{gid}] {k}={v} 非法（须 0/1/2）")
+    return (len([e for e in errs if "非法" in e or "缺项" in e or "无 games" in e]) == 0), errs
+
+
+def aggregate(sheet):
+    """聚合：每款总分/均分 + 总均分（所有款×项的均值）+ 资产对位是否触发否决。"""
+    games = sheet.get("games") or []
+    per_game, all_scores, veto_fail = [], [], []
+    for g in games:
+        sc = g.get("scores") or {}
+        vals = [sc[k] for k in ITEM_KEYS if isinstance(sc.get(k), int)]
+        total = sum(vals)
+        per_game.append({"id": g.get("id"), "total": total,
+                         "mean": round(total / len(ITEM_KEYS), 3) if vals else 0,
+                         "asset_match": sc.get(VETO_KEY)})
+        all_scores.extend(vals)
+        if sc.get(VETO_KEY) == 0:
+            veto_fail.append(g.get("id"))
+    overall_mean = round(sum(all_scores) / len(all_scores), 3) if all_scores else 0.0
+    return {"per_game": per_game, "overall_mean": overall_mean, "veto_fail": veto_fail}
+
+
+def gate(sheet):
+    """判门：总均分≥1.4 且 无款资产对位=0。返回 (passed, verdict_dict)。"""
+    ok_valid, errs = validate(sheet)
+    agg = aggregate(sheet)
+    mean_ok = agg["overall_mean"] >= MEAN_GATE
+    veto_ok = len(agg["veto_fail"]) == 0
+    passed = ok_valid and mean_ok and veto_ok
+    return passed, {
+        "passed": passed,
+        "valid": ok_valid,
+        "overall_mean": agg["overall_mean"],
+        "mean_gate": MEAN_GATE,
+        "mean_ok": mean_ok,
+        "veto_ok": veto_ok,
+        "veto_fail_games": agg["veto_fail"],
+        "per_game": agg["per_game"],
+        "errors": errs,
+    }
+
+
+def _print_verdict(v):
+    print(f"=== U7 rubric 判门 ===")
+    print(f"  评分表合法: {'✓' if v['valid'] else '✗'}")
+    print(f"  总均分: {v['overall_mean']} （门 ≥ {v['mean_gate']}）-> {'✓' if v['mean_ok'] else '✗'}")
+    print(f"  资产对位一票否决: {'✓ 无 0 分款' if v['veto_ok'] else '✗ 0 分款=' + ','.join(map(str, v['veto_fail_games']))}")
+    print("  逐款: " + "; ".join(f"{g['id']}(均{g['mean']}/对位{g['asset_match']})" for g in v["per_game"]))
+    if v["errors"]:
+        print("  问题:")
+        for e in v["errors"]:
+            print(f"    - {e}")
+    print(f"  RUBRIC_VERDICT={'PASS' if v['passed'] else 'FAIL'}")
+
+
+def _selftest():
+    """合成数据自测：覆盖 通过 / 均分不足 / 否决 / 缺项 四类。"""
+    # 1) 全 2 分 → 通过
+    s_pass = make_template([f"g{i}" for i in range(8)])
+    for g in s_pass["games"]:
+        g["scores"] = {k: 2 for k in ITEM_KEYS}
+    p, v = gate(s_pass); assert p and v["overall_mean"] == 2.0, v
+    # 2) 全 1 分 → 均分 1.0 < 1.4 → 不过（但无否决）
+    s_low = make_template([f"g{i}" for i in range(8)])
+    for g in s_low["games"]:
+        g["scores"] = {k: 1 for k in ITEM_KEYS}
+    p, v = gate(s_low); assert (not p) and v["veto_ok"] and not v["mean_ok"], v
+    # 3) 均分够但一款资产对位=0 → 否决不过
+    s_veto = make_template([f"g{i}" for i in range(8)])
+    for g in s_veto["games"]:
+        g["scores"] = {k: 2 for k in ITEM_KEYS}
+    s_veto["games"][3]["scores"]["asset_match"] = 0
+    p, v = gate(s_veto); assert (not p) and (not v["veto_ok"]) and v["veto_fail_games"] == ["g3"], v
+    # 4) 缺项 → 不合法不过
+    s_miss = make_template([f"g{i}" for i in range(8)])
+    for g in s_miss["games"]:
+        g["scores"] = {k: 2 for k in ITEM_KEYS}
+    s_miss["games"][0]["scores"]["goal"] = None
+    p, v = gate(s_miss); assert (not p) and (not v["valid"]), v
+    # 5) 边界：均分恰 1.4（5项里 3×2+2×1=8/5=1.6;构造 7/5=1.4）
+    s_edge = make_template([f"g{i}" for i in range(8)])
+    for g in s_edge["games"]:
+        g["scores"] = {"asset_match": 2, "goal": 2, "feedback": 1, "difficulty": 1, "non_degen": 1}  # 7/5=1.4
+    p, v = gate(s_edge); assert p and v["overall_mean"] == 1.4 and v["mean_ok"], v
+    # 6) 抽样：13 款抽 8、确定性可复现
+    batch = [{"gameId": f"x{i:02d}"} for i in range(13)]
+    a = sample_n(batch, 8); b = sample_n(batch, 8)
+    assert len(a) == 8 and a == b and len(set(a)) == 8, a
+    # 7) 抽样不足 N：5 款抽 8 → 全取 5
+    assert len(sample_n([{"id": i} for i in range(5)], 8)) == 5
+    print("rubric_score selftest: 7/7 PASS")
+
+
+def main():
+    ap = argparse.ArgumentParser(description="U7 rubric 打分仪（抽样/聚合/判门；逐项分人工给）")
+    ap.add_argument("--template", action="store_true", help="生成空评分表")
+    ap.add_argument("--batch", help="批 manifest（list[str|dict]）路径，配 --template 抽样")
+    ap.add_argument("-n", type=int, default=N_DEFAULT, help=f"抽样款数（默认 {N_DEFAULT}）")
+    ap.add_argument("-o", help="输出路径（默认 stdout）")
+    ap.add_argument("--score", help="已填评分表路径，聚合判门")
+    ap.add_argument("--selftest", action="store_true", help="合成数据自测")
+    args = ap.parse_args()
+
+    if args.selftest:
+        _selftest(); return 0
+    if args.template:
+        if not args.batch:
+            print("--template 需 --batch <manifest.json>", file=sys.stderr); return 2
+        with open(args.batch, encoding="utf-8") as f:
+            items = json.load(f)
+        if isinstance(items, dict):                 # 容差：{games:[...]} 或 {list:[...]}
+            items = items.get("games") or items.get("list") or items.get("data") or []
+        ids = sample_n(items, args.n)
+        tpl = make_template(ids)
+        out = json.dumps(tpl, ensure_ascii=False, indent=2)
+        if args.o:
+            with open(args.o, "w", encoding="utf-8") as f:
+                f.write(out)
+            print(f"空评分表已写 {args.o}（{len(ids)} 款，待人工填 0-2）")
+        else:
+            print(out)
+        return 0
+    if args.score:
+        with open(args.score, encoding="utf-8") as f:
+            sheet = json.load(f)
+        passed, v = gate(sheet)
+        _print_verdict(v)
+        return 0 if passed else 1
+    ap.print_help(); return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())
--- a/docs/agent-specs/2026-06-20-planB-phase2-harness/u8_feed_play_e2e.py
+++ b/docs/agent-specs/2026-06-20-planB-phase2-harness/u8_feed_play_e2e.py
@ -0,0 +1,249 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+U8 创作→生成→feed真玩 + 资产可见 e2e（Plan B Phase-2 / R4 + R11）—— turnkey 驱动。
+
+链路（一句话端到端）：create_draft → studio_generate → poll 直到生成完成 → submit_publish
+  → review_approve → feed_find_game(命中) → get_project(status=已发布) → play.cdp.cjs 九门真玩
+  → 断言四件套 + 终局遥测 + **资产对位命中(R11)**。
+
+**gated on gen-done**：编排步须 Plan A SAA 路能产「可玩游戏」后才真跑（mini-desktop，与 Plan A e2e 串行）。
+本文件**先建好 + 纯逻辑断言 selftest 现验**；gen-done 落地后 `--run` 即跑、零搭建。
+
+复用（不重造）：
+  - orchestrator/backend_gw.py `BackendGateway`：全链 API（create/generate/publish/review/feed/project）。
+  - game-runtime/games/_wg1-gen/_shared/play.cdp.cjs：九门四件套（装 bundle→驱 Chrome→verdict.json + 截图）。
+对齐契约：六类资产 = build-from-source.mjs:36；终局事件 = `game_end`（contracts 实证）。
+"""
+import argparse
+import json
+import os
+import subprocess
+import sys
+import time
+
+# 六类资产类目（冻结·对齐 game-runtime/src/host/build-from-source.mjs:36 + source-project.schema）
+ASSET_CATEGORIES = {"sprite", "character", "effect", "scene", "ui", "music"}
+GAME_END_EVENT = "game_end"          # 终局遥测事件（contracts 实证）
+PUBLISHED_STATUS = 4                  # game_project.status 已发布（U2 实证 1→4）
+ORCH_DIR = os.path.join(os.path.dirname(__file__), "..", "2026-06-09-agent-loop-v1", "orchestrator")
+PLAY_CDP = os.path.join(os.path.dirname(__file__), "..", "..", "..",
+                        "game-runtime", "games", "_wg1-gen", "_shared", "play.cdp.cjs")
+
+
+# ============== 纯逻辑断言（gen-done 前即可 selftest）==============
+
+def assert_asset_match(game_def):
+    """R11 资产对位：assets 非空 + 每项 id/category(六类)/ref(非空串) + 被 sprite render 组件引用(=可见)。"""
+    errs = []
+    gd = game_def or {}
+    assets = gd.get("assets") or []
+    if not assets:
+        errs.append("gameDefinition.assets 为空（无资产=对位不命中）")
+    asset_ids = set()
+    for i, a in enumerate(assets):
+        aid = a.get("id", f"#{i}")
+        if a.get("id") is None:
+            errs.append(f"asset[{aid}] 缺 id")
+        else:
+            asset_ids.add(str(a.get("id")))
+        if str(a.get("category")) not in ASSET_CATEGORIES:
+            errs.append(f"asset[{aid}] category 非六类(sprite/character/effect/scene/ui/music): {a.get('category')}")
+        ref = a.get("ref")
+        if not isinstance(ref, str) or not ref.strip():
+            errs.append(f"asset[{aid}] ref 须非空字符串")
+    # 资产"可见"= 至少一个 sprite render 组件引用了某 asset.id（render.asset 或 render.ref）
+    comps = gd.get("components") or []
+    used = set()
+    for c in comps:
+        if c.get("shape") == "sprite":
+            if c.get("asset") is not None:
+                used.add(str(c.get("asset")))
+            if c.get("ref") is not None:
+                used.add(str(c.get("ref")))
+    if assets and not (used & asset_ids):
+        errs.append("无 sprite render 组件引用 assets（声明了资产但没画=不可见）")
+    return (len(errs) == 0, errs)
+
+
+def assert_published(project):
+    """game_project.status == 已发布(4)。"""
+    st = (project or {}).get("status")
+    return (st == PUBLISHED_STATUS, st)
+
+
+def assert_in_feed(found_item):
+    """feed_find_game 命中（非 None/非空）。"""
+    return (bool(found_item), found_item)
+
+
+def assert_telemetry(events):
+    """遥测含 game_end 终局事件 + ≥1 事件。事件名容差 event/eventType/type 字段。"""
+    names = [(e.get("event") or e.get("eventType") or e.get("type")) for e in (events or [])]
+    return (GAME_END_EVENT in names and len(names) >= 1, names)
+
+
+# ============== 编排（gated on gen-done）==============
+
+def run_e2e(prompt, cdp_endpoint, template_id="generic", poll_s=10, timeout_s=240, evidence_dir=None):
+    """一句话端到端真跑。gen-done 后用；返回 evidence dict。失败抛 RuntimeError 带步名。"""
+    sys.path.insert(0, ORCH_DIR)
+    from backend_gw import BackendGateway  # 延迟导入：--selftest 不需依赖
+
+    gw = BackendGateway()
+    ev = {"prompt": prompt, "steps": {}, "asserts": {}}
+
+    # 1. 创作：一句话建草稿
+    draft = gw.create_draft(title=prompt[:20], template_id=template_id, prompt=prompt)
+    session_id = (draft or {}).get("id")
+    ev["steps"]["create_draft"] = {"session_id": session_id, "ok": session_id is not None}
+    if not session_id:
+        raise RuntimeError("create_draft 未拿到 sessionId")
+
+    # 2. 生成：入队 + 轮询直到完成（SAA 路）
+    task = gw.studio_generate(session_id=session_id, prompt=prompt)
+    task_id = (task or {}).get("id") or (task or {}).get("taskId")
+    deadline = time.time() + timeout_s
+    status, game_id, version_id = None, None, None
+    while time.time() < deadline:
+        t = gw.get_aigc_task(task_id)
+        status = (t or {}).get("status")
+        game_id = (t or {}).get("gameId") or game_id
+        version_id = (t or {}).get("versionId") or version_id
+        if status in ("SUCCESS", "DONE", 2, "2") and version_id:
+            break
+        if status in ("FAILED", "ERROR", 3, "3"):
+            raise RuntimeError(f"生成失败 status={status}")
+        time.sleep(poll_s)
+    ev["steps"]["generate"] = {"task_id": task_id, "status": status, "game_id": game_id, "version_id": version_id}
+    if not version_id:
+        raise RuntimeError(f"生成超时/未拿到 versionId（status={status}）")
+
+    # 3. 发布 + 审核通过
+    gw.submit_publish(game_id=game_id, version_id=version_id)
+    gw.review_approve(game_id=game_id, version_id=version_id, reason="U8 e2e auto-approve")
+    project = gw.get_project(game_id)
+    ev["steps"]["publish"] = {"project_status": (project or {}).get("status")}
+
+    # 4. feed 命中
+    found = gw.feed_find_game(game_id)
+    ev["steps"]["feed"] = {"found": bool(found)}
+
+    # 5. 资产对位（R11）：从 manifest/gameDefinition 取 assets 比对
+    gd = _extract_game_def(gw, version_id)
+    ev["asserts"]["asset_match"] = dict(zip(("ok", "errs"), assert_asset_match(gd)))
+
+    # 6. 真玩四件套（play 路二选一，见 _play_via_cdp 决策点）
+    play = _play_via_cdp(game_id, version_id, cdp_endpoint, evidence_dir)
+    ev["steps"]["play"] = play
+
+    # 7. 终局遥测（经 gw 拉该 session 事件；接口名以 backend_gw 实际为准，缺则记 gap）
+    events = _fetch_telemetry(gw, game_id, version_id)
+    ev["asserts"]["telemetry"] = dict(zip(("ok", "names"), assert_telemetry(events)))
+
+    # 汇总断言
+    ev["asserts"]["published"] = dict(zip(("ok", "status"), assert_published(project)))
+    ev["asserts"]["in_feed"] = dict(zip(("ok", "found"), assert_in_feed(found)))
+    ev["passed"] = all([
+        ev["asserts"]["published"]["ok"], ev["asserts"]["in_feed"]["ok"],
+        ev["asserts"]["asset_match"]["ok"], play.get("verdict") == "PASS",
+        ev["asserts"]["telemetry"]["ok"],
+    ])
+    return ev
+
+
+def _extract_game_def(gw, version_id):
+    """从 manifest/package 取运行时 gameDefinition（含 assets）。形态以 backend_gw get_manifest_raw 为准。"""
+    try:
+        raw = gw.get_manifest_raw(version_id)
+        m = json.loads(raw) if isinstance(raw, str) else raw
+        # 容差：manifest 可能直挂 gameDefinition / 经 package descriptor
+        return m.get("gameDefinition") or m.get("game_definition") or m
+    except Exception as e:
+        return {"_extract_error": str(e)}
+
+
+def _play_via_cdp(game_id, version_id, cdp_endpoint, evidence_dir, base="http://localhost:4320"):
+    """真玩四件套。**两条 play 路，turnkey 时按 gen-done 产出形态择一**：
+
+      路A（产品 feed 真玩·R4 权威）：经 studio 前端 :4173 `/play/<gameId>` 的 GamePlayer iframe
+        真玩（同 U5/u5u2_walk 的 player_cdp 驱动，nine-gate 式断言：非空 canvas + 交互 + readGameState）。
+        —— 当 Plan A 产「staging 发布游戏」时走此路（用户真实路径）。
+      路B（本地 bundle 九门）：`node play.cdp.cjs <gameId> --base= --cdp=`（实证 CLI：位置参 gameId + = 形式），
+        玩 games/_wg1-gen/<gameId>/ 本地 bundle（须先 build-from-source→esbuild→serve :4320）。
+        —— 当 Plan A 产「本地 gameDefinition」时走此路（生成质量门）。
+
+    本函数默认尝试路B（play.cdp.cjs 实存即调，CLI 已对齐）；路A 的前端驱动复用 orchestrator/player_cdp.py，
+    在 gen-done 接线时按产出形态切换（二者断言口径同=nine-gate）。**未决部分如实标 deferred，不伪报。**
+    """
+    if not os.path.exists(PLAY_CDP):
+        return {"verdict": "DEFERRED", "path": "A/B", "version_id": version_id,
+                "reason": f"play.cdp.cjs 不在({PLAY_CDP});路A(前端)在 gen-done 接线时用 player_cdp 驱 :4173/play/{game_id}"}
+    # 路B：本地 bundle 九门（CLI 实证 = 位置参 gameId + --base=/--cdp=）
+    cmd = ["node", PLAY_CDP, str(game_id), f"--base={base}", f"--cdp={cdp_endpoint}"]
+    try:
+        p = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
+        return {"verdict": "PASS" if p.returncode == 0 else "FAIL", "path": "B(本地bundle九门)",
+                "exit": p.returncode, "tail": (p.stdout or p.stderr)[-400:]}
+    except Exception as e:
+        return {"verdict": "ERROR", "reason": str(e)}
+
+
+def _fetch_telemetry(gw, game_id, version_id):
+    """拉该 game 的遥测事件。backend_gw 若无现成接口则返回 [] 并由调用方记 gap。"""
+    for name in ("get_telemetry", "telemetry_events", "get_game_events"):
+        fn = getattr(gw, name, None)
+        if callable(fn):
+            try:
+                return fn(game_id) or []
+            except Exception:
+                return []
+    return []   # gap：无遥测查询接口 → 终局断言记 gap（非编造为过）
+
+
+# ============== selftest（纯逻辑断言）==============
+
+def _selftest():
+    # 资产对位：合法（sprite render 引用 asset）
+    gd_ok = {"assets": [{"id": "hero", "category": "sprite", "ref": "mat-1"}],
+             "components": [{"shape": "sprite", "asset": "hero"}]}
+    ok, errs = assert_asset_match(gd_ok); assert ok, errs
+    # 空 assets → 不命中
+    ok, errs = assert_asset_match({"assets": [], "components": []}); assert not ok and any("为空" in e for e in errs), errs
+    # 非六类 + 非串 ref
+    gd_bad = {"assets": [{"id": "a", "category": "weapon", "ref": ""},
+                          {"id": "b", "category": "sprite", "ref": {"x": 1}}],
+              "components": [{"shape": "sprite", "asset": "a"}]}
+    ok, errs = assert_asset_match(gd_bad)
+    assert (not ok) and any("非六类" in e for e in errs) and sum("ref 须非空" in e for e in errs) >= 2, errs
+    # 声明资产但无 render 引用 → 不可见
+    gd_orphan = {"assets": [{"id": "h", "category": "sprite", "ref": "m"}], "components": [{"shape": "rect"}]}
+    ok, errs = assert_asset_match(gd_orphan); assert (not ok) and any("没画" in e for e in errs), errs
+    # published / in_feed / telemetry
+    assert assert_published({"status": 4})[0] and not assert_published({"status": 1})[0]
+    assert assert_in_feed({"gameId": 1})[0] and not assert_in_feed(None)[0]
+    assert assert_telemetry([{"event": "game_start"}, {"event": "game_end"}])[0]
+    assert not assert_telemetry([{"event": "game_start"}])[0]   # 缺终局
+    print("u8_feed_play_e2e selftest: 8/8 PASS（纯逻辑断言；编排步 gated on gen-done）")
+
+
+def main():
+    ap = argparse.ArgumentParser(description="U8 创作→生成→feed真玩 e2e（gated on gen-done）")
+    ap.add_argument("--selftest", action="store_true", help="纯逻辑断言自测（gen-done 前可跑）")
+    ap.add_argument("--run", action="store_true", help="真跑 e2e（须 gen-done + mini-desktop CDP）")
+    ap.add_argument("--prompt", default="一个躲避陨石的小游戏", help="一句话")
+    ap.add_argument("--cdp", default="http://localhost:9222", help="mini-desktop CDP 端点")
+    ap.add_argument("--out", help="证据输出目录")
+    args = ap.parse_args()
+    if args.selftest:
+        _selftest(); return 0
+    if args.run:
+        ev = run_e2e(args.prompt, args.cdp, evidence_dir=args.out)
+        print(json.dumps(ev, ensure_ascii=False, indent=2))
+        print(f"U8_VERDICT={'PASS' if ev.get('passed') else 'FAIL'}")
+        return 0 if ev.get("passed") else 1
+    ap.print_help(); return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())
--- a/docs/agent-specs/2026-06-20-planB-phase2-harness/u9_m0_core_gate.py
+++ b/docs/agent-specs/2026-06-20-planB-phase2-harness/u9_m0_core_gate.py
@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+U9 M0 core 联合验收门（Plan B Phase-2 / origin 003-U3）—— turnkey 门。
+
+M0 core = 「一句话真生成可玩游戏 + 三契约对齐」：
+  1. 跑 U8 e2e（create→生成→发布→feed→真玩，u8_feed_play_e2e.run）= 一句话端到端可玩；
+  2. **三契约结构核**（运行时读契约 required，防漂移）：
+     - API：CommonResult 信封（code==0 + data 非空）—— runtime/package · feed · aigc 关键响应；
+     - dispatcher：SAA 产出 = source-project.schema.json（schemaVersion/sourceHash/profile/gameDefinition）；
+     - manifest：游戏包 = game-package.schema.json（schemaVersion/gameId/versionId/templateId/gameConfig/assets/manifest/meta）。
+门：U8 PASS ∧ 三契约全对齐。
+
+**gated on gen-done + U8**：编排步须 Plan A 产可玩游戏后跑（mini-desktop）。本文件先建 + 三契约结构核 selftest 现验。
+"""
+import argparse
+import json
+import os
+import subprocess
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = os.path.abspath(os.path.join(HERE, "..", "..", ".."))     # 仓根（harness 在 docs/agent-specs/<dir>/）
+MANIFEST_SCHEMA = os.path.join(REPO, "contracts", "game-package.schema.json")
+DISPATCHER_SCHEMA = os.path.join(REPO, "contracts", "agent-loop", "source-project.schema.json")
+U8 = os.path.join(HERE, "u8_feed_play_e2e.py")
+
+
+def _required(schema_path):
+    """运行时读契约 required（防硬编码漂移）；缺文件抛错。"""
+    with open(schema_path, encoding="utf-8") as f:
+        return json.load(f).get("required") or []
+
+
+def check_required(obj, required):
+    """结构核：required 键齐则 ([], True)，否则 (缺键列表, False)。"""
+    obj = obj or {}
+    missing = [k for k in required if k not in obj]
+    return missing, len(missing) == 0
+
+
+# ============== 三契约结构核（gen-done 前可 selftest）==============
+
+def assert_api(envelope):
+    """API 契约：CommonResult 信封 code==0 + data 非空。"""
+    e = envelope or {}
+    ok = (e.get("code") == 0) and (e.get("data") is not None)
+    return ok, {"code": e.get("code"), "has_data": e.get("data") is not None}
+
+
+def assert_dispatcher(source_project):
+    """dispatcher 契约：SAA 产出对齐 source-project.schema required。"""
+    missing, ok = check_required(source_project, _required(DISPATCHER_SCHEMA))
+    return ok, {"missing": missing}
+
+
+def assert_manifest(package):
+    """manifest 契约：游戏包对齐 game-package.schema required。"""
+    missing, ok = check_required(package, _required(MANIFEST_SCHEMA))
+    return ok, {"missing": missing}
+
+
+def three_contracts(api_envelope, source_project, package):
+    """三契约联合核。返回 (passed, detail)。"""
+    a_ok, a = assert_api(api_envelope)
+    d_ok, d = assert_dispatcher(source_project)
+    m_ok, m = assert_manifest(package)
+    return (a_ok and d_ok and m_ok), {"api": {"ok": a_ok, **a},
+                                       "dispatcher": {"ok": d_ok, **d},
+                                       "manifest": {"ok": m_ok, **m}}
+
+
+# ============== 联合门（gated on gen-done + U8）==============
+
+def run_m0_core(prompt, cdp_endpoint, out_dir=None):
+    """跑 U8 e2e + 三契约核。返回 verdict dict。"""
+    # 1. U8 e2e（一句话端到端可玩）—— shell u8，读其 JSON 输出
+    cmd = [sys.executable, U8, "--run", "--prompt", prompt, "--cdp", cdp_endpoint]
+    if out_dir:
+        cmd += ["--out", out_dir]
+    p = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
+    try:
+        # u8 末行打 U8_VERDICT，前面是 JSON；取首个 { 到末个对应 }
+        txt = p.stdout
+        u8_ev = json.loads(txt[txt.index("{"):txt.rindex("}") + 1])
+    except Exception:
+        u8_ev = {"passed": False, "_parse_error": (p.stdout or p.stderr)[-400:]}
+    u8_pass = bool(u8_ev.get("passed"))
+
+    # 2. 三契约核 —— 从 U8 evidence 取 artifacts（api 信封 / source-project / manifest）
+    #    u8 在 steps 里留有 manifest/gameDef；此处按其 evidence 形态取，缺则记 gap（不伪报）。
+    api_env = (u8_ev.get("artifacts") or {}).get("api_envelope")
+    src_proj = (u8_ev.get("artifacts") or {}).get("source_project")
+    pkg = (u8_ev.get("artifacts") or {}).get("package")
+    tc_pass, tc = three_contracts(api_env, src_proj, pkg)
+    artifacts_present = all(x is not None for x in (api_env, src_proj, pkg))
+
+    passed = u8_pass and tc_pass and artifacts_present
+    return {
+        "passed": passed,
+        "u8_passed": u8_pass,
+        "three_contracts_passed": tc_pass,
+        "artifacts_present": artifacts_present,
+        "three_contracts": tc,
+        "u8": u8_ev,
+        "gap": None if artifacts_present else "U8 evidence 未带 api_envelope/source_project/package 三件 artifact——gen-done 接线时补 u8 artifacts 输出",
+    }
+
+
+# ============== selftest（三契约结构核）==============
+
+def _selftest():
+    # 用真契约文件读 required，构造 好/缺键 fixture
+    req_disp = _required(DISPATCHER_SCHEMA)
+    req_mani = _required(MANIFEST_SCHEMA)
+    assert req_disp and req_mani, "契约 required 读取为空"
+    good_disp = {k: "x" for k in req_disp}
+    good_mani = {k: "x" for k in req_mani}
+    good_api = {"code": 0, "data": {"x": 1}}
+    # 全好 → 过
+    ok, tc = three_contracts(good_api, good_disp, good_mani); assert ok, tc
+    # API code!=0 → 不过
+    ok, tc = three_contracts({"code": 500, "data": None}, good_disp, good_mani)
+    assert (not ok) and (not tc["api"]["ok"]), tc
+    # dispatcher 缺键 → 不过 + 列缺
+    bad_disp = {k: "x" for k in req_disp[1:]}   # 缺第一个
+    ok, tc = three_contracts(good_api, bad_disp, good_mani)
+    assert (not ok) and tc["dispatcher"]["missing"] == [req_disp[0]], tc
+    # manifest 缺键 → 不过
+    bad_mani = {k: "x" for k in req_mani[:-1]}   # 缺最后一个
+    ok, tc = three_contracts(good_api, good_disp, bad_mani)
+    assert (not ok) and tc["manifest"]["missing"] == [req_mani[-1]], tc
+    # 空对象 → 全缺
+    ok, tc = three_contracts({}, {}, {})
+    assert (not ok) and len(tc["dispatcher"]["missing"]) == len(req_disp), tc
+    print(f"u9_m0_core_gate selftest: 5/5 PASS（三契约结构核；读真契约 dispatcher={len(req_disp)}键 manifest={len(req_mani)}键）")
+
+
+def _print(v):
+    print("=== U9 M0 core 联合门 ===")
+    print(f"  U8 e2e（一句话端到端可玩）: {'✓' if v['u8_passed'] else '✗'}")
+    print(f"  artifacts 齐(api/dispatcher/manifest 三件): {'✓' if v['artifacts_present'] else '✗ ' + (v.get('gap') or '')}")
+    tc = v["three_contracts"]
+    print(f"  三契约: API {'✓' if tc['api']['ok'] else '✗'} | dispatcher {'✓' if tc['dispatcher']['ok'] else '✗ 缺' + str(tc['dispatcher'].get('missing'))} | manifest {'✓' if tc['manifest']['ok'] else '✗ 缺' + str(tc['manifest'].get('missing'))}")
+    print(f"  M0_CORE_VERDICT={'PASS' if v['passed'] else 'FAIL'}")
+
+
+def main():
+    ap = argparse.ArgumentParser(description="U9 M0 core 联合门（gated on gen-done + U8）")
+    ap.add_argument("--selftest", action="store_true", help="三契约结构核自测（读真契约；gen-done 前可跑）")
+    ap.add_argument("--run", action="store_true", help="真跑 M0 core（须 gen-done + mini-desktop CDP）")
+    ap.add_argument("--prompt", default="一个躲避陨石的小游戏")
+    ap.add_argument("--cdp", default="http://localhost:9222")
+    ap.add_argument("--out")
+    args = ap.parse_args()
+    if args.selftest:
+        _selftest(); return 0
+    if args.run:
+        v = run_m0_core(args.prompt, args.cdp, args.out)
+        _print(v)
+        return 0 if v["passed"] else 1
+    ap.print_help(); return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())