mirror of
https://github.com/QwenLM/qwen-code.git
synced 2025-12-19 09:33:53 +00:00
Gal Zahavi
GitHub
@@ -4,35 +4,29 @@
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
*/
|
||||
|
||||
import * as pty from '@lydell/node-pty';
|
||||
import { spawn } from 'child_process';
|
||||
import { TextDecoder } from 'util';
|
||||
import os from 'os';
|
||||
import stripAnsi from 'strip-ansi';
|
||||
import { getCachedEncodingForBuffer } from '../utils/systemEncoding.js';
|
||||
import { isBinary } from '../utils/textUtils.js';
|
||||
import pkg from '@xterm/headless';
|
||||
const { Terminal } = pkg;
|
||||
|
||||
// @ts-expect-error getFullText is not a public API.
|
||||
const getFullText = (terminal: Terminal) => {
|
||||
const buffer = terminal.buffer.active;
|
||||
const lines: string[] = [];
|
||||
for (let i = 0; i < buffer.length; i++) {
|
||||
const line = buffer.getLine(i);
|
||||
lines.push(line ? line.translateToString(true) : '');
|
||||
}
|
||||
return lines.join('\n').trim();
|
||||
};
|
||||
const SIGKILL_TIMEOUT_MS = 200;
|
||||
|
||||
/** A structured result from a shell command execution. */
|
||||
export interface ShellExecutionResult {
|
||||
/** The raw, unprocessed output buffer. */
|
||||
rawOutput: Buffer;
|
||||
/** The combined, decoded output as a string. */
|
||||
/** The combined, decoded stdout and stderr as a string. */
|
||||
output: string;
|
||||
/** The decoded stdout as a string. */
|
||||
stdout: string;
|
||||
/** The decoded stderr as a string. */
|
||||
stderr: string;
|
||||
/** The process exit code, or null if terminated by a signal. */
|
||||
exitCode: number | null;
|
||||
/** The signal that terminated the process, if any. */
|
||||
signal: number | null;
|
||||
signal: NodeJS.Signals | null;
|
||||
/** An error object if the process failed to spawn. */
|
||||
error: Error | null;
|
||||
/** A boolean indicating if the command was aborted by the user. */
|
||||
@@ -56,6 +50,8 @@ export type ShellOutputEvent =
|
||||
| {
|
||||
/** The event contains a chunk of output data. */
|
||||
type: 'data';
|
||||
/** The stream from which the data originated. */
|
||||
stream: 'stdout' | 'stderr';
|
||||
/** The decoded string chunk. */
|
||||
chunk: string;
|
||||
}
|
||||
@@ -77,7 +73,7 @@ export type ShellOutputEvent =
|
||||
*/
|
||||
export class ShellExecutionService {
|
||||
/**
|
||||
* Executes a shell command using `node-pty`, capturing all output and lifecycle events.
|
||||
* Executes a shell command using `spawn`, capturing all output and lifecycle events.
|
||||
*
|
||||
* @param commandToExecute The exact command string to run.
|
||||
* @param cwd The working directory to execute the command in.
|
||||
@@ -91,150 +87,167 @@ export class ShellExecutionService {
|
||||
cwd: string,
|
||||
onOutputEvent: (event: ShellOutputEvent) => void,
|
||||
abortSignal: AbortSignal,
|
||||
terminalColumns?: number,
|
||||
terminalRows?: number,
|
||||
): ShellExecutionHandle {
|
||||
const isWindows = os.platform() === 'win32';
|
||||
const shell = isWindows ? 'cmd.exe' : 'bash';
|
||||
const args = isWindows
|
||||
? ['/c', commandToExecute]
|
||||
: ['-c', commandToExecute];
|
||||
|
||||
let ptyProcess;
|
||||
try {
|
||||
ptyProcess = pty.spawn(shell, args, {
|
||||
cwd,
|
||||
name: 'xterm-color',
|
||||
cols: terminalColumns ?? 200,
|
||||
rows: terminalRows ?? 20,
|
||||
env: {
|
||||
...process.env,
|
||||
GEMINI_CLI: '1',
|
||||
},
|
||||
handleFlowControl: true,
|
||||
});
|
||||
} catch (e) {
|
||||
const error = e as Error;
|
||||
return {
|
||||
pid: undefined,
|
||||
result: Promise.resolve({
|
||||
rawOutput: Buffer.from(''),
|
||||
output: '',
|
||||
exitCode: 1,
|
||||
signal: null,
|
||||
error,
|
||||
aborted: false,
|
||||
pid: undefined,
|
||||
}),
|
||||
};
|
||||
}
|
||||
const child = spawn(commandToExecute, [], {
|
||||
cwd,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
// Use bash unless in Windows (since it doesn't support bash).
|
||||
// For windows, just use the default.
|
||||
shell: isWindows ? true : 'bash',
|
||||
// Use process groups on non-Windows for robust killing.
|
||||
// Windows process termination is handled by `taskkill /t`.
|
||||
detached: !isWindows,
|
||||
env: {
|
||||
...process.env,
|
||||
GEMINI_CLI: '1',
|
||||
},
|
||||
});
|
||||
|
||||
const result = new Promise<ShellExecutionResult>((resolve) => {
|
||||
const headlessTerminal = new Terminal({
|
||||
allowProposedApi: true,
|
||||
cols: terminalColumns ?? 200,
|
||||
rows: terminalRows ?? 20,
|
||||
});
|
||||
let processingChain = Promise.resolve();
|
||||
let decoder: TextDecoder | null = null;
|
||||
let output = '';
|
||||
// Use decoders to handle multi-byte characters safely (for streaming output).
|
||||
let stdoutDecoder: TextDecoder | null = null;
|
||||
let stderrDecoder: TextDecoder | null = null;
|
||||
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
const outputChunks: Buffer[] = [];
|
||||
const error: Error | null = null;
|
||||
let error: Error | null = null;
|
||||
let exited = false;
|
||||
|
||||
let isStreamingRawContent = true;
|
||||
const MAX_SNIFF_SIZE = 4096;
|
||||
let sniffedBytes = 0;
|
||||
|
||||
const handleOutput = (data: Buffer) => {
|
||||
// NOTE: The migration from `child_process` to `node-pty` means we
|
||||
// no longer have separate `stdout` and `stderr` streams. The `data`
|
||||
// buffer contains the merged output. If a drop in LLM quality is
|
||||
// observed after this change, we may need to revisit this and
|
||||
// explore ways to re-introduce that distinction.
|
||||
processingChain = processingChain.then(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
if (!decoder) {
|
||||
const encoding = getCachedEncodingForBuffer(data);
|
||||
try {
|
||||
decoder = new TextDecoder(encoding);
|
||||
} catch {
|
||||
decoder = new TextDecoder('utf-8');
|
||||
}
|
||||
}
|
||||
const handleOutput = (data: Buffer, stream: 'stdout' | 'stderr') => {
|
||||
if (!stdoutDecoder || !stderrDecoder) {
|
||||
const encoding = getCachedEncodingForBuffer(data);
|
||||
try {
|
||||
stdoutDecoder = new TextDecoder(encoding);
|
||||
stderrDecoder = new TextDecoder(encoding);
|
||||
} catch {
|
||||
// If the encoding is not supported, fall back to utf-8.
|
||||
// This can happen on some platforms for certain encodings like 'utf-32le'.
|
||||
stdoutDecoder = new TextDecoder('utf-8');
|
||||
stderrDecoder = new TextDecoder('utf-8');
|
||||
}
|
||||
}
|
||||
|
||||
outputChunks.push(data);
|
||||
outputChunks.push(data);
|
||||
|
||||
// First, check if we need to switch to binary mode.
|
||||
if (isStreamingRawContent && sniffedBytes < MAX_SNIFF_SIZE) {
|
||||
const sniffBuffer = Buffer.concat(outputChunks.slice(0, 20));
|
||||
sniffedBytes = sniffBuffer.length;
|
||||
// Binary detection logic. This only runs until we've made a determination.
|
||||
if (isStreamingRawContent && sniffedBytes < MAX_SNIFF_SIZE) {
|
||||
const sniffBuffer = Buffer.concat(outputChunks.slice(0, 20));
|
||||
sniffedBytes = sniffBuffer.length;
|
||||
|
||||
if (isBinary(sniffBuffer)) {
|
||||
isStreamingRawContent = false;
|
||||
onOutputEvent({ type: 'binary_detected' });
|
||||
}
|
||||
}
|
||||
if (isBinary(sniffBuffer)) {
|
||||
// Change state to stop streaming raw content.
|
||||
isStreamingRawContent = false;
|
||||
onOutputEvent({ type: 'binary_detected' });
|
||||
}
|
||||
}
|
||||
|
||||
// Now, based on the *current* state, either process as text or binary.
|
||||
if (isStreamingRawContent) {
|
||||
const decodedChunk = decoder.decode(data, { stream: true });
|
||||
headlessTerminal.write(decodedChunk, () => {
|
||||
const newStrippedOutput = getFullText(headlessTerminal);
|
||||
output = newStrippedOutput;
|
||||
onOutputEvent({ type: 'data', chunk: newStrippedOutput });
|
||||
resolve();
|
||||
});
|
||||
} else {
|
||||
// Once in binary mode, we only emit progress events.
|
||||
const totalBytes = outputChunks.reduce(
|
||||
(sum, chunk) => sum + chunk.length,
|
||||
0,
|
||||
);
|
||||
onOutputEvent({
|
||||
type: 'binary_progress',
|
||||
bytesReceived: totalBytes,
|
||||
});
|
||||
resolve();
|
||||
}
|
||||
}),
|
||||
);
|
||||
const decodedChunk =
|
||||
stream === 'stdout'
|
||||
? stdoutDecoder.decode(data, { stream: true })
|
||||
: stderrDecoder.decode(data, { stream: true });
|
||||
const strippedChunk = stripAnsi(decodedChunk);
|
||||
|
||||
if (stream === 'stdout') {
|
||||
stdout += strippedChunk;
|
||||
} else {
|
||||
stderr += strippedChunk;
|
||||
}
|
||||
|
||||
if (isStreamingRawContent) {
|
||||
onOutputEvent({ type: 'data', stream, chunk: strippedChunk });
|
||||
} else {
|
||||
const totalBytes = outputChunks.reduce(
|
||||
(sum, chunk) => sum + chunk.length,
|
||||
0,
|
||||
);
|
||||
onOutputEvent({ type: 'binary_progress', bytesReceived: totalBytes });
|
||||
}
|
||||
};
|
||||
|
||||
ptyProcess.onData((data) => {
|
||||
const bufferData = Buffer.from(data, 'utf-8');
|
||||
handleOutput(bufferData);
|
||||
});
|
||||
|
||||
ptyProcess.onExit(({ exitCode, signal }) => {
|
||||
exited = true;
|
||||
abortSignal.removeEventListener('abort', abortHandler);
|
||||
|
||||
processingChain.then(() => {
|
||||
const finalBuffer = Buffer.concat(outputChunks);
|
||||
|
||||
resolve({
|
||||
rawOutput: finalBuffer,
|
||||
output,
|
||||
exitCode,
|
||||
signal: signal ?? null,
|
||||
error,
|
||||
aborted: abortSignal.aborted,
|
||||
pid: ptyProcess.pid,
|
||||
});
|
||||
child.stdout.on('data', (data) => handleOutput(data, 'stdout'));
|
||||
child.stderr.on('data', (data) => handleOutput(data, 'stderr'));
|
||||
child.on('error', (err) => {
|
||||
const { stdout, stderr, finalBuffer } = cleanup();
|
||||
error = err;
|
||||
resolve({
|
||||
error,
|
||||
stdout,
|
||||
stderr,
|
||||
rawOutput: finalBuffer,
|
||||
output: stdout + (stderr ? `\n${stderr}` : ''),
|
||||
exitCode: 1,
|
||||
signal: null,
|
||||
aborted: false,
|
||||
pid: child.pid,
|
||||
});
|
||||
});
|
||||
|
||||
const abortHandler = async () => {
|
||||
if (ptyProcess.pid && !exited) {
|
||||
ptyProcess.kill('SIGHUP');
|
||||
if (child.pid && !exited) {
|
||||
if (isWindows) {
|
||||
spawn('taskkill', ['/pid', child.pid.toString(), '/f', '/t']);
|
||||
} else {
|
||||
try {
|
||||
// Kill the entire process group (negative PID).
|
||||
// SIGTERM first, then SIGKILL if it doesn't die.
|
||||
process.kill(-child.pid, 'SIGTERM');
|
||||
await new Promise((res) => setTimeout(res, SIGKILL_TIMEOUT_MS));
|
||||
if (!exited) {
|
||||
process.kill(-child.pid, 'SIGKILL');
|
||||
}
|
||||
} catch (_e) {
|
||||
// Fall back to killing just the main process if group kill fails.
|
||||
if (!exited) child.kill('SIGKILL');
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
abortSignal.addEventListener('abort', abortHandler, { once: true });
|
||||
|
||||
child.on('exit', (code: number, signal: NodeJS.Signals) => {
|
||||
const { stdout, stderr, finalBuffer } = cleanup();
|
||||
|
||||
resolve({
|
||||
rawOutput: finalBuffer,
|
||||
output: stdout + (stderr ? `\n${stderr}` : ''),
|
||||
stdout,
|
||||
stderr,
|
||||
exitCode: code,
|
||||
signal,
|
||||
error,
|
||||
aborted: abortSignal.aborted,
|
||||
pid: child.pid,
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Cleans up a process (and it's accompanying state) that is exiting or
|
||||
* erroring and returns output formatted output buffers and strings
|
||||
*/
|
||||
function cleanup() {
|
||||
exited = true;
|
||||
abortSignal.removeEventListener('abort', abortHandler);
|
||||
if (stdoutDecoder) {
|
||||
stdout += stripAnsi(stdoutDecoder.decode());
|
||||
}
|
||||
if (stderrDecoder) {
|
||||
stderr += stripAnsi(stderrDecoder.decode());
|
||||
}
|
||||
|
||||
const finalBuffer = Buffer.concat(outputChunks);
|
||||
|
||||
return { stdout, stderr, finalBuffer };
|
||||
}
|
||||
});
|
||||
|
||||
return { pid: ptyProcess.pid, result };
|
||||
return { pid: child.pid, result };
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user