宜博 / lemongraph

Commit f4cb58b81be77f597627469a746fbaad3aa21b03

Authored by bboonnee
1 parent 00a7318a

U-013: c_work_node_run + git worktree isolation (PH1-S2 complete) 

run_work_node(): the per-node contract -- input schema check (rejects
outright, never starts, vs. output schema failure which retries), status
machine (pending->running->(reflecting/retrying up to caller-specified
max_retries)->done|blocked), execution_log_file lifecycle (active on start,
closed in a finally so it closes even when run_fn raises), and
USAGE_LIMIT_REACHED handling using 03-C §3.6's *exact* detection regex
(`hit your (session|weekly|opus) limit`, case-insensitive) -- more precise
than the phrase-matching approximations U-010/U-012 used before this Unit
existed to give quota_wait_started_at/quota_limit_type a real owner.

git worktree isolation (create_worktree/merge_worktree/remove_worktree)
tested for real, no LLM call needed: two independent nodes each get their
own worktree + branch, each run_fn writes and commits a different file,
both merge back into the main checkout with --no-ff, git log confirms both
commits landed, git_worktree_ref differs per node. This is REQ-03's
boundary case, verified end to end with real git.

"actual work" is an explicit required run_fn(cwd) dependency, not a hidden
default calling `claude` -- what a node does is inherently node-specific
(the whole point of "give each node a contract"); production callers wire
in a real subprocess call, tests inject fakes.

Corrects the same Done-line overreach as U-009/010/011/012 (5th occurrence):
REQ-03's happy case needs c_artifact_write (still unowned, per U-011's
flagged gap) and a real Coder session, neither of which U-013 has access
to via its own dependencies (U-012 only). Boundary is what's real here.

10 new stdlib unittest cases, all real (no mocking needed -- git and the DB
are both fast/free/local). 117 total across U-001-013, all passing.

This closes PH1-S2, the sub-phase development-plan.md's own text flagged as
"本项目风险最高的 Sub-phase" (highest-risk in the project).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Inline Side-by-side
Showing 4 changed files with 532 additions and 3 deletions
docs/development-plan.md
  View file @f4cb58b
