import type { Readable, Writable } from "node:stream";import {	type LiveSidecarEventSelector,	normalizeSidecarEventMatcher,	SidecarEventBuffer,	SidecarEventBufferOverflow,	sidecarEventWaitAbortError,} from "./event-buffer.js";import { FrameRpcTransport } from "./frame-rpc.js";import {	type FrameTransport,	SplitLaneFrameTransport,	StdioFrameTransport,} from "./frame-stream.js";import type { LiveOwnershipScope } from "./ownership.js";import {	classifySidecarWrittenProtocolFrame,	decodeProtocolFramePayload,	encodeProtocolFramePayload,	HostProtocolFrameFactory,	type LiveControlFrame,	type LiveEventFrame,	type LiveProtocolFrame,	type LiveRequestFrame,	type LiveResponseFrame,	type LiveSidecarRequestFrame,	type LiveSidecarRequestHandler,	type LiveSidecarResponseFrame,	type ProtocolFramePayloadCodec,	resolveSidecarRequestFramePayload,} from "./protocol-frames.js";import type { LiveRequestPayload } from "./request-payloads.js";import {	SidecarRejectedError,	SidecarSilenceTimeout,} from "./sidecar-errors.js"; /** * How long the host tolerates TOTAL inbound silence (no responses, events, * sidecar requests, or heartbeats) before declaring the sidecar dead. The * sidecar heartbeats every 10s from a dedicated thread, so this allows two * missed beats plus margin; it bounds "sidecar is dead or wedged", never "this * request is slow" — individual requests have no deadline of their own. */const DEFAULT_SIDECAR_SILENCE_TIMEOUT_MS = 30_000; export interface SidecarProtocolClientOptions {	/** A combined transport remains supported for embedders and focused tests. */	frameTransport?: FrameTransport<		LiveResponseFrame | LiveEventFrame | LiveSidecarRequestFrame,		LiveProtocolFrame	>;	ordinaryFrameTransport?: FrameTransport<LiveEventFrame, LiveRequestFrame>;	controlFrameTransport?: FrameTransport<		LiveResponseFrame | LiveSidecarRequestFrame | LiveEventFrame,		LiveSidecarResponseFrame | LiveControlFrame	>;	stdin?: Writable;	stdout?: Readable;	control?: Readable & Writable;	eventBufferCapacity: number;	payloadCodec?: ProtocolFramePayloadCodec;	stderrText?: () => string;	frameError?: (error: Error) => Error;	streamEndedError?: () => Error;	/** Override the silence watchdog window. Tests only; production uses the default. */	silenceTimeoutMs?: number;	/**	 * Runs when the silence watchdog fires, before pending work is rejected.	 * The stdio layer uses it to SIGKILL the sidecar child.	 */	onSilenceExpired?: () => void;} export class SidecarProtocolClient {	private readonly eventBuffer: SidecarEventBuffer<LiveEventFrame>;	private readonly eventListeners = new Set<(event: LiveEventFrame) => void>();	private readonly silenceTimeoutMs: number;	private silenceTimer: ReturnType<typeof setInterval> | null = null;	private lastInboundAtMs = 0;	private readonly payloadCodec: ProtocolFramePayloadCodec;	private readonly stderrText: () => string;	private readonly hostFrameFactory = new HostProtocolFrameFactory();	private readonly frameTransport: FrameRpcTransport<		LiveResponseFrame | LiveEventFrame | LiveSidecarRequestFrame,		LiveRequestFrame | LiveSidecarResponseFrame | LiveControlFrame,		LiveResponseFrame,		LiveEventFrame,		LiveSidecarRequestFrame	>;	private closedError: Error | null = null;	private readonly eventWaiters = new Set<{		matches: (event: LiveEventFrame) => boolean;		resolve: (event: LiveEventFrame) => void;		reject: (error: Error) => void;		timer: ReturnType<typeof setTimeout> | null;	}>();	private sidecarRequestHandler: LiveSidecarRequestHandler | null = null; 	constructor(options: SidecarProtocolClientOptions) {		this.silenceTimeoutMs =			options.silenceTimeoutMs ?? DEFAULT_SIDECAR_SILENCE_TIMEOUT_MS;		this.eventBuffer = new SidecarEventBuffer(options.eventBufferCapacity);		this.payloadCodec = options.payloadCodec ?? "bare";		this.stderrText = options.stderrText ?? (() => "");		this.frameTransport = new FrameRpcTransport<			LiveResponseFrame | LiveEventFrame | LiveSidecarRequestFrame,			LiveRequestFrame | LiveSidecarResponseFrame | LiveControlFrame,			LiveResponseFrame,			LiveEventFrame,			LiveSidecarRequestFrame		>({			frameTransport: resolveProtocolFrameTransport(options, this.payloadCodec),			encodeFrame: (frame) =>				encodeProtocolFramePayload(frame, this.payloadCodec),			decodeFrame: (payload) =>				decodeProtocolFramePayload(payload, this.payloadCodec),			classifyFrame: classifySidecarWrittenProtocolFrame,		});		this.frameTransport.onEvent((event) => {			this.dispatchEvent(event);		});		this.frameTransport.onSidecarRequest((request) => {			void this.dispatchSidecarRequest(request);		});		this.frameTransport.onError((error) => {			this.failPermanently(options.frameError?.(error) ?? error);		});		this.frameTransport.onEnd(() => {			this.failPermanently(				options.streamEndedError?.() ??					new Error("sidecar protocol stream ended"),			);		});		this.frameTransport.onFrameActivity(() => {			this.lastInboundAtMs = performance.now();		});		this.startSilenceWatchdog(options.onSilenceExpired);	} 	/**	 * Arm the silence watchdog: ANY inbound frame resets the clock (see the	 * `onFrameActivity` tap above), and the sidecar heartbeats every 10s even	 * while busy, so sustained silence for the full window means the process	 * is dead or wedged — not slow. The check interval is unref'd so an idle	 * host process can still exit naturally.	 */	private startSilenceWatchdog(onExpired?: () => void): void {		this.lastInboundAtMs = performance.now();		const checkIntervalMs = Math.max(			Math.min(this.silenceTimeoutMs / 4, 1_000),			10,		);		this.silenceTimer = setInterval(() => {			const silenceMs = performance.now() - this.lastInboundAtMs;			if (silenceMs < this.silenceTimeoutMs) {				return;			}			this.stopSilenceWatchdog();			const error = new SidecarSilenceTimeout({				silenceMs,				stderr: this.stderrText(),			});			try {				onExpired?.();			} finally {				this.failPermanently(error);			}		}, checkIntervalMs);		this.silenceTimer.unref?.();	} 	private stopSilenceWatchdog(): void {		if (this.silenceTimer !== null) {			clearInterval(this.silenceTimer);			this.silenceTimer = null;		}	} 	setSidecarRequestHandler(handler: LiveSidecarRequestHandler | null): void {		this.sidecarRequestHandler = handler;	} 	onEvent(handler: (event: LiveEventFrame) => void): () => void {		this.eventListeners.add(handler);		return () => {			this.eventListeners.delete(handler);		};	} 	async sendRequest(input: {		ownership: LiveOwnershipScope;		payload: LiveRequestPayload;	}): Promise<LiveResponseFrame> {		if (this.closedError) {			throw this.closedError;		} 		const request = this.hostFrameFactory.createRequestFrame(input);		// No per-request deadline: only the caller knows whether an operation is		// legitimately long (a whole agent turn is one request). A dead or		// wedged sidecar rejects this via the silence watchdog instead.		const response = await this.frameTransport.sendFrame(			request.request_id,			request,		); 		if (response.payload.type === "rejected") {			throw new SidecarRejectedError(request.request_id, response.payload);		}		return response;	} 	async shutdown(reason: string): Promise<void> {		if (this.closedError) {			throw this.closedError;		}		await this.frameTransport.writeFrame(			this.hostFrameFactory.createControlFrame({ type: "shutdown", reason }),		);	} 	async waitForEvent(		matcher: LiveSidecarEventSelector | ((event: LiveEventFrame) => boolean),		timeoutMs?: number,		options?: {			signal?: AbortSignal;		},	): Promise<LiveEventFrame> {		if (this.closedError instanceof SidecarEventBufferOverflow) {			throw this.closedError;		}		const normalizedMatcher =			normalizeSidecarEventMatcher<LiveEventFrame>(matcher);		const bufferedEvent = this.eventBuffer.take(normalizedMatcher);		if (bufferedEvent) {			return bufferedEvent;		}		if (this.closedError) {			throw this.closedError;		}		if (options?.signal?.aborted) {			throw sidecarEventWaitAbortError(options.signal.reason);		} 		return await new Promise<LiveEventFrame>((resolve, reject) => {			let abortListener: (() => void) | null = null;			const waiter = {				matches: normalizedMatcher.matches,				resolve: (event: LiveEventFrame) => {					if (waiter.timer !== null) {						clearTimeout(waiter.timer);					}					if (abortListener) {						options?.signal?.removeEventListener("abort", abortListener);						abortListener = null;					}					this.eventWaiters.delete(waiter);					resolve(event);				},				reject: (error: Error) => {					if (waiter.timer !== null) {						clearTimeout(waiter.timer);					}					if (abortListener) {						options?.signal?.removeEventListener("abort", abortListener);						abortListener = null;					}					this.eventWaiters.delete(waiter);					reject(error);				},				timer:					timeoutMs === undefined						? null						: setTimeout(() => {								this.eventWaiters.delete(waiter);								reject(									new Error(										`timed out waiting for sidecar event\nstderr:\n${this.stderrText()}`,									),								);							}, timeoutMs),			};			if (options?.signal) {				abortListener = () => {					waiter.reject(sidecarEventWaitAbortError(options.signal?.reason));				};				options.signal.addEventListener("abort", abortListener, { once: true });			}			this.eventWaiters.add(waiter);		});	} 	failPermanently(		error: Error,		options?: {			replaceExisting?: (current: Error, next: Error) => boolean;		},	): void {		if (this.closedError) {			if (!options?.replaceExisting?.(this.closedError, error)) {				return;			}		}		this.closedError = error;		this.stopSilenceWatchdog();		this.rejectPending(error);	} 	dispose(): void {		this.stopSilenceWatchdog();		this.frameTransport.dispose();	} 	private async writeFrame(frame: LiveSidecarResponseFrame): Promise<void> {		await this.frameTransport.writeFrame(frame);	} 	private async dispatchSidecarRequest(		request: LiveSidecarRequestFrame,	): Promise<void> {		const payload = await resolveSidecarRequestFramePayload(			request,			this.sidecarRequestHandler,		); 		try {			await this.writeFrame(				this.hostFrameFactory.createSidecarResponseFrame({					request,					payload,				}),			);		} catch (error) {			const normalized =				error instanceof Error ? error : new Error(String(error));			this.failPermanently(normalized);		}	} 	private dispatchEvent(event: LiveEventFrame): void {		// Transport-level liveness beats from the sidecar. Their arrival already		// reset the silence watchdog at the frame layer; they carry no meaning		// for consumers and must never reach the bounded event buffer, where a		// long-idle VM would accumulate one every 10s until overflow.		if (			event.payload.type === "structured" &&			event.payload.name === "heartbeat"		) {			return;		}		for (const listener of this.eventListeners) {			try {				listener(event);			} catch {				// Event listeners are best-effort observers and must not break framing.			}		}		for (const waiter of this.eventWaiters) {			if (!waiter.matches(event)) {				continue;			}			waiter.resolve(event);			return;		}		this.bufferEvent(event);	} 	private bufferEvent(event: LiveEventFrame): void {		const overflow = this.eventBuffer.buffer(event);		if (overflow) {			this.failPermanently(overflow);		}	} 	private rejectPending(error: Error): void {		this.frameTransport.rejectAll(error);		for (const waiter of this.eventWaiters) {			waiter.reject(error);		}		this.eventWaiters.clear();	}} function resolveProtocolFrameTransport(	options: SidecarProtocolClientOptions,	payloadCodec: ProtocolFramePayloadCodec,): FrameTransport<	LiveResponseFrame | LiveEventFrame | LiveSidecarRequestFrame,	LiveRequestFrame | LiveSidecarResponseFrame | LiveControlFrame> {	const hasSplitTransport =		options.ordinaryFrameTransport !== undefined ||		options.controlFrameTransport !== undefined;	const hasStdio =		options.stdin !== undefined ||		options.stdout !== undefined ||		options.control !== undefined;	if (options.frameTransport) {		if (hasSplitTransport || hasStdio) {			throw new Error(				"SidecarProtocolClient frameTransport cannot be combined with split-lane transports or streams",			);		}		return options.frameTransport;	} 	let ordinary = options.ordinaryFrameTransport;	let control = options.controlFrameTransport;	if (hasSplitTransport) {		if (!ordinary || !control || hasStdio) {			throw new Error(				"SidecarProtocolClient requires both ordinaryFrameTransport and controlFrameTransport, without stdio streams",			);		}	} else {		if (!options.stdin || !options.stdout || !options.control) {			throw new Error(				"SidecarProtocolClient requires a combined frameTransport or stdin/stdout/control streams",			);		}		ordinary = new StdioFrameTransport<LiveEventFrame, LiveRequestFrame>({			stdin: options.stdin,			stdout: options.stdout,			encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec),			decodeFrame: (payload) => {				const frame = decodeProtocolFramePayload(payload, payloadCodec);				validateOrdinaryInboundFrame(frame);				return frame;			},		});		control = new StdioFrameTransport<			LiveResponseFrame | LiveSidecarRequestFrame | LiveEventFrame,			LiveSidecarResponseFrame | LiveControlFrame		>({			stdin: options.control,			stdout: options.control,			encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec),			decodeFrame: (payload) => {				const frame = decodeProtocolFramePayload(payload, payloadCodec);				validateControlInboundFrame(frame);				return frame;			},		});	} 	return new SplitLaneFrameTransport({		ordinary,		control,		validateOrdinaryFrame: validateOrdinaryInboundFrame,		validateControlFrame: validateControlInboundFrame,		selectWriteLane: (frame) => {			switch (frame.frame_type) {				case "request":					return "ordinary";				case "sidecar_response":				case "control":					return "control";				default: {					const invalidFrame = frame as { frame_type: string };					throw new Error(						`host frame ${invalidFrame.frame_type} is not valid on either writable protocol lane`,					);				}			}		},	});} function validateOrdinaryInboundFrame(frame: {	frame_type: string;	payload?: unknown;}): asserts frame is LiveEventFrame {	if (frame.frame_type !== "event") {		throw new Error(			`sidecar frame ${frame.frame_type} is not valid on the ordinary protocol lane`,		);	}	if (isHeartbeatFrame(frame as LiveEventFrame)) {		throw new Error(			"sidecar heartbeat event is not valid on the ordinary protocol lane",		);	}} function validateControlInboundFrame(frame: {	frame_type: string;	payload?: unknown;}): asserts frame is	| LiveResponseFrame	| LiveSidecarRequestFrame	| LiveEventFrame {	if (		frame.frame_type === "response" ||		frame.frame_type === "sidecar_request"	) {		return;	}	if (		frame.frame_type === "event" &&

Lines 1–500 of 514. Load more