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

"""
cheap_toolkit.py — 便宜档 LittleJS 六工具（对照源：Node prompt.mjs 的 TOOLS + gen.mjs done 门 + tier2 toolkit.py 组装范式）。

工具用 AgentScope FunctionTool 薄壳包 async 闭包函数（schema 据类型注解 + Google docstring 自动抽，prompt 里不手写），
闭包共享一个 CheapSession 可变状态（对 tier2 toolkit.py 的 Tier2Session）。底层：read/list/write 走 cheap_run
Python 实现、check/build shell-out node。done 门 = finish 工具（工具内复核 check+build 绿才接受，对 gen.mjs done 门）。
"""

import json
from dataclasses import dataclass
from typing import Optional

import cheap_run

# 单次工具结果注入上限（对 gen.mjs TOOL_RESULT_CAP：够一份 skill/api.d.ts，控 context）。
_TOOL_RESULT_CAP = 30000


@dataclass
class CheapSession:
    """便宜档单 run 可变状态闭包（对 tier2 toolkit.py 的 Tier2Session，LittleJS 简化版）。

    单 session 绑单 agent、单 run，不跨 run 复用（同 tier2 toolkit 的三条边界铁律）。
    """

    game_id: str
    last_check: Optional[dict] = None
    last_build: Optional[dict] = None
    finished: Optional[dict] = None  # finish 组装的产物摘要（None=未收敛，编排层据此判收敛）


def build_toolkit(session: CheapSession, *, write_whitelist=None):
    """组装便宜档 Toolkit（对 tier2 toolkit.py build_toolkit）。返回 agentscope.tool.Toolkit。

    Args:
        session: 本 run 可变状态闭包。
        write_whitelist: 可选写边界白名单（basename 集合）。None=create 路不收窄（沿用 cheap_run.write_file
            的 L3 边界，禁 L1）；A11 M4 模块重生成传 {"game-logic.js"}，把写边界收窄到只许写玩法文件——
            其余 L3（core/render/assets）连同 L1 一并禁写，确保「只改玩法、不动别处」。
    """
    from agentscope.tool import Toolkit, FunctionTool

    async def read_file(path: str) -> str:
        """读 repo 内任意文件（只读）。用它读 skill、插件 api.d.ts、_template 范例。先读手册再写码。

        Args:
            path: repo 相对路径，如 .agents/skills/littlejs-game-dev.md
        """
        r = cheap_run.read_file(path)
        if not r["ok"]:
            return "ERROR: " + r["error"]
        prefix = "[内容已截断]\n" if r.get("truncated") else ""
        return (prefix + r["content"])[:_TOOL_RESULT_CAP]

    async def list_dir(path: str) -> str:
        """列目录（只读），用于发现文件。

        Args:
            path: repo 相对路径。
        """
        r = cheap_run.list_dir(path)
        if not r["ok"]:
            return "ERROR: " + r["error"]
        return "\n".join(r["entries"])[:_TOOL_RESULT_CAP]

    async def write_file(path: str, content: str) -> str:
        """写文件到本 run 的 game 目录（仅许写 games/amgen-<id>/ 内的 L3 游戏文件）。整文件覆盖。

        Args:
            path: 如 game-runtime/games/amgen-<id>/src/game-logic.js
            content: 完整文件内容。
        """
        # M4 模块重生成：写边界白名单生效时，basename 不在白名单（即非 game-logic.js）的写一律拒——
        #   本次只重写玩法文件，其余文件保持不变（read_file 读它取上下文即可）。create 路 white=None 不收窄。
        if write_whitelist is not None:
            base = path.rsplit("/", 1)[-1]
            if base not in write_whitelist:
                allowed = "/".join(sorted(write_whitelist))
                return (f"ERROR: 本次是「只改玩法」的模块重生成，只许写 {allowed}；"
                        f"{base} 属其余文件，要保持不变、不要写它（需要上下文就 read_file 读它）。")
        r = cheap_run.write_file(session.game_id, path, content)
        return f"OK {r['bytes']}B 已写" if r["ok"] else "ERROR: " + r["error"]

    async def edit_file(path: str, old: str, new: str) -> str:
        """增量改文件：把文件里出现一次的 old 串换成 new。改小处（如一个函数几行）用它，别整文件重写。

        old 必须逐字节精确匹配文件原文（含缩进/换行）且唯一命中。找不到/不唯一会明确报错——按提示加长 old 再试。
        边界同 write_file：只许改 games/amgen-<id>/ 内 L3 游戏文件，L1 plumbing 拒改。

        Args:
            path: 如 game-runtime/games/amgen-<id>/src/game-logic.js
            old: 要被替换的原文片段（复制文件里的原样，含缩进）。
            new: 替换成的新片段。
        """
        # 写边界白名单：与 write_file 同口径（M4"只改玩法"时非 game-logic.js 一律拒）。
        if write_whitelist is not None:
            base = path.rsplit("/", 1)[-1]
            if base not in write_whitelist:
                allowed = "/".join(sorted(write_whitelist))
                return (f"ERROR: 本次是「只改玩法」的模块重生成，只许改 {allowed}；"
                        f"{base} 属其余文件，要保持不变、不要改它。")
        r = cheap_run.edit_file(session.game_id, path, old, new)
        return r.get("message", "OK edit 已应用") if r["ok"] else "ERROR: " + r["error"]

    async def check() -> str:
        """循环内快校验：node --check 各 src/*.js + 五法/导出名/红线/形状门 lint。返回 PASS 或错误清单。finish 前必须 PASS。"""
        session.last_check = cheap_run.check(session.game_id)
        r = session.last_check
        return ("PASS" if r["ok"] else "FAIL:\n" + r["output"])[:_TOOL_RESULT_CAP]

    async def build() -> str:
        """循环内 esbuild 打包（秒级，SAA 信封 __GameBundle）。返回 PASS 或编译错误。finish 前必须 PASS。"""
        session.last_build = cheap_run.build(session.game_id)
        r = session.last_build
        return ("PASS" if r["ok"] else "FAIL:\n" + r["output"])[:_TOOL_RESULT_CAP]

    async def finish(summary: str) -> str:
        """声明完成。done 门：工具内复核 check 与 build 都 PASS 才接受（不信自评；对 gen.mjs done 门）。

        Args:
            summary: 一句话：你做了一款什么游戏、核心玩法。
        """
        c = cheap_run.check(session.game_id)
        b = cheap_run.build(session.game_id) if c["ok"] else {"ok": False, "output": "check 未过，先修 check"}
        session.last_check = c
        if c["ok"]:
            session.last_build = b
        if c["ok"] and b["ok"]:
            session.finished = {"summary": summary, "gameId": session.game_id}
            return json.dumps({"ok": True, "message": "DONE 已接受（check+build 绿）"}, ensure_ascii=False)
        return json.dumps({
            "ok": False, "blockedByGate": True,
            "message": ("finish 被拒：done 前 check+build 必须都绿。"
                        f"check={'PASS' if c['ok'] else 'FAIL'} build={'PASS' if b.get('ok') else 'FAIL'}"),
            "checkOutput": (c.get("output") or "")[:2000],
            "buildOutput": (b.get("output") or "")[:2000],
        }, ensure_ascii=False)

    tools = [
        FunctionTool(read_file), FunctionTool(list_dir), FunctionTool(write_file),
        FunctionTool(edit_file), FunctionTool(check), FunctionTool(build), FunctionTool(finish),
    ]
    return Toolkit(tools=tools)