diff --git a/docs/development-plan.md b/docs/development-plan.md index 0edb2c8..f467322 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -125,8 +125,9 @@ - **U-012 [Value Path]** `c_work_graph_execute` 完整实现（`auto` 权限模式、`USAGE_LIMIT_REACHED` 退避重试 `p_usage_limit_wait`、`CLASSIFIER_BLOCKED` 处理） - 依赖：U-006, U-011 - - Done：REQ-07 全部用例通过；REQ-14（`p_usage_limit_wait`）happy 用例通过（boundary/failure 场景在 PH2-S3 补齐完整覆盖） - - 覆盖：03-C `c_work_graph_execute`；04-P `p_work_graph_execution`/`p_usage_limit_wait`；REQ-07/REQ-14 + - Done（**2026-07-31 订正**：原来写"REQ-07 全部用例通过"，但 REQ-07 happy 用例原话是"Claude Code ACP 后端下完整跑通 REQ-01~06 全部场景"——这需要 REQ-04/05/06（Validator 根因归属、Org-level Loop、断点恢复）全部就绪，那是 PH2 的事，U-012 单靠 U-006+U-011 做不到，是同一类 Done 标准过头问题第四次出现）：REQ-07 **boundary**（不加 `--bare`，冷启动开销可接受——U-006 已满足，本 Unit 未改变这点）和 **failure**（`CLASSIFIER_BLOCKED` 检测正确路由）通过；happy 用例随 PH2 收口后累计满足。REQ-14 happy 用例通过（boundary/failure 场景在 PH2-S3/U-020 补齐完整覆盖）——这条原来就写对了，不用改 + - **范围说明（2026-07-31，实现后补充）**：`p_usage_limit_wait` 退避重试落在整个 `execute_work_graph` 调用级别，不是逐个 `work_node_execution` 记录级别——U-013（`c_work_node_run`，真正拥有 `work_node_execution` 行）还没做，节点级别的额度等待追踪要等 U-013 才有意义。`CLASSIFIER_BLOCKED` 判定用单次 `-p` 调用自带的 `permission_denials` 数组长度 ≥3 做近似（03-C 原话"连续 3 次或累计 20 次"，跨调用的真正累计计数需要持久化，也是等 U-013） + - 覆盖：03-C `c_work_graph_execute`；04-P `p_work_graph_execution`/`p_usage_limit_wait`；REQ-07（boundary/failure）/REQ-14（happy） - **U-013 [Value Path]** `c_work_node_run` + `git_worktree` 隔离（并行写代码节点） - 依赖：U-012 diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 2317775..42b84be 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -24,7 +24,7 @@ {"id": "U-009", "phase": "PH1", "sub_phase": "PH1-S1", "name": "c_task_create + task 状态机（c_project_create 已在 U-002 完成）", "classification": "Value Path", "depends_on": ["U-001", "U-002"], "coverage": {"model": ["task"], "view": [], "controller": ["c_task_create"], "process": ["p_org_task_lifecycle"], "requirement": ["REQ-01"]}, "done": "REQ-01 boundary/failure 通过；happy 用例需要 c_org_node_trigger 完整实现，随 U-010 累计满足（2026-07-31 订正）"}, {"id": "U-010", "phase": "PH1", "sub_phase": "PH1-S1", "name": "c_org_node_trigger 完整实现（三段前向状态迁移，见 development-plan.md 2026-07-31 范围说明）", "classification": "Value Path", "depends_on": ["U-005", "U-009"], "coverage": {"model": [], "view": [], "controller": ["c_org_node_trigger"], "process": [], "requirement": []}, "done": "三个 Org Agent 分别可被触发（真实 claude --agent 调用），task.status 按 queued→planning_brd_prd_sdd→coding→validating 正确流转；重试/路由/blocked_pending_human 不在本 Unit 范围"}, {"id": "U-011", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_graph_plan 实现（只规划不执行，见 development-plan.md 2026-07-31 订正）", "classification": "Value Path", "depends_on": ["U-004", "U-010"], "coverage": {"model": ["artifact", "work_graph_run"], "view": [], "controller": ["c_work_graph_plan"], "process": [], "requirement": ["REQ-02"]}, "done": "graph_manifest 通过结构校验（brd/prd/sdd/benchmark 四节点+依赖链），work_graph_run(status=planning) 落地；REQ-02 happy 用例整体随 U-012/U-013 累计满足。c_artifact_write（03-C §3.7）目前没有任何 Unit 认领，是已知缺口"}, - {"id": "U-012", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_graph_execute 完整实现", "classification": "Value Path", "depends_on": ["U-006", "U-011"], "coverage": {"model": [], "view": [], "controller": ["c_work_graph_execute"], "process": ["p_work_graph_execution", "p_usage_limit_wait"], "requirement": ["REQ-07", "REQ-14"]}, "done": "REQ-07 全部通过；REQ-14 happy 用例通过"}, + {"id": "U-012", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_graph_execute 完整实现（见 development-plan.md 2026-07-31 范围说明）", "classification": "Value Path", "depends_on": ["U-006", "U-011"], "coverage": {"model": [], "view": [], "controller": ["c_work_graph_execute"], "process": ["p_work_graph_execution", "p_usage_limit_wait"], "requirement": ["REQ-07", "REQ-14"]}, "done": "REQ-07 boundary/failure 通过（happy 需 PH2 REQ-04/05/06 就绪后累计满足）；REQ-14 happy 通过。退避重试是整图级别，非逐节点；CLASSIFIER_BLOCKED 用单次调用 permission_denials≥3 近似判定"}, {"id": "U-013", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_node_run + git worktree 隔离", "classification": "Value Path", "depends_on": ["U-012"], "coverage": {"model": ["work_node_execution"], "view": [], "controller": ["c_work_node_run"], "process": [], "requirement": ["REQ-03"]}, "done": "REQ-03 happy/boundary 通过"}, {"id": "U-014", "phase": "PH1", "sub_phase": "PH1-S3", "name": "贪吃蛇场景端到端验收", "classification": "Value Path", "depends_on": ["U-011", "U-012", "U-013"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "架构文档 §11 验收标准第 1-3 条全部达成"}, diff --git a/src/test_work_graph_execute.py b/src/test_work_graph_execute.py index 6f17e06..84e105f 100644 --- a/src/test_work_graph_execute.py +++ b/src/test_work_graph_execute.py @@ -1,6 +1,8 @@ -"""U-006 tests: c_work_graph_execute's dynamic-workflow wrapper, mocked at the -subprocess boundary (CLAUDE.md Testing constraint). Real end-to-end coverage -is in src/test_work_graph_execute_e2e.py (RUN_REAL_AGENT_E2E=1), reusing the +"""U-006/U-012 tests: c_work_graph_execute's dynamic-workflow wrapper (U-006) +plus CLASSIFIER_BLOCKED / USAGE_LIMIT_REACHED detection and the +p_usage_limit_wait retry loop (U-012), mocked at the subprocess boundary +(CLAUDE.md Testing constraint). Real end-to-end coverage is in +src/test_work_graph_execute_e2e.py (RUN_REAL_AGENT_E2E=1), reusing the exact spike script the A-1 assumption was validated against. Run: python3 -m unittest src.test_work_graph_execute -v @@ -82,5 +84,120 @@ class ExecuteWorkGraphTest(unittest.TestCase): work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {}, self.SCHEMA) +class ClassifierBlockedDetectionTest(unittest.TestCase): + SCHEMA = {"type": "object"} + + def test_three_or_more_permission_denials_raises_classifier_blocked(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": True, "permission_denials": ["a", "b", "c"]} + ) + with self.assertRaises(work_graph_execute.ClassifierBlockedError): + work_graph_execute.execute_work_graph("/tmp/p", "run-1", {}, self.SCHEMA) + + def test_fewer_than_three_denials_is_generic_error_not_classifier_blocked(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": True, "permission_denials": ["a"]} + ) + with self.assertRaises(work_graph_execute.WorkGraphExecuteError) as ctx: + work_graph_execute.execute_work_graph("/tmp/p", "run-1", {}, self.SCHEMA) + self.assertNotIsInstance(ctx.exception, work_graph_execute.ClassifierBlockedError) + + +class UsageLimitDetectionTest(unittest.TestCase): + SCHEMA = {"type": "object"} + + def test_session_limit_phrase_in_stderr_on_nonzero_exit(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process( + returncode=1, stdout_obj=None, stderr="Error: You've hit your session limit" + ) + with self.assertRaises(work_graph_execute.UsageLimitReachedError) as ctx: + work_graph_execute.execute_work_graph("/tmp/p", "run-1", {}, self.SCHEMA) + self.assertEqual(ctx.exception.quota_limit_type, "session") + + def test_weekly_limit_phrase_in_error_result(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": True, "result": "You've hit your weekly limit"} + ) + with self.assertRaises(work_graph_execute.UsageLimitReachedError) as ctx: + work_graph_execute.execute_work_graph("/tmp/p", "run-1", {}, self.SCHEMA) + self.assertEqual(ctx.exception.quota_limit_type, "weekly") + + +class RetryWrapperTest(unittest.TestCase): + SCHEMA = {"type": "object"} + + def setUp(self): + self.sleeps = [] + self.clock = [0.0] + + def _sleep(self, seconds): + self.sleeps.append(seconds) + self.clock[0] += seconds + + def _now(self): + return self.clock[0] + + def test_recovers_after_backoff_no_timeout_hit(self): + responses = [ + _completed_process(returncode=1, stdout_obj=None, stderr="hit your session limit"), + _completed_process(stdout_obj={"is_error": False, "structured_output": {"ok": True}}), + ] + with patch("src.work_graph_execute.subprocess.run", side_effect=responses): + result = work_graph_execute.execute_work_graph_with_usage_limit_retry( + "/tmp/p", "run-1", {}, self.SCHEMA, sleep=self._sleep, now=self._now + ) + self.assertEqual(result, {"ok": True}) + self.assertEqual(self.sleeps, [5 * 60]) # first backoff step + + def test_backoff_schedule_escalates_then_holds_at_30(self): + # 5 failures before success -> backoff steps 5,10,20,30,30 minutes + responses = [ + _completed_process(returncode=1, stdout_obj=None, stderr="hit your session limit") + ] * 5 + [_completed_process(stdout_obj={"is_error": False, "structured_output": {"ok": True}})] + with patch("src.work_graph_execute.subprocess.run", side_effect=responses): + work_graph_execute.execute_work_graph_with_usage_limit_retry( + "/tmp/p", "run-1", {}, self.SCHEMA, sleep=self._sleep, now=self._now + ) + self.assertEqual(self.sleeps, [5 * 60, 10 * 60, 20 * 60, 30 * 60, 30 * 60]) + + def test_session_timeout_after_8_hours_raises_timeout_error(self): + def always_fails(*args, **kwargs): + return _completed_process(returncode=1, stdout_obj=None, stderr="hit your session limit") + + with patch("src.work_graph_execute.subprocess.run", side_effect=always_fails): + with self.assertRaises(work_graph_execute.UsageLimitWaitTimeoutError): + work_graph_execute.execute_work_graph_with_usage_limit_retry( + "/tmp/p", "run-1", {}, self.SCHEMA, sleep=self._sleep, now=self._now + ) + self.assertGreaterEqual(self.clock[0], 8 * 3600) + + def test_opus_quota_raises_timeout_error_immediately_no_backoff(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process( + returncode=1, stdout_obj=None, stderr="opus model quota exhausted" + ) + with self.assertRaises(work_graph_execute.UsageLimitWaitTimeoutError): + work_graph_execute.execute_work_graph_with_usage_limit_retry( + "/tmp/p", "run-1", {}, self.SCHEMA, sleep=self._sleep, now=self._now + ) + self.assertEqual(self.sleeps, []) + + def test_classifier_blocked_is_not_retried(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": True, "permission_denials": ["a", "b", "c"]} + ) + with self.assertRaises(work_graph_execute.ClassifierBlockedError): + work_graph_execute.execute_work_graph_with_usage_limit_retry( + "/tmp/p", "run-1", {}, self.SCHEMA, sleep=self._sleep, now=self._now + ) + self.assertEqual(self.sleeps, []) + self.assertEqual(run.call_count, 1) + + if __name__ == "__main__": unittest.main() diff --git a/src/work_graph_execute.py b/src/work_graph_execute.py index fa983a7..d1caad7 100644 --- a/src/work_graph_execute.py +++ b/src/work_graph_execute.py @@ -1,21 +1,69 @@ -"""c_work_graph_execute (03-C §3.5): dynamic-workflow invocation skeleton. - -U-006 scope: write an orchestration script into the target project's -`.claude/workflows/.js` and invoke it via `claude -p 'Run / with -args {...}'`, reusing the exact code path the A-1 spike validated -(spikes/a1-dynamic-workflows/, confirmed 2026-07-29). No budget-usage -bookkeeping or CLASSIFIER_BLOCKED handling yet -- that's U-012's job -("骨架" here, per development-plan.md). +"""c_work_graph_execute (03-C §3.5): dynamic-workflow invocation, full implementation. + +Writes an orchestration script into the target project's `.claude/workflows/ +.js` and invokes it via `claude -p 'Run / with args {...}'`, +reusing the exact code path the A-1 spike validated (spikes/a1-dynamic-workflows/, +confirmed 2026-07-29). + +U-012 scope, added on top of U-006's skeleton: CLASSIFIER_BLOCKED detection +(03-C §3.5) and USAGE_LIMIT_REACHED backoff-retry (04-P §4.4.1 +p_usage_limit_wait -- REQ-14's happy case only; boundary/failure is U-020). +Both are pure, time-injectable logic wrapping the same subprocess call, no +dependency on U-013's work_node_execution rows (which don't exist yet -- +see development-plan.md's 2026-07-31 scope note on this Unit for why the +retry loop operates at the whole-work-graph level rather than per node). """ import json import os import subprocess +import time + +_USAGE_LIMIT_PHRASES = ("hit your session limit", "hit your weekly limit", "usage limit") +_BACKOFF_SCHEDULE_MINUTES = (5, 10, 20, 30) # then repeats at 30 until the type's timeout +_TIMEOUT_SECONDS = {"session": 8 * 3600, "weekly": 9 * 24 * 3600} +# 03-C §3.5: "分类器连续拦截 3 次或累计 20 次" -- a single `-p` call's own +# permission_denials array already reflects that whole session's denials, so +# 3 is used as the practical per-call threshold (conservative relative to the +# 20 cumulative ceiling; true cross-call cumulative tracking needs U-013's +# per-node persistence and isn't attempted here). +_CLASSIFIER_BLOCKED_THRESHOLD = 3 class WorkGraphExecuteError(Exception): """Raised when the claude subprocess exits non-zero, reports is_error, or - produces no structured_output.""" + produces no structured_output, and it isn't one of the more specific + errors below.""" + + +class ClassifierBlockedError(WorkGraphExecuteError): + """CLASSIFIER_BLOCKED -- 03-C §3.5. Not retryable; routes to + blocked_pending_human, blocked_reason='classifier_blocked' (04-P).""" + + +class UsageLimitReachedError(WorkGraphExecuteError): + """USAGE_LIMIT_REACHED -- 04-P p_usage_limit_wait. Retryable with backoff; + see execute_work_graph_with_usage_limit_retry.""" + + def __init__(self, message, quota_limit_type): + super().__init__(message) + self.quota_limit_type = quota_limit_type + + +class UsageLimitWaitTimeoutError(WorkGraphExecuteError): + """p_usage_limit_wait exceeded its timeout -> blocked_pending_human, + blocked_reason='usage_limit_wait_timeout' (04-P §4.4.1).""" + + +def _quota_limit_type(text): + lowered = text.lower() + if "weekly" in lowered and "hit your" in lowered: + return "weekly" + if "opus" in lowered and ("limit" in lowered or "quota" in lowered): + return "opus" + if any(phrase in lowered for phrase in _USAGE_LIMIT_PHRASES): + return "session" + return None def write_workflow_script(target_folder_path, name, script_content): @@ -36,7 +84,9 @@ def execute_work_graph(target_folder_path, name, args, output_schema, timeout=60 No `--bare` (A-1 spike: skips credential/login discovery, unrelated to workflow presence). `--permission-mode auto`, never `bypassPermissions`/ `--dangerously-skip-permissions` (05 §5.3.1). Returns the parsed - `structured_output` dict. + `structured_output` dict. Raises ClassifierBlockedError or + UsageLimitReachedError instead of the generic WorkGraphExecuteError when + the response matches those specific signals. """ prompt = "Run /{} with args {}".format(name, json.dumps(args)) result = subprocess.run( @@ -53,15 +103,68 @@ def execute_work_graph(target_folder_path, name, args, output_schema, timeout=60 timeout=timeout, ) if result.returncode != 0: + quota_type = _quota_limit_type(result.stderr) + if quota_type: + raise UsageLimitReachedError( + "workflow {} exited {}: {}".format(name, result.returncode, result.stderr), quota_type + ) raise WorkGraphExecuteError( "workflow {} exited {}: {}".format(name, result.returncode, result.stderr) ) payload = json.loads(result.stdout) if payload.get("is_error"): + denials = payload.get("permission_denials") or [] + if len(denials) >= _CLASSIFIER_BLOCKED_THRESHOLD: + raise ClassifierBlockedError( + "workflow {} blocked by permission classifier ({} denials): {}".format( + name, len(denials), payload + ) + ) + quota_type = _quota_limit_type(str(payload.get("result", ""))) + if quota_type: + raise UsageLimitReachedError("workflow {} hit a usage limit: {}".format(name, payload), quota_type) raise WorkGraphExecuteError("workflow {} reported is_error: {}".format(name, payload)) structured_output = payload.get("structured_output") if structured_output is None: raise WorkGraphExecuteError("workflow {} produced no structured_output: {}".format(name, payload)) return structured_output + + +def execute_work_graph_with_usage_limit_retry( + target_folder_path, name, args, output_schema, timeout=600, + sleep=time.sleep, now=time.monotonic, +): + """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. + + `sleep`/`now` are injectable so tests don't block on real 5-30 minute waits. + """ + start = now() + backoff_index = 0 + while True: + try: + return execute_work_graph(target_folder_path, name, args, output_schema, timeout=timeout) + except UsageLimitReachedError as e: + if e.quota_limit_type == "opus": + raise UsageLimitWaitTimeoutError( + "opus quota exhausted -- caller should retry on a non-opus model: " + str(e) + ) from e + elapsed = now() - start + timeout_seconds = _TIMEOUT_SECONDS[e.quota_limit_type] + if elapsed >= timeout_seconds: + raise UsageLimitWaitTimeoutError( + "{} quota wait exceeded {}s timeout after {:.0f}s: {}".format( + e.quota_limit_type, timeout_seconds, elapsed, e + ) + ) from e + wait_minutes = _BACKOFF_SCHEDULE_MINUTES[min(backoff_index, len(_BACKOFF_SCHEDULE_MINUTES) - 1)] + sleep(wait_minutes * 60) + backoff_index += 1