恢复无执行记录的遗留运行任务

Author: javazys

docs/architecture.md

@@ -2,7 +2,7 @@
 
 > 状态：当前有效
 > 最后核对：2026-07-15
-> 适用版本：源码 `main` 分支，当前发布基线 0.1.21
+> 适用版本：源码 `main` 分支，当前开发基线 0.1.22
 
 ## 结论与适用边界
 
@@ -100,6 +100,7 @@ pending / running / failed ──cancel──> cancelled
 - 同一非空 `batchId` 在单个 Gateway 内串行；不同批次和空 `batchId` 可以并行。
 - Worker 硬超时、运行中取消和宽限期后的关闭会终止子进程树；Watchdog 在使用数据库记录的旧 PID 前，还会校验进程命令是否匹配配置的 OpenCode 可执行文件。
 - Worker 定时更新 `task_runs.heartbeatAt`。Watchdog 发现心跳过期时终止记录的 `childPid`、关闭本次 run，并按同一重试预算恢复任务。
+- Gateway 获得单实例锁后、Worker 接单前，会把不存在 active run 的遗留 `running` 任务恢复为 `pending`，覆盖进程在创建 run 前后崩溃留下的孤儿状态。
 - 输出只保留最后 64 KiB，避免单任务无限占用 Gateway 内存；发现 JSON 输出中的 `sessionID` 会写入执行记录。
 
 该实现提供 at-least-once 倾向的本地恢复，不保证 exactly-once。任务执行外部副作用时，应自行设计幂等键或可重复执行策略。

package.json

@@ -1,6 +1,6 @@
 {
     "name": "opencode-supertask",
-    "version": "0.1.21",
+    "version": "0.1.22",
     "description": "AI Agent 任务调度系统 — OpenCode 插件 + CLI + Gateway 常驻进程",
     "type": "module",
     "main": "dist/plugin/supertask.js",

src/core/services/task.service.ts

@@ -242,6 +242,28 @@ export class TaskService {
         return result.length;
     }
 
+    static async resetOrphanRunningToPending(): Promise<number> {
+        const result = await db
+            .update(tasks)
+            .set({
+                status: 'pending',
+                startedAt: null,
+                finishedAt: null,
+            })
+            .where(
+                and(
+                    eq(tasks.status, 'running'),
+                    sql`NOT EXISTS (
+                        SELECT 1 FROM ${taskRuns}
+                        WHERE ${taskRuns.taskId} = ${tasks.id}
+                          AND ${taskRuns.status} = 'running'
+                    )`,
+                ),
+            )
+            .returning();
+        return result.length;
+    }
+
     static async cancel(id: number, scope: { cwd?: string } = {}): Promise<Task | null> {
         const conditions = [
             eq(tasks.id, id),

src/gateway/index.ts

@@ -101,6 +101,16 @@ async function main() {
     const watchdog = new Watchdog(cfg);
     const scheduler = new Scheduler(cfg);
 
+    const recoveredOrphans = await TaskService.resetOrphanRunningToPending();
+    if (recoveredOrphans > 0) {
+        console.log(JSON.stringify({
+            ts: new Date().toISOString(),
+            level: 'warn',
+            msg: 'reset orphan running tasks to pending',
+            count: recoveredOrphans,
+        }));
+    }
+
     initializeGatewayHealth({
         workerPollIntervalMs: cfg.worker.pollIntervalMs,
         schedulerEnabled: cfg.scheduler.enabled,

tests/task-service.test.ts

@@ -1,6 +1,7 @@
 import { describe, test, expect, beforeEach } from 'bun:test';
 import { setupTestDb } from './helpers/mock-db';
 import { TaskService } from '../src/core/services/task.service';
+import { TaskRunService } from '../src/core/services/task-run.service';
 
 describe('TaskService', () => {
     beforeEach(() => {
@@ -900,5 +901,22 @@ describe('TaskService', () => {
             const count = await TaskService.resetRunningToPending([]);
             expect(count).toBe(0);
         });
+
+        test('启动恢复只重置没有 active run 的孤儿 running 任务', async () => {
+            const orphan = await TaskService.add({ name: '孤儿任务', agent: 'a', prompt: '恢复任务' });
+            const active = await TaskService.add({ name: '执行中任务', agent: 'a', prompt: '保持执行' });
+            const closed = await TaskService.add({ name: '已关闭 run 的任务', agent: 'a', prompt: '重新排队' });
+            await TaskService.start(orphan.id);
+            await TaskService.start(active.id);
+            await TaskService.start(closed.id);
+            await TaskRunService.create({ taskId: active.id });
+            const closedRun = await TaskRunService.create({ taskId: closed.id });
+            await TaskRunService.fail(closedRun.id, 'Gateway 在同步任务状态前退出');
+
+            expect(await TaskService.resetOrphanRunningToPending()).toBe(2);
+            expect((await TaskService.getById(orphan.id))?.status).toBe('pending');
+            expect((await TaskService.getById(closed.id))?.status).toBe('pending');
+            expect((await TaskService.getById(active.id))?.status).toBe('running');
+        });
     });
 });