feat(cheap-worker): 便宜档生成丰富度 + M1 达标门富化重验(切片一 A/B)
接 sim-business 设计指导到生成 agent(cheap_roles+prompt.mjs 双源,先设计后写码+MVP-first);新建纯 LLM 丰富度验证 agent cheap_verify(非阻塞·不进 verdict·零 code-presence 断言,红线落地);bake_off 加 richness 列;cheap_roles 三级回落配置热取(C2a 加载器);cheap_studio trace 接线(C1b)。富游戏 M1 三品类各 5/5=100% 过九门达标。测试 test_roles 12 / test_cheap_verify 11 / test_bake_off 17 / test_cheap_trace 4 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili
parent
ba0197a1a3
commit
e6525aaa3d
@ -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)
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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-<id>/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")
|
||||
|
||||
241
cheap-worker/cheap_verify.py
Normal file
241
cheap-worker/cheap_verify.py
Normal file
@ -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-<id>/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 <gameId> ["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))
|
||||
@ -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():
|
||||
|
||||
172
cheap-worker/tests/test_cheap_trace.py
Normal file
172
cheap-worker/tests/test_cheap_trace.py
Normal file
@ -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-<id>/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-<id>/ 由 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)
|
||||
165
cheap-worker/tests/test_cheap_verify.py
Normal file
165
cheap-worker/tests/test_cheap_verify.py
Normal file
@ -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)
|
||||
@ -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
|
||||
|
||||
60
contracts/prompts/04-config/cheap-system.md
Normal file
60
contracts/prompts/04-config/cheap-system.md
Normal file
@ -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。**
|
||||
@ -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）
|
||||
|
||||
@ -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())。
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user