From f6b0e1ad1cb07b63d0544acd3c864ff9ecd2a60a Mon Sep 17 00:00:00 2001 From: zjdznl Date: Tue, 13 Oct 2026 12:07:18 +0800 Subject: [PATCH] =?UTF-8?q?fix(cli):=20=E5=A4=A7=E8=BE=93=E5=87=BA?= =?UTF-8?q?=E5=86=99=E5=AE=8C=20stdout=20=E5=86=8D=E9=80=80=E5=87=BA?= =?UTF-8?q?=EF=BC=8C=E9=81=BF=E5=85=8D=20ask/hook/relay=20=E6=88=AA?= =?UTF-8?q?=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit stdout 接 pipe 时 write() 是异步的，紧跟显式 process.exit() 会丢掉未排空 的尾部（实测父进程只收到约 146 KiB）。PM2 只读 jlist 的同类问题已在 master 修复，本提交补齐其余仍会输出大文本的退出路径： - 新增 src/cli/stdout-flush.ts 的 writeAndFlush：以 write 完成回调为边界 - botmux ask：写完结果再 ack；写失败时不 ack、exit 1，daemon 仍可回收回执 - botmux hook：directive 内嵌用户文字作答，同样等写完再 exit 0 - sandbox relay：send 与 dispatch 两处统一走 mirrorRelayResponseAndExit， stdout/stderr 均写完再镜像退出码 - 测试：writeAndFlush 子进程 2 MiB 管道回归；ask --json 50 万字符 comment 端到端回归；PM2 只读 helper 大注册表（>200 KB）回归，钉住 master 的修复。 PM2 用真实 Node 启动（bun 下 execPath 不是 node） Co-Authored-By: Claude Opus 5.5 --- src/cli.ts | 59 +++++++++++++++++++----- src/cli/stdout-flush.ts | 16 +++++++ test/ask-cli.test.ts | 46 +++++++++++++++++++ test/pm2-readonly-jlist.test.ts | 79 +++++++++++++++++++++++++++++++++ test/stdout-flush.test.ts | 34 ++++++++++++++ 5 files changed, 222 insertions(+), 12 deletions(-) create mode 100644 src/cli/stdout-flush.ts create mode 100644 test/pm2-readonly-jlist.test.ts create mode 100644 test/stdout-flush.test.ts diff --git a/src/cli.ts b/src/cli.ts index 7b4f1bc489..736d4a393b 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -36,6 +36,7 @@ import { existsSync, mkdirSync, copyFileSync, createReadStream, readFileSync, wr import { underReadIsolation, sendCredFilePath } from './adapters/cli/read-isolation.js'; import { atomicWriteFileSync } from './utils/atomic-write.js'; import { readAllowedUsersResolveCache } from './utils/allowed-users-cache.js'; +import { writeAndFlush } from './cli/stdout-flush.js'; import { join, dirname, basename, resolve } from 'node:path'; import { homedir, userInfo } from 'node:os'; import { fileURLToPath } from 'node:url'; @@ -8875,9 +8876,7 @@ async function relaySend( if (preparedContentOutfile) { try { unlinkSync(preparedContentOutfile); } catch { /* */ } } if (cardOutfile) { try { unlinkSync(cardOutfile); } catch { /* */ } } if (completionProposalOutfile) { try { unlinkSync(completionProposalOutfile); } catch { /* */ } } - if (res.stdout) process.stdout.write(res.stdout); - if (res.stderr) process.stderr.write(res.stderr); - process.exit(res.code ?? 0); + await mirrorRelayResponseAndExit(res, 0); } catch { /* partial write — retry next tick */ } } await new Promise(r => setTimeout(r, 150)); @@ -8886,6 +8885,29 @@ async function relaySend( process.exit(1); } +/** + * Mirror a host relay response (`.res.json`) onto this process's stdio and + * exit with its code. The host child normally returns a short JSON, but either + * stream can carry an upstream error body of any size; because the relay exits + * explicitly, both writes must complete first or a large response is silently + * truncated at the sandbox boundary. + */ +async function mirrorRelayResponseAndExit( + res: { code?: number; stdout?: string; stderr?: string }, + defaultCode: number, +): Promise { + try { + await Promise.all([ + res.stdout ? writeAndFlush(process.stdout, res.stdout) : Promise.resolve(), + res.stderr ? writeAndFlush(process.stderr, res.stderr) : Promise.resolve(), + ]); + } catch (writeError) { + console.error(`relay: 无法完整输出 host 响应：${writeError instanceof Error ? writeError.message : String(writeError)}`); + process.exit(1); + } + process.exit(res.code ?? defaultCode); +} + /** * Relay a sandboxed cross-bot dispatch to the owning worker. The sandbox only * contributes a bounded target app/chat plus task text; the watcher forces the @@ -8965,9 +8987,7 @@ async function relayDispatch(rest: string[], relayDir: string): Promise { const res = JSON.parse(readFileSync(resPath, 'utf8')) as { code?: number; stdout?: string; stderr?: string }; try { unlinkSync(resPath); } catch { /* */ } try { unlinkSync(contentPath); } catch { /* */ } - if (res.stdout) process.stdout.write(res.stdout); - if (res.stderr) process.stderr.write(res.stderr); - process.exit(res.code ?? 1); + await mirrorRelayResponseAndExit(res, 1); } catch { /* atomic response should be complete; retry defensively */ } } await new Promise(resolve => setTimeout(resolve, 150)); @@ -14906,6 +14926,7 @@ async function cmdAsk(sub: string, rest: string[]): Promise { // result.kind==='answered' 时用 toLegacySelected 取回旧的 string（单问单选） const selected = toLegacySelected(result); + let stdoutPayload: string | undefined; if (useJson) { const out: AskJsonOutput = { // `selected` 是「单问单选」的向后兼容值（= toLegacySelected 的形状判据： @@ -14918,7 +14939,7 @@ async function cmdAsk(sub: string, rest: string[]): Promise { comment: result.kind === 'answered' ? result.comment : null, timedOut: result.kind === 'timedOut', }; - process.stdout.write(JSON.stringify(out) + '\n'); + stdoutPayload = JSON.stringify(out) + '\n'; } else if (result.kind === 'answered') { // 非 JSON 模式：单选输出 key，多选输出逗号分隔的 keys。 // @@ -14935,12 +14956,23 @@ async function cmdAsk(sub: string, rest: string[]): Promise { // N 项→逗号分隔，全程 exit 0。文字作答时 answers[0] 为空数组同样落空行（上面已在 // stderr 提示改读 --json 的 comment）。 const value = multiSelect ? (result.answers[0]?.join(',') ?? '') : (selected ?? ''); - process.stdout.write(value + '\n'); + stdoutPayload = value + '\n'; } - // Flush the actual result before acknowledging delivery. If this process - // disappears first, the daemon can still recover its unconsumed receipt. - await new Promise(resolve => process.stdout.write('', () => resolve())); + // Flush the actual result before acknowledging delivery (and before the + // explicit process.exit below): `ask --json` carries the user's free-form + // comment, whose size is unbounded, and an exit right after write() can drop + // the tail of a large pipe write. If the write fails, exit without the ack + // so the daemon can still recover its unconsumed receipt. + if (stdoutPayload !== undefined) { + try { + await writeAndFlush(process.stdout, stdoutPayload); + } catch (writeError) { + process.removeListener('SIGINT', protectAskWait); + console.error(`botmux ask: stdout 输出失败：${writeError instanceof Error ? writeError.message : String(writeError)}`); + process.exit(1); + } + } try { await postAsk({ ...body, acknowledge: true }); } catch { /* durable receipt + same-session recovery handles ambiguous delivery */ } process.removeListener('SIGINT', protectAskWait); @@ -15272,7 +15304,10 @@ async function cmdHook(cliId: string): Promise { const env = process.env as Record; const result = await runHook(payload, env, postAsk, cliId); if (result.stdout) { - console.log(result.stdout); + // The directive can embed the user's free-form answer (unbounded size); + // wait for the pipe write to complete before the explicit exit truncates it. + // A failed write still exits 0: hook mode must never block the CLI. + try { await writeAndFlush(process.stdout, result.stdout + '\n'); } catch { /* passthrough */ } } process.exit(0); } diff --git a/src/cli/stdout-flush.ts b/src/cli/stdout-flush.ts new file mode 100644 index 0000000000..7574aff87c --- /dev/null +++ b/src/cli/stdout-flush.ts @@ -0,0 +1,16 @@ +/** A minimal writable surface whose completion callback means the chunk has + * reached the underlying stream. `write()` returning false is only a + * backpressure signal; callers that must not exit early must await callback. */ +export interface CompletionWritable { + write(chunk: string, callback: (error?: Error | null) => void): boolean; +} + +/** Write one complete payload before allowing a child process to exit. */ +export function writeAndFlush(stream: CompletionWritable, chunk: string): Promise { + return new Promise((resolve, reject) => { + stream.write(chunk, error => { + if (error) reject(error); + else resolve(); + }); + }); +} diff --git a/test/ask-cli.test.ts b/test/ask-cli.test.ts index 758924abb4..1520f93d8c 100644 --- a/test/ask-cli.test.ts +++ b/test/ask-cli.test.ts @@ -286,6 +286,52 @@ describe('botmux ask — CLI boundary', () => { }); } }); + + it('--json 完整输出超出 pipe 容量的文字作答', async () => { + const dataDir = mkdtempSync(join(tmpdir(), 'botmux-ask-cli-')); + tempDirs.push(dataDir); + const comment = 'x'.repeat(500_000); + + const server = createServer((req, res) => { + req.resume(); + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ + kind: 'answered', + answers: [[]], + by: 'ou_test', + comment, + timedOut: false, + })); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + + try { + const port = (server.address() as AddressInfo).port; + const registryDir = join(dataDir, 'dashboard-daemons'); + mkdirSync(registryDir, { recursive: true }); + writeFileSync( + join(registryDir, 'cli_test.json'), + JSON.stringify({ larkAppId: 'cli_test', ipcPort: port, lastHeartbeat: Date.now() }), + ); + + const result = await runAsk(dataDir, [ + 'ask', 'buttons', '--json', '--options', 'yes,no', '请作答', + ]); + expect(result.status).toBe(0); + expect(Buffer.byteLength(result.stdout)).toBeGreaterThan(400_000); + expect(JSON.parse(result.stdout)).toMatchObject({ + selected: null, + answers: [[]], + by: 'ou_test', + comment, + timedOut: false, + }); + } finally { + await new Promise((resolve, reject) => { + server.close((err) => err ? reject(err) : resolve()); + }); + } + }, 30_000); }); describe('question file CLI', () => { diff --git a/test/pm2-readonly-jlist.test.ts b/test/pm2-readonly-jlist.test.ts new file mode 100644 index 0000000000..5bbd2c86f5 --- /dev/null +++ b/test/pm2-readonly-jlist.test.ts @@ -0,0 +1,79 @@ +import { spawnSync } from 'node:child_process'; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { dirname, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { afterEach, describe, expect, it } from 'vitest'; +import { nodeTsRunnerPrefix, resolveNodeExecutable } from './helpers/ts-runner.js'; + +const PKG_ROOT = dirname(dirname(fileURLToPath(import.meta.url))); +const PM2_PATH = join(PKG_ROOT, 'node_modules', 'pm2', 'bin', 'pm2'); +const READONLY_CLIENT = join(PKG_ROOT, 'src', 'cli', 'pm2-readonly-client.ts'); +const homes: string[] = []; + +// PM2 is a Node program: under `bun test` process.execPath is bun, so resolve a +// real Node binary instead (see test/helpers/ts-runner.ts). +const NODE_BIN = resolveNodeExecutable(); +function runPm2(args: readonly string[], options: Parameters[2]) { + if (!NODE_BIN) throw new Error('node not found on PATH; PM2 needs a real Node executable'); + return spawnSync(NODE_BIN, [PM2_PATH, ...args], options); +} + +function tempHome(): string { + const home = mkdtempSync(join(tmpdir(), 'botmux-pm2-readonly-')); + homes.push(home); + return home; +} + +afterEach(() => { + for (const home of homes.splice(0)) { + const pm2Home = join(home, '.botmux', 'pm2'); + runPm2(['kill'], { + env: { ...process.env, PM2_HOME: pm2Home }, + stdio: 'ignore', + timeout: 10_000, + }); + rmSync(home, { recursive: true, force: true }); + } +}); + +describe('PM2 read-only jlist stdout integrity', () => { + it('returns a large jlist without truncation', () => { + const home = tempHome(); + const pm2Home = join(home, '.botmux', 'pm2'); + mkdirSync(pm2Home, { recursive: true }); + expect(runPm2(['status'], { + env: { ...process.env, PM2_HOME: pm2Home }, + stdio: 'ignore', + timeout: 10_000, + }).status).toBe(0); + + const idleScript = join(home, 'idle.js'); + writeFileSync(idleScript, 'setInterval(() => {}, 1000);\n'); + for (let i = 0; i < 6; i++) { + const started = runPm2(['start', idleScript, '--name', `flush-probe-${i}`], { + env: { ...process.env, PM2_HOME: pm2Home, BOTMUX_FLUSH_PROBE: 'x'.repeat(30_000) }, + stdio: 'ignore', + timeout: 20_000, + }); + expect(started.status).toBe(0); + } + + // Same shape as the desktop observer (src/desktop/main/pm2-apps.ts): the + // helper's stdout is a pipe the parent drains, and the helper exits + // explicitly once the JSON is written. + const { command, prefixArgs } = nodeTsRunnerPrefix(); + const jlist = spawnSync(command, [...prefixArgs, READONLY_CLIENT, 'jlist'], { + cwd: PKG_ROOT, + env: { ...process.env, PM2_HOME: pm2Home }, + encoding: 'utf8', + maxBuffer: 64 * 1024 * 1024, + timeout: 20_000, + }); + expect(jlist.status, jlist.stderr).toBe(0); + const stdout = jlist.stdout; + // Keep the fixture well past the partial-flush range observed with a pipe. + expect(Buffer.byteLength(stdout)).toBeGreaterThan(200_000); + expect(JSON.parse(stdout)).toHaveLength(6); + }, 90_000); +}); diff --git a/test/stdout-flush.test.ts b/test/stdout-flush.test.ts new file mode 100644 index 0000000000..20c22acfc4 --- /dev/null +++ b/test/stdout-flush.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from 'vitest'; +import { spawnTsEvalWithRepoImports } from './helpers/ts-runner.js'; + +/** + * Child-process regression for `writeAndFlush`: a payload far beyond the pipe + * buffer (64 KiB) and the ~146 KiB observed partial-flush ceiling must reach the + * parent intact even though the child calls process.exit() right after. + */ +function runChild(bytes: number): Promise<{ code: number | null; stdout: Buffer }> { + return new Promise((resolve, reject) => { + const child = spawnTsEvalWithRepoImports(` + import { writeAndFlush } from './src/cli/stdout-flush.js'; + const payload = 'x'.repeat(${bytes}); + await writeAndFlush(process.stdout, payload); + process.exit(0); + `, { cwd: process.cwd(), stdio: ['ignore', 'pipe', 'inherit'] }); + const chunks: Buffer[] = []; + // Drain slowly so the child really hits backpressure before exiting. + child.stdout!.pause(); + setTimeout(() => child.stdout!.resume(), 300); + child.stdout!.on('data', (c: Buffer) => chunks.push(c)); + child.on('error', reject); + child.on('close', code => resolve({ code, stdout: Buffer.concat(chunks) })); + }); +} + +describe('writeAndFlush over a real pipe', () => { + it('delivers a 2 MiB payload completely before an explicit exit', async () => { + const size = 2 * 1024 * 1024; + const { code, stdout } = await runChild(size); + expect(code).toBe(0); + expect(stdout.length).toBe(size); + }, 30_000); +});