diff --git a/src/daemon/ingest.ts b/src/daemon/ingest.ts index 2829b36..5fa3bad 100644 --- a/src/daemon/ingest.ts +++ b/src/daemon/ingest.ts @@ -26,7 +26,7 @@ export function ingestRun(store: Store, log: IngestLogger, runId: string): void const recs = readOutboxSince(runId, run.lastSeq); for (const rec of recs) { try { - applyRecord(store, log, taskId, runId, rec); + applyRecord(store, log, taskId, runId, run.kind, rec); } catch (e) { // 单条映射失败不阻断后续 run 的摄取;但本条不推进游标,下轮重试。 log.error(`ingest run=${runId} seq=${rec.seq} 失败:${(e as Error).message}`); @@ -36,8 +36,8 @@ export function ingestRun(store: Store, log: IngestLogger, runId: string): void } } -/** 把一条 OutboxRecord 映射为 DB 写。 */ -function applyRecord(store: Store, log: IngestLogger, taskId: string, runId: string, rec: OutboxRecord): void { +/** 把一条 OutboxRecord 映射为 DB 写。runKind 用于区分 failed 的收尾策略(executor vs planner)。 */ +function applyRecord(store: Store, log: IngestLogger, taskId: string, runId: string, runKind: string, rec: OutboxRecord): void { switch (rec.type) { case 'started': // worker 启动自报;pid 已由 daemon spawn 时 setWorkerPid 写过,这里仅记日志。 @@ -50,8 +50,8 @@ function applyRecord(store: Store, log: IngestLogger, taskId: string, runId: str return; case 'failed': { - // 先把 executor run 收尾为 failed(带转录/会话),再走失败/重试策略。 - // failTaskAttempt 见 run 已 ended(非 started)不会重复收尾,只做重试决策。 + // 先把 run 收尾为 failed(带转录/会话),再走失败/重试策略(按 kind 分流)。 + // failTaskAttempt/failPlanAttempt 见 run 已 ended(非 started)不会重复收尾,只做重试决策。 try { store.finishRun(runId, 'failed', { error: rec.error, @@ -61,8 +61,44 @@ function applyRecord(store: Store, log: IngestLogger, taskId: string, runId: str } catch (e) { log.error(`task=${taskId} run=${runId} 收尾 failed 出错(继续重试决策):${(e as Error).message}`); } - store.failTaskAttempt(taskId, runId, rec.error); - log.info(`task=${taskId} run=${runId} 执行失败:${rec.error}`); + if (runKind === 'planner') store.failPlanAttempt(taskId, runId, rec.error); // planner:退避留 analyzing/speccing + else store.failTaskAttempt(taskId, runId, rec.error); // executor:failed→重试/needs_attention + log.info(`task=${taskId} run=${runId}(${runKind}) 失败:${rec.error}`); + return; + } + + case 'spec-result': { + // planner-spec 成功:写方案 → 转 spec_review(待你审)→ 收尾 planner run。 + store.setSpec(taskId, rec.spec); + store.transition(taskId, 'spec_review', { by: 'ingest', runId }); + store.finishRun(runId, 'succeeded', { + transcriptRef: rec.transcriptRef ?? undefined, + claudeSessionId: rec.sessionId ?? undefined, + }); + log.info(`task=${taskId} run=${runId} 方案完成 → spec_review`); + return; + } + + case 'decompose-result': { + // planner-decompose 成功:写分析 + 建子任务 + 转 plan_review(待你审)→ 收尾 planner run。 + const task = store.getTask(taskId); + if (!task) { log.error(`decompose: 任务不存在 ${taskId}`); return; } + store.setPlan(taskId, rec.plan || '(无分析说明)'); + let created = 0; + for (const sub of rec.subtasks) { + try { + store.createTask({ projectId: task.projectId, parentId: taskId, title: sub.title, complexity: sub.complexity }); + created++; + } catch (e) { + log.error(`decompose: 建子任务「${sub.title}」失败:${(e as Error).message}`); + } + } + store.transition(taskId, 'plan_review', { by: 'ingest', runId, subtasks: created }); + store.finishRun(runId, 'succeeded', { + transcriptRef: rec.transcriptRef ?? undefined, + claudeSessionId: rec.sessionId ?? undefined, + }); + log.info(`task=${taskId} run=${runId} 拆解完成 → plan_review(${created} 个子任务)`); return; } @@ -103,12 +139,11 @@ function applyRecord(store: Store, log: IngestLogger, taskId: string, runId: str } /** - * 摄取所有在途 run:对每个 executing 任务的最近一条 executor run 调 ingestRun。 + * 摄取所有在途 run(executor + planner):遍历所有 started run 调 ingestRun。 * daemon 每轮 tick 调一次,把 worker 期间累积的 outbox 落库。 */ export function ingestAll(store: Store, log: IngestLogger): void { - for (const { run } of store.executingWithLatestExecutorRun()) { - if (!run) continue; + for (const { run } of store.liveRunsWithTask()) { ingestRun(store, log, run.id); } } diff --git a/src/daemon/orchestrator.ts b/src/daemon/orchestrator.ts index 866b8b6..c11ea2a 100644 --- a/src/daemon/orchestrator.ts +++ b/src/daemon/orchestrator.ts @@ -117,17 +117,21 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: }; /** - * 本项目可领取的任务:queued(重试/孤儿)+ ready 叶子且 deps 全 done;auto-easy 只挑 easy。 - * 退避读持久化的 task.nextEligibleAt(早于它不领)。按调度分降序返回(见 model/scoring.ts)。 + * 本项目可领取的任务(三类"可执行"状态): + * ready/queued → executor(执行改动);analyzing → planner 拆解 Hard;speccing → planner 写方案 Medium。 + * 排除:已在途(inflight=有 started run 的任务)、非叶子容器、退避冷却中、deps 未全 done; + * auto-easy 只挑 easy(故不做规划——analyzing/speccing 都是 hard/medium 被排除)。按调度分降序返回。 */ - function claimable(project: Project): Array<{ task: Task; score: number }> { + function claimable(project: Project, inflight: ReadonlySet): 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)); const easyOnly = project.autonomy === 'auto-easy'; + const CLAIMABLE = new Set(['ready', 'queued', 'analyzing', 'speccing']); const candidates = tasks.filter((t) => { - if (t.status !== 'ready' && t.status !== 'queued') return false; + if (!CLAIMABLE.has(t.status)) return false; + if (inflight.has(t.id)) return false; // 已有在途 run(executor/planner) if (parents.has(t.id)) return false; // 非叶子(容器)跳过 if (easyOnly && t.complexity !== 'easy') return false; if (t.nextEligibleAt && nowMs < Date.parse(t.nextEligibleAt)) return false; // 退避冷却中 @@ -136,51 +140,66 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: return rankByScore(candidates, tasks); } - /** 领取一个任务:建分支/worktree 路径 → executing → startRun → 写 job.json → spawn worker → 记 pid。 */ + /** + * 领取一个任务并起 worker: + * - ready/queued → executor:ready→queued→executing,startRun(executor, worktree),job.runKind=executor。 + * - analyzing → planner-decompose;speccing → planner-spec:【不转状态】(留 analyzing/speccing 作"规划中"), + * startRun(planner),planner 只读跑在 repo(worktreeDir 仅记录用)。失败按 kind 走 failTaskAttempt/failPlanAttempt。 + */ function claimOne(project: Project, task: Task, score: number): void { + const isPlanner = task.status === 'analyzing' || task.status === 'speccing'; + const runKind = task.status === 'analyzing' ? 'planner-decompose' + : task.status === 'speccing' ? 'planner-spec' : 'executor'; + const dbKind = isPlanner ? 'planner' : 'executor'; + const role = isPlanner ? 'planner' : 'executor'; let runId: string | null = null; try { - if (task.status === 'ready') store.transition(task.id, 'queued', { by: 'orchestrator', score }); - const branch = branchFor(task.id); const dir = worktreeDirFor(project.repoPath, task.id); - store.transition(task.id, 'executing', { by: 'orchestrator' }); - const run = store.startRun(task.id, 'executor', { worktree: dir, branch }); + if (!isPlanner) { + if (task.status === 'ready') store.transition(task.id, 'queued', { by: 'orchestrator', score }); + store.transition(task.id, 'executing', { by: 'orchestrator' }); + } + // planner run 只读跑在主仓,worktree 记 repoPath;executor 记隔离 worktree 路径 + const run = store.startRun(task.id, dbKind, { worktree: isPlanner ? project.repoPath : dir, branch }); runId = run.id; d.writeJobSpec({ runId: run.id, - task: { ...task, status: 'executing' }, - project, - worktreeDir: dir, - branch, + task: { ...task, status: isPlanner ? task.status : 'executing' }, + project, worktreeDir: dir, branch, runKind, }); - log.info(`领取任务 ${task.id}「${task.title}」score=${score} run=${run.id} model=${pickModel(task, project, 'executor')}`); + log.info(`领取 ${runKind} ${task.id}「${task.title}」score=${score} run=${run.id} model=${pickModel(task, project, role)}`); const pid = d.spawnWorker(run.id); store.setWorkerPid(run.id, pid); - log.info(`任务 ${task.id} 已起 worker pid=${pid} worktree=${dir}`); + log.info(`任务 ${task.id} 已起 worker pid=${pid}(${runKind})`); } catch (e) { const msg = (e as Error).message; log.error(`任务 ${task.id} 领取/spawn 失败:${msg}`); try { - store.failTaskAttempt(task.id, runId, msg); + if (isPlanner) store.failPlanAttempt(task.id, runId, msg); + else store.failTaskAttempt(task.id, runId, msg); } catch (e2) { log.error(`任务 ${task.id} 失败收尾出错:${(e2 as Error).message}`); } } } - /** 领取轮:对每个 active 且 autonomy≠manual 的项目,在 countExecuting= p.concurrency) continue; - for (const { task, score } of claimable(p)) { + for (const { task, score } of claimable(p, inflight)) { if (active >= p.concurrency) break; claimOne(p, task, score); active++; @@ -200,19 +219,21 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: } } - /** 回收:对每个 executing 任务判活,死 worker → failTaskAttempt(收尾 + 重试/needs_attention)。 */ + /** 回收:对每个在途 run(executor + planner)判活,死 worker → 按 kind 收尾(executor=failTaskAttempt / planner=failPlanAttempt)。 */ function reap(): void { try { - for (const { task, run } of store.executingWithLatestExecutorRun()) { - const alive = run !== null && d.isWorkerAlive({ + for (const { task, run } of store.liveRunsWithTask()) { + const alive = d.isWorkerAlive({ pid: run.workerPid, heartbeatAgeMs: heartbeatAgeMs(run.id, d.nowMs()), startedAgeMs: d.nowMs() - Date.parse(run.startedAt), }); if (alive) continue; - log.error(`任务 ${task.id} worker 异常退出(run=${run?.id ?? '无'})→ 回收`); + log.error(`任务 ${task.id} worker 异常退出(run=${run.id} ${run.kind})→ 回收`); try { - store.failTaskAttempt(task.id, run?.id ?? null, 'worker 异常退出'); + if (run.kind === 'executor') store.failTaskAttempt(task.id, run.id, 'worker 异常退出'); + else if (run.kind === 'planner') store.failPlanAttempt(task.id, run.id, 'planner worker 异常退出'); + else store.finishRun(run.id, 'failed', { error: 'worker 异常退出' }); } catch (e) { log.error(`任务 ${task.id} 回收收尾出错:${(e as Error).message}`); } diff --git a/src/executor/models.ts b/src/executor/models.ts index 8c30730..0f20c3d 100644 --- a/src/executor/models.ts +++ b/src/executor/models.ts @@ -1,8 +1,8 @@ import type { Project, Task } from '../model/types.js'; import type { Complexity } from '../model/complexity.js'; -/** 角色:executor(执行改动)/ reviewer(执行后复审) */ -export type ModelRole = 'executor' | 'reviewer'; +/** 角色:executor(执行改动)/ reviewer(执行后复审)/ planner(拆解 Hard / 写方案 Medium) */ +export type ModelRole = 'executor' | 'reviewer' | 'planner'; /** 模型不可用时的回退链(按序取第一个 ≠ 失败模型的,同一 run 内只重试一次) */ export const MODEL_FALLBACK_CHAIN = ['claude-opus-4-8', 'claude-sonnet-4-6'] as const; @@ -19,6 +19,12 @@ const MODEL_TABLE: Record { + // planner-* 走只读的规划管线(不建 worktree、不改文件、不复审) + if (job.runKind === 'planner-spec' || job.runKind === 'planner-decompose') { + return runPlannerPipeline(job, deps, emit); + } // 1. 建 worktree(其 dir/branch 应与 job.worktreeDir/branch 一致;以 deps 真建的为准) const wt = await deps.createWorktree(job.project.repoPath, job.task.id, job.project.defaultBranch); @@ -118,3 +124,54 @@ export async function runPipeline(job: JobSpec, deps: PipelineDeps = realDeps, e }); emit({ type: 'done' }); } + +/** + * planner 管线(只读):跑一次 CC 写方案 / 拆解,emit spec-result / decompose-result。 + * - planner-spec:CC 输出即方案正文 → emit spec-result{spec}。 + * - planner-decompose:解析 CC 输出末尾的 fenced JSON{plan,subtasks} → emit decompose-result;解析失败 → failed。 + * CC 失败 / 无输出 → failed。全程总 emit 一条终态后再 emit done。 + */ +async function runPlannerPipeline(job: JobSpec, deps: PipelineDeps, emit: Emit): Promise { + const kind = job.runKind === 'planner-spec' ? 'spec' : 'decompose'; + emit({ type: 'phase', phase: kind === 'spec' ? 'speccing' : 'decomposing' }); + const r = await deps.runPlanner(job.task, job.project, kind, job.runId); + if (!r.ok || !r.finalText) { + emit({ type: 'failed', error: r.error ?? 'planner 无输出', transcriptRef: r.transcriptRef, sessionId: r.sessionId }); + emit({ type: 'done' }); + return; + } + if (kind === 'spec') { + emit({ type: 'spec-result', spec: r.finalText, transcriptRef: r.transcriptRef, sessionId: r.sessionId }); + emit({ type: 'done' }); + return; + } + const parsed = parseDecompose(r.finalText); + if (!parsed) { + emit({ type: 'failed', error: 'planner 输出未含合法的子任务 JSON 块', transcriptRef: r.transcriptRef, sessionId: r.sessionId }); + emit({ type: 'done' }); + return; + } + emit({ type: 'decompose-result', plan: parsed.plan, subtasks: parsed.subtasks, transcriptRef: r.transcriptRef, sessionId: r.sessionId }); + emit({ type: 'done' }); +} + +/** 从 CC 文本里抽取拆解结果:优先取最后一个 ```json 块,回退到任意 ``` 块,再回退到整段当 JSON。非法 → null。 */ +export function parseDecompose(text: string): DecomposeResult | null { + const blocks = [...text.matchAll(/```(?:json)?\s*([\s\S]*?)```/gi)].map((m) => m[1].trim()); + const candidates = blocks.length ? [blocks[blocks.length - 1]] : [text.trim()]; + for (const raw of candidates) { + try { + const o = JSON.parse(raw) as { plan?: unknown; subtasks?: unknown }; + if (!Array.isArray(o.subtasks)) continue; + const subtasks = o.subtasks + .filter((s): s is { title: string; complexity: 'easy' | 'medium' | 'hard' } => + !!s && typeof (s as { title?: unknown }).title === 'string' && + ['easy', 'medium', 'hard'].includes((s as { complexity?: unknown }).complexity as string)) + .map((s) => ({ title: String(s.title).trim(), complexity: s.complexity })) + .filter((s) => s.title.length > 0); + if (subtasks.length === 0) continue; + return { plan: typeof o.plan === 'string' ? o.plan : '', subtasks }; + } catch { /* 试下一个候选 */ } + } + return null; +} diff --git a/src/executor/protocol.ts b/src/executor/protocol.ts index fb836a7..a6c914b 100644 --- a/src/executor/protocol.ts +++ b/src/executor/protocol.ts @@ -54,6 +54,9 @@ export function isPlainRunId(runId: string): boolean { * worker 的唯一输入。daemon 在 spawn 前写 runs//job.json,worker 读它即可执行, * 【无需访问 DB】。携带完整 Task / Project(均 JSON 可序列化),worker 自行 pickModel/buildPrompt。 */ +/** run 类型:executor(在 worktree 执行改动)/ planner-spec(写 Medium 方案)/ planner-decompose(拆解 Hard) */ +export type RunKind = 'executor' | 'planner-spec' | 'planner-decompose'; + export interface JobSpec { runId: string; task: Task; @@ -61,6 +64,14 @@ export interface JobSpec { /** 确定性 worktree 路径与分支(daemon 预填;worker 也能由 worktree.ts 算出,传入避免重复) */ worktreeDir: string; branch: string; + /** run 类型,缺省 executor(向后兼容旧 job)。planner-* 在 repo 内只读跑 CC、不建 worktree。 */ + runKind?: RunKind; +} + +/** planner-decompose 的拆解产出:plan 文本 + 子任务列表 */ +export interface DecomposeResult { + plan: string; + subtasks: Array<{ title: string; complexity: 'easy' | 'medium' | 'hard' }>; } export function writeJobSpec(job: JobSpec): void { @@ -86,6 +97,8 @@ export interface ReviewReport { * - phase :进度阶段(executing|verifying|reviewing…),daemon 落成 event 推看板(可选) * - failed :本次尝试终态失败(executor 报错或 verify 失败)。daemon 据此 finishRun(failed)+重试决策 * - result :成功终态。daemon 据此建 reviewer/security run + setResult(四字段) + 转 exec_review + * - spec-result :planner-spec 成功终态。daemon 据此 setSpec + 转 spec_review + * - decompose-result:planner-decompose 成功终态。daemon 据此 setPlan + 建子任务 + 转 plan_review * - done :worker 即将退出(成功或失败都发,daemon 据此停止 tail) */ export type OutboxRecord = @@ -99,6 +112,12 @@ export type OutboxRecord = code: ReviewReport; security: ReviewReport; } + | { seq: number; at: string; type: 'spec-result'; spec: string; transcriptRef: string | null; sessionId: string | null } + | { + seq: number; at: string; type: 'decompose-result'; + plan: string; subtasks: Array<{ title: string; complexity: 'easy' | 'medium' | 'hard' }>; + transcriptRef: string | null; sessionId: string | null; + } | { seq: number; at: string; type: 'done' }; /** OutboxRecord 去掉 seq/at(由 appendOutbox 填) */ @@ -107,6 +126,8 @@ export type OutboxPayload = | Omit, 'seq' | 'at'> | Omit, 'seq' | 'at'> | Omit, 'seq' | 'at'> + | Omit, 'seq' | 'at'> + | Omit, 'seq' | 'at'> | Omit, 'seq' | 'at'>; /** 追加一条 outbox 记录(worker 侧调用)。seq = 现有行数+1(worker 单线程,无并发写)。返回写入的完整记录。 */ diff --git a/src/executor/runner.ts b/src/executor/runner.ts index ec0bdb0..dc0db9c 100644 --- a/src/executor/runner.ts +++ b/src/executor/runner.ts @@ -94,3 +94,69 @@ export async function runTask(task: Task, project: Project, worktree: WorktreeIn } return { ok: true, transcriptRef: cc.transcriptRef, sessionId: cc.sessionId, finalText: cc.finalText || null, modelUsed: cc.modelUsed }; } + +// ───────────────────────── planner(拆解 Hard / 写方案 Medium)───────────────────────── + +export type PlanKind = 'spec' | 'decompose'; + +export interface PlannerResult { + ok: boolean; + transcriptRef: string | null; + sessionId: string | null; + finalText: string | null; // CC 的最终文本:spec=方案正文;decompose=分析 + 末尾 fenced JSON + error?: string; +} + +/** 测试注入点(pipeline 的 planner 分支用) */ +export type PlannerFn = (task: Task, project: Project, kind: PlanKind, runId: string) => Promise; + +/** planner 提示词:只读代码库,spec=写改动方案;decompose=拆子任务并在末尾输出 fenced JSON。 */ +export function buildPlannerPrompt(task: Task, kind: PlanKind): string { + const head = [`# 任务:${task.title}`, `任务 ID:${task.id}`]; + if (kind === 'spec') { + if (task.spec) head.push('', '## 现有方案草稿(可改进/替换)', task.spec); + return [ + ...head, '', + '## 你的角色:方案作者(只读,不改任何文件)', + '只读这个代码库,为上述任务写一份「具体改动方案」。不要编辑文件、不要执行有副作用的命令。', + '## 输出(markdown 正文)', + '## 改动(要改哪些文件/模块、怎么改)', + '## 为什么(这么做的理由、取舍)', + '## 验收(怎么算完成、怎么验证)', + '这份方案将提交给人审;通过后才会有执行 agent 按它改代码。直接输出方案正文即可。', + ].join('\n'); + } + if (task.plan) head.push('', '## 现有分析草稿(可改进/替换)', task.plan); + return [ + ...head, '', + '## 你的角色:任务拆解者(只读,不改任何文件)', + '只读这个代码库,分析上述(复杂)任务,拆成 2–6 个更小、可独立交付的子任务。不要编辑文件、不要执行有副作用的命令。', + '每个子任务标注复杂度:easy(单文件机械改动/无设计)、medium(需方案、跨几处)、hard(仍需进一步拆解)。', + '## 输出(重要)', + '先写一段分析与拆解理由;然后在【最后】输出唯一一个 ```json 代码块,严格格式:', + '```json', + '{"plan":"一句话分析与拆解理由","subtasks":[{"title":"子任务标题","complexity":"easy"}]}', + '```', + 'subtasks 至少 1 个;complexity 只能是 easy|medium|hard;除这个 JSON 块外不要再写其它 ``` 代码块。', + ].join('\n'); +} + +/** + * 起 headless CC 跑一次 planner:**只读**跑在 project.repoPath(不建 worktree、不改文件)。 + * 模型用 planner 角色(默认 opus)。不抛错,失败折叠进 { ok:false, error }。 + */ +export async function runPlanner(task: Task, project: Project, kind: PlanKind, runId: string): Promise { + const model = pickModel(task, project, 'planner'); + const timeoutMs = project.timeoutMs ?? DEFAULT_TIMEOUT_MS; + const cc = await runClaude({ + prompt: buildPlannerPrompt(task, kind), + cwd: project.repoPath, // 只读跑在主仓(planner 不改文件,无需 worktree) + model, + runId, + maxTurns: 60, + timeoutMs, + permissionMode: 'acceptEdits', // planner 只读,不会触发编辑;保持工具流不被 prompt 卡住 + allowedTools: ['Read', 'Glob', 'Grep'], // 纯只读:无 Edit/Write/Bash,零副作用 + }); + return { ok: cc.ok, transcriptRef: cc.transcriptRef, sessionId: cc.sessionId, finalText: cc.finalText || null, error: cc.error }; +} diff --git a/src/executor/worker.ts b/src/executor/worker.ts index 0d9129d..53159ad 100644 --- a/src/executor/worker.ts +++ b/src/executor/worker.ts @@ -32,13 +32,14 @@ async function main(): Promise { const emit = (p: OutboxPayload): void => { appendOutbox(runId, p); }; const job = readJobSpec(runId); - // 起手汇报:pid / worktree / branch / 实际执行模型 + // 起手汇报:pid / worktree / branch / 实际执行模型(planner-* 用 planner 角色模型) + const role = job.runKind && job.runKind !== 'executor' ? 'planner' : 'executor'; emit({ type: 'started', pid: process.pid, worktree: job.worktreeDir, branch: job.branch, - model: pickModel(job.task, job.project, 'executor'), + model: pickModel(job.task, job.project, role), }); // 心跳:立即一次 + 周期刷(结束时清掉) diff --git a/src/model/status.ts b/src/model/status.ts index 68411cd..fde0cbd 100644 --- a/src/model/status.ts +++ b/src/model/status.ts @@ -57,10 +57,10 @@ export const TERMINAL_STATUS: ReadonlySet = new Set(['do /** 合法状态流转表:from → 允许的 to[] */ export const TRANSITIONS: Record = { init: ['analyzing', 'speccing', 'ready', 'cancelled'], // 按复杂度分流 - analyzing: ['plan_review', 'cancelled', 'paused'], + analyzing: ['plan_review', 'needs_attention', 'cancelled', 'paused'], // planner 拆解 / 失败超限升级 plan_review: ['decomposed', 'analyzing', 'cancelled'], // accept / reject(+意见) decomposed: ['done', 'analyzing', 'cancelled', 'paused'], // 子全 done / 重拆 - speccing: ['spec_review', 'cancelled', 'paused'], + speccing: ['spec_review', 'needs_attention', 'cancelled', 'paused'], // planner 写方案 / 失败超限升级 spec_review: ['ready', 'speccing', 'cancelled'], // accept / reject(+意见) ready: ['blocked', 'queued', 'speccing', 'analyzing', 'cancelled', 'paused'], blocked: ['ready', 'cancelled', 'paused'], diff --git a/src/store/store.ts b/src/store/store.ts index 26cac49..ff4dff8 100644 --- a/src/store/store.ts +++ b/src/store/store.ts @@ -660,11 +660,67 @@ export class Store { return this.getTask(taskId)!; } + /** + * planner(拆解 Hard / 写方案 Medium)失败的落点:收尾 planner run + 退避,任务【留在 analyzing/speccing】 + * 等退避到期重新被领取;累计 planner 失败超 project.maxRetries → 升级 needs_attention(清退避)。 + * 与 failTaskAttempt 的区别:planner 失败不把任务转 failed/queued(那是 executor 语义),只退避或升级。 + */ + failPlanAttempt(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}`); + + 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, 'planner'); + this.finishRun(r.id, 'failed', { error }); + } + + const allFailed = (this.db.prepare( + `SELECT COUNT(*) AS n FROM runs WHERE task_id = ? AND kind = 'planner' AND status = 'failed'`, + ).get(taskId) as { n: number }).n; + const priorNetFailed = Math.max(0, allFailed - 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()); + // 任务保持 analyzing/speccing(不转状态)——退避到期后重新可领,再起 planner run + this.emit(row.project_id, taskId, 'task.updated', { field: 'plan-retry', attempt, backoffMs, error }); + } else { + this.setNextEligibleAt(taskId, null); + this.transition(taskId, 'needs_attention', { by: 'failPlanAttempt', allFailed }); + } + 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); } + /** 项目内"有 started run 的任务" id 集合(多进程并发闸 + claimable 排除在途;executor 与 planner 通用)。 */ + inflightTaskIds(projectId: string): Set { + const rows = this.db.prepare( + `SELECT DISTINCT r.task_id AS tid FROM runs r JOIN tasks t ON t.id = r.task_id + WHERE r.status = 'started' AND t.project_id = ?`, + ).all(projectId) as Array<{ tid: string }>; + return new Set(rows.map((r) => r.tid)); + } + + /** 所有 started run + 其任务(reap / ingest / reconcile 通用,覆盖 executor + planner + 残留复审 run)。 */ + liveRunsWithTask(): Array<{ task: Task; run: Run }> { + const runs = this.db.prepare(`SELECT * FROM runs WHERE status = 'started' ORDER BY started_at`).all() as RunRow[]; + const out: Array<{ task: Task; run: Run }> = []; + for (const r of runs) { + const t = this.getTaskRow(r.task_id); + if (t) out.push({ task: rowToTask(t), run: rowToRun(r) }); + } + return out; + } + /** 项目内 executing 任务数(多进程执行的并发闸:每个 executing 对应一个活 worker)。 */ countExecuting(projectId: string): number { return (this.db.prepare( @@ -691,9 +747,11 @@ export class Store { */ 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 已退出'); + for (const { task, run } of this.liveRunsWithTask()) { + if (isAlive(run)) { readopted++; continue; } // worker 仍活 → re-adopt,daemon 续 ingest + if (run.kind === 'executor') this.failTaskAttempt(task.id, run.id, 'daemon 重启时发现 worker 已退出'); + else if (run.kind === 'planner') this.failPlanAttempt(task.id, run.id, 'daemon 重启时发现 planner worker 已退出'); + else this.finishRun(run.id, 'failed', { error: 'daemon 重启中断' }); // 残留复审 run:仅收尾 reclaimed++; } return { readopted, reclaimed }; diff --git a/test/ingest.test.ts b/test/ingest.test.ts index 043851c..265881d 100644 --- a/test/ingest.test.ts +++ b/test/ingest.test.ts @@ -203,3 +203,68 @@ test('ingest 续读:先 ingest 半截(started/phase),再追加 result, store.close(); }); }); + +// ───────────────────────── planner run 摄取 ───────────────────────── + +/** 真 Store + 一个 medium(speccing) 或 hard(analyzing) 任务 + 一条 planner run(模拟已领取规划)。 */ +function setupPlanning(complexity: 'medium' | 'hard'): { store: Store; taskId: string; runId: string } { + const store = new Store(':memory:'); + const p = store.createProject({ name: 'plan', repoPath: '/tmp/plan-' + Math.random(), autonomy: 'auto-approved' }); + const t = store.createTask({ projectId: p.id, title: '大任务', complexity }); + // 建任务即落 speccing(medium)/analyzing(hard),不转状态,直接起 planner run + const run = store.startRun(t.id, 'planner', { worktree: p.repoPath, branch: 'n/a' }); + store.setWorkerPid(run.id, 4243); + return { store, taskId: t.id, runId: run.id }; +} + +test('ingest planner-spec:setSpec + 转 spec_review + 收尾 planner run', () => { + withTmpDataDir(() => { + const { store, taskId, runId } = setupPlanning('medium'); + assert.equal(store.getTask(taskId)!.status, 'speccing'); + appendOutbox(runId, { type: 'started', pid: 4243, worktree: '/r', branch: 'n/a', model: 'opus' }); + appendOutbox(runId, { type: 'spec-result', spec: '## 改动\n改 a.ts', transcriptRef: '/p.jsonl', sessionId: 's' }); + appendOutbox(runId, { type: 'done' }); + ingestRun(store, noopLog, runId); + const t = store.getTask(taskId)!; + assert.equal(t.status, 'spec_review'); + assert.equal(t.spec, '## 改动\n改 a.ts'); + assert.equal(store.getRun(runId)!.status, 'succeeded'); + store.close(); + }); +}); + +test('ingest planner-decompose:setPlan + 建子任务 + 转 plan_review', () => { + withTmpDataDir(() => { + const { store, taskId, runId } = setupPlanning('hard'); + assert.equal(store.getTask(taskId)!.status, 'analyzing'); + appendOutbox(runId, { + type: 'decompose-result', plan: '拆成两步', + subtasks: [{ title: '建骨架', complexity: 'easy' }, { title: '实现', complexity: 'medium' }], + transcriptRef: null, sessionId: null, + }); + appendOutbox(runId, { type: 'done' }); + ingestRun(store, noopLog, runId); + const t = store.getTask(taskId)!; + assert.equal(t.status, 'plan_review'); + assert.equal(t.plan, '拆成两步'); + const kids = store.listTasks(t.projectId).filter((x) => x.parentId === taskId); + assert.equal(kids.length, 2, '建了 2 个子任务'); + assert.deepEqual(kids.map((k) => k.complexity).sort(), ['easy', 'medium']); + assert.equal(store.getRun(runId)!.status, 'succeeded'); + store.close(); + }); +}); + +test('ingest planner failed:走 failPlanAttempt(退避,任务留 speccing,不进 failed)', () => { + withTmpDataDir(() => { + const { store, taskId, runId } = setupPlanning('medium'); + appendOutbox(runId, { type: 'failed', error: 'planner 崩了', transcriptRef: null, sessionId: null }); + appendOutbox(runId, { type: 'done' }); + ingestRun(store, noopLog, runId); + const t = store.getTask(taskId)!; + assert.equal(t.status, 'speccing', '默认 maxRetries=2,第一次失败留 speccing 退避重试'); + assert.ok(t.nextEligibleAt, '设了退避 nextEligibleAt'); + assert.equal(store.getRun(runId)!.status, 'failed'); + store.close(); + }); +}); diff --git a/test/orchestrator.test.ts b/test/orchestrator.test.ts index efc04ce..4530090 100644 --- a/test/orchestrator.test.ts +++ b/test/orchestrator.test.ts @@ -375,3 +375,68 @@ test(`reaper 反复回收:净失败 ${DEFAULT_MAX_RETRIES} 次后 → needs_at assert.equal(failed.length, DEFAULT_MAX_RETRIES + 1); store.close(); }); + +// ───────────────────────── planner:analyzing / speccing 也被领取 ───────────────────────── + +test('claimable 纳入 speccing:medium 任务派 planner-spec(不转状态、startRun=planner、job.runKind=planner-spec)', () => { + const { store, projectId } = setup('auto-approved'); + const t = store.createTask({ projectId, title: 'feat', complexity: 'medium' }); // → speccing + assert.equal(store.getTask(t.id)!.status, 'speccing'); + const state = freshState(); + const { deps } = mockDeps(state); + const orch = createOrchestrator(store, noopLog, deps); + + orch.claimTick(); + assert.equal(state.spawned.length, 1, '派了一个 planner worker'); + assert.equal(store.getTask(t.id)!.status, 'speccing', 'planner 不转任务状态(留 speccing 作规划中)'); + const runs = store.listRuns(t.id); + assert.equal(runs.length, 1); + assert.equal(runs[0].kind, 'planner', 'run kind=planner'); + assert.equal(runs[0].status, 'started'); + assert.equal(state.jobs[0].runKind, 'planner-spec'); + + // 已在途(有 started planner run)→ 下一轮不重领 + orch.claimTick(); + assert.equal(state.spawned.length, 1, 'inflight 任务不被重复领取'); + store.close(); +}); + +test('claimable 纳入 analyzing:hard 任务派 planner-decompose', () => { + const { store, projectId } = setup('auto-approved'); + const t = store.createTask({ projectId, title: 'epic', complexity: 'hard' }); // → analyzing + assert.equal(store.getTask(t.id)!.status, 'analyzing'); + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).claimTick(); + assert.equal(state.spawned.length, 1); + assert.equal(store.getTask(t.id)!.status, 'analyzing', 'planner 不转状态'); + assert.equal(store.listRuns(t.id)[0].kind, 'planner'); + assert.equal(state.jobs[0].runKind, 'planner-decompose'); + store.close(); +}); + +test('auto-easy 不做规划:medium(speccing) 不被领取', () => { + const { store, projectId } = setup('auto-easy'); + const t = store.createTask({ projectId, title: 'feat', complexity: 'medium' }); // → speccing + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).claimTick(); + assert.equal(state.spawned.length, 0, 'auto-easy 只做 easy 执行,不规划 medium/hard'); + assert.equal(store.getTask(t.id)!.status, 'speccing'); + store.close(); +}); + +test('并发闸用 inflight:1 个 executing + 1 个 planner 占满 concurrency=2', () => { + const { store, projectId } = setup('auto-approved', 2); + const e = store.createTask({ projectId, title: 'easy', complexity: 'easy' }); // → ready + const m = store.createTask({ projectId, title: 'med', complexity: 'medium' }); // → speccing + store.createTask({ projectId, title: 'easy2', complexity: 'easy' }); // → ready(第三个) + const state = freshState(); + const { deps } = mockDeps(state); + createOrchestrator(store, noopLog, deps).claimTick(); + // concurrency=2:一个 executor(easy) + 一个 planner(medium) 占满,第三个不领 + assert.equal(state.spawned.length, 2); + assert.equal(store.getTask(e.id)!.status, 'executing'); + assert.equal(store.getTask(m.id)!.status, 'speccing'); + store.close(); +}); diff --git a/test/pipeline.test.ts b/test/pipeline.test.ts index 304bd84..6aaeede 100644 --- a/test/pipeline.test.ts +++ b/test/pipeline.test.ts @@ -1,6 +1,6 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { runPipeline, type PipelineDeps } from '../src/executor/pipeline.js'; +import { runPipeline, parseDecompose, type PipelineDeps } from '../src/executor/pipeline.js'; import type { JobSpec, OutboxPayload } from '../src/executor/protocol.js'; import type { Project, Task } from '../src/model/types.js'; import type { RunnerResult } from '../src/executor/runner.js'; @@ -55,6 +55,7 @@ function mockDeps(overrides: Partial = {}): PipelineDeps { runTask: async () => okRun, reviewCode: async () => okReview, reviewSecurity: async () => okSecurity, + runPlanner: async () => ({ ok: true, transcriptRef: '/tmp/plan.jsonl', sessionId: 'sess-plan', finalText: '方案正文' }), ...overrides, }; } @@ -181,3 +182,62 @@ test('安全审计抛错不挡:security.verdict=null、summary 含「自动复 assert.equal(result.code.verdict, 'approve'); assert.equal(emitted.at(-1)?.type, 'done'); }); + +// ───────────────────────── planner 分支 ───────────────────────── + +const medJob = (over: Partial = {}): JobSpec => ({ ...fakeJob, runKind: 'planner-spec', task: { ...fakeTask, complexity: 'medium', status: 'speccing' }, ...over }); +const hardJob = (over: Partial = {}): JobSpec => ({ ...fakeJob, runKind: 'planner-decompose', task: { ...fakeTask, complexity: 'hard', status: 'analyzing' }, ...over }); + +test('planner-spec:emit spec-result(方案=CC 文本)+ done,不复审/不建 worktree', async () => { + const { emitted, emit } = collector(); + let worktreeCalled = false; + await runPipeline(medJob(), mockDeps({ + createWorktree: async () => { worktreeCalled = true; return { dir: 'x', branch: 'y' }; }, + runPlanner: async () => ({ ok: true, transcriptRef: '/tmp/p.jsonl', sessionId: 's', finalText: '## 改动\n改 a.ts' }), + }), emit); + assert.equal(worktreeCalled, false, 'planner 不建 worktree'); + const sr = find(emitted, 'spec-result'); + assert.ok(sr, '应 emit spec-result'); + assert.equal(sr.spec, '## 改动\n改 a.ts'); + assert.equal(find(emitted, 'result'), undefined); + assert.equal(find(emitted, 'failed'), undefined); + assert.equal(emitted.at(-1)?.type, 'done'); +}); + +test('planner-decompose:解析末尾 fenced JSON → emit decompose-result + done', async () => { + const { emitted, emit } = collector(); + const txt = '分析:拆成两步。\n```json\n{"plan":"先骨架再实现","subtasks":[{"title":"建骨架","complexity":"easy"},{"title":"实现逻辑","complexity":"medium"}]}\n```'; + await runPipeline(hardJob(), mockDeps({ runPlanner: async () => ({ ok: true, transcriptRef: null, sessionId: null, finalText: txt }) }), emit); + const dr = find(emitted, 'decompose-result'); + assert.ok(dr, '应 emit decompose-result'); + assert.equal(dr.plan, '先骨架再实现'); + assert.equal(dr.subtasks.length, 2); + assert.deepEqual(dr.subtasks.map((s) => s.complexity), ['easy', 'medium']); + assert.equal(emitted.at(-1)?.type, 'done'); +}); + +test('planner-decompose:输出无合法 JSON → emit failed', async () => { + const { emitted, emit } = collector(); + await runPipeline(hardJob(), mockDeps({ runPlanner: async () => ({ ok: true, transcriptRef: null, sessionId: null, finalText: '我忘了输出 JSON' }) }), emit); + assert.ok(find(emitted, 'failed'), '解析失败应 emit failed'); + assert.equal(find(emitted, 'decompose-result'), undefined); + assert.equal(emitted.at(-1)?.type, 'done'); +}); + +test('planner:CC 失败 → emit failed', async () => { + const { emitted, emit } = collector(); + await runPipeline(medJob(), mockDeps({ runPlanner: async () => ({ ok: false, transcriptRef: null, sessionId: null, finalText: null, error: 'CC 崩了' }) }), emit); + const f = find(emitted, 'failed'); + assert.ok(f); + assert.match(f.error, /CC 崩了/); + assert.equal(emitted.at(-1)?.type, 'done'); +}); + +test('parseDecompose:取最后一个 json 块;非法/缺失 → null', () => { + assert.equal(parseDecompose('no json here'), null); + assert.equal(parseDecompose('```json\n{"subtasks":[]}\n```'), null, '空 subtasks → null'); + const ok = parseDecompose('前文\n```json\n{"subtasks":[{"title":"a","complexity":"hard"},{"title":"","complexity":"easy"},{"title":"b","complexity":"x"}]}\n```'); + assert.ok(ok); + assert.equal(ok.subtasks.length, 1, '过滤空标题与非法复杂度'); + assert.equal(ok.subtasks[0].title, 'a'); +});