Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 47 additions & 12 deletions src/cli.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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));
Expand All @@ -8886,6 +8885,29 @@ async function relaySend(
process.exit(1);
}

/**
* Mirror a host relay response (`<id>.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<never> {
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
Expand Down Expand Up @@ -8965,9 +8987,7 @@ async function relayDispatch(rest: string[], relayDir: string): Promise<void> {
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));
Expand Down Expand Up @@ -14906,6 +14926,7 @@ async function cmdAsk(sub: string, rest: string[]): Promise<void> {
// result.kind==='answered' 时用 toLegacySelected 取回旧的 string(单问单选)
const selected = toLegacySelected(result);

let stdoutPayload: string | undefined;
if (useJson) {
const out: AskJsonOutput = {
// `selected` 是「单问单选」的向后兼容值(= toLegacySelected 的形状判据:
Expand All @@ -14918,7 +14939,7 @@ async function cmdAsk(sub: string, rest: string[]): Promise<void> {
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。
//
Expand All @@ -14935,12 +14956,23 @@ async function cmdAsk(sub: string, rest: string[]): Promise<void> {
// 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<void>(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);
Expand Down Expand Up @@ -15272,7 +15304,10 @@ async function cmdHook(cliId: string): Promise<void> {
const env = process.env as Record<string, string | undefined>;
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);
}
Expand Down
16 changes: 16 additions & 0 deletions src/cli/stdout-flush.ts
Original file line number Diff line number Diff line change
@@ -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<void> {
return new Promise((resolve, reject) => {
stream.write(chunk, error => {
if (error) reject(error);
else resolve();
});
});
}
46 changes: 46 additions & 0 deletions test/ask-cli.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>((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<void>((resolve, reject) => {
server.close((err) => err ? reject(err) : resolve());
});
}
}, 30_000);
});

describe('question file CLI', () => {
Expand Down
79 changes: 79 additions & 0 deletions test/pm2-readonly-jlist.test.ts
Original file line number Diff line number Diff line change
@@ -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<typeof spawnSync>[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);
});
34 changes: 34 additions & 0 deletions test/stdout-flush.test.ts
Original file line number Diff line number Diff line change
@@ -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);
});