U-020: p_usage_limit_wait full acceptance (PH2-S3 complete)

Re-checking REQ-14's boundary case against the letter of its text found a real functional gap, not just a testing gap: U-013's opus handling only did "skip backoff" -- it never actually retried on a downgraded model, just failed immediately. REQ-14 boundary says "不走退避轮询，直接降级到非 Opus 模型重试" (skip backoff, retry immediately on a non-opus model) -- the retry was the part that was missing. Fixed: run_fn's signature is now run_fn(cwd, model=None); an opus quota signal retries immediately (no backoff) with opus_downgrade_model (default "claude-sonnet-5"). If the downgraded model somehow also reports an opus signal (shouldn't happen -- it's not using Opus), this doesn't loop forever chasing it, raises UsageLimitWaitTimeoutError instead. work_graph_execute.py's own opus handling (U-012) deliberately NOT changed to match: REQ-14's coverage only names c_work_node_run/c_org_node_trigger, and that mechanism (a multi-agent dynamic workflow, no single top-level --agent identity) has no one "the model" to downgrade -- a deeper A-1 question, documented as out of scope rather than silently left inconsistent without explanation. Also filled in a real test gap: quota_limit_type=weekly (detection + recovery, and the 9-day timeout) was never tested despite the mechanism already supporting it since U-013. 11 new/changed stdlib unittest cases across work_node_run.py's test suite (all existing run_fn lambdas updated for the new signature). 178 total across U-001-020, all passing. This closes PH2-S3 alongside U-019. REQ-14 is now genuinely complete, not just "mostly right." Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-020: p_usage_limit_wait full acceptance (PH2-S3 complete)
Re-checking REQ-14's boundary case against the letter of its text found a real functional gap, not just a testing gap: U-013's opus handling only did "skip backoff" -- it never actually retried on a downgraded model, just failed immediately. REQ-14 boundary says "不走退避轮询，直接降级到非 Opus 模型重试" (skip backoff, retry immediately on a non-opus model) -- the retry was the part that was missing. Fixed: run_fn's signature is now run_fn(cwd, model=None); an opus quota signal retries immediately (no backoff) with opus_downgrade_model (default "claude-sonnet-5"). If the downgraded model somehow also reports an opus signal (shouldn't happen -- it's not using Opus), this doesn't loop forever chasing it, raises UsageLimitWaitTimeoutError instead. work_graph_execute.py's own opus handling (U-012) deliberately NOT changed to match: REQ-14's coverage only names c_work_node_run/c_org_node_trigger, and that mechanism (a multi-agent dynamic workflow, no single top-level --agent identity) has no one "the model" to downgrade -- a deeper A-1 question, documented as out of scope rather than silently left inconsistent without explanation. Also filled in a real test gap: quota_limit_type=weekly (detection + recovery, and the 9-day timeout) was never tested despite the mechanism already supporting it since U-013. 11 new/changed stdlib unittest cases across work_node_run.py's test suite (all existing run_fn lambdas updated for the new signature). 178 total across U-001-020, all passing. This closes PH2-S3 alongside U-019. REQ-14 is now genuinely complete, not just "mostly right." Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 838667ec
Showing 5 changed files with 123 additions and 24 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/test_work_node_run.py
src/work_graph_execute.py
src/work_node_run.py
@@ -193,6 +193,7 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-020 [Risk Gate]** `p_usage_limit_wait` 完整验收（`session`/`weekly`/`opus` 三种 `quota_limit_type` + 超时兜底）
   - 依赖：U-012（骨架已在 U-012 实现，本 Unit 补齐边界/失败场景）
   - Done：REQ-14 全部用例通过（含超时转 `blocked_pending_human, blocked_reason='usage_limit_wait_timeout'`）