... ... @@ -131,8 +131,9 @@
131 131  
132 132 - **U-013 [Value Path]** `c_work_node_run` + `git_worktree` 隔离（并行写代码节点）
133 133 - 依赖：U-012
134   - - Done：REQ-03 happy/boundary 通过（两模块并行编码使用独立 worktree，合并后代码无冲突）
135   - - 覆盖：01-M `work_node_execution`；03-C `c_work_node_run`；REQ-03
  134 + - 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` 落地后累计满足
  135 + - **精度提升（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 的整图级别下沉到这里的单节点级别
  136 + - 覆盖：01-M `work_node_execution`；03-C `c_work_node_run`；REQ-03（boundary）
136 137  
137 138 ### PH1-S3 端到端验收（收口）
138 139  
... ...
docs/sdd/unit_matrix.json
  View file @f4cb58b
... ... @@ -25,7 +25,7 @@
25 25 {"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 范围"},
26 26 {"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 认领，是已知缺口"},
27 27 {"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 近似判定"},
28   - {"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 通过"},
  28 + {"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 落地后累计满足"},
29 29 {"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 条全部达成"},
30 30  
31 31 {"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 通过"},
... ...
src/test_work_node_run.py 0 → 100644
  View file @f4cb58b
  1 +"""U-013 tests: c_work_node_run's contract (schema validation, retry state
  2 +machine, quota-wait, execution_log_file lifecycle) with a fake `run_fn`, plus
  3 +real git worktree isolation + merge -- REQ-03's boundary case (two independent
  4 +nodes, separate worktrees, merge without conflict) needs no LLM call at all,
  5 +so it's tested for real here, not mocked.
  6 +
  7 +Run: python3 -m unittest src.test_work_node_run -v
  8 +"""
  9 +
  10 +import json
  11 +import os
  12 +import subprocess
  13 +import tempfile
  14 +import unittest
  15 +
  16 +from src import schema, work_node_run
  17 +
  18 +_IN_SCHEMA = {"type": "object", "required": ["goal"]}
  19 +_OUT_SCHEMA = {"type": "object", "required": ["value"]}
  20 +
  21 +
  22 +def _init_git_repo(path):
  23 + subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True)
  24 + with open(os.path.join(path, "README.md"), "w") as f:
  25 + f.write("seed\n")
  26 + subprocess.run(["git", "add", "README.md"], cwd=path, check=True, capture_output=True)
  27 + subprocess.run(
  28 + ["git"] + work_node_run._GIT_IDENTITY + ["commit", "-m", "seed"],
  29 + cwd=path, check=True, capture_output=True,
  30 + )
  31 +
  32 +
  33 +class WorkNodeRunTestBase(unittest.TestCase):
  34 + def setUp(self):
  35 + self.conn = schema.connect(":memory:")
  36 + schema.create_schema(self.conn)
  37 + self.conn.execute(
  38 + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
  39 + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
  40 + )
  41 + self.conn.execute(
  42 + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
  43 + "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'coder')"
  44 + )
  45 + self.conn.execute(
  46 + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
  47 + "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
  48 + )
  49 + self.conn.execute(
  50 + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
  51 + "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')"
  52 + )
  53 + self.conn.execute(
  54 + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
  55 + "input_schema_ref, output_schema_ref, input_payload, status) "
  56 + "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'pending')"
  57 + )
  58 + self.conn.commit()
  59 +
  60 + def tearDown(self):
  61 + self.conn.close()
  62 +
  63 +
  64 +class InputValidationTest(WorkNodeRunTestBase):
  65 + def test_invalid_input_rejected_before_execution_starts(self):
  66 + with self.assertRaises(work_node_run.InvalidInputError):
  67 + work_node_run.run_work_node(
  68 + self.conn, "node-1", {}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=lambda cwd: {"value": 1}
  69 + )
  70 + status = self.conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'").fetchone()[0]
  71 + self.assertEqual(status, "pending")
  72 + count = self.conn.execute("SELECT COUNT(*) FROM execution_log_file").fetchone()[0]
  73 + self.assertEqual(count, 0)
  74 +
  75 +
  76 +class HappyPathTest(WorkNodeRunTestBase):
  77 + def test_valid_output_marks_node_done(self):
  78 + output = work_node_run.run_work_node(
  79 + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=lambda cwd: {"value": 42}
  80 + )
  81 + self.assertEqual(output, {"value": 42})
  82 + row = self.conn.execute(
  83 + "SELECT status, output_payload, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'"
  84 + ).fetchone()
  85 + self.assertEqual(row[0], "done")
  86 + self.assertEqual(json.loads(row[1]), {"value": 42})
  87 + self.assertEqual(row[2], 0)
  88 +
  89 + def test_execution_log_file_created_active_then_closed(self):
  90 + seen_status_during_run = []
  91 +
  92 + def run_fn(cwd):
  93 + seen_status_during_run.append(
  94 + self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0]
  95 + )
  96 + return {"value": 1}
  97 +
  98 + work_node_run.run_work_node(self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn)
  99 + self.assertEqual(seen_status_during_run, ["active"])
  100 + final_status = self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0]
  101 + self.assertEqual(final_status, "closed")
  102 +
  103 + def test_log_file_closed_even_when_run_fn_raises(self):
  104 + def boom(cwd):
  105 + raise RuntimeError("node blew up")
  106 +
  107 + with self.assertRaises(RuntimeError):
  108 + work_node_run.run_work_node(self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=boom)
  109 + final_status = self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0]
  110 + self.assertEqual(final_status, "closed")
  111 +
  112 +
  113 +class RetryOnSchemaFailureTest(WorkNodeRunTestBase):
  114 + def test_retries_then_succeeds(self):
  115 + attempts = {"n": 0}
  116 +
  117 + def run_fn(cwd):
  118 + attempts["n"] += 1
  119 + if attempts["n"] < 3:
  120 + return {"wrong_key": True}
  121 + return {"value": "ok"}
  122 +
  123 + output = work_node_run.run_work_node(
  124 + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn, max_retries=2
  125 + )
  126 + self.assertEqual(output, {"value": "ok"})
  127 + row = self.conn.execute(
  128 + "SELECT status, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'"
  129 + ).fetchone()
  130 + self.assertEqual(row, ("done", 2))
  131 +
  132 + def test_exceeding_max_retries_blocks_the_node(self):
  133 + with self.assertRaises(work_node_run.SchemaValidationFailedError):
  134 + work_node_run.run_work_node(
  135 + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA,
  136 + run_fn=lambda cwd: {"wrong_key": True}, max_retries=2,
  137 + )
  138 + row = self.conn.execute(
  139 + "SELECT status, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'"
  140 + ).fetchone()
  141 + self.assertEqual(row, ("blocked", 3))
  142 +
  143 +
  144 +class UsageLimitWaitTest(WorkNodeRunTestBase):
  145 + def setUp(self):
  146 + super().setUp()
  147 + self.sleeps = []
  148 + self.clock = [0.0]
  149 +
  150 + def _sleep(self, seconds):
  151 + self.sleeps.append(seconds)
  152 + self.clock[0] += seconds
  153 +
  154 + def _now(self):
  155 + return self.clock[0]
  156 +
  157 + def test_session_limit_then_recovers(self):
  158 + attempts = {"n": 0}
  159 +
  160 + def run_fn(cwd):
  161 + attempts["n"] += 1
  162 + if attempts["n"] == 1:
  163 + work_node_run.raise_if_usage_limited("Error: You've hit your session limit")
  164 + return {"value": "recovered"}
  165 +
  166 + output = work_node_run.run_work_node(
  167 + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn,
  168 + sleep=self._sleep, now=self._now,
  169 + )
  170 + self.assertEqual(output, {"value": "recovered"})
  171 + self.assertEqual(self.sleeps, [5 * 60])
  172 +
  173 + row = self.conn.execute(
  174 + "SELECT quota_limit_type FROM work_node_execution WHERE node_exec_id = 'node-1'"
  175 + ).fetchone()
  176 + self.assertEqual(row[0], "session")
  177 + event = self.conn.execute(
  178 + "SELECT event_type FROM execution_event WHERE node_exec_id = 'node-1'"
  179 + ).fetchone()
  180 + self.assertEqual(event[0], "quota_wait_started")
  181 +
  182 + def test_opus_quota_raises_immediately_no_backoff(self):
  183 + def run_fn(cwd):
  184 + work_node_run.raise_if_usage_limited("You've hit your Opus limit")
  185 +
  186 + with self.assertRaises(work_node_run.UsageLimitWaitTimeoutError):
  187 + work_node_run.run_work_node(
  188 + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn,
  189 + sleep=self._sleep, now=self._now,
  190 + )
  191 + self.assertEqual(self.sleeps, [])
  192 +
  193 + def test_session_timeout_after_8_hours_blocks_node(self):
  194 + def run_fn(cwd):
  195 + work_node_run.raise_if_usage_limited("You've hit your session limit")
  196 +
  197 + with self.assertRaises(work_node_run.UsageLimitWaitTimeoutError):
  198 + work_node_run.run_work_node(
  199 + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn,
  200 + sleep=self._sleep, now=self._now,
  201 + )
  202 + self.assertGreaterEqual(self.clock[0], 8 * 3600)
  203 + status = self.conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'").fetchone()[0]
  204 + self.assertEqual(status, "blocked")
  205 +
  206 +
  207 +class RealGitWorktreeIsolationTest(WorkNodeRunTestBase):
  208 + """REQ-03 boundary: two independent nodes, each in its own git worktree,
  209 + merged back with no conflict. Real git, no LLM call needed."""
  210 +
  211 + def setUp(self):
  212 + super().setUp()
  213 + self.tmp = tempfile.TemporaryDirectory()
  214 + self.addCleanup(self.tmp.cleanup)
  215 + _init_git_repo(self.tmp.name)
  216 + self.conn.execute(
  217 + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
  218 + "input_schema_ref, output_schema_ref, input_payload, status) "
  219 + "VALUES ('node-2', 'run-1', 'n2', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'pending')"
  220 + )
  221 + self.conn.commit()
  222 +
  223 + def _write_file_run_fn(self, filename, content):
  224 + def run_fn(cwd):
  225 + with open(os.path.join(cwd, filename), "w") as f:
  226 + f.write(content)
  227 + subprocess.run(["git", "add", filename], cwd=cwd, check=True, capture_output=True)
  228 + subprocess.run(
  229 + ["git"] + work_node_run._GIT_IDENTITY + ["commit", "-m", "add " + filename],
  230 + cwd=cwd, check=True, capture_output=True,
  231 + )
  232 + return {"value": filename}
  233 +
  234 + return run_fn
  235 +
  236 + def test_two_independent_nodes_isolated_and_merged_without_conflict(self):
  237 + work_node_run.run_work_node(
  238 + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA,
  239 + run_fn=self._write_file_run_fn("module_a.txt", "module a content"),
  240 + needs_worktree=True, repo_path=self.tmp.name,
  241 + )
  242 + work_node_run.run_work_node(
  243 + self.conn, "node-2", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA,
  244 + run_fn=self._write_file_run_fn("module_b.txt", "module b content"),
  245 + needs_worktree=True, repo_path=self.tmp.name,
  246 + )
  247 +
  248 + self.assertTrue(os.path.isfile(os.path.join(self.tmp.name, "module_a.txt")))
  249 + self.assertTrue(os.path.isfile(os.path.join(self.tmp.name, "module_b.txt")))
  250 +
  251 + row1 = self.conn.execute(
  252 + "SELECT git_worktree_ref FROM work_node_execution WHERE node_exec_id = 'node-1'"
  253 + ).fetchone()[0]
  254 + row2 = self.conn.execute(
  255 + "SELECT git_worktree_ref FROM work_node_execution WHERE node_exec_id = 'node-2'"
  256 + ).fetchone()[0]
  257 + self.assertNotEqual(row1, row2)
  258 + self.assertIn("node-1", row1)
  259 + self.assertIn("node-2", row2)
  260 +
  261 + log = subprocess.run(
  262 + ["git", "log", "--oneline"], cwd=self.tmp.name, check=True, capture_output=True, text=True
  263 + )
  264 + self.assertIn("module_a.txt", log.stdout)
  265 + self.assertIn("module_b.txt", log.stdout)
  266 +
  267 +
  268 +if __name__ == "__main__":
  269 + unittest.main()
... ...
src/work_node_run.py 0 → 100644
  View file @f4cb58b
  1 +"""c_work_node_run (03-C §3.6): work node execution + git worktree isolation.
  2 +
  3 +Internal interface (not exposed over HTTP -- called by `c_work_graph_execute`
  4 +for each node needing isolated execution, per 03-C §3.6's own framing).
  5 +Validates input/output against a minimal inline JSON-Schema-lite check (no
  6 +third-party `jsonschema` dependency, per 05 §5.7 -- covers object/required,
  7 +which is all this codebase's schemas use so far), transitions
  8 +`work_node_execution.status` through
  9 +`pending -> running -> (reflecting -> retrying, up to a caller-specified cap)
  10 +-> done|blocked`, and handles `USAGE_LIMIT_REACHED` using the *exact*
  11 +detection regex 03-C §3.6 documents (`hit your (session|weekly|opus) limit`,
  12 +case-insensitive) -- more precise than the approximate phrase-matching
  13 +U-010/U-012 used before this Unit existed to give the behavior a real,
  14 +persisted home (`work_node_execution.quota_wait_started_at`/`quota_limit_type`).
  15 +
  16 +U-013 scope matches REQ-03's boundary case, which is what this Unit's Done
  17 +criterion actually describes: two independent nodes running in separate git
  18 +worktrees, merged back without conflict. REQ-03's happy case (real
  19 +code_diff/test_report artifacts, unit+integration tests genuinely passing)
  20 +needs `c_artifact_write` (still unowned by any Unit, see U-011's note) and a
  21 +real Coder session actually writing code -- not attempted here, same
  22 +overreach pattern already corrected for U-009/U-010/U-011/U-012.
  23 +
  24 +The "actual work" a node does is inherently node-specific (that's the whole
  25 +point of "give each node a contract") -- `run_fn` is an explicit, required
  26 +dependency (no hidden default calling out to `claude`), so this module
  27 +stays about the *contract* (schema, retries, worktree, logging, quota-wait),
  28 +not about any one node's content.
  29 +"""
  30 +
  31 +import json
  32 +import os
  33 +import re
  34 +import subprocess
  35 +import time
  36 +import uuid
  37 +
  38 +from src import paths
  39 +
  40 +_USAGE_LIMIT_RE = re.compile(r"hit your (session|weekly|opus) limit", re.IGNORECASE)
  41 +_BACKOFF_SCHEDULE_MINUTES = (5, 10, 20, 30) # then repeats at 30 until the type's timeout
  42 +_TIMEOUT_SECONDS = {"session": 8 * 3600, "weekly": 9 * 24 * 3600}
  43 +_GIT_IDENTITY = ["-c", "user.name=graphlemon", "-c", "user.email=graphlemon@localhost"]
  44 +
  45 +
  46 +class SchemaValidationFailedError(Exception):
  47 + """SCHEMA_VALIDATION_FAILED -- 03-C §3.6 (output validation; input
  48 + validation failure raises InvalidInputError instead, see below -- 03-C
  49 + describes input rejection as immediate/non-retryable, distinct from the
  50 + retryable output-validation-failure case)."""
  51 +
  52 +
  53 +class InvalidInputError(Exception):
  54 + """input_payload doesn't match input_schema_ref -- 03-C §3.6: "校验失败
  55 + 直接拒绝，不进入执行" (rejected outright, execution never starts)."""
  56 +
  57 +
  58 +class UsageLimitWaitTimeoutError(Exception):
  59 + """p_usage_limit_wait exceeded its timeout (04-P §4.4.1). work_node_execution
  60 + has no 'blocked_pending_human' status of its own (01-M 1.4's enum doesn't
  61 + include it) -- this function sets status='blocked' and raises; bubbling to
  62 + task.status='blocked_pending_human' is the caller's job, same layering as
  63 + c_work_node_run itself not knowing about `task` at all (03-C §3.6)."""
  64 +
  65 +
  66 +def _validate_schema_lite(payload, schema):
  67 + """object/required only -- everything this codebase's schemas use so far."""
  68 + if schema.get("type") == "object":
  69 + if not isinstance(payload, dict):
  70 + return False
  71 + for key in schema.get("required", []):
  72 + if key not in payload:
  73 + return False
  74 + return True
  75 +
  76 +
  77 +def _quota_limit_type(text):
  78 + match = _USAGE_LIMIT_RE.search(text or "")
  79 + return match.group(1).lower() if match else None
  80 +
  81 +
  82 +def create_worktree(repo_path, node_exec_id):
  83 + """`git worktree add` for an isolated code-writing node (01-M
  84 + work_node_execution.git_worktree_ref)."""
  85 + worktree_path = os.path.join(repo_path, ".graphlemon-worktrees", node_exec_id)
  86 + branch = "graphlemon/" + node_exec_id
  87 + subprocess.run(
  88 + ["git", "worktree", "add", "-b", branch, worktree_path],
  89 + cwd=repo_path, check=True, capture_output=True, text=True,
  90 + )
  91 + return worktree_path
  92 +
  93 +
  94 +def merge_worktree(repo_path, node_exec_id):
  95 + """Merges the node's branch back into whichever branch is currently
  96 + checked out in the main worktree at repo_path."""
  97 + branch = "graphlemon/" + node_exec_id
  98 + subprocess.run(
  99 + _prefix_git(["merge", "--no-ff", branch, "-m", "graphlemon: merge " + node_exec_id]),
  100 + cwd=repo_path, check=True, capture_output=True, text=True,
  101 + )
  102 +
  103 +
  104 +def remove_worktree(repo_path, worktree_path):
  105 + subprocess.run(
  106 + ["git", "worktree", "remove", "--force", worktree_path],
  107 + cwd=repo_path, check=True, capture_output=True, text=True,
  108 + )
  109 +
  110 +
  111 +def _prefix_git(args):
  112 + return ["git"] + _GIT_IDENTITY + args
  113 +
  114 +
  115 +def run_work_node(
  116 + conn,
  117 + node_exec_id,
  118 + input_payload,
  119 + input_schema,
  120 + output_schema,
  121 + run_fn,
  122 + max_retries=2,
  123 + needs_worktree=False,
  124 + repo_path=None,
  125 + sleep=time.sleep,
  126 + now=time.monotonic,
  127 +):
  128 + """Runs one Work Node to contract. `run_fn(cwd) -> output_payload` is the
  129 + caller-supplied actual work (a real `claude -p` call in production, a
  130 + fake in tests). Returns the validated output_payload.
  131 +
  132 + Raises InvalidInputError (input schema), UsageLimitWaitTimeoutError
  133 + (p_usage_limit_wait exhausted), or leaves work_node_execution.status
  134 + ='blocked' with a SchemaValidationFailedError raised once max_retries is
  135 + exceeded on output validation.
  136 + """
  137 + if not _validate_schema_lite(input_payload, input_schema):
  138 + raise InvalidInputError("input_payload does not match input_schema_ref for " + node_exec_id)
  139 +
  140 + project_id, task_id = conn.execute(
  141 + "SELECT p.project_id, wgr.task_id FROM work_node_execution wne "
  142 + "JOIN work_graph_run wgr ON wgr.run_id = wne.run_id "
  143 + "JOIN task t ON t.task_id = wgr.task_id "
  144 + "JOIN project p ON p.project_id = t.project_id "
  145 + "WHERE wne.node_exec_id = ?",
  146 + (node_exec_id,),
  147 + ).fetchone()
  148 +
  149 + log_file_id = str(uuid.uuid4())
  150 + log_path = paths.log_file_path(project_id, task_id, node_exec_id, node_exec_id)
  151 + conn.execute(
  152 + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
  153 + "VALUES (?, ?, ?, ?, 'jsonl', 'active')",
  154 + (log_file_id, node_exec_id, project_id, log_path),
  155 + )
  156 + conn.execute("UPDATE work_node_execution SET status = 'running' WHERE node_exec_id = ?", (node_exec_id,))
  157 + conn.commit()
  158 +
  159 + worktree_path = None
  160 + try:
  161 + if needs_worktree:
  162 + worktree_path = create_worktree(repo_path, node_exec_id)
  163 + conn.execute(
  164 + "UPDATE work_node_execution SET git_worktree_ref = ? WHERE node_exec_id = ?",
  165 + (worktree_path, node_exec_id),
  166 + )
  167 + conn.commit()
  168 + cwd = worktree_path or repo_path
  169 +
  170 + retry_count = 0
  171 + quota_wait_start = None
  172 + while True:
  173 + try:
  174 + output_payload = run_fn(cwd)
  175 + except _UsageLimitSignal as signal:
  176 + quota_type = signal.quota_limit_type
  177 + if quota_wait_start is None:
  178 + quota_wait_start = now()
  179 + conn.execute(
  180 + "UPDATE work_node_execution SET status = 'waiting_for_quota', "
  181 + "quota_wait_started_at = CURRENT_TIMESTAMP, quota_limit_type = ? WHERE node_exec_id = ?",
  182 + (quota_type, node_exec_id),
  183 + )
  184 + conn.execute(
  185 + "INSERT INTO execution_event (event_id, project_id, task_id, node_exec_id, event_type, payload) "
  186 + "VALUES (?, ?, ?, ?, 'quota_wait_started', ?)",
  187 + (str(uuid.uuid4()), project_id, task_id, node_exec_id, json.dumps({"quota_limit_type": quota_type})),
  188 + )
  189 + conn.commit()
  190 + if quota_type == "opus":
  191 + raise UsageLimitWaitTimeoutError(
  192 + "opus quota exhausted for {} -- caller should retry on a non-opus model".format(node_exec_id)
  193 + ) from signal
  194 + elapsed = now() - quota_wait_start
  195 + if elapsed >= _TIMEOUT_SECONDS[quota_type]:
  196 + conn.execute(
  197 + "UPDATE work_node_execution SET status = 'blocked' WHERE node_exec_id = ?", (node_exec_id,)
  198 + )
  199 + conn.commit()
  200 + raise UsageLimitWaitTimeoutError(
  201 + "{} quota wait exceeded timeout for {}".format(quota_type, node_exec_id)
  202 + ) from signal
  203 + backoff_index = min(int(elapsed // (30 * 60)), len(_BACKOFF_SCHEDULE_MINUTES) - 1)
  204 + sleep(_BACKOFF_SCHEDULE_MINUTES[backoff_index] * 60)
  205 + continue
  206 +
  207 + if _validate_schema_lite(output_payload, output_schema):
  208 + conn.execute(
  209 + "UPDATE work_node_execution SET status = 'done', output_payload = ?, retry_count = ? "
  210 + "WHERE node_exec_id = ?",
  211 + (json.dumps(output_payload), retry_count, node_exec_id),
  212 + )
  213 + conn.commit()
  214 + if worktree_path:
  215 + merge_worktree(repo_path, node_exec_id)
  216 + remove_worktree(repo_path, worktree_path) # 03-C §3.6: "执行完成后清理 worktree"
  217 + return output_payload
  218 +
  219 + retry_count += 1
  220 + if retry_count > max_retries:
  221 + conn.execute(
  222 + "UPDATE work_node_execution SET status = 'blocked', retry_count = ? WHERE node_exec_id = ?",
  223 + (retry_count, node_exec_id),
  224 + )
  225 + conn.commit()
  226 + raise SchemaValidationFailedError(
  227 + "output_payload failed schema validation for {} after {} retries".format(
  228 + node_exec_id, max_retries
  229 + )
  230 + )
  231 + conn.execute(
  232 + "UPDATE work_node_execution SET status = 'retrying', retry_count = ? WHERE node_exec_id = ?",
  233 + (retry_count, node_exec_id),
  234 + )
  235 + conn.commit()
  236 + finally:
  237 + conn.execute(
  238 + "UPDATE execution_log_file SET status = 'closed' WHERE log_file_id = ?", (log_file_id,)
  239 + )
  240 + conn.commit()
  241 +
  242 +
  243 +class _UsageLimitSignal(Exception):
  244 + """Internal: run_fn raises this (via raise_if_usage_limited helper) to
  245 + signal a quota hit without run_work_node having to know how run_fn talks
  246 + to `claude`."""
  247 +
  248 + def __init__(self, quota_limit_type):
  249 + super().__init__(quota_limit_type)
  250 + self.quota_limit_type = quota_limit_type
  251 +
  252 +
  253 +def raise_if_usage_limited(text):
  254 + """Call from inside a `run_fn` after a failed claude invocation: raises
  255 + _UsageLimitSignal if `text` matches 03-C §3.6's exact detection regex,
  256 + otherwise does nothing (let the caller re-raise its own error)."""
  257 + quota_type = _quota_limit_type(text)
  258 + if quota_type:
  259 + raise _UsageLimitSignal(quota_type)
... ...