767 lines
24 KiB
TypeScript
767 lines
24 KiB
TypeScript
import {
|
|
execFile,
|
|
spawn,
|
|
type ChildProcessWithoutNullStreams,
|
|
} from 'node:child_process';
|
|
import { realpathSync } from 'node:fs';
|
|
import { platform } from 'node:os';
|
|
import { Writable } from 'node:stream';
|
|
import { logger } from '../../utils/logger';
|
|
import { PiProcessError } from './process-errors';
|
|
import {
|
|
PiRpcClient,
|
|
type PiRpcCommand,
|
|
type PiRpcEvent,
|
|
type PiRpcRequestOptions,
|
|
type PiRpcResponse,
|
|
} from './rpc-client';
|
|
import { StrictLfJsonlFramer } from './rpc-framer';
|
|
import {
|
|
buildPiWorkerEnvironment,
|
|
DEFAULT_PI_RPC_TOOLS,
|
|
sanitizePiDiagnostic,
|
|
type PiWorkerLifecycleEvent,
|
|
type PiWorkerProcessOptions,
|
|
type PiWorkerProofFailure,
|
|
type PiWorkerStopReason,
|
|
type PiWorkerStopResult,
|
|
} from './worker-process';
|
|
|
|
const SERVER_CHANNEL = '@makelore/server';
|
|
const SERVER_START_TIMEOUT_MS = 60_000;
|
|
const SERVER_CONTROL_TIMEOUT_MS = 30_000;
|
|
const SERVER_SHUTDOWN_GRACE_MS = 3_000;
|
|
const SERVER_DIAGNOSTIC_BYTES = 16_000;
|
|
|
|
interface AgentServerEnvelope {
|
|
channel: string;
|
|
payload: unknown;
|
|
}
|
|
|
|
interface AgentServerThreadOptions {
|
|
cwd: string;
|
|
configDir: string;
|
|
sessionDir: string;
|
|
tools?: readonly string[];
|
|
additionalArgs: readonly string[];
|
|
env: Record<string, string>;
|
|
conversationId?: string;
|
|
workerGeneration: number;
|
|
}
|
|
|
|
export interface PiAgentServerProcessOptions {
|
|
executablePath: string;
|
|
serverPath: string;
|
|
runtimeRoot: string;
|
|
configDir: string;
|
|
shutdownGraceMs?: number;
|
|
diagnosticBytes?: number;
|
|
}
|
|
|
|
interface ChannelRecord {
|
|
rpc: PiRpcClient;
|
|
worker?: PiAgentServerWorkerProcess;
|
|
}
|
|
|
|
function positiveInteger(value: number | undefined, fallback: number, name: string): number {
|
|
const resolved = value ?? fallback;
|
|
if (!Number.isSafeInteger(resolved) || resolved <= 0) {
|
|
throw new Error(`${name} must be a positive safe integer`);
|
|
}
|
|
return resolved;
|
|
}
|
|
|
|
function boundedUtf8Tail(source: string, maxBytes: number): string {
|
|
const bytes = Buffer.from(source, 'utf8');
|
|
if (bytes.length <= maxBytes) return source;
|
|
const decoder = new TextDecoder('utf-8', { fatal: true });
|
|
for (let start = bytes.length - maxBytes; start < bytes.length; start += 1) {
|
|
try {
|
|
return decoder.decode(bytes.subarray(start));
|
|
} catch {
|
|
// Advance to the next UTF-8 code point boundary.
|
|
}
|
|
}
|
|
return '';
|
|
}
|
|
|
|
function envelopeValue(value: unknown): value is AgentServerEnvelope {
|
|
return Boolean(value)
|
|
&& typeof value === 'object'
|
|
&& !Array.isArray(value)
|
|
&& typeof (value as { channel?: unknown }).channel === 'string'
|
|
&& 'payload' in value;
|
|
}
|
|
|
|
function stringEnvironment(value: NodeJS.ProcessEnv | undefined): Record<string, string> {
|
|
return Object.fromEntries(Object.entries(value ?? {}).flatMap(([key, candidate]) => (
|
|
typeof candidate === 'string' ? [[key, candidate]] : []
|
|
)));
|
|
}
|
|
|
|
function diagnosticPathAliases(value: string | undefined): string[] {
|
|
if (!value) return [];
|
|
try {
|
|
const canonical = realpathSync.native(value);
|
|
return canonical === value ? [value] : [value, canonical];
|
|
} catch {
|
|
return [value];
|
|
}
|
|
}
|
|
|
|
function diagnosticSensitiveValues(options: PiWorkerProcessOptions): string[] {
|
|
return [...new Set([
|
|
...(options.sensitiveValues ?? []),
|
|
...[
|
|
options.cwd,
|
|
options.cliPath,
|
|
options.configDir,
|
|
options.sessionDir,
|
|
options.env?.HOME,
|
|
options.env?.USERPROFILE,
|
|
options.env?.APPDATA,
|
|
options.env?.LOCALAPPDATA,
|
|
].flatMap(diagnosticPathAliases),
|
|
].filter((value): value is string => Boolean(value)))];
|
|
}
|
|
|
|
function runExecutable(executable: string, args: readonly string[]): Promise<void> {
|
|
return new Promise((resolve, reject) => {
|
|
execFile(executable, [...args], { windowsHide: true }, (error) => {
|
|
if (error) reject(error);
|
|
else resolve();
|
|
});
|
|
});
|
|
}
|
|
|
|
class AgentServerChannelWritable extends Writable {
|
|
constructor(
|
|
private readonly channel: string,
|
|
private readonly sendEnvelope: (envelope: AgentServerEnvelope) => Promise<void>,
|
|
) {
|
|
super();
|
|
}
|
|
|
|
override _write(
|
|
chunk: Buffer | string,
|
|
_encoding: BufferEncoding,
|
|
callback: (error?: Error | null) => void,
|
|
): void {
|
|
try {
|
|
const source = Buffer.isBuffer(chunk) ? chunk.toString('utf8') : chunk;
|
|
const line = source.endsWith('\n') ? source.slice(0, -1) : source;
|
|
if (!line || line.includes('\n')) throw new Error('Agent Server channel write must contain one JSONL record');
|
|
const payload = JSON.parse(line) as unknown;
|
|
void this.sendEnvelope({ channel: this.channel, payload }).then(
|
|
() => callback(),
|
|
(error: unknown) => callback(error instanceof Error ? error : new Error(String(error))),
|
|
);
|
|
} catch (error) {
|
|
callback(error instanceof Error ? error : new Error(String(error)));
|
|
}
|
|
}
|
|
}
|
|
|
|
export class PiAgentServerWorkerProcess {
|
|
readonly generation: number;
|
|
private readonly invalidationListeners = new Set<(error: PiProcessError) => void>();
|
|
private rpc: PiRpcClient | null = null;
|
|
private invalidation: PiProcessError | null = null;
|
|
private stopFlight: Promise<PiWorkerStopResult> | null = null;
|
|
private diagnostic = '';
|
|
private failureLifecycleRecorded = false;
|
|
|
|
constructor(
|
|
private readonly server: PiAgentServerProcess,
|
|
readonly options: PiWorkerProcessOptions,
|
|
) {
|
|
this.generation = positiveInteger(options.workerGeneration, 1, 'workerGeneration');
|
|
}
|
|
|
|
get threadId(): string {
|
|
return `${this.options.conversationId ?? 'conversation'}:${this.generation}`;
|
|
}
|
|
|
|
get processId(): number | undefined {
|
|
return this.server.processId;
|
|
}
|
|
|
|
get isRunning(): boolean {
|
|
return Boolean(this.rpc) && !this.invalidation && this.server.isRunning;
|
|
}
|
|
|
|
get stderrDiagnostic(): string {
|
|
return this.diagnostic;
|
|
}
|
|
|
|
async start(): Promise<this> {
|
|
if (this.rpc) throw new Error('Pi Agent Server thread already started');
|
|
this.rpc = await this.server.openThread(this);
|
|
return this;
|
|
}
|
|
|
|
request<T = unknown>(
|
|
command: PiRpcCommand,
|
|
options?: PiRpcRequestOptions,
|
|
): Promise<PiRpcResponse<T>> {
|
|
if (this.invalidation) return Promise.reject(this.invalidation);
|
|
if (!this.rpc) {
|
|
return Promise.reject(new PiProcessError(
|
|
'PI_WORKER_START_FAILED',
|
|
'Pi Agent Server thread has not started',
|
|
{ generation: this.generation },
|
|
));
|
|
}
|
|
return this.rpc.request<T>(command, options);
|
|
}
|
|
|
|
send(command: PiRpcCommand): Promise<void> {
|
|
if (this.invalidation) return Promise.reject(this.invalidation);
|
|
if (!this.rpc) {
|
|
return Promise.reject(new PiProcessError(
|
|
'PI_WORKER_START_FAILED',
|
|
'Pi Agent Server thread has not started',
|
|
{ generation: this.generation },
|
|
));
|
|
}
|
|
return this.rpc.send(command);
|
|
}
|
|
|
|
subscribe(listener: (event: PiRpcEvent) => void): () => void {
|
|
if (!this.rpc) throw new Error('Pi Agent Server thread has not started');
|
|
return this.rpc.subscribe(listener);
|
|
}
|
|
|
|
subscribeInvalidation(listener: (error: PiProcessError) => void): () => void {
|
|
this.invalidationListeners.add(listener);
|
|
return () => this.invalidationListeners.delete(listener);
|
|
}
|
|
|
|
stop(reason: PiWorkerStopReason): Promise<PiWorkerStopResult> {
|
|
if (!this.stopFlight) this.stopFlight = this.performStop(reason);
|
|
return this.stopFlight;
|
|
}
|
|
|
|
async injectFailureForProof(failure: PiWorkerProofFailure): Promise<void> {
|
|
this.appendDiagnostic(
|
|
'[release-proof] worker failure Authorization: Bearer packaged-proof-secret\n',
|
|
);
|
|
await this.server.invalidateThreadForProof(this, failure);
|
|
}
|
|
|
|
delayNextResponseForProof(commandType: string, delayMs: number): void {
|
|
if (!this.rpc) throw new Error('Pi Agent Server thread has not started');
|
|
this.rpc.delayNextResponseForProof(commandType, delayMs);
|
|
}
|
|
|
|
invalidate(error: PiProcessError): void {
|
|
if (this.invalidation) return;
|
|
this.invalidation = error;
|
|
if (!this.failureLifecycleRecorded
|
|
&& (error.code === 'PI_RPC_PROTOCOL_ERROR' || error.code === 'PI_RPC_EXITED')) {
|
|
this.failureLifecycleRecorded = true;
|
|
this.recordLifecycle({
|
|
classification: error.code === 'PI_RPC_PROTOCOL_ERROR'
|
|
? 'protocol_invalidation'
|
|
: 'unexpected_exit',
|
|
stage: error.code === 'PI_RPC_PROTOCOL_ERROR' ? 'protocol' : 'close',
|
|
generation: this.generation,
|
|
code: error.code,
|
|
exitCode: null,
|
|
signal: null,
|
|
...(this.diagnostic ? { diagnostic: this.diagnostic } : {}),
|
|
});
|
|
}
|
|
this.rpc?.invalidate(error);
|
|
for (const listener of this.invalidationListeners) {
|
|
try {
|
|
listener(error);
|
|
} catch {
|
|
// Thread invalidation observers must not affect sibling threads.
|
|
}
|
|
}
|
|
}
|
|
|
|
private async performStop(reason: PiWorkerStopReason): Promise<PiWorkerStopResult> {
|
|
const classification = this.invalidation?.code === 'PI_RPC_PROTOCOL_ERROR'
|
|
? 'protocol_invalidation'
|
|
: this.invalidation?.code === 'PI_RPC_EXITED'
|
|
? 'unexpected_exit'
|
|
: 'intentional_stop';
|
|
const code = this.invalidation?.code ?? 'PI_WORKER_STOPPED';
|
|
this.recordLifecycle({
|
|
classification,
|
|
stage: 'stop_requested',
|
|
generation: this.generation,
|
|
code,
|
|
reason,
|
|
exitCode: null,
|
|
signal: null,
|
|
...(this.diagnostic ? { diagnostic: this.diagnostic } : {}),
|
|
});
|
|
try {
|
|
const result = await this.server.closeThread(this, reason);
|
|
this.recordLifecycle({
|
|
classification,
|
|
stage: 'stop_completed',
|
|
generation: this.generation,
|
|
code,
|
|
reason,
|
|
exitCode: result.code,
|
|
signal: result.signal,
|
|
...(this.diagnostic ? { diagnostic: this.diagnostic } : {}),
|
|
});
|
|
return result;
|
|
} catch (error) {
|
|
this.recordLifecycle({
|
|
classification,
|
|
stage: 'stop_failed',
|
|
generation: this.generation,
|
|
code: 'PI_WORKER_STOP_FAILED',
|
|
reason,
|
|
exitCode: null,
|
|
signal: null,
|
|
...(this.diagnostic ? { diagnostic: this.diagnostic } : {}),
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private appendDiagnostic(source: string): void {
|
|
this.diagnostic = boundedUtf8Tail(
|
|
`${this.diagnostic}${sanitizePiDiagnostic(source, diagnosticSensitiveValues(this.options))}`,
|
|
SERVER_DIAGNOSTIC_BYTES,
|
|
);
|
|
}
|
|
|
|
private recordLifecycle(event: PiWorkerLifecycleEvent): void {
|
|
const diagnostic = event.diagnostic
|
|
? boundedUtf8Tail(
|
|
sanitizePiDiagnostic(event.diagnostic, diagnosticSensitiveValues(this.options)),
|
|
SERVER_DIAGNOSTIC_BYTES,
|
|
)
|
|
: undefined;
|
|
const safeEvent: PiWorkerLifecycleEvent = {
|
|
...event,
|
|
...(this.options.conversationId ? { conversationId: this.options.conversationId } : {}),
|
|
...(diagnostic ? { diagnostic } : {}),
|
|
};
|
|
try {
|
|
this.options.onLifecycleEvent?.(structuredClone(safeEvent));
|
|
} catch {
|
|
// Lifecycle observers must not affect logical thread cleanup.
|
|
}
|
|
logger.warn('[PiWorkerLifecycle]', safeEvent);
|
|
}
|
|
}
|
|
|
|
export class PiAgentServerProcess {
|
|
private readonly shutdownGraceMs: number;
|
|
private readonly diagnosticBytes: number;
|
|
private readonly channels = new Map<string, ChannelRecord>();
|
|
private readonly retiredChannels = new Set<string>();
|
|
private readonly sensitiveValues = new Set<string>();
|
|
private child: ChildProcessWithoutNullStreams | null = null;
|
|
private control: PiRpcClient | null = null;
|
|
private startFlight: Promise<void> | null = null;
|
|
private stopFlight: Promise<void> | null = null;
|
|
private exitResult: Promise<{ code: number | null; signal: NodeJS.Signals | null }> | null = null;
|
|
private diagnostic = '';
|
|
private generation = 0;
|
|
private stopping = false;
|
|
private failure: PiProcessError | null = null;
|
|
|
|
constructor(private readonly options: PiAgentServerProcessOptions) {
|
|
this.shutdownGraceMs = positiveInteger(
|
|
options.shutdownGraceMs,
|
|
SERVER_SHUTDOWN_GRACE_MS,
|
|
'shutdownGraceMs',
|
|
);
|
|
this.diagnosticBytes = positiveInteger(
|
|
options.diagnosticBytes,
|
|
SERVER_DIAGNOSTIC_BYTES,
|
|
'diagnosticBytes',
|
|
);
|
|
}
|
|
|
|
get processId(): number | undefined {
|
|
return this.child?.pid;
|
|
}
|
|
|
|
get isRunning(): boolean {
|
|
return Boolean(this.child) && !this.failure;
|
|
}
|
|
|
|
get stderrDiagnostic(): string {
|
|
return this.diagnostic;
|
|
}
|
|
|
|
get activeThreadCount(): number {
|
|
return [...this.channels.values()].filter(({ worker }) => Boolean(worker)).length;
|
|
}
|
|
|
|
createWorker(options: PiWorkerProcessOptions): PiAgentServerWorkerProcess {
|
|
return new PiAgentServerWorkerProcess(this, options);
|
|
}
|
|
|
|
async openThread(worker: PiAgentServerWorkerProcess): Promise<PiRpcClient> {
|
|
await this.start();
|
|
const control = this.control;
|
|
if (!control) throw new PiProcessError('PI_WORKER_START_FAILED', 'Pi Agent Server is unavailable');
|
|
if (this.channels.has(worker.threadId)) throw new Error('Pi Agent Server thread already exists');
|
|
for (const value of diagnosticSensitiveValues(worker.options)) {
|
|
if (value) this.sensitiveValues.add(value);
|
|
}
|
|
const writable = new AgentServerChannelWritable(
|
|
worker.threadId,
|
|
async (envelope) => await this.writeEnvelope(envelope),
|
|
);
|
|
const rpc = new PiRpcClient(writable, {
|
|
generation: worker.generation,
|
|
defaultTimeoutMs: worker.options.commandTimeoutMs,
|
|
idPrefix: 'makelore-thread',
|
|
onEventListenerError: (error) => {
|
|
this.appendDiagnostic(`[thread-listener] ${error instanceof Error ? error.message : String(error)}\n`);
|
|
},
|
|
});
|
|
this.channels.set(worker.threadId, { rpc, worker });
|
|
try {
|
|
await control.request({
|
|
type: 'thread_open',
|
|
threadId: worker.threadId,
|
|
options: this.threadOptions(worker.options),
|
|
}, { timeoutMs: SERVER_START_TIMEOUT_MS });
|
|
return rpc;
|
|
} catch (error) {
|
|
this.channels.delete(worker.threadId);
|
|
this.retireChannel(worker.threadId);
|
|
const failure = error instanceof PiProcessError
|
|
? error
|
|
: new PiProcessError('PI_WORKER_START_FAILED', 'Could not open Pi Agent Server thread', {
|
|
cause: error,
|
|
generation: worker.generation,
|
|
});
|
|
rpc.invalidate(failure);
|
|
throw failure;
|
|
}
|
|
}
|
|
|
|
async closeThread(
|
|
worker: PiAgentServerWorkerProcess,
|
|
_reason: PiWorkerStopReason,
|
|
): Promise<PiWorkerStopResult> {
|
|
const record = this.channels.get(worker.threadId);
|
|
if (!record) return { mode: 'not-started', code: null, signal: null };
|
|
const control = this.control;
|
|
try {
|
|
if (control && !this.failure) {
|
|
await control.request({
|
|
type: 'thread_close',
|
|
threadId: worker.threadId,
|
|
}, { timeoutMs: SERVER_CONTROL_TIMEOUT_MS });
|
|
}
|
|
} finally {
|
|
this.channels.delete(worker.threadId);
|
|
this.retireChannel(worker.threadId);
|
|
worker.invalidate(new PiProcessError('PI_WORKER_STOPPED', 'Pi Agent Server thread stopped', {
|
|
generation: worker.generation,
|
|
}));
|
|
}
|
|
return { mode: 'stdin-close', code: 0, signal: null };
|
|
}
|
|
|
|
async invalidateThreadForProof(
|
|
worker: PiAgentServerWorkerProcess,
|
|
failure: PiWorkerProofFailure,
|
|
): Promise<void> {
|
|
await this.control?.request({
|
|
type: 'thread_close',
|
|
threadId: worker.threadId,
|
|
}, { timeoutMs: SERVER_CONTROL_TIMEOUT_MS }).catch(() => undefined);
|
|
this.channels.delete(worker.threadId);
|
|
this.retireChannel(worker.threadId);
|
|
worker.invalidate(new PiProcessError(
|
|
failure === 'protocol_invalidation' ? 'PI_RPC_PROTOCOL_ERROR' : 'PI_RPC_EXITED',
|
|
failure === 'protocol_invalidation'
|
|
? 'Injected Agent Server thread protocol failure'
|
|
: 'Injected Agent Server thread exit',
|
|
{ generation: worker.generation },
|
|
));
|
|
}
|
|
|
|
start(): Promise<void> {
|
|
if (this.startFlight) return this.startFlight;
|
|
if (this.child && !this.failure) return Promise.resolve();
|
|
this.startFlight = this.startServer().finally(() => {
|
|
this.startFlight = null;
|
|
});
|
|
return this.startFlight;
|
|
}
|
|
|
|
stop(): Promise<void> {
|
|
if (!this.stopFlight) {
|
|
this.stopFlight = this.stopServer().finally(() => {
|
|
this.stopFlight = null;
|
|
});
|
|
}
|
|
return this.stopFlight;
|
|
}
|
|
|
|
private async startServer(): Promise<void> {
|
|
this.failure = null;
|
|
this.stopping = false;
|
|
this.diagnostic = '';
|
|
const generation = ++this.generation;
|
|
const child = spawn(
|
|
this.options.executablePath,
|
|
[
|
|
// Enables the parent URL used to resolve dependencies from the packaged Pi runtime root.
|
|
'--experimental-import-meta-resolve',
|
|
this.options.serverPath,
|
|
'--runtime-root',
|
|
this.options.runtimeRoot,
|
|
],
|
|
{
|
|
cwd: this.options.runtimeRoot,
|
|
env: buildPiWorkerEnvironment(this.options.configDir),
|
|
stdio: ['pipe', 'pipe', 'pipe'],
|
|
windowsHide: true,
|
|
detached: platform() !== 'win32',
|
|
},
|
|
);
|
|
this.child = child;
|
|
this.exitResult = new Promise((resolve) => {
|
|
child.once('exit', (code, signal) => resolve({ code, signal }));
|
|
});
|
|
const writable = new AgentServerChannelWritable(
|
|
SERVER_CHANNEL,
|
|
async (envelope) => await this.writeEnvelope(envelope),
|
|
);
|
|
const control = new PiRpcClient(writable, {
|
|
generation,
|
|
defaultTimeoutMs: SERVER_CONTROL_TIMEOUT_MS,
|
|
idPrefix: 'makelore-server',
|
|
onEventListenerError: (error) => {
|
|
this.appendDiagnostic(`[server-listener] ${error instanceof Error ? error.message : String(error)}\n`);
|
|
},
|
|
});
|
|
this.control = control;
|
|
this.channels.set(SERVER_CHANNEL, { rpc: control });
|
|
|
|
child.stderr.on('data', (chunk: Buffer) => this.appendDiagnostic(chunk.toString('utf8')));
|
|
const framer = new StrictLfJsonlFramer({
|
|
onRecord: (value) => this.acceptEnvelope(value),
|
|
});
|
|
child.stdout.on('data', (chunk: Buffer) => {
|
|
try {
|
|
framer.push(chunk);
|
|
} catch (error) {
|
|
this.handleProtocolFailure(error);
|
|
}
|
|
});
|
|
child.stdout.on('end', () => {
|
|
try {
|
|
framer.finish();
|
|
} catch (error) {
|
|
this.handleProtocolFailure(error);
|
|
}
|
|
});
|
|
child.once('close', (code, signal) => {
|
|
if (this.child !== child) return;
|
|
const wasStopping = this.stopping;
|
|
if (!wasStopping && !this.failure) {
|
|
this.invalidateServer(new PiProcessError(
|
|
'PI_RPC_EXITED',
|
|
`Pi Agent Server exited unexpectedly (code=${code ?? 'null'}, signal=${signal ?? 'none'})`,
|
|
{ generation, diagnostic: this.diagnostic || undefined, exitCode: code, signal },
|
|
));
|
|
}
|
|
this.child = null;
|
|
this.control = null;
|
|
this.exitResult = null;
|
|
});
|
|
|
|
try {
|
|
await new Promise<void>((resolve, reject) => {
|
|
child.once('spawn', resolve);
|
|
child.once('error', reject);
|
|
});
|
|
await control.request({ type: 'server_initialize' }, { timeoutMs: SERVER_START_TIMEOUT_MS });
|
|
logger.info('[PiAgentServerLifecycle]', {
|
|
event: 'server.ready',
|
|
generation,
|
|
processId: child.pid,
|
|
});
|
|
} catch (error) {
|
|
const failure = error instanceof PiProcessError
|
|
? error
|
|
: new PiProcessError('PI_WORKER_START_FAILED', 'Could not start Pi Agent Server', {
|
|
cause: error,
|
|
generation,
|
|
diagnostic: this.diagnostic || undefined,
|
|
});
|
|
this.invalidateServer(failure);
|
|
await this.forceKillTree(child);
|
|
throw failure;
|
|
}
|
|
}
|
|
|
|
private async stopServer(): Promise<void> {
|
|
if (this.startFlight) await this.startFlight.catch(() => undefined);
|
|
const child = this.child;
|
|
const exitResult = this.exitResult;
|
|
if (!child || !exitResult) {
|
|
this.resetTransport();
|
|
return;
|
|
}
|
|
this.stopping = true;
|
|
await this.control?.request(
|
|
{ type: 'server_shutdown' },
|
|
{ timeoutMs: SERVER_CONTROL_TIMEOUT_MS },
|
|
).catch(() => undefined);
|
|
child.stdin.end();
|
|
const graceful = await this.waitForExit(exitResult, this.shutdownGraceMs);
|
|
if (!graceful) {
|
|
await this.forceKillTree(child);
|
|
await this.waitForExit(exitResult, this.shutdownGraceMs);
|
|
}
|
|
logger.info('[PiAgentServerLifecycle]', {
|
|
event: 'server.stopped',
|
|
generation: this.generation,
|
|
});
|
|
this.resetTransport();
|
|
}
|
|
|
|
private threadOptions(options: PiWorkerProcessOptions): AgentServerThreadOptions {
|
|
return {
|
|
cwd: options.cwd,
|
|
configDir: options.configDir,
|
|
sessionDir: options.sessionDir,
|
|
tools: [...(options.tools ?? DEFAULT_PI_RPC_TOOLS)],
|
|
additionalArgs: [...(options.additionalArgs ?? [])],
|
|
env: stringEnvironment(options.env),
|
|
...(options.conversationId ? { conversationId: options.conversationId } : {}),
|
|
workerGeneration: positiveInteger(options.workerGeneration, 1, 'workerGeneration'),
|
|
};
|
|
}
|
|
|
|
private acceptEnvelope(value: unknown): void {
|
|
if (!envelopeValue(value)) {
|
|
throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi Agent Server envelope is invalid');
|
|
}
|
|
const record = this.channels.get(value.channel);
|
|
if (!record) {
|
|
if (this.retiredChannels.has(value.channel)) return;
|
|
throw new PiProcessError(
|
|
'PI_RPC_PROTOCOL_ERROR',
|
|
`Pi Agent Server emitted an unknown channel ${value.channel}`,
|
|
);
|
|
}
|
|
record.rpc.accept(value.payload);
|
|
}
|
|
|
|
private handleProtocolFailure(error: unknown): void {
|
|
if (this.failure) return;
|
|
const message = error instanceof Error ? error.message : 'Pi Agent Server protocol failure';
|
|
this.appendDiagnostic(`[stdout-protocol] ${message}\n`);
|
|
this.invalidateServer(new PiProcessError('PI_RPC_PROTOCOL_ERROR', message, {
|
|
cause: error,
|
|
generation: this.generation,
|
|
diagnostic: this.diagnostic || undefined,
|
|
}));
|
|
if (this.child) void this.forceKillTree(this.child);
|
|
}
|
|
|
|
private invalidateServer(error: PiProcessError): void {
|
|
if (this.failure) return;
|
|
this.failure = error;
|
|
this.control?.invalidate(error);
|
|
for (const [channel, record] of [...this.channels]) {
|
|
if (channel === SERVER_CHANNEL) continue;
|
|
record.worker?.invalidate(error);
|
|
this.channels.delete(channel);
|
|
this.retireChannel(channel);
|
|
}
|
|
logger.warn('[PiAgentServerLifecycle]', {
|
|
event: 'server.invalidated',
|
|
generation: this.generation,
|
|
code: error.code,
|
|
...(this.diagnostic ? { diagnostic: this.diagnostic } : {}),
|
|
});
|
|
}
|
|
|
|
private writeEnvelope(envelope: AgentServerEnvelope): Promise<void> {
|
|
const child = this.child;
|
|
if (!child || child.stdin.destroyed || this.failure) {
|
|
return Promise.reject(this.failure ?? new PiProcessError(
|
|
'PI_RPC_WRITE_FAILED',
|
|
'Pi Agent Server input is unavailable',
|
|
{ generation: this.generation },
|
|
));
|
|
}
|
|
return new Promise((resolve, reject) => {
|
|
child.stdin.write(`${JSON.stringify(envelope)}\n`, (error?: Error | null) => {
|
|
if (error) reject(error);
|
|
else resolve();
|
|
});
|
|
});
|
|
}
|
|
|
|
private appendDiagnostic(source: string): void {
|
|
const sanitized = sanitizePiDiagnostic(source, [...this.sensitiveValues]);
|
|
this.diagnostic = boundedUtf8Tail(`${this.diagnostic}${sanitized}`, this.diagnosticBytes);
|
|
}
|
|
|
|
private retireChannel(channel: string): void {
|
|
this.retiredChannels.add(channel);
|
|
if (this.retiredChannels.size <= 256) return;
|
|
const oldest = this.retiredChannels.values().next().value as string | undefined;
|
|
if (oldest) this.retiredChannels.delete(oldest);
|
|
}
|
|
|
|
private resetTransport(): void {
|
|
const stopped = new PiProcessError('PI_WORKER_STOPPED', 'Pi Agent Server stopped', {
|
|
generation: this.generation,
|
|
});
|
|
for (const record of this.channels.values()) record.rpc.invalidate(stopped);
|
|
this.channels.clear();
|
|
this.retiredChannels.clear();
|
|
this.child = null;
|
|
this.control = null;
|
|
this.exitResult = null;
|
|
this.failure = null;
|
|
this.stopping = false;
|
|
this.sensitiveValues.clear();
|
|
}
|
|
|
|
private async forceKillTree(child: ChildProcessWithoutNullStreams): Promise<void> {
|
|
if (!child.pid) return;
|
|
if (platform() === 'win32') {
|
|
await runExecutable('taskkill.exe', ['/pid', String(child.pid), '/t', '/f']).catch(() => undefined);
|
|
return;
|
|
}
|
|
try {
|
|
process.kill(-child.pid, 'SIGKILL');
|
|
} catch {
|
|
child.kill('SIGKILL');
|
|
}
|
|
}
|
|
|
|
private async waitForExit(
|
|
exitResult: Promise<{ code: number | null; signal: NodeJS.Signals | null }>,
|
|
timeoutMs: number,
|
|
): Promise<{ code: number | null; signal: NodeJS.Signals | null } | null> {
|
|
let timeout: ReturnType<typeof setTimeout> | undefined;
|
|
try {
|
|
return await Promise.race([
|
|
exitResult,
|
|
new Promise<null>((resolve) => {
|
|
timeout = setTimeout(() => resolve(null), timeoutMs);
|
|
}),
|
|
]);
|
|
} finally {
|
|
if (timeout) clearTimeout(timeout);
|
|
}
|
|
}
|
|
}
|