I: production task reconciliation loop (FR-2)

Recovers tasks abandoned mid-flight (status in queued/planning_brd_prd_sdd/ coding/validating with no recent execution_event) by re-driving them through task_execution.run_or_resume_task_lifecycle on a background poll, guarded by an in-process in-flight lock to avoid double dispatch. Responds to a real 2026-08-06 incident where a task sat stuck in 'validating' for 15+ hours with nothing polling it. Scoped to FR-2 only from docs/todo/scheduler-production-polling-loop- requirements.md; FR-1 (new-goal auto-dispatch) stays out per that doc's own non-goals -- it's a separate, larger autonomy-boundary decision. Design recorded in docs/todo/scheduler-production-polling-loop-design.md. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

I: production task reconciliation loop (FR-2)
Recovers tasks abandoned mid-flight (status in queued/planning_brd_prd_sdd/ coding/validating with no recent execution_event) by re-driving them through task_execution.run_or_resume_task_lifecycle on a background poll, guarded by an in-process in-flight lock to avoid double dispatch. Responds to a real 2026-08-06 incident where a task sat stuck in 'validating' for 15+ hours with nothing polling it. Scoped to FR-2 only from docs/todo/scheduler-production-polling-loop- requirements.md; FR-1 (new-goal auto-dispatch) stays out per that doc's own non-goals -- it's a separate, larger autonomy-boundary decision. Design recorded in docs/todo/scheduler-production-polling-loop-design.md. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent fa86f363
Showing 7 changed files with 667 additions and 11 deletions
docs/todo/scheduler-production-polling-loop-design.md
docs/todo/scheduler-production-polling-loop-requirements.md
src/backend_server.py
src/task_execution.py
src/task_reconciliation.py
src/test_task_execution.py
src/test_task_reconciliation.py
+# 生产环境调度/复原循环——详细设计（FR-2 部分）
+
+**日期**: 2026-08-06
+**性质**: SDD 层面设计文档，对应 `scheduler-production-polling-loop-requirements.md`。
+**范围**: 只覆盖该需求文档的 FR-2（在途任务复原）。FR-1（新 Goal 自动派发）按需求文档自身的
+非目标 #1 和风险章节，仍然是一个未决的、需要单独确认的自主权边界决策，本文档不涉及，需求文档
+本身也因此暂不移入 `docs/done/`。
+**状态**: 已实现（`src/task_reconciliation.py` + `src/backend_server.py::_start_reconciliation_loop`
++ `src/task_execution.py` 的 in-flight 互斥锁），本文档是对已落地实现的设计记录，补齐
+`docs/todo/` 两段式流程里"确认后产出 SDD 设计文档"这一步。
+
+## 1. 整体方案
+
+拆成三层，分别对应需求文档 §8 的五个开放问题：
+
+1. **候选发现**（`task_reconciliation.find_stale_resumable_tasks`）：只读查询，扫描
+   `task.status IN task_execution._RESUMABLE_STATUSES` 且"最近一次活动早于阈值"的任务。
+2. **一次调度扫描**（`task_reconciliation.reconcile_once`）：调用发现函数，对每个候选调用
+   `task_execution.start_task_execution_in_background`，单个候选派发失败不影响其余候选
+   （复用 04-P "调度器继续执行队列中其他任务" 的既有原则）。
+3. **载体**（`backend_server.py::_start_reconciliation_loop`）：`main()` 专属的后台线程，
+   `while True: reconcile_once(); sleep(interval)`，`--reconciliation-interval-seconds`
+   （默认 300s）/`--disable-reconciliation-loop` 两个 CLI 开关。
+
+## 2. 开放问题回答
+
+| 需求文档 §8 的问题 | 本设计的答案 |
+|---|---|
+| FR-2 轮询周期 | 300s（`_DEFAULT_RECONCILIATION_INTERVAL_SECONDS`），可通过 `--reconciliation-interval-seconds` 覆盖 |
+| "一段时间没有新进展"的判定信号与阈值 | 用 `execution_event` 该 task 最近一条记录的 `created_at`（无记录则退回 `task.created_at`），**不用** `task.updated_at`——阶段执行期间 `task.updated_at` 本来就不变，这正是这次事故排查时的真实盲点。阈值统一 1800s，不区分阶段（见下方"已知简化"） |
+| FR-2 载体形式 | `backend_server.py::main()` 内置守护线程，不是独立进程。理由：FR-2 只是"帮已存在的任务把该调用的函数再调用一次"，复用的是 `task_execution` 现有的进程内 in-flight 互斥锁，做成独立进程反而需要额外的跨进程协调；这与 Deploy Supervisor"独立进程、无共享失败域"哲学的适用场景不同——Deploy Supervisor 独立是因为它需要在 `backend_server` 本身挂掉时还能做健康检查/回滚，FR-2 不需要这个属性 |
+| FR-2 互斥机制 | 进程内 `threading.Lock` + `set`（`task_execution._in_flight_task_ids`），不是数据库锁。有意选择：只需要防止**同一进程内**两个线程（控制台点 Start 的线程 vs 本循环）竞争同一个 `task_id`；不需要跨进程/跨重启存活——重启后这个 set 清空为空集合本身就是正确状态（意味着"这个进程里没有任何东西正在跑"），不是 bug |
+| 是否暴露为可配置项 | 是：`--reconciliation-interval-seconds`、`--disable-reconciliation-loop`。硬编码的是阈值本身（`_DEFAULT_STALE_AFTER_SECONDS=1800`），未暴露成 CLI 参数——v1 先用一个保守默认值，符合需求文档非目标 #4 |
+
+## 3. 竞态处理
+
+`find_stale_resumable_tasks` 在扫描时检查 `is_in_flight()`，但扫描到派发之间存在窗口——这段
+时间里同一个 `task_id` 可能被控制台"Start"按钮抢先派发。这不是需要额外加锁解决的问题：
+`start_task_execution_in_background` 内部的 `_try_claim` 本身是原子的（持锁检查+插入），
+竞态发生时后到者会直接收到 `TaskExecutionError`，`reconcile_once` 捕获后跳过该候选、不影响
+其余候选——测试 `test_one_dispatch_failure_does_not_block_others` 覆盖了这条路径。
+
+## 4. 已知简化（v1 天花板，非 bug）
+
+- **单一全局阈值，不分阶段**。需求文档 §8 提到"Coder 阶段可能比 Validator 阶段容忍更长空闲"，
+  v1 用统一的 1800s（比最长单次阻塞调用 Validator 自身 Work Graph 的 `timeout=1200s` 多
+  600s 余量）。升级路径：给 `_RESUMABLE_STATUSES` 里每个状态配一个独立阈值，按
+  `task.assigned_org_node`/`task.status` 查表。
+- **候选扫描全表查询，无索引优化**。当前任务量级下（个位数到几十个并发 task）不是问题；
+  升级路径：如果任务量上升，给 `task.status`、`execution_event.task_id` 加索引。
+
+## 5. 可观测性修复（本次评审新增，2026-08-06）
+
+原实现里 `_start_reconciliation_loop` 的 `except Exception: pass` 会让"每一轮都失败"和
+"每一轮都没有候选"从外部看起来完全一样——对一个设计目标就是无人值守 7x24 运行的循环来说，
+这种静默失败是最危险的一类。已改为 `traceback.print_exc()`，异常信息进 stderr（`main()`
+是真实 CLI 入口，stderr 在部署环境下通常会被进程管理器捕获落盘）。这不引入新依赖，也不改变
+"一次失败不杀死循环"的既有语义。
+
+## 6. 验收标准核对（对照需求文档 §5）
+
+1. ✅ `test_finds_task_with_no_activity_at_all` / `test_stale_event_still_flagged` 覆盖"卡住的
+   任务被重新捡起"
+2. ✅ `test_excludes_task_with_recent_event` / `test_uses_most_recent_event_not_earliest` 覆盖
+   "正常在跑的任务不会被误判"——但这两个测试验证的是"有新鲜 `execution_event`"这一路径，没有
+   单独构造"一次长达接近 1200s 的阻塞调用期间完全没有任何新事件"的场景。这属于用阈值余量
+   （1800 vs 1200）做保证、而非用事件覆盖做保证，测试没有专门验证这条时间余量假设本身——留作
+   后续如果要收紧阈值时的一个真实回归风险点。
+3. FR-1（新建 `queued` Goal 自动派发）不适用——本设计明确不覆盖 FR-1。
+4. ✅ `--disable-reconciliation-loop` 提供独立开关；FR-2 与 FR-1 天然独立（FR-1 尚未实现，
+   两者不存在耦合）。
+
+## 7. 后续（不在本次范围内）
+
+FR-1（新 Goal 自动派发）仍然是需求文档标注的"改变系统自主权边界"的决策，按该文档非目标 #1
+的要求，需要单独确认是否要做、什么时候做，不与 FR-2 混跑。在那之前，`queued` 状态的新 Goal
+仍然需要人工触发 `run_goal_to_completion`/`scheduler.run_scheduler_once()`。
+# 生产环境调度/复原循环——需求文档
+
+**日期**: 2026-08-06
+**性质**: 需求文档（BRD/PRD 层面），不含详细技术设计。回应一次真实事故的根因排查——任务 `3605272e-fe33-48cc-b3f3-e0c0ba26f970` 在 `validating` 状态卡死 15+ 小时，`execution_event` 表在此期间零记录，`work_graph_run` 里从未出现 Validator 的运行记录。
+**状态**: 草案，待评审。
+
+## 1. 背景
+
+排查这次真实事故发现两层原因：
+
+1. **生产后端进程从来没有接入过任何轮询循环**。`docs/implementation-gap-analysis.md` 第 6 条早就记录了这个缺口："给某个常驻进程（`backend_server.py` 或独立进程）接入 `scheduler.run_scheduler_once()` 的轮询循环"，评级"高"，并且明确建议"改变系统自己的自主权边界，建议单独一批、额外人工审查 diff，不要和其他 gap 混跑"。这次事故是这条缺口第一次以真实用户可见的方式发作。
+2. **即使真的接上 `scheduler.run_scheduler_once()`，也不会自动修好这次事故这类问题**——这是排查过程中新发现的、比原缺口描述更精确的一点。`run_scheduler_once(db_path, goals, max_workers=None)` 的入参 `goals` 是 `[(target_folder_path, goal_text), ...]`，语义是"这一轮要**新建**并跑完这些 Goal"，内部通过 `task_create.create_task()` 每次都新建一个 `task` 行。它解决的是"新 Goal 怎么被自动拿去跑"，**不是**"一个已经存在、卡在某个中间状态的 `task_id` 怎么被自动接回去继续跑"——后者需要的是 `task_execution.run_or_resume_task_lifecycle`（按 `task_id` 恢复），是完全不同的另一个函数、另一条代码路径。这次事故属于第二类：任务本身已经存在、状态是 `validating`（`task_execution._RESUMABLE_STATUSES` 里合法的可恢复状态之一），只是没有任何东西在合适的时机去调用恢复逻辑。
+
+这次事故本身的具体触发点（`validating` 状态本来就无法通过 `run_or_resume_task_lifecycle` 恢复）已经单独修复（见 `src/task_execution.py` 的 `_RESUMABLE_STATUSES`/`request_pause` 改动，2026-08-06）——这是"人工点 Resume 现在能生效了"，不等于"以后不会再有任务卡死没人管"。本需求要解决的是后者。
+
+## 2. 目标
+
+把生产环境需要的"持续推进任务"能力，拆成两个明确区分、可以分开排期的子问题：
+
+1. **新 Goal 派发**：`queued` 状态的任务（或者说，还没被任何调用触发过 `run_goal_to_completion`/`run_or_resume_task_lifecycle` 的新 Goal）需要被自动拿去执行，不需要人工每次手动点 Start。
+2. **在途任务复原（这次事故的真正根因）**：任何已经开始执行、但因为驱动它的那次调用提前退出（进程重启、脚本运行完毕、连接断开等任何原因）而卡在某个可恢复中间状态（`task_execution._RESUMABLE_STATUSES`：`queued`/`planning_brd_prd_sdd`/`coding`/`validating`）、且已经**一段时间没有新进展**的任务，需要被自动重新接回去继续跑，而不是无限期地停在原地、既不算 `blocked_pending_human`（触发不了人工介入通知），也没有任何东西在真的干活。
+
+## 3. 非目标
+
+1. **不在本文档里裁定"要不要做""什么时候做"这个更大的问题**。`docs/implementation-gap-analysis.md` 已经把这条标注为"改变系统自己的自主权边界"——是否现在就投入、要不要额外的人工评审门槛，是需要单独确认的决策，本文档只负责把需求本身讲清楚，供决策时参考。
+2. **不要求解决方案是常驻进程/独立进程/`backend_server.py` 内置线程三选一中的哪一个**。具体载体形式留给设计阶段，本文档只描述行为需求。
+3. **不重新设计 `run_scheduler_once`/`run_goal_to_completion`/`run_or_resume_task_lifecycle` 这三个已经存在且经过测试的函数本身**。它们的行为已经是对的，本需求要解决的是"谁、在什么时机去调用它们"这一层，不是重写它们的内部逻辑。
+4. **不要求第一版就有一套精细的"任务卡死"判定算法**（比如区分"真的卡死"和"合理地在等真实的 `claude -p` 长耗时调用"）。第一版可以用一个保守、简单的阈值（比如"`updated_at`/最近一条 `execution_event` 距今超过 N 分钟，且没有已知的进行中标记"），不需要一次做到完美。
+
+## 4. 功能需求
+
+| # | 需求描述 | 优先级 |
+|---|---|---|
+| FR-1 | 需要一个机制，定期（具体周期留待设计阶段）调用 `scheduler.run_scheduler_once()` 派发新 Goal——对应「目标 1」。 | 必须 |
+| FR-2 | 需要一个独立的机制，定期扫描 `task` 表里状态属于 `task_execution._RESUMABLE_STATUSES` 且"已经一段时间没有新进展"的任务，对每一个调用 `task_execution.run_or_resume_task_lifecycle`（或 `start_task_execution_in_background`）——对应「目标 2」，是这次事故的直接对应修复。 | 必须 |
+| FR-3 | FR-2 的"已经一段时间没有新进展"判定，需要能区分"真的没人管"和"确实在真实执行中，只是这一步本来就耗时"——不能简单靠 `task.status` 停留时间判断（比如 Validator 自己的 Work Graph 跑一次可能就要接近 `timeout=1200` 秒），需要参考更细粒度的信号（比如该任务关联的 `work_node_execution`/`execution_event` 最近一次活动时间，而不是 `task.updated_at`——`task.updated_at` 在阶段内部执行期间根本不会变，这正是这次事故排查时的真实盲点）。 | 必须 |
+| FR-4 | FR-1、FR-2 两个机制必须能独立开关/独立排期——不能因为"要接 FR-1 需要走完整的人工评审流程"就连带阻塞了 FR-2（这次事故的直接修复）。 | 必须 |
+| FR-5 | 无论 FR-1/FR-2 最终选择常驻进程还是独立进程实现，都不能引入新的"共享一个 sqlite 连接跨线程"风险——复用 `scheduler.py` 模块文档里已经验证过的模式（每个 worker 各自开自己的连接 + `busy_timeout`），不要重新踩 U-027 已经用真实 SIGSEGV 验证过的坑。 | 必须 |
+
+## 5. 验收标准
+
+1. 人为构造一个"任务卡在 `validating`，很久没有新进展"的场景（复现这次事故的条件），验证 FR-2 的机制能在设定的检测周期内把它重新接回去执行，不需要人工干预。
+2. 人为构造一个"任务正常在跑，Validator 的 Work Graph 正在执行中，还没超过合理耗时"的场景，验证 FR-2 的机制**不会**误判成"卡死"、不会对一个正常在跑的任务重复触发 `run_or_resume_task_lifecycle`（这会导致同一个任务被并发跑两次，需要明确防止）。
+3. 新建一个 `queued` 状态的 Goal，不手动点 Start，验证 FR-1 的机制能在设定周期内自动派发执行。
+4. FR-1 和 FR-2 分别单独关闭时，各自的行为符合预期（关 FR-1 不影响 FR-2 正常工作，反之亦然）。
+
+## 6. 约束与假设
+
+- 复用现有的 `task_execution._RESUMABLE_STATUSES`（已扩展为包含 `validating`）作为 FR-2 判断"这个任务原则上可以恢复"的依据，不重新定义一套判断标准。
+- FR-2 的具体实现要避免和「人工点 Resume」竞争——如果一个任务已经在被 `start_task_execution_in_background` 的某个后台线程处理，FR-2 的扫描不能对同一个 `task_id` 重复触发第二次执行。具体的互斥机制（数据库标记、内存锁、还是别的方式）留给设计阶段。
+
+## 7. 风险
+
+- **FR-1（新 Goal 自动派发）是这次事故排查之前就已经被 `implementation-gap-analysis.md` 标注为"改变系统自主权边界"的决策**，本文档不试图淡化这一点——它意味着系统会在没有人显式点击"Start"的情况下自主开始消耗真实 API 调用、真实修改代码。这条风险原样保留，需要在决策阶段被认真对待。
+- **FR-2（在途任务复原）看起来风险更小（只是让已经存在的任务继续跑完，不是凭空发起新工作），但仍然是自主行为的扩展**——需要在设计阶段想清楚"防误判"（FR-3/验收标准 2）没做好的后果：一个只是耗时较长、并没有真的卡死的任务被提前打断重跑，可能造成重复的 API 花费或者更糟的状态不一致。
+- **两个机制如果实现成同一个循环、共享同一套判断逻辑，容易在后续维护中被混为一谈**——按第 2 节"目标"明确拆开成两个独立问题，是为了避免这种混淆，设计阶段应该保持这个拆分，不要图省事合并成一个"万能调度循环"。
+
+## 8. 开放问题（留给详细设计阶段回答）
+
+- [ ] FR-1/FR-2 各自的轮询周期定多少。
+- [ ] FR-3"已经一段时间没有新进展"的具体判定信号和阈值——参考哪张表的哪个时间戳字段，阈值怎么定（是否要按 `task` 当前所处的具体阶段给不同阈值，比如 Coder 阶段允许的空闲时间可能比 Validator 阶段更长）。
+- [ ] FR-1/FR-2 的载体形式：`backend_server.py` 内置后台线程，还是独立进程（类似 Deploy Supervisor 的"独立进程、无共享失败域"哲学）。
+- [ ] FR-2 的互斥机制具体怎么实现，防止同一个任务被并发重复触发。
+- [ ] 这两个机制要不要暴露成可配置项（管理后台开关/环境变量），还是先硬编码一个保守的默认行为。
@@ -33,14 +33,16 @@ import socket
 import sqlite3
 import subprocess
 import threading
