games-development-ai/docs/plans/2026-07-02-配置控制面阶段一-护城河middleware-plan.md

date, topic, status, 关联设计, 上级, 前置, 范围

date	topic	status	关联设计	上级	前置	范围
2026-07-02	配置控制面-阶段一①	§6.8 双评审已过(Opus+Codex 双线独立收敛,2 Critical + 7 Important + 3 Minor 已文内修;Codex 另发现 I-2 异常穿透已修)· 3 材料级判断创始人已拍(续修时延接受慢重标定 / 预算耗尽至少修一次 / 尽力产物只 on-disk)· 待创始人终评审 → SDD 执行	docs/agent-specs/2026-06-30-配置控制面一次性按序实现-设计.md(§3.4 护城河 middleware / §4 阶段一)	docs/agent-specs/2026-06-30-配置控制面一次性按序实现-设计.md	docs/plans/2026-07-01-配置控制面-spike与阶段〇生产基建-plan.md(A1 续修 spike 已坐实、机制回归绿)	创始人 2026-07-02 拍板三项——① 阶段一拆两个 plan,本 plan = 阶段一①「护城河 middleware 线」(tier2 路先行、端到端验证);cheap-worker 归并 Service = 阶段一②(本 plan SDD 执行完再写)。② 九门 MCP 化 defer(run_gates 现为 FunctionTool、agent 自查已够,放行权威在 middleware 独立重跑,不在本 plan)。③ 软预算改软停交尽力产物(不再 fail-closed 断链)。

配置控制面 · 阶段一① · 护城河 middleware 线(tier2 路)· 实施计划

For agentic workers: REQUIRED SUB-SKILL:用 superpowers:subagent-driven-development(荐)或 superpowers:executing-plans 逐任务执行。步骤用 - [ ] 复选框跟踪。每个改代码的步骤都给了完整代码,不留占位。

Goal: 把 A1 已坐实的续修机制从 spike 提升为生产 middleware 并接真门(finish 点独立重跑 run_gates),把软预算从 fail-closed 断链改为优雅收尾软停,并让 tier2 生成从「control_plane 外层 resume 循环」收敛为「单次 POST + 洋葱内续修」——护城河三件(续修 / 软预算 / 门判)全落进 AgentScope 洋葱、在 tier2 路端到端验证,为阶段一② cheap 归并复用同一套 middleware 铺好地基。

Architecture: 四个任务、单链依赖。T1 统一门判据(gate_judge.py:把 tier2 verdict 归一成 GateJudgment,纯函数、零依赖、先行)是续修 check 的判据契约。T2 续修 middleware(RepairMiddleware 从 A1 spike 提进 worker/middleware.py,fake check 换成返回 GateJudgment 的真契约 + budget_exhausted 实测预算联动 + repairs>0 保护)。T3 软预算软停(CircuitBreakerMiddleware 加 soft_budget 档位:soft 档 ¥ 闸从 _trip 抛熔断改软停放行 + 超预算提醒,hard 档/cheap 保持 fail-closed;四道 on_reply 熔断不动)。T4 服务集成(Service 工厂注入续修 middleware + soft_budget + 续修模式放大超时、drive_generation 加 single_post 单回合并接收 ended + read_last_verdict 判落库、老外层 resume 留 fallback)。T1–T3 本地单测全绿;T4 本地测注入与分支逻辑、真端到端排 mini-desktop 窗口(要 chrome/esbuild)。

Tech Stack: AgentScope 2.0.2(MiddlewareBase 洋葱链、on_reasoning/on_model_call/on_system_prompt 钩子)· Python 3.12(cheap-worker/.venv,已装 agentscope 2.0.2)· pytest(tier2 侧单测型)· 现有 worker/middleware.py(CircuitBreakerMiddleware)/worker/run.py(run_gates/verdict_feedback)/service/app.py(create_app 工厂注入)/service/control_plane.py(drive_generation)。

Global Constraints(全局约束 · 每个任务隐含继承)

• AgentScope 版本钉死 == 2.0.2:读的=跑的;任何 API 疑点以已装 cheap-worker/.venv 的 2.0.2 源码为准,不凭转述。A1 spike 已对 _agent.py 的 finish 点(:857 先存 context、:875 yield finish Msg、:614-620 reply 循环收尾、:2304 _check_next_action 见 UserMsg 返 reasoning)逐条核实,续修机制建立在这些行为上。
• 门放行唯一权威 = middleware 独立重跑:续修放行与否,唯一权威是 middleware 在 finish 点自己独立跑的 run_gates,绝不采信 agent 上报的门工具结果(防便宜档 M3 漏调门 / 谎报门绿的链路假绿)。
• 软预算 = 优雅收尾、不断链(创始人 2026-07-02):¥ 预算越限时交付当前尽力产物,不 fail-closed 断链。真失控保护(死圈 / 超时 / 步数飞车 / 模型调用次数飞车)仍由 on_reply 四道熔断 fail-closed 兜住——只改 ¥ 闸这一道。
• run_gates 是同步 subprocess、~300s、要 chrome+esbuild:在 async 的 on_reasoning 里调它,必须 asyncio.to_thread 包,否则阻塞事件循环。6c6g 无 chrome/esbuild → run_gates 返回结构化诚实失败(rc=1, verdict=None),故本地单测一律 mock check、绝不真跑 run_gates;真门只在 mini-desktop 窗口验。
• middleware.py 两档共用一份:worker/middleware.py 的 CircuitBreakerMiddleware 被 tier2 与 cheap 两档共用(cheap 经 _bootstrap 复用);本 plan 对它的软停改造两档同时受益,改动须保持对 cheap 现有行为兼容(cheap 阶段一②才归并,本 plan 不碰 cheap 代码,但不能改坏共用类的构造/复用语义)。
• 中文注释 + 可追溯日志:所有代码带完整简体中文注释;续修放行 / 续修注入 / 软预算软停 / 门判定这些外部行为与错误路径必须有可追溯日志(创始人铁律)。
• 测试约定:tier2 侧 pytest 型,跑法 PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/<t>.py -v;测试文件放 tier2/gen-worker/tests/,def test_* + assert,mock/直构状态、零网络/LLM/chrome。
• 最小改动 + 复用优先:软停复用现有 on_system_prompt 软刹范式;判据复用现有 verdict_feedback;续修复用 A1 已绿的 stub 脚手架。不顺手重构无关代码。
• 提交风格:feat(<scope>): <中文> (切片一 阶段一①/TN);结尾 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>。仅创始人要求时才 commit/push。
• 落档散文标准:本 plan 散文段(Goal/rationale/风险)按资深工程师散文写——无元叙述、无 AI 造词黑话、无套话空强调。

---

执行顺序与依赖

graph LR
  T1[T1 · 统一门判据 gate_judge<br/>纯函数·先行] --> T2[T2 · 续修 RepairMiddleware<br/>提进 middleware.py·接真契约]
  T3[T3 · 软预算软停<br/>CircuitBreaker ¥闸改软停] --> T4
  T2 --> T4[T4 · 服务集成<br/>工厂注入续修 + control_plane single_post]
  T1 --> T4

• T1 先行:GateJudgment 契约是 T2 续修 check 的返回类型、T4 服务 check 的判据,纯函数零依赖、可独立测。
• T2 依赖 T1:续修 check 返回 GateJudgment(鸭子接口 .passed/.failed_gates/.feedback);T2 实现本体不 import gate_judge(解耦),但 T2 单测要 import GateJudgment 构造 check 返回值。
• T3 与 T1/T2 独立:改 CircuitBreakerMiddleware,不碰 RepairMiddleware;但 T2/T3 都改同一文件 worker/middleware.py,SDD 串行执行(T1→T2→T3→T4)避免同文件冲突。
• T4 收口:依赖 T1(判据)+ T2(续修 middleware)+ T3(软停,供 budget_exhausted 联动);真端到端排 mini-desktop 窗口。

---

Task 1:统一门判据契约 + tier2 适配器(gate_judge.py)

把 tier2 run_gates 的 verdict 归一成一个小契约 GateJudgment(passed, failed_gates, feedback),让续修 middleware 的 check 只依赖这个契约、不依赖 verdict 的具体形状。阶段一② cheap 归并时在同一文件加 judge_cheap_verdict(把 guards map 归一到同契约),续修逻辑两档共用。放行判据 decision==accept 且 L1.passed 与现有 control_plane.py:395 逐字一致,不放松。

Files:

• Create:tier2/gen-worker/worker/gate_judge.py
• Test:tier2/gen-worker/tests/test_gate_judge.py

Interfaces:

• Consumes:worker.run.verdict_feedback(verdict: dict | None, harness_log: str) -> str(现有,run.py:921;把失败 verdict 摘成给 agent 的中文回喂)。

• Produces:

  • GateJudgment(NamedTuple):字段 passed: bool / failed_gates: list[str] / feedback: str。
  • judge_tier2_verdict(gate_result: dict) -> GateJudgment:入参 = run_gates 的返回 {rc, verdict, log};出参 = 归一判据。

• Step 1:写失败测试(tier2 verdict → GateJudgment,绿/红/harness失败三态)

Create tier2/gen-worker/tests/test_gate_judge.py:

