feat(tier2): 软预算 ¥ 闸从 fail-closed 改优雅收尾软停(交尽力产物·不断链) (切片一 阶段一①/T3)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tier2/gen-worker/tests/test_budget_soft_stop.py

@@ -0,0 +1,107 @@
+"""软预算 ¥ 闸档位单测: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())

tier2/gen-worker/worker/middleware.py

@@ -195,6 +195,7 @@ class CircuitBreakerMiddleware(MiddlewareBase):
         pricing_params: Optional[dict] = None,
         enable_rmb_gate: bool = True,
         group_ratio: float | None = None,
+        soft_budget: bool = False,   # 新增:¥ 越限行为档位(False=hard fail-closed 保持现状/cheap;True=soft 软停,只 tier2 传)
     ) -> None:
         """
         Args:
@@ -209,6 +210,8 @@ class CircuitBreakerMiddleware(MiddlewareBase):
                 fetch_pricing_params 惰性活读取(best-effort,取不到则 ¥ 闸降级为次数闸)。显式传入便于测试/复用。
             enable_rmb_gate: 是否启用 ¥ 累进硬闸(False → 只走原四道;留个总开关便于排障/对照)。
             group_ratio: new-api 分组倍率(成本折算用)。
+            soft_budget: ¥ 越限行为档位。False=hard(越限抛熔断,cheap 及默认,守 ¥ 硬地板);
+                True=soft(越限软停放行 + 强提醒催 finish、不断链;只 tier2 工厂显式传 True,交尽力产物)。
 
         旋钮外置(运行时读):上述各熔断/软刹/¥ 闸/倍率旋钮未显式传入(None)→ 实例化时从生成配置层
             (generation.yaml 的 budget 区)读、default=现值;故「调成本上限/失控保护强度 = 改 YAML 重跑」。
@@ -242,6 +245,8 @@ class CircuitBreakerMiddleware(MiddlewareBase):
         # ¥ 累进硬闸配置(directional 上限 + 计费参数;参数为 None 时 on_model_call 首调惰性取价)。
         self.rmb_hard_limit = rmb_hard_limit
         self.enable_rmb_gate = enable_rmb_gate
+        # ¥ 越限行为档位:False=hard(越限抛熔断,cheap 及默认,守 ¥ 硬地板);True=soft(越限软停放行,tier2 单 POST 续修路)。
+        self.soft_budget = soft_budget
         self.group_ratio = group_ratio
         self._pricing_params = pricing_params  # {pricing, qpu, usd_rate} 或 None(惰性取价后填)
         self._pricing_fetch_tried = pricing_params is not None  # 已有则不再去取
@@ -259,15 +264,21 @@ class CircuitBreakerMiddleware(MiddlewareBase):
         # ¥ 累进硬闸:累计已花 ¥(每次模型调用后按实测 usage 折算累加)+ 当前是否走金额闸(取到活价)。
         self.spent_rmb = 0.0
         self._rmb_gate_active = self._pricing_params is not None  # 有计费参数才按金额拦,否则降级次数闸
+        # ¥ 软预算越限标记(soft 档:越限设 True + 放行本次,agent 收 on_system_prompt 强提醒后尽快 finish)。
+        self.budget_soft_tripped = False
         # 触发记录(供编排器/调试读)。
         self.tripped: dict | None = None
 
     # ── 软刹:在硬顶临近时往 system prompt 追加收敛提醒(transformer 钩子,顺序管线)──
     async def on_system_prompt(self, agent, current_prompt: str) -> str:
-        """达 soft_ratio×硬顶时,在 system prompt 末尾追加一条收敛提醒(软刹,非强制停)。
-
-        给模型一个体面收尾的机会:提醒它接近步数/预算上限、该尽快产出可交付源工程并调 finish。
-        """
+        """达 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:
@@ -363,15 +374,28 @@ class CircuitBreakerMiddleware(MiddlewareBase):
         # ① 惰性取价(仅首次;best-effort,失败则本局降级次数闸,不再重试以免每次调用都打网关)。
         self._ensure_pricing()
 
-        # ② 调用前 fail-closed 金额判(只有取到活价、即金额闸生效时才硬拦;否则降级、由次数闸兜)。
+        # ② 调用前 ¥ 判(取到活价、金额闸生效时才判;否则降级、由 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:
-                self._trip(
-                    "budget",
-                    f"¥ 累进硬闸越限 fail-closed:已花 ¥{self.spent_rmb:.4f} + 本次预估 ¥{est_rmb:.4f}"
-                    f" > 上限 ¥{self.rmb_hard_limit:.2f}(模型={model_name or '未知'})",
-                )
+                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,
+                    )
 
         # ③ 调真模型,调用后按实测 usage 折 ¥ 累加(非流式同步读;流式包一层从末块读)。
         result = await next_handler(**input_kwargs)
@@ -473,12 +497,21 @@ class CircuitBreakerMiddleware(MiddlewareBase):
                 cached_tokens=cached,
             )
             self.spent_rmb += float(detail.get("rmb", 0.0) or 0.0)
+            self._maybe_soft_trip_on_measured()  # soft 档:实测折算后越限也置软停(评审 I-1)
         except Exception as exc:  # noqa: BLE001 —— 折算异常不累加、不抛(成本少记一笔好过冤杀生成)
             print(
                 f"[tier2-circuit] ¥ 实测折算失败(本次未计入累计,不中断):{exc}",
                 flush=True,
             )
 
+    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)
+
     def _trip(self, kind: str, reason: str) -> None:
         """记下触发并抛出 Tier2CircuitBreak(供编排器 catch → 落 verdict.breakerKind)。"""
         self.tripped = {"kind": kind, "reason": reason}