phase3(daemon): 监工 tick(spawn worker)+ ingest(outbox→DB)+ reaper + reconcile 接真判活

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 13:00:50 +08:00
parent 007dffac39
commit da7c93105c
5 changed files with 765 additions and 567 deletions
@@ -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);
+  }
+}