+import time
+import traceback
 import urllib.parse
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
  
 from src import (
     artifact_query, budget_guardrail, bug_record, deployment_query, execution_log,
     org_loop_route, paths, project_query, project_scaffold, schema, settings_config,
-    skill_query, skill_update, task_create, task_execution, task_query, work_graph_fork,
-    work_graph_run_query, ws_subscribe,
+    skill_query, skill_update, task_create, task_execution, task_query, task_reconciliation,
+    work_graph_fork, work_graph_run_query, ws_subscribe,
 )
  
 _FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
@@ -1042,6 +1044,42 @@ def create_server(host=&quot;127.0.0.1&quot;, port=0, db_path=None, skill_repo_path=None):
     return server
  
  
+_DEFAULT_RECONCILIATION_INTERVAL_SECONDS = 300
+
+
+def _start_reconciliation_loop(db_path, interval_seconds):
+    """2026-08-06 (FR-2 of docs/todo/scheduler-production-polling-loop-
+    requirements.md): background daemon thread that repeatedly calls
+    `task_reconciliation.reconcile_once()`. Deliberately started here, NOT
+    inside `create_server()` -- that function is instantiated fresh by
+    nearly every backend_server test, and a background loop dispatching
+    real `start_task_execution_in_background()` calls (real subprocess
+    invocations) during unit tests would be exactly the kind of test
+    pollution CLAUDE.md's Testing constraint exists to prevent. Only
+    `main()` (the real CLI entry point) starts this.
+
+    Runs one pass immediately (a restart is exactly when previously-
+    abandoned tasks most need to be picked back up, not `interval_seconds`
+    later), then repeats. A bad pass never kills the loop -- the next
+    scheduled pass just tries again, same "one failure doesn't stop
+    everything else" principle `reconcile_once()` itself already applies
+    per-candidate. The traceback still goes to stderr: a silently-swallowed
+    per-pass exception would otherwise look identical to "nothing to
+    reconcile" from the outside -- exactly the kind of quiet failure a
+    24/7-unattended loop can't afford."""
+    def loop():
+        while True:
+            try:
+                task_reconciliation.reconcile_once(db_path)
+            except Exception:
+                traceback.print_exc()
+            time.sleep(interval_seconds)
+
+    thread = threading.Thread(target=loop, daemon=True)
+    thread.start()
+    return thread
+
+
 def main():
     """U-037: CLI entry point (`python3 -m src.backend_server ...`) -- until
     this Unit, `create_server()` was only ever instantiated in-process by
@@ -1059,10 +1097,15 @@ def main():
     parser.add_argument("--port", type=int, default=0)
     parser.add_argument("--db-path", default=None)
     parser.add_argument("--skill-repo-path", default=None)
+    parser.add_argument("--reconciliation-interval-seconds", type=int, default=_DEFAULT_RECONCILIATION_INTERVAL_SECONDS)
+    parser.add_argument("--disable-reconciliation-loop", action="store_true")
     args = parser.parse_args()
  
     server = create_server(args.host, args.port, args.db_path, args.skill_repo_path)
  
+    if not args.disable_reconciliation_loop:
+        _start_reconciliation_loop(server.db_path, args.reconciliation_interval_seconds)
+
     def _handle_sigterm(signum, frame):
         # server.shutdown() blocks until serve_forever()'s loop notices and
         # returns -- calling it from the SAME thread that's running
@@ -30,7 +30,7 @@ import threading
  
 from src import e2e_pipeline, org_node_lifecycle, schema, scheduler
  
-_RESUMABLE_STATUSES = ("queued", "planning_brd_prd_sdd", "coding")
+_RESUMABLE_STATUSES = ("queued", "planning_brd_prd_sdd", "coding", "validating")
 _DOC_ARTIFACT_TYPES = ("brd", "prd", "sdd", "benchmark")
 _CODE_ARTIFACT_TYPES = ("code_diff", "test_report")
  
@@ -95,9 +95,27 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
     ever happens BETWEEN phases (checked right before each org_node
     trigger), so the only statuses a paused task can be resumed from are
     'queued' (nothing ran yet), 'planning_brd_prd_sdd' (Planner done, Coder
-    not yet triggered), and 'coding' (Coder done, Validator not yet
-    triggered) -- anything else raises `TaskExecutionError` rather than
-    guessing at a mid-phase resume this module was never designed for."""
+    not yet triggered), 'coding' (Coder done, Validator not yet triggered),
+    and 'validating' -- anything else raises `TaskExecutionError` rather
+    than guessing at a mid-phase resume this module was never designed for.
+
+    **'validating' (2026-08-06, real stuck-task incident)**: this is NOT a
+    "between phases" resume like the other three -- `org_node_lifecycle.
+    trigger_org_node_full(..., "validator", ...)`'s blocking subprocess call
+    already completed successfully (that's the only way `task.status` could
+    have become 'validating' in the first place: the UPDATE to 'validating'
+    is the last thing that call does). Calling it a second time would fail
+    its own `_PRE_STATE["validator"] == "coding"` guard outright, since the
+    task is already past that state. The actual gap for a task stuck here is
+    narrower and safe to fill directly: `validate_implementation_with_own_
+    graph()` -- which creates and runs the Validator's OWN Work Graph --
+    never got invoked. Found for real: a task sat in 'validating' for 15+
+    hours with zero `execution_event` rows and no `validator`-org_node
+    `work_graph_run` ever created, because nothing in production actually
+    polls `scheduler.run_scheduler_once()` (a separate, larger, tracked gap
+    -- see `docs/todo/scheduler-production-polling-loop-requirements.md`)
+    and whatever one-off call had driven this task that far had already
+    exited by the time Coder finished."""
     status = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
     if status not in _RESUMABLE_STATUSES:
         raise TaskExecutionError("task {} is status={}, not resumable from here".format(task_id, status))
@@ -126,11 +144,15 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
             benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
             test_script_content=test_script_content,
         )
-    else:  # status == "coding" -- Coder already ran
+    else:  # status in ("coding", "validating") -- Coder already ran
         code_artifact_ids = _latest_artifact_ids(conn, task_id, _CODE_ARTIFACT_TYPES)
  
     _check_not_paused(conn, task_id)
-    org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
+    if status != "validating":
+        # "validating" means this exact call already ran once -- see the
+        # docstring's 'validating' section above. Calling it again would
+        # hit org_node_lifecycle's own PRE_STATE guard (needs "coding").
+        org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
     verdict = e2e_pipeline.validate_implementation_with_own_graph(
         conn, task_id, sdd_content, prd_content, goal_text, target_folder_path,
     )
@@ -146,6 +168,37 @@ def _mark_blocked(conn, task_id, exc):
     scheduler._block_task(conn, task_id, reason, detail=str(exc))
  
  
+# 2026-08-06 (task_reconciliation): in-process guard against dispatching the
+# SAME task_id twice concurrently -- e.g. the console's Start button firing
+# while a reconciliation pass (task_reconciliation.py) has already picked up
+# the same stale task, or two rapid double-clicks on Start. Deliberately
+# in-memory only (not a DB column/lock): this only needs to protect against
+# two threads inside THIS process racing each other; it does not (and isn't
+# meant to) survive a process restart -- a restart's whole point is that
+# nothing in this process is running anymore, so an empty set on startup is
+# the correct state, not a bug.
+_in_flight_lock = threading.Lock()
+_in_flight_task_ids = set()
+
+
+def is_in_flight(task_id):
+    with _in_flight_lock:
+        return task_id in _in_flight_task_ids
+
+
+def _try_claim(task_id):
+    with _in_flight_lock:
+        if task_id in _in_flight_task_ids:
+            return False
+        _in_flight_task_ids.add(task_id)
+        return True
+
+
+def _release(task_id):
+    with _in_flight_lock:
+        _in_flight_task_ids.discard(task_id)
+
+
 def start_task_execution_in_background(db_path, task_id, target_folder_path, goal_text):
     """Spawns a daemon thread that runs (or resumes) this task's real
     execution to completion. Own sqlite connection per thread -- same