"""gate_judge 单测:tier2 run_gates 的 verdict 归一成 GateJudgment(绿 / 红 / harness 未产出 三态)。
跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_gate_judge.py -v
"""
from worker.gate_judge import GateJudgment, judge_tier2_verdict


def test_green_verdict_passes():
    # decision=accept 且 L1.passed → passed=True、无失败门。
    gate_result = {"rc": 0, "log": "", "verdict": {
        "decision": "accept",
        "layerResults": {"L1": {"passed": True, "gateResults": [
            {"gate": "F_wiring", "passed": True, "fatal": True, "detail": ""},
        ]}},
    }}
    j = judge_tier2_verdict(gate_result)
    assert isinstance(j, GateJudgment)
    assert j.passed is True
    assert j.failed_gates == []


def test_red_verdict_lists_failed_gates_and_feedback():
    # 有未过致命门 → passed=False、失败门入列、feedback 非空(verdict_feedback 摘出致命门)。
    gate_result = {"rc": 0, "log": "", "verdict": {
        "decision": "accept",
        "layerResults": {"L1": {"passed": False, "gateResults": [
            {"gate": "F_wiring", "passed": True, "fatal": True, "detail": ""},
            {"gate": "E_live", "passed": False, "fatal": True, "detail": "引擎无实时帧"},
        ]}},
    }}
    j = judge_tier2_verdict(gate_result)
    assert j.passed is False
    assert "E_live" in j.failed_gates
    assert "F_wiring" not in j.failed_gates
    assert j.feedback and isinstance(j.feedback, str)


def test_decision_not_accept_is_not_passed():
    # L1.passed 为真但 decision 非 accept(如 reject)→ 仍判未过(与 control_plane 判据一致)。
    gate_result = {"rc": 0, "log": "", "verdict": {
        "decision": "reject",
        "layerResults": {"L1": {"passed": True, "gateResults": []}},
    }}
    j = judge_tier2_verdict(gate_result)
    assert j.passed is False


def test_missing_verdict_is_not_passed():
    # harness 未产出 verdict(装载即崩 / 脚本未就位)→ passed=False、失败门空、feedback 落 harness 说明。
    gate_result = {"rc": 1, "log": "serve-and-play 编排脚本未就位", "verdict": None}
    j = judge_tier2_verdict(gate_result)
    assert j.passed is False
    assert j.failed_gates == []
    assert "未产出" in j.feedback  # verdict_feedback 的 None 分支输出「真玩未产出 verdict…」


def test_rich_gates_and_findings_enter_failed_list():
    # 富游戏门未过 + P0 finding → failed_gates 用 richGame:/finding: 前缀收进(供 trace,不漏富游戏门/findings)。
    gate_result = {"rc": 0, "log": "", "verdict": {
        "decision": "accept",
        "layerResults": {"L1": {"passed": False, "gateResults": [],
            "richGameGates": {"economy": {"passed": False, "checks": []},
                              "tripleLink": {"passed": True}}}},
        "findings": [{"severity": "P0", "issue": "卡死"}, {"severity": "P2", "issue": "小瑕疵"}],
    }}
    j = judge_tier2_verdict(gate_result)
    assert j.passed is False
    assert "richGame:economy" in j.failed_gates
    assert "richGame:tripleLink" not in j.failed_gates
    assert "finding:P0" in j.failed_gates
    assert "finding:P2" not in j.failed_gates

• Step 2:跑测试,确认失败

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_gate_judge.py -v Expected:FAIL(ModuleNotFoundError: No module named 'worker.gate_judge')

• Step 3:写实现

Create tier2/gen-worker/worker/gate_judge.py:

"""统一门判据契约 + 各档 verdict 适配器(配置控制面 §3.4「两门线归一」)。

续修 middleware 的 check 只认这个契约,不认 verdict 的具体形状,故两档能共用一套续修逻辑:
  · tier2 线:run.run_gates → {rc, verdict{layerResults.L1.gateResults(list)/decision}, log} → judge_tier2_verdict
  · cheap 线:cheap_run.play → {verdict{guards(map)}} → judge_cheap_verdict(阶段一② cheap 归并时补,本 plan 不含)
放行权威唯一来源 = middleware 在 finish 点独立重跑门后经本模块判,绝不采信 agent 上报(设计 §3.4 铁律)。
"""
from __future__ import annotations

from typing import NamedTuple


class GateJudgment(NamedTuple):
    """一次门判定的归一结果。

    passed: 是否全绿放行(tier2 = decision==accept 且 L1.passed;判据不放松)。
    failed_gates: 未过门名列表(L1 九门 + richGame:<门> + finding:<severity>;供 trace/日志,放行只看 passed)。
    feedback: 给 agent 的续修回喂文字(门没绿时注入,已是人读中文)。
    """

    passed: bool
    failed_gates: list[str]
    feedback: str


def judge_tier2_verdict(gate_result: dict) -> GateJudgment:
    """把 tier2 run_gates 的返回({rc, verdict, log})归一成 GateJudgment。

    tier2 verdict 形状:顶层 decision + layerResults.L1.{passed, gateResults(list[{gate,passed,fatal,detail}])}。
    放行判据 = decision==accept 且 L1.passed(与 control_plane.py:395 现有判据逐字一致,绝不放松)。
    verdict 缺失(harness 未产出 / 装载即崩)→ 判未过,feedback 落 harness 日志摘要(verdict_feedback 兜)。
    """
    # 惰性 import:run 顶层经 worker.config 牵 agentscope,保 6c6g 仅 import gate_judge(不调本函数)不炸。
    from worker.run import verdict_feedback  # noqa: PLC0415

    gate_result = gate_result or {}
    v = gate_result.get("verdict") or {}
    l1 = ((v.get("layerResults") or {}).get("L1") or {})
    passed = v.get("decision") == "accept" and bool(l1.get("passed"))
    # 未过门收全:L1 九门 + 富游戏三门(richGame:<门>)+ P0/P1 对抗发现(finding:<severity>),
    #   与 verdict_feedback 摘要口径对齐(供 [repair] 日志/trace,不漏富游戏门与 findings;评审 M-1)。
    failed_gates = [
        g.get("gate", "?")
        for g in (l1.get("gateResults") or [])
        if not g.get("passed")
    ]
    rich = l1.get("richGameGates") or {}
    for _name in ("tripleLink", "economy", "latch"):
        if isinstance(rich.get(_name), dict) and rich[_name].get("passed") is False:
            failed_gates.append(f"richGame:{_name}")
    for _f in (v.get("findings") or []):
        if _f.get("severity") in ("P0", "P1"):
            failed_gates.append(f"finding:{_f.get('severity')}")
    feedback = verdict_feedback(v, gate_result.get("log") or "")
    return GateJudgment(passed=passed, failed_gates=failed_gates, feedback=feedback)

• Step 4:跑测试,确认通过

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_gate_judge.py -v Expected:PASS(5 passed)

• Step 5:提交

git add tier2/gen-worker/worker/gate_judge.py tier2/gen-worker/tests/test_gate_judge.py
git commit -m "feat(tier2): 统一门判据 GateJudgment + tier2 verdict 适配器 (切片一 阶段一①/T1)"

---

Task 2:续修 RepairMiddleware 提升进 middleware.py + 接真契约

把 A1 spike 里 inline 的 RepairMiddleware 提升为 worker/middleware.py 的生产类,作 CircuitBreakerMiddleware 的兄弟。三处改动:① check 契约从返回 (passed, feedback) 二元组改为返回 GateJudgment(带 failed_gates 供 trace);② 新增 budget_exhausted: () -> bool 回调,软预算耗尽时放行 finish 不再续修(与 T3 联动);③ pop_finish_claim 默认改 True(生产要干净血缘,续修前弹掉「做完了」声称)。机制本体(拦 finish → 独立跑 check → 压制 + observe 注入 → 续跑)与 A1 逐字保留、已被 A1 四断言坐实。

Files:

• Modify:tier2/gen-worker/worker/middleware.py(新增 RepairMiddleware 类;补 Msg/UserMsg import)
• Test:tier2/gen-worker/tests/test_repair_middleware.py(从 A1 spike 提升:复用 stub 脚手架 + check 返回 GateJudgment + 新增 budget_exhausted 断言)

Interfaces:

• Consumes:

  • MiddlewareBase.on_reasoning(self, agent, input_kwargs, next_handler) -> AsyncGenerator(2.0.2 洋葱钩子)。
  • agent.observe(UserMsg) / agent.state.context: list[Msg] / agentscope.message.Msg, UserMsg。
  • check 返回值的鸭子接口 .passed: bool / .failed_gates: list / .feedback: str(实例 = T1 GateJudgment;middleware 本体不 import gate_judge,鸭子解耦)。

• Produces:

  • RepairMiddleware(check, max_repairs=3, *, pop_finish_claim=True, budget_exhausted=None)。
  • check 契约 = async (agent) -> GateJudgment;budget_exhausted 契约 = () -> bool(可选,True 则放行 finish 不续修)。
  • 实例属性 repairs: int(续修次数,供 trace/断言)、last_judgment(最近一次门判定)。

• Step 1:写失败测试(复用 A1 stub、check 返回 GateJudgment、新增 budget_exhausted)

