phase3(契约层): protocol.ts + 落盘状态 + store 失败策略/判活原语
多进程执行解耦的 Phase 0 共享契约（worker 与 daemon 都 import 它）： - executor/protocol.ts: JobSpec(daemon→worker 唯一输入) + OutboxRecord(worker→daemon) + 文件 IO(job/outbox/heartbeat) + isWorkerAlive(心跳为主/boot 窗口 pid 兜底/防 pid 复用) - schema+db+types+mappers: runs.worker_pid/last_seq、tasks.next_eligible_at(持久化退避) - store: failTaskAttempt(失败/重试策略唯一落点) + setNextEligibleAt/setWorkerPid/setLastSeq + countExecuting + executingWithLatestExecutorRun + reconcileInterrupted(isAlive 注入,默认保守回收) 纯增量,140 测试全绿。worker.ts/pipeline.ts(worker 侧) 与 orchestrator/ingest/index(daemon 侧) 待接。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
@@ -29,6 +29,9 @@ export function openDb(file: string): Database.Database {
|
||||
ensureColumn(db, 'projects', 'max_retries', 'max_retries INTEGER NOT NULL DEFAULT 2'); // 失败后最大自动重试次数
|
||||
ensureColumn(db, 'projects', 'timeout_ms', 'timeout_ms INTEGER NOT NULL DEFAULT 1800000'); // 单次执行超时毫秒
|
||||
ensureColumn(db, 'tasks', 'retry_baseline', 'retry_baseline INTEGER NOT NULL DEFAULT 0'); // 手动重投时的失败基线
|
||||
ensureColumn(db, 'tasks', 'next_eligible_at', 'next_eligible_at TEXT'); // 持久化退避（多进程执行）
|
||||
ensureColumn(db, 'runs', 'worker_pid', 'worker_pid INTEGER'); // worker 进程 pid
|
||||
ensureColumn(db, 'runs', 'last_seq', 'last_seq INTEGER NOT NULL DEFAULT 0'); // outbox ingest 游标
|
||||
const schema = readFileSync(join(HERE, 'schema.sql'), 'utf8');
|
||||
db.exec(schema);
|
||||
return db;
|
||||
|
||||
@@ -16,6 +16,7 @@ export interface TaskRow {
|
||||
title: string; complexity: string; status: string; priority: number;
|
||||
deps: string; plan: string | null; spec: string | null; operations: string | null;
|
||||
result: string | null; assignee: string | null; retry_baseline: number;
|
||||
next_eligible_at: string | null;
|
||||
created_at: string; updated_at: string;
|
||||
source_ref: string | null;
|
||||
}
|
||||
@@ -27,6 +28,7 @@ export interface RunRow {
|
||||
id: string; task_id: string; kind: string; worktree: string | null; branch: string | null;
|
||||
status: string; started_at: string; ended_at: string | null;
|
||||
transcript_ref: string | null; claude_session_id: string | null; error: string | null;
|
||||
worker_pid: number | null; last_seq: number;
|
||||
}
|
||||
export interface EventRow {
|
||||
id: string; project_id: string; task_id: string | null; type: string; payload: string; at: string;
|
||||
@@ -68,6 +70,7 @@ export function rowToTask(r: TaskRow, approvals: ApprovalRecord[] = []): Task {
|
||||
plan: r.plan, spec: r.spec, operations: r.operations,
|
||||
approvals, result: r.result ? parseResult(r.result) : null,
|
||||
assignee: r.assignee as Task['assignee'], retryBaseline: r.retry_baseline ?? 0,
|
||||
nextEligibleAt: r.next_eligible_at ?? null,
|
||||
createdAt: r.created_at, updatedAt: r.updated_at,
|
||||
};
|
||||
}
|
||||
@@ -85,6 +88,7 @@ export function rowToRun(r: RunRow): Run {
|
||||
worktree: r.worktree, branch: r.branch, status: r.status as RunStatus,
|
||||
startedAt: r.started_at, endedAt: r.ended_at,
|
||||
transcriptRef: r.transcript_ref, claudeSessionId: r.claude_session_id, error: r.error,
|
||||
workerPid: r.worker_pid ?? null, lastSeq: r.last_seq ?? 0,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -36,6 +36,7 @@ CREATE TABLE IF NOT EXISTS tasks (
|
||||
result TEXT, -- JSON TaskResult
|
||||
assignee TEXT, -- agent | human
|
||||
retry_baseline INTEGER NOT NULL DEFAULT 0, -- 上次手动重投时已有的失败 run 数（重置重试计数用）
|
||||
next_eligible_at TEXT, -- 持久化退避：早于此时间不被领取（null=即刻可领）
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
source_ref TEXT -- 旧 todo 来源标识（todo:17 / todo:17/1A），项目内唯一
|
||||
@@ -66,7 +67,9 @@ CREATE TABLE IF NOT EXISTS runs (
|
||||
ended_at TEXT,
|
||||
transcript_ref TEXT,
|
||||
claude_session_id TEXT,
|
||||
error TEXT
|
||||
error TEXT,
|
||||
worker_pid INTEGER, -- 多进程执行：worker 进程 pid（daemon 写，判活用）
|
||||
last_seq INTEGER NOT NULL DEFAULT 0 -- 已 ingest 的 outbox 最大 seq（幂等游标）
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_task ON runs(task_id, started_at);
|
||||
|
||||
|
||||
+92
-10
@@ -231,8 +231,8 @@ export class Store {
|
||||
id: id('tsk'), project_id: input.projectId, parent_id: input.parentId ?? null, depth,
|
||||
title: input.title, complexity: input.complexity, status, priority: input.priority ?? 1,
|
||||
deps: JSON.stringify(input.deps ?? []), plan: null, spec: null, operations: null,
|
||||
result: null, assignee: null, retry_baseline: 0, created_at: now(), updated_at: now(),
|
||||
source_ref: null,
|
||||
result: null, assignee: null, retry_baseline: 0, next_eligible_at: null,
|
||||
created_at: now(), updated_at: now(), source_ref: null,
|
||||
};
|
||||
this.db.prepare(
|
||||
`INSERT INTO tasks (id,project_id,parent_id,depth,title,complexity,status,priority,deps,plan,spec,operations,result,assignee,retry_baseline,created_at,updated_at,source_ref)
|
||||
@@ -562,15 +562,86 @@ export class Store {
|
||||
* 卡在 executing 的任务转 failed→queued 等编排器重新领取。
|
||||
* 注：中断产生的 failed run 会计入该任务的失败次数（多次中断+真失败可能提前转 needs_attention，可接受）。
|
||||
*/
|
||||
reconcileInterrupted(): { runs: number; tasks: number } {
|
||||
const runs = this.db.prepare(`SELECT * FROM runs WHERE status = 'started'`).all() as RunRow[];
|
||||
for (const r of runs) this.finishRun(r.id, 'failed', { error: 'daemon 重启，执行中断' });
|
||||
const rows = this.db.prepare(`SELECT * FROM tasks WHERE status = 'executing'`).all() as TaskRow[];
|
||||
for (const t of rows) {
|
||||
this.transition(t.id, 'failed', { auto: 'daemon-restart' });
|
||||
this.transition(t.id, 'queued', { auto: 'daemon-restart' });
|
||||
/**
|
||||
* 失败/重试策略的唯一落点（ingest 的 failed 记录、reaper 判死、reconcile 判死 都调它）。
|
||||
* 把（可选）执行 run 收尾为 failed，按「净失败次数 vs project.maxRetries」决定：
|
||||
* 未到上限 → queued + 持久化退避（next_eligible_at = now + 指数退避）
|
||||
* 到上限 → needs_attention（清退避）
|
||||
* netFailed 相对 retry_baseline（手动重投时设的基线）。退避：min(30s·2^(n-1), 10min)。
|
||||
*/
|
||||
failTaskAttempt(taskId: string, runId: string | null, error: string): Task {
|
||||
const row = this.getTaskRow(taskId);
|
||||
if (!row) throw new StoreError(`任务不存在: ${taskId}`);
|
||||
const project = this.getProject(row.project_id);
|
||||
if (!project) throw new StoreError(`项目不存在: ${row.project_id}`);
|
||||
|
||||
// 收尾执行 run：给了 runId 且仍 started → 标 failed；没给 → 补记一条 failed（保证重试计数不漏）
|
||||
if (runId) {
|
||||
const rr = this.db.prepare(`SELECT status FROM runs WHERE id = ?`).get(runId) as { status: string } | undefined;
|
||||
if (rr && rr.status === 'started') this.finishRun(runId, 'failed', { error });
|
||||
} else {
|
||||
const r = this.startRun(taskId, 'executor');
|
||||
this.finishRun(r.id, 'failed', { error });
|
||||
}
|
||||
return { runs: runs.length, tasks: rows.length };
|
||||
|
||||
if ((this.getTaskRow(taskId)!.status as TaskStatus) !== 'failed') {
|
||||
this.transition(taskId, 'failed', { by: 'failTaskAttempt', error });
|
||||
}
|
||||
|
||||
const allFailed = (this.db.prepare(
|
||||
`SELECT COUNT(*) AS n FROM runs WHERE task_id = ? AND kind = 'executor' AND status = 'failed'`,
|
||||
).get(taskId) as { n: number }).n;
|
||||
const netFailed = Math.max(0, allFailed - row.retry_baseline);
|
||||
const priorNetFailed = Math.max(0, netFailed - 1); // 不含本次
|
||||
if (priorNetFailed < project.maxRetries) {
|
||||
const attempt = priorNetFailed + 1;
|
||||
const backoffMs = Math.min(30_000 * 2 ** (attempt - 1), 10 * 60_000);
|
||||
this.setNextEligibleAt(taskId, new Date(Date.now() + backoffMs).toISOString());
|
||||
this.transition(taskId, 'queued', { by: 'failTaskAttempt', retry: attempt, backoffMs });
|
||||
} else {
|
||||
this.setNextEligibleAt(taskId, null);
|
||||
this.transition(taskId, 'needs_attention', { by: 'failTaskAttempt', allFailed, netFailed });
|
||||
}
|
||||
return this.getTask(taskId)!;
|
||||
}
|
||||
|
||||
/** 持久化退避：早于 next_eligible_at 不被领取（null=即刻可领）。 */
|
||||
setNextEligibleAt(taskId: string, iso: string | null): void {
|
||||
this.db.prepare(`UPDATE tasks SET next_eligible_at = ?, updated_at = ? WHERE id = ?`).run(iso, now(), taskId);
|
||||
}
|
||||
|
||||
/** 项目内 executing 任务数（多进程执行的并发闸：每个 executing 对应一个活 worker）。 */
|
||||
countExecuting(projectId: string): number {
|
||||
return (this.db.prepare(
|
||||
`SELECT COUNT(*) AS n FROM tasks WHERE project_id = ? AND status = 'executing'`,
|
||||
).get(projectId) as { n: number }).n;
|
||||
}
|
||||
|
||||
/** 所有 executing 任务 + 其最近一条 executor run（reaper / reconcile 判活用）。run 可能为 null（异常）。 */
|
||||
executingWithLatestExecutorRun(): Array<{ task: Task; run: Run | null }> {
|
||||
const rows = this.db.prepare(`SELECT * FROM tasks WHERE status = 'executing'`).all() as TaskRow[];
|
||||
return rows.map((t) => {
|
||||
const rr = this.db.prepare(
|
||||
`SELECT * FROM runs WHERE task_id = ? AND kind = 'executor' ORDER BY started_at DESC LIMIT 1`,
|
||||
).get(t.id) as RunRow | undefined;
|
||||
return { task: rowToTask(t), run: rr ? rowToRun(rr) : null };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动中断恢复（多进程执行）：逐个 executing 任务按注入的 isAlive 判活——
|
||||
* 活 → re-adopt（保留 executing，daemon 续 ingest 其 outbox）；
|
||||
* 死 → failTaskAttempt（收尾 + 重试/needs_attention）。
|
||||
* isAlive 由 daemon 注入（protocol.isWorkerAlive，基于 worker_pid + 心跳）；默认保守判死（无判活信息时回收）。
|
||||
*/
|
||||
reconcileInterrupted(isAlive: (run: Run | null) => boolean = () => false): { readopted: number; reclaimed: number } {
|
||||
let readopted = 0, reclaimed = 0;
|
||||
for (const { task, run } of this.executingWithLatestExecutorRun()) {
|
||||
if (run && isAlive(run)) { readopted++; continue; }
|
||||
this.failTaskAttempt(task.id, run?.id ?? null, 'daemon 重启时发现 worker 已退出');
|
||||
reclaimed++;
|
||||
}
|
||||
return { readopted, reclaimed };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -687,6 +758,7 @@ export class Store {
|
||||
const rr: RunRow = {
|
||||
id: id('run'), task_id: taskId, kind, worktree: fields.worktree ?? null, branch: fields.branch ?? null,
|
||||
status: 'started', started_at: now(), ended_at: null, transcript_ref: null, claude_session_id: null, error: null,
|
||||
worker_pid: null, last_seq: 0,
|
||||
};
|
||||
this.db.prepare(
|
||||
`INSERT INTO runs (id,task_id,kind,worktree,branch,status,started_at,ended_at,transcript_ref,claude_session_id,error)
|
||||
@@ -707,6 +779,16 @@ export class Store {
|
||||
return rowToRun(this.db.prepare(`SELECT * FROM runs WHERE id = ?`).get(runId) as RunRow);
|
||||
}
|
||||
|
||||
/** 记录执行该 run 的 worker 进程 pid（多进程执行，daemon spawn 后写；判活用）。 */
|
||||
setWorkerPid(runId: string, pid: number): void {
|
||||
this.db.prepare(`UPDATE runs SET worker_pid = ? WHERE id = ?`).run(pid, runId);
|
||||
}
|
||||
|
||||
/** 更新已 ingest 的 outbox 最大 seq（幂等游标；daemon ingest 后写，重启续读）。 */
|
||||
setLastSeq(runId: string, seq: number): void {
|
||||
this.db.prepare(`UPDATE runs SET last_seq = ? WHERE id = ?`).run(seq, runId);
|
||||
}
|
||||
|
||||
/** 所有进行中的 run（status='started'），联 tasks 取任务标题与项目。 */
|
||||
activeRuns(): ActiveRun[] {
|
||||
const rows = this.db.prepare(
|
||||
|
||||
Reference in New Issue
Block a user