diff --git a/docs/todo/scheduler-production-polling-loop-design.md b/docs/todo/scheduler-production-polling-loop-design.md new file mode 100644 index 0000000..fcf2cc2 --- /dev/null +++ b/docs/todo/scheduler-production-polling-loop-design.md @@ -0,0 +1,77 @@ +# 生产环境调度/复原循环——详细设计（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()`。 diff --git a/docs/todo/scheduler-production-polling-loop-requirements.md b/docs/todo/scheduler-production-polling-loop-requirements.md new file mode 100644 index 0000000..d464f83 --- /dev/null +++ b/docs/todo/scheduler-production-polling-loop-requirements.md @@ -0,0 +1,64 @@ +# 生产环境调度/复原循环——需求文档 + +**日期**: 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 的互斥机制具体怎么实现，防止同一个任务被并发重复触发。 +- [ ] 这两个机制要不要暴露成可配置项（管理后台开关/环境变量），还是先硬编码一个保守的默认行为。 diff --git a/src/backend_server.py b/src/backend_server.py index d96f27c..2f85ddc 100644 --- a/src/backend_server.py +++ b/src/backend_server.py @@ -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="127.0.0.1", 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 diff --git a/src/task_execution.py b/src/task_execution.py index 6795689..bdb4701 100644 --- a/src/task_execution.py +++ b/src/task_execution.py @@ -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) ) diff --git a/src/task_reconciliation.py b/src/task_reconciliation.py new file mode 100644 index 0000000..cc36b33 --- /dev/null +++ b/src/task_reconciliation.py @@ -0,0 +1,106 @@ +"""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 diff --git a/src/test_task_execution.py b/src/test_task_execution.py index 1363f1c..bfd3d28 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -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") diff --git a/src/test_task_reconciliation.py b/src/test_task_reconciliation.py new file mode 100644 index 0000000..ca54822 --- /dev/null +++ b/src/test_task_reconciliation.py @@ -0,0 +1,172 @@ +"""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()