From 15385d6054f9954c33fccb40c80efa4a9b0e4a82 Mon Sep 17 00:00:00 2001 From: bboonnee Date: Fri, 31 Jul 2026 16:49:09 +0800 Subject: [PATCH] U-021: Graph Engine 持久化 + 断点恢复（REQ-06） --- docs/development-plan.md | 1 + docs/sdd/unit_matrix.json | 2 +- src/test_work_node_run.py | 153 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/work_node_run.py | 141 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------- 4 files changed, 285 insertions(+), 12 deletions(-) diff --git a/docs/development-plan.md b/docs/development-plan.md index 4bd85cb..88fd616 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -201,6 +201,7 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。 - **U-021 [Risk Gate]** Graph Engine 持久化状态 + 断点恢复 - 依赖：U-012, U-013 - Done：REQ-06 happy/boundary/failure 通过（执行中途中断进程后从持久化状态恢复，不重跑已完成节点） + - **实现补充（2026-07-31）**：核心问题是 U-013 `run_work_node` 的 `now` 默认值是 `time.monotonic`——单调时钟在进程重启后没有固定的墙钟含义，REQ-06 恰恰要求"进程重启后继续计算 quota 等待已经过去多久"，两者根本不兼容。改法：`now` 默认改回 `time.time`（墙钟），新增 `_parse_sqlite_timestamp()` 把 SQLite `CURRENT_TIMESTAMP` 的 UTC 字符串转成可比的 epoch 值；`run_work_node` 一开始就读出该 node_exec_id 已持久化的 `status`/`retry_count`/`quota_wait_started_at`/`git_worktree_ref`，`status='done'` 直接返回存量 `output_payload`（不重跑 `run_fn`），`retrying`/`waiting_for_quota` 从持久化的 `retry_count`/`quota_wait_started_at` 续算而不是清零重来，`needs_worktree=True` 时若已存在对应 worktree 目录则复用而不是重新 `git worktree add`（会直接失败）。新增 `resume_work_graph_run(conn, run_id)`：返回一次 `work_graph_run` 里还没 `done` 的 `node_exec_id` 列表，供调用方决定重新调度谁；对内部不一致的持久化状态（无效的 `work_graph_run.status`/节点 `status`，`status='done'` 但 `output_payload` 为空，`status='waiting_for_quota'` 但 `quota_wait_started_at` 为空）统一抛 `CorruptPersistedStateError`，不会返回一份看似正常但实际有问题的恢复计划。这几类"状态不一致"用正常写入路径其实到不了（各字段都有 `CHECK` 约束），测试里用 `PRAGMA ignore_check_constraints=ON` 模拟外部损坏的行来触发。 - 覆盖：01-M `work_graph_run`/`work_node_execution`；04-P `p_work_graph_execution`；REQ-06 --- diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 92e6569..1711d26 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -34,7 +34,7 @@ {"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 完整验收（见 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-021", "phase": "PH2", "sub_phase": "PH2-S4", "name": "Graph Engine 持久化 + 断点恢复（见 development-plan.md 2026-07-31 实现补充）", "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 通过；now 默认值从 time.monotonic 改成 time.time（墙钟）以支持跨进程重启的 quota 等待续算；新增 resume_work_graph_run() 返回未完成节点列表"}, {"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 全部通过"}, {"id": "U-023", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_goal_composer + v_bug_feedback", "classification": "Value Path", "depends_on": ["U-022"], "coverage": {"model": [], "view": ["v_goal_composer", "v_bug_feedback"], "controller": ["c_bug_report_create"], "process": [], "requirement": []}, "done": "能通过 UI 创建 task 和人工 bug_record"}, diff --git a/src/test_work_node_run.py b/src/test_work_node_run.py index 7334192..ca32e0c 100644 --- a/src/test_work_node_run.py +++ b/src/test_work_node_run.py @@ -337,5 +337,158 @@ class RealGitWorktreeIsolationTest(WorkNodeRunTestBase): self.assertIn("module_b.txt", log.stdout) +class BreakpointRecoveryTest(WorkNodeRunTestBase): + """U-021 / REQ-06: run_work_node called again for a node_exec_id that was + already mid-flight when a process restart happened.""" + + def test_done_node_returns_stored_output_without_rerunning(self): + self.conn.execute( + "UPDATE work_node_execution SET status = 'done', output_payload = ? WHERE node_exec_id = 'node-1'", + (json.dumps({"value": "already done"}),), + ) + self.conn.commit() + calls = [] + output = work_node_run.run_work_node( + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, + run_fn=lambda cwd, model=None: calls.append(1) or {"value": "should not run"}, + ) + self.assertEqual(output, {"value": "already done"}) + self.assertEqual(calls, []) + + def test_resuming_retrying_node_continues_persisted_retry_count(self): + # node-1 was already at retry_count=2 (not 0) before the restart -- + # one more failure must immediately exceed max_retries=2, proving + # the count resumed rather than restarting from zero. + self.conn.execute( + "UPDATE work_node_execution SET status = 'retrying', retry_count = 2 WHERE node_exec_id = 'node-1'" + ) + self.conn.commit() + calls = [] + + def run_fn(cwd, model=None): + calls.append(1) + return {"wrong_key": True} + + 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=run_fn, max_retries=2, + ) + self.assertEqual(len(calls), 1) + row = self.conn.execute( + "SELECT status, retry_count FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone() + self.assertEqual(row, ("blocked", 3)) + + def test_resuming_waiting_for_quota_node_uses_persisted_timestamp_not_a_fresh_clock(self): + # if elapsed-time tracking restarted at 0 on resume instead of using + # the persisted quota_wait_started_at, this would never time out. + original_ts = "2026-07-01 00:00:00" + epoch_orig = work_node_run._parse_sqlite_timestamp(original_ts) + self.conn.execute( + "UPDATE work_node_execution SET status = 'waiting_for_quota', " + "quota_wait_started_at = ?, quota_limit_type = 'session' WHERE node_exec_id = 'node-1'", + (original_ts,), + ) + self.conn.commit() + + 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): + work_node_run.run_work_node( + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn, + sleep=lambda s: None, now=lambda: epoch_orig + 8 * 3600 + 1, + ) + status = self.conn.execute( + "SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone()[0] + self.assertEqual(status, "blocked") + + def test_nonexistent_node_exec_id_raises_corrupt_state(self): + with self.assertRaises(work_node_run.CorruptPersistedStateError): + work_node_run.run_work_node( + self.conn, "no-such-node", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, + run_fn=lambda cwd, model=None: {"value": 1}, + ) + + def test_done_node_with_null_output_payload_raises_corrupt_state(self): + self.conn.execute( + "UPDATE work_node_execution SET status = 'done', output_payload = NULL WHERE node_exec_id = 'node-1'" + ) + self.conn.commit() + with self.assertRaises(work_node_run.CorruptPersistedStateError): + work_node_run.run_work_node( + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, + run_fn=lambda cwd, model=None: {"value": 1}, + ) + + def test_waiting_for_quota_node_with_null_timestamp_raises_corrupt_state(self): + self.conn.execute( + "UPDATE work_node_execution SET status = 'waiting_for_quota', quota_wait_started_at = NULL " + "WHERE node_exec_id = 'node-1'" + ) + self.conn.commit() + with self.assertRaises(work_node_run.CorruptPersistedStateError): + work_node_run.run_work_node( + self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, + run_fn=lambda cwd, model=None: {"value": 1}, + ) + + +class ResumeWorkGraphRunTest(WorkNodeRunTestBase): + """U-021 / REQ-06: resume_work_graph_run's own resume-plan + corrupt-state + detection, independent of run_work_node.""" + + def test_returns_only_non_done_node_exec_ids(self): + 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, output_payload, status) " + "VALUES ('node-2', 'run-1', 'n2', 'coder', 'skill-1', 'in.json', 'out.json', '{}', ?, 'done')", + (json.dumps({"value": 1}),), + ) + 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-3', 'run-1', 'n3', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'retrying')" + ) + self.conn.commit() + resumable = work_node_run.resume_work_graph_run(self.conn, "run-1") + self.assertEqual(sorted(resumable), ["node-1", "node-3"]) + + def test_nonexistent_run_raises_corrupt_state(self): + with self.assertRaises(work_node_run.CorruptPersistedStateError): + work_node_run.resume_work_graph_run(self.conn, "no-such-run") + + def test_invalid_run_status_raises_corrupt_state(self): + # unreachable via a normal write (work_graph_run.status has its own + # CHECK constraint) -- ignore_check_constraints simulates the kind of + # externally-corrupted row REQ-06 must still detect. + self.conn.execute("PRAGMA ignore_check_constraints = ON") + self.conn.execute("UPDATE work_graph_run SET status = 'bogus_status' WHERE run_id = 'run-1'") + self.conn.commit() + with self.assertRaises(work_node_run.CorruptPersistedStateError): + work_node_run.resume_work_graph_run(self.conn, "run-1") + + def test_done_node_missing_output_payload_raises_corrupt_state(self): + 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', '{}', 'done')" + ) + self.conn.commit() + with self.assertRaises(work_node_run.CorruptPersistedStateError): + work_node_run.resume_work_graph_run(self.conn, "run-1") + + def test_waiting_for_quota_node_missing_timestamp_raises_corrupt_state(self): + 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', '{}', 'waiting_for_quota')" + ) + self.conn.commit() + with self.assertRaises(work_node_run.CorruptPersistedStateError): + work_node_run.resume_work_graph_run(self.conn, "run-1") + + if __name__ == "__main__": unittest.main() diff --git a/src/work_node_run.py b/src/work_node_run.py index 8e65610..bc0ecd4 100644 --- a/src/work_node_run.py +++ b/src/work_node_run.py @@ -34,6 +34,7 @@ import re import subprocess import time import uuid +from datetime import datetime, timezone from src import paths @@ -41,6 +42,12 @@ _USAGE_LIMIT_RE = re.compile(r"hit your (session|weekly|opus) limit", re.IGNOREC _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"] +_VALID_WORK_GRAPH_RUN_STATUSES = { + "planning", "running", "merging", "verifying", "reflecting", "blocked", "done", "failed", +} +_VALID_NODE_STATUSES = { + "pending", "running", "reflecting", "retrying", "waiting_for_quota", "blocked", "done", "failed", +} class SchemaValidationFailedError(Exception): @@ -63,6 +70,25 @@ class UsageLimitWaitTimeoutError(Exception): c_work_node_run itself not knowing about `task` at all (03-C §3.6).""" +class CorruptPersistedStateError(Exception): + """REQ-06 failure case: persisted work_graph_run/work_node_execution state + is internally inconsistent after a restart -- must error clearly, not + silently continue on bogus state. No 03-C error code is named for this + (it's not a normal state-machine transition), but REQ-06 requires exactly + this behavior; callers should route it to task.status='blocked_pending_human' + same as c_work_node_run itself doesn't touch `task` directly (03-C §3.6).""" + + +def _parse_sqlite_timestamp(text): + """SQLite's CURRENT_TIMESTAMP is UTC 'YYYY-MM-DD HH:MM:SS' with no + timezone marker -- parsed as UTC explicitly so it's comparable against + time.time() (also UTC epoch seconds) across a process restart. A plain + monotonic clock can't do this: it has no fixed relationship to wall-clock + time and resets meaning across restarts, which is exactly REQ-06's + scenario -- so `now`'s contract here is wall-clock, not monotonic.""" + return datetime.strptime(text, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp() + + def _validate_schema_lite(payload, schema): """object/required only -- everything this codebase's schemas use so far.""" if schema.get("type") == "object": @@ -123,7 +149,7 @@ def run_work_node( needs_worktree=False, repo_path=None, sleep=time.sleep, - now=time.monotonic, + now=time.time, opus_downgrade_model="claude-sonnet-5", ): """Runs one Work Node to contract. `run_fn(cwd, model) -> output_payload` @@ -132,6 +158,14 @@ def run_work_node( opus-quota-triggered retry, in which case it's `opus_downgrade_model`. Returns the validated output_payload. + REQ-06 (breakpoint recovery): safe to call again after a process restart + for a node_exec_id that was already mid-flight. A node already 'done' + is never re-run (its stored output_payload is returned as-is); a node + interrupted mid-'retrying' or mid-'waiting_for_quota' resumes from its + persisted retry_count / quota_wait_started_at rather than restarting + those counters at zero -- `now` must be wall-clock (default time.time) + for the quota-wait resume math to mean anything across a restart. + 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 @@ -140,14 +174,38 @@ def run_work_node( 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 " + existing = conn.execute( + "SELECT p.project_id, wgr.task_id, wne.status, wne.retry_count, wne.output_payload, " + "wne.quota_wait_started_at, wne.quota_limit_type " + "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() + if existing is None: + raise CorruptPersistedStateError("no such work_node_execution: " + node_exec_id) + ( + project_id, task_id, existing_status, existing_retry_count, existing_output_payload, + existing_quota_wait_started_at, existing_quota_limit_type, + ) = existing + + if existing_status == "done": + # REQ-06 happy: never re-run an already-completed node. + if existing_output_payload is None: + raise CorruptPersistedStateError( + "work_node_execution {} is status=done but has no output_payload".format(node_exec_id) + ) + return json.loads(existing_output_payload) + if existing_status not in _VALID_NODE_STATUSES: + raise CorruptPersistedStateError( + "work_node_execution {} has invalid status {!r}".format(node_exec_id, existing_status) + ) + if existing_status == "waiting_for_quota" and not existing_quota_wait_started_at: + raise CorruptPersistedStateError( + "work_node_execution {} is status=waiting_for_quota but has no quota_wait_started_at".format(node_exec_id) + ) log_file_id = str(uuid.uuid4()) log_path = paths.log_file_path(project_id, task_id, node_exec_id, node_exec_id) @@ -162,16 +220,29 @@ def run_work_node( 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() + existing_worktree_ref = conn.execute( + "SELECT git_worktree_ref FROM work_node_execution WHERE node_exec_id = ?", (node_exec_id,) + ).fetchone()[0] + # REQ-06: resuming a node that already had a worktree from before + # the restart must reuse it, not `git worktree add` the same + # branch/path again (which would just fail). + if existing_worktree_ref and os.path.isdir(existing_worktree_ref): + worktree_path = existing_worktree_ref + else: + 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 + retry_count = existing_retry_count or 0 + quota_wait_start = ( + _parse_sqlite_timestamp(existing_quota_wait_started_at) + if existing_status == "waiting_for_quota" + else None + ) current_model = None opus_already_downgraded = False while True: @@ -274,3 +345,51 @@ def raise_if_usage_limited(text): quota_type = _quota_limit_type(text) if quota_type: raise _UsageLimitSignal(quota_type) + + +def resume_work_graph_run(conn, run_id): + """REQ-06 (04-P §4.2 p_work_graph_execution, breakpoint recovery): + after a process restart, determine which nodes in `run_id` still need + work. Nodes already 'done' are excluded (run_work_node itself also + refuses to re-run them; this is what a caller uses to decide what to + re-dispatch at all). Raises CorruptPersistedStateError instead of + returning a bogus resume plan if the persisted state doesn't hold + together -- an invalid work_graph_run.status, an invalid node status, a + 'done' node with no output, or a 'waiting_for_quota' node with no + quota_wait_started_at (the same per-node checks run_work_node applies, + checked here for every node up front rather than one at a time). + """ + run_row = conn.execute("SELECT status FROM work_graph_run WHERE run_id = ?", (run_id,)).fetchone() + if run_row is None: + raise CorruptPersistedStateError("no such work_graph_run: " + run_id) + if run_row[0] not in _VALID_WORK_GRAPH_RUN_STATUSES: + raise CorruptPersistedStateError( + "work_graph_run {} has invalid status {!r}".format(run_id, run_row[0]) + ) + + nodes = conn.execute( + "SELECT node_exec_id, status, output_payload, quota_wait_started_at FROM work_node_execution " + "WHERE run_id = ?", + (run_id,), + ).fetchall() + + resumable = [] + for node_exec_id, status, output_payload, quota_wait_started_at in nodes: + if status not in _VALID_NODE_STATUSES: + raise CorruptPersistedStateError( + "work_node_execution {} has invalid status {!r}".format(node_exec_id, status) + ) + if status == "done": + if output_payload is None: + raise CorruptPersistedStateError( + "work_node_execution {} is status=done but has no output_payload".format(node_exec_id) + ) + continue + if status == "waiting_for_quota" and not quota_wait_started_at: + raise CorruptPersistedStateError( + "work_node_execution {} is status=waiting_for_quota but has no quota_wait_started_at".format( + node_exec_id + ) + ) + resumable.append(node_exec_id) + return resumable -- libgit2 0.21.4