diff --git a/docs/development-plan.md b/docs/development-plan.md index f467322..d7bd700 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -131,8 +131,9 @@ - **U-013 [Value Path]** `c_work_node_run` + `git_worktree` 隔离（并行写代码节点） - 依赖：U-012 - - Done：REQ-03 happy/boundary 通过（两模块并行编码使用独立 worktree，合并后代码无冲突） - - 覆盖：01-M `work_node_execution`；03-C `c_work_node_run`；REQ-03 + - Done（**2026-07-31 订正**：原来写"REQ-03 happy/boundary 通过"，但 happy 用例（"产出 code_diff 与 test_report...unit+integration test 全部通过"）需要真实 Coder 会话写代码 + `c_artifact_write`（U-011 已记录的缺口，至今没有 Unit 认领），U-013 单靠 U-012 做不到——第五次出现同一类问题）：REQ-03 **boundary** 通过（两个独立节点各自用 `git worktree add` 隔离，`run_fn` 各写一个文件并 commit，`merge --no-ff` 合并回主分支，`git log` 确认两次提交都在，且 `work_node_execution.git_worktree_ref` 各不相同）；happy 用例随 `c_artifact_write` 落地后累计满足 + - **精度提升（2026-07-31，实现后补充）**：03-C §3.6 给出了 `USAGE_LIMIT_REACHED` 的精确探测正则 `/hit your (session|weekly|Opus) limit/i`，比 U-010/U-012 当时用的短语近似匹配更准——U-013 按这条正则实现，且真正写回 `work_node_execution.quota_wait_started_at`/`quota_limit_type`（U-010/U-012 阶段这两张表字段还没有真实归属）。`c_work_node_run` 是内部接口（03-C 原话"不对外暴露 HTTP，供 c_work_graph_execute 调用"），退避重试从 U-012 的整图级别下沉到这里的单节点级别 + - 覆盖：01-M `work_node_execution`；03-C `c_work_node_run`；REQ-03（boundary） ### PH1-S3 端到端验收（收口） diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 42b84be..bf1dbb4 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -25,7 +25,7 @@ {"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 完整实现（见 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-013", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_node_run + git worktree 隔离（见 development-plan.md 2026-07-31 订正）", "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 boundary 通过（真实 git worktree 隔离+merge，无冲突）；happy 需 c_artifact_write 落地后累计满足"}, {"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 条全部达成"}, {"id": "U-015", "phase": "PH2", "sub_phase": "PH2-S1", "name": "p_validator_root_cause_trace 实现", "classification": "Risk Gate", "depends_on": ["U-013"], "coverage": {"model": ["bug_record"], "view": [], "controller": ["c_bug_record_create"], "process": ["p_validator_root_cause_trace"], "requirement": ["REQ-04"]}, "done": "REQ-04 happy/boundary/failure 通过"}, diff --git a/src/test_work_node_run.py b/src/test_work_node_run.py new file mode 100644 index 0000000..3011867 --- /dev/null +++ b/src/test_work_node_run.py @@ -0,0 +1,269 @@ +"""U-013 tests: c_work_node_run's contract (schema validation, retry state +machine, quota-wait, execution_log_file lifecycle) with a fake `run_fn`, plus +real git worktree isolation + merge -- REQ-03's boundary case (two independent +nodes, separate worktrees, merge without conflict) needs no LLM call at all, +so it's tested for real here, not mocked. + +Run: python3 -m unittest src.test_work_node_run -v +""" + +import json +import os +import subprocess +import tempfile +import unittest + +from src import schema, work_node_run + +_IN_SCHEMA = {"type": "object", "required": ["goal"]} +_OUT_SCHEMA = {"type": "object", "required": ["value"]} + + +def _init_git_repo(path): + subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True) + with open(os.path.join(path, "README.md"), "w") as f: + f.write("seed\n") + subprocess.run(["git", "add", "README.md"], cwd=path, check=True, capture_output=True) + subprocess.run( + ["git"] + work_node_run._GIT_IDENTITY + ["commit", "-m", "seed"], + cwd=path, check=True, capture_output=True, + ) + + +class WorkNodeRunTestBase(unittest.TestCase): + def setUp(self): + self.conn = schema.connect(":memory:") + schema.create_schema(self.conn) + self.conn.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" + ) + self.conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'coder')" + ) + self.conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')" + ) + self.conn.execute( + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " + "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')" + ) + self.conn.execute( + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, " + "input_schema_ref, output_schema_ref, input_payload, status) " + "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'pending')" + ) + self.conn.commit() + + def tearDown(self): + self.conn.close() + + +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} + ) + status = self.conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'").fetchone()[0] + self.assertEqual(status, "pending") + count = self.conn.execute("SELECT COUNT(*) FROM execution_log_file").fetchone()[0] + self.assertEqual(count, 0) + + +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.assertEqual(output, {"value": 42}) + row = self.conn.execute( + "SELECT status, output_payload, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone() + self.assertEqual(row[0], "done") + self.assertEqual(json.loads(row[1]), {"value": 42}) + self.assertEqual(row[2], 0) + + def test_execution_log_file_created_active_then_closed(self): + seen_status_during_run = [] + + def run_fn(cwd): + seen_status_during_run.append( + self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0] + ) + return {"value": 1} + + work_node_run.run_work_node(self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn) + self.assertEqual(seen_status_during_run, ["active"]) + final_status = self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0] + self.assertEqual(final_status, "closed") + + def test_log_file_closed_even_when_run_fn_raises(self): + def boom(cwd): + raise RuntimeError("node blew up") + + with self.assertRaises(RuntimeError): + work_node_run.run_work_node(self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=boom) + final_status = self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0] + self.assertEqual(final_status, "closed") + + +class RetryOnSchemaFailureTest(WorkNodeRunTestBase): + def test_retries_then_succeeds(self): + attempts = {"n": 0} + + def run_fn(cwd): + attempts["n"] += 1 + if attempts["n"] < 3: + return {"wrong_key": True} + return {"value": "ok"} + + output = work_node_run.run_work_node( + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn, max_retries=2 + ) + self.assertEqual(output, {"value": "ok"}) + row = self.conn.execute( + "SELECT status, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone() + self.assertEqual(row, ("done", 2)) + + def test_exceeding_max_retries_blocks_the_node(self): + 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, + ) + row = self.conn.execute( + "SELECT status, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone() + self.assertEqual(row, ("blocked", 3)) + + +class UsageLimitWaitTest(WorkNodeRunTestBase): + def setUp(self): + super().setUp() + 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_session_limit_then_recovers(self): + attempts = {"n": 0} + + def run_fn(cwd): + attempts["n"] += 1 + if attempts["n"] == 1: + work_node_run.raise_if_usage_limited("Error: You've hit your session limit") + return {"value": "recovered"} + + 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(output, {"value": "recovered"}) + self.assertEqual(self.sleeps, [5 * 60]) + + row = self.conn.execute( + "SELECT quota_limit_type FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone() + self.assertEqual(row[0], "session") + event = self.conn.execute( + "SELECT event_type FROM execution_event WHERE node_exec_id = 'node-1'" + ).fetchone() + self.assertEqual(event[0], "quota_wait_started") + + def test_opus_quota_raises_immediately_no_backoff(self): + def run_fn(cwd): + work_node_run.raise_if_usage_limited("You've hit your Opus 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.assertEqual(self.sleeps, []) + + def test_session_timeout_after_8_hours_blocks_node(self): + def run_fn(cwd): + work_node_run.raise_if_usage_limited("You've hit your session 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], 8 * 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, + merged back with no conflict. Real git, no LLM call needed.""" + + def setUp(self): + super().setUp() + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + _init_git_repo(self.tmp.name) + self.conn.execute( + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, " + "input_schema_ref, output_schema_ref, input_payload, status) " + "VALUES ('node-2', 'run-1', 'n2', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'pending')" + ) + self.conn.commit() + + def _write_file_run_fn(self, filename, content): + def run_fn(cwd): + 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) + subprocess.run( + ["git"] + work_node_run._GIT_IDENTITY + ["commit", "-m", "add " + filename], + cwd=cwd, check=True, capture_output=True, + ) + return {"value": filename} + + return run_fn + + def test_two_independent_nodes_isolated_and_merged_without_conflict(self): + work_node_run.run_work_node( + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, + run_fn=self._write_file_run_fn("module_a.txt", "module a content"), + needs_worktree=True, repo_path=self.tmp.name, + ) + work_node_run.run_work_node( + self.conn, "node-2", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, + run_fn=self._write_file_run_fn("module_b.txt", "module b content"), + needs_worktree=True, repo_path=self.tmp.name, + ) + + self.assertTrue(os.path.isfile(os.path.join(self.tmp.name, "module_a.txt"))) + self.assertTrue(os.path.isfile(os.path.join(self.tmp.name, "module_b.txt"))) + + row1 = self.conn.execute( + "SELECT git_worktree_ref FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone()[0] + row2 = self.conn.execute( + "SELECT git_worktree_ref FROM work_node_execution WHERE node_exec_id = 'node-2'" + ).fetchone()[0] + self.assertNotEqual(row1, row2) + self.assertIn("node-1", row1) + self.assertIn("node-2", row2) + + log = subprocess.run( + ["git", "log", "--oneline"], cwd=self.tmp.name, check=True, capture_output=True, text=True + ) + self.assertIn("module_a.txt", log.stdout) + self.assertIn("module_b.txt", log.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/work_node_run.py b/src/work_node_run.py new file mode 100644 index 0000000..ab51991 --- /dev/null +++ b/src/work_node_run.py @@ -0,0 +1,259 @@ +"""c_work_node_run (03-C §3.6): work node execution + git worktree isolation. + +Internal interface (not exposed over HTTP -- called by `c_work_graph_execute` +for each node needing isolated execution, per 03-C §3.6's own framing). +Validates input/output against a minimal inline JSON-Schema-lite check (no +third-party `jsonschema` dependency, per 05 §5.7 -- covers object/required, +which is all this codebase's schemas use so far), transitions +`work_node_execution.status` through +`pending -> running -> (reflecting -> retrying, up to a caller-specified cap) +-> done|blocked`, and handles `USAGE_LIMIT_REACHED` using the *exact* +detection regex 03-C §3.6 documents (`hit your (session|weekly|opus) limit`, +case-insensitive) -- more precise than the approximate phrase-matching +U-010/U-012 used before this Unit existed to give the behavior a real, +persisted home (`work_node_execution.quota_wait_started_at`/`quota_limit_type`). + +U-013 scope matches REQ-03's boundary case, which is what this Unit's Done +criterion actually describes: two independent nodes running in separate git +worktrees, merged back without conflict. REQ-03's happy case (real +code_diff/test_report artifacts, unit+integration tests genuinely passing) +needs `c_artifact_write` (still unowned by any Unit, see U-011's note) and a +real Coder session actually writing code -- not attempted here, same +overreach pattern already corrected for U-009/U-010/U-011/U-012. + +The "actual work" a node does is inherently node-specific (that's the whole +point of "give each node a contract") -- `run_fn` is an explicit, required +dependency (no hidden default calling out to `claude`), so this module +stays about the *contract* (schema, retries, worktree, logging, quota-wait), +not about any one node's content. +""" + +import json +import os +import re +import subprocess +import time +import uuid + +from src import paths + +_USAGE_LIMIT_RE = re.compile(r"hit your (session|weekly|opus) limit", re.IGNORECASE) +_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} +_GIT_IDENTITY = ["-c", "user.name=graphlemon", "-c", "user.email=graphlemon@localhost"] + + +class SchemaValidationFailedError(Exception): + """SCHEMA_VALIDATION_FAILED -- 03-C §3.6 (output validation; input + validation failure raises InvalidInputError instead, see below -- 03-C + describes input rejection as immediate/non-retryable, distinct from the + retryable output-validation-failure case).""" + + +class InvalidInputError(Exception): + """input_payload doesn't match input_schema_ref -- 03-C §3.6: "校验失败 + 直接拒绝，不进入执行" (rejected outright, execution never starts).""" + + +class UsageLimitWaitTimeoutError(Exception): + """p_usage_limit_wait exceeded its timeout (04-P §4.4.1). work_node_execution + has no 'blocked_pending_human' status of its own (01-M 1.4's enum doesn't + include it) -- this function sets status='blocked' and raises; bubbling to + task.status='blocked_pending_human' is the caller's job, same layering as + c_work_node_run itself not knowing about `task` at all (03-C §3.6).""" + + +def _validate_schema_lite(payload, schema): + """object/required only -- everything this codebase's schemas use so far.""" + if schema.get("type") == "object": + if not isinstance(payload, dict): + return False + for key in schema.get("required", []): + if key not in payload: + return False + return True + + +def _quota_limit_type(text): + match = _USAGE_LIMIT_RE.search(text or "") + return match.group(1).lower() if match else None + + +def create_worktree(repo_path, node_exec_id): + """`git worktree add` for an isolated code-writing node (01-M + work_node_execution.git_worktree_ref).""" + worktree_path = os.path.join(repo_path, ".graphlemon-worktrees", node_exec_id) + branch = "graphlemon/" + node_exec_id + subprocess.run( + ["git", "worktree", "add", "-b", branch, worktree_path], + cwd=repo_path, check=True, capture_output=True, text=True, + ) + return worktree_path + + +def merge_worktree(repo_path, node_exec_id): + """Merges the node's branch back into whichever branch is currently + checked out in the main worktree at repo_path.""" + branch = "graphlemon/" + node_exec_id + subprocess.run( + _prefix_git(["merge", "--no-ff", branch, "-m", "graphlemon: merge " + node_exec_id]), + cwd=repo_path, check=True, capture_output=True, text=True, + ) + + +def remove_worktree(repo_path, worktree_path): + subprocess.run( + ["git", "worktree", "remove", "--force", worktree_path], + cwd=repo_path, check=True, capture_output=True, text=True, + ) + + +def _prefix_git(args): + return ["git"] + _GIT_IDENTITY + args + + +def run_work_node( + conn, + node_exec_id, + input_payload, + input_schema, + output_schema, + run_fn, + max_retries=2, + needs_worktree=False, + repo_path=None, + sleep=time.sleep, + now=time.monotonic, +): + """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. + + Raises InvalidInputError (input schema), UsageLimitWaitTimeoutError + (p_usage_limit_wait exhausted), or leaves work_node_execution.status + ='blocked' with a SchemaValidationFailedError raised once max_retries is + exceeded on output validation. + """ + if not _validate_schema_lite(input_payload, input_schema): + raise InvalidInputError("input_payload does not match input_schema_ref for " + node_exec_id) + + project_id, task_id = conn.execute( + "SELECT p.project_id, wgr.task_id FROM work_node_execution wne " + "JOIN work_graph_run wgr ON wgr.run_id = wne.run_id " + "JOIN task t ON t.task_id = wgr.task_id " + "JOIN project p ON p.project_id = t.project_id " + "WHERE wne.node_exec_id = ?", + (node_exec_id,), + ).fetchone() + + log_file_id = str(uuid.uuid4()) + log_path = paths.log_file_path(project_id, task_id, node_exec_id, node_exec_id) + conn.execute( + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) " + "VALUES (?, ?, ?, ?, 'jsonl', 'active')", + (log_file_id, node_exec_id, project_id, log_path), + ) + conn.execute("UPDATE work_node_execution SET status = 'running' WHERE node_exec_id = ?", (node_exec_id,)) + conn.commit() + + worktree_path = None + try: + if needs_worktree: + worktree_path = create_worktree(repo_path, node_exec_id) + conn.execute( + "UPDATE work_node_execution SET git_worktree_ref = ? WHERE node_exec_id = ?", + (worktree_path, node_exec_id), + ) + conn.commit() + cwd = worktree_path or repo_path + + retry_count = 0 + quota_wait_start = None + while True: + try: + output_payload = run_fn(cwd) + except _UsageLimitSignal as signal: + quota_type = signal.quota_limit_type + if quota_wait_start is None: + quota_wait_start = now() + conn.execute( + "UPDATE work_node_execution SET status = 'waiting_for_quota', " + "quota_wait_started_at = CURRENT_TIMESTAMP, quota_limit_type = ? WHERE node_exec_id = ?", + (quota_type, node_exec_id), + ) + conn.execute( + "INSERT INTO execution_event (event_id, project_id, task_id, node_exec_id, event_type, payload) " + "VALUES (?, ?, ?, ?, 'quota_wait_started', ?)", + (str(uuid.uuid4()), project_id, task_id, node_exec_id, json.dumps({"quota_limit_type": quota_type})), + ) + 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 + elapsed = now() - quota_wait_start + if elapsed >= _TIMEOUT_SECONDS[quota_type]: + conn.execute( + "UPDATE work_node_execution SET status = 'blocked' WHERE node_exec_id = ?", (node_exec_id,) + ) + conn.commit() + raise UsageLimitWaitTimeoutError( + "{} quota wait exceeded timeout for {}".format(quota_type, node_exec_id) + ) from signal + backoff_index = min(int(elapsed // (30 * 60)), len(_BACKOFF_SCHEDULE_MINUTES) - 1) + sleep(_BACKOFF_SCHEDULE_MINUTES[backoff_index] * 60) + continue + + if _validate_schema_lite(output_payload, output_schema): + conn.execute( + "UPDATE work_node_execution SET status = 'done', output_payload = ?, retry_count = ? " + "WHERE node_exec_id = ?", + (json.dumps(output_payload), retry_count, node_exec_id), + ) + conn.commit() + if worktree_path: + merge_worktree(repo_path, node_exec_id) + remove_worktree(repo_path, worktree_path) # 03-C §3.6: "执行完成后清理 worktree" + return output_payload + + retry_count += 1 + if retry_count > max_retries: + conn.execute( + "UPDATE work_node_execution SET status = 'blocked', retry_count = ? WHERE node_exec_id = ?", + (retry_count, node_exec_id), + ) + conn.commit() + raise SchemaValidationFailedError( + "output_payload failed schema validation for {} after {} retries".format( + node_exec_id, max_retries + ) + ) + conn.execute( + "UPDATE work_node_execution SET status = 'retrying', retry_count = ? WHERE node_exec_id = ?", + (retry_count, node_exec_id), + ) + conn.commit() + finally: + conn.execute( + "UPDATE execution_log_file SET status = 'closed' WHERE log_file_id = ?", (log_file_id,) + ) + conn.commit() + + +class _UsageLimitSignal(Exception): + """Internal: run_fn raises this (via raise_if_usage_limited helper) to + signal a quota hit without run_work_node having to know how run_fn talks + to `claude`.""" + + def __init__(self, quota_limit_type): + super().__init__(quota_limit_type) + self.quota_limit_type = quota_limit_type + + +def raise_if_usage_limited(text): + """Call from inside a `run_fn` after a failed claude invocation: raises + _UsageLimitSignal if `text` matches 03-C §3.6's exact detection regex, + otherwise does nothing (let the caller re-raise its own error).""" + quota_type = _quota_limit_type(text) + if quota_type: + raise _UsageLimitSignal(quota_type)