diff --git a/cheap-worker/bake_off.py b/cheap-worker/bake_off.py index 99d7605e..9da0d050 100644 --- a/cheap-worker/bake_off.py +++ b/cheap-worker/bake_off.py @@ -56,6 +56,32 @@ def judge_genre(runs: list) -> dict: return {**base, "passRate": rate, "status": "meets" if meets else "below", "meets": meets} +def richness_dist(runs: list) -> dict: + """单品类丰富度分布(U-B1 · 纯报告 · **与达标判定完全正交**)。 + + richness 是 LLM judge(cheap_verify)读产物源码后给的非阻塞质量信号(命中数 0..8), + 与九门 verdict.pass 的达标判定**互不影响** —— 本函数只统计分布、绝不参与 meets。 + 非降级款收集 score 算均分/区间;降级款(LLM 评分失败/超时/空 src)单列计数; + 无 richness 字段的款(对照路 run_gates=False / 老产物)直接不计。 + """ + scored = [] + degraded = 0 + for r in runs: + rv = r.get("richness") + if not isinstance(rv, dict): + continue # 没跑 richness → 不计入分布 + if rv.get("degraded") or rv.get("score") is None: + degraded += 1 + continue + try: + scored.append(int(rv["score"])) + except (TypeError, ValueError): + degraded += 1 # 脏评分值兜底为降级,绝不抛 + mean = round(sum(scored) / len(scored), 2) if scored else None + return {"max": 8, "scoredCount": len(scored), "degradedCount": degraded, + "meanScore": mean, "scores": sorted(scored)} + + def aggregate(genre_runs: dict, required_genres: list) -> dict: """按品类聚合 + 整体达标判定(质量口径)。 @@ -72,6 +98,8 @@ def aggregate(genre_runs: dict, required_genres: list) -> dict: overall = (not missing) and (not below) unconverged_total = sum(per_genre[k]["unconverged"] for k in required_genres) sample_total = sum(per_genre[k]["total"] for k in required_genres) + # U-B1:LLM 丰富度分布(非阻塞报告 · 与达标判定正交)。只观察「游戏变丰富没」,绝不改 overallMeets。 + richness_by_genre = {k: richness_dist(genre_runs.get(k, [])) for k in required_genres} return { "perGenre": per_genre, "requiredGenres": required_genres, @@ -81,6 +109,7 @@ def aggregate(genre_runs: dict, required_genres: list) -> dict: "threshold": _PASS_THRESHOLD, "unconvergedTotal": unconverged_total, # 编排未收敛款数(M2 follow-up,不计入质量达标率) "sampleTotal": sample_total, + "richnessByGenre": richness_by_genre, # U-B1:逐品类丰富度分布(LLM judge 非阻塞信号,不入达标判据) "note": "质量口径(分母=收敛款,编排未收敛剔出、单列 unconvergedTotal 归 M2);" "本机/实验室口径,生产真实分布达标在 M3 复验;判定零 LLM(纯确定性九门 verdict.pass)", } @@ -117,7 +146,8 @@ def run_one_sync(genre_key: str, brief: str, k: int, port: int, cdp: int) -> dic return {"gid": gid, "passed": bool(v and v.get("pass")), "finished": summary.get("finished"), "failedGates": (v or {}).get("failedGates"), "breakerKind": (breaker or {}).get("kind"), # 未收敛原因(归 M2 编排稳定性诊断) - "costRmb": summary.get("costRmb")} + "costRmb": summary.get("costRmb"), + "richness": summary.get("richness")} # U-B1:additive 丰富度评分(run_studio 已写 summary,非阻塞) async def run_bakeoff(genre_keys: list, n: int, conc: int = 1, offset: int = 0, @@ -180,6 +210,11 @@ def print_report(report: dict) -> None: unc = f" +{j['unconverged']} 未收敛(剔出)" if j.get("unconverged") else "" print(f"[{k}] 质量 {j['passed']}/{j['converged']} = {j['passRate']} {j['status']}" f" (原始 {j['rawPassRate']}/{j['total']} 款){unc}", file=sys.stderr) + rd = report.get("richnessByGenre", {}).get(k, {}) # U-B1:丰富度只读旁注(非阻塞·不入达标) + if rd.get("scoredCount") or rd.get("degradedCount"): + deg = f" +{rd['degradedCount']} 降级" if rd.get("degradedCount") else "" + print(f" 丰富度(LLM·非阻塞)均分 {rd.get('meanScore')}/8 " + f"评分 {rd.get('scoredCount')} 款 {rd.get('scores')}{deg}", file=sys.stderr) flag = "达标 ✅" if report["overallMeets"] else "未达标 ❌" print(f">>> 整体:{flag}(逐品类阈值 {report['threshold']}) 缺样本:{report['missingGenres']} " f"不达标:{report['belowGenres']}", file=sys.stderr) diff --git a/cheap-worker/cheap_roles.py b/cheap-worker/cheap_roles.py index bc412940..4c7effe9 100644 --- a/cheap-worker/cheap_roles.py +++ b/cheap-worker/cheap_roles.py @@ -13,8 +13,21 @@ skill / 插件 api.d.ts / 起点范例。L3 context 由 agent 经 read_file 按 system prompt 正文与 prompt.mjs 保持语义一致：入口契约 createGame({plugins,bundle,viewport}) 五法 + _forensicsView、 红线（零裸时间随机 DOM / bundle.tick / ctx.time.nowMs() / 扁平注入插件 / 方法名以 api.d.ts 为准）、 本周收口的 footgun（sceneFsm.define 逐场景调、真返 void 的绘制/音效 API 别接返回值;drawButton 已改返命中矩形、可直接接住判命中）、输入契约（handleTap/handleKey）。 + +「生成丰富度」设计步（U-A1）：除指向 littlejs-game-dev.md（怎么写代码）外，再加一个「先设计后写码」步—— +经营/养成/放置/点客类先 read sim-business-game-design.md 取「设计什么才好玩」的范式（核心循环 + 进货库存资源环 ++ 3-4 级解锁阶梯 + 数值成长 + 音效清单），过 §10 八条好玩自检，并严守 §8 MVP-first 铁律（别一稿堆满，否则 +code agent read-thrash 不收敛）。八条自检对所有品类通用、sim-business 范式只对经营类适用。**这是给生成 agent 的 +设计创作指导（散文）、不是代码校验**——丰富度的「校验」走 cheap_verify.py 的纯 LLM judge（红线：丰富不丰富的 +判断需大模型、绝不写成 code-presence/正则/断言）。prompt.mjs（Node 双源）必须同改、语义一致（仅标点全/半角差异）。 """ +# C2a：运行时热取所需标准库（无三方依赖，系统 Python 可用）。 +import os +import re +import threading +from pathlib import Path + # prompt 指 agent 去 read 的真 skill（不手抄；skill 内含两条 footgun 的 ❌→✅ 清单）。 SKILL_PATH = ".agents/skills/littlejs-game-dev.md" @@ -35,9 +48,23 @@ _SYSTEM_PROMPT = """你是 A-model 游戏生成 agent。目标产物 = 一款** 2. read_file('⟦G⟧/README.md') —— 这份克隆起点的文件职责 +「你写什么 vs 调什么」边界 + ⚠bundle.tick 坑。 3. 用到某插件就 read_file 它的 api.d.ts（如 game-runtime/src/plugins/scene-fsm/api.d.ts、.../session-score/api.d.ts）看精确签名。 4. 你的起点 ⟦G⟧/src/game-logic.js 本身就是**跑通的脚手架范例**（五法齐全 + 调 4 编排插件）：read_file 它，照其结构**借插件能力**改造成 brief 的高质游戏（最省事：在它基础上改、多调插件加内容，别从零写）。 +5. **若 brief 属经营 / 养成 / 放置 / 点客这类「攒资源→成长→解锁」品类**：read_file('.agents/skills/sim-business-game-design.md') 取「设计什么才好玩」的范式（6 心理引擎 / §1 玩法范式含进货补货环 / §8 设计配方 / §10 反无趣 8 条自检）。littlejs-game-dev 教你「代码怎么写」、sim-business 教你「设计什么才好玩」——配对用：先想清好玩、再写正确。 + +【先设计后写码：决定这游戏好不好玩（关键，别跳过）】 +能跑 ≠ 好玩。动手写码前，先据 brief 在心里（或 game-logic.js 顶部一小段注释里）定一份**轻量玩法设计**，再实现： +- **核心循环**（一句话：玩家做什么 → 得什么即时反馈 → 怎么变强）； +- **资源环**（经营/放置类必含）：「进货 → 库存 → 售卖收钱 → 缺货补货」的软币循环，制造「赚→进→卖→再赚」的张力； +- **3–4 级解锁阶梯**：攒够阈值解锁新商品/区域/能力，任意时刻都露出「下一个锁」； +- **数值成长**：产出/成本随级上升、略带滚雪球感，别平淡线性； +- **音效清单**：收钱「叮」/ 升级欢呼 / 解锁号角（经 plugins.audioMusic；宁可程序化也别没有反馈音）。 +定完过一遍 sim-business §10 的 **8 条好玩自检**（①即时反馈 ②可见成长 ③下一个解锁 ④30 秒内首次升级/解锁 ⑤数值滚雪球 ⑥情感锚〔萌角色/拥有物〕 ⑦放置回归惊喜 ⑧音反馈）——命中越多越好玩，命中 ≤3 ≈ 能玩但无趣。 +**8 条自检对所有品类通用**（动作/消除/跑酷也照它要即时反馈 + 可见成长 + 音反馈）；**经营/养成/放置/点客类**再按 sim-business 取资源环/解锁阶梯/客流节奏范式。 + +【MVP-first 铁律（钉死·关乎你能不能收敛，别一稿堆满）】 +设计太满 → 你实现负担过重 → read/write 反复跳、跑不收敛（实测：满配设计循环截停、精简设计 9 步收敛）。首版**只做可玩核心**：核心循环 + 1 个主机制 + 1 个资源环（进货）+ 3–4 级解锁 + 基础数值/音效，**商品 ≤3 种起步**。**离线收益 / 看广告位 / 雇员 / 多档 BGM / hitstop / 6+ 商品 一律标「后续·MVP 不做」**、别塞进首版。**先出能玩的核心，再谈丰富。** 【步骤】 -A 读手册（1）+ 读你的起点 game-logic.js（4）→ B 按 brief 定玩法（核心循环 / 胜负 / 计分 / 数值）→ C 读要用的插件 api.d.ts → D write_file 改写 **game-logic.js**（按需拆 core/render）实现玩法 → E 调 check，有错改到 PASS → F 调 build，有错改到 PASS → G check+build 都绿后调 finish。 +A 读手册（1+5）+ 读你的起点 game-logic.js（4）→ B 据 brief 定**轻量玩法设计**（核心循环 / 资源环 / 解锁阶梯 / 数值 / 音效，过 §10 八条自检 + 守 MVP-first）→ C 读要用的插件 api.d.ts → D write_file 改写 **game-logic.js**（按需拆 core/render）实现玩法 → E 调 check，有错改到 PASS → F 调 build，有错改到 PASS → G check+build 都绿后调 finish。 【红线（check 会拦，违反则 finish 被拒）】 - game-logic.js 必须 **export function createGame({ plugins, bundle, viewport })**（命名导出、别改名、别默认导出）；**零引擎 import**（引擎经 boot.ctx.getEngine()）。 @@ -68,6 +95,224 @@ A 读手册（1）+ 读你的起点 game-logic.js（4）→ B 按 brief 定玩 _DEFAULT_SCAFFOLD_DESC = "一个跑通的脚手架起点「点圆得分」：menu→play→over 骨架 + 五法齐全 + 编排插件示范" +# ── C2a：cheap_roles 运行时热取配置 ───────────────────────────────────────────── +# 把 _SYSTEM_PROMPT 从「硬编码字符串」改成「运行时从 contracts/prompts 读 + mtime 缓存 + 回落内置」。 +# 复用 tier2/gen-worker/worker/prompts.py 同款思路（registry → .md 剥 frontmatter → 回落内置）， +# 但在本文件自包含实现（cheap_roles.py 无 tier2 import 依赖，tests/ 可裸运行）。 +# +# 取值三级（任一级取到即用）： +# ① env CHEAP_PROMPTS_DIR 指向目录里的 contracts/prompts/registry.yaml（临时换 prompt 不改仓内文件）； +# ② 仓内 contracts/prompts/registry.yaml 引用的 04-config/cheap-system.md 正文（默认）； +# ③ 内置 _SYSTEM_PROMPT（兜底，保证无任何外部依赖时行为 == 改造前，逐字节一致）。 +# +# 不变量（本相红线）： +# 没人改 .md、没设 env 时，build_system_prompt(...) 输出必须 == 改造前（纯内置路）逐字节一致。 +# 为此 04-config/cheap-system.md 正文 = _SYSTEM_PROMPT 逐字节对齐（含 ⟦⟧ 占位符）。 +# +# best-effort 铁律：任何读失败/registry 缺/正文空 → 回落 _SYSTEM_PROMPT + 一次性告警，绝不抛。 + +_CHEAP_PROMPT_ID = "config.cheap-system" + +# 本文件在 cheap-worker/cheap_roles.py → parents[1] = 仓库根 +_CR_REPO_ROOT = Path(__file__).resolve().parents[1] +_CR_DEFAULT_PROMPTS_DIR = _CR_REPO_ROOT / "contracts" / "prompts" + +# 运行时读状态（mtime 缓存 + 线程安全，与 tier2/worker/prompts.py 同款） +_cr_lock = threading.Lock() +_cr_registry_cache: dict | None = None +_cr_registry_mtime: float | None = None +_cr_registry_path_cache: str | None = None +_cr_file_cache: dict = {} # path_str → (mtime, body) +_cr_warned: set = set() # 防重复告警刷屏 + + +def _cr_warn_once(tag: str, msg: str) -> None: + """同一 tag 只告警一次（防 best-effort 路在循环里刷屏）；带 [cheap-roles] 前缀。""" + if tag in _cr_warned: + return + _cr_warned.add(tag) + print(f"[cheap-roles] {msg}", flush=True) + + +def _cr_prompts_dir() -> Path: + """当前生效的 contracts/prompts/ 目录：env CHEAP_PROMPTS_DIR > 默认仓内 contracts/prompts/。""" + p = os.environ.get("CHEAP_PROMPTS_DIR") + return Path(p) if p else _CR_DEFAULT_PROMPTS_DIR + + +def _cr_strip_inline_comment(value: str) -> str: + """剥 registry 值后的行内注释（# 后非引号内的内容），再剥成对外层引号。 + 与 tier2/worker/prompts.py _strip_inline_comment 同逻辑（无三方依赖）。 + """ + out: list = [] + in_quote: str | None = None + for ch in value: + if in_quote: + if ch == in_quote: + in_quote = None + out.append(ch) + elif ch in ("'", '"'): + in_quote = ch + out.append(ch) + elif ch == "#": + break # 注释起点（不在引号内） + else: + out.append(ch) + s = "".join(out).strip() + if len(s) >= 2 and s[0] == s[-1] and s[0] in ("'", '"'): + s = s[1:-1] + return s + + +def _cr_strip_frontmatter(text: str) -> str: + """剥 .md 文件的 YAML frontmatter（首行 --- 到下一个 ---），逐字节返回正文。 + + 字节保真（本相红线）：正文**不做 .strip()**——prompt 文本的首尾空白是正文的一部分， + 必须逐字节还原才能保证「registry 命中 == 内置 _SYSTEM_PROMPT」的不变量。 + 写文件时正文前固定垫一个空行（约定：闭合 --- 后紧随 \\n\\n），此处跳过那一个空行。 + 与 tier2/worker/prompts.py _strip_frontmatter 同逻辑。 + """ + lines = text.split("\n") + if not lines or lines[0].strip() != "---": + return text # 无 frontmatter：整篇即正文，逐字节返回 + close_idx = None + for idx in range(1, len(lines)): + if lines[idx].strip() == "---": + close_idx = idx + break + if close_idx is None: + return text # frontmatter 未闭合：退化为整篇当正文（best-effort） + rest = lines[close_idx + 1:] + # 跳过紧随闭合 --- 的第一个空行（写文件时固定垫的那个），其余逐字节保留 + if rest and rest[0] == "": + rest = rest[1:] + return "\n".join(rest) + + +def _cr_load_registry() -> dict: + """运行时读 registry.yaml，返回 {id: {file, version,...}}（mtime 缓存，best-effort 绝不抛）。""" + global _cr_registry_cache, _cr_registry_mtime, _cr_registry_path_cache + path = _cr_prompts_dir() / "registry.yaml" + path_str = str(path) + with _cr_lock: + try: + mtime = path.stat().st_mtime if path.exists() else None + except OSError: + mtime = None + + # 命中缓存（路径 + mtime 均未变）→ 直接返回 + if (_cr_registry_cache is not None + and _cr_registry_path_cache == path_str + and _cr_registry_mtime == mtime): + return _cr_registry_cache + + # registry 不存在：缓存空索引 + 一次性告警 + if mtime is None: + _cr_warn_once(f"reg-missing:{path_str}", + f"registry 不存在（{path_str}）→ cheap_roles 回落内置 _SYSTEM_PROMPT") + _cr_registry_cache, _cr_registry_mtime, _cr_registry_path_cache = {}, None, path_str + return _cr_registry_cache + + # 解析最小 YAML 子集（best-effort：任何异常落空索引 + 告警） + entries: dict = {} + try: + current_id: str | None = None + in_prompts = False + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.rstrip("\n") + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + if re.match(r"^prompts\s*:\s*(#.*)?$", line): + in_prompts = True + continue + if not in_prompts: + continue + m = re.match(r"^\s*-\s+id\s*:\s*(.+)$", line) + if m: + pid = _cr_strip_inline_comment(m.group(1)) + current_id = pid or None + if current_id: + entries[current_id] = {"id": current_id} + continue + m = re.match(r"^\s+([A-Za-z_][A-Za-z0-9_]*)\s*:\s*(.*)$", line) + if m and current_id is not None: + entries[current_id][m.group(1)] = _cr_strip_inline_comment(m.group(2)) + continue + # prompts 段内遇顶层新键（无缩进）→ 列表段结束 + if re.match(r"^[A-Za-z_]", line): + in_prompts = False + current_id = None + except Exception as exc: # noqa: BLE001 registry 解析失败：降级回落内置 + _cr_warn_once(f"reg-parse:{path_str}", + f"registry 解析失败（{path_str}: {exc}）→ cheap_roles 回落内置 _SYSTEM_PROMPT") + entries = {} + + _cr_registry_cache, _cr_registry_mtime, _cr_registry_path_cache = entries, mtime, path_str + return _cr_registry_cache + + +def _cr_read_body(rel_file: str) -> str | None: + """读 registry file 字段所指 .md，剥 frontmatter 返回正文（mtime 缓存，best-effort 绝不抛）。 + + 读不到 / 空文件 / 任何 IO 异常 → 返回 None（上层回落内置 _SYSTEM_PROMPT）。 + """ + path = (_cr_prompts_dir() / rel_file).resolve() + path_str = str(path) + with _cr_lock: + try: + mtime = path.stat().st_mtime if path.exists() else None + except OSError: + mtime = None + if mtime is None: + return None # 文件不存在：上层回落内置 + cached = _cr_file_cache.get(path_str) + if cached is not None and cached[0] == mtime: + return cached[1] + try: + body = _cr_strip_frontmatter(path.read_text(encoding="utf-8")) + except Exception as exc: # noqa: BLE001 读异常：降级回落内置 + _cr_warn_once(f"file-read:{path_str}", + f"prompt 文件读失败（{path_str}: {exc}）→ 回落内置 _SYSTEM_PROMPT") + return None + _cr_file_cache[path_str] = (mtime, body) + return body + + +def _load_system_prompt() -> str: + """运行时取 cheap_roles system prompt 原文（三级回落，best-effort，绝不抛）。 + + ① env CHEAP_PROMPTS_DIR + registry → .md 文件正文； + ② 仓内 contracts/prompts/ + registry → .md 文件正文（默认）； + ③ 内置 _SYSTEM_PROMPT（兜底；保证默认行为逐字节不变）。 + """ + registry = _cr_load_registry() + entry = registry.get(_CHEAP_PROMPT_ID) + if not entry: + _cr_warn_once(f"id-missing:{_CHEAP_PROMPT_ID}", + f"registry 未注册 {_CHEAP_PROMPT_ID} → 回落内置 _SYSTEM_PROMPT") + return _SYSTEM_PROMPT + rel_file = entry.get("file") + if not rel_file: + _cr_warn_once(f"file-field:{_CHEAP_PROMPT_ID}", + f"registry 条目 {_CHEAP_PROMPT_ID} 缺 file 字段 → 回落内置 _SYSTEM_PROMPT") + return _SYSTEM_PROMPT + body = _cr_read_body(rel_file) + if not body: + _cr_warn_once(f"body-empty:{_CHEAP_PROMPT_ID}", + f"prompt 文件缺失或正文空（{rel_file}）→ 回落内置 _SYSTEM_PROMPT（行为不变）") + return _SYSTEM_PROMPT + return body + + +def _cr_reload_cache() -> None: + """强制丢弃所有缓存，下次 _load_system_prompt 重读（测试 / 手动热更用）。""" + global _cr_registry_cache, _cr_registry_mtime, _cr_registry_path_cache, _cr_file_cache + with _cr_lock: + _cr_registry_cache, _cr_registry_mtime, _cr_registry_path_cache = None, None, None + _cr_file_cache = {} + + def build_system_prompt(game_id: str, scaffold_desc: str = None) -> str: """据 game_id 拼出本 run 的 system prompt（把 ⟦SCAFFOLD_DESC⟧/⟦G⟧ 占位替换）。 @@ -79,8 +324,11 @@ def build_system_prompt(game_id: str, scaffold_desc: str = None) -> str: Returns: 薄 system prompt 字符串（指 agent read 真 skill，不手抄）。 """ + # C2a：运行时热取 prompt 文本（registry → .md → 内置 _SYSTEM_PROMPT 三级回落）； + # 任何读失败都回落 _SYSTEM_PROMPT，保证输出与改造前逐字节一致（本相红线）。 + base = _load_system_prompt() g = f"game-runtime/games/amgen-{game_id}" - return _SYSTEM_PROMPT.replace("⟦SCAFFOLD_DESC⟧", scaffold_desc or _DEFAULT_SCAFFOLD_DESC).replace("⟦G⟧", g) + return base.replace("⟦SCAFFOLD_DESC⟧", scaffold_desc or _DEFAULT_SCAFFOLD_DESC).replace("⟦G⟧", g) def _contract_block() -> str: diff --git a/cheap-worker/cheap_studio.py b/cheap-worker/cheap_studio.py index 9c668070..83ecc054 100644 --- a/cheap-worker/cheap_studio.py +++ b/cheap-worker/cheap_studio.py @@ -23,12 +23,22 @@ import _bootstrap # noqa: E402,F401 from cheap_roles import build_system_prompt # noqa: E402 from cheap_toolkit import CheapSession, build_toolkit # noqa: E402 import cheap_run # noqa: E402 +import cheap_verify # noqa: E402 U-A2：便宜档「丰富度」LLM 验证 agent（非阻塞·只报告） # tier2 框架层（import config 触发代理旁路 + 加载 agentscope，必须在裸 import agentscope/openai 之前）。 from worker import config # noqa: E402 from worker.middleware import ( # noqa: E402 CircuitBreakerMiddleware, Tier2TraceMiddleware, Tier2CircuitBreak, ) +# C1b trace 落库：复用 tier2 observability 的 JSONL sink（cheap-worker 不另造 sink）。 +# 非阻塞铁律：导入失败只告警，回落到无 sink 模式，绝不阻断生成主链。 +try: + from observability.trace import make_jsonl_sink as _make_jsonl_sink # noqa: E402 + _TRACE_SINK_AVAILABLE = True +except Exception as _import_err: + _TRACE_SINK_AVAILABLE = False + print(f"[cheap_studio] trace sink import 失败（回落无 sink 模式，不影响生成）：{_import_err}", + flush=True) # AgentScope（此时 agentscope 已由 config import 链加载、代理旁路已装）。 from agentscope.agent import Agent, ReActConfig # noqa: E402 from agentscope.message import UserMsg # noqa: E402 @@ -111,7 +121,7 @@ def build_trace_source(verdict, driver_type, attempts, stage, model) -> dict: async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens=16000, port=4320, cdp_port=9222, run_gates=True, system_prompt=None, initial_kick=None, write_whitelist=None, prepare=None, - scaffold_template=None, scaffold_desc=None): + scaffold_template=None, scaffold_desc=None, verify_richness_enabled=True): """跑便宜档一局生成（scaffold → ReAct 写 src/ → done 门 → 收口 stage+smoke+九门）。返回 run-summary dict。 run_gates=True（默认，spike 单局）：收口跑全套 stage→smoke→ensure_play_spec→九门 play。 @@ -144,7 +154,19 @@ async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens= # on_model_call 在每次裸模型调用前按「已花 + 本次预估」判、越线即 fail-closed 抛 Tier2CircuitBreak(budget)； # 取价不可达时 _ensure_pricing 降级为次数闸（既有行为，不静默超支、不静默阻断）。步数/超时仍用 tier2 默认。 breaker = CircuitBreakerMiddleware(enable_rmb_gate=True, rmb_hard_limit=10.0) - tracer = Tier2TraceMiddleware(trace_id=game_id) + # C1b trace 落库：把生成轨迹按五字段（traceId/step/cost/verdict/timestamp + ext） + # 追加写到产物目录 games/amgen-/trace.jsonl，供成本对账 / replay 用。 + # 非阻塞铁律：接线异常只告警 + 回落无 sink 模式，绝不阻断生成主链。 + # 为什么落盘路径选产物目录：与 run-summary.json / evidence/ 同目录，随产物一起留存，运维按 gameId 定位直观。 + _trace_sink = None + if _TRACE_SINK_AVAILABLE: + try: + _trace_path = cheap_run.game_dir(game_id) / "trace.jsonl" + _trace_sink = _make_jsonl_sink(_trace_path) + _rec(f"trace sink 接线 → {_trace_path}") + except Exception as _sink_err: + _rec(f"trace sink 构造异常（只告警，回落无 sink，不阻断生成）：{_sink_err}") + tracer = Tier2TraceMiddleware(trace_id=game_id, sink=_trace_sink) writer = Agent( name="cheap-writer", @@ -244,6 +266,8 @@ async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens= "tokens": {"in": tok_in, "out": tok_out, "total": tok_in + tok_out}, "doneSummary": session.finished.get("summary") if session.finished else None, "wallSec": round(time.time() - t0, 1), + # C1b trace 收口摘要：{traceId, steps, dropped}，供编排器对账（trace.jsonl 落盘路径已在接线处 _rec）。 + "trace": tracer.summary(), } # ── M3b U1：additive 富化 9d trace 源维度（verdictFull/driverType/models/stage）── # 收口段已算出但没存的维度结构化进 summary，供 result_out 组 payload["trace"]、后端 D11 算分； @@ -251,6 +275,24 @@ async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens= stage = _furthest_stage(finished=finished, staged=staged, smoke_ran=(smoke_ok is not None), played=(verdict is not None)) summary.update(build_trace_source(verdict, driver_type, attempts, stage, _bootstrap.SPIKE_MODEL)) + + # ── U-A2：便宜档「丰富度」LLM 评分（非阻塞·只报告·不进 verdict / 不改 ok 达标）── + # 架构红线：玩法「丰富不丰富」的校验需大模型能力，绝不写成 code-presence/正则/断言（违反创始人红线）。 + # 故纯走 LLM judge（cheap_verify.verify_richness），结果 additive 写 summary["richness"]： + # · 不进 verdict、不改 summary["ok"]/达标判定、不动任何既有键（result_out._build_trace 只读 named 字段，不读 richness）； + # · judge 用独立 model 实例，token 不污染生成成本台账（costRmb 仍只算生成）； + # · 非阻塞双保险：verify_richness 内部已 try/except 兜底 degraded，这里再包一层，verify 失败绝不影响 run。 + # 只在 run_gates（完整 spike 收口）且 finished（有真产物）时跑；对照路（run_gates=False）零改动、不触发 LLM。 + if verify_richness_enabled and run_gates and finished: + try: + summary["richness"] = await cheap_verify.verify_richness(game_id, brief=brief) + rv = summary["richness"] + _rec(f"丰富度 LLM 评分 score={rv.get('score')}/{rv.get('max')} degraded={rv.get('degraded')}") + except Exception as e: # noqa: BLE001 非阻塞铁律：richness 绝不影响生成主链 + summary["richness"] = {"score": None, "degraded": True, + "reason": f"richness 接线异常：{type(e).__name__}: {e}"} + _rec(f"丰富度评分接线异常（已降级，不影响 run）：{type(e).__name__}: {e}") + ev_dir = cheap_run.game_dir(game_id) / "evidence" ev_dir.mkdir(parents=True, exist_ok=True) (ev_dir / "run-summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") diff --git a/cheap-worker/cheap_verify.py b/cheap-worker/cheap_verify.py new file mode 100644 index 00000000..33488fbc --- /dev/null +++ b/cheap-worker/cheap_verify.py @@ -0,0 +1,241 @@ +""" +cheap_verify.py — 便宜档「丰富度」LLM 验证 agent（非阻塞·只报告）。 + +架构红线（创始人多轮强调，绝不可破）：**项目代码只做机械确定的事**；玩法/美术/音乐的「丰富」—— +无论生成还是校验——都需大模型能力，**绝不写成代码校验**（code-presence/静态扫描/正则/断言一律禁）。 +故本模块只做两件 100% 机械确定的事，「丰富不丰富」的判断本身完全交 LLM judge： + ① 读产物 L3 源码（确定性文件 I/O）+ 拼 judge prompt（确定性字符串）； + ② 解析 LLM 返回的 JSON 成结构化评分（确定性解析 + degraded 兜底）。 +本模块**不含任何丰富度的代码断言**——8 条好玩清单只是【喂给 LLM 的评分尺】，命中与否全由 judge 读码后裁。 +（注：sim-business-game-design.md §10 写了「4 条可机检 code-presence」子集，那与本红线冲突——本模块刻意不照它写代码校验。） + +非阻塞铁律：LLM 调失败/超时/解析失败 → 返回 {score:None, degraded:True, reason:...}， +**绝不抛、绝不阻断生成、绝不进 verdict / 不改达标判定**。接线方（cheap_studio）再包一层 try/except 双保险。 + +被 cheap_studio.run_studio 收口段调用（九门 play 之后、写 run-summary 之前），结果 additive 写 summary["richness"]。 +""" + +import asyncio + +import cheap_run # 仅 stdlib 依赖（读产物 src 文件）；不 import agentscope/_bootstrap，故纯函数可在无 agentscope 环境单测。 + +# ── sim-business §10 反「无趣」8 条好玩清单（judge 的评分 rubric）── +# 这是【喂给 LLM judge 的打分尺】，不是代码校验：丰富不丰富全由 judge 读源码后逐条裁。 +# 与 .agents/skills/sim-business-game-design.md §10 同源（那边是 design 自检，这里是产物 judge 评分）。 +# 每项 = (标准名, 含义)；judge 按标准名逐条回命中与否 + 理由。 +RICHNESS_CHECKLIST = [ + ("即时反馈", "每个主要操作有「飘字 +N + 音效 + 粒子」即时回报"), + ("可见成长", "有数字/规模肉眼可见地变大（分数飙升/店铺扩张/等级上涨）"), + ("下一个解锁", "任意时刻玩家眼前都有「再攒一点就解锁 X」的钩子"), + ("30秒爽点", "开局 30 秒内有第一次升级/解锁"), + ("数值滚雪球", "成长有「越来越快」的暴富段，而非平淡线性"), + ("情感锚", "有萌角色/拥有物让玩家「想养大它」"), + ("放置回归", "离线回来有惊喜（离线收益弹窗等）"), + ("音反馈", "收益/升级/解锁有声音（哪怕程序化）"), +] + +_MAX = len(RICHNESS_CHECKLIST) # = 8 + +# 只读 L3 游戏本体（玩法/画面/数值/资产）；跳过 L1 plumbing（host-config/game.js/index.html—— +# 固定脚手架、与「游戏丰富不丰富」无关，喂给 judge 只会稀释信号）。 +_L3_SOURCE_FILES = ("game-logic.js", "core.js", "render.js", "balance.js", "assets.js") +_MAX_SRC_BYTES = 24000 # 喂 judge 的源码总量上限（控 token / 成本；超限按字节截断）。 +_JUDGE_MAX_TOKENS = 3000 # judge 输出小（8 条短理由 + 一句点评），3000 足够、M3 无 thinking。 + +# judge 的 system / user 框定——只评分、不改码、要看源码真实现、输出严格 JSON。 +_JUDGE_SYSTEM = ( + "你是轻量小游戏的【丰富度评审 agent】。任务 = 读一款便宜档 LittleJS 小游戏的源码，" + "只评判它「作为一款游戏够不够丰富、好玩」，逐条给出 8 条好玩清单的命中与否 + 一句中文理由，" + "最后输出严格 JSON。你只评分、不修改代码、不阻断发布——这是非阻塞的质量信号。" + "评判要看源码里**真实实现**的玩法、数值成长、解锁阶梯、即时反馈与音效，别被空壳或注释骗。" +) + + +def _degraded(reason: str) -> dict: + """降级结果（非阻塞铁律）：score=None + degraded=True；score 为 None 表示「这次没评出来」，绝不参与达标。""" + return {"score": None, "max": _MAX, "hits": [], "notes": None, "degraded": True, "reason": reason} + + +def _coerce_bool(v) -> bool: + """把 judge 可能给的多形态命中值归一为 bool（true/1/"是"/"命中" 等 → True；其余 → False）。""" + if isinstance(v, bool): + return v + if isinstance(v, (int, float)): + return v > 0 + if isinstance(v, str): + return v.strip().lower() in ("true", "1", "yes", "y", "是", "命中", "hit", "对", "✓", "√") + return False + + +def parse_judge_output(text, *, checklist=RICHNESS_CHECKLIST) -> dict: + """纯函数：把 judge 的 LLM 文本输出解析成结构化丰富度评分（含 degraded 兜底）。可单测、零网络、零 agentscope。 + + 成功 → {score:int(命中数 0..8), max:8, hits:[{name,hit,why}*8], notes:str, degraded:False} + 解析失败（空/非 JSON 对象/缺 checks 数组）→ _degraded(...)（score=None, degraded=True）。 + + 对齐策略：优先按 name 对齐 judge 的 checks 到 8 条标准项；name 对不上的按位置兜底（容忍模型改名/英文名）。 + 本函数**不含任何丰富度的代码判断**——只搬运 judge 的逐条裁决并计命中数。 + """ + if not isinstance(text, str) or not text.strip(): + return _degraded("judge 输出为空") + try: + # json_repair 容忍 markdown 围栏 / 前后赘语 / 尾逗号等 LLM 常见脏输出。 + import json_repair + data = json_repair.loads(text) + except Exception as e: # noqa: BLE001 解析层任何异常都降级，绝不抛 + return _degraded(f"judge 输出解析失败：{type(e).__name__}") + if not isinstance(data, dict): + return _degraded("judge 输出不是 JSON 对象") + checks_raw = data.get("checks") + if not isinstance(checks_raw, list) or not checks_raw: + return _degraded("judge 输出缺 checks 数组") + + by_name = {} + for c in checks_raw: + if isinstance(c, dict): + nm = str(c.get("name", "")).strip() + if nm: + by_name[nm] = c + + hits = [] + score = 0 + for i, (name, _meaning) in enumerate(checklist): + c = by_name.get(name) + if c is None and i < len(checks_raw) and isinstance(checks_raw[i], dict): + c = checks_raw[i] # name 对不上 → 位置兜底 + hit = _coerce_bool(c.get("hit")) if isinstance(c, dict) else False + why = (str(c.get("why", "")).strip()[:200]) if isinstance(c, dict) else "" + hits.append({"name": name, "hit": hit, "why": why}) + if hit: + score += 1 + notes = str(data.get("notes", "")).strip()[:500] + return {"score": score, "max": len(checklist), "hits": hits, "notes": notes, "degraded": False} + + +def _collect_sources(game_id: str, *, max_bytes: int = _MAX_SRC_BYTES) -> str: + """读产物 L3 源码（game-logic/core/render/balance/assets）拼成带文件头的一段文本，总量截断到 max_bytes。 + + 只读 L3 游戏本体；读不到的文件跳过；全空 → 返回空串（上层据此降级、跳过 LLM、不白烧钱）。 + """ + src_dir = cheap_run.game_dir(game_id) / "src" + parts = [] + total = 0 + for name in _L3_SOURCE_FILES: + p = src_dir / name + try: + if not p.is_file(): + continue + txt = p.read_text(encoding="utf-8", errors="ignore") + except OSError: + continue # 单文件读失败不致命，跳过即可 + chunk = f"\n// ===== {name} =====\n{txt}\n" + parts.append(chunk) + total += len(chunk.encode("utf-8")) + if total >= max_bytes: + break + joined = "".join(parts) + enc = joined.encode("utf-8") + if len(enc) > max_bytes: # 字节级截断兜底（防单个大文件超限撑爆 token） + joined = enc[:max_bytes].decode("utf-8", "ignore") + "\n// [源码已截断]" + return joined + + +def _build_judge_user(src_text: str, brief: str = "") -> str: + """拼 judge 的 user 消息：brief（可选）+ 8 条清单 + 产物源码 + 严格 JSON 输出契约（纯确定性字符串拼接）。""" + checklist_lines = "\n".join( + f"{i + 1}. {name}：{meaning}" for i, (name, meaning) in enumerate(RICHNESS_CHECKLIST) + ) + brief_block = f"这款游戏的 brief（玩家想要的）：{brief}\n\n" if brief else "" + return ( + f"{brief_block}" + "下面是这款便宜档小游戏的 L3 源码。请逐条评判它是否命中这 8 条好玩清单，" + "据**源码里真实实现了的玩法/数值/音效/解锁**判断（别被注释或空壳骗：比如只 import 了 audioMusic " + "但收益处没真调 playSfx，则「音反馈」不算命中）：\n\n" + f"{checklist_lines}\n\n" + "=== 源码开始 ===\n" + f"{src_text}\n" + "=== 源码结束 ===\n\n" + "严格只输出以下 JSON（不要任何额外文字、不要 markdown 围栏）：\n" + '{"checks":[{"name":"即时反馈","hit":true,"why":"一句中文理由"}, … 共 8 条，' + 'name 用上面 8 条的中文名、hit 为 true/false], "notes":"整体一句话点评"}' + ) + + +def _extract_text(resp) -> str: + """从 ChatResponse 抽纯文本：content 是 TextBlock/ToolCall... 序列，只取 text 块（兼容 pydantic 块与 dict 块）。""" + content = getattr(resp, "content", None) + if content is None and isinstance(resp, dict): + content = resp.get("content") + if isinstance(content, str): + return content + parts = [] + for b in (content or []): + if isinstance(b, dict): + if b.get("type") == "text" and isinstance(b.get("text"), str): + parts.append(b["text"]) + elif getattr(b, "type", None) == "text": + t = getattr(b, "text", None) + if isinstance(t, str): + parts.append(t) + return "".join(parts) + + +async def verify_richness(game_id: str, *, brief: str = "", model=None, sources=None, + timeout: float = 90.0, max_src_bytes: int = _MAX_SRC_BYTES) -> dict: + """生成完成后跑一次 LLM 丰富度评分（非阻塞·只报告）。**绝不抛、绝不阻断、绝不进 verdict / 不改达标**。 + + Args: + game_id: 本 run 游戏 id（读 games/amgen-/src/ 下 L3 源码）。 + brief: 本局 brief（喂 judge 做上下文，提升「情感锚/解锁」等判断质量）；缺省空串。 + model: 可选注入的 LLM 客户端（单测用 fake model 注入、零网络）；None → 用 _bootstrap.build_cheap_model + 新建**独立** M3 实例（judge token 不污染生成成本台账 costRmb——richness 是非阻塞 add-on）。 + sources: 可选直接注入的源码文本（单测用，绕过文件 I/O）；None → 从 game_id 收集。 + timeout: judge LLM 调用超时秒数；超时 → degraded（绝不卡死收口）。 + max_src_bytes: 喂 judge 的源码上限。 + + Returns: + 成功 → parse_judge_output 的结构化评分（含 judgeTokens 观测，best-effort）； + 任何失败（src 空/LLM 异常/超时/解析失败）→ _degraded(...)（score=None, degraded=True, reason）。 + """ + try: + src_text = sources if sources is not None else _collect_sources(game_id, max_bytes=max_src_bytes) + if not src_text or not src_text.strip(): + return _degraded("产物 src 为空/读不到，跳过 LLM 评分") + + # 惰性 import：agentscope / _bootstrap 较重，且让上面的 parse_judge_output 等纯函数能在无 agentscope 环境单测。 + from agentscope.message import SystemMsg, UserMsg + m = model + if m is None: + import _bootstrap # 代理旁路 + key 注入由 _bootstrap.build_cheap_model 内部处理 + m = _bootstrap.build_cheap_model(max_tokens=_JUDGE_MAX_TOKENS) + + sys_msg = SystemMsg(name="system", content=_JUDGE_SYSTEM) + user_msg = UserMsg(name="user", content=_build_judge_user(src_text, brief)) + # 硬超时包裹真模型调用：judge 卡住绝不能拖死收口（非阻塞铁律）。 + resp = await asyncio.wait_for(m([sys_msg, user_msg]), timeout=timeout) + result = parse_judge_output(_extract_text(resp)) + + # best-effort 附 judge token（外部 LLM 调用可观测；与生成成本台账隔离——judge 用独立 model 实例）。 + try: + ti, to = m.usage_sum() + result["judgeTokens"] = {"in": ti, "out": to, "total": ti + to} + except Exception: # noqa: BLE001 观测字段失败不影响评分主体 + pass + return result + except (asyncio.TimeoutError, TimeoutError): + return _degraded(f"judge LLM 超时（>{timeout}s）") + except Exception as e: # noqa: BLE001 非阻塞铁律：任何异常（网络/import/解析）都降级，绝不抛断生成主链 + return _degraded(f"judge 异常：{type(e).__name__}: {e}") + + +if __name__ == "__main__": + # 便捷自跑：python cheap-worker/cheap_verify.py ["brief"] —— 对已有产物真跑一次评分。 + import json + import sys + from pathlib import Path + + sys.path.insert(0, str(Path(__file__).resolve().parent)) + gid = sys.argv[1] if len(sys.argv) > 1 else "cheap-smoke1" + bf = sys.argv[2] if len(sys.argv) > 2 else "" + out = asyncio.run(verify_richness(gid, brief=bf)) + print(json.dumps(out, ensure_ascii=False, indent=2)) diff --git a/cheap-worker/tests/test_bake_off.py b/cheap-worker/tests/test_bake_off.py index 0230319f..ceed95d8 100644 --- a/cheap-worker/tests/test_bake_off.py +++ b/cheap-worker/tests/test_bake_off.py @@ -110,6 +110,48 @@ def test_aggregate_unconverged_total_surfaced(): assert r["overallMeets"] is True and r["unconvergedTotal"] == 3 and r["sampleTotal"] == 18 +# ───────────────────────── 丰富度分布(U-B1 · additive · 与达标正交)───────────────────────── + +def _pr(n, score): + """n 款过九门 + richness 评分 score(非降级)。""" + return [{"passed": True, "finished": True, + "richness": {"score": score, "max": 8, "degraded": False}} for _ in range(n)] + + +def _pd(n): + """n 款过九门 + richness 降级(LLM 评分失败/超时,score=None)。""" + return [{"passed": True, "finished": True, + "richness": {"score": None, "max": 8, "degraded": True}} for _ in range(n)] + + +def test_richness_dist_basic(): + """丰富度分布:非降级款收集 score 算均分/分布,降级款单列计数(纯报告)。""" + d = B.richness_dist(_pr(2, 6) + _pr(1, 8) + _pd(1)) + assert d["scoredCount"] == 3 and d["degradedCount"] == 1 + assert d["meanScore"] == round((6 + 6 + 8) / 3, 2) and d["scores"] == [6, 6, 8] and d["max"] == 8 + + +def test_richness_dist_all_degraded_or_missing(): + """全降级 / 无 richness 字段(对照路/老产物)→ meanScore=None、scoredCount=0,不报错。""" + d = B.richness_dist(_pd(2) + _p(1)) # _p 无 richness 字段 + assert d["scoredCount"] == 0 and d["meanScore"] is None and d["degradedCount"] == 2 + + +def test_aggregate_richness_additive_does_not_change_meets(): + """红线:richness 是 additive 报告 —— 加 richness 后 overallMeets / perGenre 达标判定与无 richness 时完全一致。""" + bare = {"a": _p(4) + _f(1)} # 无 richness + rich = {"a": _pr(4, 7) + _f(1)} # 4 过门款带 richness=7、1 质量挂 + rb = B.aggregate(bare, ["a"]) + rr = B.aggregate(rich, ["a"]) + # 达标判定完全不受 richness 影响(判据只看 verdict.pass,与 richness 正交) + assert rr["overallMeets"] == rb["overallMeets"] is True + assert rr["perGenre"]["a"]["passRate"] == rb["perGenre"]["a"]["passRate"] == 0.8 + assert rr["perGenre"]["a"]["meets"] == rb["perGenre"]["a"]["meets"] is True + # richness 分布 additive 出现,不混进达标判据 + assert rr["richnessByGenre"]["a"]["meanScore"] == 7.0 + assert rb["richnessByGenre"]["a"]["scoredCount"] == 0 # 无 richness 款 → 空分布,不报错 + + # ───────────────────────── 并发上限 / 零 LLM ───────────────────────── def test_clamp_conc_upper_cap(): diff --git a/cheap-worker/tests/test_cheap_trace.py b/cheap-worker/tests/test_cheap_trace.py new file mode 100644 index 00000000..41be1561 --- /dev/null +++ b/cheap-worker/tests/test_cheap_trace.py @@ -0,0 +1,172 @@ +"""test_cheap_trace.py — C1b cheap-worker trace 落盘单测（TDD：先红后绿）。 + +验证 cheap-worker 生成轨迹落盘： + 1. make_jsonl_sink + TraceAdapter → emit 事件 → trace.jsonl 落盘； + 2. 每行是合法 JSON； + 3. 五字段（traceId / step / cost / verdict / timestamp）齐全； + 4. sink 写失败 → 只告警不抛（非阻塞铁律）； + 5. cheap_studio.run_studio 通过 Tier2TraceMiddleware(sink=...) 接线后， + 产物目录 games/amgen-/trace.jsonl 存在（可选 e2e，此处为集成桩）。 + +复用 tier2 observability/trace.py 的 JsonlFileSink / make_jsonl_sink / TraceAdapter， +不另造 sink（最小改动原则；cheap-worker 已通过 _bootstrap 把 tier2/gen-worker 加进 sys.path）。 + +跑法（无 agentscope 依赖，系统 Python 即可）： + cheap-worker/.venv/bin/python cheap-worker/tests/test_cheap_trace.py +""" + +import sys +import os +import json +import tempfile +from pathlib import Path + +# sys.path：把 cheap-worker/ 和 tier2/gen-worker/ 加进去 +# （tier2/gen-worker 含 observability 包，cheap-worker 含 cheap_studio 等） +_HERE = os.path.dirname(os.path.abspath(__file__)) +_CW = os.path.dirname(_HERE) # cheap-worker/ +_REPO_ROOT = os.path.dirname(_CW) # repo root +_GEN_WORKER = os.path.join(_REPO_ROOT, "tier2", "gen-worker") + +for _p in (_CW, _GEN_WORKER): + if _p not in sys.path: + sys.path.insert(0, _p) + +# C1b 复用 tier2 trace 基础件（cheap-worker 不另造 sink） +from observability.trace import ( # noqa: E402 + JsonlFileSink, make_jsonl_sink, TraceAdapter, +) + + +# ── 测试一：核心落盘 + 五字段完整性（happy path）──────────────────────────────── +def test_cheap_trace_creates_file_and_five_fields(): + """emit 三条桩事件 → trace.jsonl 落盘；每行合法 JSON；五字段齐；内容正确。""" + with tempfile.TemporaryDirectory() as tmpdir: + out = Path(tmpdir) / "trace.jsonl" + adapter = TraceAdapter( + trace_id="c1btrace-test", + sink=make_jsonl_sink(out), + ) + + # 便宜档生成中会产生的典型事件形态（dict 形式，无需真 AgentScope 环境） + fake_events = [ + # ReplyStartEvent：一轮 ReAct 回复开始 + {"type": "ReplyStartEvent", "created_at": "2026-06-30T10:00:00", + "reply_id": "r1", "name": "cheap-writer"}, + # ModelCallEndEvent：模型调用结束，带 token 用量 + {"type": "ModelCallEndEvent", "created_at": "2026-06-30T10:00:01", + "input_tokens": 200, "output_tokens": 80}, + # ToolResultEndEvent：工具调用结果，带 verdict + {"type": "ToolResultEndEvent", "created_at": "2026-06-30T10:00:02", + "state": "success"}, + ] + for e in fake_events: + adapter.ingest(e) + + # ① 文件已落盘 + assert out.exists(), f"trace.jsonl 应已落盘，但不存在：{out}" + + lines = out.read_text(encoding="utf-8").strip().split("\n") + # ② 三条事件对应三行 + assert len(lines) == 3, f"应有 3 行，实际 {len(lines)} 行：\n" + "\n".join(lines) + + # ③ 每行合法 JSON + 五字段齐全 + CORE_FIELDS = {"traceId", "step", "cost", "verdict", "timestamp"} + for i, line in enumerate(lines): + obj = json.loads(line) # 若非合法 JSON 此处抛 JSONDecodeError + missing = CORE_FIELDS - set(obj.keys()) + assert not missing, f"第 {i} 行缺五字段 {missing}：{line}" + assert obj["traceId"] == "c1btrace-test", ( + f"第 {i} 行 traceId 错（期望 c1btrace-test，实际 {obj['traceId']}）") + assert obj["step"] == i, ( + f"第 {i} 行 step 错（期望 {i}，实际 {obj['step']}）") + + # ④ ModelCallEnd 步 cost.tokens 正确 + second = json.loads(lines[1]) + assert second["cost"] is not None, "ModelCallEnd 步 cost 应非空" + assert second["cost"]["tokens"]["in"] == 200, "input_tokens 应为 200" + assert second["cost"]["tokens"]["out"] == 80, "output_tokens 应为 80" + + # ⑤ ToolResultEnd 步 verdict=success + third = json.loads(lines[2]) + assert third["verdict"] == "success", ( + f"ToolResultEnd 步 verdict 应为 success，实际 {third['verdict']}") + + print(f" [ok] traceId={json.loads(lines[0])['traceId']}, {len(lines)} 行, 五字段齐") + + +# ── 测试二：嵌套目录自动创建 ───────────────────────────────────────────────────── +def test_cheap_trace_auto_creates_dir(): + """游戏目录若不存在，sink 应自动创建并写入（amgen-/ 由 scaffold 建，但 best-effort 时序容错）。""" + with tempfile.TemporaryDirectory() as tmpdir: + out = Path(tmpdir) / "subdir" / "amgen-c1b" / "trace.jsonl" + sink = make_jsonl_sink(out) + step = { + "traceId": "c1b-auto", "step": 0, "cost": None, + "verdict": None, "timestamp": "2026-06-30T00:00:00", + "ext": {"raw": {"event": "ReplyStartEvent"}}, + } + sink(step) + assert out.exists(), f"嵌套目录下 trace.jsonl 应自动创建：{out}" + obj = json.loads(out.read_text(encoding="utf-8").strip()) + assert obj["traceId"] == "c1b-auto" + print(f" [ok] auto-mkdir: {out}") + + +# ── 测试三：写失败只告警不抛（非阻塞铁律）──────────────────────────────────────── +def test_cheap_trace_write_fail_no_raise(): + """sink 写失败（不可写路径）→ 只 print 告警，绝不抛，不阻断调用方（非阻塞铁律）。""" + bad = Path("/proc/nonexistent_cheap_trace_test/trace.jsonl") + sink = JsonlFileSink(bad) + try: + sink({"traceId": "t-fail", "step": 0, "cost": None, "verdict": None, + "timestamp": "", "ext": {"raw": {"event": "X"}}}) + print(" [ok] 写失败只告警不抛") + except Exception as e: + raise AssertionError(f"sink 写失败时不应抛异常，但抛了：{e}") from e + + +# ── 测试四：sink 内部抛 → adapter.ingest best-effort 不传播（非阻塞）────────────── +def test_cheap_trace_adapter_sink_fail_no_raise(): + """adapter 内层 sink 崩溃 → adapter.ingest 只记 dropped，不向外抛（non-blocking 铁律）。""" + def _bad_sink(step): + raise IOError("模拟磁盘满写失败") + + adapter = TraceAdapter(trace_id="t-sink-fail", sink=_bad_sink) + # 正常 emit 一条（sink 内部崩溃） + result = adapter.ingest( + {"type": "ReplyStartEvent", "created_at": "2026-06-30T10:00:00", "reply_id": "r1"} + ) + # ingest 应返回 TraceStep（映射成功），只是 sink 写失败记 dropped + assert result is not None or adapter.dropped > 0, ( + "sink 崩溃时要么返回 step（sink 失败只计 dropped），要么 dropped>0") + print(f" [ok] sink 崩溃 → dropped={adapter.dropped}，adapter.ingest 不抛") + + +# ── 入口 ──────────────────────────────────────────────────────────────────────── +if __name__ == "__main__": + tests = [ + ("test_cheap_trace_creates_file_and_five_fields", + test_cheap_trace_creates_file_and_five_fields), + ("test_cheap_trace_auto_creates_dir", + test_cheap_trace_auto_creates_dir), + ("test_cheap_trace_write_fail_no_raise", + test_cheap_trace_write_fail_no_raise), + ("test_cheap_trace_adapter_sink_fail_no_raise", + test_cheap_trace_adapter_sink_fail_no_raise), + ] + passed = failed = 0 + for name, fn in tests: + try: + print(f"[RUN] {name}") + fn() + print(f"[PASS] {name}") + passed += 1 + except Exception as exc: + import traceback + print(f"[FAIL] {name}: {exc}") + traceback.print_exc() + failed += 1 + print(f"\n结果：{passed} 通过 / {failed} 失败") + import sys as _sys + _sys.exit(0 if failed == 0 else 1) diff --git a/cheap-worker/tests/test_cheap_verify.py b/cheap-worker/tests/test_cheap_verify.py new file mode 100644 index 00000000..0112ee70 --- /dev/null +++ b/cheap-worker/tests/test_cheap_verify.py @@ -0,0 +1,165 @@ +""" +test_cheap_verify.py — U-A2 便宜档「丰富度」LLM 验证 agent 单测。 + +两层： + ① parse_judge_output 纯函数（解析 LLM 输出 → 结构化评分 + degraded 兜底）——无 agentscope、零网络， + plain python 即可跑：cheap-worker/.venv/bin/python cheap-worker/tests/test_cheap_verify.py + ② verify_richness 非阻塞契约（fake model 注入，零真网络）——验证「LLM 失败/超时/解析失败/空 src → degraded、 + 绝不抛、绝不阻断」。需 agentscope.message（venv 有），缺则自动跳过。 + +跑：cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_cheap_verify.py -q +""" + +import asyncio +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/ +import cheap_verify # noqa: E402 + +try: + import agentscope.message # noqa: E402,F401 + _HAS_AS = True +except Exception: # noqa: BLE001 无 agentscope 时 ② 类测试跳过（纯函数 ① 不受影响） + _HAS_AS = False + + +def _full_checks(hits): + """据布尔列表造 8 条 checks（name 用 cheap_verify 的标准名）。""" + return [{"name": n, "hit": h, "why": "理由"} for (n, _m), h in zip(cheap_verify.RICHNESS_CHECKLIST, hits)] + + +# ───────────────────────── ① parse_judge_output 纯函数（无 agentscope/无网络）───────────────────────── + +def test_parse_valid_full(): + """合法 8 条 JSON → 命中数正确、非 degraded、hits/notes 落位。""" + text = json.dumps({"checks": _full_checks([True, True, False, True, False, False, False, True]), + "notes": "还行"}, ensure_ascii=False) + r = cheap_verify.parse_judge_output(text) + assert r["degraded"] is False + assert r["score"] == 4 and r["max"] == 8 + assert len(r["hits"]) == 8 and r["hits"][0]["name"] == "即时反馈" + assert r["notes"] == "还行" + + +def test_parse_fenced_json(): + """markdown 围栏 + 前后赘语 → json_repair 兜得住、正确解析。""" + body = json.dumps({"checks": _full_checks([True] * 8), "notes": "满"}, ensure_ascii=False) + text = "这是我的评分：\n```json\n" + body + "\n```\n以上。" + r = cheap_verify.parse_judge_output(text) + assert r["degraded"] is False and r["score"] == 8 + + +def test_parse_garbage_degraded(): + """完全非 JSON → degraded（score=None），不抛。""" + r = cheap_verify.parse_judge_output("完全不是 JSON 的一段话") + assert r["degraded"] is True and r["score"] is None and r["max"] == 8 + + +def test_parse_none_and_empty_degraded(): + """None / 空串 → degraded。""" + assert cheap_verify.parse_judge_output(None)["degraded"] is True + assert cheap_verify.parse_judge_output("")["degraded"] is True + assert cheap_verify.parse_judge_output(" ")["degraded"] is True + + +def test_parse_non_dict_degraded(): + """JSON 数组（非对象）→ degraded。""" + assert cheap_verify.parse_judge_output("[1,2,3]")["degraded"] is True + + +def test_parse_missing_checks_degraded(): + """有 JSON 对象但缺 checks 数组 → degraded（不静默判 0 分）。""" + assert cheap_verify.parse_judge_output(json.dumps({"notes": "x"}))["degraded"] is True + assert cheap_verify.parse_judge_output(json.dumps({"checks": []}))["degraded"] is True + + +def test_parse_coerce_and_positional(): + """hit 用字符串/数字 + name 对不上时按位置兜底（容忍模型改名）。""" + checks = [{"hit": "true", "why": "a"}, {"hit": 1}, {"hit": "否"}] + [{"hit": False}] * 5 + r = cheap_verify.parse_judge_output(json.dumps({"checks": checks})) + assert r["degraded"] is False + assert r["score"] == 2 # 前两条 "true"/1 命中，第三条「否」不命中，其余 False + + +# ───────────────────────── ② verify_richness 非阻塞契约（fake model，零真网络）───────────────────────── + +class _FakeResp: + """mock ChatResponse：content 用 dict 文本块（_extract_text 兼容 dict 块）。""" + def __init__(self, text): + self.content = [{"type": "text", "text": text}] + + +class _FakeModel: + """async 可调用 mock LLM：返回固定文本 / 抛异常 / 睡眠（测超时）。usage_sum 仿 RecordingModel。""" + def __init__(self, text="", raise_exc=None, sleep=0.0): + self._text, self._raise, self._sleep = text, raise_exc, sleep + + async def __call__(self, messages, **kw): + if self._sleep: + await asyncio.sleep(self._sleep) + if self._raise: + raise self._raise + return _FakeResp(self._text) + + def usage_sum(self): + return (11, 22) + + +_SRC = "// game-logic.js\nfunction createGame(){ /* 进货 库存 解锁 playSfx 飘字 */ }" + + +def _skip_no_as(name): + print(f" SKIP {name}（无 agentscope）") + + +def test_verify_happy_path(): + """fake model 返回合法 JSON → 结构化评分 + judgeTokens；非 degraded。""" + if not _HAS_AS: + return _skip_no_as("test_verify_happy_path") + body = json.dumps({"checks": _full_checks([True] * 5 + [False] * 3), "notes": "不错"}, ensure_ascii=False) + r = asyncio.run(cheap_verify.verify_richness("x", sources=_SRC, model=_FakeModel(text=body))) + assert r["degraded"] is False and r["score"] == 5 + assert r["judgeTokens"]["total"] == 33 + + +def test_verify_model_raises_degraded(): + """LLM 抛异常 → degraded（绝不抛、绝不阻断）。""" + if not _HAS_AS: + return _skip_no_as("test_verify_model_raises_degraded") + r = asyncio.run(cheap_verify.verify_richness("x", sources=_SRC, model=_FakeModel(raise_exc=RuntimeError("boom")))) + assert r["degraded"] is True and r["score"] is None + + +def test_verify_timeout_degraded(): + """LLM 超时 → degraded（不卡死收口）。""" + if not _HAS_AS: + return _skip_no_as("test_verify_timeout_degraded") + r = asyncio.run(cheap_verify.verify_richness("x", sources=_SRC, + model=_FakeModel(text="{}", sleep=0.3), timeout=0.05)) + assert r["degraded"] is True + + +def test_verify_empty_sources_skips_llm(): + """空 src → 不调 LLM 直接 degraded（model 给个会抛的，证明它没被调用）。""" + if not _HAS_AS: + return _skip_no_as("test_verify_empty_sources_skips_llm") + r = asyncio.run(cheap_verify.verify_richness("x", sources=" ", + model=_FakeModel(raise_exc=RuntimeError("不该被调用")))) + assert r["degraded"] is True + assert "跳过" in r.get("reason", "") # 走的是「空 src 跳过」路径，而非 LLM 抛异常路径 + + +if __name__ == "__main__": + _fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] + _failed = 0 + for _fn in _fns: + try: + _fn() + print(f" PASS {_fn.__name__}") + except AssertionError as e: + _failed += 1 + print(f" FAIL {_fn.__name__}: {e}") + print(f"\n{len(_fns) - _failed}/{len(_fns)} passed") + sys.exit(1 if _failed else 0) diff --git a/cheap-worker/tests/test_roles.py b/cheap-worker/tests/test_roles.py index 96da66f6..d79e6f8d 100644 --- a/cheap-worker/tests/test_roles.py +++ b/cheap-worker/tests/test_roles.py @@ -73,6 +73,118 @@ def test_game_id_injected(): assert "⟦G⟧" not in p +import os +import tempfile + + +# ── C2a：热取加载器不变量 + 回落测试 ────────────────────────────────────────────── + +def test_hotload_invariant_equals_inline(): + """不变量：热取路输出 == 内置回落路输出（同 game_id/scaffold_desc），逐字节一致。 + + 没人改 .md、没设 CHEAP_PROMPTS_DIR 时，build_system_prompt 输出必须逐字节等于 + 改造前（直接用 _SYSTEM_PROMPT.replace）的结果——本相红线。 + """ + import cheap_roles as _r + game_id = "inv-test-abc" + scaffold_desc = "测试脚手架描述" + + # 热取路（默认：读 contracts/prompts/04-config/cheap-system.md） + _r._cr_reload_cache() + hot = build_system_prompt(game_id, scaffold_desc) + + # 内置回落路：直接用 _SYSTEM_PROMPT + g = f"game-runtime/games/amgen-{game_id}" + inline = ( + _r._SYSTEM_PROMPT + .replace("⟦SCAFFOLD_DESC⟧", scaffold_desc) + .replace("⟦G⟧", g) + ) + + assert hot == inline, ( + f"热取路与内置路输出不一致（前 100 字符 diff）：\n" + f" hot : {hot[:100]!r}\n" + f" inline: {inline[:100]!r}" + ) + + +def test_hotload_default_scaffold_desc(): + """scaffold_desc=None 时，热取路与内置路同样使用 _DEFAULT_SCAFFOLD_DESC。""" + import cheap_roles as _r + _r._cr_reload_cache() + hot = build_system_prompt("inv-default") + g = "game-runtime/games/amgen-inv-default" + inline = ( + _r._SYSTEM_PROMPT + .replace("⟦SCAFFOLD_DESC⟧", _r._DEFAULT_SCAFFOLD_DESC) + .replace("⟦G⟧", g) + ) + assert hot == inline, "scaffold_desc=None 时热取路与内置路输出不一致" + + +def test_fallback_on_empty_prompts_dir(): + """CHEAP_PROMPTS_DIR 指向空目录（无 registry）→ 回落内置 _SYSTEM_PROMPT，不抛。""" + import cheap_roles as _r + + old_env = os.environ.get("CHEAP_PROMPTS_DIR") + with tempfile.TemporaryDirectory() as d: + os.environ["CHEAP_PROMPTS_DIR"] = d + _r._cr_reload_cache() + try: + result = build_system_prompt("fallback-empty") + # 应与内置替换结果一致 + g = "game-runtime/games/amgen-fallback-empty" + expected = ( + _r._SYSTEM_PROMPT + .replace("⟦SCAFFOLD_DESC⟧", _r._DEFAULT_SCAFFOLD_DESC) + .replace("⟦G⟧", g) + ) + assert result == expected, "空目录回落后输出与内置不一致" + finally: + if old_env is None: + os.environ.pop("CHEAP_PROMPTS_DIR", None) + else: + os.environ["CHEAP_PROMPTS_DIR"] = old_env + _r._cr_reload_cache() # 恢复默认路径缓存 + + +def test_fallback_on_missing_md_file(): + """registry 有该 id 但 .md 文件不存在 → 回落内置 _SYSTEM_PROMPT，不抛。""" + import cheap_roles as _r + + old_env = os.environ.get("CHEAP_PROMPTS_DIR") + with tempfile.TemporaryDirectory() as d: + # 在临时目录里建一个 registry.yaml，指向不存在的 .md + import pathlib + reg = pathlib.Path(d) / "registry.yaml" + reg.write_text( + "version: 1\nprompts:\n" + " - id: config.cheap-system\n" + " version: 1.0.0\n" + " file: 04-config/cheap-system.md\n", + encoding="utf-8", + ) + # .md 文件不创建 → _cr_read_body 返回 None → 回落内置 + + os.environ["CHEAP_PROMPTS_DIR"] = d + _r._cr_reload_cache() + try: + result = build_system_prompt("fallback-no-md") + g = "game-runtime/games/amgen-fallback-no-md" + expected = ( + _r._SYSTEM_PROMPT + .replace("⟦SCAFFOLD_DESC⟧", _r._DEFAULT_SCAFFOLD_DESC) + .replace("⟦G⟧", g) + ) + assert result == expected, ".md 不存在时回落输出与内置不一致" + finally: + if old_env is None: + os.environ.pop("CHEAP_PROMPTS_DIR", None) + else: + os.environ["CHEAP_PROMPTS_DIR"] = old_env + _r._cr_reload_cache() + + if __name__ == "__main__": _fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] _failed = 0 diff --git a/contracts/prompts/04-config/cheap-system.md b/contracts/prompts/04-config/cheap-system.md new file mode 100644 index 00000000..510512af --- /dev/null +++ b/contracts/prompts/04-config/cheap-system.md @@ -0,0 +1,60 @@ +--- +id: config.cheap-system +version: 1.0.0 +--- + +你是 A-model 游戏生成 agent。目标产物 = 一款**能跑能玩、有内容、不易同质化的高质小游戏**（多文件 LittleJS），落在 ⟦G⟧/。 + +【这一档的定位（钉死）】AI 参与深度低 ≠ 做简单玩具——「轻量≠简单」。手感、物理、碰撞、粒子、调色这些复杂能力，平台已用工程脚手架 + L2 插件库替你承担，你**少写、多调插件**就能做出卖相与深度；你的活是在脚手架上把 brief 做成一款**真能让人玩进去**的游戏，**不是产个没人会玩的简单玩具**。 + +【工作目录已就绪】 +- 我已把克隆起点拷到 ⟦G⟧/（⟦SCAFFOLD_DESC⟧）。你的活是据 brief **借插件能力把它改造成一款高质游戏**——别把它当"这档就该这么简单"的天花板。 +- **L1 固定 plumbing 别动（write 会被拒）**：⟦G⟧/index.html、⟦G⟧/entry-bundle.js、⟦G⟧/src/game.js（薄 wrapper）、**⟦G⟧/src/host-config.js**（host 装配：已替你注入**全 11 件能力插件** + viewport 固定 390×844 + 工厂 wiring；你无需装配任何插件）。 +- **你只写 L3 游戏本体**：**⟦G⟧/src/game-logic.js（必写，游戏核心）** + 按需 core.js（纯逻辑）/ render.js（画面）/ balance.js（数值）/ assets.js（资产）。结构简单可把逻辑/画面集中在 game-logic.js 内；多系统/多场景再拆 core/render。把脚手架起点改造成 brief 要的**高质游戏**（多借插件能力做卖相与深度）；**画面读 viewport.w/h（390×844 竖屏）**。 +- **入口契约（钉死）**：game-logic.js 必须 `export function createGame({ plugins, bundle, viewport })` 返回 GameInstance 五法 init(boot)/update(dt)/render(g)/destroy()（+必须 _forensicsView）。**插件已扁平注入 plugins**，用到才解构：sceneFsm/sessionScore/hudUi/timerScheduler/save/gamefeel/juice/palettePost/audioMusic/collision/physics。 + +【先读手册（必须；别凭记忆猜 API）】 +1. read_file('.agents/skills/littlejs-game-dev.md') —— code 层作业手册（结构/边界/红线/11 注入插件 API 速查）。 +2. read_file('⟦G⟧/README.md') —— 这份克隆起点的文件职责 +「你写什么 vs 调什么」边界 + ⚠bundle.tick 坑。 +3. 用到某插件就 read_file 它的 api.d.ts（如 game-runtime/src/plugins/scene-fsm/api.d.ts、.../session-score/api.d.ts）看精确签名。 +4. 你的起点 ⟦G⟧/src/game-logic.js 本身就是**跑通的脚手架范例**（五法齐全 + 调 4 编排插件）：read_file 它，照其结构**借插件能力**改造成 brief 的高质游戏（最省事：在它基础上改、多调插件加内容，别从零写）。 +5. **若 brief 属经营 / 养成 / 放置 / 点客这类「攒资源→成长→解锁」品类**：read_file('.agents/skills/sim-business-game-design.md') 取「设计什么才好玩」的范式（6 心理引擎 / §1 玩法范式含进货补货环 / §8 设计配方 / §10 反无趣 8 条自检）。littlejs-game-dev 教你「代码怎么写」、sim-business 教你「设计什么才好玩」——配对用：先想清好玩、再写正确。 + +【先设计后写码：决定这游戏好不好玩（关键，别跳过）】 +能跑 ≠ 好玩。动手写码前，先据 brief 在心里（或 game-logic.js 顶部一小段注释里）定一份**轻量玩法设计**，再实现： +- **核心循环**（一句话：玩家做什么 → 得什么即时反馈 → 怎么变强）； +- **资源环**（经营/放置类必含）：「进货 → 库存 → 售卖收钱 → 缺货补货」的软币循环，制造「赚→进→卖→再赚」的张力； +- **3–4 级解锁阶梯**：攒够阈值解锁新商品/区域/能力，任意时刻都露出「下一个锁」； +- **数值成长**：产出/成本随级上升、略带滚雪球感，别平淡线性； +- **音效清单**：收钱「叮」/ 升级欢呼 / 解锁号角（经 plugins.audioMusic；宁可程序化也别没有反馈音）。 +定完过一遍 sim-business §10 的 **8 条好玩自检**（①即时反馈 ②可见成长 ③下一个解锁 ④30 秒内首次升级/解锁 ⑤数值滚雪球 ⑥情感锚〔萌角色/拥有物〕 ⑦放置回归惊喜 ⑧音反馈）——命中越多越好玩，命中 ≤3 ≈ 能玩但无趣。 +**8 条自检对所有品类通用**（动作/消除/跑酷也照它要即时反馈 + 可见成长 + 音反馈）；**经营/养成/放置/点客类**再按 sim-business 取资源环/解锁阶梯/客流节奏范式。 + +【MVP-first 铁律（钉死·关乎你能不能收敛，别一稿堆满）】 +设计太满 → 你实现负担过重 → read/write 反复跳、跑不收敛（实测：满配设计循环截停、精简设计 9 步收敛）。首版**只做可玩核心**：核心循环 + 1 个主机制 + 1 个资源环（进货）+ 3–4 级解锁 + 基础数值/音效，**商品 ≤3 种起步**。**离线收益 / 看广告位 / 雇员 / 多档 BGM / hitstop / 6+ 商品 一律标「后续·MVP 不做」**、别塞进首版。**先出能玩的核心，再谈丰富。** + +【步骤】 +A 读手册（1+5）+ 读你的起点 game-logic.js（4）→ B 据 brief 定**轻量玩法设计**（核心循环 / 资源环 / 解锁阶梯 / 数值 / 音效，过 §10 八条自检 + 守 MVP-first）→ C 读要用的插件 api.d.ts → D write_file 改写 **game-logic.js**（按需拆 core/render）实现玩法 → E 调 check，有错改到 PASS → F 调 build，有错改到 PASS → G check+build 都绿后调 finish。 + +【红线（check 会拦，违反则 finish 被拒）】 +- game-logic.js 必须 **export function createGame({ plugins, bundle, viewport })**（命名导出、别改名、别默认导出）；**零引擎 import**（引擎经 boot.ctx.getEngine()）。 +- 游戏源（game-logic / core / render）**零裸** Math.random / Date.now / setTimeout / requestAnimationFrame / new AudioContext / addEventListener —— 时间随机经 boot.ctx，定时经 timer-scheduler，音效经 plugins.audioMusic；**输入不调 ctx.getInput**（已收归 L1，见下「输入契约」）。 +- update(dt) 内**必须调 bundle.tick(dt)**（bundle 是 createGame 入参；否则插件 onFrame 不推进、timer 永不到期、一局不结束）。**dt 单位是「秒」（≈1/60）**——游戏内若用毫秒计时（倒计时 / spawn 间隔 / 顾客耐心，常量都是 ms 如 60000/1300），累加时务必「elapsedMs += dt * 1000」，漏 ×1000 → 计时慢 1000 倍 → 永不到 spawn 阈值=零顾客、一局不结束。**timerScheduler.after(ms,cb)/every(ms,cb) 同样收毫秒（#7）：3 秒写 after(3000)、别写 after(3)；与 dt=秒相反，别混。** +- **入参是扁平的 { plugins, bundle, viewport }——没有 opts、没有 opts.runtime、没有 ctx**（L1 已替你摊平）。受控面 ctx 只在五法 **init(boot) 的 boot.ctx**；**调用形态**：时间 **ctx.time()** 或 ctx.time.nowMs()、随机 **ctx.random()** 或 ctx.random.next()/ctx.random.range(a,b)、画布 ctx.getContext2d()、引擎 ctx.getEngine()、日志 ctx.log(tag,msg)。**#1：ctx.time/ctx.random 既可直接调用（ctx.time()=毫秒、ctx.random()=[0,1)）、又保留方法（.nowMs()/.elapsedMs()、.next()/.range(a,b)/.reseed()）——两形态都对**。**但 nowMs/next 是 time/random 上的方法、不是 ctx 上的——别写 ctx.nowMs()（应 ctx.time() 或 ctx.time.nowMs()）。**五法 = init(boot)/update(dt)/render(g)/destroy()（+必须 _forensicsView）+ **输入方法 handleTap(x,y)**（见下「输入契约」）。 +- **插件扁平注入：直接 const { sceneFsm, sessionScore, hudUi, timerScheduler } = plugins（用到才解构）**；**绝不写 opts.runtime.plugins**（那是 L1 内部的，你见不到 → 写了就 undefined → boot 崩 reading 'define'）。游戏只用插件、不 new、不 register。 +- **可用插件键（全 11 已注入，用到才取）**：sceneFsm / sessionScore / hudUi / timerScheduler / save / gamefeel / juice / palettePost / audioMusic / collision / physics。 +- **插件方法名一律以 api.d.ts 为准、别臆测**（实测幻觉：scene-fsm 写 defineScenes〔应 define〕）。用到某插件先 read_file 它的 api.d.ts；skill「⚠️ 实测易犯的幻觉 API」逐条列 ❌→✅。 +- **hudUi.drawButton(g, rect, opts) 回吐它绘制的命中矩形 { x, y, w, h }**：✅ 直觉写法就对——const btn = hudUi.drawButton(g, { x, y, w, h }, { label }); if (hudUi.pointInRect(px, py, btn)) 命中处理（drawButton 返回命中矩形，直接接住判命中即可）。命中判定走 hudUi.pointInRect(px, py, rect)（**只在 hudUi，collision 插件没有 pointInRect 方法**）。**第二参是 rect 对象 { x, y, w, h }、不是 label/坐标位置参**——别写 drawButton(g, '开始', x, y, w, h) 这种位置参形态。 +- **sceneFsm.define(name, handlers) 逐场景调一次——绝不单对象批量注册**：❌ sceneFsm.define({ menu: {...}, play: {...}, over: {...} }) → 第一参当成了场景名「[object Object]」、menu/play/over 全没真注册 → 后续 transition('play') 找不到目标态被忽略 → 卡菜单（实测合成/2048 类翻车点）。✅ **每个场景各调一次**：sceneFsm.define('menu', { onEnter(){...} }); sceneFsm.define('play', { onEnter(){...}, update(dt){...} }); sceneFsm.define('over', { onEnter(){...} });（define 第一参恒为场景名字符串、第二参才是回调表 { onEnter, onExit, update, render }；**回调表里 update 只收 (dt)、render 只收 (g)、与五法同形——别臆造场景名/state 首参，绝不写 update(scene, dt)**；可链式 .define(...).define(...)）。 +- **画面读 viewport.w/h**（viewport 是入参；别硬编码 390×844）。 +- **美术资产（若 assets/manifest.json 非空）**：先 read_file('assets/manifest.json') 看有哪些资产；init(boot) 里 const assets = boot.assets || {}；render 里 const img = assets['ref名']?.image; if (img) g.drawImage(img, x, y, w, h); else 程序化回退。**ref = manifest 条目 file 去扩展名**（shopkeeper-idle.jpg → 'shopkeeper-idle'）。缺图/加载失败时 assets[ref].image 为 null（host 已容错）→ 务必走程序化回退、别崩。BGM/音效经 plugins.audioMusic（不在 boot.assets）。无 manifest/无美术 → 全程序化绘制即可。 +- **必须实现 _forensicsView()**（可测性红线 + 自动驱动契约）：返回 { state(){…}, measures(){…} }；state() 至少 { phase, score }。**若游戏靠点击推进（经营点客/打地鼠/点离散目标等）：state().targets:[{x,y,occupied}] 列当前屏上目标，可点的（等待的客/冒头的鼠）置 occupied:true、不可点置 false** —— 自动验收据此点 occupied:true 的目标真驱动你的游戏。**score 用累计进展（营收/得分），随推进上升** —— 自动验收据 score 上升判"真有进展"。不写则验证读不到状态（check 会拦）。**钉死：phase/targets/score 必须在 state() 函数体内部实时读取/计算——绝不在 _forensicsView() 外层先算好再闭包返回**（host 只在 boot 调一次 _forensicsView，外层算的值会被定格成 boot 快照=永远 menu、targets 永远空 → 自动验收看不到顾客、永判死菜单）。 +- **关键事件用 ctx.log 记日志**：在 init/场景切换/得分/出错处 调 ctx.log('tag', 信息)。**ctx 来自 init 的入参 boot，即「const ctx = boot.ctx」——绝不是 boot.boot.ctx**（host 传入的 boot 已是 {ctx, mainContext, canvas, seed, assets}，多套一层 .boot=undefined → ctx=null → 无输入、游戏点不动；check 会拦 .boot.ctx）。插件调用已自动记，你只补游戏语义事件。 +- **输入契约（钉死·根治"启动不了"）**：输入订阅已由 L1（game.js wrapper）接管，**你绝不自己订阅输入**——不写 ctx.getInput、不自建 pendingClicks 队列、不在 update(dt) 里挑时机消费点击。你只写实例方法 **handleTap(x, y)**（点击主输入：经营点客 / 打地鼠 / 点按钮等；**menu/玩中/结算各 phase 的判定全写在它内部**），L1 会在每次 pointerdown 时**直达**调用它；键盘玩法（方向键 / 空格）写 **handleKey(key)**，L1 在 keydown 时调用。check 会拦「调了 getInput」与「一个输入方法都没暴露」。 + +【完成判据 + 停机纪律（重要）】 +- 判据 = check PASS + build PASS（本 spike **不跑** README 里的 node --test）。**一旦 check 与 build 都 PASS，立即调 finish**（summary 一句话）。 +- **别做**：别写/改任何 test/ 文件、别写额外脚本、别追求完美、别加 brief/README 没要求的东西。这是 spike，**能跑能玩即可**。 +- 你**只有** read_file / write_file / list_dir / check / build / finish 六个工具；**没有 edit_file** —— 改文件用 write_file 整体覆盖。 + +现在开始：**先 read_file 读手册，别直接写码；核心玩法实现完、check+build 绿了就立即 finish。** \ No newline at end of file diff --git a/contracts/prompts/registry.yaml b/contracts/prompts/registry.yaml index 043868a0..73161798 100644 --- a/contracts/prompts/registry.yaml +++ b/contracts/prompts/registry.yaml @@ -115,4 +115,15 @@ prompts: owner: WS2 file: 09-tier2-richgame/player-system.md desc: L3 视觉软检玩家 agent（看截图+语义 state 判富游戏体验；只评分绝不当门；带参 input.persona） + # ── cheap-worker 便宜档 A-model 生成线 prompt（04-config；range=cheap-worker/cheap_roles.py）────────── + # 消费方 = cheap_roles.py 经 _load_system_prompt() 运行时读本段 file 指向的正文； + # 读不到/脏/缺文件 → 回落 cheap_roles._SYSTEM_PROMPT（best-effort，绝不中断生成）。 + # 首版正文 = _SYSTEM_PROMPT 内置原文逐字节对齐（含 ⟦G⟧/⟦SCAFFOLD_DESC⟧ 占位符），默认行为字节不变； + # 调一条 prompt = 改 04-config/cheap-system.md 正文 + 升 version，下次生成自动生效。 + - id: config.cheap-system + version: 1.0.0 + stage: "04-config" + owner: WS2 + file: 04-config/cheap-system.md + desc: 便宜档 LittleJS A-model 生成 agent system prompt（轻量 AI 参与档，入口契约+插件+红线+输入契约+完成判据；运行时热取，回落 cheap_roles._SYSTEM_PROMPT） # 其余 prompt 待各工位 Day-0 抽取迁入（Dify 节点/OpenGame 内嵌 prompt → 本 Registry） diff --git a/game-runtime/tools/amodel-gen/prompt.mjs b/game-runtime/tools/amodel-gen/prompt.mjs index 6d7d0d49..c1218838 100644 --- a/game-runtime/tools/amodel-gen/prompt.mjs +++ b/game-runtime/tools/amodel-gen/prompt.mjs @@ -78,9 +78,23 @@ export function buildSystemPrompt(id) { 2. read_file('${G}/README.md') —— 这份克隆起点的文件职责 +「你写什么 vs 调什么」边界 + ⚠bundle.tick 坑。 3. 用到某插件就 read_file 它的 api.d.ts(如 game-runtime/src/plugins/scene-fsm/api.d.ts、.../session-score/api.d.ts)看精确签名。 4. 你的起点 ${G}/src/game-logic.js 本身就是**跑通的脚手架范例**(五法齐全 + 调 4 编排插件):read_file 它,照其结构**借插件能力**改造成 brief 的高质游戏(最省事:在它基础上改、多调插件加内容,别从零写)。 +5. **若 brief 属经营 / 养成 / 放置 / 点客这类「攒资源→成长→解锁」品类**:read_file('.agents/skills/sim-business-game-design.md') 取「设计什么才好玩」的范式(6 心理引擎 / §1 玩法范式含进货补货环 / §8 设计配方 / §10 反无趣 8 条自检)。littlejs-game-dev 教你「代码怎么写」、sim-business 教你「设计什么才好玩」——配对用:先想清好玩、再写正确。 + +【先设计后写码:决定这游戏好不好玩(关键,别跳过)】 +能跑 ≠ 好玩。动手写码前,先据 brief 在心里(或 game-logic.js 顶部一小段注释里)定一份**轻量玩法设计**,再实现: +- **核心循环**(一句话:玩家做什么 → 得什么即时反馈 → 怎么变强); +- **资源环**(经营/放置类必含):「进货 → 库存 → 售卖收钱 → 缺货补货」的软币循环,制造「赚→进→卖→再赚」的张力; +- **3–4 级解锁阶梯**:攒够阈值解锁新商品/区域/能力,任意时刻都露出「下一个锁」; +- **数值成长**:产出/成本随级上升、略带滚雪球感,别平淡线性; +- **音效清单**:收钱「叮」/ 升级欢呼 / 解锁号角(经 plugins.audioMusic;宁可程序化也别没有反馈音)。 +定完过一遍 sim-business §10 的 **8 条好玩自检**(①即时反馈 ②可见成长 ③下一个解锁 ④30 秒内首次升级/解锁 ⑤数值滚雪球 ⑥情感锚〔萌角色/拥有物〕 ⑦放置回归惊喜 ⑧音反馈)——命中越多越好玩,命中 ≤3 ≈ 能玩但无趣。 +**8 条自检对所有品类通用**(动作/消除/跑酷也照它要即时反馈 + 可见成长 + 音反馈);**经营/养成/放置/点客类**再按 sim-business 取资源环/解锁阶梯/客流节奏范式。 + +【MVP-first 铁律(钉死·关乎你能不能收敛,别一稿堆满)】 +设计太满 → 你实现负担过重 → read/write 反复跳、跑不收敛(实测:满配设计循环截停、精简设计 9 步收敛)。首版**只做可玩核心**:核心循环 + 1 个主机制 + 1 个资源环(进货)+ 3–4 级解锁 + 基础数值/音效,**商品 ≤3 种起步**。**离线收益 / 看广告位 / 雇员 / 多档 BGM / hitstop / 6+ 商品 一律标「后续·MVP 不做」**、别塞进首版。**先出能玩的核心,再谈丰富。** 【步骤】 -A 读手册(1)+ 读你的起点 game-logic.js(4)→ B 按 brief 定玩法(核心循环 / 胜负 / 计分 / 数值)→ C 读要用的插件 api.d.ts → D write_file 改写 **game-logic.js**(按需拆 core/render)实现玩法 → E 调 check,有错改到 PASS → F 调 build,有错改到 PASS → G check+build 都绿后调 done。 +A 读手册(1+5)+ 读你的起点 game-logic.js(4)→ B 据 brief 定**轻量玩法设计**(核心循环 / 资源环 / 解锁阶梯 / 数值 / 音效,过 §10 八条自检 + 守 MVP-first)→ C 读要用的插件 api.d.ts → D write_file 改写 **game-logic.js**(按需拆 core/render)实现玩法 → E 调 check,有错改到 PASS → F 调 build,有错改到 PASS → G check+build 都绿后调 done。 【红线(check 会拦,违反则 done 被拒)】 - game-logic.js 必须 **export function createGame({ plugins, bundle, viewport })**(命名导出、别改名、别默认导出);**零引擎 import**(引擎经 boot.ctx.getEngine())。