Create tier2/gen-worker/tests/test_repair_middleware.py(stub 模型脚手架逐字复用 cheap-worker/tests/test_repair_middleware_spike.py,已随 A1 绿;仅 check 改返回 GateJudgment、新增 budget_exhausted 断言):

"""RepairMiddleware(提升进 worker/middleware.py 后)单测:续修机制 + GateJudgment 契约 + budget_exhausted 放行。
stub 模型脚手架复用 A1 spike(cheap-worker/tests/test_repair_middleware_spike.py),已坐实机制;本测验提升后的生产类。
跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_repair_middleware.py -v
"""
import asyncio
from typing import Any, AsyncGenerator, Type

from agentscope.agent import Agent, ReActConfig
from agentscope.model import ChatModelBase, ChatResponse
from agentscope.message import Msg, TextBlock, UserMsg
from agentscope.tool import Toolkit
from agentscope.credential import CredentialBase
from pydantic import BaseModel

from worker.middleware import RepairMiddleware
from worker.gate_judge import GateJudgment


# ── stub 模型(确定性,不打真 LLM;逐字采 A1 spike 的 2.0.2 verbatim 范式)──
class MockCredential(CredentialBase):
    @classmethod
    def get_chat_model_class(cls) -> Type["ChatModelBase"]:
        return MockModel


class MockModel(ChatModelBase):
    class Parameters(BaseModel):
        pass

    def __init__(self, model: str = "mock-model", stream: bool = True,
                 context_size: int = 1000, mock_chat_responses: list | None = None) -> None:
        super().__init__(credential=MockCredential(), model=model, stream=stream,
                         parameters=MockModel.Parameters(), context_size=context_size)
        self.mock_chat_responses = mock_chat_responses or []
        self.cnt = 0

    def set_responses(self, mock_responses: list) -> None:
        self.mock_chat_responses = mock_responses
        self.stream = not all(isinstance(_, ChatResponse) for _ in mock_responses)
        self.cnt = 0

    async def _call_api(self, *args: Any, **kwargs: Any):
        mock_responses = self.mock_chat_responses[self.cnt]
        self.cnt += 1
        if isinstance(mock_responses, list):
            async def _stream() -> AsyncGenerator[ChatResponse, None]:
                for response in mock_responses:
                    yield response
            return _stream()
        if isinstance(mock_responses, ChatResponse):
            return mock_responses
        raise AssertionError


def _text(m) -> str:
    try:
        return m.get_text_content() or ""
    except Exception:  # noqa: BLE001
        return ""


def _agent(model, mw, max_iters=5) -> Agent:
    return Agent(name="tier2_writer", system_prompt="你是 tier2 富游戏生成 agent",
                 model=model, toolkit=Toolkit(), middlewares=[mw],
                 react_config=ReActConfig(max_iters=max_iters))


def test_intercept_inject_and_continue():
    # 模型两次都想 finish;门第一次没过、第二次过 → 门被独立跑两次(压制一次)、放行第二版、注入消息在 context。
    calls = {"n": 0}

    async def check(agent) -> GateJudgment:
        calls["n"] += 1
        if calls["n"] >= 2:
            return GateJudgment(passed=True, failed_gates=[], feedback="")
        return GateJudgment(passed=False, failed_gates=["H_progress"], feedback="H_progress 门未过(无进展)")

    model = MockModel()
    model.set_responses([
        ChatResponse(content=[TextBlock(text="第一版游戏,做完了")], is_last=True),
        ChatResponse(content=[TextBlock(text="已修复进展问题,再交付")], is_last=True),
    ])
    mw = RepairMiddleware(check, max_repairs=3)
    final = asyncio.run(_agent(model, mw).reply(UserMsg(name="user", content="生成一个经营游戏")))
    assert calls["n"] == 2, "门应被独立跑两次(压制一次)"
    assert mw.repairs == 1, "应续修一次"
    assert "已修复" in _text(final), "放行的应是第二版"


def test_intercept_injects_user_repair_message():
    # 单独验注入:门恒不过一轮后放行,context 里有一条 role=user 的续修消息(含 feedback)。
    calls = {"n": 0}

    async def check(agent) -> GateJudgment:
        calls["n"] += 1
        return GateJudgment(passed=calls["n"] >= 2, failed_gates=["E_live"], feedback="E_live 门未过")

    model = MockModel()
    model.set_responses([
        ChatResponse(content=[TextBlock(text="一版")], is_last=True),
        ChatResponse(content=[TextBlock(text="二版")], is_last=True),
    ])
    mw = RepairMiddleware(check, max_repairs=3)
    agent = _agent(model, mw)
    asyncio.run(agent.reply(UserMsg(name="user", content="生成")))
    assert any(isinstance(m, Msg) and getattr(m, "role", None) == "user"
               and "E_live" in _text(m) for m in agent.state.context)


def test_max_repairs_cap_releases():
    # 门恒不过、max_repairs=2 → 续修 2 次后放行、不无限跑。
    async def check(agent) -> GateJudgment:
        return GateJudgment(passed=False, failed_gates=["x"], feedback="恒不过")

    model = MockModel()
    model.set_responses([ChatResponse(content=[TextBlock(text=f"第{i}版")], is_last=True) for i in range(6)])
    mw = RepairMiddleware(check, max_repairs=2)
    final = asyncio.run(_agent(model, mw, max_iters=10).reply(UserMsg(name="user", content="生成")))
    assert mw.repairs == 2
    assert _text(final)


def test_budget_exhausted_still_repairs_once_then_releases():
    # 软预算耗尽 + 门恒没绿:首个 finish 仍先续修一次(repairs>0 保护),第二个 finish 才因预算放行(交尽力产物)。
    async def check(agent) -> GateJudgment:
        return GateJudgment(passed=False, failed_gates=["E_live"], feedback="门没绿")

    model = MockModel()
    model.set_responses([
        ChatResponse(content=[TextBlock(text="一版没绿")], is_last=True),
        ChatResponse(content=[TextBlock(text="二版尽力产物,交付")], is_last=True),
        ChatResponse(content=[TextBlock(text="不该到这三版")], is_last=True),
    ])
    mw = RepairMiddleware(check, max_repairs=5, budget_exhausted=lambda: True)
    final = asyncio.run(_agent(model, mw).reply(UserMsg(name="user", content="生成")))
    assert mw.repairs == 1, "首个未绿 finish 必先续修一次(预算不旁路续修),之后才因预算放行"
    assert "尽力产物" in _text(final)


def test_pop_finish_claim_default_clears_stale_claim():
    # 默认 pop_finish_claim=True → 续修前弹掉第一版「做完了」声称,context 不残留。
    calls = {"n": 0}

    async def check(agent) -> GateJudgment:
        calls["n"] += 1
        return GateJudgment(passed=calls["n"] >= 2, failed_gates=["x"], feedback="修")

    model = MockModel()
    model.set_responses([
        ChatResponse(content=[TextBlock(text="UNIQUE_CLAIM_一版做完了")], is_last=True),
        ChatResponse(content=[TextBlock(text="二版交付")], is_last=True),
    ])
    mw = RepairMiddleware(check, max_repairs=3)  # 默认 pop_finish_claim=True
    agent = _agent(model, mw)
    asyncio.run(agent.reply(UserMsg(name="user", content="生成")))
    assert not any("UNIQUE_CLAIM" in _text(m) for m in agent.state.context)

说明:注入断言由独立用例 test_intercept_injects_user_repair_message 承担(持 agent 引用直接读 agent.state.context);test_intercept_inject_and_continue 只验「门跑两次 + 续修一次 + 放行第二版」三点,职责单一。

• Step 2:跑测试,确认失败

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_repair_middleware.py -v Expected:FAIL(ImportError: cannot import name 'RepairMiddleware' from 'worker.middleware')

• Step 3:在 middleware.py 补 import 并新增 RepairMiddleware 类

先确认 tier2/gen-worker/worker/middleware.py 顶部已 import Msg/UserMsg;若无则补(与现有 agentscope import 同段):

from agentscope.message import Msg, UserMsg  # 续修 middleware:拦 finish Msg、observe 注入 UserMsg

在 CircuitBreakerMiddleware 类之后(文件内 middleware 类集中处)新增:

class RepairMiddleware(MiddlewareBase):
    """on_reasoning 拦 finish → 独立重跑门 → 没绿则压制 finish + 注入续跑(护城河续修,配置控制面 §3.4)。

    取代 control_plane 外层 resume 循环:整局在单次 POST 内于 finish 点拦截续修,不再每轮重开 POST。
    放行权威唯一 = 本 middleware 在 finish 点独立跑的 check(重跑 run_gates 经 gate_judge 归一),
    绝不采信 agent 上报的门工具结果(防漏调门 / 谎报门绿的链路假绿)。

    机制(A1 spike 对已装 2.0.2 _agent.py 逐条坐实):
      finish = _reasoning_impl 无 tool call 时 yield 纯文本 Msg(:875);它穿过 on_reasoning 洋葱链(:745)
      才到 reply 循环(:614-620),故本 middleware 能在它到达前吞掉(不 yield)。吞掉后本轮无 Msg →
      reply 循环不走 :614 return → cur_iter++(:670)→ while 未封顶再进推理;注入走 observe(UserMsg)使
      context[-1] 为 role=user → _get_last_msg 返 None(:2263)→ _check_next_action 强制 reasoning(:2304)。
    坑(已处理):finish 文本在 :857 先存进 context,压制 yield 抹不掉 → pop_finish_claim=True 时 context.pop() 清掉。

    check 契约:async (agent) -> GateJudgment(passed, failed_gates, feedback)。
    budget_exhausted 契约:() -> bool(可选,读实测已花 ¥);True 且已至少续修过一次(repairs>0)才因预算放行 finish
    (首个未绿 finish 一定先修一次,护城河续修不被软预算旁路;判据用实测已花 ¥、非预估)。
    """

    def __init__(self, check, max_repairs: int = 3, *, pop_finish_claim: bool = True,
                 budget_exhausted=None) -> None:
        """
        Args:
            check: async (agent) -> GateJudgment;finish 点独立重跑门的判据(tier2 = 重跑 run_gates 经 gate_judge)。
            max_repairs: 续修硬顶(防无限续;另有 ReActConfig.max_iters 兜底)。
            pop_finish_claim: 续修前是否把 agent 刚存进 context 的「做完了」文本弹掉(干净血缘;生产默认 True)。
            budget_exhausted: 可选 () -> bool(读实测已花 ¥ 是否超上限);True 且 repairs>0 才因预算放行 finish
                (首个未绿 finish 必先修一次;软预算不旁路续修,与 CircuitBreaker ¥ 闸联动)。
        """
        self._check = check
        self._max_repairs = max_repairs
        self._pop_finish_claim = pop_finish_claim
        self._budget_exhausted = budget_exhausted
        self.repairs = 0            # 续修次数(供 trace/断言/收口读)
        self.last_judgment = None   # 最近一次门判定(GateJudgment;供 trace/收口读)

    async def on_reasoning(self, agent, input_kwargs, next_handler):
        async for item in next_handler(**input_kwargs):
            if not isinstance(item, Msg):
                yield item  # 非 finish 的事件流(ModelCallStart/Text* 等)原样透传
                continue
            # —— 拦到 finish(纯文本 Msg):独立重跑门判放行 ——
            judgment = await self._check(agent)
            self.last_judgment = judgment
            # 软预算耗尽放行加「至少续修一次」保护(repairs>0):首个门没绿的 finish 一定先修一次,
            #   护城河续修不被软预算旁路(评审 I1、创始人 2026-07-02 拍板:至少修一次再因预算放行)。
            budget_out = bool(self.repairs > 0 and self._budget_exhausted and self._budget_exhausted())
            if judgment.passed or self.repairs >= self._max_repairs or budget_out:
                reason = ("门绿" if judgment.passed
                          else ("续修达上限" if self.repairs >= self._max_repairs else "软预算耗尽"))
                # 放行 finish:门绿 / 续修耗尽 / 软预算耗尽(后两者交尽力产物、不断链)。
                print(f"[repair] 放行 finish(repairs={self.repairs} 原因={reason} "
                      f"未过门={judgment.failed_gates})", flush=True)
                yield item
                return
            # —— 门没绿 + 有续修预算:压制 finish + 注入续跑 ——
            self.repairs += 1
            if self._pop_finish_claim and agent.state.context and isinstance(agent.state.context[-1], Msg):
                agent.state.context.pop()  # 弹掉刚存进的「做完了」声称(_agent.py:857 先存;干净血缘)
            print(f"[repair] 门未绿续修 #{self.repairs}:未过门={judgment.failed_gates}", flush=True)
            # 注入「修」:必须 role=user 纯文本(system/tool/thinking 会让 _handle_incoming_messages 抛 ValueError)。
            await agent.observe(UserMsg(
                name="gate",
                content=f"以下验收门未通过,请据反馈修复后再交付,不要直接结束:\n{judgment.feedback}"))
            return  # 吞掉 finish(不 yield)→ 本轮无 Msg → reply 循环 cur_iter++ 续跑

• Step 4:跑测试,确认通过

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_repair_middleware.py -v Expected:PASS(5 passed —— intercept/注入/封顶/预算放行/pop 血缘)

• Step 5:回归 A1 spike(确认提升未破坏原机制)

Run:cheap-worker/.venv/bin/python cheap-worker/tests/test_repair_middleware_spike.py Expected:4/4 passed(A1 spike 仍绿;提升是加法,不改 spike 文件)

• Step 6:提交

git add tier2/gen-worker/worker/middleware.py tier2/gen-worker/tests/test_repair_middleware.py
git commit -m "feat(tier2): 续修 RepairMiddleware 提升进 middleware.py + GateJudgment 真契约 + budget_exhausted 联动 (切片一 阶段一①/T2)"

---

Task 3:软预算 ¥ 闸从 fail-closed 改优雅收尾软停

CircuitBreakerMiddleware 的 on_model_call ¥ 累进硬闸现状是越限即 _trip 抛熔断(fail-closed 断链)。设计 §3.4 要「超目标优雅收尾、不断链」。改法:越限时设 budget_soft_tripped 标记 + 放行本次调用,并在软刹 on_system_prompt 里加一条「预算已达上限、立即 finish」的强提醒——agent 收到后基于当前工程状态尽快 finish,续修 middleware 经 budget_exhausted 见预算耗尽即放行(T2 已实现)。四道 on_reply 熔断(步数 / 模型调用次数飞车 / 死圈 / 超时)保持 fail-closed 不动:¥ 预算越限是「花够了、该交付」,而次数飞车 / 死圈 / 超时是「真失控」,后者仍须硬熔断。

Files:

• Modify:tier2/gen-worker/worker/middleware.py(CircuitBreakerMiddleware.__init__ 加 soft_budget 档位参数 / reset 加 budget_soft_tripped 初始化 / on_model_call ¥ 闸按档位软停或 fail-closed / _accumulate_call_rmb 后实测越限也置软停 / on_system_prompt 加超预算提醒)
• Test:tier2/gen-worker/tests/test_budget_soft_stop.py(tier2 soft 档软停 + hard 档仍 fail-closed 两口径都锁)
• 回归(不改、Step 7 须跑绿):cheap-worker/tests/test_budget_gate.py(cheap 默认 hard、越 ¥10 仍抛 Tier2CircuitBreak——验证共用类改动未误伤 cheap 硬地板,评审 C2/C-1)

Interfaces:

• Consumes:CircuitBreakerMiddleware(现有,middleware.py:172)的 __init__/on_model_call(:343)/on_system_prompt(:265)/reset(:249)/_accumulate_call_rmb(:456)/spent_rmb/rmb_hard_limit/_rmb_gate_active。

• Produces:

  • 构造参数 soft_budget: bool = False(默认 hard=保持现状 fail-closed;只 tier2 工厂显式传 True,cheap 保持 hard——共用类不误伤 cheap 硬地板)。
  • 实例属性 budget_soft_tripped: bool(soft 档下 ¥ 越限标记;供 on_system_prompt 提醒 + T4 budget_exhausted 参考)。
  • 契约变更(仅 soft 档):soft_budget=True 时 on_model_call 越 ¥ 上限不抛熔断、改软停放行;soft_budget=False(cheap 及默认)保持抛 Tier2CircuitBreak(kind="budget") fail-closed 不变。

• Step 1:写失败测试(soft 档软停放行 / hard 档仍抛熔断 / 默认 hard / 实测越限置位 / 软刹提醒 / reset 清 / on_reply 熔断回归)

Create tier2/gen-worker/tests/test_budget_soft_stop.py:

"""软预算 ¥ 闸档位单测:soft 档越限软停放行 + 强提醒;hard 档(cheap 及默认)仍 fail-closed 抛熔断。
跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_budget_soft_stop.py -v
"""
import asyncio

import pytest

from worker.middleware import CircuitBreakerMiddleware, Tier2CircuitBreak


class _FakeModel:
    model = "MiniMax-M3"


class _FakeResponse:
    usage = None  # 无 usage → _accumulate_call_rmb 跳过,不干扰软停断言


async def _passthrough(**kwargs):
    return _FakeResponse()


def _breaker_over_budget(*, soft_budget: bool) -> CircuitBreakerMiddleware:
    # 直构:金额闸激活 + 已花超上限,下次调用必越限(pricing_params 传入 → 不打网关取价)。
    b = CircuitBreakerMiddleware(rmb_hard_limit=1.0, soft_budget=soft_budget,
                                 pricing_params={"pricing": {}, "qpu": 1, "usd_rate": 0})
    b._rmb_gate_active = True
    b.spent_rmb = 2.0  # 已花 ¥2 > 上限 ¥1;预估按 0(无 pricing)仍越限
    return b


def test_soft_over_budget_does_not_raise_and_passes_through():
    # soft 档:越限放行本次调用、设软停标记、不抛熔断。
    b = _breaker_over_budget(soft_budget=True)
    result = asyncio.run(b.on_model_call(None, {"current_model": _FakeModel()}, _passthrough))
    assert isinstance(result, _FakeResponse), "soft 档越限应放行本次调用、不抛熔断"
    assert b.budget_soft_tripped is True, "soft 档越限应设软停标记"


