From 53efe32a6e3efda3cfb30dd8b68a2b34f66740ff Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Sat, 13 Jun 2026 15:03:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(executor):=20SDK=20=E6=8A=96=E5=8A=A8=20re?= =?UTF-8?q?sume=20=E7=BB=AD=E8=B7=91=20+=20worker=20=E6=B1=A0=E5=8C=96?= =?UTF-8?q?=E8=AF=84=E4=BC=B0=E6=9A=82=E7=BC=93=20[tsk=5F74Acz1Nh4K1J]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 可选优化任务，先评估收益再做： - worker 池化：评估后暂缓。当前「一 run ↔ 一 pid」模型干净（判活/回收/ re-adopt 全靠每 run 一 pid+心跳，worker 与 DB 隔离 + 重启 re-adopt 天然 崩溃恢复），池化会打破该不变量且只在高并发量下兑现收益。现规模保持 「每任务一进程 + 失败从头重跑」简单模型，仅记入 DESIGN.md。 - SDK resume：实现 worker 内部健壮性小优化。cc.ts 单次会话因流式异常中断 （流断/没收到 result，非超时取消、非 max_turns 终态）且已拿到 sessionId 时， 用 resume 接着原会话续跑一次（保留已干的活，不从头重来），剩余预算不足留 60s。 与「重启 re-adopt / 整体失败 daemon 从头重跑」正交不替代；与模型回退互斥； MAESTRO_SDK_RESUME=0 可关闭。query 改为可注入便于单测。 测试：新增 test/cc.test.ts（6 例：续跑成功/终态不续/无 session 不续/ 开关关闭/超时不续/模型回退）。npm test 全绿 163/163。 Co-Authored-By: Claude Opus 4.8 --- DESIGN.md | 2 + README.md | 1 + src/executor/cc.ts | 67 ++++++++++++++++++++---- test/cc.test.ts | 128 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 187 insertions(+), 11 deletions(-) create mode 100644 test/cc.test.ts diff --git a/DESIGN.md b/DESIGN.md index e8368f8..fc0deca 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -207,9 +207,11 @@ daemon 启动时订阅 Store 事件流，符合条件时用 `osascript display n - ✅ **Phase 2（自动化，已完成）**：编排器 + score 调度 + Agent SDK executor（Easy/Medium/Hard worktree 自动执行）+ verify + 双复审（code review + 安全审计）+ 自动合并（--no-ff + 逃生口）。 - ✅ **Phase 3（打磨，核心已完成）**：macOS 原生通知 + Claude 订阅额度透传 + 归档详情时间线 + 模型分级（复杂度→模型映射 + env 覆盖 + 回退链）+ 项目 logo + 侧栏拖动排序。 - 🔲 **后续**：并发资源精细监控 / 指标面板 / 可选远程看板 / PR 平台集成（GitHub PR 链接）。 +- 🟡 **worker 池化（已评估，暂缓）**：长驻 worker 从队列拉任务、省 spawn 冷启动开销，仅在 spawn 量很大时才有收益。当前模型是干净的「一 run ↔ 一 pid」：`setWorkerPid`/`isWorkerAlive`/reaper/`reconcileInterrupted` 全靠「每 run 一个 pid + 一份心跳」判活与恢复，且 worker 与 DB 完全隔离 + 重启 re-adopt 天然得到崩溃恢复。池化会打破该不变量（一个池 worker 在跑哪个 run？单进程崩溃波及 N 个任务、共享 worktree 冲突），收益只在高并发量下兑现。现规模下**保持「每任务一进程 + 失败从头重跑」的简单模型**，不引入池化。 ## 20. 风险与注意事项 - **Agent SDK headless 执行可靠性**：verify_cmd 把关 + 双复审 + needs_attention 人工兜底。executor run 最多 100 轮、30min 超时兜底；reviewer run 最多 40 轮、15min 超时兜底。 +- **SDK 抖动 → resume 续跑**（`cc.ts`）：单次会话若因流式异常中断（流断 / 没收到 result，**非**超时取消、**非** max_turns 等终态），且已拿到 sessionId，则在 worker 进程内用 `resume` 接着原会话续跑一次（保留已干的活，不从头重来），剩余预算不足时至少留 60s。这是 worker 内部健壮性小优化，与「重启靠 re-adopt、整体失败靠 daemon 从头重跑」**正交、不替代**它们；`MAESTRO_SDK_RESUME=0` 可关闭。与模型不可用回退（换模型重开会话）互斥。 - **自动执行安全**：worktree 隔离（改动不进主检出）+ 不自动 push + merge 需 exec_review accept + 安全审计 run 专门检查凭证/注入/权限问题。 - **中断恢复**：daemon 重启后 `reconcileInterrupted()` 自动恢复执行中断的任务（标 failed → queued）。 - **长任务上下文超限**：100 turns / 30min 兜底，超时后转 failed → 重试。 diff --git a/README.md b/README.md index f573b69..b9d9981 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,7 @@ exec_review accept 时自动 `--no-ff` 合并：若 defaultBranch 在主检出 | `MAESTRO_DATA_DIR` | SQLite + worktree 数据目录 | `~/.maestro` | | `MAESTRO_ORCH_INTERVAL` | 编排器轮询间隔（秒），`0`=关闭 | `15` | | `MAESTRO_NOTIFY` | `0` 关闭 macOS 通知 | 开启 | +| `MAESTRO_SDK_RESUME` | `0` 关闭 SDK 抖动时的 session resume 续跑（worker 内部健壮性优化） | 开启 | | `MAESTRO_MODEL_EASY` | executor Easy 模型覆盖 | `claude-sonnet-4-6` | | `MAESTRO_MODEL_MEDIUM` | executor Medium 模型覆盖 | `claude-opus-4-8` | | `MAESTRO_MODEL_HARD` | executor Hard 模型覆盖 | `claude-fable-5` | diff --git a/src/executor/cc.ts b/src/executor/cc.ts index 9159295..a5103ba 100644 --- a/src/executor/cc.ts +++ b/src/executor/cc.ts @@ -35,21 +35,43 @@ interface AttemptResult { finalText: string; sessionId: string | null; error?: string; + /** + * 失败是否属于「会话异常中断」——即流式断了 / 没收到 result 就结束（SDK 偶发抖动）。 + * 这类失败可凭 sessionId 用 resume 续跑（见 runClaude）。区别于： + * - 超时/取消（abort，预算已耗尽，续跑无意义）→ false + * - 收到 result 但 subtype=error_*（max_turns 等终态，重来只会再撞墙）→ false + */ + resumable?: boolean; } +/** runClaude 的 query 注入点：生产用 SDK 的 query，测试传 fake（避免真起 CC）。 */ +export type CCQuery = typeof query; + +/** resume 续跑至少留这么多预算（避免首攻已耗掉大半 timeout 后续跑没时间干活）。 */ +const RESUME_MIN_TIMEOUT_MS = 60_000; + /** 单次 headless CC 会话：流式消息逐行写 transcript，不抛错（失败折叠进 error） */ -async function attempt(opts: CCOptions, model: string, out: WriteStream): Promise { +async function attempt( + opts: CCOptions, + model: string, + out: WriteStream, + queryImpl: CCQuery, + extra?: { resume?: string; timeoutMs?: number }, +): Promise { let sessionId: string | null = null; let finalText = ''; let resultOk = false; let resultError: string | undefined; let sawResult = false; + let resumable = false; + let timedOut = false; const abort = new AbortController(); - const killer = setTimeout(() => abort.abort(new Error('执行超时')), opts.timeoutMs); + const timeoutMs = extra?.timeoutMs ?? opts.timeoutMs; + const killer = setTimeout(() => { timedOut = true; abort.abort(new Error('执行超时')); }, timeoutMs); try { - const q = query({ + const q = queryImpl({ prompt: opts.prompt, options: { cwd: opts.cwd, @@ -59,6 +81,7 @@ async function attempt(opts: CCOptions, model: string, out: WriteStream): Promis abortController: abort, model, allowedTools: opts.allowedTools, + ...(extra?.resume ? { resume: extra.resume } : {}), // 续跑：从该 session 加载历史接着干 }, }); @@ -73,44 +96,66 @@ async function attempt(opts: CCOptions, model: string, out: WriteStream): Promis const txt = (message as { result?: unknown }).result; if (typeof txt === 'string') finalText = txt; } else { + // 收到了 result（终态错误，如 max_turns）→ 不可 resume（重来只会再撞同一堵墙） const errs = 'errors' in message && Array.isArray(message.errors) ? message.errors.join('; ') : ''; resultError = `Claude Code 结束于 ${message.subtype}${errs ? `：${errs}` : ''}`; } } } - if (!sawResult && !resultError) resultError = '未收到 result 消息（会话异常结束）'; + if (!sawResult && !resultError) { + // 流跑完了却没见 result：会话异常结束 → 可 resume 续跑 + resultError = '未收到 result 消息（会话异常结束）'; + resumable = true; + } } catch (e) { + // 流式中途抛错：超时/取消导致的 abort 不可 resume；其余（网络抖动/传输错）可 resume resultError = `Claude Code 执行异常：${(e as Error).message}`; resultOk = false; + resumable = !timedOut; } finally { clearTimeout(killer); } - if (!resultOk) return { ok: false, finalText: '', sessionId, error: resultError ?? '未知错误' }; + if (!resultOk) return { ok: false, finalText: '', sessionId, error: resultError ?? '未知错误', resumable }; return { ok: true, finalText, sessionId }; } /** - * 跑一次 headless CC（runner / reviewer 公用）。 - * 模型可用性兜底：失败且错误信息像模型不可用（not_found/invalid/permission 等）时， - * 自动用回退链取一个 ≠ 原模型的模型在同一 run 内重试一次；回退记入 transcript 与 finalText。 + * 跑一次 headless CC（runner / reviewer 公用）。两层「同一 run 内重试一次」的兜底（互斥，按需触发）： + * + * 1. **SDK 抖动 → resume 续跑**：首攻因会话异常中断（流断/没收到 result，非超时/取消、非 max_turns 终态） + * 且已拿到 sessionId 时，用 `resume` 接着原会话续跑一次（保留已干的活，不从头重来）。 + * 这是 worker 进程内部的健壮性小优化，与「重启靠 re-adopt、失败靠 daemon 从头重跑」正交、不替代它们。 + * `MAESTRO_SDK_RESUME=0` 可关闭。 + * 2. **模型不可用 → 回退**：错误信息像模型不可用（not_found/invalid/permission 等）时， + * 用回退链取一个 ≠ 原模型的模型重试一次；回退记入 transcript 与 finalText。 + * + * queryImpl 默认用 SDK 的 query，测试可注入 fake。 */ -export async function runClaude(opts: CCOptions): Promise { +export async function runClaude(opts: CCOptions, queryImpl: CCQuery = query): Promise { const dir = transcriptDir(); mkdirSync(dir, { recursive: true }); const transcriptRef = join(dir, `${opts.runId}.jsonl`); const out = createWriteStream(transcriptRef, { flags: 'a' }); + const t0 = Date.now(); try { - let r = await attempt(opts, opts.model, out); + let r = await attempt(opts, opts.model, out, queryImpl); let modelUsed = opts.model; let fellBack = false; + // 1. SDK 抖动续跑：会话异常中断 + 有 sessionId + 非模型错误（那归回退处理） + 未关闭 + if (!r.ok && r.resumable && r.sessionId && !isModelError(r.error ?? '') && process.env.MAESTRO_SDK_RESUME !== '0') { + out.write(`${JSON.stringify({ type: 'maestro.session_resume', sessionId: r.sessionId, reason: r.error })}\n`); + const remaining = Math.max(opts.timeoutMs - (Date.now() - t0), RESUME_MIN_TIMEOUT_MS); + r = await attempt(opts, opts.model, out, queryImpl, { resume: r.sessionId, timeoutMs: remaining }); + } + if (!r.ok && r.error && isModelError(r.error)) { const fb = pickFallbackModel(opts.model); if (fb) { out.write(`${JSON.stringify({ type: 'maestro.model_fallback', from: opts.model, to: fb, reason: r.error })}\n`); - r = await attempt(opts, fb, out); + r = await attempt(opts, fb, out, queryImpl); modelUsed = fb; fellBack = true; } diff --git a/test/cc.test.ts b/test/cc.test.ts new file mode 100644 index 0000000..c96d1aa --- /dev/null +++ b/test/cc.test.ts @@ -0,0 +1,128 @@ +import { test, before, after } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { runClaude, type CCQuery, type CCOptions } from '../src/executor/cc.js'; + +// cc.ts 的 SDK resume 续跑 / 模型回退逻辑：注入 fake query，不真起 Claude Code。 + +let dataDir: string; +let prevData: string | undefined; +let prevResume: string | undefined; + +before(() => { + dataDir = mkdtempSync(join(tmpdir(), 'maestro-cc-')); + prevData = process.env.MAESTRO_DATA_DIR; + prevResume = process.env.MAESTRO_SDK_RESUME; + process.env.MAESTRO_DATA_DIR = dataDir; // transcript 落临时目录 + delete process.env.MAESTRO_SDK_RESUME; // 默认开启 resume +}); +after(() => { + if (prevData === undefined) delete process.env.MAESTRO_DATA_DIR; else process.env.MAESTRO_DATA_DIR = prevData; + if (prevResume === undefined) delete process.env.MAESTRO_SDK_RESUME; else process.env.MAESTRO_SDK_RESUME = prevResume; + rmSync(dataDir, { recursive: true, force: true }); +}); + +type Script = (args: { prompt: string; options: Record }) => AsyncGenerator; + +/** 串接多段脚本：第 i 次调用走第 i 段（越界用最后一段）；记录每次的 options（看 resume）。 */ +function makeQuery(scripts: Script[]): { fn: CCQuery; calls: Record[] } { + const calls: Record[] = []; + let i = 0; + const fn = ((args: { prompt: string; options: Record }) => { + calls.push(args.options); + const script = scripts[Math.min(i, scripts.length - 1)]; + i += 1; + return script(args); + }) as unknown as CCQuery; + return { fn, calls }; +} + +function baseOpts(runId: string, timeoutMs = 30_000): CCOptions { + return { prompt: 'do it', cwd: dataDir, model: 'claude-x', runId, maxTurns: 10, timeoutMs, allowedTools: [] }; +} + +// 段：拿到 sessionId 后流式中途抛错（SDK 抖动） +const transientThrow: Script = async function* (args) { + yield { type: 'system', session_id: 'sess-1' }; + void args; + throw new Error('socket hang up'); +}; +// 段：一条成功 result +const successResult: Script = async function* () { + yield { type: 'result', subtype: 'success', is_error: false, result: '完工', session_id: 'sess-1' }; +}; + +test('SDK 抖动（流中途抛错）+ 有 sessionId → resume 续跑一次并成功', async () => { + const { fn, calls } = makeQuery([transientThrow, successResult]); + const r = await runClaude(baseOpts('r-resume'), fn); + assert.equal(r.ok, true, '续跑后应成功'); + assert.equal(r.finalText, '完工'); + assert.equal(r.fellBack, false, '这是 resume，不是模型回退'); + assert.equal(calls.length, 2, '应正好两次 attempt（首攻 + resume）'); + assert.equal(calls[0].resume, undefined, '首攻不带 resume'); + assert.equal(calls[1].resume, 'sess-1', 'resume 应续上首攻的 sessionId'); +}); + +test('终态错误（result=error_max_turns）→ 不 resume（重来只会再撞墙）', async () => { + const terminal: Script = async function* () { + yield { type: 'result', subtype: 'error_max_turns', is_error: true, session_id: 'sess-1' }; + }; + const { fn, calls } = makeQuery([terminal, successResult]); + const r = await runClaude(baseOpts('r-terminal'), fn); + assert.equal(r.ok, false); + assert.equal(calls.length, 1, 'max_turns 是终态，不应触发 resume'); + assert.match(r.error ?? '', /error_max_turns/); +}); + +test('首攻无 sessionId（连 system 都没拿到就抛）→ 无可续之 session，不 resume', async () => { + const noSession: Script = async function* () { + throw new Error('connect ECONNREFUSED'); + }; + const { fn, calls } = makeQuery([noSession, successResult]); + const r = await runClaude(baseOpts('r-nosess'), fn); + assert.equal(r.ok, false); + assert.equal(calls.length, 1, '没有 sessionId 无法 resume'); +}); + +test('MAESTRO_SDK_RESUME=0 → 即便可续也不 resume', async () => { + process.env.MAESTRO_SDK_RESUME = '0'; + try { + const { fn, calls } = makeQuery([transientThrow, successResult]); + const r = await runClaude(baseOpts('r-disabled'), fn); + assert.equal(r.ok, false); + assert.equal(calls.length, 1, '开关关闭时不应 resume'); + } finally { + delete process.env.MAESTRO_SDK_RESUME; + } +}); + +test('超时（abort）→ 不 resume（预算已耗尽）', async () => { + // 拿到 session 后挂起，直到 abort 信号触发才抛错；配极小 timeout 强制超时 + const hangUntilAbort: Script = async function* (args) { + yield { type: 'system', session_id: 'sess-1' }; + await new Promise((_res, rej) => { + const ac = args.options.abortController as AbortController; + ac.signal.addEventListener('abort', () => rej(new Error('aborted'))); + }); + }; + const { fn, calls } = makeQuery([hangUntilAbort, successResult]); + const r = await runClaude(baseOpts('r-timeout', 40), fn); + assert.equal(r.ok, false); + assert.equal(calls.length, 1, '超时导致的 abort 不可 resume'); +}); + +test('模型不可用错误 → 走模型回退（非 resume），续跑标记 fellBack', async () => { + // 首攻报模型不可用（终态 result，非中断）→ 不 resume，应触发模型回退 + const modelErr: Script = async function* () { + yield { type: 'result', subtype: 'error_during_execution', is_error: true, session_id: 'sess-1', errors: ['model not_found'] }; + }; + const { fn, calls } = makeQuery([modelErr, successResult]); + const r = await runClaude(baseOpts('r-modelfb'), fn); + assert.equal(r.ok, true, '回退模型后成功'); + assert.equal(r.fellBack, true); + assert.equal(calls.length, 2); + assert.equal(calls[1].resume, undefined, '模型回退是重开会话，不带 resume'); + assert.notEqual(calls[1].model, calls[0].model, '回退应换了模型'); +});