feat(cleanup): exec_review 合并后回收 runs/<runId>/ 工作目录 + 定期扫尾 [tsk_6uziW1Dg-ftQ]

每个 run 在 ~/.maestro/runs/<runId>/ 留下 job.json/outbox.ndjson/heartbeat，合并（exec_review accept）后不再需要，需回收避免无限堆积。 - protocol.ts：新增 removeRunDir / isPlainRunId（幂等删 runDir，runId 白名单防越界） - cleanup.ts（新）： - cleanupTaskRunArtifacts：合并/接受后即时回收某任务全部 run 工作目录 - sweepRunArtifacts：周期扫尾终态/孤儿 runDir 与（可选）转录，按保留天数 - isRunCleanable：仅终态(done/cancelled)/孤儿可回收，活跃/在途一律保留 - 双保险：任务状态判定 + heartbeat 新鲜则跳过，绝不误删活跃 run - 保留策略可配：MAESTRO_RUN_RETENTION_DAYS(默认3) / MAESTRO_TRANSCRIPT_RETENTION_DAYS(默认0=永久保留，便于排查) - server.ts：exec_review accept 合并成功后即时清 runDir（与 removeWorktree 同处） - index.ts：startCleanupLoop 定时扫尾（MAESTRO_CLEANUP_INTERVAL 秒，默认3600，0=关闭） - test/cleanup.test.ts：终态清/在途留/孤儿清/保留期/heartbeat 兜底/转录开关/配置测试：npm test → 164 passed；npm run typecheck → clean Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 15:02:23 +08:00
parent f020e15137
commit a365aaf06a
5 changed files with 396 additions and 1 deletions
@@ -0,0 +1,166 @@
+// Run 产物回收（Phase 3 多进程执行的扫尾）。
+//
+// 每个 run 在两处留下产物：
+//   - runs/<runId>/        ── job.json / outbox.ndjson / heartbeat（纯运行期脚手架，落库后无价值）
+//   - <transcriptDir>/<runId>.jsonl  ── 转录（事后排查有价值，默认长保留）
+//
+// 终态任务（done/cancelled）的产物不再需要；orphan（任务已删，run 行被 FK 级联清掉）的产物同理。
+// 本模块提供：
+//   - cleanupTaskRunArtifacts：合并/接受后即时回收某任务的全部 run 工作目录（server 调）。
+//   - sweepRunArtifacts      ：周期扫尾，按保留天数回收终态/孤儿 runDir 与（可选）转录（daemon 调）。
+//
+// 安全铁律：只删【终态或孤儿】run 的产物；活跃/在途/exec_review 的一律保留。
+// 双保险：① 任务状态判定 ② heartbeat 新鲜则跳过（兜底防误删仍在写的 run）。
+
+import { existsSync, readdirSync, rmSync, statSync } from 'node:fs';
+import { join } from 'node:path';
+import { Store } from '../store/index.js';
+import { TERMINAL_STATUS } from '../model/status.js';
+import {
+  runsBase, runDir, removeRunDir, isPlainRunId, heartbeatAgeMs, HEARTBEAT_GRACE_MS,
+} from './protocol.js';
+import { transcriptDir } from './cc.js';
+
+const DAY_MS = 24 * 60 * 60 * 1000;
+
+export interface CleanupLogger {
+  info(msg: string): void;
+  error(msg: string): void;
+}
+
+export interface CleanupConfig {
+  /** runs/<runId>/ 保留天数（自目录 mtime 起算）；0 = 终态后立即可回收。 */
+  runRetentionDays: number;
+  /** <transcriptDir>/<runId>.jsonl 保留天数；<=0 = 永不自动清理（默认，便于事后排查）。 */
+  transcriptRetentionDays: number;
+}
+
+export const DEFAULT_CLEANUP_CONFIG: CleanupConfig = {
+  runRetentionDays: 3,
+  transcriptRetentionDays: 0,
+};
+
+/** 从环境变量读保留策略（覆盖默认）。非法值回退默认。 */
+export function loadCleanupConfig(env: NodeJS.ProcessEnv = process.env): CleanupConfig {
+  const num = (raw: string | undefined, fallback: number): number => {
+    if (raw === undefined || raw.trim() === '') return fallback;
+    const n = Number(raw);
+    return Number.isFinite(n) && n >= 0 ? n : fallback;
+  };
+  return {
+    runRetentionDays: num(env.MAESTRO_RUN_RETENTION_DAYS, DEFAULT_CLEANUP_CONFIG.runRetentionDays),
+    transcriptRetentionDays: num(env.MAESTRO_TRANSCRIPT_RETENTION_DAYS, DEFAULT_CLEANUP_CONFIG.transcriptRetentionDays),
+  };
+}
+
+/**
+ * 该 run 的工作目录是否可回收：终态任务 / orphan（run 行或任务已不在 DB）→ 可；活跃/在途 → 否。
+ * 不看保留天数与 heartbeat（那是调用方叠加的策略/兜底）。
+ */
+export function isRunCleanable(store: Store, runId: string): boolean {
+  const run = store.getRun(runId);
+  if (!run) return true;                       // orphan：任务已删，run 行被 FK 级联清除 → 回收
+  const task = store.getTask(run.taskId);
+  if (!task) return true;                      // 同上（防御）
+  return TERMINAL_STATUS.has(task.status);     // done / cancelled
+}
+
+/** heartbeat 仍新鲜（有进程在写）→ 不可删，兜底防误删活跃 run。 */
+function heartbeatFresh(runId: string, now: number): boolean {
+  const age = heartbeatAgeMs(runId, now);
+  return age !== null && age < HEARTBEAT_GRACE_MS;
+}
+
+/**
+ * 即时回收某任务全部 run 的工作目录（runs/<runId>/）。合并成功 / 接受后由 server 调。
+ * 只删工作目录，不动转录（保留排查价值）。失败只计日志，绝不抛（不阻断主流程）。
+ * 返回实际清理的目录数。
+ */
+export function cleanupTaskRunArtifacts(store: Store, taskId: string, log: CleanupLogger): number {
+  let n = 0;
+  try {
+    for (const run of store.listRuns(taskId)) {
+      if (heartbeatFresh(run.id, Date.now())) continue;   // 极少见：仍有进程在写则跳过
+      try {
+        removeRunDir(run.id);
+        n++;
+      } catch (e) {
+        log.error(`回收 run 目录 ${run.id} 失败（不影响结果）：${(e as Error).message}`);
+      }
+    }
+  } catch (e) {
+    log.error(`回收任务 ${taskId} 的 run 产物失败（不影响结果）：${(e as Error).message}`);
+  }
+  return n;
+}
+
+export interface SweepResult {
+  runDirs: number;       // 清理的 runs/<runId>/ 目录数
+  transcripts: number;   // 清理的转录文件数
+}
+
+/**
+ * 周期扫尾：遍历 runs/ 与 transcripts/，回收终态/孤儿且超过保留期的产物。
+ * - runDir：任务终态/孤儿 且 目录 mtime 早于 runRetentionDays 且 heartbeat 不新鲜 → 删。
+ * - 转录：transcriptRetentionDays>0 时，任务终态/孤儿 且 文件 mtime 早于该天数 → 删（活跃任务永不删）。
+ * 单条失败不阻断其余。返回清理计数。
+ */
+export function sweepRunArtifacts(
+  store: Store,
+  cfg: CleanupConfig,
+  log: CleanupLogger,
+  now: number = Date.now(),
+): SweepResult {
+  const res: SweepResult = { runDirs: 0, transcripts: 0 };
+
+  // ── runs/<runId>/ ──
+  const base = runsBase();
+  if (existsSync(base)) {
+    let entries: string[] = [];
+    try { entries = readdirSync(base); } catch (e) { log.error(`读取 runs 目录失败：${(e as Error).message}`); }
+    const cutoff = now - cfg.runRetentionDays * DAY_MS;
+    for (const runId of entries) {
+      if (!isPlainRunId(runId)) continue;
+      const dir = runDir(runId);
+      try {
+        const st = statSync(dir);
+        if (!st.isDirectory()) continue;
+        if (!isRunCleanable(store, runId)) continue;          // 活跃/在途 → 保留
+        if (st.mtimeMs > cutoff) continue;                    // 未过保留期
+        if (heartbeatFresh(runId, now)) continue;             // 仍在写 → 兜底保留
+        removeRunDir(runId);
+        res.runDirs++;
+      } catch (e) {
+        log.error(`扫尾 run 目录 ${runId} 失败：${(e as Error).message}`);
+      }
+    }
+  }
+
+  // ── <transcriptDir>/<runId>.jsonl ──（默认关闭）
+  if (cfg.transcriptRetentionDays > 0) {
+    const tdir = transcriptDir();
+    if (existsSync(tdir)) {
+      let files: string[] = [];
+      try { files = readdirSync(tdir); } catch (e) { log.error(`读取转录目录失败：${(e as Error).message}`); }
+      const cutoff = now - cfg.transcriptRetentionDays * DAY_MS;
+      for (const file of files) {
+        if (!file.endsWith('.jsonl')) continue;
+        const runId = file.slice(0, -'.jsonl'.length);
+        if (!isPlainRunId(runId)) continue;
+        const fp = join(tdir, file);
+        try {
+          const st = statSync(fp);
+          if (!st.isFile()) continue;
+          if (!isRunCleanable(store, runId)) continue;        // 活跃任务的转录 → 永不删
+          if (st.mtimeMs > cutoff) continue;                  // 未过保留期
+          rmSync(fp, { force: true });
+          res.transcripts++;
+        } catch (e) {
+          log.error(`扫尾转录 ${file} 失败：${(e as Error).message}`);
+        }
+      }
+    }
+  }
+
+  return res;
+}
@@ -10,7 +10,7 @@

 import {
  existsSync, mkdirSync, readFileSync, writeFileSync, appendFileSync,
-  statSync, utimesSync, closeSync, openSync,
+  statSync, utimesSync, closeSync, openSync, rmSync,
 } from 'node:fs';
 import { homedir } from 'node:os';
 import { join } from 'node:path';
@@ -33,6 +33,21 @@ export function jobPath(runId: string): string { return join(runDir(runId), 'job
 export function outboxPath(runId: string): string { return join(runDir(runId), 'outbox.ndjson'); }
 export function heartbeatPath(runId: string): string { return join(runDir(runId), 'heartbeat'); }

+/**
+ * 删除单个 run 的工作目录 runs/<runId>/（job.json + outbox.ndjson + heartbeat）。
+ * 幂等（目录不存在不报错）。合并/终态后由 daemon 侧回收调用——worker 自身从不删自己的工作目录。
+ * 仅接受形如 run_xxx 的纯 id（无路径分隔符），避免越界删除。
+ */
+export function removeRunDir(runId: string): void {
+  if (!isPlainRunId(runId)) return;
+  rmSync(runDir(runId), { recursive: true, force: true });
+}
+
+/** runId 合法性：仅字母数字与 -_，杜绝 . / 等可越界字符（清理路径计算的安全前提）。 */
+export function isPlainRunId(runId: string): boolean {
+  return /^[A-Za-z0-9_-]+$/.test(runId);
+}
+
 // ───────────────────────── daemon → worker：JobSpec ─────────────────────────

 /**