merge phase3-C: daemon 侧（监工 + ingest + 接线 + 测试）

2026-06-13 13:02:13 +08:00
parent b0ef60ab58 da7c93105c
commit adf88e82cd
5 changed files with 765 additions and 567 deletions
@@ -6,6 +6,7 @@ import { registerStatic } from '../api/static.js';
 import { loadConfig } from './config.js';
 import { startOrchestrator } from './orchestrator.js';
 import { startNotifier } from './notify.js';
+import { isWorkerAlive, heartbeatAgeMs } from '../executor/protocol.js';
 import { syncProject, todoJsonPath } from '../sync/todo-sync.js';

 /**
@@ -53,7 +54,15 @@ async function main(): Promise<void> {
  const cfg = loadConfig();
  const store = new Store(cfg.dbFile);
  const rec = store.reconcileDeps();   // 启动对账：ready↔blocked 按依赖纠正存量数据
-  const itr = store.reconcileInterrupted();   // 中断恢复：上次退出时在跑的任务重新入队
+  // 中断恢复（多进程执行）：用 worker_pid + 心跳真判活——worker 仍活则 re-adopt（daemon 续 ingest 其 outbox），
+  // 死则回收（failTaskAttempt 收尾 + 重试/needs_attention）。
+  const itr = store.reconcileInterrupted((run) =>
+    run !== null && isWorkerAlive({
+      pid: run.workerPid,
+      heartbeatAgeMs: heartbeatAgeMs(run.id),
+      startedAgeMs: Date.now() - Date.parse(run.startedAt),
+    }),
+  );
  const app = buildServer({ store, logger: true });

  registerStatic(app);          // Web 看板（web/ 静态文件）
@@ -0,0 +1,114 @@
+import { Store } from '../store/index.js';
+import { readOutboxSince, type OutboxRecord } from '../executor/protocol.js';
+
+/**
+ * Outbox → DB 摄取（daemon 是唯一 DB 写者）。
+ *
+ * worker 全程不碰 DB，只把进度/结果追加进 runs/<runId>/outbox.ndjson。daemon 在每轮 tick 里把
+ * 这些记录读出来，按 seq 升序逐条映射成 DB 写（finishRun / setResult / transition / failTaskAttempt …）。
+ *
+ * 幂等靠 run.lastSeq 游标：只处理 seq>lastSeq 的记录，每处理一条就 setLastSeq(seq)。崩溃重启续读时
+ * 已落库的 seq 不会重做——故 result/failed/done 这类终态记录即便文件还在，也只会被消费一次。
+ */
+export interface IngestLogger {
+  info(msg: string): void;
+  error(msg: string): void;
+}
+
+/**
+ * 摄取单个 run 的新 outbox 记录（seq>run.lastSeq），按序落库。
+ * run 不存在则忽略（已被删/异常）。每条处理后推进 lastSeq 游标。
+ */
+export function ingestRun(store: Store, log: IngestLogger, runId: string): void {
+  const run = store.getRun(runId);
+  if (!run) return;
+  const taskId = run.taskId;
+  const recs = readOutboxSince(runId, run.lastSeq);
+  for (const rec of recs) {
+    try {
+      applyRecord(store, log, taskId, runId, rec);
+    } catch (e) {
+      // 单条映射失败不阻断后续 run 的摄取；但本条不推进游标，下轮重试。
+      log.error(`ingest run=${runId} seq=${rec.seq} 失败：${(e as Error).message}`);
+      return;
+    }
+    store.setLastSeq(runId, rec.seq);
+  }
+}
+
+/** 把一条 OutboxRecord 映射为 DB 写。 */
+function applyRecord(store: Store, log: IngestLogger, taskId: string, runId: string, rec: OutboxRecord): void {
+  switch (rec.type) {
+    case 'started':
+      // worker 启动自报；pid 已由 daemon spawn 时 setWorkerPid 写过，这里仅记日志。
+      log.info(`worker 启动 task=${taskId} run=${runId} pid=${rec.pid} model=${rec.model}`);
+      return;
+
+    case 'phase':
+      // 进度阶段，仅日志（看板事件可后续接）。
+      log.info(`task=${taskId} run=${runId} 阶段=${rec.phase}`);
+      return;
+
+    case 'failed': {
+      // 先把 executor run 收尾为 failed（带转录/会话），再走失败/重试策略。
+      // failTaskAttempt 见 run 已 ended（非 started）不会重复收尾，只做重试决策。
+      try {
+        store.finishRun(runId, 'failed', {
+          error: rec.error,
+          transcriptRef: rec.transcriptRef ?? undefined,
+          claudeSessionId: rec.sessionId ?? undefined,
+        });
+      } 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}`);
+      return;
+    }
+
+    case 'result': {
+      // 成功终态：建 reviewer/security 两条复审 run + 收尾，写 result 四字段，转 exec_review，收尾 executor run。
+      const cr = store.startRun(taskId, 'reviewer', { worktree: rec.worktree, branch: rec.branch });
+      store.finishRun(cr.id, 'succeeded', { transcriptRef: rec.code.transcriptRef ?? undefined });
+
+      const sr = store.startRun(taskId, 'security', { worktree: rec.worktree, branch: rec.branch });
+      store.finishRun(sr.id, 'succeeded', { transcriptRef: rec.security.transcriptRef ?? undefined });
+
+      store.setResult(taskId, {
+        branch: rec.branch,
+        worktree: rec.worktree,
+        diffSummary: rec.diffSummary,
+        commits: rec.commits,
+        prUrl: null,
+        summary: rec.code.summary,
+        verdict: rec.code.verdict,
+        securitySummary: rec.security.summary,
+        securityVerdict: rec.security.verdict,
+        mergeTaskId: null,
+      });
+      store.transition(taskId, 'exec_review', { by: 'ingest', runId });
+      store.finishRun(runId, 'succeeded', {
+        transcriptRef: rec.executor.transcriptRef ?? undefined,
+        claudeSessionId: rec.executor.sessionId ?? undefined,
+      });
+      log.info(`task=${taskId} run=${runId} 执行完成 → exec_review（${rec.commits.length} commits）`);
+      return;
+    }
+
+    case 'done':
+      // worker 即将退出：终态标记。run 已由 result/failed 收尾，这里无须额外 DB 写。
+      log.info(`task=${taskId} run=${runId} worker 退出`);
+      return;
+  }
+}
+
+/**
+ * 摄取所有在途 run：对每个 executing 任务的最近一条 executor run 调 ingestRun。
+ * daemon 每轮 tick 调一次，把 worker 期间累积的 outbox 落库。
+ */
+export function ingestAll(store: Store, log: IngestLogger): void {
+  for (const { run } of store.executingWithLatestExecutorRun()) {
+    if (!run) continue;
+    ingestRun(store, log, run.id);
+  }
+}
@@ -1,66 +1,107 @@
+import { spawn } from 'node:child_process';
+import { fileURLToPath } from 'node:url';
+import { dirname, join } from 'node:path';
 import { Store } from '../store/index.js';
-import type { Project, Task, ReviewVerdict } from '../model/types.js';
+import type { Project, Task } from '../model/types.js';
 import { rankByScore } from '../model/scoring.js';
-import { createWorktree, worktreeDiff, type WorktreeDiff, type WorktreeInfo } from '../executor/worktree.js';
-import { runTask, type RunnerFn } from '../executor/runner.js';
-import { runVerify, type VerifyFn } from '../executor/verify.js';
-import { reviewCode, reviewSecurity, type ReviewerFn } from '../executor/reviewer.js';
+import { branchFor, worktreeDirFor } from '../executor/worktree.js';
+import { writeJobSpec, isWorkerAlive, heartbeatAgeMs, type JobSpec } from '../executor/protocol.js';
 import { pickModel } from '../executor/models.js';
+import { ingestAll, type IngestLogger } from './ingest.js';

-/** 失败后默认最多自动重试次数（项目级 maxRetries 未设置时回退此值） */
+/** 失败后默认最多自动重试次数（项目级 maxRetries 未设置时回退此值）。失败/退避策略本体在 store.failTaskAttempt。 */
 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;
  error(msg: string): void;
 }

-/** 依赖注入点：测试传 mock，生产用真实现 */
+/**
+ * worker 入口解析：返回 spawn 用的 [cmd, ...args]（不含 runId，由调用方追加）。
+ * - 默认：node <本文件同级 ../executor/worker.js>（编译后 dist 布局：daemon/ 与 executor/ 同级）。
+ * - env MAESTRO_WORKER_CMD 覆盖整条命令（空格分隔），供测试 / tsx 跑 .ts 入口用，
+ *   例：MAESTRO_WORKER_CMD="npx tsx src/executor/worker.ts"。
+ */
+export function workerEntry(): string[] {
+  const override = process.env.MAESTRO_WORKER_CMD?.trim();
+  if (override) return override.split(/\s+/);
+  const here = dirname(fileURLToPath(import.meta.url));
+  return ['node', join(here, '..', 'executor', 'worker.js')];
+}
+
+/**
+ * spawn worker 时传给子进程的最小 env：只保留 PATH/HOME/LANG 与 ANTHROPIC_ / CLAUDE_ 前缀变量，
+ * 不把 daemon 全量 env（含 MAESTRO_ 控制变量、无关密钥）漏给独立 worker 进程。
+ * MAESTRO_DATA_DIR 例外透传——worker 必须和 daemon 看同一个 runs/ 目录（job.json/outbox 路径一致）。
+ */
+export function workerEnv(src: NodeJS.ProcessEnv = process.env): NodeJS.ProcessEnv {
+  const out: NodeJS.ProcessEnv = {};
+  for (const k of ['PATH', 'HOME', 'LANG', 'MAESTRO_DATA_DIR', 'MAESTRO_WORKER_CMD'] as const) {
+    if (src[k] !== undefined) out[k] = src[k];
+  }
+  for (const [k, v] of Object.entries(src)) {
+    if (v !== undefined && (k.startsWith('ANTHROPIC_') || k.startsWith('CLAUDE_'))) out[k] = v;
+  }
+  return out;
+}
+
+/** 默认 spawnWorker：detached + 脱离 stdio + 收敛 env，unref 后自存活（daemon 退出不杀 worker）。返回子进程 pid。 */
+function defaultSpawnWorker(runId: string): number {
+  const [cmd, ...args] = workerEntry();
+  const child = spawn(cmd, [...args, runId], {
+    detached: true,
+    stdio: 'ignore',
+    env: workerEnv(),
+  });
+  child.unref();
+  if (child.pid === undefined) throw new Error('spawn worker 未返回 pid');
+  return child.pid;
+}
+
+/** 依赖注入点：生产用默认实现，测试传 mock（不真 spawn / 不真判活 / ingest no-op）。 */
 export interface OrchestratorDeps {
-  runner: RunnerFn;
-  reviewCode: ReviewerFn;        // code review（kind=reviewer 的 run）
-  reviewSecurity: ReviewerFn;    // 安全审计（kind=security 的 run）
-  verify: VerifyFn;
-  createWorktree: (repoPath: string, taskId: string, baseBranch: string) => Promise<WorktreeInfo>;
-  worktreeDiff: (repoPath: string, dir: string, branch: string, baseBranch: string) => Promise<WorktreeDiff>;
-  /** 当前时间戳（ms）；测试注入可控时钟（默认 Date.now） */
+  /** spawn 一个独立 worker 进程跑该 run，返回其 pid。 */
+  spawnWorker: (runId: string) => number;
+  /** worker 是否存活（reaper 判死用），默认 protocol.isWorkerAlive。 */
+  isWorkerAlive: typeof isWorkerAlive;
+  /** 把所有在途 run 的 outbox 摄取入库（daemon 唯一写者），默认 ingest.ingestAll。 */
+  ingestAll: (store: Store, log: IngestLogger) => void;
+  /** 写 worker 输入 job.json，默认 protocol.writeJobSpec（测试可 mock，避免落盘）。 */
+  writeJobSpec: (job: JobSpec) => void;
+  /** 当前时间戳（ms）；测试注入可控时钟（默认 Date.now）。 */
  nowMs: () => number;
 }

 export interface Orchestrator {
-  /** 跑一轮领取（同步领取 + 异步执行，不阻塞）。错误只记日志。 */
+  /** 跑一轮：先 ingest（outbox→DB）+ reaper（回收死 worker），再领新任务 spawn worker。错误只记日志、不抛。 */
  tick(): void;
-  /** 等待所有在途执行收尾（测试用） */
-  drain(): Promise<void>;
-  /** 在途任务 id（防同任务重复领取） */
-  readonly inflight: ReadonlyMap<string, string>;
+  /** 仅领取/spawn 那一步（测试细粒度断言用，不含 ingest/reaper）。 */
+  claimTick(): void;
+  /** ingest 所有在途 run（测试用）。 */
+  ingest(): void;
+  /** 回收死 worker（测试用）。 */
+  reap(): void;
 }

 /**
- * 编排器核心（与定时器解耦，便于测试）。
- * 每轮对 status=active 且 autonomy≠manual 的项目：在途数 < concurrency 时领任务——
- * ready 叶子（deps 全 done；auto-easy 只领 easy）或 queued（重试/重启遗留）。
- * 成功 → setResult + exec_review；失败 → failed → 重试 ≤MAX_RETRIES 次 → needs_attention。
+ * 编排器（监工）：自身不跑 agent，只领任务 + spawn 独立 worker + 把 worker 的 outbox 摄取入库 + 回收死 worker。
+ * 全部 DB 写都在 daemon 进程内（含 ingest 落的 run/result/transition），WS 推送照常经 store.subscribe 触发。
+ * 无内存在途态——并发闸读 store.countExecuting，退避读 task.nextEligibleAt，故跨 daemon 重启天然持久。
 */
 export function createOrchestrator(store: Store, log: OrchestratorLogger, deps: Partial<OrchestratorDeps> = {}): Orchestrator {
-  const d: OrchestratorDeps = { runner: runTask, reviewCode, reviewSecurity, verify: runVerify, createWorktree, worktreeDiff, nowMs: Date.now, ...deps };
-  const inflight = new Map<string, string>();       // taskId → projectId
-  const pending = new Set<Promise<void>>();
-  /** 退避表：taskId → 下次可领取时间戳（ms）；daemon 重启后清空，已有 queued 任务即刻可领 */
-  const backoffUntil = new Map<string, number>();
+  const d: OrchestratorDeps = {
+    spawnWorker: defaultSpawnWorker,
+    isWorkerAlive,
+    ingestAll,
+    writeJobSpec,
+    nowMs: Date.now,
+    ...deps,
+  };

  /**
   * 本项目可领取的任务：queued（重试/孤儿）+ ready 叶子且 deps 全 done；auto-easy 只挑 easy。
-   * 按调度分降序返回（score = 自身分 + 已完成依赖分 + 等待解锁的 blocked 任务分，见 model/scoring.ts）。
+   * 退避读持久化的 task.nextEligibleAt（早于它不领）。按调度分降序返回（见 model/scoring.ts）。
   */
  function claimable(project: Project): Array<{ task: Task; score: number }> {
    const nowMs = d.nowMs();
@@ -69,178 +110,113 @@ export function createOrchestrator(store: Store, log: OrchestratorLogger, deps:
    const parents = new Set(tasks.filter((t) => t.parentId).map((t) => t.parentId as string));
    const easyOnly = project.autonomy === 'auto-easy';
    const candidates = tasks.filter((t) => {
-      if (inflight.has(t.id)) return false;
      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;               // 退避冷却中
+      if (t.nextEligibleAt && nowMs < Date.parse(t.nextEligibleAt)) return false;  // 退避冷却中
      return t.deps.every((dep) => byId.get(dep)?.status === 'done');
    });
    return rankByScore(candidates, tasks);
  }

-  /** 跑一个独立复审 run（kind=reviewer/security）。任何失败折叠为 summary=失败原因、verdict=null，不抛错、不挡任务。 */
-  async function runOneReview(
-    kind: 'reviewer' | 'security',
-    fn: ReviewerFn,
-    task: Task,
-    project: Project,
-    wt: WorktreeInfo,
-    executorReport: string,
-  ): Promise<{ summary: string | null; verdict: ReviewVerdict | null }> {
-    const label = kind === 'reviewer' ? 'code review' : '安全审计';
-    let reviewRunId: string | null = null;
-    try {
-      const review = store.startRun(task.id, kind, { worktree: wt.dir, branch: wt.branch });
-      reviewRunId = review.id;
-      log.info(`${label} 任务 ${task.id} run=${review.id} model=${pickModel(task, project, 'reviewer')}`);
-      const rv = await fn(task, project, wt, review.id, executorReport);
-      store.finishRun(review.id, 'succeeded', {
-        transcriptRef: rv.transcriptRef ?? undefined,
-        claudeSessionId: rv.sessionId ?? undefined,
-      });
-      log.info(`任务 ${task.id} ${label} 完成 verdict=${rv.verdict ?? '（未解析到）'}`);
-      return { summary: rv.summary, verdict: rv.verdict };
-    } catch (e) {
-      const reMsg = (e as Error).message;
-      if (reviewRunId) {
-        try { store.finishRun(reviewRunId, 'failed', { error: reMsg }); } catch { /* 收尾失败不影响主流程 */ }
-      }
-      log.error(`任务 ${task.id} ${label} 失败（不挡任务，照常进 exec_review）：${reMsg}`);
-      return { summary: `自动复审失败：${reMsg}`, verdict: null };
-    }
-  }
-
-  /** 单任务全流程：executing → worktree → run → verify → code review run → 安全审计 run（均失败不挡）→ setResult(四字段) → exec_review / failed(重试) */
-  async function executeTask(project: Project, task: Task): Promise<void> {
+  /** 领取一个任务：建分支/worktree 路径 → executing → startRun → 写 job.json → spawn worker → 记 pid。 */
+  function claimOne(project: Project, task: Task, score: number): void {
    let runId: string | null = null;
-    let runClosed = false;
    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 wt = await d.createWorktree(project.repoPath, task.id, project.defaultBranch);
-      const run = store.startRun(task.id, 'executor', { worktree: wt.dir, branch: wt.branch });
+      const run = store.startRun(task.id, 'executor', { worktree: dir, branch });
      runId = run.id;
-      log.info(`执行任务 ${task.id}「${task.title}」 run=${run.id} worktree=${wt.dir}`);

-      const rr = await d.runner(task, project, wt, run.id);
-      if (!rr.ok) {
-        store.finishRun(run.id, 'failed', {
-          error: rr.error ?? '执行失败',
-          transcriptRef: rr.transcriptRef ?? undefined,
-          claudeSessionId: rr.sessionId ?? undefined,
-        });
-        runClosed = true;
-        throw new Error(rr.error ?? '执行失败');
-      }
-
-      const vr = await d.verify(project, wt.dir, run.id);
-      if (!vr.ok) {
-        store.finishRun(run.id, 'failed', {
-          error: vr.error ?? 'verify 失败',
-          transcriptRef: rr.transcriptRef ?? undefined,
-          claudeSessionId: rr.sessionId ?? undefined,
-        });
-        runClosed = true;
-        throw new Error(vr.error ?? 'verify 失败');
-      }
-
-      const diff = await d.worktreeDiff(project.repoPath, wt.dir, wt.branch, project.defaultBranch);
-
-      // 双复审（顺序）：code review run（kind=reviewer）→ 安全审计 run（kind=security）。任一失败不挡任务。
-      const code = await runOneReview('reviewer', d.reviewCode, task, project, wt, rr.finalText ?? '');
-      const sec = await runOneReview('security', d.reviewSecurity, task, project, wt, rr.finalText ?? '');
-
-      store.setResult(task.id, {
-        branch: wt.branch, worktree: wt.dir,
-        diffSummary: diff.diffSummary, commits: diff.commits, prUrl: null,
-        summary: code.summary, verdict: code.verdict,
-        securitySummary: sec.summary, securityVerdict: sec.verdict,
-        mergeTaskId: null,
+      d.writeJobSpec({
+        runId: run.id,
+        task: { ...task, status: 'executing' },
+        project,
+        worktreeDir: dir,
+        branch,
      });
-      store.transition(task.id, 'exec_review', { by: 'orchestrator', runId: run.id });
-      store.finishRun(run.id, 'succeeded', {
-        transcriptRef: rr.transcriptRef ?? undefined,
-        claudeSessionId: rr.sessionId ?? undefined,
-      });
-      log.info(`任务 ${task.id} 执行完成 → exec_review（${diff.commits.length} commits）`);
+
+      log.info(`领取任务 ${task.id}「${task.title}」score=${score} run=${run.id} model=${pickModel(task, project, 'executor')}`);
+      const pid = d.spawnWorker(run.id);
+      store.setWorkerPid(run.id, pid);
+      log.info(`任务 ${task.id} 已起 worker pid=${pid} worktree=${dir}`);
    } catch (e) {
      const msg = (e as Error).message;
-      log.error(`任务 ${task.id} 执行失败：${msg}`);
+      log.error(`任务 ${task.id} 领取/spawn 失败：${msg}`);
      try {
-        if (runId && !runClosed) {
-          store.finishRun(runId, 'failed', { error: msg });                 // worktree 创建后抛错（diff 等）时收尾
-        } else if (!runId) {
-          // run 还没建（如 createWorktree 失败）：补记一条 failed run，保证重试计数不漏
-          const r = store.startRun(task.id, 'executor');
-          store.finishRun(r.id, 'failed', { error: msg });
-        }
-        store.transition(task.id, 'failed', { by: 'orchestrator', error: msg });
-        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 {
-          backoffUntil.delete(task.id);
-          store.transition(task.id, 'needs_attention', { by: 'orchestrator', allFailed, netFailed });
-          log.error(`任务 ${task.id} 净失败 ${netFailed} 次（上限 ${maxRetries}）→ needs_attention`);
-        }
+        store.failTaskAttempt(task.id, runId, msg);
      } catch (e2) {
        log.error(`任务 ${task.id} 失败收尾出错：${(e2 as Error).message}`);
      }
    }
  }

-  function tick(): void {
+  /** 领取轮：对每个 active 且 autonomy≠manual 的项目，在 countExecuting<concurrency 时按 claimable 领新任务并 spawn worker。 */
+  function claimTick(): void {
    try {
      for (const p of store.listProjects()) {
        if (p.status !== 'active' || p.autonomy === 'manual') continue;
-        let active = 0;
-        for (const pid of inflight.values()) if (pid === p.id) active++;
+        let active = store.countExecuting(p.id);
        if (active >= p.concurrency) continue;

        for (const { task, score } of claimable(p)) {
          if (active >= p.concurrency) break;
-          try {
-            if (task.status === 'ready') store.transition(task.id, 'queued', { by: 'orchestrator', score });
-          } catch (e) {
-            log.error(`任务 ${task.id} 入队失败：${(e as Error).message}`);
-            continue;
-          }
-          log.info(`领取任务 ${task.id}「${task.title}」score=${score} model=${pickModel(task, p, 'executor')}`);
-          inflight.set(task.id, p.id);
+          claimOne(p, task, score);
          active++;
-          const job: Promise<void> = executeTask(p, { ...task, status: 'queued' })
-            .catch((e) => log.error(`任务 ${task.id} 执行异常：${(e as Error).message}`))
-            .finally(() => {
-              inflight.delete(task.id);
-              pending.delete(job);
-            });
-          pending.add(job);
        }
      }
    } catch (e) {
-      log.error(`编排器轮询失败：${(e as Error).message}`);
+      log.error(`编排器领取轮失败：${(e as Error).message}`);
    }
  }

-  async function drain(): Promise<void> {
-    while (pending.size > 0) await Promise.allSettled([...pending]);
+  /** ingest 所有在途 run 的 outbox（daemon 唯一 DB 写者把 worker 进度/结果落库）。 */
+  function ingest(): void {
+    try {
+      d.ingestAll(store, log);
+    } catch (e) {
+      log.error(`ingest 轮失败：${(e as Error).message}`);
+    }
  }

-  return { tick, drain, inflight };
+  /** 回收：对每个 executing 任务判活，死 worker → failTaskAttempt（收尾 + 重试/needs_attention）。 */
+  function reap(): void {
+    try {
+      for (const { task, run } of store.executingWithLatestExecutorRun()) {
+        const alive = run !== null && 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 ?? '无'}）→ 回收`);
+        try {
+          store.failTaskAttempt(task.id, run?.id ?? null, 'worker 异常退出');
+        } catch (e) {
+          log.error(`任务 ${task.id} 回收收尾出错：${(e as Error).message}`);
+        }
+      }
+    } catch (e) {
+      log.error(`编排器回收轮失败：${(e as Error).message}`);
+    }
+  }
+
+  function tick(): void {
+    ingest();      // 先把 worker 进度/结果落库（可能把 executing→exec_review，腾出并发槽）
+    reap();        // 再回收死 worker（可能把 executing→queued/needs_attention，腾出并发槽）
+    claimTick();   // 最后领新任务
+  }
+
+  return { tick, claimTick, ingest, reap };
 }

 /**
 * 接线入口：MAESTRO_ORCH_INTERVAL（秒）控制轮询间隔，默认 15，0=关闭。
- * 返回 timer 供 shutdown 时 clearInterval。
+ * 每轮依次 ingest → reaper → 领取（见 tick）。返回 timer 供 shutdown 时 clearInterval。
 */
 export function startOrchestrator(
  store: Store,
@@ -255,6 +231,6 @@ export function startOrchestrator(
  const orch = createOrchestrator(store, app.log, deps);
  const timer = setInterval(() => orch.tick(), intervalSec * 1000);
  timer.unref();
-  app.log.info(`编排器已启用：每 ${intervalSec}s 一轮领取（autonomy≠manual 的 active 项目）`);
+  app.log.info(`编排器已启用：每 ${intervalSec}s 一轮（ingest→回收→领取，autonomy≠manual 的 active 项目）`);
  return timer;
 }