def test_hard_mode_still_fail_closed():
    # hard 档(默认,cheap 用):越限仍抛 Tier2CircuitBreak(共用类改动不误伤 cheap 硬地板)。
    b = _breaker_over_budget(soft_budget=False)
    with pytest.raises(Tier2CircuitBreak):
        asyncio.run(b.on_model_call(None, {"current_model": _FakeModel()}, _passthrough))


def test_default_is_hard():
    # 不传 soft_budget → 默认 hard(保护 cheap:共用类默认行为不变)。
    b = CircuitBreakerMiddleware(rmb_hard_limit=1.0,
                                 pricing_params={"pricing": {}, "qpu": 1, "usd_rate": 0})
    b._rmb_gate_active = True
    b.spent_rmb = 2.0
    with pytest.raises(Tier2CircuitBreak):
        asyncio.run(b.on_model_call(None, {"current_model": _FakeModel()}, _passthrough))


def test_under_budget_no_soft_trip():
    # 未越限:正常放行、不设软停标记。
    b = CircuitBreakerMiddleware(rmb_hard_limit=100.0, soft_budget=True,
                                 pricing_params={"pricing": {}, "qpu": 1, "usd_rate": 0})
    b._rmb_gate_active = True
    b.spent_rmb = 0.0
    result = asyncio.run(b.on_model_call(None, {"current_model": _FakeModel()}, _passthrough))
    assert isinstance(result, _FakeResponse)
    assert b.budget_soft_tripped is False


def test_soft_trip_set_on_measured_overspend():
    # soft 档:即便调用前预估未越限,调用后实测折算使 spent 超上限 → 也置软停(评审 I-1 实测置位)。
    b = CircuitBreakerMiddleware(rmb_hard_limit=1.0, soft_budget=True,
                                 pricing_params={"pricing": {}, "qpu": 1, "usd_rate": 0})
    b._rmb_gate_active = True
    b.spent_rmb = 2.0  # 模拟实测折算后已越上限
    b._maybe_soft_trip_on_measured()  # _accumulate_call_rmb 尾部调它
    assert b.budget_soft_tripped is True


def test_soft_trip_injects_finish_reminder():
    b = _breaker_over_budget(soft_budget=True)
    b.budget_soft_tripped = True
    out = asyncio.run(b.on_system_prompt(None, "原始系统提示"))
    assert "原始系统提示" in out
    assert "finish" in out and "预算" in out


def test_reset_clears_soft_trip():
    b = _breaker_over_budget(soft_budget=True)
    b.budget_soft_tripped = True
    b.reset()
    assert b.budget_soft_tripped is False


def test_on_reply_circuit_still_fail_closed():
    # 回归护栏:四道 on_reply 熔断仍 fail-closed(改软停只碰 ¥ 闸,不碰失控保护)。
    # 触发次数飞车:max_model_calls=1,喂两个 ModelCallStartEvent → 第二个越限抛熔断。
    b = CircuitBreakerMiddleware(max_model_calls=1, enable_rmb_gate=False)

    async def _events(**kwargs):
        for _ in range(2):
            yield type("ModelCallStartEvent", (), {})()

    async def _drive():
        async for _ in b.on_reply(None, {}, _events):
            pass

    with pytest.raises(Tier2CircuitBreak):
        asyncio.run(_drive())

注:test_on_reply_circuit_still_fail_closed 只为锁住「软停改造未误伤四道熔断」这条回归护栏。若 on_reply 的事件类型判定(type(evt).__name__ == "ModelCallStartEvent")对动态构造的类不便断言,实现时改用 worker 已有的真 ModelCallStartEvent 类型构造事件(从 agentscope 事件模块 import),保持「第二次模型调用越 max_model_calls=1 → 抛 Tier2CircuitBreak」的断言不变。

• Step 2:跑测试,确认失败

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_budget_soft_stop.py -v Expected:FAIL(TypeError: __init__() got an unexpected keyword argument 'soft_budget';及无 budget_soft_tripped / _maybe_soft_trip_on_measured)

• Step 3:改 init 加 soft_budget 档位参数 + reset 加 budget_soft_tripped 初始化

在 CircuitBreakerMiddleware.__init__(middleware.py:184-197)的关键字参数末尾加 soft_budget(默认 False=hard):

        enable_rmb_gate: bool = True,
        group_ratio: float | None = None,
        soft_budget: bool = False,   # 新增:¥ 越限行为档位(False=hard fail-closed 保持现状/cheap;True=soft 软停,只 tier2 传)
    ) -> None:

在 __init__ 体内(self.enable_rmb_gate = enable_rmb_gate 一带、self.reset() 之前)存档位:

        # ¥ 越限行为档位:False=hard(越限抛熔断,cheap 及默认,守 ¥ 硬地板);True=soft(越限软停放行,tier2 单 POST 续修路)。
        self.soft_budget = soft_budget

在 reset(middleware.py:249)的 ¥ 闸状态初始化处(self.spent_rmb = 0.0 附近)加:

        # ¥ 软预算越限标记(soft 档:越限设 True + 放行本次,agent 收 on_system_prompt 强提醒后尽快 finish)。
        self.budget_soft_tripped = False

• Step 4:改 on_model_call —— ¥ 闸按档位软停(soft)或 fail-closed(hard);加实测越限置位

把 on_model_call(middleware.py:366-373)里的越限 _trip 分支替换为按档位分流:

        # ② 调用前 ¥ 判(取到活价、金额闸生效时才判;否则降级、由 on_reply 次数闸兜)。
        if self._rmb_gate_active:
            est_rmb = self._estimate_call_rmb(model_name)
            if self.spent_rmb + est_rmb > self.rmb_hard_limit:
                if not self.soft_budget:
                    # hard 档(cheap 及默认):保持现状 fail-closed 抛熔断,守 ¥ 硬地板。
                    self._trip(
                        "budget",
                        f"¥ 累进硬闸越限 fail-closed:已花 ¥{self.spent_rmb:.4f} + 本次预估 ¥{est_rmb:.4f}"
                        f" > 上限 ¥{self.rmb_hard_limit:.2f}(模型={model_name or '未知'})",
                    )
                elif not self.budget_soft_tripped:
                    # soft 档(tier2 单 POST 续修路):越限不断链,设软停标记 + 放行本次调用,让 agent 收
                    #   on_system_prompt 强提醒后基于当前工程状态尽快 finish(交尽力产物)。续修 middleware 经
                    #   budget_exhausted 见实测耗尽 + 已至少续修一次即放行 finish(评审 C2/I1、创始人 2026-07-02)。
                    self.budget_soft_tripped = True
                    print(
                        f"[tier2-circuit] ¥ 软预算越限 → 优雅收尾软停(不断链):已花 ¥{self.spent_rmb:.4f}"
                        f" + 本次预估 ¥{est_rmb:.4f} > 上限 ¥{self.rmb_hard_limit:.2f}"
                        f"(模型={model_name or '未知'});提醒 agent 尽快 finish。",
                        flush=True,
                    )

