diff --git a/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-interactive-io_2026-08-21-21-00.json b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-interactive-io_2026-08-21-21-00.json new file mode 100644 index 0000000000..6dbd0f8be6 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-interactive-io_2026-08-21-21-00.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon-protocol", + "comment": "Add request-scoped stdin, raw-mode control, and terminal fallback contracts.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon-protocol", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-daemon-transport/mojazayeri-interactive-io_2026-08-21-21-00.json b/common/changes/@rushstack/rush-daemon-transport/mojazayeri-interactive-io_2026-08-21-21-00.json new file mode 100644 index 0000000000..eed82891ba --- /dev/null +++ b/common/changes/@rushstack/rush-daemon-transport/mojazayeri-interactive-io_2026-08-21-21-00.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon-transport", + "comment": "Serialize and backpressure asynchronous incoming frame handlers.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon-transport", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-interactive-io_2026-08-21-21-00.json b/common/changes/@rushstack/rush-daemon/mojazayeri-interactive-io_2026-08-21-21-00.json new file mode 100644 index 0000000000..aaf540d233 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-interactive-io_2026-08-21-21-00.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Route request-scoped interactive input and signal PTY-only in-process fallback.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/reviews/api/rush-daemon-protocol.api.md b/common/reviews/api/rush-daemon-protocol.api.md index e880b9645e..68a328e076 100644 --- a/common/reviews/api/rush-daemon-protocol.api.md +++ b/common/reviews/api/rush-daemon-protocol.api.md @@ -21,7 +21,10 @@ export const DAEMON_CONTROL_MESSAGE_KINDS: readonly [ 'unsubscribe', 'ping', 'pong', -'error' +'error', +'setRawMode', +'rawModeChanged', +'terminalPolicy' ]; // @beta @@ -43,6 +46,9 @@ export const DAEMON_EVENT_TYPES: readonly [ 'extension' ]; +// @beta +export const DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR: number; + // @beta export const DAEMON_PROTOCOL_VERSION: IDaemonProtocolVersion; @@ -50,7 +56,7 @@ export const DAEMON_PROTOCOL_VERSION: IDaemonProtocolVersion; export type DaemonCommandOutcome = 'success' | 'success-with-warning' | 'failure' | 'aborted'; // @beta -export type DaemonControlMessage = IDaemonHelloMessage | IDaemonHelloAckMessage | IDaemonSubscribeMessage | IDaemonUnsubscribeMessage | IDaemonPingMessage | IDaemonPongMessage | IDaemonErrorMessage; +export type DaemonControlMessage = IDaemonHelloMessage | IDaemonHelloAckMessage | IDaemonSubscribeMessage | IDaemonUnsubscribeMessage | IDaemonPingMessage | IDaemonPongMessage | IDaemonErrorMessage | IDaemonSetRawModeMessage | IDaemonRawModeChangedMessage | IDaemonTerminalPolicyMessage; // @beta export type DaemonControlMessageKind = (typeof DAEMON_CONTROL_MESSAGE_KINDS)[number]; @@ -115,6 +121,15 @@ export class DaemonProtocolError extends Error { // @beta export type DaemonProtocolErrorCode = 'frameTooLarge' | 'unknownFrameType' | 'malformedPayload' | 'malformedControlMessage' | 'protocolVersionMismatch'; +// @beta +export type DaemonTerminalPolicyDecision = 'runInDaemon' | 'requiresInProcess'; + +// @beta +export type DaemonTerminalPolicyReason = 'controllingTerminalRequired'; + +// @beta +export type DaemonTerminalRequirement = 'none' | 'interactiveInput' | 'controllingTerminal'; + // @beta export type DaemonVerbosity = 'quiet' | 'normal' | 'verbose' | 'debug'; @@ -127,6 +142,9 @@ export function decodeDaemonEventFrame(payload: Uint8Array): IDaemonEventEnvelop // @beta export function decodeDaemonLogChunk(payload: Uint8Array): IDaemonLogChunk; +// @beta +export function decodeDaemonStdinChunk(payload: Uint8Array): IDaemonStdinChunk; + // @beta export const DEFAULT_MAX_PAYLOAD_BYTES: number; @@ -145,6 +163,9 @@ export function encodeDaemonFrames(frames: readonly IDaemonFrame[]): Uint8Array[ // @beta export function encodeDaemonLogChunk(log: IDaemonLogChunk): Uint8Array; +// @beta +export function encodeDaemonStdinChunk(input: IDaemonStdinChunk): Uint8Array; + // @beta export const FRAME_HEADER_BYTES: number; @@ -159,6 +180,7 @@ export interface IDaemonClientCaps { readonly colorLevel?: number; readonly columns?: number; readonly isTTY: boolean; + readonly supportsInteractiveIO?: boolean; readonly verbosity?: DaemonVerbosity; } @@ -314,11 +336,13 @@ export interface IDaemonPhasedOperationSelection { // @beta export interface IDaemonPhasedRequest { + readonly acceptsStdin?: boolean; readonly commandName: string; readonly engineShape: IDaemonPhasedEngineShape; readonly environment: Readonly>; readonly operationSelection: ReadonlyArray; readonly requestId: string; + readonly terminalRequirement?: DaemonTerminalRequirement; } // @beta @@ -359,6 +383,34 @@ export interface IDaemonProtocolVersion { readonly minor: number; } +// @beta +export interface IDaemonRawModeChangedMessage { + // (undocumented) + readonly kind: 'rawModeChanged'; + // (undocumented) + readonly payload: { + readonly enabled: boolean; + readonly requestId: string; + }; +} + +// @beta +export interface IDaemonSetRawModeMessage { + // (undocumented) + readonly kind: 'setRawMode'; + // (undocumented) + readonly payload: { + readonly enabled: boolean; + readonly requestId: string; + }; +} + +// @beta +export interface IDaemonStdinChunk { + readonly chunk: Uint8Array; + readonly requestId: string; +} + // @beta export interface IDaemonSubscribeMessage { // (undocumented) @@ -367,6 +419,24 @@ export interface IDaemonSubscribeMessage { readonly payload: IDaemonClientCaps; } +// @beta +export interface IDaemonTerminalPolicyMessage { + // (undocumented) + readonly kind: 'terminalPolicy'; + // (undocumented) + readonly payload: IDaemonTerminalPolicyResult; +} + +// @beta +export interface IDaemonTerminalPolicyResult { + // (undocumented) + readonly decision: DaemonTerminalPolicyDecision; + // (undocumented) + readonly reason?: DaemonTerminalPolicyReason; + // (undocumented) + readonly requestId: string; +} + // @beta export interface IDaemonUnsubscribeMessage { // (undocumented) @@ -411,6 +481,9 @@ export const LENGTH_FIELD_OFFSET: number; // @beta export const MAX_OPERATION_ID_BYTES: number; +// @beta +export const MAX_REQUEST_ID_BYTES: number; + // @beta export function negotiateDaemonHello(hello: IDaemonHelloMessage, localVersion: IDaemonProtocolVersion, sessionId: string): DaemonHandshakeOutcome; @@ -430,6 +503,12 @@ export class ProtocolVersionMismatchError extends DaemonProtocolError { readonly expectedMajor: number; } +// @beta +export const REQUEST_ID_LENGTH_BYTES: number; + +// @beta +export const REQUEST_ID_LENGTH_OFFSET: number; + // @beta export const RUSHD_EXTENSION_NAMESPACE: 'rushd'; diff --git a/common/reviews/api/rush-daemon-transport.api.md b/common/reviews/api/rush-daemon-transport.api.md index c69a18ee09..c198668b6f 100644 --- a/common/reviews/api/rush-daemon-transport.api.md +++ b/common/reviews/api/rush-daemon-transport.api.md @@ -19,7 +19,7 @@ export class DaemonFrameConnection { constructor(socket: net.Socket); closeAsync(): Promise; onClosed(handler: (error: Error | undefined) => void): void; - onFrame(handler: (frame: IDaemonFrame) => void): void; + onFrame(handler: (frame: IDaemonFrame) => void | Promise): void; sendFrameAsync(frame: IDaemonFrame): Promise; // @internal get socket(): net.Socket; diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index 17b4f625c1..a984f44f83 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -7,12 +7,15 @@ /// import * as childProcess from 'node:child_process'; +import type { DaemonTerminalRequirement } from '@rushstack/rush-daemon-protocol'; import type { GetInputsSnapshotAsyncFn } from '@microsoft/rush-lib'; import type { IDaemonCommandResult } from '@rushstack/rush-daemon-protocol'; import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; import type { IDaemonPaths } from '@rushstack/rush-daemon-transport'; import type { IDaemonPhasedRequest } from '@rushstack/rush-daemon-protocol'; import type { IDaemonPhasedRequestResult } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonSetRawModeMessage } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; import type { IInputsSnapshot } from '@microsoft/rush-lib'; import type { IOperationGraph } from '@microsoft/rush-lib'; import type { ITerminal } from '@rushstack/terminal'; @@ -27,6 +30,16 @@ export type CreateWorkspaceEngineComponentsAsync = (options: ICreateWorkspaceEng // @beta export type CreateWorkspaceSessionComponentsAsync = (options: ICreateWorkspaceSessionComponentsOptions) => Promise; +// @beta +export class DaemonRequiresInProcessError extends Error { + constructor(policy: IDaemonTerminalPolicyResult); + // (undocumented) + readonly policy: IDaemonTerminalPolicyResult; +} + +// @beta +export function evaluateDaemonTerminalPolicy(requestId: string, requirement?: DaemonTerminalRequirement): IDaemonTerminalPolicyResult; + // @beta export type GlobalCommandExecutor = (context: IGlobalCommandExecutionContext) => Promise; @@ -62,6 +75,28 @@ export interface ICreateWorkspaceSessionComponentsOptions { readonly rushConfiguration: RushConfiguration; } +// @beta +export interface IDaemonInteractiveConnection { + // (undocumented) + readonly abortSignal: AbortSignal; + // (undocumented) + registerRequest(options: IDaemonInteractiveRequestOptions): IInteractiveRequestSession; + // (undocumented) + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; +} + +// @beta +export interface IDaemonInteractiveRequestOptions { + // (undocumented) + readonly abortSignal: AbortSignal; + // (undocumented) + readonly acceptsStdin: boolean; + // (undocumented) + readonly onFailure: (error: Error) => void; + // (undocumented) + readonly requestId: string; +} + // @beta export interface IGlobalCommandEnvironment { // (undocumented) @@ -81,6 +116,8 @@ export interface IGlobalCommandExecutionContext { // (undocumented) readonly environment: IGlobalCommandEnvironment; // (undocumented) + readonly interactiveInput: IInteractiveRequestSession | undefined; + // (undocumented) registerDisposable(disposable: AsyncDisposable): void; spawnChild(command: string, args: ReadonlyArray, options?: IGlobalCommandSpawnOptions): childProcess.ChildProcessWithoutNullStreams; // (undocumented) @@ -100,8 +137,10 @@ export interface IGlobalCommandExecutionResult { // @beta export interface IGlobalCommandRequestClient { readonly abortSignal: AbortSignal; + readonly interactiveSession?: IInteractiveRequestSession; writeResultAsync(result: IDaemonCommandResult): Promise; writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; } // @beta @@ -112,6 +151,8 @@ export interface IGlobalCommandSpawnOptions { // (undocumented) readonly environmentOverlay?: Readonly; // (undocumented) + readonly forwardInput?: boolean; + // (undocumented) readonly forwardOutput?: boolean; // (undocumented) readonly shell?: boolean | string; @@ -121,12 +162,53 @@ export interface IGlobalCommandSpawnOptions { // @beta export interface IGlobalCommandTerminalProperties { + // (undocumented) + readonly acceptsStdin?: boolean; // (undocumented) readonly columns: number | undefined; // (undocumented) readonly isTTY: boolean; // (undocumented) readonly supportsColor: boolean; + // (undocumented) + readonly terminalRequirement?: DaemonTerminalRequirement; +} + +// @beta +export interface IInteractiveRequestControlClient { + // (undocumented) + readonly abortSignal: AbortSignal; + writeRawModeControlAsync(message: IDaemonSetRawModeMessage): Promise; +} + +// @beta +export interface IInteractiveRequestInputSink { + // (undocumented) + writeInputAsync(chunk: Uint8Array): Promise; +} + +// @beta +export interface IInteractiveRequestRegistrationOptions { + // (undocumented) + readonly acceptsStdin: boolean; + // (undocumented) + readonly client: IInteractiveRequestControlClient; + // (undocumented) + readonly onFailure: (error: Error) => void; + // (undocumented) + readonly requestId: string; +} + +// @beta +export interface IInteractiveRequestSession { + // (undocumented) + attachInputSink(sink: IInteractiveRequestInputSink): Disposable; + // (undocumented) + finishAsync(): Promise; + // (undocumented) + readonly requestId: string; + // (undocumented) + setRawModeAsync(enabled: boolean): Promise; } // @beta @@ -141,14 +223,35 @@ export interface IMapWorkspaceInvalidationsOptions { readonly operationGraph: IOperationGraph; } +// @beta +export class InteractiveInputRoutingError extends Error { + constructor(code: InteractiveInputRoutingErrorCode, message: string); + // (undocumented) + readonly code: InteractiveInputRoutingErrorCode; +} + +// @beta +export type InteractiveInputRoutingErrorCode = 'duplicateRequest' | 'unknownRequest' | 'completedRequest' | 'nonInteractiveRequest'; + +// @beta +export class InteractiveRequestInputRouter { + // (undocumented) + register(options: IInteractiveRequestRegistrationOptions): IInteractiveRequestSession; + // (undocumented) + routeStdinFrameAsync(payload: Uint8Array): Promise; +} + // @beta export interface IPhasedRequestClient { readonly abortSignal: AbortSignal; getNextEventSequence(): number; + readonly interactiveInputSink?: IInteractiveRequestInputSink; + readonly interactiveSession?: IInteractiveRequestSession; readonly sessionId: string; writeEventAsync(event: IDaemonEventEnvelope): Promise; writeLogChunkAsync(operationId: string, stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; writeResultAsync(result: IDaemonPhasedRequestResult): Promise; + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; } // @public @@ -201,6 +304,7 @@ export interface IRushDaemonHostOptions { readonly createWorkspaceSessionAsync?: WorkspaceSessionFactory; readonly daemonVersion: string; readonly onError?: (error: Error) => void; + readonly onInteractiveConnection?: (connection: IDaemonInteractiveConnection) => void; readonly repoRoot: string; readonly rushVersion: string; readonly startupOptions?: Readonly>; diff --git a/libraries/rush-daemon-protocol/README.md b/libraries/rush-daemon-protocol/README.md index 05647aa49f..c06ed6c9ec 100644 --- a/libraries/rush-daemon-protocol/README.md +++ b/libraries/rush-daemon-protocol/README.md @@ -22,6 +22,8 @@ The engine-agnostic **wire layer** spoken by every client of the Rush daemon (`r resolved it against a real warm operation graph. - **Final command result contract** — one typed success, warning, failure, or abort outcome with the authoritative Rush-compatible exit code, delivered after request output drains. +- **Interactive request contracts** — request-tagged stdin frames preserve arbitrary bytes, while + acknowledged raw-mode controls and typed terminal-policy results remain scoped to one request. Part of the Rush 6 / rushd re-architecture: [microsoft/rushstack#5894](https://github.com/microsoft/rushstack/issues/5894). diff --git a/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts b/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts index 49d87c195a..542c45e723 100644 --- a/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts +++ b/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts @@ -1,19 +1,21 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. -import { isDaemonControlMessageKind } from './DaemonControlMessage'; +import { isDaemonControlMessageKind } from './DaemonControlKinds'; import { DaemonProtocolError } from './DaemonProtocolError'; import { isDaemonVerbosity } from './DaemonVerbosity'; - -/** Returns `true` when `value` is a non-null object usable as a control record. @beta */ +import { + validateInteractiveCapability, + validateRawModeControl, + validateTerminalPolicyControl +} from './InteractiveControlValidation'; +/** Returns `true` when `value` is a non-null control record. @beta */ export function isDaemonControlRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null; } - function fail(reason: string): never { throw new DaemonProtocolError('malformedControlMessage', reason); } - function requireRecordField(record: Record, field: string): Record { const value: unknown = record[field]; if (!isDaemonControlRecord(value)) { @@ -21,13 +23,11 @@ function requireRecordField(record: Record, field: string): Rec } return value; } - function requireStringField(record: Record, field: string): void { if (typeof record[field] !== 'string') { fail(`Control message field "${field}" must be a string.`); } } - function requireNumberField(record: Record, field: string): void { if (typeof record[field] !== 'number') { fail(`Control message field "${field}" must be a number.`); @@ -55,6 +55,7 @@ function validateSubscribe(payload: Record): void { if (typeof payload.isTTY !== 'boolean') { fail('Subscribe message payload.isTTY must be a boolean.'); } + validateInteractiveCapability(payload); requireSubscribeVerbosity(payload); } @@ -80,14 +81,13 @@ const VALIDATORS_BY_KIND: Record = { unsubscribe: noopValidator, ping: noopValidator, pong: validatePong, - error: validateError + error: validateError, + setRawMode: validateRawModeControl, + rawModeChanged: validateRawModeControl, + terminalPolicy: validateTerminalPolicyControl }; -/** Structurally validates a parsed control message. - * @throws {@link DaemonProtocolError} when the value is not a well-formed control message. - * - * @beta - */ +/** Structurally validates a parsed control message. @beta */ export function validateDaemonControlMessage(value: unknown): void { if (!isDaemonControlRecord(value)) { fail('Control frame payload is not a JSON object.'); diff --git a/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts b/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts index db0e76b511..baac9379ee 100644 --- a/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts +++ b/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts @@ -16,6 +16,8 @@ import type { DaemonVerbosity } from './DaemonVerbosity'; export interface IDaemonClientCaps { /** Whether the client's output is an interactive TTY. */ readonly isTTY: boolean; + /** Whether the client supports request-scoped stdin and acknowledged raw-mode control. */ + readonly supportsInteractiveIO?: boolean; /** The verbosity subset this client receives. Defaults to `normal`. */ readonly verbosity?: DaemonVerbosity; /** The client's terminal width in columns, when known. */ diff --git a/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts b/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts new file mode 100644 index 0000000000..cd10b5a1e2 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts @@ -0,0 +1,29 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** Every control message `kind` discriminant. @beta */ +export const DAEMON_CONTROL_MESSAGE_KINDS: readonly [ + 'hello', + 'helloAck', + 'subscribe', + 'unsubscribe', + 'ping', + 'pong', + 'error', + 'setRawMode', + 'rawModeChanged', + 'terminalPolicy' +] = [ + 'hello', 'helloAck', 'subscribe', 'unsubscribe', 'ping', 'pong', 'error', + 'setRawMode', 'rawModeChanged', 'terminalPolicy' +]; + +/** The union of control message `kind` discriminants. @beta */ +export type DaemonControlMessageKind = (typeof DAEMON_CONTROL_MESSAGE_KINDS)[number]; + +const CONTROL_KIND_SET: ReadonlySet = new Set(DAEMON_CONTROL_MESSAGE_KINDS); + +/** Returns `true` when `value` is a control message `kind`. @beta */ +export function isDaemonControlMessageKind(value: unknown): value is DaemonControlMessageKind { + return typeof value === 'string' && CONTROL_KIND_SET.has(value); +} diff --git a/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts b/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts index c20a04a3a5..0619f5ff5f 100644 --- a/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts +++ b/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts @@ -2,6 +2,11 @@ // See LICENSE in the project root for license information. import type { IDaemonClientCaps } from './DaemonClientCaps'; +import type { + IDaemonRawModeChangedMessage, + IDaemonSetRawModeMessage, + IDaemonTerminalPolicyMessage +} from './DaemonInteractiveControl'; import type { IDaemonPongMessage } from './DaemonPongMessage'; import type { DaemonProtocolErrorCode } from './DaemonProtocolError'; import type { IDaemonProtocolVersion } from './DaemonProtocolVersion'; @@ -66,30 +71,7 @@ export type DaemonControlMessage = | IDaemonUnsubscribeMessage | IDaemonPingMessage | IDaemonPongMessage - | IDaemonErrorMessage; - -/** - * The runtime list of control message `kind` discriminants, from which - * {@link DaemonControlMessageKind} is derived (single source of truth). - * - * @beta - */ -export const DAEMON_CONTROL_MESSAGE_KINDS: readonly [ - 'hello', - 'helloAck', - 'subscribe', - 'unsubscribe', - 'ping', - 'pong', - 'error' -] = ['hello', 'helloAck', 'subscribe', 'unsubscribe', 'ping', 'pong', 'error']; - -/** The union of control message `kind` discriminants, derived from the runtime list. @beta */ -export type DaemonControlMessageKind = (typeof DAEMON_CONTROL_MESSAGE_KINDS)[number]; - -const CONTROL_KIND_SET: ReadonlySet = new Set(DAEMON_CONTROL_MESSAGE_KINDS); - -/** Returns `true` when `value` is a control message `kind`. @beta */ -export function isDaemonControlMessageKind(value: unknown): value is DaemonControlMessageKind { - return typeof value === 'string' && CONTROL_KIND_SET.has(value); -} + | IDaemonErrorMessage + | IDaemonSetRawModeMessage + | IDaemonRawModeChangedMessage + | IDaemonTerminalPolicyMessage; diff --git a/libraries/rush-daemon-protocol/src/DaemonInteractiveControl.ts b/libraries/rush-daemon-protocol/src/DaemonInteractiveControl.ts new file mode 100644 index 0000000000..9748cbac97 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonInteractiveControl.ts @@ -0,0 +1,28 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonTerminalPolicyResult } from './DaemonTerminalPolicy'; + +/** Asks the thin client to change raw mode for one active request. @beta */ +export interface IDaemonSetRawModeMessage { + readonly kind: 'setRawMode'; + readonly payload: { + readonly enabled: boolean; + readonly requestId: string; + }; +} + +/** Confirms that the thin client applied a request-scoped raw-mode change. @beta */ +export interface IDaemonRawModeChangedMessage { + readonly kind: 'rawModeChanged'; + readonly payload: { + readonly enabled: boolean; + readonly requestId: string; + }; +} + +/** Reports whether a command can run in the daemon or requires client-side fallback. @beta */ +export interface IDaemonTerminalPolicyMessage { + readonly kind: 'terminalPolicy'; + readonly payload: IDaemonTerminalPolicyResult; +} diff --git a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts index ead42dcee1..eda6676677 100644 --- a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts +++ b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts @@ -2,6 +2,7 @@ // See LICENSE in the project root for license information. import type { IDaemonCommandResult } from './DaemonCommandResult'; +import type { DaemonTerminalRequirement } from './DaemonTerminalPolicy'; /** * The enabled state assigned to one selected operation by a phased request. @@ -44,6 +45,8 @@ export interface IDaemonPhasedEngineShape { * @beta */ export interface IDaemonPhasedRequest { + /** Whether the command accepts request-scoped stdin bytes. */ + readonly acceptsStdin?: boolean; /** The parsed phased command name. */ readonly commandName: string; /** The exact warm engine shape against which the selection was resolved. */ @@ -54,6 +57,8 @@ export interface IDaemonPhasedRequest { readonly operationSelection: ReadonlyArray; /** A client-generated identifier unique within the connection. */ readonly requestId: string; + /** Terminal capability needed by the resolved command. */ + readonly terminalRequirement?: DaemonTerminalRequirement; } /** diff --git a/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts b/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts index e4053bdc79..f9e96f17e5 100644 --- a/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts +++ b/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts @@ -1,6 +1,9 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. +/** The first additive protocol minor that supports request-scoped interactive I/O. @beta */ +export const DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR: number = 3; + /** * A rushd wire protocol version. * @@ -34,7 +37,7 @@ export interface IDaemonProtocolVersion { */ export const DAEMON_PROTOCOL_VERSION: IDaemonProtocolVersion = { major: 0, - minor: 2 + minor: DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR }; /** diff --git a/libraries/rush-daemon-protocol/src/DaemonTerminalPolicy.ts b/libraries/rush-daemon-protocol/src/DaemonTerminalPolicy.ts new file mode 100644 index 0000000000..c50ac3b240 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonTerminalPolicy.ts @@ -0,0 +1,22 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** The terminal capability required by a resolved command. @beta */ +export type DaemonTerminalRequirement = 'none' | 'interactiveInput' | 'controllingTerminal'; + +/** The server-side execution decision for a resolved command. @beta */ +export type DaemonTerminalPolicyDecision = 'runInDaemon' | 'requiresInProcess'; + +/** Why a command cannot run in the daemon. @beta */ +export type DaemonTerminalPolicyReason = 'controllingTerminalRequired'; + +/** + * A request-scoped decision that a future thin client can use to select daemon or in-process execution. + * + * @beta + */ +export interface IDaemonTerminalPolicyResult { + readonly decision: DaemonTerminalPolicyDecision; + readonly reason?: DaemonTerminalPolicyReason; + readonly requestId: string; +} diff --git a/libraries/rush-daemon-protocol/src/FrameConstants.ts b/libraries/rush-daemon-protocol/src/FrameConstants.ts index 9f52f21f1c..71f41fd30d 100644 --- a/libraries/rush-daemon-protocol/src/FrameConstants.ts +++ b/libraries/rush-daemon-protocol/src/FrameConstants.ts @@ -39,5 +39,14 @@ export const MAX_OPERATION_ID_BYTES: number = 65535; /** The offset of the operation-id length prefix within a log frame payload. @beta */ export const OPERATION_ID_LENGTH_OFFSET: number = 0; +/** The byte length of the request-id length prefix used by stdin frames (`u16` little-endian). @beta */ +export const REQUEST_ID_LENGTH_BYTES: number = 2; + +/** The maximum byte length of a request id in a stdin frame (`u16` range). @beta */ +export const MAX_REQUEST_ID_BYTES: number = 65535; + +/** The offset of the request-id length prefix within a stdin frame payload. @beta */ +export const REQUEST_ID_LENGTH_OFFSET: number = 0; + /** The offset of the frame payload within a serialized frame. @beta */ export const PAYLOAD_OFFSET: number = FRAME_HEADER_BYTES; diff --git a/libraries/rush-daemon-protocol/src/InteractiveControlValidation.ts b/libraries/rush-daemon-protocol/src/InteractiveControlValidation.ts new file mode 100644 index 0000000000..a768f89fbb --- /dev/null +++ b/libraries/rush-daemon-protocol/src/InteractiveControlValidation.ts @@ -0,0 +1,50 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { DaemonProtocolError } from './DaemonProtocolError'; + +const EMPTY_LENGTH: number = 0; + +function fail(reason: string): never { + throw new DaemonProtocolError('malformedControlMessage', reason); +} + +function requireRequestId(payload: Record): void { + if (typeof payload.requestId !== 'string' || payload.requestId.length === EMPTY_LENGTH) { + fail('Interactive control message payload.requestId must be a nonempty string.'); + } +} + +/** Validates optional interactive capability negotiation. @internal */ +export function validateInteractiveCapability(payload: Record): void { + if (payload.supportsInteractiveIO !== undefined && typeof payload.supportsInteractiveIO !== 'boolean') { + fail('Subscribe message payload.supportsInteractiveIO must be a boolean.'); + } +} + +/** Validates a request-scoped raw-mode command or acknowledgement. @internal */ +export function validateRawModeControl(payload: Record): void { + requireRequestId(payload); + if (typeof payload.enabled !== 'boolean') { + fail('Interactive control message payload.enabled must be a boolean.'); + } +} + +/** Validates a request-scoped terminal execution policy result. @internal */ +export function validateTerminalPolicyControl(payload: Record): void { + requireRequestId(payload); + validateTerminalPolicyDecision(payload.decision); + validateTerminalPolicyReason(payload.reason); +} + +function validateTerminalPolicyDecision(decision: unknown): void { + if (decision !== 'runInDaemon' && decision !== 'requiresInProcess') { + fail('Terminal policy payload.decision is not recognized.'); + } +} + +function validateTerminalPolicyReason(reason: unknown): void { + if (reason !== undefined && reason !== 'controllingTerminalRequired') { + fail('Terminal policy payload.reason is not recognized.'); + } +} diff --git a/libraries/rush-daemon-protocol/src/StdinFrameCodec.ts b/libraries/rush-daemon-protocol/src/StdinFrameCodec.ts new file mode 100644 index 0000000000..5d7b78003b --- /dev/null +++ b/libraries/rush-daemon-protocol/src/StdinFrameCodec.ts @@ -0,0 +1,77 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { DaemonProtocolError } from './DaemonProtocolError'; +import { WIRE_TEXT_ENCODER } from './DaemonWireText'; +import { + MAX_REQUEST_ID_BYTES, + REQUEST_ID_LENGTH_BYTES, + REQUEST_ID_LENGTH_OFFSET +} from './FrameConstants'; + +const LITTLE_ENDIAN: boolean = true; +const REQUEST_ID_TEXT_DECODER: InstanceType = new TextDecoder('utf-8', { + fatal: true +}); + +/** One request-tagged chunk of raw stdin bytes. @beta */ +export interface IDaemonStdinChunk { + /** Raw bytes that must not be decoded or re-encoded. */ + readonly chunk: Uint8Array; + /** The active request that owns the input. */ + readonly requestId: string; +} + +/** Serializes stdin as `[u16 LE requestIdBytes][requestId utf8][raw chunk]`. @beta */ +export function encodeDaemonStdinChunk(input: IDaemonStdinChunk): Uint8Array { + const idBytes: Uint8Array = WIRE_TEXT_ENCODER.encode(input.requestId); + if (idBytes.length > MAX_REQUEST_ID_BYTES) { + throw new DaemonProtocolError( + 'malformedPayload', + `Request id is ${idBytes.length} bytes, exceeding the maximum of ${MAX_REQUEST_ID_BYTES}.` + ); + } + const payload: Uint8Array = new Uint8Array(REQUEST_ID_LENGTH_BYTES + idBytes.length + input.chunk.length); + new DataView(payload.buffer).setUint16(REQUEST_ID_LENGTH_OFFSET, idBytes.length, LITTLE_ENDIAN); + payload.set(idBytes, REQUEST_ID_LENGTH_BYTES); + payload.set(input.chunk, REQUEST_ID_LENGTH_BYTES + idBytes.length); + return payload; +} + +/** Parses a stdin payload without interpreting or transforming its raw input bytes. @beta */ +export function decodeDaemonStdinChunk(payload: Uint8Array): IDaemonStdinChunk { + if (payload.length < REQUEST_ID_LENGTH_BYTES) { + throw new DaemonProtocolError( + 'malformedPayload', + 'Stdin frame payload is too short to contain a request id length.' + ); + } + const idLength: number = new DataView(payload.buffer, payload.byteOffset).getUint16( + REQUEST_ID_LENGTH_OFFSET, + LITTLE_ENDIAN + ); + const chunkOffset: number = REQUEST_ID_LENGTH_BYTES + idLength; + if (payload.length < chunkOffset) { + throw new DaemonProtocolError( + 'malformedPayload', + `Stdin frame declared a request id of ${idLength} bytes but the payload is ${payload.length} bytes.` + ); + } + const idBytes: Uint8Array = payload.subarray(REQUEST_ID_LENGTH_BYTES, chunkOffset); + return { + chunk: payload.slice(chunkOffset), + requestId: decodeRequestId(idBytes) + }; +} + +function decodeRequestId(idBytes: Uint8Array): string { + try { + return REQUEST_ID_TEXT_DECODER.decode(idBytes); + } catch (error) { + throw new DaemonProtocolError( + 'malformedPayload', + 'Stdin frame request id is not valid UTF-8.', + { cause: error } + ); + } +} diff --git a/libraries/rush-daemon-protocol/src/index.ts b/libraries/rush-daemon-protocol/src/index.ts index b6b0841f12..96e586d52b 100644 --- a/libraries/rush-daemon-protocol/src/index.ts +++ b/libraries/rush-daemon-protocol/src/index.ts @@ -17,20 +17,31 @@ export type { IDaemonFrame } from './DaemonFrame'; export { DaemonFrameType, isDaemonFrameType } from './DaemonFrameType'; export { DEFAULT_MAX_PAYLOAD_BYTES, FRAME_HEADER_BYTES, LENGTH_FIELD_BYTES, LENGTH_FIELD_OFFSET, - MAX_OPERATION_ID_BYTES, OPERATION_ID_LENGTH_BYTES, OPERATION_ID_LENGTH_OFFSET, PAYLOAD_OFFSET, + MAX_OPERATION_ID_BYTES, MAX_REQUEST_ID_BYTES, OPERATION_ID_LENGTH_BYTES, OPERATION_ID_LENGTH_OFFSET, + PAYLOAD_OFFSET, REQUEST_ID_LENGTH_BYTES, REQUEST_ID_LENGTH_OFFSET, TYPE_FIELD_BYTES, TYPE_FIELD_OFFSET } from './FrameConstants'; export { encodeDaemonFrame, encodeDaemonFrames } from './FrameEncoder'; export { DaemonFrameDecoder, type IDaemonFrameDecoderOptions } from './FrameDecoder'; export { DaemonProtocolError, ProtocolVersionMismatchError } from './DaemonProtocolError'; export type { DaemonProtocolErrorCode, IDaemonProtocolErrorOptions } from './DaemonProtocolError'; -export { DAEMON_PROTOCOL_VERSION, isDaemonProtocolCompatible } from './DaemonProtocolVersion'; +export { + DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR, + DAEMON_PROTOCOL_VERSION, + isDaemonProtocolCompatible +} from './DaemonProtocolVersion'; export type { IDaemonProtocolVersion } from './DaemonProtocolVersion'; export type { IDaemonClientCaps } from './DaemonClientCaps'; -export { DAEMON_CONTROL_MESSAGE_KINDS, isDaemonControlMessageKind } from './DaemonControlMessage'; -export type { DaemonControlMessage, DaemonControlMessageKind, DaemonEmptyPayload } from './DaemonControlMessage'; +export { DAEMON_CONTROL_MESSAGE_KINDS, isDaemonControlMessageKind } from './DaemonControlKinds'; +export type { DaemonControlMessageKind } from './DaemonControlKinds'; +export type { DaemonControlMessage, DaemonEmptyPayload } from './DaemonControlMessage'; export type { IDaemonErrorMessage, IDaemonHelloAckMessage, IDaemonHelloMessage } from './DaemonControlMessage'; export type { IDaemonPingMessage, IDaemonSubscribeMessage, IDaemonUnsubscribeMessage } from './DaemonControlMessage'; +export type { + IDaemonRawModeChangedMessage, + IDaemonSetRawModeMessage, + IDaemonTerminalPolicyMessage +} from './DaemonInteractiveControl'; export type { IDaemonPongMessage } from './DaemonPongMessage'; export { isDaemonControlRecord, validateDaemonControlMessage } from './ControlMessageValidation'; export { decodeDaemonControlMessage, encodeDaemonControlMessage } from './ControlFrameCodec'; @@ -39,6 +50,17 @@ export { createDaemonHello, createDaemonHelloAck, negotiateDaemonHello } from '. export type { DaemonHandshakeOutcome } from './DaemonHandshake'; export type { DaemonJsonNull, DaemonJsonValue } from './DaemonJsonValue'; export type { DaemonCommandOutcome, IDaemonCommandResult } from './DaemonCommandResult'; +export type { + DaemonTerminalPolicyDecision, + DaemonTerminalPolicyReason, + DaemonTerminalRequirement, + IDaemonTerminalPolicyResult +} from './DaemonTerminalPolicy'; +export { + decodeDaemonStdinChunk, + encodeDaemonStdinChunk, + type IDaemonStdinChunk +} from './StdinFrameCodec'; export { DAEMON_EVENT_TYPES, isDaemonEventType, type DaemonEventType } from './DaemonEventType'; export type { DaemonEventPrivacy, IDaemonEventEnvelope, IDaemonEventScope, IDaemonEventSource } from './DaemonEventEnvelope'; export { isDaemonEventEnvelope, validateDaemonEventEnvelope } from './DaemonEventValidation'; diff --git a/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts b/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts index 11a854a0ba..ad32785fe9 100644 --- a/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts +++ b/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts @@ -14,7 +14,15 @@ const DAEMON_VERSION: string = '5.178.1'; const MESSAGES: readonly DaemonControlMessage[] = [ { kind: 'hello', payload: { protocolVersion: DAEMON_PROTOCOL_VERSION } }, { kind: 'helloAck', payload: { protocolVersion: DAEMON_PROTOCOL_VERSION, sessionId: 's-1' } }, - { kind: 'subscribe', payload: { isTTY: true, verbosity: 'verbose', columns: COLUMNS } }, + { + kind: 'subscribe', + payload: { + isTTY: true, + supportsInteractiveIO: true, + verbosity: 'verbose', + columns: COLUMNS + } + }, { kind: 'unsubscribe', payload: {} }, { kind: 'ping', payload: {} }, { kind: 'pong', payload: { uptimeMs: UPTIME_MS } }, diff --git a/libraries/rush-daemon-protocol/src/test/InteractiveControl.test.ts b/libraries/rush-daemon-protocol/src/test/InteractiveControl.test.ts new file mode 100644 index 0000000000..890401a6f9 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/test/InteractiveControl.test.ts @@ -0,0 +1,43 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { decodeDaemonControlMessage, encodeDaemonControlMessage } from '../ControlFrameCodec'; +import type { DaemonControlMessage } from '../DaemonControlMessage'; + +import { captureProtocolError } from './TestVectors'; + +const MESSAGES: readonly DaemonControlMessage[] = [ + { + kind: 'subscribe', + payload: { isTTY: true, supportsInteractiveIO: true } + }, + { kind: 'setRawMode', payload: { enabled: true, requestId: 'request-1' } }, + { kind: 'rawModeChanged', payload: { enabled: false, requestId: 'request-1' } }, + { + kind: 'terminalPolicy', + payload: { + decision: 'requiresInProcess', + reason: 'controllingTerminalRequired', + requestId: 'request-1' + } + } +]; + +it('round-trips interactive capability and request-scoped controls', () => { + for (const message of MESSAGES) { + expect(decodeDaemonControlMessage(encodeDaemonControlMessage(message))).toEqual(message); + } +}); + +it('rejects malformed interactive capability and control messages', () => { + const invalidMessages: readonly string[] = [ + '{"kind":"subscribe","payload":{"isTTY":true,"supportsInteractiveIO":"yes"}}', + '{"kind":"setRawMode","payload":{"enabled":"yes","requestId":"r"}}', + '{"kind":"terminalPolicy","payload":{"decision":"allocatePty","requestId":"r"}}' + ]; + for (const json of invalidMessages) { + expect(captureProtocolError(() => decodeDaemonControlMessage(Buffer.from(json))).code).toBe( + 'malformedControlMessage' + ); + } +}); diff --git a/libraries/rush-daemon-protocol/src/test/StdinFrameCodec.test.ts b/libraries/rush-daemon-protocol/src/test/StdinFrameCodec.test.ts new file mode 100644 index 0000000000..c4607552c3 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/test/StdinFrameCodec.test.ts @@ -0,0 +1,78 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonFrame } from '../DaemonFrame'; +import { DaemonFrameType } from '../DaemonFrameType'; +import { MAX_REQUEST_ID_BYTES } from '../FrameConstants'; +import { DaemonFrameDecoder } from '../FrameDecoder'; +import { encodeDaemonFrame } from '../FrameEncoder'; +import { decodeDaemonStdinChunk, encodeDaemonStdinChunk } from '../StdinFrameCodec'; + +import { FIRST_INDEX, NON_UTF8_BYTES, SINGLE_COUNT, captureProtocolError } from './TestVectors'; + +const EMPTY_BYTES: number = 0; +const FIRST_INPUT_BYTE: number = 0xff; +const SECOND_INPUT_BYTE: number = 0x00; +const THIRD_INPUT_BYTE: number = 0x80; +const SPLIT_OFFSET: number = 3; +const TOO_LONG_ID_BYTES: number = MAX_REQUEST_ID_BYTES + SINGLE_COUNT; + +function expectBytesEqual(actual: Uint8Array, expected: Uint8Array): void { + expect(Buffer.from(actual).equals(Buffer.from(expected))).toBe(true); +} + +it('round-trips request-tagged non-UTF-8 stdin without transforming bytes', () => { + const decoded: ReturnType = decodeDaemonStdinChunk( + encodeDaemonStdinChunk({ chunk: NON_UTF8_BYTES, requestId: 'request-a' }) + ); + expect(decoded.requestId).toBe('request-a'); + expectBytesEqual(decoded.chunk, NON_UTF8_BYTES); +}); + +it('preserves split and interleaved stdin frame boundaries', () => { + const decoder: DaemonFrameDecoder = new DaemonFrameDecoder(); + const frames: Uint8Array[] = [ + encodeDaemonFrame({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ chunk: Uint8Array.of(FIRST_INPUT_BYTE), requestId: 'request-a' }) + }), + encodeDaemonFrame({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ chunk: Uint8Array.of(SECOND_INPUT_BYTE), requestId: 'request-b' }) + }), + encodeDaemonFrame({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ chunk: Uint8Array.of(THIRD_INPUT_BYTE), requestId: 'request-a' }) + }) + ]; + const wire: Buffer = Buffer.concat(frames.map((frame: Uint8Array) => Buffer.from(frame))); + const decodedFrames: IDaemonFrame[] = [ + ...decoder.push(wire.subarray(FIRST_INDEX, SPLIT_OFFSET)), + ...decoder.push(wire.subarray(SPLIT_OFFSET)) + ]; + const decoded = decodedFrames.map((frame: IDaemonFrame) => decodeDaemonStdinChunk(frame.payload)); + expect(decoded.map(({ requestId }) => requestId)).toEqual(['request-a', 'request-b', 'request-a']); + expect(decoded.map(({ chunk }) => chunk[FIRST_INDEX])).toEqual([ + FIRST_INPUT_BYTE, + SECOND_INPUT_BYTE, + THIRD_INPUT_BYTE + ]); +}); + +it('rejects malformed request id prefixes', () => { + expect(() => encodeDaemonStdinChunk({ + chunk: new Uint8Array(EMPTY_BYTES), + requestId: 'x'.repeat(TOO_LONG_ID_BYTES) + })).toThrow(); + expect(captureProtocolError(() => decodeDaemonStdinChunk(Uint8Array.of(SINGLE_COUNT))).code).toBe( + 'malformedPayload' + ); + const malformedIdPayload: Uint8Array = Uint8Array.of( + NON_UTF8_BYTES.length, + EMPTY_BYTES, + ...NON_UTF8_BYTES + ); + expect(captureProtocolError(() => decodeDaemonStdinChunk(malformedIdPayload)).code).toBe( + 'malformedPayload' + ); +}); diff --git a/libraries/rush-daemon-transport/README.md b/libraries/rush-daemon-transport/README.md index 52e6b64d79..ce43f12231 100644 --- a/libraries/rush-daemon-transport/README.md +++ b/libraries/rush-daemon-transport/README.md @@ -11,7 +11,7 @@ The workspace-keyed socket/pipe **transport** for the Rush daemon (`rushd`): `\\.\pipe\rushd-` named pipes on Windows. - **`net` listener and connector** — framed with [`@rushstack/rush-daemon-protocol`](https://www.npmjs.com/package/@rushstack/rush-daemon-protocol), - with backpressure-aware writes. + with backpressure-aware writes and serialized async frame handlers for inbound flow control. - **PID/lockfile handling** — stale sockets and dead PIDs are detected (two-factor: PID liveness plus a connect probe) and reclaimed without manual cleanup. diff --git a/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts b/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts index dce6c5a545..2ac6417002 100644 --- a/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts +++ b/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts @@ -18,26 +18,24 @@ import { DaemonTransportError, DaemonTransportErrorCode } from './DaemonTranspor export class DaemonFrameConnection { private readonly _socket: net.Socket; private readonly _decoder: DaemonFrameDecoder = new DaemonFrameDecoder(); - private _frameHandler: ((frame: IDaemonFrame) => void) | undefined; + private _frameHandler: ((frame: IDaemonFrame) => void | Promise) | undefined; private _closedHandler: ((error: Error | undefined) => void) | undefined; private _closedError: Error | undefined; - + private _receiveQueue: Promise = Promise.resolve(); public constructor(socket: net.Socket) { this._socket = socket; socket.on('data', (chunk: Buffer) => this._onData(chunk)); socket.on('error', (error: Error) => this._onError(error)); socket.on('close', () => this._onClose()); } - - /** Registers the frame handler invoked for each decoded frame. */ - public onFrame(handler: (frame: IDaemonFrame) => void): void { + /** Registers the serialized, backpressured handler invoked for each decoded frame. */ + public onFrame(handler: (frame: IDaemonFrame) => void | Promise): void { this._frameHandler = handler; } /** Registers the close handler, invoked at most once with the cause. */ public onClosed(handler: (error: Error | undefined) => void): void { this._closedHandler = handler; } - /** Encodes and writes a frame, resolving when the socket has drained it. @throws {@link DaemonTransportError} when closed. */ public async sendFrameAsync(frame: IDaemonFrame): Promise { this._assertOpen(); @@ -45,13 +43,11 @@ export class DaemonFrameConnection { await once(this._socket, 'drain'); } } - /** Half-closes the writable side and releases the socket. */ public async closeAsync(): Promise { this._socket.end(); this._socket.destroySoon(); } - /** The wrapped socket, for the internal raw-write test hook. @internal */ public get socket(): net.Socket { return this._socket; @@ -73,15 +69,18 @@ export class DaemonFrameConnection { this._fail(error); return; } - for (const frame of frames) { - this._dispatchFrame(frame); - } + this._socket.pause(); + this._receiveQueue = this._receiveQueue + .then(() => this._dispatchFramesAsync(frames)) + .then(() => { + this._socket.resume(); + }) + .catch((error: unknown) => this._fail(error)); } - private _dispatchFrame(frame: IDaemonFrame): void { - try { - this._frameHandler?.(frame); - } catch (error) { - this._fail(error); + + private async _dispatchFramesAsync(frames: ReadonlyArray): Promise { + for (const frame of frames) { + await this._frameHandler?.(frame); } } diff --git a/libraries/rush-daemon-transport/src/test/AsyncFrameHandler.test.ts b/libraries/rush-daemon-transport/src/test/AsyncFrameHandler.test.ts new file mode 100644 index 0000000000..eb0ae25f3d --- /dev/null +++ b/libraries/rush-daemon-transport/src/test/AsyncFrameHandler.test.ts @@ -0,0 +1,68 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { DaemonFrameType } from '@rushstack/rush-daemon-protocol'; + +import type { DaemonFrameConnection } from '../DaemonFrameConnection'; +import type { IDaemonPaths } from '../DaemonPaths'; + +import { createDeferred, createTestDaemonPaths, startTestDaemonPair } from './TestDaemonFixture'; +import type { IDeferred, ITestDaemonPair } from './TestDaemonFixture'; + +const EMPTY_TOTAL: number = 0; +const FIRST_COUNT: number = 1; +const FRAME_COUNT: number = 3; + +interface IHandlerState { + active: number; + maximumActive: number; + received: number; +} + +function createHandler( + state: IHandlerState, + firstHandler: IDeferred, + firstReceived: IDeferred, + allReceived: IDeferred +): () => Promise { + return async (): Promise => { + state.active++; + state.maximumActive = Math.max(state.maximumActive, state.active); + state.received++; + if (state.received === FIRST_COUNT) { + firstReceived.resolve(); + await firstHandler.promise; + } + state.active--; + if (state.received === FRAME_COUNT) allReceived.resolve(); + }; +} + +async function sendFramesAsync(server: DaemonFrameConnection): Promise { + await Promise.all( + Array.from({ length: FRAME_COUNT }, () => + server.sendFrameAsync({ kind: DaemonFrameType.stdin, payload: new Uint8Array() }) + ) + ); +} + +it('awaits each incoming frame handler before dispatching the next frame', async () => { + const paths: IDaemonPaths = createTestDaemonPaths(); + const pair: ITestDaemonPair = await startTestDaemonPair(paths); + const firstHandler: IDeferred = createDeferred(); + const firstReceived: IDeferred = createDeferred(); + const allReceived: IDeferred = createDeferred(); + const state: IHandlerState = { active: EMPTY_TOTAL, maximumActive: EMPTY_TOTAL, received: EMPTY_TOTAL }; + try { + pair.client.onFrame(createHandler(state, firstHandler, firstReceived, allReceived)); + await sendFramesAsync(await pair.serverSide); + await firstReceived.promise; + expect(state.received).toBe(FIRST_COUNT); + firstHandler.resolve(); + await allReceived.promise; + expect(state.maximumActive).toBe(FIRST_COUNT); + } finally { + await pair.client.closeAsync(); + await pair.listener.closeAsync(); + } +}); diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index e9390f22ba..dc682030d0 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -59,5 +59,15 @@ The existing `RushCommandLineParser`, `BaseRushAction`, and some built-in/global process-global state. This layer therefore does not pretend that arbitrary existing actions are daemon-safe: the integration must supply already resolved command logic that consumes `IGlobalCommandExecutionContext`, including `spawnChild()` for command-local subprocesses. Adapting the complete action surface remains bounded by the open -[rushstack#5895](https://github.com/microsoft/rushstack/issues/5895) engine/action prerequisite work. Interactive -stdin/raw-mode/PTY support, scheduling classification, and shared-build merging belong to later layers. +[rushstack#5895](https://github.com/microsoft/rushstack/issues/5895) engine/action prerequisite work. `InteractiveRequestInputRouter` supplies the opt-in WS2.7 boundary for connection-scoped input. The WS1 stdin +frame carries a request identifier plus untouched raw bytes; frames are serialized per request through an injected +sink while separate requests remain isolated. Global command integrations can bind that sink directly to a spawned +child process. Both global and phased routes stop accepting input on abort/disconnect and await input drain plus an +acknowledged cooked-mode restoration before publishing the exact-once command result. The daemon never reads or +mutates its own stdin or raw-mode state. + +Terminal width remains the immutable request-start value established by WS2.5. The thin client owns resize and +rendering, so this layer does not forward `SIGWINCH`. Commands declaring a real controlling-terminal requirement +receive a typed `requiresInProcess` policy result and are not executed by rushd; no pseudo-terminal is allocated or +emulated. The future WS4 client will perform the actual in-process fallback. Scheduling classification and +shared-build merging remain later layers. diff --git a/libraries/rush-daemon/src/DaemonControlSession.ts b/libraries/rush-daemon/src/DaemonControlSession.ts index 85c3821132..55b79b5ab7 100644 --- a/libraries/rush-daemon/src/DaemonControlSession.ts +++ b/libraries/rush-daemon/src/DaemonControlSession.ts @@ -5,6 +5,7 @@ import { randomUUID } from 'node:crypto'; import { DAEMON_PROTOCOL_VERSION, + DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR, DaemonFrameType, DaemonProtocolError, decodeDaemonControlMessage, @@ -19,24 +20,43 @@ import type { } from '@rushstack/rush-daemon-protocol'; import type { DaemonFrameConnection } from '@rushstack/rush-daemon-transport'; +import { + DaemonInteractiveConnection +} from './DaemonInteractiveConnection'; +import type { IDaemonInteractiveConnection } from './DaemonInteractiveConnection'; +import { + InteractiveInputRoutingError, + isInteractiveRequestInputFailure +} from './InteractiveRequestInputRouter'; + export interface IDaemonControlSessionOptions { readonly daemonVersion: string; readonly startedAtMs: number; + readonly onInteractiveConnection?: (connection: IDaemonInteractiveConnection) => void; readonly onClosed: (session: DaemonControlSession, error: Error | undefined) => void; readonly onError: (error: Error) => void; } export class DaemonControlSession { private readonly _connection: DaemonFrameConnection; + private readonly _interactiveConnection: DaemonInteractiveConnection; private readonly _options: IDaemonControlSessionOptions; private _handshakeComplete: boolean = false; + private _peerSupportsInteractiveProtocol: boolean = false; private _sendQueue: Promise = Promise.resolve(); public constructor(connection: DaemonFrameConnection, options: IDaemonControlSessionOptions) { this._connection = connection; this._options = options; + this._interactiveConnection = new DaemonInteractiveConnection( + (message: DaemonControlMessage) => this._enqueueSendAsync(message) + ); connection.onFrame((frame: IDaemonFrame) => this._onFrame(frame)); - connection.onClosed((error: Error | undefined) => options.onClosed(this, error)); + connection.onClosed((error: Error | undefined) => { + this._interactiveConnection.close(error); + options.onClosed(this, error); + }); + options.onInteractiveConnection?.(this._interactiveConnection); } public closeAsync(): Promise { @@ -44,6 +64,16 @@ export class DaemonControlSession { } private _onFrame(frame: IDaemonFrame): void { + if (frame.kind === DaemonFrameType.stdin) { + if (!this._handshakeComplete) { + throw new DaemonProtocolError( + 'malformedControlMessage', + 'The first frame on a connection must be a hello control message.' + ); + } + void this._completeInputAsync(this._interactiveConnection.routeStdinFrameAsync(frame.payload)); + return; + } if (frame.kind !== DaemonFrameType.controlJson) { throw new DaemonProtocolError( 'malformedControlMessage', @@ -53,6 +83,12 @@ export class DaemonControlSession { const message: DaemonControlMessage = decodeDaemonControlMessage(frame.payload); if (!this._handshakeComplete) { this._handleHello(message); + } else if (this._interactiveConnection.handleControlMessage(message)) { + return; + } else if (message.kind === 'subscribe') { + this._interactiveConnection.setEnabled( + this._peerSupportsInteractiveProtocol && message.payload.supportsInteractiveIO === true + ); } else if (message.kind === 'ping') { this._send(this._createPong()); } else { @@ -77,6 +113,8 @@ export class DaemonControlSession { ); if (outcome.accepted) { this._handshakeComplete = true; + this._peerSupportsInteractiveProtocol = + message.payload.protocolVersion.minor >= DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR; this._send(outcome.ack); } else { const errorMessage: IDaemonErrorMessage = { @@ -99,14 +137,24 @@ export class DaemonControlSession { } private _send(message: DaemonControlMessage, closeAfterSend: boolean = false): void { + void this._enqueueSendAsync(message, closeAfterSend).catch((error: unknown) => + this._handleSendErrorAsync(error) + ); + } + + private _enqueueSendAsync( + message: DaemonControlMessage, + closeAfterSend: boolean = false + ): Promise { const frame: IDaemonFrame = { kind: DaemonFrameType.controlJson, payload: encodeDaemonControlMessage(message) }; - this._sendQueue = this._sendQueue + const sendPromise: Promise = this._sendQueue .then(() => this._connection.sendFrameAsync(frame)) - .then(() => (closeAfterSend ? this._connection.closeAsync() : undefined)) - .catch((error: unknown) => this._handleSendErrorAsync(error)); + .then(() => (closeAfterSend ? this._connection.closeAsync() : undefined)); + this._sendQueue = sendPromise.catch(() => undefined); + return sendPromise; } private async _handleSendErrorAsync(error: unknown): Promise { @@ -114,4 +162,17 @@ export class DaemonControlSession { this._options.onError(normalizedError); await this._connection.closeAsync(); } + + private async _completeInputAsync(inputPromise: Promise): Promise { + try { + await inputPromise; + } catch (error) { + if ( + !isInteractiveRequestInputFailure(error) && + !(error instanceof InteractiveInputRoutingError && error.code === 'completedRequest') + ) { + await this._handleSendErrorAsync(error); + } + } + } } diff --git a/libraries/rush-daemon/src/DaemonInteractiveConnection.ts b/libraries/rush-daemon/src/DaemonInteractiveConnection.ts new file mode 100644 index 0000000000..e1ceeae23e --- /dev/null +++ b/libraries/rush-daemon/src/DaemonInteractiveConnection.ts @@ -0,0 +1,251 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + DaemonControlMessage, + IDaemonRawModeChangedMessage, + IDaemonSetRawModeMessage, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; + +import { + InteractiveRequestInputRouter +} from './InteractiveRequestInputRouter'; +import type { + IInteractiveRequestControlClient, + IInteractiveRequestSession +} from './InteractiveRequestInputRouter'; + +interface IRawModeAcknowledgement { + readonly enabled: boolean; + readonly reject: (error: Error) => void; + readonly resolve: () => void; +} + +interface IRawModeWaiter { + readonly acknowledgement: IRawModeAcknowledgement; + readonly promise: Promise; +} + +/** Options for registering one command with its connection-owned input router. @beta */ +export interface IDaemonInteractiveRequestOptions { + readonly abortSignal: AbortSignal; + readonly acceptsStdin: boolean; + readonly onFailure: (error: Error) => void; + readonly requestId: string; +} + +/** Interactive I/O services owned by one live daemon connection. @beta */ +export interface IDaemonInteractiveConnection { + readonly abortSignal: AbortSignal; + registerRequest(options: IDaemonInteractiveRequestOptions): IInteractiveRequestSession; + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; +} + +type SendDaemonControlMessageAsync = (message: DaemonControlMessage) => Promise; + +/** Implements request input and acknowledged raw-mode control for one live connection. @internal */ +export class DaemonInteractiveConnection implements IDaemonInteractiveConnection { + readonly #abandonedRawModeEnableRequestIds: Set = new Set(); + readonly #abortController: AbortController = new AbortController(); + readonly #inputRouter: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + readonly #pendingRawModeByRequestId: Map = new Map(); + readonly #sendControlMessageAsync: SendDaemonControlMessageAsync; + #enabled: boolean = false; + #rawModeOwnerRequestId: string | undefined; + #rawModeTail: Promise = Promise.resolve(); + + public constructor(sendControlMessageAsync: SendDaemonControlMessageAsync) { + this.#sendControlMessageAsync = sendControlMessageAsync; + } + + public get abortSignal(): AbortSignal { + return this.#abortController.signal; + } + + public setEnabled(enabled: boolean): void { + this.#enabled = enabled; + } + + public registerRequest(options: IDaemonInteractiveRequestOptions): IInteractiveRequestSession { + if (options.acceptsStdin) { + this.#assertEnabled(); + } + const requestAbortSignal: AbortSignal = AbortSignal.any([this.abortSignal, options.abortSignal]); + const client: IInteractiveRequestControlClient = { + abortSignal: requestAbortSignal, + writeRawModeControlAsync: (message: IDaemonSetRawModeMessage): Promise => + this.#queueRawModeControlAsync(message, requestAbortSignal) + }; + return this.#inputRouter.register({ ...options, client }); + } + + public async routeStdinFrameAsync(payload: Uint8Array): Promise { + this.#assertEnabled(); + await this.#inputRouter.routeStdinFrameAsync(payload); + } + + public handleControlMessage(message: DaemonControlMessage): boolean { + if (message.kind !== 'rawModeChanged') { + return false; + } + this.#assertEnabled(); + this.#acknowledgeRawMode(message); + return true; + } + + public writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise { + this.#assertEnabled(); + return this.#sendControlMessageAsync({ kind: 'terminalPolicy', payload: result }); + } + + public close(error: Error | undefined): void { + const reason: Error = error ?? new Error('The daemon client connection closed.'); + this.#abortController.abort(reason); + for (const acknowledgement of this.#pendingRawModeByRequestId.values()) { + acknowledgement.reject(reason); + } + this.#abandonedRawModeEnableRequestIds.clear(); + this.#pendingRawModeByRequestId.clear(); + this.#rawModeOwnerRequestId = undefined; + } + + #queueRawModeControlAsync( + message: IDaemonSetRawModeMessage, + requestAbortSignal: AbortSignal + ): Promise { + const transition: Promise = this.#rawModeTail.then(() => + this.#applyRawModeControlAsync(message, requestAbortSignal) + ); + this.#rawModeTail = transition.catch(() => undefined); + return transition; + } + + async #applyRawModeControlAsync( + message: IDaemonSetRawModeMessage, + requestAbortSignal: AbortSignal + ): Promise { + this.#assertEnabled(); + const { enabled, requestId } = message.payload; + if (enabled) { + if (requestAbortSignal.aborted) { + throw normalizeAbortReason(requestAbortSignal); + } + if (this.#rawModeOwnerRequestId === requestId) { + return; + } + if (this.#rawModeOwnerRequestId !== undefined) { + throw new Error( + `Raw mode is already owned by interactive request "${this.#rawModeOwnerRequestId}".` + ); + } + this.#rawModeOwnerRequestId = requestId; + await this.#sendRawModeControlAsync(message, requestAbortSignal); + } else if (this.#rawModeOwnerRequestId === requestId) { + await this.#sendRawModeControlAsync(message, this.abortSignal); + this.#rawModeOwnerRequestId = undefined; + } + } + + async #sendRawModeControlAsync( + message: IDaemonSetRawModeMessage, + abortSignal: AbortSignal + ): Promise { + if (this.#pendingRawModeByRequestId.has(message.payload.requestId)) { + throw new Error(`Request "${message.payload.requestId}" already has a pending raw-mode change.`); + } + if (abortSignal.aborted) { + throw normalizeAbortReason(abortSignal); + } + const { acknowledgement, promise }: IRawModeWaiter = createRawModeWaiter( + message.payload.enabled, + abortSignal, + () => this.#pendingRawModeByRequestId.delete(message.payload.requestId) + ); + this.#pendingRawModeByRequestId.set(message.payload.requestId, acknowledgement); + if (abortSignal.aborted) { + acknowledgement.reject(normalizeAbortReason(abortSignal)); + return await promise; + } + let sendStarted: boolean = false; + try { + sendStarted = true; + await this.#sendControlMessageAsync(message); + } catch (error) { + acknowledgement.reject(normalizeError(error)); + } + try { + await promise; + } catch (error) { + if ( + message.payload.enabled && + sendStarted && + abortSignal.aborted && + !this.abortSignal.aborted + ) { + this.#abandonedRawModeEnableRequestIds.add(message.payload.requestId); + } + throw error; + } + } + + #acknowledgeRawMode(message: IDaemonRawModeChangedMessage): void { + if ( + message.payload.enabled && + this.#abandonedRawModeEnableRequestIds.delete(message.payload.requestId) + ) { + return; + } + const acknowledgement: IRawModeAcknowledgement | undefined = + this.#pendingRawModeByRequestId.get(message.payload.requestId); + if (!acknowledgement || acknowledgement.enabled !== message.payload.enabled) { + throw new Error(`Unexpected raw-mode acknowledgement for request "${message.payload.requestId}".`); + } + acknowledgement.resolve(); + } + + #assertEnabled(): void { + if (!this.#enabled) { + throw new Error('The daemon client did not negotiate request-scoped interactive I/O.'); + } + if (this.abortSignal.aborted) { + throw this.abortSignal.reason; + } + } +} + +function createRawModeWaiter( + enabled: boolean, + abortSignal: AbortSignal, + onSettled: () => void +): IRawModeWaiter { + let resolvePromise: () => void = () => undefined; + let rejectPromise: (error: Error) => void = () => undefined; + const promise: Promise = new Promise((resolve, reject) => { + resolvePromise = resolve; + rejectPromise = reject; + }); + let removeAbortListener: () => void = () => undefined; + const settle = (callback: () => void): void => { + removeAbortListener(); + onSettled(); + callback(); + }; + const acknowledgement: IRawModeAcknowledgement = { + enabled, + reject: (error: Error) => settle(() => rejectPromise(error)), + resolve: () => settle(resolvePromise) + }; + const onAbort = (): void => acknowledgement.reject(normalizeAbortReason(abortSignal)); + removeAbortListener = () => abortSignal.removeEventListener('abort', onAbort); + abortSignal.addEventListener('abort', onAbort, { once: true }); + return { acknowledgement, promise }; +} + +function normalizeAbortReason(abortSignal: AbortSignal): Error { + return normalizeError(abortSignal.reason ?? new Error('The interactive request was aborted.')); +} + +function normalizeError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} diff --git a/libraries/rush-daemon/src/DaemonTerminalPolicy.ts b/libraries/rush-daemon/src/DaemonTerminalPolicy.ts new file mode 100644 index 0000000000..272a848e30 --- /dev/null +++ b/libraries/rush-daemon/src/DaemonTerminalPolicy.ts @@ -0,0 +1,37 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + DaemonTerminalRequirement, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; + +/** + * A typed signal that a thin client must execute the command in-process. + * + * @beta + */ +export class DaemonRequiresInProcessError extends Error { + public readonly policy: IDaemonTerminalPolicyResult; + + public constructor(policy: IDaemonTerminalPolicyResult) { + super('The command requires a real controlling terminal and cannot run in the Rush daemon.'); + this.name = 'DaemonRequiresInProcessError'; + this.policy = policy; + } +} + +/** Evaluates the terminal requirement without probing or mutating daemon stdio. @beta */ +export function evaluateDaemonTerminalPolicy( + requestId: string, + requirement: DaemonTerminalRequirement = 'none' +): IDaemonTerminalPolicyResult { + if (requirement === 'controllingTerminal') { + return { + decision: 'requiresInProcess', + reason: 'controllingTerminalRequired', + requestId + }; + } + return { decision: 'runInDaemon', requestId }; +} diff --git a/libraries/rush-daemon/src/GlobalCommandExecutionContext.ts b/libraries/rush-daemon/src/GlobalCommandExecutionContext.ts index 879f7527fd..c9ee9425d8 100644 --- a/libraries/rush-daemon/src/GlobalCommandExecutionContext.ts +++ b/libraries/rush-daemon/src/GlobalCommandExecutionContext.ts @@ -15,6 +15,10 @@ import { type IResolvedGlobalCommandRequest } from './GlobalCommandRequest'; import type { IGlobalCommandRequestClient } from './GlobalCommandRequestClient'; +import type { + IInteractiveRequestInputSink, + IInteractiveRequestSession +} from './InteractiveRequestInputRouter'; import type { IWorkspaceSession } from './WorkspaceSession'; const MAX_PENDING_TERMINAL_BYTES: number = 1024 * 1024; @@ -26,6 +30,7 @@ const MAX_PENDING_TERMINAL_BYTES: number = 1024 * 1024; */ export interface IGlobalCommandSpawnOptions { readonly environmentOverlay?: Readonly; + readonly forwardInput?: boolean; readonly forwardOutput?: boolean; readonly shell?: boolean | string; readonly windowsHide?: boolean; @@ -40,6 +45,7 @@ export interface IGlobalCommandExecutionContext { readonly abortSignal: AbortSignal; readonly cwd: string; readonly environment: IGlobalCommandEnvironment; + readonly interactiveInput: IInteractiveRequestSession | undefined; readonly terminal: ITerminal; readonly terminalProperties: IGlobalCommandTerminalProperties; readonly workspaceSession: IWorkspaceSession; @@ -159,6 +165,7 @@ export class GlobalCommandExecutionContext #requestAborted: boolean = false; public readonly terminal: ITerminal; + public readonly interactiveInput: IInteractiveRequestSession | undefined; public readonly workspaceSession: IWorkspaceSession; public constructor( @@ -168,6 +175,7 @@ export class GlobalCommandExecutionContext ) { this.#request = request; this.#client = client; + this.interactiveInput = client.interactiveSession; this.workspaceSession = workspaceSession; this.#onClientAbort = () => this.#abortRequest(client.abortSignal.reason); this.#writer = new OrderedTerminalWriter(client, (error: Error) => this.#abortRequest(error)); @@ -231,6 +239,9 @@ export class GlobalCommandExecutionContext const trackedChild: ITrackedChild = { completion }; this.#trackedChildren.add(trackedChild); void completion.then(() => this.#trackedChildren.delete(trackedChild)); + if (options.forwardInput === true) { + this.#attachChildInput(child); + } if (options.forwardOutput !== false) { this.#forwardChildOutput(child.stdout, 'stdout'); this.#forwardChildOutput(child.stderr, 'stderr'); @@ -238,6 +249,22 @@ export class GlobalCommandExecutionContext return child; } + #attachChildInput(child: childProcess.ChildProcessWithoutNullStreams): void { + if (!this.interactiveInput) { + throw new Error('The global command did not register an interactive input session.'); + } + const sink: IInteractiveRequestInputSink = { + writeInputAsync: (chunk: Uint8Array): Promise => writeChildInputAsync(child, chunk) + }; + const attachment: Disposable = this.interactiveInput.attachInputSink(sink); + this.registerDisposable({ + [Symbol.asyncDispose]: (): Promise => { + attachment[Symbol.dispose](); + return Promise.resolve(); + } + }); + } + public async [Symbol.asyncDispose](): Promise { if (this.#closed) { return; @@ -354,3 +381,18 @@ function throwCleanupErrors(cleanupErrors: unknown[]): void { function normalizeError(error: unknown): Error { return error instanceof Error ? error : new Error(String(error)); } + +function writeChildInputAsync( + child: childProcess.ChildProcessWithoutNullStreams, + chunk: Uint8Array +): Promise { + return new Promise((resolve, reject) => { + child.stdin.write(chunk, (error: Error | null | undefined) => { + if (error) { + reject(error); + } else { + resolve(); + } + }); + }); +} diff --git a/libraries/rush-daemon/src/GlobalCommandRequest.ts b/libraries/rush-daemon/src/GlobalCommandRequest.ts index 7969d525d6..3e12693ea7 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequest.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequest.ts @@ -5,6 +5,7 @@ import * as fs from 'node:fs'; import * as path from 'node:path'; import { EnvironmentMap } from '@rushstack/node-core-library'; +import type { DaemonTerminalRequirement } from '@rushstack/rush-daemon-protocol'; import type { IWorkspaceSession } from './WorkspaceSession'; @@ -14,8 +15,10 @@ import type { IWorkspaceSession } from './WorkspaceSession'; * @beta */ export interface IGlobalCommandTerminalProperties { + readonly acceptsStdin?: boolean; readonly columns: number | undefined; readonly isTTY: boolean; + readonly terminalRequirement?: DaemonTerminalRequirement; readonly supportsColor: boolean; } @@ -170,9 +173,25 @@ function resolveTerminalProperties( if (typeof terminal.isTTY !== 'boolean' || typeof terminal.supportsColor !== 'boolean') { throw new Error('Global command terminal TTY and color properties must be boolean values.'); } + if (terminal.acceptsStdin !== undefined && typeof terminal.acceptsStdin !== 'boolean') { + throw new Error('Global command terminal acceptsStdin must be a boolean value.'); + } + if ( + terminal.terminalRequirement !== undefined && + terminal.terminalRequirement !== 'none' && + terminal.terminalRequirement !== 'interactiveInput' && + terminal.terminalRequirement !== 'controllingTerminal' + ) { + throw new Error('Global command terminal requirement is not recognized.'); + } + if (terminal.terminalRequirement === 'interactiveInput' && terminal.acceptsStdin !== true) { + throw new Error('Global command interactive input requires acceptsStdin to be true.'); + } return Object.freeze({ + acceptsStdin: terminal.acceptsStdin ?? false, columns: terminal.columns, isTTY: terminal.isTTY, + terminalRequirement: terminal.terminalRequirement ?? 'none', supportsColor: terminal.supportsColor }); } diff --git a/libraries/rush-daemon/src/GlobalCommandRequestClient.ts b/libraries/rush-daemon/src/GlobalCommandRequestClient.ts index 2f922e4a8c..565661d290 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequestClient.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequestClient.ts @@ -1,7 +1,12 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. -import type { IDaemonCommandResult } from '@rushstack/rush-daemon-protocol'; +import type { + IDaemonCommandResult, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; + +import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; /** * A client-scoped destination for one global command request. @@ -15,10 +20,15 @@ import type { IDaemonCommandResult } from '@rushstack/rush-daemon-protocol'; export interface IGlobalCommandRequestClient { /** Aborted by the transport when the request is cancelled or disconnected. */ readonly abortSignal: AbortSignal; + /** The request-scoped stdin/control lifecycle when one was registered by the transport integration. */ + readonly interactiveSession?: IInteractiveRequestSession; /** Writes one request-scoped terminal chunk through the client's backpressured destination. */ writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; + /** Signals that the client must execute this request in-process instead. */ + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; + /** Writes the final command result after every preceding terminal chunk has drained. */ writeResultAsync(result: IDaemonCommandResult): Promise; } diff --git a/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts index 15377f2773..dbe064de3f 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts @@ -1,7 +1,10 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. -import type { IDaemonCommandResult } from '@rushstack/rush-daemon-protocol'; +import type { + IDaemonCommandResult, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; import { createGlobalCommandResult } from './CommandResultPolicy'; import type { IGlobalCommandExecutionContext } from './GlobalCommandExecutionContext'; @@ -13,6 +16,11 @@ import { validateResolvedGlobalCommandRequest } from './GlobalCommandRequest'; import type { IGlobalCommandRequestClient } from './GlobalCommandRequestClient'; +import { + DaemonRequiresInProcessError, + evaluateDaemonTerminalPolicy +} from './DaemonTerminalPolicy'; +import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; import type { IWorkspaceSession } from './WorkspaceSession'; /** @@ -72,6 +80,19 @@ export class GlobalCommandRequestRouter { client: IGlobalCommandRequestClient ): Promise { validateResolvedGlobalCommandRequest(request, this.#workspaceSession); + const interactiveSession: IInteractiveRequestSession | undefined = validateInteractiveSession( + request, + client + ); + const policy: IDaemonTerminalPolicyResult = evaluateDaemonTerminalPolicy( + request.requestId, + request.terminal.terminalRequirement + ); + if (policy.decision === 'requiresInProcess') { + await interactiveSession?.finishAsync(); + await client.writeTerminalPolicyAsync(policy); + throw new DaemonRequiresInProcessError(policy); + } const context: GlobalCommandExecutionContext = new GlobalCommandExecutionContext( request, client, @@ -103,6 +124,11 @@ export class GlobalCommandRequestRouter { } catch (error) { cleanupError = error; } + try { + await interactiveSession?.finishAsync(); + } catch (error) { + cleanupError = combineExecutionAndCleanupErrors(cleanupError, error); + } aborted ||= context.requestAborted; const combinedError: unknown = combineExecutionAndCleanupErrors(executionError, cleanupError); let result: IDaemonCommandResult; @@ -121,6 +147,20 @@ export class GlobalCommandRequestRouter { requestId: request.requestId }); } + + function validateInteractiveSession( + resolvedRequest: IResolvedGlobalCommandRequest, + requestClient: IGlobalCommandRequestClient + ): IInteractiveRequestSession | undefined { + const session: IInteractiveRequestSession | undefined = requestClient.interactiveSession; + if (session && session.requestId !== resolvedRequest.requestId) { + throw new Error('The interactive input session does not belong to the global command request.'); + } + if (resolvedRequest.terminal.acceptsStdin === true && !session) { + throw new Error('The interactive global command does not have a registered input session.'); + } + return session; + } await client.writeResultAsync(result); return result; } diff --git a/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts b/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts new file mode 100644 index 0000000000..2a52f9599b --- /dev/null +++ b/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts @@ -0,0 +1,340 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { decodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonSetRawModeMessage } from '@rushstack/rush-daemon-protocol'; + +/** Why an incoming stdin frame cannot be routed. @beta */ +export type InteractiveInputRoutingErrorCode = + | 'duplicateRequest' + | 'unknownRequest' + | 'completedRequest' + | 'nonInteractiveRequest'; + +/** A request-scoped stdin routing failure. @beta */ +export class InteractiveInputRoutingError extends Error { + public readonly code: InteractiveInputRoutingErrorCode; + + public constructor(code: InteractiveInputRoutingErrorCode, message: string) { + super(message); + this.name = 'InteractiveInputRoutingError'; + this.code = code; + } +} + +/** A backpressured destination for arbitrary stdin bytes. @beta */ +export interface IInteractiveRequestInputSink { + writeInputAsync(chunk: Uint8Array): Promise; +} + +/** A client-side terminal control destination for one daemon request. @beta */ +export interface IInteractiveRequestControlClient { + readonly abortSignal: AbortSignal; + /** Resolves after the thin client acknowledges that it applied the requested terminal state. */ + writeRawModeControlAsync(message: IDaemonSetRawModeMessage): Promise; +} + +/** Registration options for one request on a connection-scoped input router. @beta */ +export interface IInteractiveRequestRegistrationOptions { + readonly acceptsStdin: boolean; + readonly client: IInteractiveRequestControlClient; + readonly onFailure: (error: Error) => void; + readonly requestId: string; +} + +/** The request-owned interactive lifecycle exposed to command integrations. @beta */ +export interface IInteractiveRequestSession { + readonly requestId: string; + attachInputSink(sink: IInteractiveRequestInputSink): Disposable; + finishAsync(): Promise; + setRawModeAsync(enabled: boolean): Promise; +} + +interface IRequestState { + readonly acceptsStdin: boolean; + readonly client: IInteractiveRequestControlClient; + readonly onAbort: () => void; + readonly onFailure: (error: Error) => void; + accepting: boolean; + failure: Error | undefined; + inputSink: IInteractiveRequestInputSink | undefined; + inputTail: Promise; + pendingInputBytes: number; + pendingInputFrameCount: number; + rawModeRequested: boolean; + rawModeTail: Promise; + requestId: string; + sinkWaiters: Array<{ + readonly reject: (error: Error) => void; + readonly resolve: (sink: IInteractiveRequestInputSink) => void; + }>; +} + +const MAX_COMPLETED_REQUEST_IDS: number = 256; +const MAX_PENDING_INPUT_BYTES: number = 1024 * 1024; +const MAX_PENDING_INPUT_FRAMES: number = 256; +const requestInputFailures: WeakSet = new WeakSet(); + +/** @internal */ +export function isInteractiveRequestInputFailure(error: unknown): error is Error { + return error instanceof Error && requestInputFailures.has(error); +} + +class InteractiveRequestSession implements IInteractiveRequestSession { + readonly #state: IRequestState; + readonly #onFinished: () => void; + #finishPromise: Promise | undefined; + + public constructor(state: IRequestState, onFinished: () => void) { + this.#state = state; + this.#onFinished = onFinished; + } + + public get requestId(): string { + return this.#state.requestId; + } + + public attachInputSink(sink: IInteractiveRequestInputSink): Disposable { + const state: IRequestState = this.#state; + assertAcceptingInput(state); + if (state.inputSink) { + throw new Error(`Interactive request "${state.requestId}" already has an input sink.`); + } + state.inputSink = sink; + for (const waiter of state.sinkWaiters.splice(0)) { + waiter.resolve(sink); + } + return { + [Symbol.dispose]: (): void => { + if (state.inputSink === sink) { + state.inputSink = undefined; + } + } + }; + } + + public setRawModeAsync(enabled: boolean): Promise { + const state: IRequestState = this.#state; + if (!state.accepting) { + return Promise.reject(createRoutingError('completedRequest', state.requestId)); + } + if (!state.acceptsStdin) { + return Promise.reject(createRoutingError('nonInteractiveRequest', state.requestId)); + } + if (enabled) { + state.rawModeRequested = true; + } + return queueRawModeAsync(state, enabled); + } + + public finishAsync(): Promise { + const state: IRequestState = this.#state; + stopAcceptingInput(state); + this.#finishPromise ??= finishStateAsync(state).finally(this.#onFinished); + return this.#finishPromise; + } +} + +/** + * Routes request-tagged stdin frames with per-request ordering and backpressure. + * + * @remarks + * One instance belongs to one client connection. Completed request records remain as tombstones so late stdin is + * rejected distinctly from input for an unknown request. + * + * @beta + */ +export class InteractiveRequestInputRouter { + readonly #stateByRequestId: Map = new Map(); + readonly #completedRequestIds: Set = new Set(); + + public register(options: IInteractiveRequestRegistrationOptions): IInteractiveRequestSession { + validateRequestId(options.requestId); + if (this.#stateByRequestId.has(options.requestId) || this.#completedRequestIds.has(options.requestId)) { + throw createRoutingError('duplicateRequest', options.requestId); + } + const state: IRequestState = createRequestState(options); + this.#stateByRequestId.set(options.requestId, state); + return new InteractiveRequestSession(state, () => this.#completeRequest(options.requestId, state)); + } + + public async routeStdinFrameAsync(payload: Uint8Array): Promise { + const { chunk, requestId } = decodeDaemonStdinChunk(payload); + const state: IRequestState | undefined = this.#stateByRequestId.get(requestId); + if (!state) { + if (this.#completedRequestIds.has(requestId)) { + throw createRoutingError('completedRequest', requestId); + } + throw createRoutingError('unknownRequest', requestId); + } + await queueInputAsync(state, chunk); + } + + #completeRequest(requestId: string, state: IRequestState): void { + if (this.#stateByRequestId.get(requestId) !== state) { + return; + } + this.#stateByRequestId.delete(requestId); + this.#completedRequestIds.add(requestId); + if (this.#completedRequestIds.size > MAX_COMPLETED_REQUEST_IDS) { + const oldestRequestId: string | undefined = this.#completedRequestIds.values().next().value; + if (oldestRequestId !== undefined) { + this.#completedRequestIds.delete(oldestRequestId); + } + } + } +} + +function createRequestState(options: IInteractiveRequestRegistrationOptions): IRequestState { + const state: IRequestState = { + acceptsStdin: options.acceptsStdin, + accepting: !options.client.abortSignal.aborted, + client: options.client, + failure: undefined, + inputSink: undefined, + inputTail: Promise.resolve(), + onAbort: () => stopAcceptingInput(state), + onFailure: options.onFailure, + pendingInputBytes: 0, + pendingInputFrameCount: 0, + rawModeRequested: false, + rawModeTail: Promise.resolve(), + requestId: options.requestId, + sinkWaiters: [] + }; + options.client.abortSignal.addEventListener('abort', state.onAbort, { once: true }); + return state; +} + +function queueInputAsync(state: IRequestState, chunk: Uint8Array): Promise { + assertAcceptingInput(state); + reserveInputCapacity(state, chunk.byteLength); + const writePromise: Promise = state.inputTail.then(async () => { + assertAcceptingInput(state); + const sink: IInteractiveRequestInputSink = await getInputSinkAsync(state); + assertAcceptingInput(state); + try { + await sink.writeInputAsync(chunk); + } catch (error) { + const normalizedError: Error = createInputFailure(error); + failState(state, normalizedError); + throw normalizedError; + } + }); + const trackedPromise: Promise = writePromise.finally(() => + releaseInputCapacity(state, chunk.byteLength) + ); + state.inputTail = trackedPromise.catch((error: unknown) => handleQueuedInputError(state, error)); + return trackedPromise; +} + +function getInputSinkAsync(state: IRequestState): Promise { + if (state.inputSink) { + return Promise.resolve(state.inputSink); + } + return new Promise((resolve, reject) => { + state.sinkWaiters.push({ reject, resolve }); + }); +} + +function queueRawModeAsync(state: IRequestState, enabled: boolean): Promise { + const message: IDaemonSetRawModeMessage = { + kind: 'setRawMode', + payload: { enabled, requestId: state.requestId } + }; + const writePromise: Promise = state.rawModeTail.then(() => + state.client.writeRawModeControlAsync(message) + ); + state.rawModeTail = writePromise.catch((error: unknown) => failState(state, error)); + return writePromise; +} + +async function finishStateAsync(state: IRequestState): Promise { + state.client.abortSignal.removeEventListener('abort', state.onAbort); + await state.inputTail; + if (state.rawModeRequested) { + await queueRawModeAsync(state, false); + } + await state.rawModeTail; + if (state.failure) { + throw state.failure; + } +} + +function assertAcceptingInput(state: IRequestState): void { + if (!state.accepting) { + throw createRoutingError('completedRequest', state.requestId); + } + if (!state.acceptsStdin) { + throw createRoutingError('nonInteractiveRequest', state.requestId); + } + if (state.failure) { + throw state.failure; + } +} + +function failState(state: IRequestState, error: unknown): void { + const normalizedError: Error = normalizeError(error); + if (!state.failure) { + state.failure = normalizedError; + stopAcceptingInput(state); + state.onFailure(normalizedError); + } +} + +function reserveInputCapacity(state: IRequestState, byteLength: number): void { + if ( + state.pendingInputFrameCount >= MAX_PENDING_INPUT_FRAMES || + state.pendingInputBytes + byteLength > MAX_PENDING_INPUT_BYTES + ) { + const error: Error = createInputFailure( + new Error(`Interactive request "${state.requestId}" exceeded its pending stdin buffer limit.`) + ); + failState(state, error); + throw error; + } + state.pendingInputBytes += byteLength; + state.pendingInputFrameCount++; +} + +function releaseInputCapacity(state: IRequestState, byteLength: number): void { + state.pendingInputBytes -= byteLength; + state.pendingInputFrameCount--; +} + +function handleQueuedInputError(state: IRequestState, error: unknown): void { + if (!(error instanceof InteractiveInputRoutingError) && !isInteractiveRequestInputFailure(error)) { + failState(state, error); + } +} + +function createInputFailure(error: unknown): Error { + const normalizedError: Error = normalizeError(error); + requestInputFailures.add(normalizedError); + return normalizedError; +} + +function normalizeError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + +function stopAcceptingInput(state: IRequestState): void { + state.accepting = false; + const error: InteractiveInputRoutingError = createRoutingError('completedRequest', state.requestId); + for (const waiter of state.sinkWaiters.splice(0)) { + waiter.reject(error); + } +} + +function validateRequestId(requestId: string): void { + if (requestId.length === 0 || requestId.trim() !== requestId) { + throw new Error(`Invalid interactive request id: "${requestId}".`); + } +} + +function createRoutingError( + code: InteractiveInputRoutingErrorCode, + requestId: string +): InteractiveInputRoutingError { + return new InteractiveInputRoutingError(code, `Cannot route stdin for request "${requestId}": ${code}.`); +} diff --git a/libraries/rush-daemon/src/PhasedRequestClient.ts b/libraries/rush-daemon/src/PhasedRequestClient.ts index 046a4163c2..d8dfbd48b8 100644 --- a/libraries/rush-daemon/src/PhasedRequestClient.ts +++ b/libraries/rush-daemon/src/PhasedRequestClient.ts @@ -3,9 +3,15 @@ import type { IDaemonEventEnvelope, - IDaemonPhasedRequestResult + IDaemonPhasedRequestResult, + IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; +import type { + IInteractiveRequestInputSink, + IInteractiveRequestSession +} from './InteractiveRequestInputRouter'; + /** * A client-scoped destination for one routed phased request. * @@ -18,6 +24,10 @@ import type { export interface IPhasedRequestClient { /** Aborted by the transport when the request is cancelled or disconnected. */ readonly abortSignal: AbortSignal; + /** The request-scoped stdin/control lifecycle when one was registered by the transport integration. */ + readonly interactiveSession?: IInteractiveRequestSession; + /** The integration-owned destination for stdin accepted by this phased request. */ + readonly interactiveInputSink?: IInteractiveRequestInputSink; /** The connection session identifier used in structured event envelopes. */ readonly sessionId: string; @@ -34,6 +44,9 @@ export interface IPhasedRequestClient { chunk: Uint8Array ): Promise; + /** Signals that the client must execute this request in-process instead. */ + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; + /** Writes the final command result after every preceding event and log chunk has drained. */ writeResultAsync(result: IDaemonPhasedRequestResult): Promise; } diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts index 1f751c4a8a..64bfa9c3c4 100644 --- a/libraries/rush-daemon/src/PhasedRequestRouter.ts +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -12,12 +12,18 @@ import type { IDaemonPhasedEngineShape, IDaemonPhasedOperationSelection, IDaemonPhasedRequest, - IDaemonPhasedRequestResult + IDaemonPhasedRequestResult, + IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; import { PhasedRequestEventSink } from './PhasedRequestEventSink'; import { PhasedRequestEventMultiplexer } from './PhasedRequestEventMultiplexer'; import type { IPhasedRequestClient } from './PhasedRequestClient'; +import { + DaemonRequiresInProcessError, + evaluateDaemonTerminalPolicy +} from './DaemonTerminalPolicy'; +import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; import { RequestExclusivityClass, RequestScheduler, @@ -71,6 +77,20 @@ export class PhasedRequestRouter { request: IDaemonPhasedRequest, client: IPhasedRequestClient ): Promise { + validateRequestIdentity(request); + const interactiveSession: IInteractiveRequestSession | undefined = validateInteractiveSession( + request, + client + ); + const policy: IDaemonTerminalPolicyResult = evaluateDaemonTerminalPolicy( + request.requestId, + request.terminalRequirement + ); + if (policy.decision === 'requiresInProcess') { + await interactiveSession?.finishAsync(); + await client.writeTerminalPolicyAsync(policy); + throw new DaemonRequiresInProcessError(policy); + } const graph: IDualEmitOperationGraph = getDualEmitGraph(this.#workspaceSession); const routingState: IGraphRoutingState = getGraphRoutingState(graph); let lease: IRequestLease; @@ -84,14 +104,25 @@ export class PhasedRequestRouter { error instanceof RequestSchedulerError && error.code === RequestSchedulerErrorCode.Aborted ) { - return await writeAbortedResultAsync(request.requestId, client); + return await writeAbortedResultAsync(request.requestId, client, interactiveSession); } throw error; } + let inputAttachment: Disposable | undefined; try { - return await this.#executeAdmittedAsync(request, client, graph, routingState); + inputAttachment = attachInteractiveInput(request, client, interactiveSession); + return await this.#executeAdmittedAsync( + request, + client, + graph, + routingState, + interactiveSession + ); + } catch (error) { + return await finishAfterRoutingErrorAsync(interactiveSession, error); } finally { + inputAttachment?.[Symbol.dispose](); lease.release(); } } @@ -100,9 +131,9 @@ export class PhasedRequestRouter { request: IDaemonPhasedRequest, client: IPhasedRequestClient, graph: IDualEmitOperationGraph, - routingState: IGraphRoutingState + routingState: IGraphRoutingState, + interactiveSession: IInteractiveRequestSession | undefined ): Promise { - validateRequestIdentity(request); validateEngineShape(request.engineShape, this.#workspaceSession.engineShape); const operationById: ReadonlyMap = indexOperations(graph.operations); const selection: IResolvedSelection = resolveSelection(request.operationSelection, operationById); @@ -110,9 +141,11 @@ export class PhasedRequestRouter { try { warningsAllowedByEnvironment = parseWarningsAllowedByEnvironment(request.environment); } catch (error) { + const cleanupErrors: unknown[] = []; + await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); const result: IDaemonPhasedRequestResult = createPhasedCommandResult({ aborted: client.abortSignal.aborted, - error, + error: combineErrors(error, cleanupErrors), graphStatus: graph.status, operationOutcomes: [], requestId: request.requestId, @@ -124,14 +157,14 @@ export class PhasedRequestRouter { } if (client.abortSignal.aborted) { - return await writeAbortedResultAsync(request.requestId, client); + return await writeAbortedResultAsync(request.requestId, client, interactiveSession); } if (graph.hasScheduledIteration || graph.status === OperationStatus.Executing) { throw new Error('The warm workspace operation graph is not idle.'); } await this.#workspaceSession.reconcileInvalidationsAsync(); if (client.abortSignal.aborted) { - return await writeAbortedResultAsync(request.requestId, client); + return await writeAbortedResultAsync(request.requestId, client, interactiveSession); } applySelection(graph, selection); @@ -205,6 +238,7 @@ export class PhasedRequestRouter { } catch (error) { cleanupErrors.push(error); } + await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); await abortTail; cleanupErrors.push(...abortErrors.slice(observedAbortErrorCount)); const result: IDaemonPhasedRequestResult = createPhasedCommandResult({ @@ -265,6 +299,20 @@ function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingStat function validateRequestIdentity(request: IDaemonPhasedRequest): void { validateNonemptyName(request.requestId, 'request id'); validateNonemptyName(request.commandName, 'command name'); + if (request.acceptsStdin !== undefined && typeof request.acceptsStdin !== 'boolean') { + throw new Error('Phased request acceptsStdin must be a boolean value.'); + } + if ( + request.terminalRequirement !== undefined && + request.terminalRequirement !== 'none' && + request.terminalRequirement !== 'interactiveInput' && + request.terminalRequirement !== 'controllingTerminal' + ) { + throw new Error('Phased request terminal requirement is not recognized.'); + } + if (request.terminalRequirement === 'interactiveInput' && request.acceptsStdin !== true) { + throw new Error('Phased request interactive input requires acceptsStdin to be true.'); + } } function validateNonemptyName(value: string, kind: string): void { @@ -398,11 +446,14 @@ function compareOperations(left: Operation, right: Operation): number { async function writeAbortedResultAsync( requestId: string, - client: IPhasedRequestClient + client: IPhasedRequestClient, + interactiveSession: IInteractiveRequestSession | undefined ): Promise { + const cleanupErrors: unknown[] = []; + await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); const result: IDaemonPhasedRequestResult = createPhasedCommandResult({ aborted: true, - error: undefined, + error: combineErrors(undefined, cleanupErrors), graphStatus: OperationStatus.Aborted, operationOutcomes: [], requestId, @@ -413,6 +464,63 @@ async function writeAbortedResultAsync( return result; } +function validateInteractiveSession( + request: IDaemonPhasedRequest, + client: IPhasedRequestClient +): IInteractiveRequestSession | undefined { + const session: IInteractiveRequestSession | undefined = client.interactiveSession; + if (session && session.requestId !== request.requestId) { + throw new Error('The interactive input session does not belong to the phased request.'); + } + if (request.acceptsStdin === true && !session) { + throw new Error('The interactive phased request does not have a registered input session.'); + } + if (request.acceptsStdin === true && !client.interactiveInputSink) { + throw new Error('The interactive phased request does not have an input sink bridge.'); + } + return session; +} + +function attachInteractiveInput( + request: IDaemonPhasedRequest, + client: IPhasedRequestClient, + session: IInteractiveRequestSession | undefined +): Disposable | undefined { + if (request.acceptsStdin !== true) { + return undefined; + } + if (!session || !client.interactiveInputSink) { + throw new Error('The interactive phased request input bridge is unavailable.'); + } + return session.attachInputSink(client.interactiveInputSink); +} + +async function collectInteractiveCleanupErrorAsync( + session: IInteractiveRequestSession | undefined, + cleanupErrors: unknown[] +): Promise { + try { + await session?.finishAsync(); + } catch (error) { + cleanupErrors.push(error); + } +} + +async function finishAfterRoutingErrorAsync( + session: IInteractiveRequestSession | undefined, + routingError: unknown +): Promise { + try { + await session?.finishAsync(); + } catch (cleanupError) { + throw new AggregateError( + [routingError, cleanupError], + 'The phased request failed and could not restore its interactive terminal state.' + ); + } + throw routingError; +} + function combineErrors(executionError: unknown, cleanupErrors: unknown[]): unknown { if (executionError !== undefined && cleanupErrors.length > 0) { return new AggregateError( diff --git a/libraries/rush-daemon/src/RushDaemonHost.ts b/libraries/rush-daemon/src/RushDaemonHost.ts index 15cc941fc3..3b78da4358 100644 --- a/libraries/rush-daemon/src/RushDaemonHost.ts +++ b/libraries/rush-daemon/src/RushDaemonHost.ts @@ -15,6 +15,7 @@ import type { } from '@rushstack/rush-daemon-transport'; import { DaemonControlSession } from './DaemonControlSession'; +import type { IDaemonInteractiveConnection } from './DaemonInteractiveConnection'; import { WorkspaceSession } from './WorkspaceSession'; import type { IWorkspaceSession, WorkspaceSessionFactory } from './WorkspaceSession'; import { WorkspaceSessionProvider } from './WorkspaceSessionProvider'; @@ -31,6 +32,8 @@ export interface IRushDaemonHostOptions { readonly daemonVersion: string; /** Reports connection-level failures. */ readonly onError?: (error: Error) => void; + /** Receives the request-scoped interactive broker owned by each accepted connection. */ + readonly onInteractiveConnection?: (connection: IDaemonInteractiveConnection) => void; /** The repository root containing rush.json. */ readonly repoRoot: string; /** The selected Rush version used to isolate the workspace transport. */ @@ -96,6 +99,7 @@ export class RushDaemonHost { const session: DaemonControlSession = new DaemonControlSession(connection, { daemonVersion: options.daemonVersion, startedAtMs, + onInteractiveConnection: options.onInteractiveConnection, onClosed: (closedSession: DaemonControlSession, error: Error | undefined) => { sessions.delete(closedSession); if (error) { diff --git a/libraries/rush-daemon/src/index.ts b/libraries/rush-daemon/src/index.ts index 6d3282d61b..7a3622c159 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -3,6 +3,23 @@ /// +export { + type IDaemonInteractiveConnection, + type IDaemonInteractiveRequestOptions +} from './DaemonInteractiveConnection'; +export { + DaemonRequiresInProcessError, + evaluateDaemonTerminalPolicy +} from './DaemonTerminalPolicy'; +export { + type IInteractiveRequestControlClient, + type IInteractiveRequestInputSink, + type IInteractiveRequestRegistrationOptions, + type IInteractiveRequestSession, + InteractiveInputRoutingError, + type InteractiveInputRoutingErrorCode, + InteractiveRequestInputRouter +} from './InteractiveRequestInputRouter'; export { type IRequestLease, type IRequestSchedulerAcquireOptions, diff --git a/libraries/rush-daemon/src/test/DaemonInteractiveConnection.test.ts b/libraries/rush-daemon/src/test/DaemonInteractiveConnection.test.ts new file mode 100644 index 0000000000..d1cf11aa05 --- /dev/null +++ b/libraries/rush-daemon/src/test/DaemonInteractiveConnection.test.ts @@ -0,0 +1,147 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { DaemonControlMessage } from '@rushstack/rush-daemon-protocol'; + +import { DaemonInteractiveConnection } from '../DaemonInteractiveConnection'; +import type { IInteractiveRequestSession } from '../InteractiveRequestInputRouter'; + +it('cancels an unacknowledged raw-mode entry but still acknowledges restoration', async () => { + const sentControls: DaemonControlMessage[] = []; + let markEnterSent: (() => void) | undefined; + let markRestoreSent: (() => void) | undefined; + const enterSent: Promise = new Promise((resolve) => { + markEnterSent = resolve; + }); + const restoreSent: Promise = new Promise((resolve) => { + markRestoreSent = resolve; + }); + const holder: { connection?: DaemonInteractiveConnection } = {}; + const connection: DaemonInteractiveConnection = new DaemonInteractiveConnection( + (message: DaemonControlMessage): Promise => { + sentControls.push(message); + if (message.kind === 'setRawMode') { + if (message.payload.enabled) { + markEnterSent?.(); + } else { + markRestoreSent?.(); + } + } + return Promise.resolve(); + } + ); + holder.connection = connection; + connection.setEnabled(true); + const requestAbortController: AbortController = new AbortController(); + const session: IInteractiveRequestSession = connection.registerRequest({ + abortSignal: requestAbortController.signal, + acceptsStdin: true, + onFailure: () => undefined, + requestId: 'raw-abort' + }); + + const enterRawModePromise: Promise = session.setRawModeAsync(true); + await enterSent; + requestAbortController.abort(new Error('request cancelled')); + + await expect(enterRawModePromise).rejects.toThrow('request cancelled'); + const finishPromise: Promise = session.finishAsync(); + await restoreSent; + expect(() => + connection.handleControlMessage({ + kind: 'rawModeChanged', + payload: { enabled: true, requestId: 'raw-abort' } + }) + ).not.toThrow(); + connection.handleControlMessage({ + kind: 'rawModeChanged', + payload: { enabled: false, requestId: 'raw-abort' } + }); + await expect(finishPromise).rejects.toThrow('request cancelled'); + expect( + sentControls + .filter((message): message is Extract => + message.kind === 'setRawMode' + ) + .map(({ payload }) => payload.enabled) + ).toEqual([true, false]); +}); + +it('rejects interactive traffic until the client negotiates support', async () => { + const connection: DaemonInteractiveConnection = new DaemonInteractiveConnection(() => + Promise.resolve() + ); + expect(() => + connection.registerRequest({ + abortSignal: new AbortController().signal, + acceptsStdin: true, + onFailure: () => undefined, + requestId: 'not-negotiated' + }) + ).toThrow('did not negotiate'); + const stdinPromise: Promise = connection.routeStdinFrameAsync(Uint8Array.of(0)); + expect(stdinPromise).toBeInstanceOf(Promise); + await expect(stdinPromise).rejects.toThrow('did not negotiate'); + expect(() => connection.writeTerminalPolicyAsync({ + decision: 'runInDaemon', + requestId: 'not-negotiated' + })).toThrow('did not negotiate'); +}); + +it('serializes concurrent raw-mode requests and preserves exclusive ownership', async () => { + const sentControls: DaemonControlMessage[] = []; + const failures: Error[] = []; + const holder: { connection?: DaemonInteractiveConnection } = {}; + const connection: DaemonInteractiveConnection = new DaemonInteractiveConnection( + (message: DaemonControlMessage): Promise => { + sentControls.push(message); + if (message.kind === 'setRawMode') { + queueMicrotask(() => + holder.connection?.handleControlMessage({ + kind: 'rawModeChanged', + payload: message.payload + }) + ); + } + return Promise.resolve(); + } + ); + holder.connection = connection; + connection.setEnabled(true); + const first: IInteractiveRequestSession = connection.registerRequest({ + abortSignal: new AbortController().signal, + acceptsStdin: true, + onFailure: () => undefined, + requestId: 'first-owner' + }); + const second: IInteractiveRequestSession = connection.registerRequest({ + abortSignal: new AbortController().signal, + acceptsStdin: true, + onFailure: (error: Error) => failures.push(error), + requestId: 'second-owner' + }); + + const firstRawMode: Promise = first.setRawModeAsync(true); + const secondRawMode: Promise = second.setRawModeAsync(true); + await firstRawMode; + await expect(secondRawMode).rejects.toThrow('already owned'); + await expect(second.finishAsync()).rejects.toThrow('already owned'); + expect(failures).toHaveLength(1); + expect(rawModePayloads(sentControls)).toEqual([{ enabled: true, requestId: 'first-owner' }]); + + await first.finishAsync(); + expect(rawModePayloads(sentControls)).toEqual([ + { enabled: true, requestId: 'first-owner' }, + { enabled: false, requestId: 'first-owner' } + ]); +}); + +function rawModePayloads( + messages: DaemonControlMessage[] +): Array<{ enabled: boolean; requestId: string }> { + return messages + .filter((message): message is Extract => + message.kind === 'setRawMode' + ) + .map(({ payload }) => payload); +} diff --git a/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts index 91bb5a1153..b7d3bf1d58 100644 --- a/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts +++ b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts @@ -5,8 +5,14 @@ import * as fs from 'node:fs'; import * as path from 'node:path'; import { SubprocessTerminator } from '@rushstack/node-core-library'; -import type { IDaemonCommandResult } from '@rushstack/rush-daemon-protocol'; +import { encodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; +import type { + IDaemonCommandResult, + IDaemonSetRawModeMessage, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; +import { DaemonRequiresInProcessError } from '../DaemonTerminalPolicy'; import type { IGlobalCommandExecutionContext } from '../GlobalCommandExecutionContext'; import type { IResolvedGlobalCommandRequest, @@ -18,6 +24,10 @@ import { type IGlobalCommandExecutionResult, type IGlobalCommandRequestResult } from '../GlobalCommandRequestRouter'; +import { + InteractiveRequestInputRouter, + type IInteractiveRequestSession +} from '../InteractiveRequestInputRouter'; import { TestWorkspaceSession, TEST_REPO_ROOT } from './TestWorkspaceSession'; const TEXT_DECODER: InstanceType = new TextDecoder(); @@ -33,7 +43,9 @@ class TestGlobalCommandClient implements IGlobalCommandRequestClient { public readonly abortController: AbortController = new AbortController(); public readonly chunks: IClientChunk[] = []; public readonly results: IDaemonCommandResult[] = []; + public readonly policies: IDaemonTerminalPolicyResult[] = []; public readonly writeOrder: Array<'chunk' | 'result'> = []; + public interactiveSession: IInteractiveRequestSession | undefined; public onWriteAsync: ((chunk: IClientChunk) => Promise) | undefined; public onResultAsync: ((result: IDaemonCommandResult) => Promise) | undefined; @@ -56,6 +68,11 @@ class TestGlobalCommandClient implements IGlobalCommandRequestClient { this.results.push(result); this.writeOrder.push('result'); } + + public writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise { + this.policies.push(result); + return Promise.resolve(); + } } function createRequestOptions( @@ -756,4 +773,164 @@ describe(GlobalCommandRequestRouter.name, () => { ).rejects.toThrow('not resolved for this workspace session'); expect(executor).not.toHaveBeenCalled(); }); + + it('forwards raw stdin bytes to an injected child process with backpressure', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + const inputRouter: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const requestId: string = 'child-input'; + client.interactiveSession = inputRouter.register({ + acceptsStdin: true, + client: { + abortSignal: client.abortSignal, + writeRawModeControlAsync: (): Promise => Promise.resolve() + }, + onFailure: (error: Error) => client.abortController.abort(error), + requestId + }); + const options: IResolveGlobalCommandRequestOptions = createRequestOptions( + requestId, + FIRST_CWD, + {}, + 80 + ); + const request: IResolvedGlobalCommandRequest = router.resolveRequest({ + ...options, + terminal: { ...options.terminal, acceptsStdin: true } + }); + let markChildStarted: (() => void) | undefined; + const childStarted: Promise = new Promise((resolve) => { + markChildStarted = resolve; + }); + const resultPromise: Promise = router.executeAsync( + request, + async (context: IGlobalCommandExecutionContext): Promise => { + const child = context.spawnChild( + process.execPath, + ['-e', "process.stdin.once('data',b=>{process.stdout.write(Buffer.from(b).toString('hex'));process.exit(0)})"], + { forwardInput: true } + ); + child.once('spawn', () => markChildStarted?.()); + await new Promise((resolve) => child.once('close', () => resolve())); + return { exitCode: 0 }; + }, + client + ); + await childStarted; + const inputBytes: Uint8Array = Uint8Array.of(0xff, 0x00, 0x80); + await inputRouter.routeStdinFrameAsync(encodeDaemonStdinChunk({ chunk: inputBytes, requestId })); + await resultPromise; + + expect(client.chunks.map(({ text }) => text).join('')).toBe('ff0080'); + }); + + it('restores raw mode after output drain and before the final result', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + const lifecycleOrder: string[] = []; + const inputRouter: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const requestId: string = 'raw-lifecycle'; + client.interactiveSession = inputRouter.register({ + acceptsStdin: true, + client: { + abortSignal: client.abortSignal, + writeRawModeControlAsync: (message: IDaemonSetRawModeMessage): Promise => { + lifecycleOrder.push(`raw:${message.payload.enabled}`); + return Promise.resolve(); + } + }, + onFailure: (error: Error) => client.abortController.abort(error), + requestId + }); + client.onWriteAsync = async (): Promise => { + await new Promise((resolve) => setImmediate(resolve)); + lifecycleOrder.push('output'); + }; + client.onResultAsync = (): Promise => { + lifecycleOrder.push('result'); + return Promise.resolve(); + }; + const options: IResolveGlobalCommandRequestOptions = createRequestOptions( + requestId, + FIRST_CWD, + {}, + 80 + ); + + await router.executeAsync( + router.resolveRequest({ + ...options, + terminal: { ...options.terminal, acceptsStdin: true } + }), + async (context: IGlobalCommandExecutionContext): Promise => { + await context.interactiveInput?.setRawModeAsync(true); + context.terminal.writeLine('prompt'); + return { exitCode: 0 }; + }, + client + ); + + expect(lifecycleOrder).toEqual(['raw:true', 'output', 'raw:false', 'result']); + }); + + it('rejects interactive input requirements without stdin capability', () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const options: IResolveGlobalCommandRequestOptions = createRequestOptions( + 'invalid-interactive-input', + FIRST_CWD, + {}, + 80 + ); + + expect(() => + router.resolveRequest({ + ...options, + terminal: { ...options.terminal, terminalRequirement: 'interactiveInput' } + }) + ).toThrow('requires acceptsStdin'); + }); + + it('signals typed in-process fallback without executing a PTY-only command', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + const executor: jest.Mock, [IGlobalCommandExecutionContext]> = + jest.fn(async (context: IGlobalCommandExecutionContext) => { + void context; + return { exitCode: 0 }; + }); + const options: IResolveGlobalCommandRequestOptions = createRequestOptions( + 'pty-only', + FIRST_CWD, + {}, + 80 + ); + + await expect( + router.executeAsync( + router.resolveRequest({ + ...options, + terminal: { ...options.terminal, terminalRequirement: 'controllingTerminal' } + }), + executor, + client + ) + ).rejects.toBeInstanceOf(DaemonRequiresInProcessError); + expect(client.policies).toEqual([ + { + decision: 'requiresInProcess', + reason: 'controllingTerminalRequired', + requestId: 'pty-only' + } + ]); + expect(executor).not.toHaveBeenCalled(); + expect(client.results).toHaveLength(0); + }); }); diff --git a/libraries/rush-daemon/src/test/InteractiveRequestInputLimits.test.ts b/libraries/rush-daemon/src/test/InteractiveRequestInputLimits.test.ts new file mode 100644 index 0000000000..76fba0ff31 --- /dev/null +++ b/libraries/rush-daemon/src/test/InteractiveRequestInputLimits.test.ts @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { encodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; + +import { InteractiveRequestInputRouter } from '../InteractiveRequestInputRouter'; +import type { IInteractiveRequestSession } from '../InteractiveRequestInputRouter'; + +const PENDING_FRAME_LIMIT: number = 256; +const REQUEST_ID: string = 'bounded-input'; + +it('bounds pending stdin without turning overflow into a connection failure', async () => { + const abortController: AbortController = new AbortController(); + const failures: Error[] = []; + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const session: IInteractiveRequestSession = router.register({ + acceptsStdin: true, + client: { + abortSignal: abortController.signal, + writeRawModeControlAsync: (): Promise => Promise.resolve() + }, + onFailure: (error: Error) => failures.push(error), + requestId: REQUEST_ID + }); + const pendingRoutes: Promise[] = Array.from({ length: PENDING_FRAME_LIMIT }, () => + routeAsync(router).catch(() => undefined) + ); + + await expect(routeAsync(router)).rejects.toThrow('pending stdin buffer limit'); + await Promise.all(pendingRoutes); + await expect(session.finishAsync()).rejects.toThrow('pending stdin buffer limit'); + expect(failures).toHaveLength(1); +}); + +function routeAsync(router: InteractiveRequestInputRouter): Promise { + return router.routeStdinFrameAsync( + encodeDaemonStdinChunk({ chunk: new Uint8Array(), requestId: REQUEST_ID }) + ); +} diff --git a/libraries/rush-daemon/src/test/InteractiveRequestInputRouter.test.ts b/libraries/rush-daemon/src/test/InteractiveRequestInputRouter.test.ts new file mode 100644 index 0000000000..16838becc4 --- /dev/null +++ b/libraries/rush-daemon/src/test/InteractiveRequestInputRouter.test.ts @@ -0,0 +1,241 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { encodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonSetRawModeMessage } from '@rushstack/rush-daemon-protocol'; + +import { InteractiveRequestInputRouter } from '../InteractiveRequestInputRouter'; +import type { + IInteractiveRequestControlClient, + IInteractiveRequestInputSink, + IInteractiveRequestSession +} from '../InteractiveRequestInputRouter'; + +class TestControlClient implements IInteractiveRequestControlClient { + public readonly abortController: AbortController = new AbortController(); + public readonly controls: IDaemonSetRawModeMessage[] = []; + public onControlAsync: ((message: IDaemonSetRawModeMessage) => Promise) | undefined; + + public get abortSignal(): AbortSignal { + return this.abortController.signal; + } + + public async writeRawModeControlAsync(message: IDaemonSetRawModeMessage): Promise { + this.controls.push(message); + await this.onControlAsync?.(message); + } +} + +function register( + router: InteractiveRequestInputRouter, + requestId: string, + client: TestControlClient, + acceptsStdin: boolean = true +): { failures: Error[]; session: IInteractiveRequestSession } { + const failures: Error[] = []; + const session: IInteractiveRequestSession = router.register({ + acceptsStdin, + client, + onFailure: (error: Error) => failures.push(error), + requestId + }); + return { failures, session }; +} + +function routeAsync( + router: InteractiveRequestInputRouter, + requestId: string, + chunk: Uint8Array +): Promise { + return router.routeStdinFrameAsync(encodeDaemonStdinChunk({ chunk, requestId })); +} + +describe(InteractiveRequestInputRouter.name, () => { + it('returns promise rejections for malformed and request-ineligible input', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const malformedPromise: Promise = router.routeStdinFrameAsync(Uint8Array.of(0)); + expect(malformedPromise).toBeInstanceOf(Promise); + await expect(malformedPromise).rejects.toMatchObject({ code: 'malformedPayload' }); + + const session: IInteractiveRequestSession = register( + router, + 'non-interactive', + new TestControlClient(), + false + ).session; + const ineligiblePromise: Promise = routeAsync( + router, + 'non-interactive', + Uint8Array.of(1) + ); + expect(ineligiblePromise).toBeInstanceOf(Promise); + await expect(ineligiblePromise).rejects.toMatchObject({ code: 'nonInteractiveRequest' }); + await session.finishAsync(); + }); + + it('backpressures stdin that arrives before its input sink is attached', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const session: IInteractiveRequestSession = register( + router, + 'sink-race', + new TestControlClient() + ).session; + const writes: Uint8Array[] = []; + let settled: boolean = false; + const writePromise: Promise = routeAsync(router, 'sink-race', Uint8Array.of(1)); + void writePromise.finally(() => { + settled = true; + }); + await Promise.resolve(); + expect(settled).toBe(false); + + session.attachInputSink({ + writeInputAsync: (chunk: Uint8Array): Promise => { + writes.push(chunk); + return Promise.resolve(); + } + }); + await writePromise; + expect(writes).toEqual([Uint8Array.of(1)]); + }); + + it('preserves bytes and ordered backpressure without cross-request blocking', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const first = register(router, 'first', new TestControlClient()).session; + const second = register(router, 'second', new TestControlClient()).session; + const writes: string[] = []; + let releaseFirst: (() => void) | undefined; + first.attachInputSink({ + writeInputAsync: async (chunk: Uint8Array): Promise => { + writes.push(`first:${Buffer.from(chunk).toString('hex')}`); + if (writes.length === 1) { + await new Promise((resolve) => { + releaseFirst = resolve; + }); + } + } + }); + second.attachInputSink({ + writeInputAsync: (chunk: Uint8Array): Promise => { + writes.push(`second:${Buffer.from(chunk).toString('hex')}`); + return Promise.resolve(); + } + }); + + const firstWrite: Promise = routeAsync(router, 'first', Uint8Array.of(0xff, 0x80)); + const queuedFirstWrite: Promise = routeAsync(router, 'first', Uint8Array.of(0x00)); + await routeAsync(router, 'second', Uint8Array.of(0x7f)); + expect(writes).toEqual(['first:ff80', 'second:7f']); + releaseFirst?.(); + await Promise.all([firstWrite, queuedFirstWrite]); + expect(writes).toEqual(['first:ff80', 'second:7f', 'first:00']); + }); + + it('rejects unknown, non-interactive, aborted, and completed request input', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + await expect(routeAsync(router, 'missing', Uint8Array.of(1))).rejects.toMatchObject({ + code: 'unknownRequest' + }); + const nonInteractive = register(router, 'plain', new TestControlClient(), false).session; + await expect(routeAsync(router, 'plain', Uint8Array.of(1))).rejects.toMatchObject({ + code: 'nonInteractiveRequest' + }); + await nonInteractive.finishAsync(); + const client: TestControlClient = new TestControlClient(); + const interactive: IInteractiveRequestSession = register(router, 'active', client).session; + interactive.attachInputSink({ writeInputAsync: (): Promise => Promise.resolve() }); + client.abortController.abort(); + await expect(routeAsync(router, 'active', Uint8Array.of(1))).rejects.toMatchObject({ + code: 'completedRequest' + }); + await interactive.finishAsync(); + }); + + it('serializes raw-mode transitions and restores cooked mode before finishing', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const client: TestControlClient = new TestControlClient(); + const session: IInteractiveRequestSession = register(router, 'raw', client).session; + + await session.setRawModeAsync(true); + await session.finishAsync(); + + expect(client.controls.map(({ payload }) => payload)).toEqual([ + { enabled: true, requestId: 'raw' }, + { enabled: false, requestId: 'raw' } + ]); + }); + + it('restores cooked mode after cancellation races request cleanup', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const client: TestControlClient = new TestControlClient(); + const session: IInteractiveRequestSession = register(router, 'abort-raw', client).session; + await session.setRawModeAsync(true); + + client.abortController.abort(new Error('connection closed')); + await session.finishAsync(); + + expect(client.controls.map(({ payload }) => payload.enabled)).toEqual([true, false]); + }); + + it('does not deliver queued stdin after cancellation', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const client: TestControlClient = new TestControlClient(); + const session: IInteractiveRequestSession = register(router, 'queued-abort', client).session; + const writes: number[] = []; + let releaseFirst: (() => void) | undefined; + let markFirstStarted: (() => void) | undefined; + const firstStarted: Promise = new Promise((resolve) => { + markFirstStarted = resolve; + }); + session.attachInputSink({ + writeInputAsync: async (chunk: Uint8Array): Promise => { + writes.push(chunk[0]); + markFirstStarted?.(); + await new Promise((resolve) => { + releaseFirst = resolve; + }); + } + }); + const firstWrite: Promise = routeAsync(router, 'queued-abort', Uint8Array.of(1)); + const secondWrite: Promise = routeAsync(router, 'queued-abort', Uint8Array.of(2)); + const secondWriteRejection: Promise = expect(secondWrite).rejects.toMatchObject({ + code: 'completedRequest' + }); + await firstStarted; + client.abortController.abort(); + releaseFirst?.(); + + await firstWrite; + await secondWriteRejection; + await session.finishAsync(); + expect(writes).toEqual([1]); + }); + + it('attempts raw-mode restoration after a control write failure', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const client: TestControlClient = new TestControlClient(); + client.onControlAsync = ({ payload }): Promise => + payload.enabled ? Promise.reject(new Error('raw mode write failed')) : Promise.resolve(); + const { failures, session } = register(router, 'raw-failure', client); + + await expect(session.setRawModeAsync(true)).rejects.toThrow('raw mode write failed'); + await expect(session.finishAsync()).rejects.toThrow('raw mode write failed'); + + expect(client.controls.map(({ payload }) => payload.enabled)).toEqual([true, false]); + expect(failures).toHaveLength(1); + }); + + it('stops a request after an input write failure and reports it once', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const { failures, session } = register(router, 'write-failure', new TestControlClient()); + const sink: IInteractiveRequestInputSink = { + writeInputAsync: (): Promise => Promise.reject(new Error('stdin closed')) + }; + session.attachInputSink(sink); + + await expect(routeAsync(router, 'write-failure', Uint8Array.of(1))).rejects.toThrow('stdin closed'); + await expect(routeAsync(router, 'write-failure', Uint8Array.of(2))).rejects.toThrow('completedRequest'); + await expect(session.finishAsync()).rejects.toThrow('stdin closed'); + expect(failures).toHaveLength(1); + }); +}); diff --git a/libraries/rush-daemon/src/test/PhasedRequestInput.test.ts b/libraries/rush-daemon/src/test/PhasedRequestInput.test.ts new file mode 100644 index 0000000000..79fdba6c90 --- /dev/null +++ b/libraries/rush-daemon/src/test/PhasedRequestInput.test.ts @@ -0,0 +1,81 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { OperationStatus } from '@microsoft/rush-lib'; +import { encodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonPhasedRequest } from '@rushstack/rush-daemon-protocol'; + +import { InteractiveRequestInputRouter } from '../InteractiveRequestInputRouter'; +import { PhasedRequestRouter } from '../PhasedRequestRouter'; +import { + TEST_ENGINE_SHAPE, + TestOperationRunner, + TestPhasedRequestClient, + createRoutingFixture +} from './PhasedRequestRouterTestUtilities'; +import type { ITestRoutingFixture } from './PhasedRequestRouterTestUtilities'; + +const OPERATION_ID: string = 'project-a (_phase:test)'; +const REQUEST_ID: string = 'phased-input'; +const INPUT_BYTE: number = 0x7f; + +function createRequest(overrides: Partial = {}): IDaemonPhasedRequest { + return { + commandName: 'build', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: [{ enabledState: true, operationId: OPERATION_ID }], + requestId: REQUEST_ID, + ...overrides + }; +} + +it('bridges request-scoped stdin into phased execution', async () => { + const inputRouter: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + const received: Uint8Array[] = []; + const runner: TestOperationRunner = new TestOperationRunner( + OPERATION_ID, + OperationStatus.Success, + async (): Promise => { + await inputRouter.routeStdinFrameAsync( + encodeDaemonStdinChunk({ chunk: Uint8Array.of(INPUT_BYTE), requestId: REQUEST_ID }) + ); + } + ); + const fixture: ITestRoutingFixture = createRoutingFixture(new Map([[OPERATION_ID, runner]])); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + client.interactiveSession = inputRouter.register({ + acceptsStdin: true, + client: { + abortSignal: client.abortSignal, + writeRawModeControlAsync: (): Promise => Promise.resolve() + }, + onFailure: (error: Error) => client.abortController.abort(error), + requestId: REQUEST_ID + }); + client.interactiveInputSink = { + writeInputAsync: (chunk: Uint8Array): Promise => { + received.push(chunk); + return Promise.resolve(); + } + }; + + await new PhasedRequestRouter(fixture.session).executeAsync( + createRequest({ acceptsStdin: true, terminalRequirement: 'interactiveInput' }), + client + ); + + expect(received).toEqual([Uint8Array.of(INPUT_BYTE)]); +}); + +it('rejects interactive input requirements without stdin capability', async () => { + const fixture: ITestRoutingFixture = createRoutingFixture( + new Map([[OPERATION_ID, new TestOperationRunner(OPERATION_ID)]]) + ); + await expect( + new PhasedRequestRouter(fixture.session).executeAsync( + createRequest({ terminalRequirement: 'interactiveInput' }), + new TestPhasedRequestClient() + ) + ).rejects.toThrow('requires acceptsStdin'); +}); diff --git a/libraries/rush-daemon/src/test/PhasedRequestInteractive.test.ts b/libraries/rush-daemon/src/test/PhasedRequestInteractive.test.ts new file mode 100644 index 0000000000..1654fc23b7 --- /dev/null +++ b/libraries/rush-daemon/src/test/PhasedRequestInteractive.test.ts @@ -0,0 +1,91 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IDaemonPhasedRequest, + IDaemonSetRawModeMessage +} from '@rushstack/rush-daemon-protocol'; + +import { DaemonRequiresInProcessError } from '../DaemonTerminalPolicy'; +import { InteractiveRequestInputRouter } from '../InteractiveRequestInputRouter'; +import { PhasedRequestRouter } from '../PhasedRequestRouter'; +import { + TEST_ENGINE_SHAPE, + TestOperationRunner, + TestPhasedRequestClient, + createRoutingFixture +} from './PhasedRequestRouterTestUtilities'; +import type { ITestRoutingFixture } from './PhasedRequestRouterTestUtilities'; + +const OPERATION_ID: string = 'project-a (_phase:test)'; + +function createRequest(overrides: Partial = {}): IDaemonPhasedRequest { + return { + commandName: 'build', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: [{ enabledState: true, operationId: OPERATION_ID }], + requestId: 'interactive-request', + ...overrides + }; +} + +function createFixture(): ITestRoutingFixture { + return createRoutingFixture(new Map([[OPERATION_ID, new TestOperationRunner(OPERATION_ID)]])); +} + +it('restores phased-request raw mode before publishing the command result', async () => { + const fixture: ITestRoutingFixture = createFixture(); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + const lifecycleOrder: string[] = []; + client.interactiveSession = new InteractiveRequestInputRouter().register({ + acceptsStdin: true, + client: { + abortSignal: client.abortSignal, + writeRawModeControlAsync: (message: IDaemonSetRawModeMessage): Promise => { + lifecycleOrder.push(`raw:${message.payload.enabled}`); + return Promise.resolve(); + } + }, + onFailure: (error: Error) => client.abortController.abort(error), + requestId: 'interactive-request' + }); + client.interactiveInputSink = { + writeInputAsync: (): Promise => Promise.resolve() + }; + client.onWriteAsync = (write): Promise => { + if (write.result) lifecycleOrder.push('result'); + return Promise.resolve(); + }; + await client.interactiveSession.setRawModeAsync(true); + + await new PhasedRequestRouter(fixture.session).executeAsync( + createRequest({ acceptsStdin: true, terminalRequirement: 'interactiveInput' }), + client + ); + + expect(lifecycleOrder).toEqual(['raw:true', 'raw:false', 'result']); +}); + +it('signals requiresInProcess without scheduling a PTY-only phased request', async () => { + const fixture: ITestRoutingFixture = createFixture(); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + + await expect( + new PhasedRequestRouter(fixture.session).executeAsync( + createRequest({ terminalRequirement: 'controllingTerminal' }), + client + ) + ).rejects.toBeInstanceOf(DaemonRequiresInProcessError); + + expect(client.policies).toEqual([ + { + decision: 'requiresInProcess', + reason: 'controllingTerminalRequired', + requestId: 'interactive-request' + } + ]); + expect(scheduleSpy).not.toHaveBeenCalled(); + expect(client.writes).toHaveLength(0); +}); diff --git a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts index 251de684a5..cc5efef46e 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts @@ -17,9 +17,12 @@ import { OperationGraph } from '@microsoft/rush-lib/lib/logic/operations/Operati import type { IOperationGraphOptions } from '@microsoft/rush-lib/lib/logic/operations/OperationGraph'; import type { IDaemonEventEnvelope, - IDaemonPhasedRequestResult + IDaemonPhasedRequestResult, + IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; +import type { IInteractiveRequestSession } from '../InteractiveRequestInputRouter'; +import type { IInteractiveRequestInputSink } from '../InteractiveRequestInputRouter'; import type { IPhasedRequestClient } from '../PhasedRequestClient'; import type { IWorkspaceEngineShape, @@ -59,6 +62,9 @@ export class TestPhasedRequestClient implements IPhasedRequestClient { public readonly abortController: AbortController = new AbortController(); public readonly sessionId: string = 'test-session'; public readonly writes: ITestClientWrite[] = []; + public readonly policies: IDaemonTerminalPolicyResult[] = []; + public interactiveInputSink: IInteractiveRequestInputSink | undefined; + public interactiveSession: IInteractiveRequestSession | undefined; public onWriteAsync: ((write: ITestClientWrite) => Promise) | undefined; readonly #sequenceState: { next: number }; @@ -101,6 +107,11 @@ export class TestPhasedRequestClient implements IPhasedRequestClient { await this.onWriteAsync?.(write); this.writes.push(write); } + + public writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise { + this.policies.push(result); + return Promise.resolve(); + } } export class TestOperationRunner implements IOperationRunner { diff --git a/libraries/rush-daemon/src/test/RushDaemonHost.test.ts b/libraries/rush-daemon/src/test/RushDaemonHost.test.ts index 020d7e5ed3..c7f4db778e 100644 --- a/libraries/rush-daemon/src/test/RushDaemonHost.test.ts +++ b/libraries/rush-daemon/src/test/RushDaemonHost.test.ts @@ -10,7 +10,8 @@ import { DaemonFrameType, createDaemonHello, decodeDaemonControlMessage, - encodeDaemonControlMessage + encodeDaemonControlMessage, + encodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; import type { DaemonControlMessage, @@ -29,6 +30,8 @@ import type { import { RushDaemonHost } from '../RushDaemonHost'; import type { IRushDaemonHostOptions } from '../RushDaemonHost'; +import type { IDaemonInteractiveConnection } from '../DaemonInteractiveConnection'; +import type { IInteractiveRequestSession } from '../InteractiveRequestInputRouter'; import { serveRushDaemonAsync } from '../serveRushDaemon'; import type { IWorkspaceSession } from '../WorkspaceSession'; import { TestWorkspaceSession } from './TestWorkspaceSession'; @@ -37,6 +40,7 @@ const RUSH_VERSION: string = '5.178.1'; const DAEMON_VERSION: string = '0.1.0-test'; const WINDOWS_PIPE_PREFIX: string = '\\\\.\\pipe\\rushd-'; const REPO_PREFIX: string = 'rush-daemon-host-test-'; +const INPUT_BYTE: number = 0xff; const testRepoRoots: Set = new Set(); @@ -122,6 +126,270 @@ describe(RushDaemonHost.name, () => { } }); + it('routes stdin frames through a connection-scoped async input router', async () => { + let markRouted: (() => void) | undefined; + let interactiveConnection: IDaemonInteractiveConnection | undefined; + const routed: Promise = new Promise((resolve) => { + markRouted = resolve; + }); + const received: Array<{ chunk: Uint8Array; requestId: string }> = []; + const inputHost: RushDaemonHost = await RushDaemonHost.startAsync( + createHostOptions(createTestRepoRoot(), { + onInteractiveConnection: (connection: IDaemonInteractiveConnection) => { + interactiveConnection = connection; + } + }) + ); + const inputClient: DaemonFrameConnection = await connectDaemonAsync(inputHost.paths.socketPath); + try { + await exchangeControlAsync(inputClient, createDaemonHello(DAEMON_PROTOCOL_VERSION)); + await inputClient.sendFrameAsync({ + kind: DaemonFrameType.controlJson, + payload: encodeDaemonControlMessage({ + kind: 'subscribe', + payload: { isTTY: true, supportsInteractiveIO: true } + }) + }); + await exchangeControlAsync(inputClient, { kind: 'ping', payload: {} }); + if (!interactiveConnection) { + throw new Error('The host did not expose its interactive connection.'); + } + const requestAbortController: AbortController = new AbortController(); + const requestSession: IInteractiveRequestSession = interactiveConnection.registerRequest({ + abortSignal: requestAbortController.signal, + acceptsStdin: true, + onFailure: (error: Error) => requestAbortController.abort(error), + requestId: 'request-input' + }); + requestSession.attachInputSink({ + writeInputAsync: (chunk: Uint8Array): Promise => { + received.push({ chunk, requestId: 'request-input' }); + markRouted?.(); + return Promise.resolve(); + } + }); + await inputClient.sendFrameAsync({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ + chunk: Uint8Array.of(INPUT_BYTE), + requestId: 'request-input' + }) + }); + await routed; + expect(received).toEqual([{ chunk: Uint8Array.of(INPUT_BYTE), requestId: 'request-input' }]); + const rawModes: boolean[] = []; + inputClient.onFrame(async (frame: IDaemonFrame): Promise => { + const message: DaemonControlMessage = decodeDaemonControlMessage(frame.payload); + if (message.kind === 'setRawMode') { + rawModes.push(message.payload.enabled); + await inputClient.sendFrameAsync({ + kind: DaemonFrameType.controlJson, + payload: encodeDaemonControlMessage({ + kind: 'rawModeChanged', + payload: message.payload + }) + }); + } + }); + await requestSession.setRawModeAsync(true); + await requestSession.finishAsync(); + expect(rawModes).toEqual([true, false]); + } finally { + await inputClient.closeAsync(); + await inputHost.closeAsync(); + } + }); + + it('handles control frames while stdin waits for its request sink', async () => { + let interactiveConnection: IDaemonInteractiveConnection | undefined; + const inputHost: RushDaemonHost = await RushDaemonHost.startAsync( + createHostOptions(createTestRepoRoot(), { + onInteractiveConnection: (connection: IDaemonInteractiveConnection) => { + interactiveConnection = connection; + } + }) + ); + const inputClient: DaemonFrameConnection = await connectDaemonAsync(inputHost.paths.socketPath); + try { + await exchangeControlAsync(inputClient, createDaemonHello(DAEMON_PROTOCOL_VERSION)); + await inputClient.sendFrameAsync({ + kind: DaemonFrameType.controlJson, + payload: encodeDaemonControlMessage({ + kind: 'subscribe', + payload: { isTTY: true, supportsInteractiveIO: true } + }) + }); + await exchangeControlAsync(inputClient, { kind: 'ping', payload: {} }); + if (!interactiveConnection) { + throw new Error('The host did not expose its interactive connection.'); + } + const requestSession: IInteractiveRequestSession = interactiveConnection.registerRequest({ + abortSignal: new AbortController().signal, + acceptsStdin: true, + onFailure: () => undefined, + requestId: 'waiting-input' + }); + await inputClient.sendFrameAsync({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ + chunk: Uint8Array.of(INPUT_BYTE), + requestId: 'waiting-input' + }) + }); + + await expect( + exchangeControlAsync(inputClient, { kind: 'ping', payload: {} }) + ).resolves.toMatchObject({ kind: 'pong' }); + let markInputDelivered: (() => void) | undefined; + const inputDelivered: Promise = new Promise((resolve) => { + markInputDelivered = resolve; + }); + requestSession.attachInputSink({ + writeInputAsync: (): Promise => { + markInputDelivered?.(); + return Promise.resolve(); + } + }); + await inputDelivered; + await requestSession.finishAsync(); + } finally { + await inputClient.closeAsync(); + await inputHost.closeAsync(); + } + }); + + it('keeps the connection alive when pending stdin is cancelled', async () => { + const errors: Error[] = []; + let interactiveConnection: IDaemonInteractiveConnection | undefined; + const inputHost: RushDaemonHost = await RushDaemonHost.startAsync( + createHostOptions(createTestRepoRoot(), { + onError: (error: Error) => errors.push(error), + onInteractiveConnection: (connection: IDaemonInteractiveConnection) => { + interactiveConnection = connection; + } + }) + ); + const inputClient: DaemonFrameConnection = await connectDaemonAsync(inputHost.paths.socketPath); + try { + await exchangeControlAsync(inputClient, createDaemonHello(DAEMON_PROTOCOL_VERSION)); + await inputClient.sendFrameAsync({ + kind: DaemonFrameType.controlJson, + payload: encodeDaemonControlMessage({ + kind: 'subscribe', + payload: { isTTY: true, supportsInteractiveIO: true } + }) + }); + await exchangeControlAsync(inputClient, { kind: 'ping', payload: {} }); + if (!interactiveConnection) { + throw new Error('The host did not expose its interactive connection.'); + } + const requestAbortController: AbortController = new AbortController(); + interactiveConnection.registerRequest({ + abortSignal: requestAbortController.signal, + acceptsStdin: true, + onFailure: () => undefined, + requestId: 'cancelled-input' + }); + await inputClient.sendFrameAsync({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ + chunk: Uint8Array.of(INPUT_BYTE), + requestId: 'cancelled-input' + }) + }); + await exchangeControlAsync(inputClient, { kind: 'ping', payload: {} }); + requestAbortController.abort(); + + await expect( + exchangeControlAsync(inputClient, { kind: 'ping', payload: {} }) + ).resolves.toMatchObject({ kind: 'pong' }); + expect(errors).toEqual([]); + } finally { + await inputClient.closeAsync(); + await inputHost.closeAsync(); + } + }); + + it('keeps the connection alive after a request input sink fails', async () => { + let failureConnection: IDaemonInteractiveConnection | undefined; + const failureHost: RushDaemonHost = await RushDaemonHost.startAsync( + createHostOptions(createTestRepoRoot(), { + onInteractiveConnection: (connection: IDaemonInteractiveConnection) => { + failureConnection = connection; + } + }) + ); + const failureClient: DaemonFrameConnection = await connectDaemonAsync(failureHost.paths.socketPath); + try { + await exchangeControlAsync(failureClient, createDaemonHello(DAEMON_PROTOCOL_VERSION)); + await failureClient.sendFrameAsync({ + kind: DaemonFrameType.controlJson, + payload: encodeDaemonControlMessage({ + kind: 'subscribe', + payload: { isTTY: true, supportsInteractiveIO: true } + }) + }); + await exchangeControlAsync(failureClient, { kind: 'ping', payload: {} }); + if (!failureConnection) { + throw new Error('The host did not expose its interactive connection.'); + } + let reportFailure: ((error: Error) => void) | undefined; + const failureReported: Promise = new Promise((resolve) => { + reportFailure = resolve; + }); + const failedRequest: IInteractiveRequestSession = failureConnection.registerRequest({ + abortSignal: new AbortController().signal, + acceptsStdin: true, + onFailure: (error: Error) => reportFailure?.(error), + requestId: 'failed-input' + }); + failedRequest.attachInputSink({ + writeInputAsync: (): Promise => Promise.reject(new Error('request stdin failed')) + }); + let markSurvivorInput: (() => void) | undefined; + const survivorInput: Promise = new Promise((resolve) => { + markSurvivorInput = resolve; + }); + const survivingRequest: IInteractiveRequestSession = failureConnection.registerRequest({ + abortSignal: new AbortController().signal, + acceptsStdin: true, + onFailure: () => undefined, + requestId: 'surviving-input' + }); + survivingRequest.attachInputSink({ + writeInputAsync: (): Promise => { + markSurvivorInput?.(); + return Promise.resolve(); + } + }); + + await failureClient.sendFrameAsync({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ + chunk: Uint8Array.of(INPUT_BYTE), + requestId: 'failed-input' + }) + }); + expect((await failureReported).message).toBe('request stdin failed'); + await failureClient.sendFrameAsync({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ + chunk: Uint8Array.of(INPUT_BYTE), + requestId: 'surviving-input' + }) + }); + await survivorInput; + await expect(failedRequest.finishAsync()).rejects.toThrow('request stdin failed'); + await survivingRequest.finishAsync(); + await expect( + exchangeControlAsync(failureClient, { kind: 'ping', payload: {} }) + ).resolves.toMatchObject({ kind: 'pong' }); + } finally { + await failureClient.closeAsync(); + await failureHost.closeAsync(); + } + }); + it('signals readiness only after the listener and lockfile are available', async () => { const controller: AbortController = new AbortController(); let readyPaths: IDaemonPaths | undefined;