U-021: Graph Engine 持久化 + 断点恢复（REQ-06）

now 默认值从 time.monotonic 改成 time.time（墙钟），使 quota 等待的已耗时计算能在进程重启后正确续算；run_work_node 对已 done 的节点直接返回存量结果不重跑，对 retrying/waiting_for_quota 节点从持久化的 retry_count/quota_wait_started_at 续算而非清零；needs_worktree=True 时复用已存在的 worktree 而不是重新 create。新增 resume_work_graph_run() 返回一次 run 里尚未完成的 node_exec_id 列表，并对内部不一致的持久化状态统一抛 CorruptPersistedStateError。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-021: Graph Engine 持久化 + 断点恢复（REQ-06）
now 默认值从 time.monotonic 改成 time.time（墙钟），使 quota 等待的已耗时计算能在进程重启后正确续算；run_work_node 对已 done 的节点直接返回存量结果不重跑，对 retrying/waiting_for_quota 节点从持久化的 retry_count/quota_wait_started_at 续算而非清零；needs_worktree=True 时复用已存在的 worktree 而不是重新 create。新增 resume_work_graph_run() 返回一次 run 里尚未完成的 node_exec_id 列表，并对内部不一致的持久化状态统一抛 CorruptPersistedStateError。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent ee6787ba
Showing 4 changed files with 285 additions and 12 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/test_work_node_run.py
src/work_node_run.py
@@ -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
  
 ---
@@ -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"},
@@ -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()
@@ -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&quot;hit your (session|weekly|opus) limit&quot;, 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