diff --git a/docs/development-plan.md b/docs/development-plan.md index 984ca66..4bd85cb 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -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 断点恢复 diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 09751d2..92e6569 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -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 全部通过"}, diff --git a/src/test_work_node_run.py b/src/test_work_node_run.py index 3011867..7334192 100644 --- a/src/test_work_node_run.py +++ b/src/test_work_node_run.py @@ -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) diff --git a/src/work_graph_execute.py b/src/work_graph_execute.py index a1b1351..d95f32b 100644 --- a/src/work_graph_execute.py +++ b/src/work_graph_execute.py @@ -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. """ diff --git a/src/work_node_run.py b/src/work_node_run.py index ab51991..8e65610 100644 --- a/src/work_node_run.py +++ b/src/work_node_run.py @@ -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(