games-development-ai/cheap-worker/hard_genre_xtheme.py

"""hard_genre_xtheme.py — 轻A 换皮鲁棒性【跨主题】验证(it5 的 brief 与模板同题=偏软,本轮换主题真考换皮)。

it5 的 5 个 brief 恰与模板品类同题(heritage=榫卯、feiyi 模板也=榫卯),换皮近乎恒等、验不出"跨主题换皮"能力。
本脚本给每个路由品类换一个【不同主题】的 brief(仍命中同一 per-genre 模板+写锁),真考:模型能否只写 game-logic
把一款「榫卯」模板换皮成「剪纸」、把「点客」换皮成「奶茶店」等——render/core 锁死下,靠数据位驱动出不同主题。

跑:cheap-worker/.venv/bin/python cheap-worker/hard_genre_xtheme.py（NO_PROXY + 凭据档自动解析 NEWAPI_KEY）。
"""

import asyncio
import json
import time
from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parent))
import cheap_studio  # noqa: E402
from cheap_genre_route import route_genre  # noqa: E402
from cheap_roles import SCAFFOLD_DESC_BY_TEMPLATE  # noqa: E402

# 跨主题 brief——每个仍路由到同一 per-genre 模板(关键词命中),但主题换了,真考换皮。
XTHEME_BRIEFS = [
    # 非遗:榫卯→剪纸(仍命中 feiyi:非遗/剪纸)
    ("feiyi-jianzhi", "一个还原传统剪纸技艺的非遗小游戏:按刻纸工序(起稿→折纸→刻镂→揭裱)逐步操作,每步有稳度节奏判定,完成得窗花成品评分"),
    # TRPG:奇幻地牢→赛博骰战(仍命中 trpg:掷骰/地下城)
    ("trpg-cyber", "一个赛博朋克掷骰爬塔:每层地下城遇一个黑客守卫,点击掷骰(可见骰点)按攻防结算,击败进下一层,失败结算,每3层升级选芯片天赋"),
    # 经营:通用点客→奶茶店(仍命中 shop:经营/奶茶店/顾客/点单)
    ("shop-milktea", "一个经营奶茶店:顾客排队点单,按配方调对饮品上对得分,收银攒钱升级解锁新品与自动出杯设备"),
    # 解谜:通用→数字华容道(仍命中 puzzle:解谜/华容道)
    ("puzzle-huarong", "一个数字华容道解谜:滑动数字块把它们按序归位,规则逐关递进,卡壳给提示,通关计步数炫耀成绩"),
    # 剧情:通用→校园悬疑(仍命中 story:互动小说/分支/多结局)
    ("story-campus", "一个校园悬疑互动小说:每屏一段剧情+2-3个选项,选择影响走向,揭开三个不同真相结局,有结局图鉴"),
]


async def _run_one(sem, tag, brief, port, cdp):
    async with sem:
        gid = f"xt-{tag}"
        t0 = time.time()
        scaffold_template, routed_genre = route_genre(brief)
        # 2026-07-08 创始人纠偏:不写锁,全放开(只锁 L1_FIXED 通用 plumbing)。
        write_whitelist = None
        scaffold_desc = SCAFFOLD_DESC_BY_TEMPLATE.get(scaffold_template) if scaffold_template else None
        try:
            summary = await cheap_studio.run_studio(
                gid, brief, run_gates=True, port=port, cdp_port=cdp,
                scaffold_template=scaffold_template, scaffold_desc=scaffold_desc,
                write_whitelist=write_whitelist, genre=routed_genre)
        except Exception as e:  # noqa: BLE001
            return {"tag": tag, "gid": gid, "template": scaffold_template, "locked": write_whitelist is not None,
                    "ok": False, "accepted": False, "floorPass": None, "acceptanceVersion": None, "verdictPass": None,
                    "fail": f"{type(e).__name__}: {e}", "wallSec": round(time.time() - t0, 1)}
        v = summary.get("verdict") or {}
        floor = summary.get("floor") or {}
        rich = summary.get("richness") or {}
        pt = summary.get("playtest") or {}                  # 测试员段(W-AXIS-V2 波2 批账观测口径)
        return {
            "tag": tag, "gid": gid, "template": scaffold_template or "_template(通用)",
            "locked": write_whitelist is not None,
            "ok": bool(summary.get("ok")),
            "accepted": bool(summary.get("accepted")),      # W-AXIS-V2:最终验收(四门 ∧ 测试员)
            "floorPass": floor.get("pass"),                 # 预筛权威 = 四门投影(取代 verdictPass)
            "acceptanceVersion": summary.get("acceptanceVersion"),
            "verdictPass": v.get("pass"),                   # harness 原生九门/七门,仅对照、不作判定输入
            "playtest": {                                   # 测试员裁决摘要(裁决/degraded/rolls/成本)
                "verdict": pt.get("verdict"), "degraded": pt.get("degraded"),
                "rollCount": pt.get("rollCount"), "costRmb": pt.get("costRmb"),
            } if pt else None,
            "failedGates": v.get("failedGates"), "attempts": summary.get("attempts"),
            "costRmb": summary.get("costRmb"), "richness": rich.get("score"),
            "wallSec": round(time.time() - t0, 1),
        }


async def main():
    sem = asyncio.Semaphore(2)
    tasks = [_run_one(sem, tag, brief, 4380 + i * 2, 9282 + i * 2)
             for i, (tag, brief) in enumerate(XTHEME_BRIEFS)]
    results = await asyncio.gather(*tasks)
    print("\n\n========== 轻A 跨主题换皮验证 ==========")
    passed = 0
    for r in results:
        acc = bool(r.get("accepted"))
        fp = r.get("floorPass") is True
        passed += 1 if acc else 0
        mark = "✅" if acc else ("⚠️测试员拒" if fp else ("⚠️FALSE-PASS" if r.get("ok") else "❌"))
        pt = r.get("playtest") or {}
        print(f"{mark} {r['tag']:16s} tpl={r.get('template')} lock={r.get('locked')} "
              f"accepted={r.get('accepted')} floorPass={r.get('floorPass')} ver={r.get('acceptanceVersion')} "
              f"verdictPass(对照)={r.get('verdictPass')} playtest={pt.get('verdict')}/rolls={pt.get('rollCount')} "
              f"ptDegraded={pt.get('degraded')} failedGates={r.get('failedGates')} "
              f"attempts={r.get('attempts')} rich={r.get('richness')}/12 "
              f"cost={r.get('costRmb')} ptCost=¥{pt.get('costRmb')} wall={r.get('wallSec')}s {('FAIL='+r['fail']) if r.get('fail') else ''}")
    print(f">>> 跨主题换皮达标(accepted 最终验收 = 四门 ∧ 测试员):{passed}/{len(results)}  达标率={passed/len(results):.0%}")
    out = Path(__file__).resolve().parent / "results" / "hard-genre-xtheme.json"
    out.parent.mkdir(exist_ok=True)
    out.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f">>> 结果 → {out}")


if __name__ == "__main__":
    asyncio.run(main())