diff --git a/src/api/server.ts b/src/api/server.ts index f325faa..91284f5 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -52,6 +52,8 @@ export function buildServer(opts: ApiOptions): FastifyInstance { verifyCmd: b.verifyCmd === undefined ? undefined : (b.verifyCmd === null ? null : String(b.verifyCmd)), autonomy: b.autonomy as never, model: b.model === undefined ? undefined : (b.model === null ? null : String(b.model)), concurrency: b.concurrency === undefined ? undefined : Number(b.concurrency), + maxRetries: b.maxRetries === undefined ? undefined : Number(b.maxRetries), + timeoutMs: b.timeoutMs === undefined ? undefined : Number(b.timeoutMs), })); }); @@ -73,6 +75,8 @@ export function buildServer(opts: ApiOptions): FastifyInstance { if (b.verifyCmd !== undefined) patch.verifyCmd = b.verifyCmd === null ? null : String(b.verifyCmd); if (b.model !== undefined) patch.model = b.model === null ? null : String(b.model); if (b.logo !== undefined) patch.logo = b.logo === null || b.logo === '' ? null : String(b.logo); + if (b.maxRetries !== undefined) patch.maxRetries = Number(b.maxRetries); + if (b.timeoutMs !== undefined) patch.timeoutMs = Number(b.timeoutMs); return projectOut(store.patchProject(id, patch)); }); @@ -238,6 +242,12 @@ export function buildServer(opts: ApiOptions): FastifyInstance { return store.deleteTask(id); }); + // needs_attention 任务一键重投：重置重试基线 + 转 queued，让编排器下一轮重新领取 + app.post('/api/tasks/:id/requeue', (req) => { + const { id } = req.params as { id: string }; + return store.requeueTask(id); + }); + // 审批闸：accept / reject（reject 必带 reason） // exec 闸的 accept = 通过并合并（PR 闭环）：先 merge 再 decide；merge 失败 → 400，任务保留在审核闸。 app.post('/api/tasks/:id/decide', async (req) => { diff --git a/src/daemon/orchestrator.ts b/src/daemon/orchestrator.ts index 6bf418e..ea4ca8a 100644 --- a/src/daemon/orchestrator.ts +++ b/src/daemon/orchestrator.ts @@ -7,8 +7,17 @@ import { runVerify, type VerifyFn } from '../executor/verify.js'; import { reviewCode, reviewSecurity, type ReviewerFn } from '../executor/reviewer.js'; import { pickModel } from '../executor/models.js'; -/** 失败后最多自动重试次数（重试 2 次 = 最多 3 次执行），之后 → needs_attention */ -export const MAX_RETRIES = 2; +/** 失败后默认最多自动重试次数（项目级 maxRetries 未设置时回退此值） */ +export const DEFAULT_MAX_RETRIES = 2; + +/** 指数退避基础等待（第 n 次重试等待 BASE * 2^(n-1)，上限 MAX） */ +const BACKOFF_BASE_MS = 30_000; // 30s +const BACKOFF_MAX_MS = 10 * 60_000; // 10min + +/** 第 attempt 次重试（1-indexed）的等待毫秒 */ +export function computeBackoffMs(attempt: number): number { + return Math.min(BACKOFF_BASE_MS * (2 ** (attempt - 1)), BACKOFF_MAX_MS); +} export interface OrchestratorLogger { info(msg: string): void; @@ -23,6 +32,8 @@ export interface OrchestratorDeps { verify: VerifyFn; createWorktree: (repoPath: string, taskId: string, baseBranch: string) => Promise; worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise; + /** 当前时间戳（ms）；测试注入可控时钟（默认 Date.now） */ + nowMs: () => number; } export interface Orchestrator { @@ -41,15 +52,18 @@ export interface Orchestrator { * 成功 → setResult + exec_review；失败 → failed → 重试 ≤MAX_RETRIES 次 → needs_attention。 */ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: Partial = {}): Orchestrator { - const d: OrchestratorDeps = { runner: runTask, reviewCode, reviewSecurity, verify: runVerify, createWorktree, worktreeDiff, ...deps }; + const d: OrchestratorDeps = { runner: runTask, reviewCode, reviewSecurity, verify: runVerify, createWorktree, worktreeDiff, nowMs: Date.now, ...deps }; const inflight = new Map(); // taskId → projectId const pending = new Set>(); + /** 退避表：taskId → 下次可领取时间戳（ms）；daemon 重启后清空，已有 queued 任务即刻可领 */ + const backoffUntil = new Map(); /** * 本项目可领取的任务：queued（重试/孤儿）+ ready 叶子且 deps 全 done；auto-easy 只挑 easy。 * 按调度分降序返回（score = 自身分 + 已完成依赖分 + 等待解锁的 blocked 任务分，见 model/scoring.ts）。 */ function claimable(project: Project): Array<{ task: Task; score: number }> { + const nowMs = d.nowMs(); const tasks = store.listTasks(project.id); const byId = new Map(tasks.map((t) => [t.id, t])); const parents = new Set(tasks.filter((t) => t.parentId).map((t) => t.parentId as string)); @@ -59,6 +73,8 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: if (t.status !== 'ready' && t.status !== 'queued') return false; if (parents.has(t.id)) return false; // 非叶子（容器）跳过 if (easyOnly && t.complexity !== 'easy') return false; + const until = backoffUntil.get(t.id); + if (until !== undefined && nowMs < until) return false; // 退避冷却中 return t.deps.every((dep) => byId.get(dep)?.status === 'done'); }); return rankByScore(candidates, tasks); @@ -160,14 +176,21 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: store.finishRun(r.id, 'failed', { error: msg }); } store.transition(task.id, 'failed', { by: 'orchestrator', error: msg }); - const failedRuns = store.listRuns(task.id).filter((r) => r.kind === 'executor' && r.status === 'failed').length; - const priorFailed = Math.max(0, failedRuns - 1); // 不含本次 - if (priorFailed < MAX_RETRIES) { - store.transition(task.id, 'queued', { by: 'orchestrator', retry: priorFailed + 1 }); - log.info(`任务 ${task.id} 重新入队（第 ${priorFailed + 1} 次重试，下一轮领取）`); + const allFailed = store.listRuns(task.id).filter((r) => r.kind === 'executor' && r.status === 'failed').length; + // 相对于上次手动重投基线的净失败次数（retryBaseline 在 requeueTask 时设置） + const netFailed = Math.max(0, allFailed - task.retryBaseline); + const priorNetFailed = Math.max(0, netFailed - 1); // 不含本次 + const maxRetries = project.maxRetries; + if (priorNetFailed < maxRetries) { + const attempt = priorNetFailed + 1; // 本次是第几次重试 + const backoffMs = computeBackoffMs(attempt); + backoffUntil.set(task.id, d.nowMs() + backoffMs); + store.transition(task.id, 'queued', { by: 'orchestrator', retry: attempt, backoffMs }); + log.info(`任务 ${task.id} 重新入队（第 ${attempt} 次重试，退避 ${backoffMs}ms 后可领）`); } else { - store.transition(task.id, 'needs_attention', { by: 'orchestrator', failedRuns }); - log.error(`任务 ${task.id} 连续失败 ${failedRuns} 次 → needs_attention`); + backoffUntil.delete(task.id); + store.transition(task.id, 'needs_attention', { by: 'orchestrator', allFailed, netFailed }); + log.error(`任务 ${task.id} 净失败 ${netFailed} 次（上限 ${maxRetries}）→ needs_attention`); } } catch (e2) { log.error(`任务 ${task.id} 失败收尾出错：${(e2 as Error).message}`); diff --git a/src/executor/runner.ts b/src/executor/runner.ts index 41fc8e4..ec0bdb0 100644 --- a/src/executor/runner.ts +++ b/src/executor/runner.ts @@ -20,7 +20,8 @@ export interface RunnerResult { export type RunnerFn = (task: Task, project: Project, worktree: WorktreeInfo, runId: string) => Promise; const DEFAULT_MAX_TURNS = 100; -const DEFAULT_TIMEOUT_MS = 30 * 60_000; // 整体兜底超时 30min +/** 若项目未配置 timeoutMs 时的全局兜底超时（30min） */ +const DEFAULT_TIMEOUT_MS = 30 * 60_000; /** 组装任务提示词：标题 + operations/spec 全文 + 执行约束 */ export function buildPrompt(task: Task): string { @@ -58,13 +59,14 @@ async function ensureCommitted(dir: string, task: Task): Promise { */ export async function runTask(task: Task, project: Project, worktree: WorktreeInfo, runId: string): Promise { const model = pickModel(task, project, 'executor'); + const timeoutMs = project.timeoutMs ?? DEFAULT_TIMEOUT_MS; const cc = await runClaude({ prompt: buildPrompt(task), cwd: worktree.dir, model, runId, maxTurns: DEFAULT_MAX_TURNS, - timeoutMs: DEFAULT_TIMEOUT_MS, + timeoutMs, permissionMode: 'acceptEdits', // worktree 内自动接受编辑 allowedTools: [ 'Read', 'Edit', 'Write', 'Glob', 'Grep', diff --git a/src/mcp/index.ts b/src/mcp/index.ts index 716ab4a..d41cff1 100644 --- a/src/mcp/index.ts +++ b/src/mcp/index.ts @@ -271,6 +271,20 @@ server.registerTool( run(() => api('GET', projectId ? `/api/approvals?projectId=${encodeURIComponent(projectId)}` : '/api/approvals')), ); +server.registerTool( + 'requeue_task', + { + title: '重投 needs_attention 任务', + description: + '将 needs_attention 状态的任务重置重试计数并重新入队（queued），让编排器下一轮自动领取执行。只适用于 needs_attention 状态，其他状态会返回错误。', + inputSchema: { + taskId: z.string().min(1).describe('任务 id（tsk_ 开头）'), + }, + }, + async ({ taskId }) => + run(() => api('POST', `/api/tasks/${encodeURIComponent(taskId)}/requeue`, {})), +); + // ---------- 启动 ---------- async function main(): Promise { diff --git a/src/model/types.ts b/src/model/types.ts index 75b3c48..884c1ed 100644 --- a/src/model/types.ts +++ b/src/model/types.ts @@ -19,6 +19,8 @@ export interface Project { autonomy: Autonomy; model: string | null; concurrency: number; // 每项目并发执行上限 + maxRetries: number; // 失败后最大自动重试次数（默认 2） + timeoutMs: number; // 单次执行超时毫秒（默认 1800000 = 30min） status: 'active' | 'paused'; logo: string | null; // 自定义 logo（URL/仓库内相对路径；null=自动解析） sortOrder: number; // 侧栏排序（小在前） @@ -67,6 +69,7 @@ export interface Task { approvals: ApprovalRecord[]; result: TaskResult | null; assignee: 'agent' | 'human' | null; + retryBaseline: number; // 上次手动重投时已有的失败 run 数（重置重试计数用） createdAt: string; updatedAt: string; } diff --git a/src/store/db.ts b/src/store/db.ts index 5198547..37b5aeb 100644 --- a/src/store/db.ts +++ b/src/store/db.ts @@ -26,6 +26,9 @@ export function openDb(file: string): Database.Database { ensureColumn(db, 'projects', 'last_sync_at', 'last_sync_at TEXT'); ensureColumn(db, 'projects', 'logo', 'logo TEXT'); // 自定义 logo（URL/仓库内相对路径，null=自动解析） ensureColumn(db, 'projects', 'sort_order', 'sort_order INTEGER NOT NULL DEFAULT 0'); // 侧栏排序 + 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'); // 手动重投时的失败基线 const schema = readFileSync(join(HERE, 'schema.sql'), 'utf8'); db.exec(schema); return db; diff --git a/src/store/mappers.ts b/src/store/mappers.ts index 77d91de..13c10f5 100644 --- a/src/store/mappers.ts +++ b/src/store/mappers.ts @@ -7,14 +7,16 @@ import type { RunKind, RunStatus, EventType } from '../model/types.js'; export interface ProjectRow { id: string; name: string; repo_path: string; default_branch: string; verify_cmd: string | null; autonomy: string; model: string | null; - concurrency: number; status: string; created_at: string; + concurrency: number; max_retries: number; timeout_ms: number; + status: string; created_at: string; last_sync_at: string | null; logo: string | null; sort_order: number; } export interface TaskRow { id: string; project_id: string; parent_id: string | null; depth: number; 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; created_at: string; updated_at: string; + result: string | null; assignee: string | null; retry_baseline: number; + created_at: string; updated_at: string; source_ref: string | null; } export interface ApprovalRow { @@ -34,7 +36,8 @@ export function rowToProject(r: ProjectRow): Project { return { id: r.id, name: r.name, repoPath: r.repo_path, defaultBranch: r.default_branch, verifyCmd: r.verify_cmd, autonomy: r.autonomy as Autonomy, model: r.model, - concurrency: r.concurrency, status: r.status as Project['status'], createdAt: r.created_at, + concurrency: r.concurrency, maxRetries: r.max_retries ?? 2, timeoutMs: r.timeout_ms ?? 1_800_000, + status: r.status as Project['status'], createdAt: r.created_at, lastSyncAt: r.last_sync_at, logo: r.logo, sortOrder: r.sort_order, }; } @@ -64,7 +67,8 @@ export function rowToTask(r: TaskRow, approvals: ApprovalRecord[] = []): Task { priority: r.priority, deps: JSON.parse(r.deps) as string[], plan: r.plan, spec: r.spec, operations: r.operations, approvals, result: r.result ? parseResult(r.result) : null, - assignee: r.assignee as Task['assignee'], createdAt: r.created_at, updatedAt: r.updated_at, + assignee: r.assignee as Task['assignee'], retryBaseline: r.retry_baseline ?? 0, + createdAt: r.created_at, updatedAt: r.updated_at, }; } diff --git a/src/store/schema.sql b/src/store/schema.sql index a6d70a6..0965981 100644 --- a/src/store/schema.sql +++ b/src/store/schema.sql @@ -11,6 +11,8 @@ CREATE TABLE IF NOT EXISTS projects ( autonomy TEXT NOT NULL DEFAULT 'manual', -- manual | auto-easy | auto-approved model TEXT, concurrency INTEGER NOT NULL DEFAULT 1, + max_retries INTEGER NOT NULL DEFAULT 2, -- 失败后最大自动重试次数 + timeout_ms INTEGER NOT NULL DEFAULT 1800000, -- 单次执行超时毫秒（默认 30min） status TEXT NOT NULL DEFAULT 'active', -- active | paused created_at TEXT NOT NULL, last_sync_at TEXT, -- 最近一次 todo.json 同步时间 @@ -31,8 +33,9 @@ CREATE TABLE IF NOT EXISTS tasks ( plan TEXT, -- Hard：分析 + 拆解 spec TEXT, -- Medium：改动 + 理由 operations TEXT, -- Easy：将执行的操作 - result TEXT, -- JSON TaskResult - assignee TEXT, -- agent | human + result TEXT, -- JSON TaskResult + assignee TEXT, -- agent | human + retry_baseline INTEGER NOT NULL DEFAULT 0, -- 上次手动重投时已有的失败 run 数（重置重试计数用） created_at TEXT NOT NULL, updated_at TEXT NOT NULL, source_ref TEXT -- 旧 todo 来源标识（todo:17 / todo:17/1A），项目内唯一 diff --git a/src/store/store.ts b/src/store/store.ts index b0a2087..0495a0e 100644 --- a/src/store/store.ts +++ b/src/store/store.ts @@ -20,6 +20,7 @@ export class StoreError extends Error {} export interface CreateProjectInput { name: string; repoPath: string; defaultBranch?: string; verifyCmd?: string | null; autonomy?: Autonomy; model?: string | null; concurrency?: number; + maxRetries?: number; timeoutMs?: number; } export interface CreateTaskInput { projectId: string; title: string; complexity: Complexity; @@ -28,6 +29,7 @@ export interface CreateTaskInput { export interface PatchProjectInput { autonomy?: Autonomy; concurrency?: number; verifyCmd?: string | null; model?: string | null; status?: 'active' | 'paused'; logo?: string | null; + maxRetries?: number; timeoutMs?: number; } export interface PatchTaskInput { title?: string; priority?: number; complexity?: Complexity; @@ -89,17 +91,25 @@ export class Store { // ---------- Projects ---------- createProject(input: CreateProjectInput): Project { + if (input.maxRetries !== undefined && (!Number.isInteger(input.maxRetries) || input.maxRetries < 0)) { + throw new StoreError('maxRetries 必须是 >=0 的整数'); + } + if (input.timeoutMs !== undefined && (!Number.isFinite(input.timeoutMs) || input.timeoutMs < 1000)) { + throw new StoreError('timeoutMs 必须是 >=1000 的数字（毫秒）'); + } const maxOrder = (this.db.prepare(`SELECT COALESCE(MAX(sort_order), -1) AS m FROM projects`).get() as { m: number }).m; const row: ProjectRow = { id: id('prj'), name: input.name, repo_path: input.repoPath, default_branch: input.defaultBranch ?? 'main', verify_cmd: input.verifyCmd ?? null, autonomy: input.autonomy ?? 'manual', model: input.model ?? null, - concurrency: input.concurrency ?? 1, status: 'active', created_at: now(), + concurrency: input.concurrency ?? 1, max_retries: input.maxRetries ?? 2, + timeout_ms: input.timeoutMs ?? 1_800_000, + status: 'active', created_at: now(), last_sync_at: null, logo: null, sort_order: maxOrder + 1, }; this.db.prepare( - `INSERT INTO projects (id,name,repo_path,default_branch,verify_cmd,autonomy,model,concurrency,status,created_at,last_sync_at,logo,sort_order) - VALUES (@id,@name,@repo_path,@default_branch,@verify_cmd,@autonomy,@model,@concurrency,@status,@created_at,@last_sync_at,@logo,@sort_order)`, + `INSERT INTO projects (id,name,repo_path,default_branch,verify_cmd,autonomy,model,concurrency,max_retries,timeout_ms,status,created_at,last_sync_at,logo,sort_order) + VALUES (@id,@name,@repo_path,@default_branch,@verify_cmd,@autonomy,@model,@concurrency,@max_retries,@timeout_ms,@status,@created_at,@last_sync_at,@logo,@sort_order)`, ).run(row); this.emit(row.id, null, 'task.created', { kind: 'project', name: row.name }); return rowToProject(row); @@ -159,6 +169,18 @@ export class Store { if (patch.verifyCmd !== undefined) { sets.push('verify_cmd = @verify_cmd'); args.verify_cmd = patch.verifyCmd; } if (patch.model !== undefined) { sets.push('model = @model'); args.model = patch.model; } if (patch.logo !== undefined) { sets.push('logo = @logo'); args.logo = patch.logo; } + if (patch.maxRetries !== undefined) { + if (!Number.isInteger(patch.maxRetries) || patch.maxRetries < 0) { + throw new StoreError('maxRetries 必须是 >=0 的整数'); + } + sets.push('max_retries = @max_retries'); args.max_retries = patch.maxRetries; + } + if (patch.timeoutMs !== undefined) { + if (!Number.isFinite(patch.timeoutMs) || patch.timeoutMs < 1000) { + throw new StoreError('timeoutMs 必须是 >=1000 的数字（毫秒）'); + } + sets.push('timeout_ms = @timeout_ms'); args.timeout_ms = patch.timeoutMs; + } if (sets.length > 0) { this.db.prepare(`UPDATE projects SET ${sets.join(', ')} WHERE id = @id`).run(args); @@ -209,12 +231,12 @@ 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, created_at: now(), updated_at: now(), + result: null, assignee: null, retry_baseline: 0, 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,created_at,updated_at,source_ref) - VALUES (@id,@project_id,@parent_id,@depth,@title,@complexity,@status,@priority,@deps,@plan,@spec,@operations,@result,@assignee,@created_at,@updated_at,@source_ref)`, + `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) + VALUES (@id,@project_id,@parent_id,@depth,@title,@complexity,@status,@priority,@deps,@plan,@spec,@operations,@result,@assignee,@retry_baseline,@created_at,@updated_at,@source_ref)`, ).run(row); this.emit(input.projectId, row.id, 'task.created', { title: row.title, complexity: row.complexity, status }); return rowToTask(row); @@ -552,6 +574,26 @@ export class Store { return this.getTask(taskId)!; } + /** + * 手动一键重投 needs_attention 任务： + * 1. 记录当前失败 run 数为新基线（重置重试计数）； + * 2. 转 needs_attention → queued，让编排器下一轮重新领取。 + */ + requeueTask(taskId: string): Task { + const row = this.getTaskRow(taskId); + if (!row) throw new StoreError(`任务不存在: ${taskId}`); + if (row.status !== 'needs_attention') { + throw new StoreError(`只有 needs_attention 状态的任务才可重投（当前状态：${row.status}）`); + } + const failedCount = (this.db.prepare( + `SELECT COUNT(*) AS n FROM runs WHERE task_id = ? AND kind = 'executor' AND status = 'failed'`, + ).get(taskId) as { n: number }).n; + this.db.prepare(`UPDATE tasks SET retry_baseline = ?, updated_at = ? WHERE id = ?`).run(failedCount, now(), taskId); + this.transition(taskId, 'queued', { by: 'user', action: 'requeue' }); + this.emit(row.project_id, taskId, 'task.updated', { field: 'retryBaseline', retryBaseline: failedCount }); + return this.getTask(taskId)!; + } + // ---------- Runs ---------- startRun(taskId: string, kind: Run['kind'], fields: Partial> = {}): Run { const row = this.getTaskRow(taskId); diff --git a/test/orchestrator.test.ts b/test/orchestrator.test.ts index 47829f1..c0a0c81 100644 --- a/test/orchestrator.test.ts +++ b/test/orchestrator.test.ts @@ -1,7 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { Store } from '../src/store/index.js'; -import { createOrchestrator, MAX_RETRIES, type OrchestratorDeps } from '../src/daemon/orchestrator.js'; +import { createOrchestrator, DEFAULT_MAX_RETRIES, computeBackoffMs, type OrchestratorDeps } from '../src/daemon/orchestrator.js'; import type { RunnerResult } from '../src/executor/runner.js'; import type { ReviewResult } from '../src/executor/reviewer.js'; import type { Autonomy } from '../src/model/types.js'; @@ -39,10 +39,17 @@ function mockDeps(overrides: Partial = {}): OrchestratorDeps { runner: async () => okRun, reviewCode: async () => okReview, reviewSecurity: async () => okSecurity, + nowMs: Date.now, ...overrides, }; } +/** 可前进的测试时钟（用于退避相关测试） */ +function makeClock(initialMs = 0): { nowMs: () => number; advance: (ms: number) => void } { + let t = initialMs; + return { nowMs: () => t, advance: (ms) => { t += ms; } }; +} + function setup(autonomy: Autonomy, concurrency = 1): { store: Store; projectId: string } { const store = new Store(':memory:'); const p = store.createProject({ @@ -291,32 +298,165 @@ test('verify 不过：按失败处理（run failed + 重新入队）', async () store.close(); }); -test(`失败重试：重试 ${MAX_RETRIES} 次后 → needs_attention（共 ${MAX_RETRIES + 1} 次失败 run）`, async () => { +test(`失败重试：重试 ${DEFAULT_MAX_RETRIES} 次后 → needs_attention（共 ${DEFAULT_MAX_RETRIES + 1} 次失败 run）`, async () => { const { store, projectId } = setup('auto-easy'); const t = store.createTask({ projectId, title: 'flaky', complexity: 'easy' }); let attempts = 0; + // 使用快进时钟跳过退避冷却（每次 tick 前将时钟推进 10min） + const clock = makeClock(); const orch = createOrchestrator(store, noopLog, mockDeps({ runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: `boom #${attempts}` }; }, + nowMs: clock.nowMs, })); - for (let i = 1; i <= MAX_RETRIES; i++) { + for (let i = 1; i <= DEFAULT_MAX_RETRIES; i++) { + clock.advance(10 * 60_000); // 跳过退避冷却 orch.tick(); await orch.drain(); assert.equal(store.getTask(t.id)!.status, 'queued', `第 ${i} 次失败后应重新入队`); } + clock.advance(10 * 60_000); // 跳过最后一次退避 orch.tick(); // 最后一次重试也失败 await orch.drain(); assert.equal(store.getTask(t.id)!.status, 'needs_attention'); - assert.equal(attempts, MAX_RETRIES + 1); + assert.equal(attempts, DEFAULT_MAX_RETRIES + 1); const failed = store.listRuns(t.id).filter((r) => r.status === 'failed'); - assert.equal(failed.length, MAX_RETRIES + 1); + assert.equal(failed.length, DEFAULT_MAX_RETRIES + 1); + clock.advance(10 * 60_000); orch.tick(); // needs_attention 不会再被领取 await orch.drain(); - assert.equal(attempts, MAX_RETRIES + 1); + assert.equal(attempts, DEFAULT_MAX_RETRIES + 1); + store.close(); +}); + +test('项目级 maxRetries=1：只重试 1 次就 → needs_attention', async () => { + const store = new Store(':memory:'); + const p = store.createProject({ + name: 'orch', repoPath: '/tmp/orch-repo-retries-' + Math.random(), + autonomy: 'auto-easy', maxRetries: 1, + }); + const projectId = p.id; + const t = store.createTask({ projectId, title: 'fail1', complexity: 'easy' }); + + let attempts = 0; + const clock = makeClock(); + const orch = createOrchestrator(store, noopLog, mockDeps({ + runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, + nowMs: clock.nowMs, + })); + + // 第 1 次执行失败 → 重新入队（还有 1 次重试机会） + orch.tick(); + await orch.drain(); + assert.equal(store.getTask(t.id)!.status, 'queued'); + + // 第 2 次失败 → 超过 maxRetries=1 → needs_attention（快进时钟跳过退避） + clock.advance(10 * 60_000); + orch.tick(); + await orch.drain(); + assert.equal(store.getTask(t.id)!.status, 'needs_attention'); + assert.equal(attempts, 2); + store.close(); +}); + +test('项目级 maxRetries=0：首次失败直接 needs_attention（不重试）', async () => { + const store = new Store(':memory:'); + const p = store.createProject({ + name: 'orch', repoPath: '/tmp/orch-repo-zero-' + Math.random(), + autonomy: 'auto-easy', maxRetries: 0, + }); + const t = store.createTask({ projectId: p.id, title: 'fail0', complexity: 'easy' }); + + let attempts = 0; + const orch = createOrchestrator(store, noopLog, mockDeps({ + runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, + })); + + orch.tick(); + await orch.drain(); + assert.equal(store.getTask(t.id)!.status, 'needs_attention'); + assert.equal(attempts, 1); + store.close(); +}); + +test('computeBackoffMs：指数退避公式，30s 基数，上限 10min', () => { + assert.equal(computeBackoffMs(1), 30_000); + assert.equal(computeBackoffMs(2), 60_000); + assert.equal(computeBackoffMs(3), 120_000); + assert.equal(computeBackoffMs(4), 240_000); + assert.equal(computeBackoffMs(10), 10 * 60_000); // capped at 10min + assert.equal(computeBackoffMs(100), 10 * 60_000); +}); + +test('退避冷却：首次失败重入队后，退避期内不被领取；退避过期后正常领取', async () => { + const { store, projectId } = setup('auto-easy'); + const t = store.createTask({ projectId, title: 'backoff', complexity: 'easy' }); + + let attempts = 0; + // 使用可控时钟：初始时间 0，退避 30s（attempt=1 → BACKOFF_BASE=30000ms） + const clock = makeClock(0); + const orch = createOrchestrator(store, noopLog, mockDeps({ + runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, + nowMs: clock.nowMs, + })); + + // 第 1 次失败（t=0）→ 重入队 + 退避（backoffUntil = 0 + 30000 = 30000ms） + orch.tick(); + await orch.drain(); + assert.equal(store.getTask(t.id)!.status, 'queued'); + assert.equal(attempts, 1); + + // t=29999：退避期内，不被领取 + clock.advance(29_999); + orch.tick(); + await orch.drain(); + assert.equal(attempts, 1, '退避期内不应再次执行'); + assert.equal(store.getTask(t.id)!.status, 'queued'); + + // t=30001：退避过期，任务应被重新领取 + clock.advance(2); + orch.tick(); + await orch.drain(); + assert.equal(attempts, 2, '退避过期后应再次执行'); + store.close(); +}); + +test('requeueTask：needs_attention → queued，重置重试基线，再次允许重试', async () => { + const { store, projectId } = setup('auto-easy'); + const t = store.createTask({ projectId, title: 'requeue', complexity: 'easy' }); + + let attempts = 0; + const clock = makeClock(); + const orch = createOrchestrator(store, noopLog, mockDeps({ + runner: async () => { attempts++; return { ok: false, transcriptRef: null, sessionId: null, error: 'boom' }; }, + nowMs: clock.nowMs, + })); + + // 耗尽默认重试次数 → needs_attention（每次需快进时钟跳过退避） + for (let i = 0; i <= DEFAULT_MAX_RETRIES; i++) { + clock.advance(10 * 60_000); + orch.tick(); + await orch.drain(); + } + assert.equal(store.getTask(t.id)!.status, 'needs_attention'); + assert.equal(attempts, DEFAULT_MAX_RETRIES + 1); + + // 手动重投：重置基线，转 queued + const requeued = store.requeueTask(t.id); + assert.equal(requeued.status, 'queued'); + assert.equal(requeued.retryBaseline, DEFAULT_MAX_RETRIES + 1); + + // 重投后编排器应能再次执行（快进时钟确保无退避阻拦） + clock.advance(10 * 60_000); + orch.tick(); + await orch.drain(); + assert.equal(attempts, DEFAULT_MAX_RETRIES + 2, '重投后应再次执行'); + // 第 1 次净失败后应重入队（基线已重置，净失败=1 < maxRetries=2） + assert.equal(store.getTask(t.id)!.status, 'queued'); store.close(); }); diff --git a/test/store.test.ts b/test/store.test.ts index 24a8439..a3df7b6 100644 --- a/test/store.test.ts +++ b/test/store.test.ts @@ -301,3 +301,69 @@ test('合并失败补救：建最高优先级 easy 任务，且幂等不重复 assert.notEqual(fresh.id, rem.id, '旧补救任务已结束 → 建新的'); s.close(); }); + +test('patchProject：maxRetries/timeoutMs 可配 + 校验', () => { + const s = freshStore(); + const p = s.createProject({ name: 'cfg', repoPath: '/tmp/cfg-' + Math.random() }); + // 默认值 + assert.equal(p.maxRetries, 2); + assert.equal(p.timeoutMs, 1_800_000); + + // 更新 maxRetries + timeoutMs + const updated = s.patchProject(p.id, { maxRetries: 5, timeoutMs: 60_000 }); + assert.equal(updated.maxRetries, 5); + assert.equal(updated.timeoutMs, 60_000); + + // 校验：maxRetries 必须 >=0 的整数 + assert.throws(() => s.patchProject(p.id, { maxRetries: -1 }), /maxRetries/); + assert.throws(() => s.patchProject(p.id, { maxRetries: 1.5 }), /maxRetries/); + + // 校验：timeoutMs 必须 >=1000 + assert.throws(() => s.patchProject(p.id, { timeoutMs: 500 }), /timeoutMs/); + s.close(); +}); + +test('createProject：maxRetries/timeoutMs 自定义初值', () => { + const s = freshStore(); + const p = s.createProject({ name: 'custom', repoPath: '/tmp/custom-' + Math.random(), maxRetries: 0, timeoutMs: 120_000 }); + assert.equal(p.maxRetries, 0); + assert.equal(p.timeoutMs, 120_000); + s.close(); +}); + +test('requeueTask：needs_attention → queued，重置 retryBaseline', () => { + const s = freshStore(); + const p = s.createProject({ name: 'rq', repoPath: '/tmp/rq-' + Math.random() }); + const t = s.createTask({ projectId: p.id, title: 'flaky', complexity: 'easy' }); + + // 推进到 needs_attention（模拟 3 次失败 run） + s.transition(t.id, 'queued'); + s.transition(t.id, 'executing'); + s.transition(t.id, 'failed'); + s.transition(t.id, 'queued'); + s.transition(t.id, 'executing'); + s.transition(t.id, 'failed'); + s.transition(t.id, 'queued'); + s.transition(t.id, 'executing'); + s.transition(t.id, 'failed'); + s.transition(t.id, 'needs_attention'); + + // 模拟 3 条 failed executor run + const r1 = s.startRun(t.id, 'executor'); + s.finishRun(r1.id, 'failed', { error: 'boom1' }); + const r2 = s.startRun(t.id, 'executor'); + s.finishRun(r2.id, 'failed', { error: 'boom2' }); + const r3 = s.startRun(t.id, 'executor'); + s.finishRun(r3.id, 'failed', { error: 'boom3' }); + + assert.equal(s.getTask(t.id)!.retryBaseline, 0); + + // 一键重投 + const requeued = s.requeueTask(t.id); + assert.equal(requeued.status, 'queued'); + assert.equal(requeued.retryBaseline, 3); // 基线设为当前失败 run 数 + + // 只有 needs_attention 状态才可重投 + assert.throws(() => s.requeueTask(t.id), /needs_attention/); + s.close(); +}); diff --git a/test/sync.test.ts b/test/sync.test.ts index 912dbef..0d8ee9c 100644 --- a/test/sync.test.ts +++ b/test/sync.test.ts @@ -169,17 +169,23 @@ test('迁移：旧版库（无 source_ref/last_sync_at）打开后补列且数 const projCols = (db.prepare(`PRAGMA table_info(projects)`).all() as Array<{ name: string }>).map((c) => c.name); const taskCols = (db.prepare(`PRAGMA table_info(tasks)`).all() as Array<{ name: string }>).map((c) => c.name); assert.ok(projCols.includes('last_sync_at')); + assert.ok(projCols.includes('max_retries')); + assert.ok(projCols.includes('timeout_ms')); assert.ok(taskCols.includes('source_ref')); + assert.ok(taskCols.includes('retry_baseline')); db.close(); - // Store 能正常读旧数据，新字段为 null + // Store 能正常读旧数据，新字段使用默认值 const s = new Store(file); const projects = s.listProjects(); assert.equal(projects.length, 1); assert.equal(projects[0].name, '旧项目'); assert.equal(projects[0].lastSyncAt, null); + assert.equal(projects[0].maxRetries, 2); // 默认值 + assert.equal(projects[0].timeoutMs, 1_800_000); // 默认 30min const t = s.getTask('tsk_old'); assert.equal(t?.title, '旧任务'); + assert.equal(t?.retryBaseline, 0); // 默认基线 // 新方法在迁移后的旧库上可用 s.setSourceRef('tsk_old', 'todo:99'); assert.equal(s.getTaskBySourceRef('prj_old', 'todo:99')?.id, 'tsk_old');