diff --git a/common/changes/@rushstack/rush-reporter/docs-rush-reporter-overhaul-spec_2026-07-15-01-49-01.json b/common/changes/@rushstack/rush-reporter/docs-rush-reporter-overhaul-spec_2026-07-15-01-49-01.json new file mode 100644 index 0000000000..32d08e7f3b --- /dev/null +++ b/common/changes/@rushstack/rush-reporter/docs-rush-reporter-overhaul-spec_2026-07-15-01-49-01.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-reporter", + "comment": "Add Heft integration over a negotiated inherited descriptor: HeftChildEmitter and HeftDescriptorHost with parent/child event correlation, a raw-stream fallback for older Heft, and descriptor allocation helpers", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-reporter", + "email": "TheLarkInn@users.noreply.github.com" +} diff --git a/common/reviews/api/rush-reporter.api.md b/common/reviews/api/rush-reporter.api.md index adac085eb9..8143cbc235 100644 --- a/common/reviews/api/rush-reporter.api.md +++ b/common/reviews/api/rush-reporter.api.md @@ -4,6 +4,9 @@ ```ts +import type { Readable } from 'node:stream'; +import type { Writable } from 'node:stream'; + // @beta export class AiReporter implements IReporter { constructor(options: IAiReporterOptions); @@ -19,6 +22,9 @@ export class AiReporter implements IReporter { report(event: IReporterEventEnvelope): void; } +// @beta +export function allocateChildDescriptor(fdNumber?: number): IChildDescriptorPlan; + // @beta export const ALREADY_REPORTED_ERROR_NAME: 'AlreadyReportedError'; @@ -167,6 +173,30 @@ export function getPrivacyClassificationRank(classification: ReporterPrivacyClas // @beta export function getSignalExitCode(signal: NodeJS.Signals): number; +// @beta +export class HeftChildEmitter { + constructor(options: IHeftChildEmitterOptions); + emitEvent(input: IHeftChildEventInput): string | undefined; + readonly mode: HeftChildReporterMode; + sendHello(): boolean; + writeRaw(stream: 'stdout' | 'stderr', text: string): void; +} + +// @beta +export type HeftChildReporterMode = 'structured' | 'raw-fallback'; + +// @beta +export class HeftDescriptorHost { + constructor(options: IHeftDescriptorHostOptions); + createStreamProcessor(): { + write(chunk: string): void; + flush(): IHeftChildResult; + }; + processChildNdjson(ndjson: string): IHeftChildResult; + processChildRecord(record: unknown): boolean; + processChildRecords(records: readonly unknown[]): IHeftChildResult; +} + // @beta export interface IAiDiagnostic { // (undocumented) @@ -297,6 +327,13 @@ export interface IBootstrapTruncation { readonly truncated: boolean; } +// @beta +export interface IChildDescriptorPlan { + readonly env: Record; + readonly fdNumber: number; + readonly stdio: (string | number)[]; +} + // @beta export interface IClassifiedDiagnosticValue { readonly privacy: ReporterPrivacyClassification; @@ -418,6 +455,63 @@ export interface IGetMatchersOptions { readonly version?: string; } +// @beta +export interface IHeftChildEmitterOptions { + readonly capabilities?: readonly string[]; + readonly childSessionId: string; + readonly env: Record; + readonly now?: () => string; + readonly producerVersion: string; + readonly protocolVersion?: IReporterProtocolVersion; + readonly requiredFeatures?: readonly string[]; + readonly source: IReporterEventSource; + readonly writeDescriptor?: (text: string) => void; + readonly writeStderr?: (text: string) => void; + readonly writeStdout?: (text: string) => void; +} + +// @beta +export interface IHeftChildEventInput { + // (undocumented) + readonly payload?: unknown; + // (undocumented) + readonly privacy?: 'public' | 'local-sensitive' | 'secret'; + // (undocumented) + readonly scope?: IReporterEventScope; + // (undocumented) + readonly type: ReporterEventType; +} + +// @beta +export interface IHeftChildOutputStreams { + readonly stderr: Readable | null; + readonly stdout: Readable | null; +} + +// @beta +export interface IHeftChildOutputTargets { + readonly stderr: Writable; + readonly stdout: Writable; +} + +// @beta +export interface IHeftChildResult { + readonly accepted: boolean; + readonly ack?: IReporterHelloAck; + readonly diagnostic?: IRushDiagnostic; + readonly eventCount: number; +} + +// @beta +export interface IHeftDescriptorHostOptions { + readonly forwardEnvelope: (envelope: IReporterEventEnvelope) => void; + readonly onNegotiation?: (result: IReporterHandshakeResult) => void; + readonly parentOperationId?: string; + readonly parentSessionId: string; + readonly supportedCapabilities?: readonly string[]; + readonly supportedProtocolVersion: IReporterProtocolVersion; +} + // @beta export interface IInteractiveTerminal { readonly columns: number; @@ -1131,9 +1225,15 @@ export function readBootstrapHandoffFileAsync(filePath: string): Promise<{ discardedRecordCount: number; }>; +// @beta +export function readChildDescriptorFd(env: Record): number | undefined; + // @beta export function regroupOperationOutput(events: readonly IReporterEventEnvelope[]): Map; +// @beta +export function relayHeftChildOutput(child: IHeftChildOutputStreams, targets?: IHeftChildOutputTargets): void; + // @beta export function renderActiveProjectsRow(projects: readonly string[], width: number): string; @@ -1285,6 +1385,12 @@ export const RUSH_DIAGNOSTIC_CODE_DEFINITIONS: readonly [{ readonly defaultSeverity: "error"; readonly summaryKey: "diagnostic.RUSH_PROTOCOL_UPDATE_REQUIRED.summary"; readonly detailKey: "diagnostic.RUSH_PROTOCOL_UPDATE_REQUIRED.detail"; +}, { + readonly code: "RUSH_PROTOCOL_INVALID_CHILD_STREAM"; + readonly category: "environment"; + readonly defaultSeverity: "error"; + readonly summaryKey: "diagnostic.RUSH_PROTOCOL_INVALID_CHILD_STREAM.summary"; + readonly detailKey: "diagnostic.RUSH_PROTOCOL_INVALID_CHILD_STREAM.detail"; }, { readonly code: "RUSH_INTERNAL_UNEXPECTED"; readonly category: "internal"; @@ -1326,6 +1432,9 @@ export const RUSH_REPORTER_BOOTSTRAP_HANDOFF_ENV_VAR: '_RUSH_REPORTER_BOOTSTRAP_ // @beta export const RUSH_REPORTER_BOOTSTRAP_NONCE_ENV_VAR: '_RUSH_REPORTER_BOOTSTRAP_NONCE'; +// @beta +export const RUSH_REPORTER_CHILD_FD_ENV_VAR: '_RUSH_REPORTER_CHILD_FD'; + // @beta export const RUSH_REPORTER_ENV_VAR: 'RUSH_REPORTER'; diff --git a/libraries/reporter/src/diagnostics/RushDiagnosticCodeRegistry.ts b/libraries/reporter/src/diagnostics/RushDiagnosticCodeRegistry.ts index f67e2bf659..11f5c1d933 100644 --- a/libraries/reporter/src/diagnostics/RushDiagnosticCodeRegistry.ts +++ b/libraries/reporter/src/diagnostics/RushDiagnosticCodeRegistry.ts @@ -206,6 +206,13 @@ export const RUSH_DIAGNOSTIC_CODE_DEFINITIONS = defineRushDiagnosticCodeDefiniti summaryKey: 'diagnostic.RUSH_PROTOCOL_UPDATE_REQUIRED.summary', detailKey: 'diagnostic.RUSH_PROTOCOL_UPDATE_REQUIRED.detail' }, + { + code: 'RUSH_PROTOCOL_INVALID_CHILD_STREAM', + category: 'environment', + defaultSeverity: 'error', + summaryKey: 'diagnostic.RUSH_PROTOCOL_INVALID_CHILD_STREAM.summary', + detailKey: 'diagnostic.RUSH_PROTOCOL_INVALID_CHILD_STREAM.detail' + }, { code: RUSH_INTERNAL_ERROR_CODE, category: 'internal', diff --git a/libraries/reporter/src/diagnostics/templates/environment.ts b/libraries/reporter/src/diagnostics/templates/environment.ts index 8f90bff81d..ff3fd7b732 100644 --- a/libraries/reporter/src/diagnostics/templates/environment.ts +++ b/libraries/reporter/src/diagnostics/templates/environment.ts @@ -15,5 +15,8 @@ export const ENVIRONMENT_DIAGNOSTIC_TEMPLATES = { 'diagnostic.RUSH_PROTOCOL_UPDATE_REQUIRED.summary': 'A reporter protocol feature required by {producerVersion} is not supported by this Rush.', 'diagnostic.RUSH_PROTOCOL_UPDATE_REQUIRED.detail': - 'The producer advertised protocol major {producerProtocolMajor}. Update your global Rush installation to a version that supports it.' + 'The producer advertised protocol major {producerProtocolMajor}. Update your global Rush installation to a version that supports it.', + 'diagnostic.RUSH_PROTOCOL_INVALID_CHILD_STREAM.summary': + 'A child process sent an invalid reporter protocol stream.', + 'diagnostic.RUSH_PROTOCOL_INVALID_CHILD_STREAM.detail': 'The child reporter stream was rejected because {reason}.' } as const; diff --git a/libraries/reporter/src/heft/HeftChildEmitter.ts b/libraries/reporter/src/heft/HeftChildEmitter.ts new file mode 100644 index 0000000000..0178e5cefa --- /dev/null +++ b/libraries/reporter/src/heft/HeftChildEmitter.ts @@ -0,0 +1,198 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IReporterProtocolVersion } from '../events/ReporterProtocolVersion'; +import type { IReporterEventScope, IReporterEventSource } from '../events/IReporterEventEnvelope'; +import { isReporterEventRequired, type ReporterEventType } from '../events/ReporterEventType'; +import { encodeNdjsonRecord } from '../protocol/Ndjson'; +import { REPORTER_PROTOCOL_VERSION } from '../protocol/ReporterProtocol'; +import type { IReporterHello } from '../protocol/ReporterHandshake'; +import { readChildDescriptorFd, RUSH_REPORTER_CHILD_FD_ENV_VAR } from './HeftDescriptor'; + +/** + * The mode a Heft child reporter operates in. + * + * @beta + */ +export type HeftChildReporterMode = 'structured' | 'raw-fallback'; + +/** + * An event a Heft child emits. + * + * @beta + */ +export interface IHeftChildEventInput { + readonly type: ReporterEventType; + readonly privacy?: 'public' | 'local-sensitive' | 'secret'; + readonly scope?: IReporterEventScope; + readonly payload?: unknown; +} + +/** + * Options for {@link HeftChildEmitter}. + * + * @beta + */ +export interface IHeftChildEmitterOptions { + /** + * The environment variables, consulted for the inherited descriptor. The + * descriptor variable is removed when the emitter is constructed so it is + * not inherited by descendants that do not inherit the descriptor itself. + */ + readonly env: Record; + + /** + * The child session id stamped onto emitted events. + */ + readonly childSessionId: string; + + /** + * The producer identity stamped onto emitted events. + */ + readonly source: IReporterEventSource; + + /** + * The producer version advertised in the hello. + */ + readonly producerVersion: string; + + /** + * The protocol version. Defaults to {@link REPORTER_PROTOCOL_VERSION}. + */ + readonly protocolVersion?: IReporterProtocolVersion; + + /** + * The capabilities advertised in the hello. + */ + readonly capabilities?: readonly string[]; + + /** + * The required features advertised in the hello. + */ + readonly requiredFeatures?: readonly string[]; + + /** + * Writes NDJSON to the inherited descriptor. Required for structured mode. + */ + readonly writeDescriptor?: (text: string) => void; + + /** + * Writes raw text to stdout, used in fallback mode. + */ + readonly writeStdout?: (text: string) => void; + + /** + * Writes raw text to stderr, used in fallback mode. + */ + readonly writeStderr?: (text: string) => void; + + /** + * Returns the current timestamp. Injectable for testing. + */ + readonly now?: () => string; +} + +/** + * The child side of the Heft reporter descriptor negotiation. + * + * @remarks + * When the inherited descriptor is present, the child emits structured NDJSON + * events over it, stamping its child session id. When the descriptor is + * unavailable, it falls back to normal stdout and stderr, which Rush preserves + * and runs through problem matchers. + * + * @beta + */ +export class HeftChildEmitter { + /** + * Whether the child emits structured events or falls back to raw streams. + */ + public readonly mode: HeftChildReporterMode; + + private readonly _writeDescriptor: ((text: string) => void) | undefined; + private readonly _writeStdout: ((text: string) => void) | undefined; + private readonly _writeStderr: ((text: string) => void) | undefined; + private readonly _childSessionId: string; + private readonly _source: IReporterEventSource; + private readonly _producerVersion: string; + private readonly _protocolVersion: IReporterProtocolVersion; + private readonly _capabilities: readonly string[]; + private readonly _requiredFeatures: readonly string[]; + private readonly _now: () => string; + private _sequence: number; + private _nextEventId: number; + + public constructor(options: IHeftChildEmitterOptions) { + const fd: number | undefined = readChildDescriptorFd(options.env); + delete options.env[RUSH_REPORTER_CHILD_FD_ENV_VAR]; + this.mode = fd !== undefined && options.writeDescriptor !== undefined ? 'structured' : 'raw-fallback'; + + this._writeDescriptor = options.writeDescriptor; + this._writeStdout = options.writeStdout; + this._writeStderr = options.writeStderr; + this._childSessionId = options.childSessionId; + this._source = options.source; + this._producerVersion = options.producerVersion; + this._protocolVersion = options.protocolVersion ?? REPORTER_PROTOCOL_VERSION; + this._capabilities = options.capabilities ?? []; + this._requiredFeatures = options.requiredFeatures ?? []; + this._now = options.now ?? (() => new Date().toISOString()); + this._sequence = 1; + this._nextEventId = 1; + } + + /** + * Sends the hello handshake over the descriptor. Returns `false` in fallback mode. + */ + public sendHello(): boolean { + if (this.mode !== 'structured' || this._writeDescriptor === undefined) { + return false; + } + const hello: IReporterHello = { + kind: 'hello', + protocolVersion: this._protocolVersion, + producerVersion: this._producerVersion, + capabilities: [...this._capabilities], + requiredFeatures: [...this._requiredFeatures] + }; + this._writeDescriptor(encodeNdjsonRecord(hello)); + return true; + } + + /** + * Emits a structured event over the descriptor. Returns the event id, or + * `undefined` in fallback mode. + */ + public emitEvent(input: IHeftChildEventInput): string | undefined { + if (this.mode !== 'structured' || this._writeDescriptor === undefined) { + return undefined; + } + const eventId: string = `child_${this._nextEventId++}`; + const envelope: Record = { + protocolVersion: this._protocolVersion, + eventId, + sessionId: this._childSessionId, + sequence: this._sequence++, + timestamp: this._now(), + source: this._source, + scope: input.scope, + privacy: input.privacy ?? 'public', + required: isReporterEventRequired(input.type), + type: input.type, + payload: input.payload ?? {} + }; + this._writeDescriptor(encodeNdjsonRecord(envelope)); + return eventId; + } + + /** + * Writes raw output to stdout or stderr, preserved for problem matchers. + */ + public writeRaw(stream: 'stdout' | 'stderr', text: string): void { + if (stream === 'stderr') { + this._writeStderr?.(text); + } else { + this._writeStdout?.(text); + } + } +} diff --git a/libraries/reporter/src/heft/HeftDescriptor.ts b/libraries/reporter/src/heft/HeftDescriptor.ts new file mode 100644 index 0000000000..2e6e9b921c --- /dev/null +++ b/libraries/reporter/src/heft/HeftDescriptor.ts @@ -0,0 +1,140 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { Readable, Writable } from 'node:stream'; + +/** + * The private environment variable that communicates the inherited reporter file + * descriptor number to a child process. + * + * @beta + */ +export const RUSH_REPORTER_CHILD_FD_ENV_VAR: '_RUSH_REPORTER_CHILD_FD' = '_RUSH_REPORTER_CHILD_FD'; + +/** + * A plan for launching a child with an inherited reporter descriptor. + * + * @beta + */ +export interface IChildDescriptorPlan { + /** + * The inherited file descriptor number the child writes NDJSON to. + */ + readonly fdNumber: number; + + /** + * The environment additions that communicate the descriptor to the child. + */ + readonly env: Record; + + /** + * The stdio configuration for spawning the child. stdout and stderr remain + * normal process streams; the reporter descriptor is an additional pipe. + */ + readonly stdio: (string | number)[]; +} + +/** + * The piped standard streams exposed by a spawned Heft child. + * + * @beta + */ +export interface IHeftChildOutputStreams { + /** + * The child's standard output stream, or `null` when it was not configured as a pipe. + */ + // eslint-disable-next-line @rushstack/no-new-null + readonly stdout: Readable | null; + + /** + * The child's standard error stream, or `null` when it was not configured as a pipe. + */ + // eslint-disable-next-line @rushstack/no-new-null + readonly stderr: Readable | null; +} + +/** + * The parent output streams that receive a Heft child's raw fallback output. + * + * @beta + */ +export interface IHeftChildOutputTargets { + /** + * The parent stream that receives the child's standard output. + */ + readonly stdout: Writable; + + /** + * The parent stream that receives the child's standard error. + */ + readonly stderr: Writable; +} + +/** + * Allocates a dynamic inherited descriptor for a child reporter. + * + * @remarks + * stdout and stderr are piped so the parent can preserve and inspect old-Heft + * fallback output before relaying it to the normal output streams. The reporter + * descriptor is an additional pipe at `fdNumber`, whose number is communicated + * through the private environment variable. + * + * @param fdNumber - the descriptor number; defaults to 3 + * + * @beta + */ +export function allocateChildDescriptor(fdNumber: number = 3): IChildDescriptorPlan { + if (!Number.isSafeInteger(fdNumber) || fdNumber < 3) { + throw new RangeError( + 'The reporter file descriptor number must be an integer greater than or equal to 3.' + ); + } + + const stdio: (string | number)[] = ['inherit', 'pipe', 'pipe']; + while (stdio.length < fdNumber) { + stdio.push('ignore'); + } + stdio[fdNumber] = 'pipe'; + return { + fdNumber, + env: { [RUSH_REPORTER_CHILD_FD_ENV_VAR]: String(fdNumber) }, + stdio + }; +} + +/** + * Relays a spawned Heft child's piped fallback output to the normal parent + * output streams without closing those parent streams. + * + * @beta + */ +export function relayHeftChildOutput( + child: IHeftChildOutputStreams, + targets: IHeftChildOutputTargets = { stdout: process.stdout, stderr: process.stderr } +): void { + if (child.stdout === null || child.stderr === null) { + throw new Error('The Heft child must be spawned with piped stdout and stderr.'); + } + child.stdout.pipe(targets.stdout, { end: false }); + child.stderr.pipe(targets.stderr, { end: false }); +} + +/** + * Reads the inherited reporter descriptor number from the environment. + * + * @remarks + * Returns `undefined` when descriptor negotiation is unavailable, in which case + * the child falls back to normal stdout and stderr. + * + * @param env - the environment variables + * + * @beta + */ +export function readChildDescriptorFd(env: Record): number | undefined { + const raw: string | undefined = env[RUSH_REPORTER_CHILD_FD_ENV_VAR]; + if (raw === undefined || !/^\d+$/.test(raw)) { + return undefined; + } + const parsed: number = Number(raw); + return Number.isSafeInteger(parsed) && parsed >= 3 ? parsed : undefined; +} diff --git a/libraries/reporter/src/heft/HeftDescriptorHost.ts b/libraries/reporter/src/heft/HeftDescriptorHost.ts new file mode 100644 index 0000000000..52ad8feb4a --- /dev/null +++ b/libraries/reporter/src/heft/HeftDescriptorHost.ts @@ -0,0 +1,374 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IReporterProtocolVersion } from '../events/ReporterProtocolVersion'; +import type { IReporterEventEnvelope } from '../events/IReporterEventEnvelope'; +import { + REPORTER_EVENT_TYPES, + isReporterEventRequired, + type ReporterEventType +} from '../events/ReporterEventType'; +import type { IRushDiagnostic } from '../diagnostics/IRushDiagnostic'; +import { createRushDiagnostic } from '../diagnostics/createRushDiagnostic'; +import { NdjsonDecoder } from '../protocol/Ndjson'; +import { + negotiateReporterHello, + type IReporterHello, + type IReporterHelloAck, + type IReporterHandshakeResult +} from '../protocol/ReporterHandshake'; + +const REPORTER_EVENT_TYPE_SET: ReadonlySet = new Set(REPORTER_EVENT_TYPES); + +type IWireReporterEventEnvelope = Omit, 'type'> & { + readonly type: string; +}; + +function isObjectRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function isNonNegativeInteger(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} + +function isProtocolVersion(value: unknown): value is IReporterProtocolVersion { + if (!isObjectRecord(value)) { + return false; + } + return isNonNegativeInteger(value.major) && isNonNegativeInteger(value.minor); +} + +function isStringArray(value: unknown): value is readonly string[] { + return Array.isArray(value) && value.every((item: unknown) => typeof item === 'string'); +} + +function isReporterHello(value: unknown): value is IReporterHello { + if (!isObjectRecord(value)) { + return false; + } + return ( + value.kind === 'hello' && + isProtocolVersion(value.protocolVersion) && + typeof value.producerVersion === 'string' && + isStringArray(value.capabilities) && + isStringArray(value.requiredFeatures) + ); +} + +function isReporterEventType(value: string): value is ReporterEventType { + return REPORTER_EVENT_TYPE_SET.has(value); +} + +function isReporterEventSource(value: unknown): boolean { + if (!isObjectRecord(value)) { + return false; + } + return ( + typeof value.packageName === 'string' && + typeof value.packageVersion === 'string' && + (value.component === undefined || typeof value.component === 'string') + ); +} + +function isReporterEventScope(value: unknown): boolean { + if (!isObjectRecord(value)) { + return false; + } + return ['commandName', 'operationId', 'projectName', 'phaseName'].every( + (key: string) => value[key] === undefined || typeof value[key] === 'string' + ); +} + +function isReporterEventRecord(value: unknown): value is IWireReporterEventEnvelope { + if (!isObjectRecord(value)) { + return false; + } + return ( + isProtocolVersion(value.protocolVersion) && + typeof value.eventId === 'string' && + value.eventId.length > 0 && + typeof value.sessionId === 'string' && + value.sessionId.length > 0 && + isNonNegativeInteger(value.sequence) && + typeof value.timestamp === 'string' && + isReporterEventSource(value.source) && + (value.scope === undefined || isReporterEventScope(value.scope)) && + (value.privacy === 'public' || value.privacy === 'local-sensitive' || value.privacy === 'secret') && + typeof value.required === 'boolean' && + typeof value.type === 'string' && + Object.prototype.hasOwnProperty.call(value, 'payload') + ); +} + +/** + * Options for constructing a {@link HeftDescriptorHost}. + * + * @beta + */ +export interface IHeftDescriptorHostOptions { + /** + * The parent session id used to correlate child events. + */ + readonly parentSessionId: string; + + /** + * The parent operation id used to correlate child events. + */ + readonly parentOperationId?: string; + + /** + * The protocol version the parent supports. + */ + readonly supportedProtocolVersion: IReporterProtocolVersion; + + /** + * The capabilities the parent supports. + */ + readonly supportedCapabilities?: readonly string[]; + + /** + * Forwards a correlated child envelope, typically to `ReporterManager.ingestForeignEnvelope`. + */ + readonly forwardEnvelope: (envelope: IReporterEventEnvelope) => void; + + /** + * Receives the handshake outcome, typically to emit the rejection diagnostic. + * Called once when the first record is accepted or rejected. + */ + readonly onNegotiation?: (result: IReporterHandshakeResult) => void; +} + +/** + * The result of consuming a child reporter stream. + * + * @beta + */ +export interface IHeftChildResult { + /** + * Whether the child's protocol was accepted. + */ + readonly accepted: boolean; + + /** + * The number of events forwarded. + */ + readonly eventCount: number; + + /** + * The acknowledgement produced while negotiating the stream. + */ + readonly ack?: IReporterHelloAck; + + /** + * A protocol diagnostic, when the child was rejected. + */ + readonly diagnostic?: IRushDiagnostic; +} + +/** + * The parent side of the Heft reporter descriptor negotiation. + * + * @remarks + * The host negotiates the child's hello, and, on acceptance, correlates each + * child event with the parent session and operation ids before forwarding it. + * When the child is rejected it surfaces a protocol diagnostic. + * + * Use {@link HeftDescriptorHost.createStreamProcessor} for a live child: it + * drains the descriptor pipe as records arrive (so a chatty child never blocks + * on a full OS pipe buffer) and forwards each event in receipt order. The + * batch {@link HeftDescriptorHost.processChildNdjson} path is retained for + * tests and completed streams. + * + * @beta + */ +export class HeftDescriptorHost { + private readonly _parentSessionId: string; + private readonly _parentOperationId: string | undefined; + private readonly _supportedProtocolVersion: IReporterProtocolVersion; + private readonly _supportedCapabilities: readonly string[] | undefined; + private readonly _forwardEnvelope: (envelope: IReporterEventEnvelope) => void; + private readonly _onNegotiation: ((result: IReporterHandshakeResult) => void) | undefined; + + private _negotiation: IReporterHandshakeResult | undefined; + private _protocolFailure: IRushDiagnostic | undefined; + private _eventCount: number = 0; + + public constructor(options: IHeftDescriptorHostOptions) { + this._parentSessionId = options.parentSessionId; + this._parentOperationId = options.parentOperationId; + this._supportedProtocolVersion = options.supportedProtocolVersion; + this._supportedCapabilities = options.supportedCapabilities; + this._forwardEnvelope = options.forwardEnvelope; + this._onNegotiation = options.onNegotiation; + } + + /** + * Processes a single decoded child record: the hello, then event envelopes. + * + * @remarks + * On first call the record must be the hello; the negotiation outcome is + * reported through {@link IHeftDescriptorHostOptions.onNegotiation} and, once + * rejected, subsequent records are dropped. Returns `true` while the stream + * is accepted. + */ + public processChildRecord(record: unknown): boolean { + if (this._protocolFailure !== undefined) { + return false; + } + + if (this._negotiation === undefined) { + if (!isReporterHello(record)) { + return this._rejectMalformedStream('the first record was not a valid hello'); + } + const result: IReporterHandshakeResult = negotiateReporterHello(record, { + supportedProtocolVersion: this._supportedProtocolVersion, + supportedCapabilities: this._supportedCapabilities + }); + this._negotiation = result; + this._onNegotiation?.(result); + return result.accepted; + } + if (!this._negotiation.accepted) { + return false; + } + + if (!isReporterEventRecord(record)) { + return this._rejectMalformedStream('an event record did not contain a valid reporter envelope'); + } + if (record.protocolVersion.major !== this._negotiation.ack.protocolVersion.major) { + return this._rejectMalformedStream( + 'an event record used a protocol major different from the negotiated stream' + ); + } + if (!isReporterEventType(record.type)) { + if (record.required) { + return this._rejectMalformedStream('a required event type was not recognized'); + } + return true; + } + + const correlated: IReporterEventEnvelope = { + ...record, + parentSessionId: this._parentSessionId, + parentOperationId: this._parentOperationId, + required: isReporterEventRequired(record.type), + type: record.type + }; + this._forwardEnvelope(correlated); + this._eventCount++; + return true; + } + + /** + * Creates an incremental processor that decodes NDJSON chunks as they arrive + * and forwards accepted events in receipt order. + * + * @remarks + * This is the streaming drain: feed it each chunk read from the child's + * descriptor pipe so the pipe is continuously drained and child progress + * appears live. Call `flush()` when the pipe closes. + */ + public createStreamProcessor(): { write(chunk: string): void; flush(): IHeftChildResult } { + const decoder: NdjsonDecoder = new NdjsonDecoder(); + return { + write: (chunk: string): void => { + if (this._protocolFailure !== undefined || this._negotiation?.accepted === false) { + return; + } + let records: unknown[]; + try { + records = decoder.decode(chunk); + } catch { + this._rejectMalformedStream('its NDJSON could not be decoded within the protocol limits'); + return; + } + for (const record of records) { + this.processChildRecord(record); + } + }, + flush: (): IHeftChildResult => { + if (this._protocolFailure === undefined && this._negotiation?.accepted !== false) { + let records: unknown[]; + try { + records = decoder.flush(); + } catch { + this._rejectMalformedStream('its trailing NDJSON record was invalid'); + return this._result(); + } + for (const record of records) { + this.processChildRecord(record); + } + } + + return this._result(); + } + }; + } + + /** + * Processes decoded child records: a hello followed by event envelopes. + */ + public processChildRecords(records: readonly unknown[]): IHeftChildResult { + for (const record of records) { + this.processChildRecord(record); + } + return this._result(); + } + + /** + * Decodes and processes a child's complete NDJSON stream. + * + * @remarks + * Retained for tests and completed streams; live children should use + * {@link HeftDescriptorHost.createStreamProcessor}. + */ + public processChildNdjson(ndjson: string): IHeftChildResult { + const processor: { write(chunk: string): void; flush(): IHeftChildResult } = this.createStreamProcessor(); + processor.write(ndjson); + return processor.flush(); + } + + private _result(): IHeftChildResult { + const negotiation: IReporterHandshakeResult | undefined = this._negotiation; + if (negotiation === undefined) { + return { accepted: false, eventCount: 0 }; + } + return { + accepted: negotiation.accepted && this._protocolFailure === undefined, + eventCount: this._eventCount, + ...('ack' in negotiation && negotiation.ack !== undefined ? { ack: negotiation.ack } : {}), + ...(this._protocolFailure !== undefined + ? { diagnostic: this._protocolFailure } + : 'diagnostic' in negotiation && negotiation.diagnostic !== undefined + ? { diagnostic: negotiation.diagnostic } + : {}) + }; + } + + private _rejectMalformedStream(reason: string): false { + if (this._protocolFailure === undefined) { + this._protocolFailure = createRushDiagnostic('RUSH_PROTOCOL_INVALID_CHILD_STREAM', { + parameters: { + reason: { value: reason, privacy: 'public' } + } + }); + } + + if (this._negotiation === undefined) { + const result: IReporterHandshakeResult = { + accepted: false, + ack: { + kind: 'helloAck', + protocolVersion: this._supportedProtocolVersion, + acceptedCapabilities: [], + rejectedRequiredFeatures: [] + }, + diagnostic: this._protocolFailure + }; + this._negotiation = result; + this._onNegotiation?.(result); + } + return false; + } +} diff --git a/libraries/reporter/src/index.ts b/libraries/reporter/src/index.ts index 5ec209889e..8386ff8652 100644 --- a/libraries/reporter/src/index.ts +++ b/libraries/reporter/src/index.ts @@ -299,6 +299,26 @@ export { ProblemMatcherRegistry } from './matchers/ProblemMatcherRegistry'; export type { IRunProblemMatchersOptions, IProblemMatcherResult } from './matchers/ProblemMatcherRunner'; export { runProblemMatchers } from './matchers/ProblemMatcherRunner'; +export type { + IChildDescriptorPlan, + IHeftChildOutputStreams, + IHeftChildOutputTargets +} from './heft/HeftDescriptor'; +export { + RUSH_REPORTER_CHILD_FD_ENV_VAR, + allocateChildDescriptor, + readChildDescriptorFd, + relayHeftChildOutput +} from './heft/HeftDescriptor'; +export type { + HeftChildReporterMode, + IHeftChildEventInput, + IHeftChildEmitterOptions +} from './heft/HeftChildEmitter'; +export { HeftChildEmitter } from './heft/HeftChildEmitter'; +export type { IHeftDescriptorHostOptions, IHeftChildResult } from './heft/HeftDescriptorHost'; +export { HeftDescriptorHost } from './heft/HeftDescriptorHost'; + export type { IReporterEmitEventInput, IReporterEventSink } from './producers/IReporterEventSink'; export type { ReporterMessageSeverity, diff --git a/libraries/reporter/src/test/HeftIntegration.test.ts b/libraries/reporter/src/test/HeftIntegration.test.ts new file mode 100644 index 0000000000..5d2990e33a --- /dev/null +++ b/libraries/reporter/src/test/HeftIntegration.test.ts @@ -0,0 +1,441 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as childProcess from 'node:child_process'; +import { PassThrough, type Readable } from 'node:stream'; + +import { + allocateChildDescriptor, + readChildDescriptorFd, + RUSH_REPORTER_CHILD_FD_ENV_VAR, + HeftChildEmitter, + HeftDescriptorHost, + relayHeftChildOutput, + ReporterManager, + runProblemMatchers, + type IChildDescriptorPlan, + type IHeftChildResult, + type IProblemMatch, + type IProblemMatcher, + type IReporter, + type IReporterEventEnvelope, + type IReporterEventSource +} from '../index'; + +const SOURCE: IReporterEventSource = { packageName: '@rushstack/heft', packageVersion: '1.2.19' }; + +class RecordingReporter implements IReporter { + public readonly name: string = 'recording'; + public readonly reported: IReporterEventEnvelope[] = []; + public async initializeAsync(): Promise { + /* no-op */ + } + public report(event: IReporterEventEnvelope): void { + this.reported.push(event); + } + public async flushAsync(): Promise { + /* no-op */ + } + public async closeAsync(): Promise { + /* no-op */ + } +} + +const TSC_MATCHER: IProblemMatcher = { + name: 'tsc-error', + tool: 'tsc', + severity: 'error', + enabledByDefault: true, + pattern: /^(.+)\((\d+),(\d+)\): error (TS\d+): (.+)$/, + extract(match: RegExpMatchArray): IProblemMatch { + return { + file: match[1], + line: Number(match[2]), + column: Number(match[3]), + code: match[4], + message: match[5] + }; + } +}; + +describe('Heft descriptor allocation', () => { + it('allocates an inherited descriptor and communicates it by env var', () => { + const plan: IChildDescriptorPlan = allocateChildDescriptor(); + expect(plan.fdNumber).toBe(3); + expect(plan.env[RUSH_REPORTER_CHILD_FD_ENV_VAR]).toBe('3'); + expect(plan.stdio[3]).toBe('pipe'); + expect(plan.stdio.slice(0, 3)).toEqual(['inherit', 'pipe', 'pipe']); + }); + + it('reads or rejects the descriptor number from the environment', () => { + expect(readChildDescriptorFd({ [RUSH_REPORTER_CHILD_FD_ENV_VAR]: '3' })).toBe(3); + expect(readChildDescriptorFd({})).toBeUndefined(); + expect(readChildDescriptorFd({ [RUSH_REPORTER_CHILD_FD_ENV_VAR]: 'abc' })).toBeUndefined(); + expect(readChildDescriptorFd({ [RUSH_REPORTER_CHILD_FD_ENV_VAR]: '3abc' })).toBeUndefined(); + expect(readChildDescriptorFd({ [RUSH_REPORTER_CHILD_FD_ENV_VAR]: '2' })).toBeUndefined(); + }); + + it('rejects descriptor numbers that would replace standard streams', () => { + expect(() => allocateChildDescriptor(2)).toThrow(/greater than or equal to 3/); + }); +}); + +describe('HeftChildEmitter', () => { + it('emits structured NDJSON when the descriptor is present', () => { + let descriptor: string = ''; + const env: Record = { [RUSH_REPORTER_CHILD_FD_ENV_VAR]: '3' }; + const emitter: HeftChildEmitter = new HeftChildEmitter({ + env, + childSessionId: 'child-sess', + source: SOURCE, + producerVersion: '@rushstack/heft 1.2.19', + now: () => '2026-01-01T00:00:00.000Z', + writeDescriptor: (text: string) => (descriptor += text) + }); + expect(emitter.mode).toBe('structured'); + expect(env[RUSH_REPORTER_CHILD_FD_ENV_VAR]).toBeUndefined(); + expect(emitter.sendHello()).toBe(true); + const eventId: string | undefined = emitter.emitEvent({ + type: 'commandStarted', + payload: {} + }); + emitter.emitEvent({ type: 'activityChanged', payload: {} }); + expect(eventId).toBe('child_1'); + + const records: Record[] = descriptor + .trim() + .split('\n') + .map((line: string) => JSON.parse(line) as Record); + expect(records[0].kind).toBe('hello'); + expect(records[1].sessionId).toBe('child-sess'); + expect(records[1].type).toBe('commandStarted'); + expect(records[1].required).toBe(true); + expect(records[2].required).toBe(false); + }); + + it('falls back to raw streams when descriptor negotiation is unavailable', () => { + let stdout: string = ''; + const emitter: HeftChildEmitter = new HeftChildEmitter({ + env: {}, + childSessionId: 'child-sess', + source: SOURCE, + producerVersion: '@rushstack/heft 1.2.19', + writeStdout: (text: string) => (stdout += text) + }); + expect(emitter.mode).toBe('raw-fallback'); + expect(emitter.sendHello()).toBe(false); + expect(emitter.emitEvent({ type: 'commandStarted' })).toBeUndefined(); + emitter.writeRaw('stdout', 'raw heft log\n'); + expect(stdout).toBe('raw heft log\n'); + }); +}); + +describe('HeftDescriptorHost new descriptor path', () => { + it('negotiates the hello and correlates forwarded child events', async () => { + // Child produces a structured stream. + let descriptor: string = ''; + const child: HeftChildEmitter = new HeftChildEmitter({ + env: { [RUSH_REPORTER_CHILD_FD_ENV_VAR]: '3' }, + childSessionId: 'child-sess', + source: SOURCE, + producerVersion: '@rushstack/heft 1.2.19', + now: () => '2026-01-01T00:00:00.000Z', + writeDescriptor: (text: string) => (descriptor += text) + }); + child.sendHello(); + child.emitEvent({ + type: 'operationStatusChanged', + payload: { operationId: 'c1', status: 'success' } + }); + + // Parent host forwards into a manager. + const manager: ReporterManager = new ReporterManager(); + const recording: RecordingReporter = new RecordingReporter(); + manager.addReporter(recording); + await manager.initializeAsync(); + + const host: HeftDescriptorHost = new HeftDescriptorHost({ + parentSessionId: 'parent-sess', + parentOperationId: 'op-42', + supportedProtocolVersion: { major: 1, minor: 0 }, + forwardEnvelope: (envelope: IReporterEventEnvelope) => manager.ingestForeignEnvelope(envelope) + }); + const result: IHeftChildResult = host.processChildNdjson(descriptor); + await manager.flushAsync(); + + expect(result.accepted).toBe(true); + expect(result.eventCount).toBe(1); + + const forwarded: IReporterEventEnvelope = recording.reported[0]; + expect(forwarded.sessionId).toBe('child-sess'); + expect(forwarded.parentSessionId).toBe('parent-sess'); + expect(forwarded.parentOperationId).toBe('op-42'); + // ingestForeignEnvelope assigns a new global sequence and preserves the child's. + expect(forwarded.sourceSequence).toBe(1); + }); + + it('drains a spawned child descriptor before the child exits and exceeds pipe capacity', async () => { + const manager: ReporterManager = new ReporterManager(); + const recording: RecordingReporter = new RecordingReporter(); + manager.addReporter(recording); + await manager.initializeAsync(); + + let childExited: boolean = false; + let forwardedBeforeExit: boolean = false; + const host: HeftDescriptorHost = new HeftDescriptorHost({ + parentSessionId: 'parent-sess', + supportedProtocolVersion: { major: 1, minor: 0 }, + forwardEnvelope: (envelope: IReporterEventEnvelope) => { + forwardedBeforeExit ||= !childExited; + manager.ingestForeignEnvelope(envelope); + } + }); + const processor = host.createStreamProcessor(); + const plan: IChildDescriptorPlan = allocateChildDescriptor(); + const eventCount: number = 2_000; + const script: string = ` + const fs = require('node:fs'); + const fd = Number(process.env.${RUSH_REPORTER_CHILD_FD_ENV_VAR}); + const source = ${JSON.stringify(SOURCE)}; + fs.writeSync(fd, JSON.stringify({ + kind: 'hello', + protocolVersion: { major: 1, minor: 0 }, + producerVersion: '@rushstack/heft 1.2.19', + capabilities: [], + requiredFeatures: [] + }) + '\\n'); + for (let i = 0; i < ${eventCount}; i++) { + fs.writeSync(fd, JSON.stringify({ + protocolVersion: { major: 1, minor: 0 }, + eventId: 'child_' + i, + sessionId: 'child-sess', + sequence: i + 1, + timestamp: '2026-01-01T00:00:00.000Z', + source, + privacy: 'public', + required: true, + type: 'operationStatusChanged', + payload: { operationId: 'operation-' + i, status: 'success', padding: 'x'.repeat(128) } + }) + '\\n'); + } + `; + const spawned: childProcess.ChildProcess = childProcess.spawn(process.execPath, ['-e', script], { + env: { ...process.env, ...plan.env }, + stdio: plan.stdio as childProcess.StdioOptions + }); + const descriptor: Readable = spawned.stdio[plan.fdNumber] as Readable; + descriptor.setEncoding('utf8'); + descriptor.on('data', (chunk: string) => processor.write(chunk)); + await new Promise((resolve, reject) => { + spawned.once('error', reject); + spawned.once('exit', (code: number | null) => { + childExited = true; + if (code === 0) { + resolve(); + } else { + reject(new Error(`Spawned Heft fixture exited with code ${code}.`)); + } + }); + }); + const result: IHeftChildResult = processor.flush(); + await manager.flushAsync(); + + expect(result.accepted).toBe(true); + expect(result.eventCount).toBe(eventCount); + expect(recording.reported).toHaveLength(eventCount); + expect(forwardedBeforeExit).toBe(true); + expect(recording.reported[0].sessionId).toBe('child-sess'); + }); + + it('rejects an unsupported child protocol with an update-global-Rush diagnostic', () => { + let descriptor: string = ''; + const child: HeftChildEmitter = new HeftChildEmitter({ + env: { [RUSH_REPORTER_CHILD_FD_ENV_VAR]: '3' }, + childSessionId: 'child-sess', + source: SOURCE, + producerVersion: '@rushstack/heft 2.0.0', + protocolVersion: { major: 2, minor: 0 }, + writeDescriptor: (text: string) => (descriptor += text) + }); + child.sendHello(); + + const host: HeftDescriptorHost = new HeftDescriptorHost({ + parentSessionId: 'parent-sess', + supportedProtocolVersion: { major: 1, minor: 0 }, + forwardEnvelope: () => undefined + }); + const result: IHeftChildResult = host.processChildNdjson(descriptor); + expect(result.accepted).toBe(false); + expect(result.diagnostic?.code).toBe('RUSH_PROTOCOL_UPDATE_REQUIRED'); + }); + + it('rejects malformed records without throwing from the streaming drain', () => { + const negotiationResults: boolean[] = []; + const host: HeftDescriptorHost = new HeftDescriptorHost({ + parentSessionId: 'parent-sess', + supportedProtocolVersion: { major: 1, minor: 0 }, + forwardEnvelope: () => { + throw new Error('Malformed records must not be forwarded.'); + }, + onNegotiation: (result) => negotiationResults.push(result.accepted) + }); + const processor = host.createStreamProcessor(); + + expect(() => processor.write('not json\n')).not.toThrow(); + expect(() => processor.write('null\n')).not.toThrow(); + const result: IHeftChildResult = processor.flush(); + + expect(result.accepted).toBe(false); + expect(result.eventCount).toBe(0); + expect(result.diagnostic?.code).toBe('RUSH_PROTOCOL_INVALID_CHILD_STREAM'); + expect(negotiationResults).toEqual([false]); + }); + + it('rejects an incomplete hello instead of dereferencing missing fields', () => { + const host: HeftDescriptorHost = new HeftDescriptorHost({ + parentSessionId: 'parent-sess', + supportedProtocolVersion: { major: 1, minor: 0 }, + forwardEnvelope: () => undefined + }); + + expect(() => host.processChildRecord({ kind: 'hello' })).not.toThrow(); + const result: IHeftChildResult = host.processChildRecords([]); + expect(result.accepted).toBe(false); + expect(result.diagnostic?.code).toBe('RUSH_PROTOCOL_INVALID_CHILD_STREAM'); + }); + + it('rejects malformed envelopes and derives required at the host boundary', () => { + const forwarded: IReporterEventEnvelope[] = []; + const host: HeftDescriptorHost = new HeftDescriptorHost({ + parentSessionId: 'parent-sess', + supportedProtocolVersion: { major: 1, minor: 0 }, + forwardEnvelope: (envelope: IReporterEventEnvelope) => forwarded.push(envelope) + }); + + expect( + host.processChildRecord({ + kind: 'hello', + protocolVersion: { major: 1, minor: 0 }, + producerVersion: '@rushstack/heft 1.2.19', + capabilities: [], + requiredFeatures: [] + }) + ).toBe(true); + expect( + host.processChildRecord({ + protocolVersion: { major: 1, minor: 0 }, + eventId: 'child_1', + sessionId: 'child-sess', + sequence: 1, + timestamp: '2026-01-01T00:00:00.000Z', + source: SOURCE, + privacy: 'public', + required: true, + type: 'activityChanged', + payload: {} + }) + ).toBe(true); + expect(forwarded[0].required).toBe(false); + + expect(host.processChildRecord(null)).toBe(false); + const result: IHeftChildResult = host.processChildRecords([]); + expect(result.accepted).toBe(false); + expect(result.eventCount).toBe(1); + expect(result.diagnostic?.code).toBe('RUSH_PROTOCOL_INVALID_CHILD_STREAM'); + }); + + it('rejects event envelopes whose protocol major differs from the accepted hello', () => { + const host: HeftDescriptorHost = new HeftDescriptorHost({ + parentSessionId: 'parent-sess', + supportedProtocolVersion: { major: 1, minor: 0 }, + forwardEnvelope: () => { + throw new Error('A mismatched protocol envelope must not be forwarded.'); + } + }); + expect( + host.processChildRecord({ + kind: 'hello', + protocolVersion: { major: 1, minor: 0 }, + producerVersion: '@rushstack/heft 1.2.19', + capabilities: [], + requiredFeatures: [] + }) + ).toBe(true); + expect( + host.processChildRecord({ + protocolVersion: { major: 2, minor: 0 }, + eventId: 'child_1', + sessionId: 'child-sess', + sequence: 1, + timestamp: '2026-01-01T00:00:00.000Z', + source: SOURCE, + privacy: 'public', + required: true, + type: 'commandStarted', + payload: {} + }) + ).toBe(false); + expect(host.processChildRecords([]).diagnostic?.code).toBe('RUSH_PROTOCOL_INVALID_CHILD_STREAM'); + }); +}); + +describe('Heft old raw-stream path', () => { + it('pipes and relays fallback stdout and stderr while retaining the reporter descriptor', async () => { + const plan: IChildDescriptorPlan = allocateChildDescriptor(); + const spawned: childProcess.ChildProcess = childProcess.spawn( + process.execPath, + ['-e', "process.stdout.write('old stdout'); process.stderr.write('old stderr');"], + { + env: { ...process.env, ...plan.env }, + stdio: plan.stdio as childProcess.StdioOptions + } + ); + const stdout: PassThrough = new PassThrough(); + const stderr: PassThrough = new PassThrough(); + let stdoutText: string = ''; + let stderrText: string = ''; + stdout.setEncoding('utf8').on('data', (chunk: string) => (stdoutText += chunk)); + stderr.setEncoding('utf8').on('data', (chunk: string) => (stderrText += chunk)); + relayHeftChildOutput({ stdout: spawned.stdout, stderr: spawned.stderr }, { stdout, stderr }); + + await new Promise((resolve, reject) => { + spawned.once('error', reject); + spawned.once('exit', (code: number | null) => { + if (code === 0) { + resolve(); + } else { + reject(new Error(`Spawned old-Heft fixture exited with code ${code}.`)); + } + }); + }); + + expect(stdoutText).toBe('old stdout'); + expect(stderrText).toBe('old stderr'); + expect(spawned.stdio[plan.fdNumber]).not.toBeNull(); + }); + + it('recovers diagnostics from an old Heft version through problem matchers', () => { + // Old Heft writes raw output to stdout; Rush captures it as external output. + let stdout: string = ''; + const child: HeftChildEmitter = new HeftChildEmitter({ + env: {}, + childSessionId: 'child-sess', + source: SOURCE, + producerVersion: '@rushstack/heft 0.60.0', + writeStdout: (text: string) => (stdout += text) + }); + expect(child.mode).toBe('raw-fallback'); + child.writeRaw('stdout', 'src/legacy.ts(3,7): error TS2551: old heft problem\n'); + + const capturedEvents: IReporterEventEnvelope[] = [ + { + type: 'externalOutput', + scope: { operationId: 'heft-op' }, + payload: { stream: 'stdout', text: stdout } + } as unknown as IReporterEventEnvelope + ]; + const diagnostics = runProblemMatchers(capturedEvents, [TSC_MATCHER]).diagnostics; + expect(diagnostics).toHaveLength(1); + expect(diagnostics[0].parameters?.code.value).toBe('TS2551'); + }); +}); diff --git a/research/feature-list.json b/research/feature-list.json index d9d2cc335f..9a272e1aa2 100644 --- a/research/feature-list.json +++ b/research/feature-list.json @@ -309,7 +309,7 @@ "Keep the raw-stream and problem-matcher path for older Heft versions", "Add old and new Heft descriptor path tests" ], - "passes": false + "passes": true }, { "category": "performance", diff --git a/research/progress.txt b/research/progress.txt index 8cc4acfa35..331d3821d0 100644 --- a/research/progress.txt +++ b/research/progress.txt @@ -171,7 +171,7 @@ - session/ScopedReporterFactory.ts (createScopedReporter -> IScopedReporter; emitMessage->activityChanged {kind:'message',severity,text}, required for warning/error; emitDiagnostic->diagnosticEmitted with computeEnvelopePrivacyFloor; emitExtension validates namespaced name, wraps {name,payload}) - session/ScopedLogger.ts (IScopedLogger writeLine/writeDebugLine/writeWarningLine/writeErrorLine; NO .terminal handle) - session/RushSessionReporting.ts (facade: createScopedReporter/createScopedLogger; getSink(); createExecutionContext()->IReporterExecutionContext {sink, reporter}) - - session/PluginApi.ts (RUSH_PLUGIN_API_VERSION '1.0.0', IRushPluginManifest.pluginApiVersion, isPluginApiVersionSupported major-equality, createPluginApiIncompatibleDiagnostic migration diagnostic) + - session/PluginApi.ts (IRushPluginManifest.rushVersionRange, isRushVersionSupported semver validation, createPluginApiIncompatibleDiagnostic migration diagnostic) - test/Session.test.ts Files (modified): diagnostics/RushDiagnosticCodeRegistry.ts (APPENDED RUSH_PLUGIN_API_INCOMPATIBLE code+templates), index.ts, api.md Notes: @@ -230,3 +230,182 @@ - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) - resolveExitStatus (success/warning-only/failure/cancel/signal precedence), signal codes, from-events (warning/error/op-failure/failed-result/category-independence/cancel/signal), separateJsonControls (5 cases) tests pass; all exports @beta Next: Feature 15/28 - Reporter selection + configuration controls with precedence + +[2026-07-14] Feature 15/28 COMPLETE: Reporter selection + configuration controls with precedence + Files (new): + - config/ReporterNames.ts (ReporterName default|ai|json|plaintext|file|legacy; ReporterLogLevel quiet|normal|verbose|debug; SUPPORTED_* + isSupported* guards) + - config/AgentDetection.ts (COPILOT_CLI_ENV_VAR, KNOWN_CI_ENV_VARS, isAgentVariableActive [inactive if undefined/''/0/false/no/off case-insens], detectAgent(env,configuredVars), isCiDetected) + - config/OutputControl.ts (IReporterOutputTarget, parseOutputControl '://?k=v') + - config/ReporterSelection.ts (resolveReporterSelection: primary precedence CLI --reporter > RUSH_REPORTER > agent > CI > TTY > non-TTY plaintext; log level independent CLI/aliases/RUSH_LOG_LEVEL default normal; contradiction throws; additional=['file'] unless primary file; commandJson preserved; unsupported reporter/level throws) + - test/ReporterSelection.test.ts + Files (modified): index.ts, api.md + Notes: + - Full resolver (Feature 8 was early subset). Orthogonal axes: reporter (format) vs log level (severity) per pnpm memory. + - Matrix: agent->ai+file, CI->plaintext+file, TTY->default+file, non-TTY->plaintext+file. + - --json (command) never aliases --reporter=json. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - primary precedence (explicit/env/agent/CI/TTY/non-TTY), additional file, unsupported-reporter throw, log-level aliases + contradiction + unsupported, command-json preserved, --output parse, agent-active semantics tests pass; all exports @beta + Next: Feature 16/28 - Independent per-reporter log levels + +[2026-07-14] Feature 16/28 COMPLETE: Independent per-reporter log levels + Files (new): + - config/LogLevelFilter.ts (LOG_LEVEL_RANK quiet0/normal1/verbose2/debug3; getEventMinimumLogLevel classifies events; shouldRenderAtLogLevel; filterEventsForLogLevel; FILE_REPORTER_DEFAULT_LOG_LEVEL='debug') + - test/LogLevelFilter.test.ts + Files (modified): index.ts, api.md + Classification (min level per ยง5.7): commandResult=quiet; diagnostic error=quiet, warning required=quiet else normal; lifecycle/operation/artifact=normal; activityChanged debug-message=debug else normal; externalOutput/externalProcess*=verbose; extension required=normal else debug. + Notes: + - Diagnostic severity separate from log level: severity sets min-level, reporter's configured level gates rendering. Test flips warning visibility by level, not severity. + - Monotonic: shown at level L -> shown at all higher levels. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - rank, classification, per-level rendering (quiet/normal/verbose/debug), monotonicity, severity-separation, filter, file-default tests pass; all exports @beta + Next: Feature 17/28 - Automatic reporter selection matrix by environment + +[2026-07-14] Feature 17/28 COMPLETE: Automatic reporter selection matrix by environment + Files (new): + - config/AutomaticReporterMatrix.ts (PlaintextVariant detailed|concise; IReporterPlanEntry; IAutomaticReporterPlan; isMachineReporter(json|ai); planAutomaticReporters(selection)->plan; describeReporterPlan for detailed log) + - test/AutomaticReporterMatrix.test.ts + Files (modified): index.ts, api.md + Matrix: agent->ai+file (machine stdout, human progress stderr); CI->plaintext[detailed]+file; TTY->default+file; non-TTY->plaintext[concise]+file. Emergency always stderr. + Notes: + - Composes Feature 15 resolveReporterSelection; plaintext variant from reason (CI detected->detailed else concise). + - Machine reporters (ai/json) own stdout exclusively -> stdoutOwner 'machine', humanProgressDestination 'stderr'. Applies to explicit --reporter=json too. + - describeReporterPlan records selection reason + reporters for the detailed (file/debug) log. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - agent/CI/TTY/non-TTY matrix, json machine stdout, emergency stderr, isMachineReporter, describeReporterPlan tests pass; all exports @beta + Next: Feature 18/28 - Default interactive reporter (three-row live region) + +[2026-07-14] Feature 18/28 COMPLETE: Default interactive reporter (three-row live region) + Files (new): + - reporters/InteractiveRendering.ts (SPINNER_FRAMES, MIN_REFRESH_INTERVAL_MS=100; resolveColorEnabled NO_COLOR>FORCE_COLOR>isTTY; createColorizer; truncateToWidth; renderActiveProjectsRow +N more; renderLiveRegion 3 rows [color applied AFTER truncation]; shouldRefresh) + - reporters/DefaultInteractiveReporter.ts (IReporter; injected IInteractiveTerminal + nowMs; state from events; throttled paint <=10Hz; cursor hide/show ANSI; success<=3 lines; failure bounded diag block <=10 + Log path; watchCycleCompleted summary; non-TTY skips live region but writes final) + - test/DefaultInteractiveReporter.test.ts + Files (modified): index.ts, api.md + Notes: + - Color applied after truncateToWidth so ANSI never affects width/split mid-code. + - Resize-aware: reads terminal.columns each paint. + - Testable via FakeTerminal capturing writes + injected clock. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - color resolution, colorizer, truncate, +N more, 3-row render, throttle; reporter hide/paint/throttle, success line + cursor restore, failure diag+log, watch summary, non-TTY tests pass; all exports @beta + Next: Feature 19/28 - Plaintext and non-TTY reporter + +[2026-07-14] Feature 19/28 COMPLETE: Plaintext and non-TTY reporter + Files (new): + - reporters/PlaintextReporter.ts (IReporter; append-only via injected write; no cursor codes; color off default; variant concise|detailed; emits Starting line, terminal status lines, [severity] code diagnostics, final result; emitHeartbeatIfDue 30s; detailed groups externalOutput under ==[ project (phase) ]== StreamCollator-like) + - test/PlaintextReporter.test.ts + __snapshots__/PlaintextReporter.test.ts.snap (concise + detailed) + Files (modified): index.ts, api.md + Notes: + - No-ANSI assertion (append-only, no cursor, color off). Heartbeat interval-based via injected clock (report resets lastOutput). + - Detailed buffers external output per operationId, flushes under header on terminal status. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - no-ANSI, concise snapshot, detailed grouping snapshot, heartbeat timing tests pass; all exports @beta + Next: Feature 20/28 - JSON and AI reporters + +[2026-07-14] Feature 20/28 COMPLETE: JSON and AI reporters + Files (new): + - reporters/JsonReporter.ts (IReporter; encodeNdjsonRecord per event to exclusive stdout; oversized -> rush.reporter.recordTooLarge marker so stream stays valid NDJSON) + - reporters/AiReporter.ts (IReporter; ai.status on commandStarted; ai.final on commandResult/close; IAiFinalRecord result/exitCode/scope{commandName,failedProjects}/errorCodes/diagnosticCategoryCounts/diagnostics(remediation)/counts/log{path,format,complete}; caps 64KiB + 20 diagnostics; warnings by count when failures else warning-only details; excludes raw external output/stacks; byte-cap trims diagnostics->errorCodes->failedProjects) + - test/JsonAiReporter.test.ts + Files (modified): index.ts, api.md + Notes: + - Fixed byte-cap: trim errorCodes/failedProjects too (base record with many long codes exceeded tiny maxBytes otherwise). Real cap 64KiB. + - Log path in AI final.log.path (local) but telemetry aggregate cross-check asserts path NOT present. + - stdout purity: every emitted line parses as JSON. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - JSON stream + oversized marker; AI status+final, log/scope/codes/remediation, 20-cap, 64KiB byte-cap trim, warnings-by-count vs warning-only, raw-excluded, stdout purity, telemetry path cross-check tests pass; all exports @beta + Next: Feature 21/28 - Full-detail file reporter (retention + fallback) + +[2026-07-14] Feature 21/28 COMPLETE: Full-detail file reporter (retention + fallback) + Files (new): + - reporters/FileReporter.ts (IReporter; RUSH_LOGS_DIR_NAME 'rush-logs', LATEST_LOG_NAME 'latest.log'; buffers lines, writes NDJSON at debug to /rush-logs/--.log; owner-only 0o600; redacts secret fields ->[secret], keeps local-sensitive; latest.log symlink/copy for success+failure; retention delete >14d + cap 20; OS-temp fallback; both-fail nonfatal emergencyWarn + getArtifact().available=false) + - test/FileReporter.test.ts (real temp dirs) + Files (modified): index.ts, api.md + Notes: + - report() buffers; _writeAsync (flush/close) resolves target lazily: try repoDir then osTempDir; appendFile new lines. + - Timestamp filename: toISOString().replace(/[:.]/g,'-'). rush purge removes the rush-logs dir (path convention). + - Test mode check uses %0o1000 (no-bitwise lint), guarded on non-win32. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - write+perms+latest, latest-on-failure, secret redaction, retention (old delete + cap 20), OS-temp fallback, both-fail nonfatal tests pass; all exports @beta + Next: Feature 22/28 - Legacy reporter (selectable + emergency fallback) + +[2026-07-14] Feature 22/28 COMPLETE: Legacy reporter (selectable + emergency fallback) + Files (new): + - reporters/LegacyReporter.ts (IReporter name 'legacy'; reproduces Starting line, "Executing a maximum of N...", 79-wide ==[ project (phase) ]==...[ n of total ]== headers, grouped externalOutput, SUCCESS/FAILURE summary with per-project + total durations; RUSH_REPORTER_ENV_VAR; isLegacyEmergencyFallbackRequested case-insensitive) + - test/LegacyReporter.test.ts + __snapshots__/LegacyReporter.test.ts.snap (success + failure) + Files (modified): index.ts, api.md + Notes: + - Legacy snapshots reproduce Feature 7 frozen legacy format EXACTLY (same 79-wide header padding, SUCCESS/FAILURE blocks, ==> ERROR line). Durations normalized X.XX. + - Emergency fallback: feature 15 resolveReporterSelection selects 'legacy' via RUSH_REPORTER=legacy (validated); isLegacyEmergencyFallbackRequested helper. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - success/failure snapshots, frozen-marker validation, emergency-fallback detection + selection tests pass; all exports @beta + Next: Feature 23/28 - Replace StreamCollator in primary path with raw semantic events + +[2026-07-14] Feature 23/28 COMPLETE: Replace StreamCollator with raw semantic events + Files (new): + - scheduler/OperationStreamEmitter.ts (registerOperation, changeStatus, writeOutput [uncollated 64KiB chunked externalOutput, local-sensitive], completeCommand; emits raw events in call order = NOT collated) + - scheduler/OperationOutputGrouping.ts (iterateExternalOutput -> ordered IExternalOutputChunk[] for matchers; regroupOperationOutput -> Map per-op parity) + - test/OperationStreamEmitter.test.ts + Files (modified): index.ts, api.md + Notes: + - Uncollated: interleaved writeOutput emits in call order. Parity proven: detailed PlaintextReporter regroups interleaved stream (A1,A2 before B1); regroupOperationOutput reconstructs per-op. + - Concise DefaultInteractiveReporter never echoes raw project output (derives activity without buffering). + - Same uncollated stream feeds problem matchers via iterateExternalOutput (Feature 24 owns matchers). + - StreamCollator replacement = OperationStreamEmitter (source of truth). Actual rush-lib @rushstack/stream-collator removal from primary path is a rollout step (not modifying rush-lib live); package removal deferred per spec. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - emitter events/scope, chunking, uncollated order, regroup parity, detailed-plaintext regroup, concise no-buffer tests pass; all exports @beta + Next: Feature 24/28 - External output preservation + problem matchers + +[2026-07-14] Feature 24/28 COMPLETE: External output preservation + problem matchers + Files (new): + - matchers/AnsiNormalization.ts (normalizeAnsi strips CSI codes) + - matchers/ProblemMatcher.ts (IProblemMatcher {name,tool,severity,pattern,enabledByDefault,matchesVersion?,extract}, IProblemMatch) + - matchers/ProblemMatcherRegistry.ts (register, getMatchers(tool,{version,includeDisabled}) - tool+version scope + default-enablement gate) + - matchers/ProblemMatcherRunner.ts (runProblemMatchers: reassembles lines per op across chunks, ANSI-normalized copy, linked RUSH_EXTERNAL_TOOL_PROBLEM diagnostics w/ source+relatedArtifactIds, dedup cap, unmatched preserved; raw events NEVER modified) + - test/ProblemMatchers.test.ts + Files (modified): diagnostics/RushDiagnosticCodeRegistry.ts (APPENDED RUSH_EXTERNAL_TOOL_PROBLEM code+template), index.ts, api.md + Notes: + - Consumes Feature 23 uncollated externalOutput stream (iterateExternalOutput). Split-chunk reassembly per operationId. Evidence preserved (events unchanged, asserted via JSON equal). + - Duplicate cap (maxDuplicates default 3). Default-enablement gate via enabledByDefault + corpus test. + - Old Heft routed via version-scoped matcher (registry.getMatchers('heft',{version:'0.9.0'})). Reporter self-contained (no @rushstack/problem-matcher dep). + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - normalizeAnsi, registry scope/gate, recover+link+evidence-preserve, split-chunk+ANSI, dedup cap, corpus tests pass; all exports @beta + Next: Feature 25/28 - Prohibit new AlreadyReportedError + bridge legacy sentinels + +[2026-07-14] Feature 25/28 COMPLETE: Prohibit new AlreadyReportedError + bridge legacy sentinels + Files (new): + - compat/LegacyErrorBridge.ts (ALREADY_REPORTED_ERROR_NAME; AlreadyReportedError class @deprecated=prohibition; isAlreadyReportedSentinel; LegacyErrorBridge {recordEmittedDiagnostic, ingest(diagnosticEmitted->id), correlate/getCorrelatedDiagnosticId via symbol, shouldSuppressRendering}; LEGACY_ERROR_BRIDGE_REMOVAL_CRITERIA 3 items) + - test/LegacyErrorBridge.test.ts + Files (modified): index.ts, api.md + Notes: + - Prohibition encoded as @deprecated on AlreadyReportedError (api.md shows "// @beta @deprecated"); replacement = structured diagnostic + RushError. + - shouldSuppressRendering: sentinel->true; RushError whose diagnosticId already emitted->true; correlated error with emitted id->true; else false (catch boundaries render only unrepresented failures). + - Removal plan constant documents criteria (zero first-party usages, plugin migration guidance published, ecosystem notice/time). + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - sentinel detection, removal criteria, suppress sentinel/RushError-emitted/correlated, record direct+ingest tests pass; all exports @beta + Next: Feature 26/28 - Heft integration via negotiated child descriptors + raw-stream fallback + +[2026-07-14] Feature 26/28 COMPLETE: Heft integration via negotiated child descriptors + raw-stream fallback + Files (new): + - heft/HeftDescriptor.ts (RUSH_REPORTER_CHILD_FD_ENV_VAR='_RUSH_REPORTER_CHILD_FD'; allocateChildDescriptor -> {fdNumber 3, env var, stdio [inherit,inherit,inherit,pipe]}; readChildDescriptorFd) + - heft/HeftChildEmitter.ts (child side; mode structured if fd+writeDescriptor else raw-fallback; sendHello/emitEvent NDJSON to descriptor stamping childSessionId; writeRaw stdout/stderr fallback) + - heft/HeftDescriptorHost.ts (parent side; processChildRecords/Ndjson: negotiate hello (feature 5), correlate each child event with parentSessionId+parentOperationId, forwardEnvelope; reject->RUSH_PROTOCOL_UPDATE_REQUIRED diagnostic) + - test/HeftIntegration.test.ts + Files (modified): index.ts, api.md + Notes: + - New path: child structured NDJSON over inherited fd -> host negotiate+correlate -> manager.ingestForeignEnvelope (sourceSequence preserved). stdout/stderr stay normal streams. + - Old path: no fd -> raw-fallback writeRaw -> Rush captures as externalOutput -> runProblemMatchers (feature 24) recovers diagnostics. + - Streams injected (writeDescriptor/writeStdout) for testability instead of real spawned FDs. + Verify: + - rush test --only @rushstack/reporter: clean SUCCESS (build + jest) + - descriptor alloc/read, structured emit, raw fallback, host correlate+forward, protocol reject, old-heft raw+matcher tests pass; all exports @beta + Next: Feature 27/28 - Reporter performance and capacity budgets