+  - **实现补充（2026-07-31）**：这次不只是补测试——重新对照 REQ-14 boundary 用例原文发现，U-013 时 `work_node_run.py` 的 opus 处理只做到"不退避"，没做到"直接降级到非 Opus 模型重试"（原来是探测到 opus 额度直接抛错，从没真正重试过）。补上了：`run_fn` 签名改成 `run_fn(cwd, model=None)`，探测到 opus 信号后立即（不等待）用 `opus_downgrade_model`（默认 `claude-sonnet-5`）重试一次；如果降级后还是收到 opus 信号（不合理，非 opus 模型不该撞 opus 专属额度），直接判超时，不会无限重试。顺手把之前漏测的 `weekly` 额度类型（探测恢复 + 9 天超时）也补齐了。`work_graph_execute.py`（U-012）里同名的 opus 立即失败逻辑**没有**跟着改——REQ-14 覆盖列表只写了 `c_work_node_run`/`c_org_node_trigger`，没有 `c_work_graph_execute`，而且那条路径是多 Agent 动态工作流，没有一个单一的"the model"可以整体降级，属于更深的 A-1 动态工作流问题，本 Unit 不动
   - 覆盖：04-P `p_usage_limit_wait`；REQ-14
 ### PH2-S4 断点恢复
@@ -33,7 +33,7 @@
     {"id": "U-017", "phase": "PH2", "sub_phase": "PH2-S1", "name": "人为植入 bug 完整闭环验证（见 development-plan.md 2026-07-31 实现补充）", "classification": "Risk Gate", "depends_on": ["U-016"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "架构文档 §11 验收标准第 4 条达成：run_fix_loop 串联 U-015 判定 + U-016 路由 + 真实 Coder 修复 + 真实 Validator 复核"},
     {"id": "U-018", "phase": "PH2", "sub_phase": "PH2-S2", "name": "c_skill_update + c_skill_rollback（见 development-plan.md 2026-07-31 实现补充）", "classification": "Risk Gate", "depends_on": ["U-015"], "coverage": {"model": ["skill", "skill_iteration"], "view": [], "controller": ["c_skill_update", "c_skill_rollback"], "process": [], "requirement": ["REQ-08"]}, "done": "REQ-08 全部通过；顺带解决了 U-015 的 skill_defect 缺口（bug_record.skill_iteration_suggestion_id 由 update_skill 回填）"},
     {"id": "U-019", "phase": "PH2", "sub_phase": "PH2-S3", "name": "budget_guardrail_config + c_budget_check（见 development-plan.md 2026-07-31 实现补充）", "classification": "Risk Gate", "depends_on": ["U-012"], "coverage": {"model": ["budget_guardrail_config"], "view": [], "controller": ["c_budget_check"], "process": ["p_budget_guardrail"], "requirement": ["REQ-10"]}, "done": "REQ-10 happy/boundary 通过；boundary 复用 U-016 manual_resume，完整 c_budget_config_update UI 是 U-030 的事"},
-    {"id": "U-020", "phase": "PH2", "sub_phase": "PH2-S3", "name": "p_usage_limit_wait 完整验收", "classification": "Risk Gate", "depends_on": ["U-012"], "coverage": {"model": [], "view": [], "controller": [], "process": ["p_usage_limit_wait"], "requirement": ["REQ-14"]}, "done": "REQ-14 全部通过（含超时兜底）"},
+    {"id": "U-020", "phase": "PH2", "sub_phase": "PH2-S3", "name": "p_usage_limit_wait 完整验收（见 development-plan.md 2026-07-31 实现补充）", "classification": "Risk Gate", "depends_on": ["U-012"], "coverage": {"model": [], "view": [], "controller": [], "process": ["p_usage_limit_wait"], "requirement": ["REQ-14"]}, "done": "REQ-14 全部通过（含超时兜底）；补上了 opus 真实降级重试（U-013 时只做到不退避，没做到真重试）+ 之前漏测的 weekly 类型"},
     {"id": "U-021", "phase": "PH2", "sub_phase": "PH2-S4", "name": "Graph Engine 持久化 + 断点恢复", "classification": "Risk Gate", "depends_on": ["U-012", "U-013"], "coverage": {"model": ["work_graph_run", "work_node_execution"], "view": [], "controller": [], "process": ["p_work_graph_execution"], "requirement": ["REQ-06"]}, "done": "REQ-06 happy/boundary/failure 通过"},
     {"id": "U-022", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_project_setup + c_pick_folder 全平台", "classification": "Value Path", "depends_on": ["U-007", "U-008"], "coverage": {"model": [], "view": ["v_project_setup"], "controller": ["c_pick_folder"], "process": [], "requirement": ["REQ-11"]}, "done": "REQ-11 全部通过"},
@@ -65,7 +65,7 @@ class InputValidationTest(WorkNodeRunTestBase):
     def test_invalid_input_rejected_before_execution_starts(self):
         with self.assertRaises(work_node_run.InvalidInputError):
             work_node_run.run_work_node(
-                self.conn, "node-1", {}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=lambda cwd: {"value": 1}
+                self.conn, "node-1", {}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=lambda cwd, model=None: {"value": 1}
             )
         status = self.conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'").fetchone()[0]
         self.assertEqual(status, "pending")
@@ -76,7 +76,7 @@ class InputValidationTest(WorkNodeRunTestBase):
 class HappyPathTest(WorkNodeRunTestBase):
     def test_valid_output_marks_node_done(self):
         output = work_node_run.run_work_node(
-            self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=lambda cwd: {"value": 42}
+            self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=lambda cwd, model=None: {"value": 42}
         )
         self.assertEqual(output, {"value": 42})
         row = self.conn.execute(
@@ -89,7 +89,7 @@ class HappyPathTest(WorkNodeRunTestBase):
     def test_execution_log_file_created_active_then_closed(self):
         seen_status_during_run = []
-        def run_fn(cwd):
+        def run_fn(cwd, model=None):
             seen_status_during_run.append(
                 self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0]
             )
@@ -101,7 +101,7 @@ class HappyPathTest(WorkNodeRunTestBase):
         self.assertEqual(final_status, "closed")
     def test_log_file_closed_even_when_run_fn_raises(self):
-        def boom(cwd):
+        def boom(cwd, model=None):
             raise RuntimeError("node blew up")
         with self.assertRaises(RuntimeError):
@@ -114,7 +114,7 @@ class RetryOnSchemaFailureTest(WorkNodeRunTestBase):
     def test_retries_then_succeeds(self):
         attempts = {"n": 0}
-        def run_fn(cwd):
+        def run_fn(cwd, model=None):
             attempts["n"] += 1
             if attempts["n"] < 3:
                 return {"wrong_key": True}
@@ -133,7 +133,7 @@ class RetryOnSchemaFailureTest(WorkNodeRunTestBase):
         with self.assertRaises(work_node_run.SchemaValidationFailedError):
             work_node_run.run_work_node(
                 self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA,
-                run_fn=lambda cwd: {"wrong_key": True}, max_retries=2,
+                run_fn=lambda cwd, model=None: {"wrong_key": True}, max_retries=2,
             )
         row = self.conn.execute(
             "SELECT status, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'"
@@ -157,7 +157,7 @@ class UsageLimitWaitTest(WorkNodeRunTestBase):
     def test_session_limit_then_recovers(self):
         attempts = {"n": 0}
-        def run_fn(cwd):
+        def run_fn(cwd, model=None):
             attempts["n"] += 1
             if attempts["n"] == 1:
                 work_node_run.raise_if_usage_limited("Error: You've hit your session limit")
@@ -179,8 +179,46 @@ class UsageLimitWaitTest(WorkNodeRunTestBase):
         ).fetchone()
         self.assertEqual(event[0], "quota_wait_started")
-    def test_opus_quota_raises_immediately_no_backoff(self):
-        def run_fn(cwd):
+    def test_opus_quota_retries_immediately_on_downgraded_model_no_backoff(self):
+        # REQ-14 boundary: "不走退避轮询，直接降级到非 Opus 模型重试" -- an actual
+        # retry with a different model, not just detect-and-fail.
+        calls = []
+
+        def run_fn(cwd, model=None):
+            calls.append(model)
+            if len(calls) == 1:
+                work_node_run.raise_if_usage_limited("You've hit your Opus limit")
+            return {"value": "recovered on " + str(model)}
+
+        output = work_node_run.run_work_node(
+            self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn,
+            sleep=self._sleep, now=self._now,
+        )
+        self.assertEqual(calls, [None, "claude-sonnet-5"])
+        self.assertEqual(output, {"value": "recovered on claude-sonnet-5"})
+        self.assertEqual(self.sleeps, [])  # no backoff wait for opus
+        status = self.conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'").fetchone()[0]
+        self.assertEqual(status, "done")
+
+    def test_opus_quota_custom_downgrade_model_is_used(self):
+        calls = []
+
+        def run_fn(cwd, model=None):
+            calls.append(model)
+            if len(calls) == 1:
+                work_node_run.raise_if_usage_limited("You've hit your Opus limit")
+            return {"value": "ok"}
+
+        work_node_run.run_work_node(
+            self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn,
+            sleep=self._sleep, now=self._now, opus_downgrade_model="claude-fable-5",
+        )
+        self.assertEqual(calls, [None, "claude-fable-5"])
+
+    def test_opus_quota_persisting_after_downgrade_raises_timeout(self):
+        # a non-opus model shouldn't hit an opus-specific quota again -- if it
+        # somehow does, this doesn't loop forever chasing it
+        def run_fn(cwd, model=None):
             work_node_run.raise_if_usage_limited("You've hit your Opus limit")
         with self.assertRaises(work_node_run.UsageLimitWaitTimeoutError):
@@ -191,7 +229,7 @@ class UsageLimitWaitTest(WorkNodeRunTestBase):
         self.assertEqual(self.sleeps, [])
     def test_session_timeout_after_8_hours_blocks_node(self):
-        def run_fn(cwd):
+        def run_fn(cwd, model=None):
             work_node_run.raise_if_usage_limited("You've hit your session limit")
         with self.assertRaises(work_node_run.UsageLimitWaitTimeoutError):
@@ -203,6 +241,40 @@ class UsageLimitWaitTest(WorkNodeRunTestBase):
         status = self.conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'").fetchone()[0]
         self.assertEqual(status, "blocked")
+    def test_weekly_limit_then_recovers(self):
+        # previously untested quota_limit_type -- session/opus were covered,
+        # weekly wasn't, even though the detection regex and timeout table
+        # both already supported it
+        attempts = {"n": 0}
+
+        def run_fn(cwd, model=None):
+            attempts["n"] += 1
+            if attempts["n"] == 1:
+                work_node_run.raise_if_usage_limited("Error: You've hit your weekly limit")
+            return {"value": "recovered"}
+
+        work_node_run.run_work_node(
+            self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn,
+            sleep=self._sleep, now=self._now,
+        )
+        quota_type = self.conn.execute(
+            "SELECT quota_limit_type FROM work_node_execution WHERE node_exec_id = 'node-1'"
+        ).fetchone()[0]
+        self.assertEqual(quota_type, "weekly")
+
+    def test_weekly_timeout_after_9_days_blocks_node(self):
+        def run_fn(cwd, model=None):
+            work_node_run.raise_if_usage_limited("You've hit your weekly limit")
+
+        with self.assertRaises(work_node_run.UsageLimitWaitTimeoutError):
+            work_node_run.run_work_node(
+                self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn,
+                sleep=self._sleep, now=self._now,
+            )
+        self.assertGreaterEqual(self.clock[0], 9 * 24 * 3600)
+        status = self.conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'").fetchone()[0]
+        self.assertEqual(status, "blocked")
+
 class RealGitWorktreeIsolationTest(WorkNodeRunTestBase):
     """REQ-03 boundary: two independent nodes, each in its own git worktree,
@@ -221,7 +293,7 @@ class RealGitWorktreeIsolationTest(WorkNodeRunTestBase):
         self.conn.commit()
     def _write_file_run_fn(self, filename, content):
-        def run_fn(cwd):
+        def run_fn(cwd, model=None):
             with open(os.path.join(cwd, filename), "w") as f:
                 f.write(content)
             subprocess.run(["git", "add", filename], cwd=cwd, check=True, capture_output=True)
@@ -141,11 +141,20 @@ def execute_work_graph_with_usage_limit_retry(
     """Wraps execute_work_graph with p_usage_limit_wait (04-P §4.4.1): a
     session/weekly quota hit backs off 5->10->20->30min (repeating at 30)
     and retries until that type's timeout, then raises
-    UsageLimitWaitTimeoutError. An opus quota hit skips backoff entirely
-    (04-P: "直接...降级到非 Opus 模型重试" -- the actual model swap is a
-    caller-level decision, not this function's; it raises immediately so the
-    caller can decide). ClassifierBlockedError is never retried (03-C:
-    not retryable, needs human review) and propagates immediately.
+    UsageLimitWaitTimeoutError. ClassifierBlockedError is never retried
+    (03-C: not retryable, needs human review) and propagates immediately.
+
+    An opus quota hit skips backoff and raises immediately, unlike
+    src/work_node_run.py's run_work_node (U-020), which actually retries on
+    a downgraded model. Deliberately not matched here: REQ-14's coverage
+    (06 节) only names c_work_node_run/c_org_node_trigger, not
+    c_work_graph_execute, and this function's underlying mechanism (a
+    dynamic-workflow script with no single top-level --agent identity, per
+    03-C §3.5) has no one "the model" to swap the way a single-agent call
+    does -- individual agent() calls inside the workflow script would each
+    need their own override, a deeper A-1 dynamic-workflow question this
+    Unit doesn't attempt. Caller still has to decide what "retry on a
+    non-opus model" means here.
     `sleep`/`now` are injectable so tests don't block on real 5-30 minute waits.
     """
@@ -124,10 +124,13 @@ def run_work_node(
     repo_path=None,
     sleep=time.sleep,
     now=time.monotonic,
+    opus_downgrade_model="claude-sonnet-5",
 ):
-    """Runs one Work Node to contract. `run_fn(cwd) -> output_payload` is the
-    caller-supplied actual work (a real `claude -p` call in production, a
-    fake in tests). Returns the validated output_payload.
+    """Runs one Work Node to contract. `run_fn(cwd, model) -> output_payload`
+    is the caller-supplied actual work (a real `claude -p` call in
+    production, a fake in tests); `model` is None unless this call is an
+    opus-quota-triggered retry, in which case it's `opus_downgrade_model`.
+    Returns the validated output_payload.
     Raises InvalidInputError (input schema), UsageLimitWaitTimeoutError
     (p_usage_limit_wait exhausted), or leaves work_node_execution.status
@@ -169,9 +172,11 @@ def run_work_node(
         retry_count = 0
         quota_wait_start = None
+        current_model = None
+        opus_already_downgraded = False
         while True:
             try:
-                output_payload = run_fn(cwd)
+                output_payload = run_fn(cwd, model=current_model)
             except _UsageLimitSignal as signal:
                 quota_type = signal.quota_limit_type
                 if quota_wait_start is None:
@@ -188,9 +193,21 @@ def run_work_node(
                     )
                     conn.commit()
                 if quota_type == "opus":
-                    raise UsageLimitWaitTimeoutError(
-                        "opus quota exhausted for {} -- caller should retry on a non-opus model".format(node_exec_id)
-                    ) from signal
+                    # 04-P §4.4.1: "不用等待退避——直接把该节点这次调用降级到非
+                    # Opus 模型重试", not just detect-and-fail. Only one
+                    # downgrade attempt -- if the downgraded model *also*
+                    # reports an opus-quota signal, something's wrong (a
+                    # non-opus model shouldn't hit an opus-specific quota),
+                    # so this doesn't loop forever chasing it.
+                    if opus_already_downgraded:
+                        raise UsageLimitWaitTimeoutError(
+                            "opus quota signal persisted for {} even after downgrading to {}".format(
+                                node_exec_id, opus_downgrade_model
+                            )
+                        ) from signal
+                    current_model = opus_downgrade_model
+                    opus_already_downgraded = True
+                    continue
                 elapsed = now() - quota_wait_start
                 if elapsed >= _TIMEOUT_SECONDS[quota_type]:
                     conn.execute(