From 8f700c4b422f0c0d269bdb42dbbdc287111906bd Mon Sep 17 00:00:00 2001 From: lili Date: Wed, 8 Jul 2026 02:23:36 -0700 Subject: [PATCH] =?UTF-8?q?fix(cheap-gen):=20=E6=AD=BB=E5=9C=88=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20fix=E2=91=A2=20brief=20=E8=90=BD=20evidence/brief.j?= =?UTF-8?q?son=20=E4=BE=9B=E6=A0=B9=E5=9B=A0=E5=8F=96=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题:一局 brief 原只内联在 run-summary.json,收口末尾才落——run 硬失败/挂死(如熔断 os._exit)时收不了口、brief 丢失,死圈根因取证断链;且无完整性戳、无限长策略。 改动(cheap_studio.py):scaffold 就绪后即调 _persist_brief_forensics 落 game_dir/evidence/brief.json,字段 {brief(限长)、bytes、sha256、truncated、promptMax、source}: · 早落——硬失败/挂死也留得下,取证不断链; · sha256/bytes 按【完整原文】算——核 worker 收到的 brief 与后端下发字节一致(查是否 brief 被截断/篡改害翻车),截断不掩盖完整性核对; · 超 PROMPT_MAX=1000 截断 brief 字段 + truncated=True——脱敏 + 防超长撑爆; · 落 evidence/(forensics,不随 src/ 进素材市场交易),【绝不】落工程根——源工程会被交易、 brief 是用户私有输入,不得泄进可交易源; · 非阻塞铁律:落盘任何异常只告警、绝不咬生成主链(try/except 兜底)。 测:新增 test_brief_forensics.py 4 测(落 evidence 非工程根/字段齐 + sha256 按完整原文/ 超长截断但摘要仍完整/空与 None 非阻塞),+ 回归 test_edit_file_toolkit 6 测,pytest exit=0; cheap_studio import smoke 通过(import hashlib + 接线未破坏加载)。 待续(本计划剩项):fix①-B 共享 tier2 熔断窄口径纠偏(最险,另做)、n=5 收敛环。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cheap-worker/cheap_studio.py | 31 ++++++++ cheap-worker/tests/test_brief_forensics.py | 86 ++++++++++++++++++++++ 2 files changed, 117 insertions(+) create mode 100644 cheap-worker/tests/test_brief_forensics.py diff --git a/cheap-worker/cheap_studio.py b/cheap-worker/cheap_studio.py index 6cdd67b7..0eca18e2 100644 --- a/cheap-worker/cheap_studio.py +++ b/cheap-worker/cheap_studio.py @@ -12,6 +12,7 @@ cheap_studio.py — 便宜档 ReAct 主编排（spike 核心；对照源：Node CLI：cheap-worker/.venv/bin/python cheap-worker/cheap_studio.py --id cheap-run1 --brief "一个简单的点击得分小游戏" """ +import hashlib import json import sys import time @@ -56,6 +57,35 @@ def _rec(msg: str) -> None: print(f"[cheap_studio] {msg}", file=sys.stderr, flush=True) +# fix③(死圈修复 2026-07-08):brief 落 evidence/brief.json 供根因取证。 +# 为什么单独落而不只靠 run-summary.brief:① 早落(scaffold 后即写)——run 硬失败/挂死时 run-summary 收不了口、 +# brief.json 已在盘上,取证不断链;② 带 sha256 完整性戳(核 worker 收到的 brief 与后端下发字节一致,查"是不是 brief +# 被截断/篡改害的生成翻车");③ 限长脱敏(超 PROMPT_MAX 截断 + truncated 标记),防超长 brief 撑爆落盘。 +# 落 evidence/(forensics 目录,不随 src/ 进素材市场交易),绝不落工程根——源工程会被交易,brief 是用户私有输入、不得泄进可交易源。 +_BRIEF_PROMPT_MAX = 1000 # brief 落盘限长(字符);超出截断 + truncated=True,sha256 仍按【完整原文】算(截断不掩盖完整性核对) + + +def _persist_brief_forensics(game_id: str, brief: str, source: str = "job.brief") -> None: + """把本局 brief 落 evidence/brief.json(取证用,见上方注释)。非阻塞:任何异常只告警、绝不咬生成主链。""" + try: + text = brief or "" + raw = text.encode("utf-8") + rec = { + "brief": text[:_BRIEF_PROMPT_MAX], # 限长:超长只留前 N 字符(脱敏 + 防撑爆) + "bytes": len(raw), # 完整 brief 的 UTF-8 字节数 + "sha256": hashlib.sha256(raw).hexdigest(), # 完整 brief 摘要:核 worker↔后端字节一致 + "truncated": len(text) > _BRIEF_PROMPT_MAX, # 是否因限长被截 + "promptMax": _BRIEF_PROMPT_MAX, + "source": source, # 来源溯源(默认 job.brief) + } + ev_dir = cheap_run.game_dir(game_id) / "evidence" + ev_dir.mkdir(parents=True, exist_ok=True) + (ev_dir / "brief.json").write_text(json.dumps(rec, ensure_ascii=False, indent=2), encoding="utf-8") + _rec(f"brief 落盘取证 → {ev_dir / 'brief.json'} bytes={rec['bytes']} truncated={rec['truncated']}") + except Exception as e: # noqa: BLE001 非阻塞铁律:取证落盘绝不影响生成主链 + _rec(f"brief 落盘异常(已忽略,不影响生成):{type(e).__name__}: {e}") + + def _bypass_state() -> AgentState: """无人值守跳过工具 ASK（否则 FunctionTool 默认 check_permissions 返回 ASK 卡死；对 studio.py:_bypass_state）。""" return AgentState(permission_context=PermissionContext(mode=PermissionMode.BYPASS)) @@ -193,6 +223,7 @@ async def run_studio(game_id, brief, *, max_iters=40, max_resumes=6, max_tokens= if not sc["ok"]: return {"ok": False, "gameId": game_id, "fail": "准备失败：" + sc["output"]} _rec(f"准备就绪 amgen-{game_id}（{'scaffold clone _template + SAA 信封' if prepare is None else 'modify: base 源已预备'}）") + _persist_brief_forensics(game_id, brief) # fix③:scaffold 后即落 brief 取证(硬失败/挂死也留得下,不进可交易工程根) session = CheapSession(game_id=game_id) toolkit = build_toolkit(session, write_whitelist=write_whitelist) # modify 收窄到只许写 game-logic.js diff --git a/cheap-worker/tests/test_brief_forensics.py b/cheap-worker/tests/test_brief_forensics.py new file mode 100644 index 00000000..f92ca8f9 --- /dev/null +++ b/cheap-worker/tests/test_brief_forensics.py @@ -0,0 +1,86 @@ +"""test_brief_forensics.py — fix③ brief 落盘取证单测(死圈修复 2026-07-08)。 + +守的不变量: + · brief 落 game_dir/evidence/brief.json(forensics 目录),【绝不】落工程根 game_dir/(源工程可进素材市场交易、brief 是用户私有); + · 字段齐 {brief, bytes, sha256, truncated, promptMax, source}; + · sha256 / bytes 按【完整原文】算(截断不掩盖完整性核对); + · 超 PROMPT_MAX 截断:brief 字段限长、truncated=True,但 sha256 仍是完整原文摘要; + · 非阻塞:落盘异常不抛(此处以正常路径为主,异常路径由 try/except 兜底保证)。 + +用真 games/ 下临时 game_dir(amgen-brieftest*),teardown 清理,不碰真产物。 +跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_brief_forensics.py -q +""" + +import hashlib +import json +import shutil +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # → cheap-worker/ +import cheap_run # noqa: E402 +import cheap_studio # noqa: E402 +import pytest # noqa: E402 + +_GID = "brieftest001" + + +@pytest.fixture(autouse=True) +def _fixture_game_dir(): + gd = cheap_run.game_dir(_GID) + gd.mkdir(parents=True, exist_ok=True) + try: + yield gd + finally: + shutil.rmtree(gd, ignore_errors=True) + + +def _read_brief_json(): + return json.loads((cheap_run.game_dir(_GID) / "evidence" / "brief.json").read_text(encoding="utf-8")) + + +def test_brief_persisted_to_evidence_not_root(): + brief = "做一个点击得分的小游戏" + cheap_studio._persist_brief_forensics(_GID, brief) + gd = cheap_run.game_dir(_GID) + # 落 evidence/,不落工程根 + assert (gd / "evidence" / "brief.json").exists() + assert not (gd / "brief.json").exists(), "brief 绝不能落工程根(源工程会被交易)" + rec = _read_brief_json() + assert rec["brief"] == brief + assert rec["bytes"] == len(brief.encode("utf-8")) + assert rec["sha256"] == hashlib.sha256(brief.encode("utf-8")).hexdigest() + assert rec["truncated"] is False + assert rec["source"] == "job.brief" + assert rec["promptMax"] == cheap_studio._BRIEF_PROMPT_MAX + + +def test_long_brief_truncated_but_hash_is_full(): + brief = "关" * 3000 # 远超 PROMPT_MAX + cheap_studio._persist_brief_forensics(_GID, brief, source="test.long") + rec = _read_brief_json() + assert rec["truncated"] is True + assert len(rec["brief"]) == cheap_studio._BRIEF_PROMPT_MAX # brief 字段被限长 + assert rec["bytes"] == len(brief.encode("utf-8")) # bytes 按完整原文 + assert rec["sha256"] == hashlib.sha256(brief.encode("utf-8")).hexdigest() # 摘要按完整原文 + assert rec["source"] == "test.long" + + +def test_empty_brief_ok(): + cheap_studio._persist_brief_forensics(_GID, "") + rec = _read_brief_json() + assert rec["brief"] == "" + assert rec["bytes"] == 0 + assert rec["sha256"] == hashlib.sha256(b"").hexdigest() + assert rec["truncated"] is False + + +def test_non_blocking_on_bad_input(): + # None brief 不应抛(非阻塞铁律:取证落盘绝不咬生成) + cheap_studio._persist_brief_forensics(_GID, None) # type: ignore[arg-type] + rec = _read_brief_json() + assert rec["brief"] == "" + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-q"]))