merge: worker 池化 + SDK resume 抗抖动 [tsk_74Acz1Nh4K1J]
手工正确合并(破除补救死循环):带上 src/executor/cc.ts 池化 + SDK resume + test/cc.test.ts(typecheck + 189 测试全绿)。DESIGN.md 冲突取 main 新版(多进程 re-adopt / worker 不碰 DB 描述),保留 246 行 SDK-resume 新增条目,丢弃分支过时描述。 此前补救任务只啃了 DESIGN.md 冲突文件、没带 cc.ts 代码,导致真代码一直没落地、原任务卡 exec_review、每次重试又生新补救(xbtd6GpS4NYg→Yb8eFAX2VUxZ)的死循环,至此终结。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
@@ -107,6 +107,7 @@ exec_review accept 时自动 `--no-ff` 合并：若 defaultBranch 在主检出
|
||||
| `MAESTRO_ORCH_INTERVAL` | 编排器轮询间隔（秒），`0`=关闭 | `15` |
|
||||
| `MAESTRO_WORKER_CMD` | 覆盖 worker 进程启动命令（空格分隔），供 tsx 跑 .ts 入口/测试用 | `node dist/executor/worker.js` |
|
||||
| `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` |
|
||||
|
||||
+56
-11
@@ -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<AttemptResult> {
|
||||
async function attempt(
|
||||
opts: CCOptions,
|
||||
model: string,
|
||||
out: WriteStream,
|
||||
queryImpl: CCQuery,
|
||||
extra?: { resume?: string; timeoutMs?: number },
|
||||
): Promise<AttemptResult> {
|
||||
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<CCResult> {
|
||||
export async function runClaude(opts: CCOptions, queryImpl: CCQuery = query): Promise<CCResult> {
|
||||
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;
|
||||
}
|
||||
|
||||
+128
@@ -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<string, unknown> }) => AsyncGenerator<unknown>;
|
||||
|
||||
/** 串接多段脚本：第 i 次调用走第 i 段（越界用最后一段）；记录每次的 options（看 resume）。 */
|
||||
function makeQuery(scripts: Script[]): { fn: CCQuery; calls: Record<string, unknown>[] } {
|
||||
const calls: Record<string, unknown>[] = [];
|
||||
let i = 0;
|
||||
const fn = ((args: { prompt: string; options: Record<string, unknown> }) => {
|
||||
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, '回退应换了模型');
|
||||
});
|
||||
Reference in New Issue
Block a user