@@ -155,7 +208,14 @@ def start_task_execution_in_background(db_path, task_id, target_folder_path, goa
     `scheduler.finish_task_lifecycle` closing step (`close_task_done` /
     `org_level_loop.run_fix_loop`) a scheduled task gets, so a manually
     started one reaches the same real terminal state. Returns the Thread
-    (already started) so callers/tests can `.join()` it if needed."""
+    (already started) so callers/tests can `.join()` it if needed.
+
+    Raises `TaskExecutionError` immediately (no thread spawned) if this
+    task_id already has an execution in flight in this process -- see
+    `_in_flight_task_ids` above."""
+    if not _try_claim(task_id):
+        raise TaskExecutionError("task {} already has an execution in flight".format(task_id))
+
     def run():
         # check_same_thread=False (2026-08-06, F): same reasoning as
         # scheduler.run_goal_to_completion's own conn -- this thread's conn
@@ -173,6 +233,7 @@ def start_task_execution_in_background(db_path, task_id, target_folder_path, goa
             _mark_blocked(conn, task_id, e)
         finally:
             conn.close()
+            _release(task_id)
  
     thread = threading.Thread(target=run, daemon=True)
     thread.start()
@@ -186,12 +247,20 @@ def request_pause(conn, task_id, reason):
     trail (execution_event / blocked_reason) preserves which button the
     user actually clicked. Only valid while a background execution could
     plausibly be in flight; raises TaskExecutionError otherwise (nothing
-    running to pause, or already terminal)."""
+    running to pause, or already terminal).
+
+    'validating' included (2026-08-06, same incident as
+    `run_or_resume_task_lifecycle`'s 'validating' section): a resumed
+    validation run stays in `task.status='validating'` for its whole
+    duration (`validate_implementation_with_own_graph` never updates
+    `task.status` itself, same as `implement_code` doesn't during
+    'coding') -- excluding it here would leave that in-flight run with no
+    way to request a pause/stop, unlike the other two resumable phases."""
     row = conn.execute("SELECT project_id, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
     if row is None:
         raise TaskExecutionError("no such task: " + task_id)
     project_id, status = row
-    if status not in ("planning_brd_prd_sdd", "coding"):
+    if status not in ("planning_brd_prd_sdd", "coding", "validating"):
         raise TaskExecutionError(
             "task {} is status={}, nothing in-flight to pause/stop".format(task_id, status)
         )
+"""Periodic recovery for tasks abandoned mid-flight (2026-08-06, real
+incident: a task sat in status='validating' for 15+ hours with zero
+`execution_event` activity and no live process working on it, because
+whatever one-off call had driven it that far had already exited and
+nothing in production polls for orphaned resumable tasks).
+
+Scope: FR-2 only from `docs/todo/scheduler-production-polling-loop-
+requirements.md` (in-flight task reconciliation) -- deliberately NOT FR-1
+(new-goal auto-dispatch via `scheduler.run_scheduler_once()`), which is a
+separate, larger, deliberately-not-bundled decision (see that doc's
+non-goals/risks: "改变系统自主权边界"). This module only ever resumes a
+task that already exists and was already in progress; it never creates a
+new task or starts work nobody asked for.
+
+A task only becomes a reconciliation candidate if ALL of:
+1. `task.status` is in `task_execution._RESUMABLE_STATUSES`, and
+2. it has no execution happening in THIS process right now
+   (`task_execution.is_in_flight()` -- covers the common case where the
+   console's Start button and this loop share the same guard, so an
+   actively-running task is never double-dispatched), and
+3. its most recent `execution_event` is older than `stale_after_seconds`
+   (or it has none at all, in which case `task.created_at` is used) --
+   deliberately NOT `task.updated_at`, which does not change while a
+   phase is genuinely in progress (`implement_code`/
+   `validate_implementation_with_own_graph` never touch it mid-phase --
+   only the `task_status_changed` transitions do). Using
+   `task.updated_at` was the real blind spot during this incident's own
+   diagnosis: it looked "stale" for reasons unrelated to whether the task
+   was actually still being worked on.
+
+`_DEFAULT_STALE_AFTER_SECONDS` (1800s = 30 min) is comfortably above the
+longest normal single-node blocking call in this codebase (Validator's
+own work graph, `timeout=1200s`) so a node that's still genuinely inside
+one blocking `claude -p` call is never misdiagnosed as abandoned -- see
+the requirements doc's FR-3/non-goal #4 for why this doesn't need to be
+more precise than that for v1.
+"""
+
+import time
+from datetime import datetime, timezone
+
+from src import schema, task_execution
+
+_DEFAULT_STALE_AFTER_SECONDS = 1800
+
+
+def _to_epoch(sqlite_timestamp):
+    # Same parsing as work_node_run._parse_sqlite_timestamp: sqlite's
+    # CURRENT_TIMESTAMP / *_at defaults are UTC 'YYYY-MM-DD HH:MM:SS' with
+    # no timezone marker.
+    return datetime.strptime(sqlite_timestamp, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp()
+
+
+def find_stale_resumable_tasks(conn, stale_after_seconds=_DEFAULT_STALE_AFTER_SECONDS, now=time.time):
+    """Returns [{"task_id", "target_folder_path", "goal_text"}, ...] for
+    every task that's a genuine reconciliation candidate. Read-only --
+    does not dispatch anything, kept separate from `reconcile_once()` so
+    candidate selection is independently testable without spinning up
+    real background threads."""
+    cutoff = now() - stale_after_seconds
+    placeholders = ",".join("?" * len(task_execution._RESUMABLE_STATUSES))
+    rows = conn.execute(
+        "SELECT t.task_id, p.target_folder_path, t.goal_text, t.created_at, "
+        "(SELECT MAX(e.created_at) FROM execution_event e WHERE e.task_id = t.task_id) AS last_event_at "
+        "FROM task t JOIN project p ON p.project_id = t.project_id "
+        "WHERE t.status IN ({})".format(placeholders),
+        task_execution._RESUMABLE_STATUSES,
+    ).fetchall()
+
+    candidates = []
+    for task_id, target_folder_path, goal_text, created_at, last_event_at in rows:
+        if task_execution.is_in_flight(task_id):
+            continue
+        last_activity = _to_epoch(last_event_at or created_at)
+        if last_activity >= cutoff:
+            continue
+        candidates.append({"task_id": task_id, "target_folder_path": target_folder_path, "goal_text": goal_text})
+    return candidates
+
+
+def reconcile_once(db_path, stale_after_seconds=_DEFAULT_STALE_AFTER_SECONDS, now=time.time):
+    """One reconciliation pass: find candidates, dispatch each via
+    `task_execution.start_task_execution_in_background()`. Uses its own
+    short-lived connection for the scan, closed before dispatching -- the
+    background threads each open their own (same pattern as
+    `scheduler.py`). Returns the candidate list (dispatched or not) for
+    callers/tests to inspect.
+
+    A candidate that fails to dispatch (e.g. `TaskExecutionError` because
+    something else claimed it between the scan and here -- a real race,
+    not a bug) is skipped, not treated as a fatal error for the whole
+    pass: 04-P's "调度器继续执行队列中其他任务" applies here the same way
+    it does in `scheduler.run_scheduler_once`."""
+    conn = schema.connect(db_path)
+    try:
+        candidates = find_stale_resumable_tasks(conn, stale_after_seconds, now)
+    finally:
+        conn.close()
+    for c in candidates:
+        try:
+            task_execution.start_task_execution_in_background(
+                db_path, c["task_id"], c["target_folder_path"], c["goal_text"]
+            )
+        except Exception:
+            pass
+    return candidates
@@ -10,6 +10,7 @@ import json
 import os
 import subprocess
 import tempfile
+import threading
 import unittest
 from unittest.mock import patch
  
@@ -136,6 +137,52 @@ class ResumeFromPlanningTest(TaskExecutionTestBase):
         self.assertEqual(mock_trigger.call_args[0][0], "validator")
  
  
+class ResumeFromValidatingTest(TaskExecutionTestBase):
+    """2026-08-06 real incident: a task stuck in 'validating' for 15+ hours
+    with no validator work_graph_run ever created, because nothing in
+    production polls scheduler.run_scheduler_once() and whatever one-off
+    call drove it that far had already exited. Resuming from 'validating'
+    must skip BOTH Planner and Coder (already done) AND the
+    trigger_org_node_full("validator", ...) readiness call (already ran
+    once -- that's literally how status became 'validating'), going
+    straight to validate_implementation_with_own_graph()."""
+
+    def test_skips_readiness_trigger_runs_validation_only(self):
+        self._insert_task("task-1", status="validating", assigned_org_node="validator")
+        self._write_artifact("task-1", "prd", "# PRD content")
+        self._write_artifact("task-1", "sdd", "# SDD content")
+        self._write_artifact("task-1", "benchmark", "# Benchmark content")
+        self._write_artifact("task-1", "code_diff", "diff content")
+        self._write_artifact("task-1", "test_report", "report content")
+
+        with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger) as mock_ready, \
+             patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
+             patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger:
+            mock_execute.side_effect = [
+                {"content": "# Validator's own Benchmark"},
+                {"content": "N/A: no dataset needed"},
+                {"content": "import sys\nsys.exit(0)\n"},
+            ]
+            result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
+
+        self.assertEqual(result["verdict"]["verdict"], "PASS")
+        self.assertEqual(result["code_artifact_ids"]["code_diff"], "art-code_diff")
+        # the readiness trigger (trigger_org_node_full's own PRE_STATE-guarded
+        # call) must NOT fire a second time -- it already succeeded once to
+        # get the task into 'validating' in the first place, and calling it
+        # again would hit org_node_lifecycle's own "needs status=coding" guard.
+        mock_ready.assert_not_called()
+        # only the real work_graph_run's own final call happened
+        mock_trigger.assert_called_once()
+        self.assertEqual(mock_trigger.call_args[0][0], "validator")
+        # a brand-new validator work_graph_run was created (none existed before)
+        rows = self.conn.execute(
+            "SELECT run_id FROM work_graph_run WHERE task_id = 'task-1' AND org_node = 'validator'"
+        ).fetchall()
+        self.assertEqual(len(rows), 1)
+
+
 class ResumeUnresumableStatusTest(TaskExecutionTestBase):
     def test_done_task_raises(self):
         self._insert_task("task-1", status="done")
@@ -207,6 +254,14 @@ class RequestPauseTest(TaskExecutionTestBase):
         result = task_execution.request_pause(self.conn, "task-1", "user_stopped")
         self.assertEqual(result["blocked_reason"], "user_stopped")
  
+    def test_pauses_a_validating_task(self):
+        # 2026-08-06: 'validating' stays task.status for the whole duration
+        # of a resumed validation run (validate_implementation_with_own_graph
+        # never updates task.status itself), so it must be pausable too.
+        self._insert_task("task-1", status="validating")
+        result = task_execution.request_pause(self.conn, "task-1", "user_paused")
+        self.assertEqual(result["status"], "blocked_pending_human")
+
     def test_cannot_pause_a_queued_task(self):
         self._insert_task("task-1", status="queued")
         with self.assertRaises(task_execution.TaskExecutionError):
@@ -228,6 +283,17 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
     per thread would be a completely separate, empty database)."""
  
     def setUp(self):
+        # Defense-in-depth against the 2026-08-06 pollution incident (see
+        # test_second_call_while_in_flight_raises_no_second_thread's own
+        # comment): `_in_flight_task_ids` is a process-wide global, and
+        # "task-1" is reused across many unrelated tests/files -- if any
+        # test's mocking is incomplete enough that a spawned thread outlives
+        # its own join(timeout=...), the claim leaks and silently fails
+        # OTHER tests depending on run order. Unconditionally clearing here
+        # makes that failure mode impossible regardless of what any
+        # individual test does right or wrong.
+        task_execution._release("task-1")
+        self.addCleanup(lambda: task_execution._release("task-1"))
         self.repo = tempfile.TemporaryDirectory()
         self.addCleanup(self.repo.cleanup)
         subprocess.run(["git", "init"], cwd=self.repo.name, check=True, capture_output=True)
@@ -300,6 +366,65 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
         status = conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
         conn.close()
         self.assertEqual(status, "done")
+        # 2026-08-06 (task_reconciliation): the in-flight guard must release
+        # once the thread is done, so a LATER dispatch of the same task_id
+        # (e.g. a future reconciliation pass) isn't permanently blocked by a
+        # completed run.
+        self.assertFalse(task_execution.is_in_flight("task-1"))
+
+    def test_second_call_while_in_flight_raises_no_second_thread(self):
+        # 2026-08-06 (task_reconciliation): guards against the console's
+        # Start button and a reconciliation pass racing to dispatch the SAME
+        # task_id concurrently -- the second call must fail fast, not spawn
+        # a second thread that would run the same task twice.
+        #
+        # Fully mocks the same real work the OTHER StartTaskExecutionIn
+        # BackgroundTest tests mock (write_workflow_script/execute_work_graph/
+        # org_node_trigger.trigger_org_node), not just the readiness trigger
+        # this test blocks on -- an earlier version of this test only
+        # patched the readiness trigger, so once unblocked the thread went on
+        # to attempt REAL subprocess/file work, outlived this test's own
+        # join(timeout=10), and left "task-1" permanently claimed in the
+        # process-wide `_in_flight_task_ids` set -- silently failing every
+        # OTHER test in the suite that reuses the literal task_id "task-1"
+        # (nearly all of them), depending on run order. Caught by running
+        # the full discover -s src suite, not by running this file alone.
+        self._write_artifact("task-1", "prd", "# PRD")
+        self._write_artifact("task-1", "sdd", "# SDD")
+        self._write_artifact("task-1", "benchmark", "# Benchmark")
+        self._write_artifact("task-1", "code_diff", "diff")
+        self._write_artifact("task-1", "test_report", "report")
+
+        release_first_call = threading.Event()
+        entered_first_call = threading.Event()
+
+        def blocking_trigger(org_node, prompt, target_folder_path, timeout=300):
+            entered_first_call.set()
+            release_first_call.wait(timeout=10)
+            return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout)
+
+        with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=blocking_trigger), \
+             patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
+             patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS"):
+            mock_execute.side_effect = [
+                {"content": "# Validator's own Benchmark"},
+                {"content": "N/A: no dataset needed"},
+                {"content": "import sys\nsys.exit(0)\n"},
+            ]
+            thread = task_execution.start_task_execution_in_background(
+                self.db_path, "task-1", self.repo.name, "a goal",
+            )
+            entered_first_call.wait(timeout=10)
+            self.assertTrue(task_execution.is_in_flight("task-1"))
+            with self.assertRaises(task_execution.TaskExecutionError):
+                task_execution.start_task_execution_in_background(
+                    self.db_path, "task-1", self.repo.name, "a goal",
+                )
+            release_first_call.set()
+            thread.join(timeout=10)
+
+        self.assertFalse(task_execution.is_in_flight("task-1"))
  
     def test_generic_failure_marks_blocked(self):
         self._write_artifact("task-1", "prd", "# PRD")
+"""task_reconciliation.py tests (2026-08-06, real 15h-stuck-task incident):
+mocked at the subprocess boundary for reconcile_once, real sqlite for
+candidate-selection logic -- same split as test_task_execution.py.
+
+Run: python3 -m unittest src.test_task_reconciliation -v
+"""
+
+import subprocess
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+from src import project_scaffold, schema, task_execution, task_reconciliation
+
+_NOW_EPOCH = 2_000_000_000  # fixed reference point, independent of wall clock
+
+
+def _ts(epoch_seconds):
+    return datetime.fromtimestamp(epoch_seconds, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+
+class ReconciliationTestBase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        project = project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
+        self.project_id = project["project_id"]
+
+    def tearDown(self):
+        self.conn.close()
+
+    def _insert_task(self, task_id, status, created_at_epoch, assigned_org_node="planner"):
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES (?, ?, 'a goal', 'goal', ?, ?, ?)",
+            (task_id, self.project_id, status, assigned_org_node, _ts(created_at_epoch)),
+        )
+        self.conn.commit()
+
+    def _insert_event(self, task_id, created_at_epoch, event_type="log_appended"):
+        self.conn.execute(
+            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload, created_at) "
+            "VALUES (?, ?, ?, ?, '{}', ?)",
+            ("ev-" + task_id + "-" + str(created_at_epoch), self.project_id, task_id, event_type, _ts(created_at_epoch)),
+        )
+        self.conn.commit()
+
+
+class FindStaleResumableTasksTest(ReconciliationTestBase):
+    def test_finds_task_with_no_activity_at_all(self):
+        self._insert_task("task-1", "coding", created_at_epoch=_NOW_EPOCH - 3600)
+        result = task_reconciliation.find_stale_resumable_tasks(
+            self.conn, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+        )
+        self.assertEqual([c["task_id"] for c in result], ["task-1"])
+
+    def test_excludes_task_with_recent_event(self):
+        self._insert_task("task-1", "validating", created_at_epoch=_NOW_EPOCH - 7200)
+        self._insert_event("task-1", created_at_epoch=_NOW_EPOCH - 60)  # 1 min ago -- fresh
+        result = task_reconciliation.find_stale_resumable_tasks(
+            self.conn, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+        )
+        self.assertEqual(result, [])
+
+    def test_uses_most_recent_event_not_earliest(self):
+        self._insert_task("task-1", "coding", created_at_epoch=_NOW_EPOCH - 7200)
+        self._insert_event("task-1", created_at_epoch=_NOW_EPOCH - 5000)  # old
+        self._insert_event("task-1", created_at_epoch=_NOW_EPOCH - 60)  # recent -- MAX() must win
+        result = task_reconciliation.find_stale_resumable_tasks(
+            self.conn, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+        )
+        self.assertEqual(result, [])
+
+    def test_stale_event_still_flagged(self):
+        self._insert_task("task-1", "coding", created_at_epoch=_NOW_EPOCH - 7200)
+        self._insert_event("task-1", created_at_epoch=_NOW_EPOCH - 5000)  # > 1800s ago
+        result = task_reconciliation.find_stale_resumable_tasks(
+            self.conn, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+        )
+        self.assertEqual([c["task_id"] for c in result], ["task-1"])
+
+    def test_excludes_non_resumable_status(self):
+        self._insert_task("task-1", "done", created_at_epoch=_NOW_EPOCH - 100000)
+        self._insert_task("task-2", "blocked_pending_human", created_at_epoch=_NOW_EPOCH - 100000)
+        result = task_reconciliation.find_stale_resumable_tasks(
+            self.conn, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+        )
+        self.assertEqual(result, [])
+
+    def test_excludes_in_flight_task(self):
+        self._insert_task("task-1", "validating", created_at_epoch=_NOW_EPOCH - 100000)
+        with patch.object(task_execution, "is_in_flight", return_value=True):
+            result = task_reconciliation.find_stale_resumable_tasks(
+                self.conn, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+            )
+        self.assertEqual(result, [])
+
+    def test_candidate_carries_target_folder_path_and_goal(self):
+        self._insert_task("task-1", "queued", created_at_epoch=_NOW_EPOCH - 100000)
+        result = task_reconciliation.find_stale_resumable_tasks(
+            self.conn, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+        )
+        self.assertEqual(result, [{"task_id": "task-1", "target_folder_path": self.tmp.name, "goal_text": "a goal"}])
+
+
+class ReconcileOnceTest(unittest.TestCase):
+    """Real file db (own connection per dispatched thread, same reasoning
+    as StartTaskExecutionInBackgroundTest) -- dispatch itself is mocked so
+    this stays at the "did we correctly decide to call it" boundary, not
+    a re-test of run_or_resume_task_lifecycle's own mechanics."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
+        self.tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+        self.tmp_db.close()
+        self.db_path = self.tmp_db.name
+        self.addCleanup(lambda: __import__("os").path.exists(self.db_path) and __import__("os").remove(self.db_path))
+
+        conn = schema.connect(self.db_path)
+        schema.create_schema(conn)
+        project = project_scaffold.create_project(conn, self.tmp.name, init_git_if_missing=False)
+        self.project_id = project["project_id"]
+        conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES ('stale-1', ?, 'goal 1', 'goal', 'coding', 'coder', ?)",
+            (self.project_id, _ts(_NOW_EPOCH - 100000)),
+        )
+        conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES ('fresh-1', ?, 'goal 2', 'goal', 'coding', 'coder', ?)",
+            (self.project_id, _ts(_NOW_EPOCH - 10)),
+        )
+        conn.commit()
+        conn.close()
+
+    def test_dispatches_only_stale_candidates(self):
+        with patch.object(task_execution, "start_task_execution_in_background") as mock_start:
+            result = task_reconciliation.reconcile_once(
+                self.db_path, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+            )
+        self.assertEqual([c["task_id"] for c in result], ["stale-1"])
+        mock_start.assert_called_once_with(self.db_path, "stale-1", self.tmp.name, "goal 1")
+
+    def test_one_dispatch_failure_does_not_block_others(self):
+        conn = schema.connect(self.db_path)
+        conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, created_at) "
+            "VALUES ('stale-2', ?, 'goal 3', 'goal', 'validating', 'validator', ?)",
+            (self.project_id, _ts(_NOW_EPOCH - 100000)),
+        )
+        conn.commit()
+        conn.close()
+
+        with patch.object(
+            task_execution, "start_task_execution_in_background",
+            side_effect=[task_execution.TaskExecutionError("already in flight"), None],
+        ) as mock_start:
+            result = task_reconciliation.reconcile_once(
+                self.db_path, stale_after_seconds=1800, now=lambda: _NOW_EPOCH
+            )
+        self.assertEqual(len(result), 2)
+        self.assertEqual(mock_start.call_count, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()