(删掉原无条件 self._trip("budget", ...);其后 # ③ 调真模型… 起的放行代码不变。)

再在 _accumulate_call_rmb(middleware.py:456)的 self.spent_rmb += float(...) 之后调 self._maybe_soft_trip_on_measured(),并新增该方法(治评审 I-1:预估未触发但实测已超的漏网):

    def _maybe_soft_trip_on_measured(self) -> None:
        """soft 档:调用后实测折算使已花 ¥ 越上限时也置软停(评审 I-1;hard 档不置,仍靠调用前 fail-closed)。"""
        if self.soft_budget and self._rmb_gate_active and not self.budget_soft_tripped \
                and self.spent_rmb > self.rmb_hard_limit:
            self.budget_soft_tripped = True
            print(f"[tier2-circuit] ¥ 软预算实测越限 → 软停(不断链):已花 ¥{self.spent_rmb:.4f}"
                  f" > 上限 ¥{self.rmb_hard_limit:.2f};提醒 agent 尽快 finish。", flush=True)

• Step 5:改 on_system_prompt —— 加超预算强提醒

在 on_system_prompt(middleware.py:265)的 near_step/near_budget 判定之前加超预算分支(软停比软刹更强、不再留探索空间):

    async def on_system_prompt(self, agent, current_prompt: str) -> str:
        """达 soft_ratio×硬顶时软刹提醒;¥ 软预算越限时更强地催 finish(软停,均非强制断链)。"""
        if self.budget_soft_tripped:
            # ¥ 软预算已越限:强提醒立即 finish(比 near_* 软刹更强,不再留探索空间)。
            return (
                current_prompt
                + "\n\n<system-reminder>本次生成的 ¥ 预算已达上限。请立即停止任何新的探索或修改,"
                "基于当前工程的最好状态直接调用 finish 交付源工程。继续消耗预算不会带来更多产出。</system-reminder>"
            )
        near_step = self.tool_calls >= self.soft_ratio * self.max_tool_calls
        near_budget = self.model_calls >= self.soft_ratio * self.max_model_calls
        if near_step or near_budget:
            return (
                current_prompt
                + "\n\n<system-reminder>你已接近本次会话的步数/预算上限。"
                "请停止发散探索,基于当前最好的工程状态尽快收敛:补齐能过 L1 门的最小可玩闭环,"
                "然后调用 finish 交付源工程。再不收敛将触发硬熔断、本次作废。</system-reminder>"
            )
        return current_prompt

• Step 6:跑测试,确认通过

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_budget_soft_stop.py -v Expected:PASS(8 passed)

• Step 7:回归 tier2 + cheap 双档(确认共用类改造未破坏现有熔断、且 cheap 仍 fail-closed)

Run(tier2):PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/ -k "middleware or circuit or budget or repair" -v Expected:全 PASS(现有熔断 / trace / 续修 / 软停用例都绿)

Run(cheap 回归——验证共用类改动未误伤 cheap 硬地板):cheap-worker/.venv/bin/python cheap-worker/tests/test_budget_gate.py Expected:PASS(cheap 默认 hard 档、越 ¥10 仍 fail-closed 抛 Tier2CircuitBreak)

• Step 8:提交

git add tier2/gen-worker/worker/middleware.py tier2/gen-worker/tests/test_budget_soft_stop.py
git commit -m "feat(tier2): 软预算 ¥ 闸从 fail-closed 改优雅收尾软停(交尽力产物·不断链) (切片一 阶段一①/T3)"

---

Task 4:服务集成 —— 工厂注入续修 middleware + control_plane 单 POST 化

把续修 middleware 接进 Service:_tier2_middlewares_factory 在每回合注入 RepairMiddleware(check = 独立重跑 tier2 run_gates 经 to_thread+异常兜底+judge_tier2_verdict,budget_exhausted = 读同回合 breaker 实测已花 ¥ 是否超上限)+ 给 breaker 传 soft_budget=True 与续修模式放大的超时旋钮。extra 注入序 [tracer, repair, breaker](框架另在其外前置 InboxMiddleware——每轮 drain inbox 后透传全部 evt、不吞 finish,故 repair 仍是 on_reasoning 最内层、能拦原始 finish)。消费方 drive_generation 加 single_post 开关:新路只发一次 kick、等这一次(内部续修多轮)回合真结束(接收 _wait_for_turn_end 的 ended)、用 read_last_verdict 读服务端续修已落的 verdict(不重跑门,避免冗余 ~300s + chrome 端口竞态)判落库;老外层 resume 循环整体保留作 fallback。single_post 的附带收益:整局一个 POST = 一个 breaker 实例贯穿续修全程,spent_rmb 在同一实例内自然累计——解决现状「每回合新 breaker、跨回合 ¥ 不累计」的已知缺口(阶段〇 follow-up),无需持久化。真实门判开销:续修在服务端跑 N 次真门、消费端 0 次(只读 verdict);真端到端(要 chrome/esbuild)排 mini-desktop 窗口,本地单测覆盖注入结构与分支逻辑。

Files:

• Modify:tier2/gen-worker/service/app.py(_tier2_middlewares_factory 注入续修 middleware)
• Modify:tier2/gen-worker/service/control_plane.py(drive_generation 加 single_post 参数与单回合分支)
• Test:tier2/gen-worker/tests/test_repair_wiring.py(工厂注入结构 + single_post 分支逻辑,mock)

Interfaces:

• Consumes:RepairMiddleware(T2)、judge_tier2_verdict(T1)、CircuitBreakerMiddleware(soft_budget=, wall_timeout_s=, step_timeout_s=)+.spent_rmb/rmb_hard_limit/_rmb_gate_active(T3)、worker.run.run_gates(现有)、worker.genconfig.get(现有)、_wait_for_turn_end(...) -> {ended, reason, endEvent}(现有,control_plane:144)。

• Produces:

  • worker.run.read_last_verdict(game_id) -> dict | None:读工程 evidence/verdict.json(服务端续修已落),不重跑门。
  • _tier2_middlewares_factory 返回 [Tier2TraceMiddleware, RepairMiddleware, CircuitBreakerMiddleware](extra 注入序 tracer 外、repair 中、breaker 内;框架在其外另前置 InboxMiddleware/StateChange/ToolOffload,repair 仍是 on_reasoning 最内层、能拦原始 finish)。
  • drive_generation(..., single_post: bool = True, ...):single_post=True 单回合 + 洋葱内续修 + 接收 ended + read_last_verdict 判落库;False = 现有外层 resume(fallback)。

• Step 1:写失败测试(工厂注入结构 + single_post 分支)

Create tier2/gen-worker/tests/test_repair_wiring.py:

"""服务集成单测:middleware 工厂注入续修(注入序 + soft_budget + 续修超时)+ drive_generation single_post 分支
(接收 ended + read_last_verdict 判落库,mock、无真 Service)。
跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_repair_wiring.py -v
"""
import asyncio

from worker.middleware import CircuitBreakerMiddleware, RepairMiddleware, Tier2TraceMiddleware


def test_factory_injects_repair_with_soft_budget_and_repair_timeouts():
    # extra 注入序 [tracer, repair, breaker];breaker 走 soft_budget=True + 续修模式放大超时(评审 C1/C2)。
    from service.app import _tier2_middlewares_factory
    mws = asyncio.run(_tier2_middlewares_factory("u1", "a1", "s1"))
    assert [type(m).__name__ for m in mws] == [
        "Tier2TraceMiddleware", "RepairMiddleware", "CircuitBreakerMiddleware"]
    tracer, repair, breaker = mws
    assert isinstance(tracer, Tier2TraceMiddleware)
    assert isinstance(repair, RepairMiddleware)
    assert isinstance(breaker, CircuitBreakerMiddleware)
    assert breaker.soft_budget is True, "tier2 续修路走 soft 档(cheap 仍 hard)"
    assert breaker.wall_timeout_s > 1800, "续修模式墙钟须放大容纳 N 次串行真门"
    assert breaker.step_timeout_s > 300, "续修模式单步静默阈值须 > 单次门判上限"


def test_repair_budget_exhausted_reads_measured_spend():
    # 续修 budget_exhausted 读同回合 breaker 实测已花 ¥ 是否超上限(非预估标记):未超 False、超 True。
    from service.app import _tier2_middlewares_factory
    _, repair, breaker = asyncio.run(_tier2_middlewares_factory("u1", "a1", "s1"))
    breaker._rmb_gate_active = True
    breaker.rmb_hard_limit = 3.0
    breaker.spent_rmb = 0.0
    assert repair._budget_exhausted() is False
    breaker.spent_rmb = 3.5
    assert repair._budget_exhausted() is True


def test_single_post_persists_on_green(monkeypatch):
    # single_post + 回合真结束(ended=True)+ 读到门绿 verdict → 落库、attempts=1、消费端不重跑门(评审 I5)。
    from service import control_plane as cp
    import service.bootstrap as _bs
    import worker.run as _run

    calls = {"wait": 0, "read": 0, "gates": 0}

    async def _fake_start(*a, **k):
        return {"agent_id": "a1", "session_id": "s1"}

    async def _fake_wait(*a, **k):
        calls["wait"] += 1
        return {"ended": True, "reason": "REPLY_END", "endEvent": {}}

    def _fake_read_verdict(gid):
        calls["read"] += 1
        return {"decision": "accept", "layerResults": {"L1": {"passed": True, "gateResults": []}}}

    def _fake_gates(*a, **k):
        calls["gates"] += 1  # 消费端不应调它
        return {"rc": 0, "verdict": None, "log": ""}

    monkeypatch.setattr(_bs, "start_new_game", _fake_start)
    monkeypatch.setattr(cp, "_wait_for_turn_end", _fake_wait)
    monkeypatch.setattr(_run, "read_last_verdict", _fake_read_verdict)
    monkeypatch.setattr(_run, "run_gates", _fake_gates)
    monkeypatch.setattr(cp, "_collect_on_disk_project",
                        lambda gid: {"source_project": {"ok": 1}, "file_list": [{"path": "game.js", "content": "x"}]})
    monkeypatch.setattr(_run, "persist_source_project",
                        lambda gid, sp, fl, **k: {"id": gid, "versionId": "v1", "sourceHash": "abc"})

    res = asyncio.run(cp.drive_generation("http://x", "g1", "做个游戏", single_post=True, max_resumes=6))
    assert res["finished"] is True
    assert res["attempts"] == 1
    assert calls["wait"] == 1 and calls["read"] == 1
    assert calls["gates"] == 0, "消费端应读 verdict、不重跑门(评审 I5)"
    assert res["store_addressing"]["versionId"] == "v1"


def test_single_post_not_green_stays_on_disk(monkeypatch):
    # single_post 回合真结束但门没绿(软停/续修耗尽):不落库、reason=single_post_gates_not_green、attempts=1。
    from service import control_plane as cp
    import service.bootstrap as _bs
    import worker.run as _run

    async def _fake_start(*a, **k):
        return {"agent_id": "a1", "session_id": "s1"}

    async def _fake_wait(*a, **k):
        return {"ended": True, "reason": "EXCEED_MAX_ITERS", "endEvent": {}}

    def _fake_read_verdict(gid):
        return {"decision": "accept", "layerResults": {"L1": {"passed": False, "gateResults": [
            {"gate": "E_live", "passed": False, "fatal": True, "detail": ""}]}}}

    monkeypatch.setattr(_bs, "start_new_game", _fake_start)
    monkeypatch.setattr(cp, "_wait_for_turn_end", _fake_wait)
    monkeypatch.setattr(_run, "read_last_verdict", _fake_read_verdict)

    res = asyncio.run(cp.drive_generation("http://x", "g1", "做个游戏", single_post=True))
    assert res["finished"] is False
    assert res["attempts"] == 1
    assert res["stopped_reason"] == "single_post_gates_not_green"


def test_single_post_turn_not_ended_does_not_evaluate(monkeypatch):
    # SSE 超时未真结束(ended=False):不评中间态、不读 verdict、不落库、reason 标 not_ended(评审 I2)。
    from service import control_plane as cp
    import service.bootstrap as _bs
    import worker.run as _run

    calls = {"read": 0}

    async def _fake_start(*a, **k):
        return {"agent_id": "a1", "session_id": "s1"}

    async def _fake_wait(*a, **k):
        return {"ended": False, "reason": "total_timeout", "endEvent": None}

    def _fake_read_verdict(gid):
        calls["read"] += 1
        return None

    monkeypatch.setattr(_bs, "start_new_game", _fake_start)
    monkeypatch.setattr(cp, "_wait_for_turn_end", _fake_wait)
    monkeypatch.setattr(_run, "read_last_verdict", _fake_read_verdict)

    res = asyncio.run(cp.drive_generation("http://x", "g1", "做个游戏", single_post=True))
    assert res["finished"] is False
    assert res["attempts"] == 1
    assert res["stopped_reason"].startswith("single_post_turn_not_ended")
    assert calls["read"] == 0, "回合未真结束不应读 verdict、不评中间态"

• Step 2:跑测试,确认失败

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_repair_wiring.py -v Expected:FAIL(工厂仍返 [tracer, breaker]、breaker 无 soft_budget;drive_generation 无 single_post;worker.run 无 read_last_verdict)

• Step 3:改 app.py 工厂 —— 注入续修 middleware

把 _tier2_middlewares_factory(app.py:171-194)改为:

async def _tier2_middlewares_factory(user_id: str, agent_id: str, session_id: str) -> list:
    """extra_agent_middlewares 工厂:每回合产 [trace, 续修, 熔断] 三件。

    注入序(列表序=外→内):tracer / 续修 / 熔断;框架在其外另前置 InboxMiddleware(挂 on_reasoning、每轮 drain
    inbox 后透传全部 evt、不吞 finish Msg,故不阻断续修拦截)+ StateChange/ToolOffload(只挂 on_acting)。
    实际 on_reasoning 链 = [InboxMiddleware, tracer, repair];repair 仍是最内层、能拦到 _reasoning 产的原始 finish。
    续修 check = finish 点独立重跑 tier2 run_gates(同步 subprocess、to_thread 防阻塞)+ 异常兜底(评审 I-2:
    run_gates 非 subprocess 段也可能抛,不兜会穿透 on_reasoning → Service 无 REPLY_END、只能靠超时兜)。game_id=session_id。
    breaker 传 soft_budget=True(软停、不误伤 cheap hard 档)+ 续修模式放大的超时(容纳 N 次串行真门,评审 C1)。
    budget_exhausted 读实测已花 ¥ 超上限(非预估标记;配 T2 的 repairs>0 保护,软预算不旁路续修)。
    """
    import asyncio  # noqa: PLC0415
    from worker import genconfig, run as _run  # noqa: PLC0415 —— 惰性 import 红线
    from worker.gate_judge import GateJudgment, judge_tier2_verdict  # noqa: PLC0415
    from worker.middleware import (  # noqa: PLC0415
        CircuitBreakerMiddleware,
        RepairMiddleware,
        Tier2TraceMiddleware,
    )

    tracer = Tier2TraceMiddleware(trace_id=session_id)
    max_repairs = genconfig.get("iteration", "max_resumes", 6)
    gate_timeout_s = 300  # run_gates 单次真门上限(run.py:812 的 timeout_s 默认)
    # 续修模式重标定 breaker 超时(评审 C1):整局在一个 reply 内串行跑 (max_repairs+1) 次真门,墙钟/单步静默
    #   都要覆盖它,否则续修被自己的门判撑爆 timeout 熔断。directional,待 mini-desktop 量真实门时延后收紧(见风险段)。
    repair_wall_s = genconfig.get("budget", "repair_wall_timeout_s",
                                  (max_repairs + 1) * (gate_timeout_s + 120) + 300)  # 门 + 单轮推理 + 余量
    repair_step_s = genconfig.get("budget", "repair_step_timeout_s", gate_timeout_s + 180)  # > 单次门判上限,治静默误触
    breaker = CircuitBreakerMiddleware(
        soft_budget=True,               # tier2 单 POST 续修路走软停(cheap 仍默认 hard)
        wall_timeout_s=repair_wall_s,   # 放大墙钟容纳 N 次串行真门
        step_timeout_s=repair_step_s,   # 放大单步静默阈值 > 单次门判上限(门判期间无 evt 不误杀)
    )

    game_id = session_id  # 评门 game_id = session_id(九工具据此管工程目录)

    async def _check(agent):
        # 独立重跑门:run_gates 同步 subprocess(~300s、起 chrome),to_thread 防阻塞;异常兜底为「门未过」续修。
        #   play_spec 服务态先 None(run_gates 内按品类默认 business-sim 驱动;定制驱动随后续接口下发,列 followup)。
        try:
            gate_result = await asyncio.to_thread(_run.run_gates, game_id, None)
        except Exception as e:  # noqa: BLE001 —— 门执行任何异常都判未过续修,绝不穿透 on_reasoning(评审 I-2)
            print(f"[repair] run_gates 异常(判未过、续修): {type(e).__name__}: {e}", flush=True)
            return GateJudgment(passed=False, failed_gates=["run_gates_error"],
                                feedback=f"门执行异常,请检查工程可构建/可运行后再交付:{type(e).__name__}: {e}")
        return judge_tier2_verdict(gate_result)

    repair = RepairMiddleware(
        check=_check,
        max_repairs=max_repairs,
        # 实测已花 ¥ 超上限(非预估软停标记):配 T2 的 repairs>0 保护,首个未绿 finish 必先修一次再因预算放行。
        budget_exhausted=lambda: (
            breaker._rmb_gate_active and breaker.spent_rmb >= breaker.rmb_hard_limit),
    )
    return [tracer, repair, breaker]

• Step 4:run.py 加 read_last_verdict + control_plane single_post 分支(接收 ended、读 verdict 不重跑)

先在 tier2/gen-worker/worker/run.py(run_gates 旁)加读已落 verdict 的纯函数:

def read_last_verdict(game_id: str) -> dict | None:
    """读工程 evidence/verdict.json(服务端续修 middleware 最后一次 run_gates 已落),不重跑门。

    single_post 消费端据此判落库(评审 I5:避免消费端再起一次 ~300s 真门 + 与服务端 chrome 端口 4330/9322 竞态)。
    文件不存在 / 解析失败 → None(判未过,不伪造门绿)。
    """
    vpath = _workdir(game_id) / "evidence" / "verdict.json"
    if not vpath.exists():
        return None
    try:
        return json.loads(vpath.read_text(encoding="utf-8"))
    except Exception as e:  # noqa: BLE001 —— 解析失败按 None(判未过),不中断
        print(f"[tier2-run] read_last_verdict 解析失败(按未产出): {type(e).__name__}: {e}", flush=True)
        return None

给 drive_generation(control_plane.py:287)签名加 single_post: bool = True;惰性 import 处加 from worker.gate_judge import judge_tier2_verdict;在启动拿到 agent_id/session_id、算出 gate_game_id 之后、现有 last_verdict: dict | None = None 与外层 for attempt 循环之前,插入 single_post 分支:

    # ── ②' 单 POST 路(阶段一①):续修在 Service 端 on_reasoning middleware 内完成,消费方只发一次 kick、
    #      等这一次(内部续修多轮)回合真结束、读服务端已落 verdict 判落库;不再外层多轮 resume。老循环留 fallback。──
    if single_post:
        # SSE 总超时须覆盖续修 N 次串行真门(评审 C-2):调用方未显式放大时按 budget.repair_wall_timeout_s 兜底,
        #   与工厂 breaker 的 repair_wall_s 同源对齐;绝不用默认 1200s(否则续修中的合法局被误判 not_ended、丢产物)。
        sse_timeout = max(sse_turn_timeout_s,
                          genconfig.get("budget", "repair_wall_timeout_s",
                                        (max_resumes + 1) * (300 + 120) + 300))
        turn = await _wait_for_turn_end(
            base_url, agent_id, session_id, user_id=user_id,
            timeout_s=sse_timeout, idle_timeout_s=sse_idle_timeout_s,
            trace_path=trace_path, attempt=0)
        result["attempts"] = 1
        if not turn.get("ended"):
            # 回合未真结束(SSE 总超时/断流):服务端可能仍在续修写盘,评中间态会误判 + 与写盘竞态 → 不评、不落库。
            print(f"[tier2-control] game={game_id} single_post 回合未真结束"
                  f"(reason={turn.get('reason')}),不评中间态、不落库。", flush=True)
            result["stopped_reason"] = f"single_post_turn_not_ended:{turn.get('reason')}"
            result["wall_s"] = round(time.perf_counter() - t0, 1)
            return result
        # 回合真结束(REPLY_END/EXCEED_MAX_ITERS):读服务端续修已落的 verdict(不重跑门,评审 I5)。
        v = _run.read_last_verdict(gate_game_id)
        judgment = judge_tier2_verdict({"rc": 0, "verdict": v, "log": ""})
        result["last_verdict"] = v or None
        if judgment.passed:
            # 门绿:据 on-disk 源工程落库(逻辑同外层循环 c 分支)。
            built = _collect_on_disk_project(gate_game_id)
            if built:
                addr = _run.persist_source_project(
                    gate_game_id, built["source_project"], built["file_list"], now_ts=time.time())
                result["store_addressing"] = addr
                result["finished"] = bool(addr)
                result["stopped_reason"] = "gates_green_persisted" if addr else "gates_green_but_persist_failed"
            else:
                result["stopped_reason"] = "gates_green_but_no_on_disk_source"
        else:
            # 软停/续修耗尽放行的尽力产物:门没绿 → 只留 on-disk workdir、不入 store(创始人 2026-07-02:质量门守住)。
            print(f"[tier2-control] game={game_id} single_post 门未绿(未过门={judgment.failed_gates}),"
                  "尽力产物留 on-disk workdir、不入库。", flush=True)
            result["stopped_reason"] = "single_post_gates_not_green"
        result["wall_s"] = round(time.perf_counter() - t0, 1)
        print(f"[tier2-control] game={game_id} single_post 结束: finished={result['finished']} "
              f"reason={result['stopped_reason']} wall={result['wall_s']}s", flush=True)
        return result

    last_verdict: dict | None = None
    # ── ② 外层有界 resume 循环(fallback:single_post=False;续修 middleware spike 不成时回落这条,设计 §6)──
    for attempt in range(max_resumes + 1):
        # …(现有循环代码整段保留,不改)…

SSE 超时已在分支内兜底(落码,非仅 prose):single_post 下 sse_timeout = max(调用方传入, genconfig budget.repair_wall_timeout_s),与工厂 breaker 的 repair_wall_s 同源对齐,不用默认 1200s(否则续修中的合法局被误判 not_ended)。真值 directional、待 mini-desktop 量真实门时延后收紧(见风险段)。

(single_post=True 时 return,不进老 for attempt 循环;single_post=False 走现有外层 resume(fallback),现有循环代码整段保留不改。)

• Step 5:跑测试,确认通过

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_repair_wiring.py -v Expected:PASS(5 passed —— 工厂 soft_budget+超时 / budget_exhausted 实测 / single_post 门绿落库·门没绿留 on-disk·回合未结束不评)

• Step 6:跑 tier2 全量单测(确认服务集成未破坏现有)

Run:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/ -v Expected:全 PASS(现有 admin/trace/nacos/guardrails/续修/软停/判据/wiring 全绿)

• Step 7:提交

git add tier2/gen-worker/service/app.py tier2/gen-worker/service/control_plane.py tier2/gen-worker/tests/test_repair_wiring.py
git commit -m "feat(tier2): Service 工厂注入续修 middleware + drive_generation single_post 单回合(老 resume 留 fallback) (切片一 阶段一①/T4)"

---

随 mini-desktop 后端窗口验证 TODO(真门要 chrome/esbuild)

本地单测覆盖判据 / 续修机制 / 软停档位 / 注入结构 / 分支逻辑;下列「真门端到端 + 需真实时延标定」须在 mini-desktop(有 chrome+esbuild)按创始人窗口验:

• 先量真实门时延(定超时的前提):量单次 run_gates 真实时延(plan 假定 ~300s 是 timeout 上限,典型可能 60-120s);据此收紧工厂的 repair_wall_timeout_s/repair_step_timeout_s 与 single_post 的 sse_turn_timeout_s(directional 值须 > (max_repairs+1)×真实门时延 + 余量,否则续修被自己的门判撑爆 breaker/SSE 超时,评审 C1/C-2)。
• 续修真拦早退 + 至少修一次:tier2 一局真生成,agent 首轮想 finish 但门没绿 → 续修 middleware 拦住、注入失败门反馈、agent 续跑修 → 直到门绿 finish 或续修/预算耗尽放行。观察 [repair] 日志的续修轮次与放行原因;确认软预算耗尽也至少续修一次(repairs>0 才因预算放行)。
• 放行权威 = 独立重跑:构造 agent 谎报门绿(或漏调门工具)也不放行——放行只认 middleware 独立重跑的 run_gates。
• 单 POST 收敛 + 不撞超时:drive_generation(single_post=True) 一次 kick 走完整局(内部续修多轮)、attempts==1、回合真结束(ended=True)后读 verdict 落库;确认续修 N 轮真门未触发 breaker 墙钟/单步静默熔断、未触发 SSE 总超时;与老 single_post=False 外层 resume 路对照产物一致。
• 消费端不重跑门:single_post 落库据 read_last_verdict 读服务端已落 verdict,消费端不再起 chrome 跑门(无冗余 ~300s、无 4330/9322 端口竞态)。
• 软预算优雅收尾·尽力产物只 on-disk:把 rmb_hard_limit 压到极小,一局超预算 → agent 收「预算已达上限」提醒后尽快 finish、续修 middleware 见实测 budget_exhausted 放行,不断链失败(对比改造前 fail-closed 会抛熔断作废);门没绿的尽力产物留 on-disk workdir、不入 store(创始人拍板:质量门守住)。
• 续修与 InboxMiddleware 并存:服务态框架前置 InboxMiddleware(on_reasoning 外层每轮 drain inbox);确认它与续修 middleware 并存时 context/续跑判定正常(A1 spike 是裸 agent、未覆盖此场景,评审 I4)。
• 九门仍全绿:一局正常生成的九门(+富游戏三门)结果与改造前一致,续修未放松任何门。
• cheap 硬地板不变:mini-desktop 跑一局 cheap 生成,确认 cheap 仍走 hard 档、超 ¥10 仍 fail-closed(共用类改动未误伤,评审 C2)。

风险与回滚

• 续修 middleware 拦 finish 的机制:A1 spike 已对 2.0.2 逐条坐实、四断言绿;T2 是提升(加 GateJudgment 契约 + budget_exhausted + repairs>0 保护),机制不变。回滚 = drive_generation(single_post=False) 走老外层 resume 循环(本 plan 完整保留),续修 middleware 不注入即退回阶段〇前状态。
• 续修串行真门撞 breaker/SSE 超时(评审 C1/C-2):single_post 一局在一个 reply 内串行跑最多 (max_repairs+1) 次 ~300s 真门(最坏 ~2100s)。本 plan 靠续修模式放大 breaker 墙钟/单步静默 + SSE 总超时容纳(工厂 repair_wall_s/repair_step_s、single_post sse_turn_timeout_s),真值 directional、待 mini-desktop 量真实门时延后收紧(窗口 TODO 首条)。性能优化(门结果缓存 / 增量门缩短单局)列 follow-up、不在本 plan(创始人 2026-07-02:接受慢、先重标定跑通)。
• 软停改共用类不误伤 cheap(评审 C2/C-1):CircuitBreakerMiddleware 两档共用,软停用 soft_budget 档位隔离——默认 hard(cheap 及一切现有调用保持 fail-closed 硬地板),只 tier2 工厂显式传 soft。Step 7 回归 cheap-worker/tests/test_budget_gate.py 锁住 cheap 仍越 ¥10 抛熔断。cheap 归并续修在阶段一②,届时再评是否切 soft。
• 软预算判据双源(评审 I1/I-1):on_model_call 用预估提前置软停标记(只驱动 on_system_prompt 提醒;当前 reply 若已是 finish 则本轮看不到、下一轮生效),外加实测越限也置位;续修放行的 budget_exhausted 用实测已花 ¥ + repairs>0 保护——首个未绿 finish 必先修一次,软预算不旁路续修。
• 尽力产物只 on-disk(创始人拍板 I3):软停/续修耗尽放行的门没绿产物留 on-disk workdir、不入 store——质量门是护城河、没过门的不进正式库;「不断链」= 优雅结束 + 产物可查、非正式交付。
• 洋葱序含框架前置 InboxMiddleware(评审 I4):服务态实际 on_reasoning 链 = [InboxMiddleware, tracer, repair];repair 仍是最内层、能拦 _reasoning 产的原始 finish(InboxMiddleware 每轮 drain inbox 后透传全部 evt、不吞 finish,已核 2.0.2 源码)。tracer 在续修外层,续修吞掉的中间轮次 finish 尝试它看不到(续修对 trace 透明的预期语义;[repair] 日志兜底)。A1 spike 是裸 agent、未覆盖 InboxMiddleware 并存,列窗口 TODO 真端到端首验。
• play_spec 服务态为 None:续修 check 重跑 run_gates(game_id, None) 用品类默认 business-sim 驱动;定制驱动规格随后续接口下发(与 CLI 差异点,列 follow-up、不阻塞本 plan)。