feat(planB-p2): U8 feed真玩 e2e 驱动 turnkey（R4+R11）——编排全链 + 资产对位断言
create→生成(SAA)→发布→审核→feed→真玩→断言；复用 orchestrator/backend_gw.py 全链 API + play.cdp.cjs 九门： - 纯逻辑断言（selftest 8/8）：资产对位(六类/非空串ref/render 引用=可见) / published(status4) / in-feed / 终局遥测(game_end) - 编排步 gated on gen-done（须 Plan A 产可玩游戏 + mini-desktop）；play 路二选一（产品前端 :4173 / 本地 bundle 九门）按产出形态择，已对齐 play.cdp.cjs 实证 CLI（位置参 gameId + --base=/--cdp=） - 遥测查询接口缺则记 gap（不伪报） turnkey：gen-done 落地 --run 即跑。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili
parent
bd70188e9b
commit
4261a638a0
@ -9,7 +9,7 @@
|
||||
| 单元 | 文件 | 状态 | 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.*`（接 player_cdp + play.cdp.cjs + feed CDP） | ⏳ 待建 | 一句话→SAA 生成→发布→feed→试玩→断言（非空帧/交互截图/published/遥测/资产命中） |
|
||||
| **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.*`（组合 U8 + 三契约核） | ⏳ 待建 | 跑 U8 e2e + 核 API/dispatcher/manifest 三契约对齐 |
|
||||
|
||||
## 门禁判据（plan 003 对齐）
|
||||
|
||||
@ -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())
|
||||
Loading…
x
Reference in New Issue
Block a user