fix(cheap-gen): 死圈修复 fix③ brief 落 evidence/brief.json 供根因取证

问题:一局 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) <noreply@anthropic.com>
This commit is contained in:
lili 2026-07-08 02:23:36 -07:00
parent b63fd1b8f7
commit 8f700c4b42
2 changed files with 117 additions and 0 deletions

View File

@ -12,6 +12,7 @@ cheap_studio.py — 便宜档 ReAct 主编排spike 核心对照源Node
CLIcheap-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

View File

@ -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"]))