From 96a8f7fc5b007adc8562340939d20ffb14e19674 Mon Sep 17 00:00:00 2001 From: TheLarkInn Date: Fri, 28 Aug 2026 06:41:03 +0000 Subject: [PATCH 1/3] Add feature-flagged operation event adapter Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d6318e80-5da9-4858-a147-817e8692f10e --- apps/rush/src/RushFrontend.ts | 3 +- apps/rush/src/RushReporterHost.ts | 46 +++- apps/rush/src/test/RushFrontend.test.ts | 8 +- apps/rush/src/test/RushReporterHost.test.ts | 66 +++++- ...5a-operation-adapter_2026-08-28-06-35.json | 11 + ...5a-operation-adapter_2026-08-28-06-35.json | 11 + common/reviews/api/rush-lib.api.md | 3 + common/reviews/api/rush-reporter.api.md | 22 +- .../reporter/src/config/LogLevelFilter.ts | 2 + .../reporter/src/events/ReporterEventType.ts | 8 +- libraries/reporter/src/index.ts | 2 + .../reporter/src/lifecycle/LifecycleEvents.ts | 44 ++++ .../reporter/src/protocol/ReporterProtocol.ts | 2 +- .../src/scheduler/OperationStreamEmitter.ts | 68 +++++- .../src/test/IReporterEventEnvelope.test.ts | 4 +- .../reporter/src/test/LogLevelFilter.test.ts | 2 + .../src/test/OperationStreamEmitter.test.ts | 40 +++- libraries/reporter/src/test/Protocol.test.ts | 6 +- libraries/reporter/src/test/Telemetry.test.ts | 2 +- .../test/__snapshots__/Goldens.test.ts.snap | 2 +- .../logic/operations/OperationEventSink.ts | 10 +- .../operations/OperationExecutionRecord.ts | 16 +- .../src/logic/operations/OperationGraph.ts | 24 +- .../operations/ReporterOperationEventSink.ts | 168 ++++++++++++-- .../test/OperationGraphEventSink.test.ts | 217 +++++++++++++++++- .../src/pluginFramework/RushSession.ts | 43 ++++ 26 files changed, 756 insertions(+), 74 deletions(-) create mode 100644 common/changes/@microsoft/rush/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json create mode 100644 common/changes/@rushstack/rush-reporter/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json diff --git a/apps/rush/src/RushFrontend.ts b/apps/rush/src/RushFrontend.ts index 044a060d6b..00caadebf4 100644 --- a/apps/rush/src/RushFrontend.ts +++ b/apps/rush/src/RushFrontend.ts @@ -161,7 +161,8 @@ export async function launchRushFrontendAsync(options: IRushFrontendOptions): Pr ...launchOptions, reporter: { eventSink: reporterHost.sink, - sessionId + sessionId, + operationStreamEnabled: reporterHost.selection.enabled }, reporterCloseAsync }; diff --git a/apps/rush/src/RushReporterHost.ts b/apps/rush/src/RushReporterHost.ts index dfa9b84a7e..82dc2d4f90 100644 --- a/apps/rush/src/RushReporterHost.ts +++ b/apps/rush/src/RushReporterHost.ts @@ -23,6 +23,7 @@ import { type IReporterEventEnvelope, type IReporterEventSink, type IReporterOutputTarget, + type ReporterEventType, type ReporterLogLevel, type ReporterName } from '@rushstack/rush-reporter'; @@ -70,6 +71,13 @@ export interface IInitializedRushReporterHost { const REPORTER_VALUE_FLAGS: ReadonlySet = new Set(['--reporter', '--output', '--log-level']); const ALL_REPORTER_VALUE_FLAGS: readonly string[] = ['--reporter', '--output', '--log-level']; const REPORTER_SELECTION_FLAG: readonly string[] = ['--reporter']; +const DEFERRED_OPERATION_EVENT_TYPES: ReadonlySet = new Set([ + 'operationRegistered', + 'operationStatusChanged', + 'operationStreamClosed', + 'operationCompleted', + 'externalOutput' +]); interface IParsedReporterControls { readonly reporters: readonly string[]; @@ -111,6 +119,38 @@ class LogLevelReporter implements IReporter { } } +/** + * Keeps operation presentation on the legacy collator until R5B transfers terminal ownership. + */ +class DeferredOperationPresentationReporter implements IReporter { + public readonly name: string; + + private readonly _reporter: IReporter; + + public constructor(reporter: IReporter) { + this._reporter = reporter; + this.name = reporter.name; + } + + public initializeAsync(context: IReporterContext): Promise { + return this._reporter.initializeAsync(context); + } + + public report(event: IReporterEventEnvelope): void { + if (!DEFERRED_OPERATION_EVENT_TYPES.has(event.type)) { + this._reporter.report(event); + } + } + + public flushAsync(): Promise { + return this._reporter.flushAsync(); + } + + public closeAsync(): Promise { + return this._reporter.closeAsync(); + } +} + class ExplicitOutputReporter implements IReporter { public readonly name: string; @@ -610,7 +650,11 @@ export async function initializeRushReporterHostAsync( if (selection.enabled) { const primaryReporter: IReporter | undefined = createPrimaryReporter(selection, stdout, env); if (primaryReporter) { - host.manager.addReporter(new LogLevelReporter(primaryReporter, selection.logLevel), { + const presentationReporter: IReporter = + selection.reporter === 'file' + ? primaryReporter + : new DeferredOperationPresentationReporter(primaryReporter); + host.manager.addReporter(new LogLevelReporter(presentationReporter, selection.logLevel), { destination: selection.reporter === 'file' ? 'file:auto' : 'stdout' }); } diff --git a/apps/rush/src/test/RushFrontend.test.ts b/apps/rush/src/test/RushFrontend.test.ts index 233b3d7e2d..4a59142d85 100644 --- a/apps/rush/src/test/RushFrontend.test.ts +++ b/apps/rush/src/test/RushFrontend.test.ts @@ -178,7 +178,7 @@ function emitCommandStarted(sink: IReporterEventSink): void { } describe(launchRushFrontendAsync.name, () => { - it('creates the authoritative host before invoking the bundled rush-lib and passes only its sink', async () => { + it('creates the authoritative host before invoking the bundled rush-lib and passes only its channel', async () => { const order: string[] = []; let receivedOptions: IRushFrontendLaunchOptions | undefined; const processLifecycle: ITestProcessLifecycle = createTestProcessLifecycle(); @@ -207,7 +207,8 @@ describe(launchRushFrontendAsync.name, () => { expect(process.argv).toEqual(['node', 'rush', 'build', '--json']); expect(receivedOptions?.reporter).toEqual({ eventSink: expect.objectContaining({ emit: expect.any(Function) }), - sessionId: expect.any(String) + sessionId: expect.any(String), + operationStreamEnabled: false }); expect(receivedOptions).not.toHaveProperty('selection'); expect(receivedOptions).not.toHaveProperty('host'); @@ -249,7 +250,8 @@ describe(launchRushFrontendAsync.name, () => { expect(createSessionId).toHaveBeenCalledTimes(1); expect(receivedOptions?.reporter).toEqual({ eventSink: initialized.sink, - sessionId: 'session-from-frontend' + sessionId: 'session-from-frontend', + operationStreamEnabled: false }); await initialized.closeAsync(); expect(order).toEqual(['host', 'close']); diff --git a/apps/rush/src/test/RushReporterHost.test.ts b/apps/rush/src/test/RushReporterHost.test.ts index fc3e630773..7846cb673c 100644 --- a/apps/rush/src/test/RushReporterHost.test.ts +++ b/apps/rush/src/test/RushReporterHost.test.ts @@ -44,6 +44,45 @@ function emitCommandStarted(sink: IReporterEventSink): void { }); } +function emitOperationEvents(sink: IReporterEventSink): void { + const base = { + protocolVersion: { major: 1, minor: 1 }, + sessionId: 'session', + source: { packageName: '@microsoft/rush-lib', packageVersion: '5.178.1' }, + scope: { commandName: 'build', operationId: 'project#phase' } + } as const; + sink.emit({ + ...base, + privacy: 'public', + type: 'operationRegistered', + payload: { operationId: 'project#phase', projectName: 'project', phaseName: 'phase' } + }); + sink.emit({ + ...base, + privacy: 'public', + type: 'operationStatusChanged', + payload: { operationId: 'project#phase', previousStatus: 'queued', status: 'executing' } + }); + sink.emit({ + ...base, + privacy: 'local-sensitive', + type: 'externalOutput', + payload: { stream: 'stdout', text: 'raw operation output\n' } + }); + sink.emit({ + ...base, + privacy: 'public', + type: 'operationStreamClosed', + payload: { operationId: 'project#phase' } + }); + sink.emit({ + ...base, + privacy: 'public', + type: 'operationCompleted', + payload: { operationId: 'project#phase', status: 'success' } + }); +} + describe(resolveRushReporterSelection.name, () => { it('preserves the legacy path without an explicit opt-in in TTY, non-TTY, CI, and agent environments', () => { for (const testCase of [ @@ -436,7 +475,12 @@ describe(initializeRushReporterHostAsync.name, () => { let stdoutText: string = ''; try { const initialized = await initializeRushReporterHostAsync({ - argv: ['build', '--reporter=json', `--output=json://${outputPath}`], + argv: [ + 'build', + '--reporter=json', + '--log-level=debug', + `--output=json://${outputPath}?logLevel=debug` + ], env: {}, stdout: { isTTY: false, @@ -448,12 +492,28 @@ describe(initializeRushReporterHostAsync.name, () => { }); emitCommandStarted(initialized.sink); + emitOperationEvents(initialized.sink); const firstClose: Promise = initialized.closeAsync(); expect(initialized.closeAsync()).toBe(firstClose); await firstClose; - expect(JSON.parse(stdoutText).type).toBe('commandStarted'); - expect(JSON.parse(await fs.promises.readFile(outputPath, 'utf8')).type).toBe('commandStarted'); + const stdoutEvents: Record[] = stdoutText + .trim() + .split('\n') + .map((line: string) => JSON.parse(line) as Record); + const fileEvents: Record[] = (await fs.promises.readFile(outputPath, 'utf8')) + .trim() + .split('\n') + .map((line: string) => JSON.parse(line) as Record); + expect(stdoutEvents.map(({ type }) => type)).toEqual(['commandStarted']); + expect(fileEvents.map(({ type }) => type)).toEqual([ + 'commandStarted', + 'operationRegistered', + 'operationStatusChanged', + 'externalOutput', + 'operationStreamClosed', + 'operationCompleted' + ]); } finally { await fs.promises.rm(directory, { recursive: true, force: true }); } diff --git a/common/changes/@microsoft/rush/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json b/common/changes/@microsoft/rush/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json new file mode 100644 index 0000000000..2e0fd43f62 --- /dev/null +++ b/common/changes/@microsoft/rush/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@microsoft/rush", + "comment": "Emit feature-flagged phase-aware operation registration, status, raw output, stream-close, and completion events while preserving the legacy StreamCollator output path.", + "type": "patch" + } + ], + "packageName": "@microsoft/rush", + "email": "TheLarkInn@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-reporter/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json b/common/changes/@rushstack/rush-reporter/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json new file mode 100644 index 0000000000..fe11e3b0e3 --- /dev/null +++ b/common/changes/@rushstack/rush-reporter/copilot-reporter-r5a-operation-adapter_2026-08-28-06-35.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-reporter", + "comment": "Extend OperationStreamEmitter with silent registration metadata, previous status, stream-close, and operation-completion events.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-reporter", + "email": "TheLarkInn@users.noreply.github.com" +} diff --git a/common/reviews/api/rush-lib.api.md b/common/reviews/api/rush-lib.api.md index 66da397e0c..4b5f481e45 100644 --- a/common/reviews/api/rush-lib.api.md +++ b/common/reviews/api/rush-lib.api.md @@ -682,6 +682,7 @@ export interface IOperationGraphContext extends ICreateOperationsContext { export interface _IOperationGraphEventSink { onActivity?(text: string, options?: _IOperationActivityOptions): void; onOperationChunk?(operationId: string, chunk: ITerminalChunk): void; + onOperationCompleted?(result: IOperationExecutionResult): void; onOperationHeader?(operationId: string, completedOperations: number, totalOperations: number): void; onOperationRegistered?(operationId: string, silent: boolean): void; onOperationStatusChanged?(result: IOperationExecutionResult, previousStatus: OperationStatus): void; @@ -1014,6 +1015,8 @@ export interface IRushSessionOptions { // @beta export interface IRushSessionReporterOptions { readonly eventSink: IReporterEventSink; + // @internal + readonly operationStreamEnabled?: boolean; readonly sessionId: string; } diff --git a/common/reviews/api/rush-reporter.api.md b/common/reviews/api/rush-reporter.api.md index ecf09047db..029dea99d8 100644 --- a/common/reviews/api/rush-reporter.api.md +++ b/common/reviews/api/rush-reporter.api.md @@ -604,20 +604,34 @@ export interface IOldEngineOutputAdapterOptions { readonly source: IReporterEventSource; } +// @beta +export interface IOperationCompletedPayload { + readonly durationMs?: number; + readonly operationId: string; + readonly status: OperationStatus; +} + // @beta export interface IOperationRegisteredPayload { readonly operationId: string; readonly phaseName?: string; readonly projectName?: string; + readonly silent?: boolean; } // @beta export interface IOperationStatusChangedPayload { readonly durationMs?: number; readonly operationId: string; + readonly previousStatus?: OperationStatus; readonly status: OperationStatus; } +// @beta +export interface IOperationStreamClosedPayload { + readonly operationId: string; +} + // @beta export interface IOperationStreamEmitterOptions { readonly maxChunkBytes?: number; @@ -1251,11 +1265,13 @@ export type OperationStatus = 'ready' | 'waiting' | 'queued' | 'executing' | 'su // @beta export class OperationStreamEmitter { constructor(options: IOperationStreamEmitterOptions); - changeStatus(operationId: string, status: OperationStatus, durationMs?: number): string; + changeStatus(operationId: string, status: OperationStatus, durationMs?: number, previousStatus?: OperationStatus): string; + closeOperationStream(operationId: string): string; completeCommand(commandName: string, succeeded: boolean, exitCode: number, operationCounts?: { readonly [status: string]: number; }): string; - registerOperation(operationId: string, projectName?: string, phaseName?: string): string; + completeOperation(operationId: string, status: OperationStatus, durationMs?: number): string; + registerOperation(operationId: string, projectName?: string, phaseName?: string, silent?: boolean): string; writeOutput(operationId: string, stream: 'stdout' | 'stderr', text: string): string[]; } @@ -1328,7 +1344,7 @@ export function renderActiveProjectsRow(projects: readonly string[], width: numb export function renderLiveRegion(state: ILiveRegionState, options: IRenderLiveRegionOptions): string[]; // @beta -export const REPORTER_EVENT_TYPES: readonly ["sessionStarted", "sessionCompleted", "commandStarted", "commandCompleted", "operationRegistered", "operationStatusChanged", "activityChanged", "watchCycleCompleted", "diagnosticEmitted", "messageEmitted", "externalProcessStarted", "externalOutput", "externalProcessCompleted", "artifactAvailable", "commandResult", "extension"]; +export const REPORTER_EVENT_TYPES: readonly ["sessionStarted", "sessionCompleted", "commandStarted", "commandCompleted", "operationRegistered", "operationStatusChanged", "activityChanged", "watchCycleCompleted", "diagnosticEmitted", "messageEmitted", "externalProcessStarted", "externalOutput", "externalProcessCompleted", "artifactAvailable", "commandResult", "extension", "operationStreamClosed", "operationCompleted"]; // @beta export const REPORTER_KNOWN_CAPABILITIES: readonly []; diff --git a/libraries/reporter/src/config/LogLevelFilter.ts b/libraries/reporter/src/config/LogLevelFilter.ts index f644a0edda..6ea09cdf5f 100644 --- a/libraries/reporter/src/config/LogLevelFilter.ts +++ b/libraries/reporter/src/config/LogLevelFilter.ts @@ -59,6 +59,7 @@ export function getEventMinimumLogLevel(event: IReporterEventEnvelope): case 'sessionStarted': case 'commandStarted': case 'operationStatusChanged': + case 'operationCompleted': case 'watchCycleCompleted': case 'artifactAvailable': return 'normal'; @@ -79,6 +80,7 @@ export function getEventMinimumLogLevel(event: IReporterEventEnvelope): case 'externalProcessCompleted': return 'verbose'; case 'externalOutput': + case 'operationStreamClosed': return 'debug'; case 'extension': return 'normal'; diff --git a/libraries/reporter/src/events/ReporterEventType.ts b/libraries/reporter/src/events/ReporterEventType.ts index f0c99004fc..7e7010b953 100644 --- a/libraries/reporter/src/events/ReporterEventType.ts +++ b/libraries/reporter/src/events/ReporterEventType.ts @@ -30,6 +30,8 @@ * | `artifactAvailable` | yes | `normal` | * | `commandResult` | yes | `quiet` | * | `extension` | yes | `normal` | + * | `operationStreamClosed` | yes | `debug` | + * | `operationCompleted` | yes | `normal` | * * Coalescing a replaceable `activityChanged` event under queue pressure leaves * gaps in the delivered `sequence` values; gaps are legal and are not a @@ -57,7 +59,9 @@ export const REPORTER_EVENT_TYPES = [ 'externalProcessCompleted', 'artifactAvailable', 'commandResult', - 'extension' + 'extension', + 'operationStreamClosed', + 'operationCompleted' ] as const; /** @@ -83,4 +87,4 @@ export type ReporterEventType = (typeof REPORTER_EVENT_TYPES)[number]; */ export function isReporterEventRequired(type: ReporterEventType): boolean { return type !== 'activityChanged'; -} \ No newline at end of file +} diff --git a/libraries/reporter/src/index.ts b/libraries/reporter/src/index.ts index fcff5af94f..2167262849 100644 --- a/libraries/reporter/src/index.ts +++ b/libraries/reporter/src/index.ts @@ -177,6 +177,8 @@ export type { ICommandCompletedPayload, IOperationRegisteredPayload, IOperationStatusChangedPayload, + IOperationStreamClosedPayload, + IOperationCompletedPayload, ICommandResultPayload, IWatchCycleCompletedPayload } from './lifecycle/LifecycleEvents'; diff --git a/libraries/reporter/src/lifecycle/LifecycleEvents.ts b/libraries/reporter/src/lifecycle/LifecycleEvents.ts index e142e9ef32..81b2775271 100644 --- a/libraries/reporter/src/lifecycle/LifecycleEvents.ts +++ b/libraries/reporter/src/lifecycle/LifecycleEvents.ts @@ -113,6 +113,11 @@ export interface IOperationRegisteredPayload { * The phase the operation belongs to. */ readonly phaseName?: string; + + /** + * Whether the operation is architectural and normally omitted from visible summaries. + */ + readonly silent?: boolean; } /** @@ -131,12 +136,51 @@ export interface IOperationStatusChangedPayload { */ readonly status: OperationStatus; + /** + * The status immediately preceding this transition. + */ + readonly previousStatus?: OperationStatus; + /** * The operation duration in milliseconds when known. */ readonly durationMs?: number; } +/** + * The payload of an `operationStreamClosed` event. + * + * @beta + */ +export interface IOperationStreamClosedPayload { + /** + * The operation whose output stream has closed. + */ + readonly operationId: string; +} + +/** + * The payload of an `operationCompleted` event. + * + * @beta + */ +export interface IOperationCompletedPayload { + /** + * The completed operation. + */ + readonly operationId: string; + + /** + * The terminal operation status. + */ + readonly status: OperationStatus; + + /** + * The final operation duration in milliseconds when known. + */ + readonly durationMs?: number; +} + /** * The payload of a `commandResult` event. * diff --git a/libraries/reporter/src/protocol/ReporterProtocol.ts b/libraries/reporter/src/protocol/ReporterProtocol.ts index 40119adea8..15dc630563 100644 --- a/libraries/reporter/src/protocol/ReporterProtocol.ts +++ b/libraries/reporter/src/protocol/ReporterProtocol.ts @@ -15,7 +15,7 @@ import type { IReporterProtocolVersion } from '../events/ReporterProtocolVersion */ export const REPORTER_PROTOCOL_VERSION: IReporterProtocolVersion = { major: 1, - minor: 0 + minor: 1 }; /** diff --git a/libraries/reporter/src/scheduler/OperationStreamEmitter.ts b/libraries/reporter/src/scheduler/OperationStreamEmitter.ts index 259c935a1e..241217781e 100644 --- a/libraries/reporter/src/scheduler/OperationStreamEmitter.ts +++ b/libraries/reporter/src/scheduler/OperationStreamEmitter.ts @@ -49,11 +49,11 @@ export interface IOperationStreamEmitterOptions { * * @remarks * The operation scheduler uses this to publish operation registration, status - * transitions, raw output chunks, and the aggregate command result. Output - * chunks are emitted immediately in call order and are never collated, so the - * concise reporter can derive activity without buffering, the detailed and file - * reporters can own grouping, and problem matchers can consume the same - * uncollated source stream. + * transitions, raw output chunks, stream close, operation completion, and the + * aggregate command result. Output chunks are emitted immediately in call order + * and are never collated, so the concise reporter can derive activity without + * buffering, the detailed and file reporters can own grouping, and problem + * matchers can consume the same uncollated source stream. * * @beta */ @@ -71,8 +71,7 @@ export class OperationStreamEmitter { this._source = options.source; this._scope = options.scope; this._protocolVersion = options.protocolVersion ?? REPORTER_PROTOCOL_VERSION; - const maxChunkBytes: number = - options.maxChunkBytes ?? REPORTER_PROTOCOL_LIMITS.externalOutputChunkBytes; + const maxChunkBytes: number = options.maxChunkBytes ?? REPORTER_PROTOCOL_LIMITS.externalOutputChunkBytes; if ( !Number.isInteger(maxChunkBytes) || maxChunkBytes < 4 || @@ -88,10 +87,20 @@ export class OperationStreamEmitter { /** * Emits an operation registration event. */ - public registerOperation(operationId: string, projectName?: string, phaseName?: string): string { + public registerOperation( + operationId: string, + projectName?: string, + phaseName?: string, + silent?: boolean + ): string { return this._emit( 'operationRegistered', - { operationId, projectName, phaseName }, + { + operationId, + projectName, + phaseName, + ...(silent === undefined ? {} : { silent }) + }, { operationId, projectName, phaseName }, 'public' ); @@ -100,10 +109,20 @@ export class OperationStreamEmitter { /** * Emits an operation status transition. */ - public changeStatus(operationId: string, status: OperationStatus, durationMs?: number): string { + public changeStatus( + operationId: string, + status: OperationStatus, + durationMs?: number, + previousStatus?: OperationStatus + ): string { return this._emit( 'operationStatusChanged', - { operationId, status, durationMs }, + { + operationId, + status, + ...(previousStatus === undefined ? {} : { previousStatus }), + ...(durationMs === undefined ? {} : { durationMs }) + }, { operationId }, 'public' ); @@ -146,6 +165,25 @@ export class OperationStreamEmitter { return eventIds; } + /** + * Emits the authoritative signal that no more output will be emitted for an operation. + */ + public closeOperationStream(operationId: string): string { + return this._emit('operationStreamClosed', { operationId }, { operationId }, 'public'); + } + + /** + * Emits the final outcome of an operation. + */ + public completeOperation(operationId: string, status: OperationStatus, durationMs?: number): string { + return this._emit( + 'operationCompleted', + { operationId, status, ...(durationMs === undefined ? {} : { durationMs }) }, + { operationId }, + 'public' + ); + } + /** * Emits the aggregate command result. */ @@ -164,7 +202,13 @@ export class OperationStreamEmitter { } private _emit( - type: 'operationRegistered' | 'operationStatusChanged' | 'externalOutput' | 'commandResult', + type: + | 'operationRegistered' + | 'operationStatusChanged' + | 'operationStreamClosed' + | 'operationCompleted' + | 'externalOutput' + | 'commandResult', payload: unknown, scopeOverride: IReporterEventScope, privacy: 'public' | 'local-sensitive' | 'secret' diff --git a/libraries/reporter/src/test/IReporterEventEnvelope.test.ts b/libraries/reporter/src/test/IReporterEventEnvelope.test.ts index 949f38df08..0786651183 100644 --- a/libraries/reporter/src/test/IReporterEventEnvelope.test.ts +++ b/libraries/reporter/src/test/IReporterEventEnvelope.test.ts @@ -27,7 +27,9 @@ describe('ReporterEventType', () => { 'externalProcessCompleted', 'artifactAvailable', 'commandResult', - 'extension' + 'extension', + 'operationStreamClosed', + 'operationCompleted' ]); }); diff --git a/libraries/reporter/src/test/LogLevelFilter.test.ts b/libraries/reporter/src/test/LogLevelFilter.test.ts index b1c6569ffe..a50fa6d3df 100644 --- a/libraries/reporter/src/test/LogLevelFilter.test.ts +++ b/libraries/reporter/src/test/LogLevelFilter.test.ts @@ -40,6 +40,7 @@ describe('getEventMinimumLogLevel', () => { it('classifies standard lifecycle and non-required warnings as normal', () => { expect(getEventMinimumLogLevel(ev('commandStarted', { commandName: 'build' }))).toBe('normal'); expect(getEventMinimumLogLevel(ev('operationStatusChanged', { status: 'success' }))).toBe('normal'); + expect(getEventMinimumLogLevel(ev('operationCompleted', { status: 'success' }))).toBe('normal'); expect(getEventMinimumLogLevel(ev('diagnosticEmitted', { severity: 'warning' }, false))).toBe('normal'); }); @@ -47,6 +48,7 @@ describe('getEventMinimumLogLevel', () => { expect(getEventMinimumLogLevel(ev('operationRegistered', {}))).toBe('normal'); expect(getEventMinimumLogLevel(ev('externalProcessStarted', {}))).toBe('verbose'); expect(getEventMinimumLogLevel(ev('externalOutput', { stream: 'stdout', text: 'x' }))).toBe('debug'); + expect(getEventMinimumLogLevel(ev('operationStreamClosed', {}))).toBe('debug'); expect(getEventMinimumLogLevel(ev('messageEmitted', { severity: 'debug', text: 'd' }))).toBe('debug'); expect(getEventMinimumLogLevel(ev('extension', { name: 'a.b' }, false))).toBe('normal'); }); diff --git a/libraries/reporter/src/test/OperationStreamEmitter.test.ts b/libraries/reporter/src/test/OperationStreamEmitter.test.ts index a6b47eaf74..680cdb49e1 100644 --- a/libraries/reporter/src/test/OperationStreamEmitter.test.ts +++ b/libraries/reporter/src/test/OperationStreamEmitter.test.ts @@ -61,10 +61,12 @@ describe('OperationStreamEmitter', () => { it('emits registration, status, output, and result with operation scope', () => { const sink: CapturingSink = new CapturingSink(); const emitter: OperationStreamEmitter = makeEmitter(sink); - emitter.registerOperation('op1', 'project-a', 'build'); - emitter.changeStatus('op1', 'executing'); + emitter.registerOperation('op1', 'project-a', 'build', false); + emitter.changeStatus('op1', 'executing', 0, 'queued'); emitter.writeOutput('op1', 'stdout', 'hello\n'); - emitter.changeStatus('op1', 'success', 100); + emitter.changeStatus('op1', 'success', 100, 'executing'); + emitter.closeOperationStream('op1'); + emitter.completeOperation('op1', 'success', 100); emitter.completeCommand('build', true, 0, { success: 1 }); expect(sink.inputs.map((i) => i.type)).toEqual([ @@ -72,15 +74,45 @@ describe('OperationStreamEmitter', () => { 'operationStatusChanged', 'externalOutput', 'operationStatusChanged', + 'operationStreamClosed', + 'operationCompleted', 'commandResult' ]); expect(sink.inputs[2].scope).toEqual({ commandName: 'build', operationId: 'op1' }); expect(sink.inputs[2].privacy).toBe('local-sensitive'); - expect(sink.inputs[3].payload).toMatchObject({ operationId: 'op1', status: 'success', durationMs: 100 }); + expect(sink.inputs[0].payload).toMatchObject({ operationId: 'op1', silent: false }); + expect(sink.inputs[3].payload).toMatchObject({ + operationId: 'op1', + previousStatus: 'executing', + status: 'success', + durationMs: 100 + }); // externalOutput is protected (never coalesced/dropped); the manager derives `required`. expect(isReporterEventRequired('externalOutput')).toBe(true); }); + it('records silent metadata and orders close before completion', () => { + const sink: CapturingSink = new CapturingSink(); + const emitter: OperationStreamEmitter = makeEmitter(sink); + emitter.registerOperation('silent-op', 'project-a', '_phase:synthetic', true); + emitter.changeStatus('silent-op', 'noOp', 0, 'ready'); + emitter.closeOperationStream('silent-op'); + emitter.completeOperation('silent-op', 'noOp', 0); + + expect(sink.inputs.map(({ type }) => type)).toEqual([ + 'operationRegistered', + 'operationStatusChanged', + 'operationStreamClosed', + 'operationCompleted' + ]); + expect(sink.inputs[0].payload).toEqual({ + operationId: 'silent-op', + projectName: 'project-a', + phaseName: '_phase:synthetic', + silent: true + }); + }); + it('splits raw output into uncollated chunks', () => { const sink: CapturingSink = new CapturingSink(); const emitter: OperationStreamEmitter = makeEmitter(sink, 4); diff --git a/libraries/reporter/src/test/Protocol.test.ts b/libraries/reporter/src/test/Protocol.test.ts index b602d9d5cc..8efe5138ba 100644 --- a/libraries/reporter/src/test/Protocol.test.ts +++ b/libraries/reporter/src/test/Protocol.test.ts @@ -18,6 +18,7 @@ import { describe('ReporterProtocol', () => { it('advertises protocol major 1 and the specified byte limits', () => { expect(REPORTER_PROTOCOL_VERSION.major).toBe(1); + expect(REPORTER_PROTOCOL_VERSION.minor).toBe(1); expect(REPORTER_PROTOCOL_LIMITS.bootstrapBufferBytes).toBe(1024 * 1024); expect(REPORTER_PROTOCOL_LIMITS.ndjsonRecordBytes).toBe(1024 * 1024); expect(REPORTER_PROTOCOL_LIMITS.externalOutputChunkBytes).toBe(64 * 1024); @@ -175,10 +176,7 @@ describe('negotiateReporterHello', () => { it('rejects a malformed wire hello with a predictable validation error', () => { expect(() => - negotiateReporterHello( - { kind: 'hello' }, - { supportedProtocolVersion: { major: 1, minor: 0 } } - ) + negotiateReporterHello({ kind: 'hello' }, { supportedProtocolVersion: { major: 1, minor: 0 } }) ).toThrow(InvalidReporterHelloError); expect(() => negotiateReporterHello( diff --git a/libraries/reporter/src/test/Telemetry.test.ts b/libraries/reporter/src/test/Telemetry.test.ts index 423b8a78cf..fed71783a8 100644 --- a/libraries/reporter/src/test/Telemetry.test.ts +++ b/libraries/reporter/src/test/Telemetry.test.ts @@ -85,7 +85,7 @@ describe('TelemetrySubscriber', () => { expect(aggregate.diagnosticCodes).toEqual(['RUSH_OPERATION_FAILED']); expect(aggregate.diagnosticCategoryCounts).toEqual({ operation: 1 }); expect(aggregate.reporterMode).toBe('default'); - expect(aggregate.protocolVersion).toEqual({ major: 1, minor: 0 }); + expect(aggregate.protocolVersion).toEqual({ major: 1, minor: 1 }); expect(aggregate.producerVersions).toEqual(['@microsoft/rush-lib@5.177.2']); // The subscriber runs alongside a rendering reporter and does not consume events from it. diff --git a/libraries/reporter/src/test/__snapshots__/Goldens.test.ts.snap b/libraries/reporter/src/test/__snapshots__/Goldens.test.ts.snap index 76b6b15a1c..e46cd3c517 100644 --- a/libraries/reporter/src/test/__snapshots__/Goldens.test.ts.snap +++ b/libraries/reporter/src/test/__snapshots__/Goldens.test.ts.snap @@ -3,7 +3,7 @@ exports[`compatibility goldens advertises the current protocol version as the negotiation baseline 1`] = ` Object { "major": 1, - "minor": 0, + "minor": 1, } `; diff --git a/libraries/rush-lib/src/logic/operations/OperationEventSink.ts b/libraries/rush-lib/src/logic/operations/OperationEventSink.ts index cda9311ebb..b9c9c7932d 100644 --- a/libraries/rush-lib/src/logic/operations/OperationEventSink.ts +++ b/libraries/rush-lib/src/logic/operations/OperationEventSink.ts @@ -46,10 +46,7 @@ export interface IOperationGraphEventSink { * Invoked synchronously on every operation status transition. The result's * `status`, `error`, and `stopwatch` reflect the new state. */ - onOperationStatusChanged?( - result: IOperationExecutionResult, - previousStatus: OperationStatus - ): void; + onOperationStatusChanged?(result: IOperationExecutionResult, previousStatus: OperationStatus): void; /** * Invoked when an operation's collated output is about to be displayed, @@ -72,6 +69,11 @@ export interface IOperationGraphEventSink { */ onOperationStreamClosed?(operationId: string): void; + /** + * Invoked after the operation stream is closed and the final outcome is authoritative. + */ + onOperationCompleted?(result: IOperationExecutionResult): void; + /** * Invoked for each human-oriented status line written to the terminal, * carrying the plain (pre-colorization) text. diff --git a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts index 47317ecfac..a1c3bf5f0a 100644 --- a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts +++ b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts @@ -173,6 +173,7 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera private _status: OperationStatus; private _stateHash: string | undefined; private _stateHashComponents: IOperationStateHashComponents | undefined; + private _operationStreamClosed: boolean = false; public constructor(operation: Operation, context: IOperationExecutionRecordContext) { const { runner, associatedPhase, associatedProject, enabled } = operation; @@ -283,6 +284,18 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera return !this.enabled || this.runner.silent; } + /** + * Notifies observers that this iteration cannot emit more output for the operation. + * + * @internal + */ + public closeOperationStream(): void { + if (!this._operationStreamClosed) { + this._operationStreamClosed = true; + this._context.eventSink?.onOperationStreamClosed?.(this.name); + } + } + public getStateHash(): string { if (this._stateHash === undefined) { const { dependencies, local, config } = this.getStateHashComponents(); @@ -486,9 +499,6 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera } finally { if (this.isTerminal) { this._collatedWriter?.close(); - if (this._collatedWriter) { - this._context.eventSink?.onOperationStreamClosed?.(this.name); - } this.stdioSummarizer.close(); this.problemCollector.close(); } diff --git a/libraries/rush-lib/src/logic/operations/OperationGraph.ts b/libraries/rush-lib/src/logic/operations/OperationGraph.ts index 58e6c692aa..316e90cea5 100644 --- a/libraries/rush-lib/src/logic/operations/OperationGraph.ts +++ b/libraries/rush-lib/src/logic/operations/OperationGraph.ts @@ -676,9 +676,8 @@ export class OperationGraph implements IOperationGraph { operation, iterationContext ); - executionRecords.set(operation, executionRecord); - eventSink?.onOperationRegistered?.(executionRecord.name, executionRecord.silent); + executionRecords.set(operation, executionRecord); } for (const [operation, record] of executionRecords) { @@ -717,6 +716,10 @@ export class OperationGraph implements IOperationGraph { return; } + for (const executionRecord of executionRecords.values()) { + eventSink?.onOperationRegistered?.(executionRecord.name, executionRecord.silent); + } + this._setScheduledIteration(iterationContext); // Notify listeners that an iteration has been scheduled with the planned operation records try { @@ -961,6 +964,8 @@ export class OperationGraph implements IOperationGraph { } } for (const record of executionRecords.values()) { + record.closeOperationStream(); + eventSink?.onOperationCompleted?.(record); record.stdioSummarizer.close(); record.problemCollector.close(); } @@ -1295,10 +1300,9 @@ function _handleOperationNoOp(record: OperationExecutionRecord, context: IStatef function _handleOperationSuccess(record: OperationExecutionRecord, context: IStatefulExecutionContext): void { const stopwatch: IStopwatchResult = _getOperationStopwatch(record); if (!record.silent) { - record.eventSink?.onActivity?.( - `"${record.name}" completed successfully in ${stopwatch.toString()}.`, - { operationId: record.name } - ); + record.eventSink?.onActivity?.(`"${record.name}" completed successfully in ${stopwatch.toString()}.`, { + operationId: record.name + }); record.collatedWriter.terminal.writeStdoutLine( Colorize.green(`"${record.name}" completed successfully in ${stopwatch.toString()}.`) ); @@ -1315,10 +1319,10 @@ function _handleOperationSuccessWithWarning( ): void { const stopwatch: IStopwatchResult = _getOperationStopwatch(record); if (!record.silent) { - record.eventSink?.onActivity?.( - `"${record.name}" completed with warnings in ${stopwatch.toString()}.`, - { operationId: record.name, stderr: true } - ); + record.eventSink?.onActivity?.(`"${record.name}" completed with warnings in ${stopwatch.toString()}.`, { + operationId: record.name, + stderr: true + }); record.collatedWriter.terminal.writeStderrLine( Colorize.yellow(`"${record.name}" completed with warnings in ${stopwatch.toString()}.`) ); diff --git a/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts b/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts index 11a100e68b..10ba5f15bd 100644 --- a/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts +++ b/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts @@ -5,14 +5,16 @@ import { createRushDiagnostic, type IRushDiagnostic, type LifecycleEmitter, + type OperationStreamEmitter, type OperationStatus as ReporterOperationStatus } from '@rushstack/rush-reporter'; -import type { ITerminalChunk } from '@rushstack/terminal'; +import { TerminalChunkKind, type ITerminalChunk } from '@rushstack/terminal'; import type { RushSession } from '../../pluginFramework/RushSession'; import { _correlateRushSessionError, - _getRushSessionLifecycleEmitter + _getRushSessionLifecycleEmitter, + _getRushSessionOperationStreamEmitter } from '../../pluginFramework/RushSession'; import type { IOperationExecutionResult } from './IOperationExecutionResult'; import type { IOperationGraphEventSink, IOperationActivityOptions } from './OperationEventSink'; @@ -28,11 +30,19 @@ interface IReporterOperation { readonly projectName: string; readonly registeredOperationIds: Set; readonly statuses: Map; + readonly streamEmitter: OperationStreamEmitter | undefined; + readonly closedOperationIds: Set; + readonly completedResults: Map; lastEmittedStatus: ReporterOperationStatus | undefined; silent: boolean; + streamClosed: boolean; } class ReporterOperationEventSink implements IOperationGraphEventSink { + public readonly onOperationChunk: ((operationId: string, chunk: ITerminalChunk) => void) | undefined; + public readonly onOperationStreamClosed: ((operationId: string) => void) | undefined; + public readonly onOperationCompleted: ((result: IOperationExecutionResult) => void) | undefined; + private readonly _operationsByLegacyId: Map = new Map(); private readonly _diagnosedOperations: Set = new Set(); private readonly _rushSession: RushSession; @@ -56,6 +66,15 @@ class ReporterOperationEventSink implements IOperationGraphEventSink { if (!emitter) { continue; } + const streamEmitter: OperationStreamEmitter | undefined = _getRushSessionOperationStreamEmitter( + rushSession, + { + commandName, + operationId, + projectName, + phaseName + } + ); reporterOperation = { emitter, legacyOperationIds: new Set(), @@ -64,8 +83,12 @@ class ReporterOperationEventSink implements IOperationGraphEventSink { projectName, registeredOperationIds: new Set(), statuses: new Map(), + streamEmitter, + closedOperationIds: new Set(), + completedResults: new Map(), lastEmittedStatus: undefined, - silent: true + silent: true, + streamClosed: false }; operationsByReporterId.set(operationId, reporterOperation); } @@ -73,6 +96,16 @@ class ReporterOperationEventSink implements IOperationGraphEventSink { reporterOperation.silent &&= !operation.enabled || operation.runner?.silent === true; this._operationsByLegacyId.set(operation.name, reporterOperation); } + + if (Array.from(this._operationsByLegacyId.values()).some(({ streamEmitter }) => !!streamEmitter)) { + this.onOperationChunk = (operationId, chunk) => this._onOperationChunk(operationId, chunk); + this.onOperationStreamClosed = (operationId) => this._onOperationStreamClosed(operationId); + this.onOperationCompleted = (result) => this._onOperationCompleted(result); + } else { + this.onOperationChunk = undefined; + this.onOperationStreamClosed = undefined; + this.onOperationCompleted = undefined; + } } public get isEnabled(): boolean { @@ -88,25 +121,37 @@ class ReporterOperationEventSink implements IOperationGraphEventSink { if (operation.registeredOperationIds.size === operation.legacyOperationIds.size) { operation.registeredOperationIds.clear(); operation.statuses.clear(); + operation.closedOperationIds.clear(); + operation.completedResults.clear(); operation.lastEmittedStatus = undefined; operation.silent = true; + operation.streamClosed = false; this._diagnosedOperations.delete(operation.operationId); } operation.registeredOperationIds.add(operationId); operation.silent &&= silent; - if (operation.registeredOperationIds.size !== operation.legacyOperationIds.size || operation.silent) { + if (operation.registeredOperationIds.size !== operation.legacyOperationIds.size) { return; } - operation.emitter.emitOperationRegistered({ - operationId: operation.operationId, - projectName: operation.projectName, - phaseName: operation.phaseName - }); + if (operation.streamEmitter) { + operation.streamEmitter.registerOperation( + operation.operationId, + operation.projectName, + operation.phaseName, + operation.silent + ); + } else if (!operation.silent) { + operation.emitter.emitOperationRegistered({ + operationId: operation.operationId, + projectName: operation.projectName, + phaseName: operation.phaseName + }); + } } - public onOperationStatusChanged(result: IOperationExecutionResult): void { + public onOperationStatusChanged(result: IOperationExecutionResult, previousStatus: OperationStatus): void { const operation: IReporterOperation | undefined = this._operationsByLegacyId.get(result.operation.name); if (!operation) { return; @@ -137,12 +182,22 @@ class ReporterOperationEventSink implements IOperationGraphEventSink { if (status === undefined || status === operation.lastEmittedStatus) { return; } + const aggregatePreviousStatus: ReporterOperationStatus | undefined = + operation.lastEmittedStatus ?? + (operation.legacyOperationIds.size === 1 ? _toReporterStatus(previousStatus) : undefined); operation.lastEmittedStatus = status; - if (!operation.silent) { - const durationMs: number | undefined = - operation.legacyOperationIds.size === 1 && result.stopwatch.startTime !== undefined - ? result.stopwatch.duration * 1000 - : undefined; + const durationMs: number | undefined = + operation.legacyOperationIds.size === 1 && result.stopwatch.startTime !== undefined + ? result.stopwatch.duration * 1000 + : undefined; + if (operation.streamEmitter) { + operation.streamEmitter.changeStatus( + operation.operationId, + status, + durationMs, + aggregatePreviousStatus + ); + } else if (!operation.silent) { operation.emitter.emitOperationStatusChanged({ operationId: operation.operationId, status, @@ -150,11 +205,58 @@ class ReporterOperationEventSink implements IOperationGraphEventSink { }); } } + + private _onOperationChunk(operationId: string, chunk: ITerminalChunk): void { + const operation: IReporterOperation | undefined = this._operationsByLegacyId.get(operationId); + if (!operation?.streamEmitter) { + return; + } + + if (chunk.kind === TerminalChunkKind.Stdout) { + operation.streamEmitter.writeOutput(operation.operationId, 'stdout', chunk.text); + } else if (chunk.kind === TerminalChunkKind.Stderr) { + operation.streamEmitter.writeOutput(operation.operationId, 'stderr', chunk.text); + } + } + + private _onOperationStreamClosed(operationId: string): void { + const operation: IReporterOperation | undefined = this._operationsByLegacyId.get(operationId); + if (!operation?.streamEmitter || operation.streamClosed) { + return; + } + operation.closedOperationIds.add(operationId); + if (operation.closedOperationIds.size === operation.legacyOperationIds.size) { + operation.streamClosed = true; + operation.streamEmitter.closeOperationStream(operation.operationId); + } + } + + private _onOperationCompleted(result: IOperationExecutionResult): void { + const operation: IReporterOperation | undefined = this._operationsByLegacyId.get(result.operation.name); + if (!operation?.streamEmitter) { + return; + } + + operation.completedResults.set(result.operation.name, result); + if (operation.completedResults.size !== operation.legacyOperationIds.size) { + return; + } + const status: ReporterOperationStatus = _getAggregateTerminalStatus( + Array.from(operation.completedResults.values(), ({ status: resultStatus }) => resultStatus) + ); + const durationMs: number | undefined = _getAggregateDurationMs(operation.completedResults); + operation.streamEmitter.completeOperation( + operation.operationId, + status, + durationMs + ); + } } class CompositeOperationGraphEventSink implements IOperationGraphEventSink { public readonly onOperationChunk: ((operationId: string, chunk: ITerminalChunk) => void) | undefined; public readonly onOperationStreamClosed: ((operationId: string) => void) | undefined; + public readonly onOperationCompleted: ((result: IOperationExecutionResult) => void) | undefined; private readonly _first: IOperationGraphEventSink; private readonly _second: IOperationGraphEventSink; @@ -176,6 +278,13 @@ class CompositeOperationGraphEventSink implements IOperationGraphEventSink { second.onOperationStreamClosed?.(operationId); } : undefined; + this.onOperationCompleted = + first.onOperationCompleted || second.onOperationCompleted + ? (result) => { + first.onOperationCompleted?.(result); + second.onOperationCompleted?.(result); + } + : undefined; } public onOperationRegistered(operationId: string, silent: boolean): void { @@ -200,7 +309,7 @@ class CompositeOperationGraphEventSink implements IOperationGraphEventSink { } /** - * Adds status-only reporter emission without changing the graph's visible output or raw chunk routing. + * Adds reporter emission without changing the graph's visible output or collator routing. * * @internal */ @@ -312,3 +421,30 @@ function _isTerminalStatus(status: OperationStatus): boolean { return false; } } + +function _getAggregateDurationMs( + results: ReadonlyMap +): number | undefined { + let startTime: number | undefined; + let endTime: number | undefined; + for (const result of results.values()) { + if (result.stopwatch.startTime !== undefined) { + startTime = + startTime === undefined + ? result.stopwatch.startTime + : Math.min(startTime, result.stopwatch.startTime); + } + if (result.stopwatch.endTime !== undefined) { + endTime = + endTime === undefined ? result.stopwatch.endTime : Math.max(endTime, result.stopwatch.endTime); + } + } + if (startTime !== undefined && endTime !== undefined) { + return Math.max(0, endTime - startTime); + } + if (results.size === 1) { + const result: IOperationExecutionResult | undefined = results.values().next().value; + return result?.stopwatch.startTime === undefined ? undefined : result.stopwatch.duration * 1000; + } + return undefined; +} diff --git a/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts b/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts index 19f7bf16d6..92bec43e5a 100644 --- a/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts +++ b/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts @@ -35,7 +35,12 @@ jest.mock('../ProjectLogWritable', () => { }); import type { IReporterEmitEventInput, IReporterEventSink } from '@rushstack/rush-reporter'; -import { MockWritable, StringBufferTerminalProvider, type ITerminalChunk } from '@rushstack/terminal'; +import { + MockWritable, + StringBufferTerminalProvider, + TerminalProviderSeverity, + type ITerminalChunk +} from '@rushstack/terminal'; import type { CollatedTerminal } from '@rushstack/stream-collator'; import type { IPhase } from '../../../api/CommandLineConfiguration'; @@ -85,6 +90,8 @@ class RecordingSink implements IOperationGraphEventSink { public readonly headers: [string, number, number][] = []; public readonly activities: string[] = []; public readonly chunks: Map = new Map(); + public readonly closed: string[] = []; + public readonly completed: [string, string][] = []; public onOperationRegistered(operationId: string, silent: boolean): void { this.registered.push([operationId, silent]); @@ -106,6 +113,12 @@ class RecordingSink implements IOperationGraphEventSink { } chunks.push(chunk.text); } + public onOperationStreamClosed(operationId: string): void { + this.closed.push(operationId); + } + public onOperationCompleted(result: IOperationExecutionResult): void { + this.completed.push([result.operation.name, result.status]); + } } class CapturingReporterSink implements IReporterEventSink { @@ -167,6 +180,11 @@ describe('OperationGraph event sink (dual-emit)', () => { expect(sink.activities.some((line: string) => line.includes('"alpha" completed successfully'))).toBe( true ); + expect([...sink.closed].sort()).toEqual(['alpha', 'beta']); + expect([...sink.completed].sort()).toEqual([ + ['alpha', OperationStatus.Success], + ['beta', OperationStatus.Success] + ]); }); it('emits raw per-operation chunks even in quiet mode, matching the collated stream', async () => { @@ -235,7 +253,11 @@ describe('OperationGraph event sink (dual-emit)', () => { const rushSession: RushSession = new RushSession({ terminalProvider: new StringBufferTerminalProvider(), getIsDebugMode: () => false, - reporter: { eventSink: reporterSink, sessionId: 'operation-shadow' } + reporter: { + eventSink: reporterSink, + sessionId: 'operation-shadow', + operationStreamEnabled: false + } }); const createFailingOperation = (): Operation => createOperation( @@ -279,6 +301,174 @@ describe('OperationGraph event sink (dual-emit)', () => { expect(mockWritable.getAllOutput()).toEqual(plainWritable.getAllOutput()); }); + it('emits the opted-in canonical stream without duplicating or losing operation chunks', async () => { + const stdoutText: string = `${'a'.repeat(64 * 1024 + 7)}\n`; + const stderrText: string = 'stderr detail\n'; + const createOutputRunner = (): IOperationRunner => ({ + name: '@scope/project (_phase:build)', + reportTiming: true, + silent: false, + cacheable: false, + warningsAreAllowed: true, + isNoOp: false, + executeAsync: async (context: IOperationRunnerContext) => + await context.runWithTerminalAsync( + async (terminal, terminalProvider) => { + void terminal; + terminalProvider.write(stdoutText, TerminalProviderSeverity.log); + terminalProvider.write(stderrText, TerminalProviderSeverity.error); + return OperationStatus.SuccessWithWarning; + }, + { createLogFile: false, logFileSuffix: '' } + ), + getConfigHash: () => 'mock' + }); + + const plainWritable: MockWritable = new MockWritable(); + await new OperationGraph( + new Set([createOperation('@scope/project', createOutputRunner(), mockPhase, '@scope/project')]), + createGraphOptions(plainWritable, false) + ).executeAsync({}); + + const reporterSink: CapturingReporterSink = new CapturingReporterSink(); + const rushSession: RushSession = new RushSession({ + terminalProvider: new StringBufferTerminalProvider(), + getIsDebugMode: () => false, + reporter: { + eventSink: reporterSink, + sessionId: 'operation-stream', + operationStreamEnabled: true + } + }); + const streamedWritable: MockWritable = new MockWritable(); + const graph: OperationGraph = new OperationGraph( + new Set([createOperation('@scope/project', createOutputRunner(), mockPhase, '@scope/project')]), + createGraphOptions(streamedWritable, false) + ); + + attachReporterOperationEventSink(graph, rushSession, 'build'); + await graph.executeAsync({}); + + expect(streamedWritable.getAllOutput()).toEqual(plainWritable.getAllOutput()); + + const operationEvents: IReporterEmitEventInput[] = reporterSink.inputs.filter( + ({ scope }) => scope?.operationId === '@scope/project#phase' + ); + expect(operationEvents[0]).toMatchObject({ + type: 'operationRegistered', + payload: { + operationId: '@scope/project#phase', + projectName: '@scope/project', + phaseName: 'phase', + silent: false + } + }); + + const statusEvents: IReporterEmitEventInput[] = operationEvents.filter( + ({ type }) => type === 'operationStatusChanged' + ); + expect(statusEvents.map(({ payload }) => payload)).toEqual([ + expect.objectContaining({ previousStatus: 'ready', status: 'queued' }), + expect.objectContaining({ previousStatus: 'queued', status: 'executing' }), + expect.objectContaining({ previousStatus: 'executing', status: 'successWithWarnings' }) + ]); + + const outputEvents: IReporterEmitEventInput[] = operationEvents.filter( + ({ type }) => type === 'externalOutput' + ); + expect( + outputEvents.every( + ({ payload }) => Buffer.byteLength((payload as { text: string }).text, 'utf8') <= 64 * 1024 + ) + ).toBe(true); + const stdoutChunks: string = outputEvents + .filter(({ payload }) => (payload as { stream: string }).stream === 'stdout') + .map(({ payload }) => (payload as { text: string }).text) + .join(''); + const stderrChunks: string = outputEvents + .filter(({ payload }) => (payload as { stream: string }).stream === 'stderr') + .map(({ payload }) => (payload as { text: string }).text) + .join(''); + expect(stdoutChunks).toBe(stdoutText); + expect(stderrChunks).toBe(stderrText); + + const closedIndex: number = operationEvents.findIndex(({ type }) => type === 'operationStreamClosed'); + const completedIndex: number = operationEvents.findIndex(({ type }) => type === 'operationCompleted'); + expect(closedIndex).toBeGreaterThan(operationEvents.lastIndexOf(outputEvents.at(-1)!)); + expect(completedIndex).toBeGreaterThan(closedIndex); + expect(operationEvents[completedIndex].payload).toMatchObject({ + operationId: '@scope/project#phase', + status: 'successWithWarnings' + }); + }); + + it('reports silent operation metadata and outcomes on the opted-in stream', async () => { + const silentRunner: IOperationRunner = { + name: 'silent synthetic', + reportTiming: false, + silent: true, + cacheable: false, + warningsAreAllowed: false, + isNoOp: false, + executeAsync: async () => OperationStatus.Success, + getConfigHash: () => 'silent' + }; + const reporterSink: CapturingReporterSink = new CapturingReporterSink(); + const rushSession: RushSession = new RushSession({ + terminalProvider: new StringBufferTerminalProvider(), + getIsDebugMode: () => false, + reporter: { + eventSink: reporterSink, + sessionId: 'silent-operation', + operationStreamEnabled: true + } + }); + const graph: OperationGraph = new OperationGraph( + new Set([ + createOperation('visible', new MockOperationRunner('visible'), mockPhase, '@scope/visible'), + createOperation('silent synthetic', silentRunner, mockPhase, '@scope/project') + ]), + createGraphOptions(mockWritable, false) + ); + + attachReporterOperationEventSink(graph, rushSession, 'build'); + await graph.executeAsync({}); + + expect(reporterSink.inputs).toContainEqual( + expect.objectContaining({ + type: 'operationRegistered', + payload: expect.objectContaining({ + operationId: '@scope/project#phase', + silent: true + }) + }) + ); + expect(reporterSink.inputs).toContainEqual( + expect.objectContaining({ + type: 'operationCompleted', + payload: expect.objectContaining({ + operationId: '@scope/project#phase', + status: 'success' + }) + }) + ); + }); + + it('does not attach an operation adapter when the session has no reporter sink', () => { + const rushSession: RushSession = new RushSession({ + terminalProvider: new StringBufferTerminalProvider(), + getIsDebugMode: () => false + }); + const graph: OperationGraph = new OperationGraph( + new Set([createOperation('no sink', new MockOperationRunner('no sink'))]), + createGraphOptions(mockWritable, false) + ); + + attachReporterOperationEventSink(graph, rushSession, 'build'); + + expect(graph.eventSink).toBeUndefined(); + }); + it('aggregates sharded records across mixed outcomes and repeated watch-style iterations', async () => { const reporterSink: CapturingReporterSink = new CapturingReporterSink(); const rushSession: RushSession = new RushSession({ @@ -440,7 +630,11 @@ describe('OperationGraph event sink (dual-emit)', () => { const rushSession: RushSession = new RushSession({ terminalProvider: new StringBufferTerminalProvider(), getIsDebugMode: () => false, - reporter: { eventSink: reporterSink, sessionId: 'operation-retries' } + reporter: { + eventSink: reporterSink, + sessionId: 'operation-retries', + operationStreamEnabled: true + } }); const compilePhase: IPhase = { ...mockPhase, @@ -484,6 +678,16 @@ describe('OperationGraph event sink (dual-emit)', () => { '@scope/project#_phase:compile', '@scope/project#_phase:test' ]); + expect( + reporterSink.inputs + .filter(({ type }) => type === 'operationCompleted') + .map(({ scope }) => scope?.operationId) + ).toEqual([ + '@scope/project#_phase:compile', + '@scope/project#_phase:test', + '@scope/project#_phase:compile', + '@scope/project#_phase:test' + ]); for (const event of reporterSink.inputs.filter(({ type }) => type === 'operationStatusChanged')) { expect(event.scope?.operationId).toBe(`@scope/project#${event.scope?.phaseName}`); expect((event.payload as { operationId: string }).operationId).toBe(event.scope?.operationId); @@ -508,7 +712,11 @@ describe('OperationGraph event sink (dual-emit)', () => { const rushSession: RushSession = new RushSession({ terminalProvider: new StringBufferTerminalProvider(), getIsDebugMode: () => false, - reporter: { eventSink: reporterSink, sessionId: 'output-parity' } + reporter: { + eventSink: reporterSink, + sessionId: 'output-parity', + operationStreamEnabled: false + } }); const shadowWritable: MockWritable = new MockWritable(); const shadowGraph: OperationGraph = new OperationGraph( @@ -516,6 +724,7 @@ describe('OperationGraph event sink (dual-emit)', () => { createGraphOptions(shadowWritable, false) ); attachReporterOperationEventSink(shadowGraph, rushSession, 'build'); + expect(shadowGraph.eventSink?.onOperationChunk).toBeUndefined(); await shadowGraph.executeAsync({}); expect(shadowWritable.getAllOutput()).toEqual(plainWritable.getAllOutput()); expect(reporterSink.inputs.some(({ type }) => type === 'externalOutput')).toBe(false); diff --git a/libraries/rush-lib/src/pluginFramework/RushSession.ts b/libraries/rush-lib/src/pluginFramework/RushSession.ts index e52cbb0ad0..9525b92dc4 100644 --- a/libraries/rush-lib/src/pluginFramework/RushSession.ts +++ b/libraries/rush-lib/src/pluginFramework/RushSession.ts @@ -5,6 +5,7 @@ import { InternalError, PackageJsonLookup, type IPackageJson } from '@rushstack/ import { LifecycleEmitter, LegacyErrorBridge, + OperationStreamEmitter, RushSessionReporting, TelemetrySubscriber, isReporterEventRequired, @@ -49,6 +50,17 @@ export interface IRushSessionReporterOptions { * The identifier assigned to this Rush session by the frontend. */ readonly sessionId: string; + + /** + * Enables raw semantic operation events for the pre-major reporter opt-in path. + * + * @remarks + * When false or omitted, Rush retains the shadow lifecycle-only behavior and + * does not tap operation output. + * + * @internal + */ + readonly operationStreamEnabled?: boolean; } /** @@ -293,6 +305,22 @@ function _createLifecycleEmitter( }); } +function _createOperationStreamEmitter( + state: IRushSessionReportingState | undefined, + scope?: IReporterEventScope +): OperationStreamEmitter | undefined { + if (!state) { + return undefined; + } + + return new OperationStreamEmitter({ + sink: state.eventSink, + sessionId: state.sessionId, + source: state.source, + scope: scope ? { ...scope } : undefined + }); +} + function _getSessionState(rushSession: RushSession): IRushSessionState { const state: IRushSessionState | undefined = _rushSessionStates.get(rushSession); if (!state) { @@ -451,6 +479,21 @@ export function _getRushSessionLifecycleEmitter( return _createLifecycleEmitter(_getSessionState(rushSession).reporting, scope); } +/** + * Creates the raw operation stream emitter only for the pre-major opt-in path. + * + * @internal + */ +export function _getRushSessionOperationStreamEmitter( + rushSession: RushSession, + scope?: IReporterEventScope +): OperationStreamEmitter | undefined { + const state: IRushSessionState = _getSessionState(rushSession); + return state.options.reporter?.operationStreamEnabled + ? _createOperationStreamEmitter(state.reporting, scope) + : undefined; +} + /** * Returns the current allowlisted reporter telemetry projection. * From 22511f317951d728e6383513e14202f4d51d6767 Mon Sep 17 00:00:00 2001 From: TheLarkInn Date: Fri, 28 Aug 2026 08:10:04 +0000 Subject: [PATCH 2/3] Fix operation reporter stream edge cases Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d6318e80-5da9-4858-a147-817e8692f10e --- .../operations/CacheableOperationPlugin.ts | 10 +- .../logic/operations/OperationEventSink.ts | 3 +- .../operations/OperationExecutionRecord.ts | 40 +++--- .../src/logic/operations/OperationGraph.ts | 33 +++-- .../operations/ReporterOperationEventSink.ts | 6 +- .../test/OperationGraphEventSink.test.ts | 124 +++++++++++++++++- 6 files changed, 178 insertions(+), 38 deletions(-) diff --git a/libraries/rush-lib/src/logic/operations/CacheableOperationPlugin.ts b/libraries/rush-lib/src/logic/operations/CacheableOperationPlugin.ts index 6f254f52fc..de28b6a499 100644 --- a/libraries/rush-lib/src/logic/operations/CacheableOperationPlugin.ts +++ b/libraries/rush-lib/src/logic/operations/CacheableOperationPlugin.ts @@ -762,15 +762,17 @@ export class CacheableOperationPlugin implements IPhasedCommandPlugin { cacheConsoleWritable = collatedWriter; } - let cacheCollatedTerminal: CollatedTerminal; + let cacheDestination: TerminalWritable; if (cacheProjectLogWritable) { - const cacheSplitterTransform: SplitterTransform = new SplitterTransform({ + cacheDestination = new SplitterTransform({ destinations: [cacheConsoleWritable, cacheProjectLogWritable] }); - cacheCollatedTerminal = new CollatedTerminal(cacheSplitterTransform); } else { - cacheCollatedTerminal = new CollatedTerminal(cacheConsoleWritable); + cacheDestination = cacheConsoleWritable; } + const cacheCollatedTerminal: CollatedTerminal = new CollatedTerminal( + record.addOperationChunkTap(cacheDestination) + ); const buildCacheTerminalProvider: CollatedTerminalProvider = new CollatedTerminalProvider( cacheCollatedTerminal, diff --git a/libraries/rush-lib/src/logic/operations/OperationEventSink.ts b/libraries/rush-lib/src/logic/operations/OperationEventSink.ts index b9c9c7932d..45f9f2a11e 100644 --- a/libraries/rush-lib/src/logic/operations/OperationEventSink.ts +++ b/libraries/rush-lib/src/logic/operations/OperationEventSink.ts @@ -57,8 +57,7 @@ export interface IOperationGraphEventSink { /** * Invoked for each chunk of an operation's raw output, upstream of any - * quiet-mode filtering. Concatenated chunks for one operation exactly match - * what the collated sink receives for that operation. + * newline normalization or quiet-mode filtering. */ onOperationChunk?(operationId: string, chunk: ITerminalChunk): void; diff --git a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts index a1c3bf5f0a..61cfe514a4 100644 --- a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts +++ b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts @@ -296,6 +296,27 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera } } + /** + * Adds the reporter's lossless operation-output tap ahead of any legacy presentation transforms. + * + * @internal + */ + public addOperationChunkTap(destination: TerminalWritable): TerminalWritable { + const eventSink: IOperationGraphEventSink | undefined = this._context.eventSink; + if (!eventSink?.onOperationChunk) { + return destination; + } + + return new SplitterTransform({ + destinations: [ + destination, + new OperationChunkTap(this.name, (operationId, chunk) => + eventSink.onOperationChunk?.(operationId, chunk) + ) + ] + }); + } + public getStateHash(): string { if (this._stateHash === undefined) { const { dependencies, local, config } = this.getStateHashComponents(); @@ -409,25 +430,12 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera newlineKind: NewlineKind.Lf // for StdioSummarizer }); - const chunkTapDestinations: TerminalWritable[] = []; - const eventSink: IOperationGraphEventSink | undefined = this._context.eventSink; - if (eventSink?.onOperationChunk) { - // Tap the stream upstream of the quiet-mode discard so the sink observes - // the exact bytes the collated writer would receive, regardless of verbosity. - chunkTapDestinations.push( - new OperationChunkTap(this.name, (operationId, chunk) => - eventSink.onOperationChunk?.(operationId, chunk) - ) - ); - } - const splitterTransform1: SplitterTransform = new SplitterTransform({ destinations: [ this.quietMode ? new DiscardStdoutTransform({ destination: this.collatedWriter }) : this.collatedWriter, - stderrLineTransform, - ...chunkTapDestinations + stderrLineTransform ] }); @@ -437,7 +445,9 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera ensureNewlineAtEnd: true }); - const collatedTerminal: CollatedTerminal = new CollatedTerminal(normalizeNewlineTransform); + const collatedTerminal: CollatedTerminal = new CollatedTerminal( + this.addOperationChunkTap(normalizeNewlineTransform) + ); const terminalProvider: CollatedTerminalProvider = new CollatedTerminalProvider(collatedTerminal, { debugEnabled: this.debugMode }); diff --git a/libraries/rush-lib/src/logic/operations/OperationGraph.ts b/libraries/rush-lib/src/logic/operations/OperationGraph.ts index 316e90cea5..22def845d9 100644 --- a/libraries/rush-lib/src/logic/operations/OperationGraph.ts +++ b/libraries/rush-lib/src/logic/operations/OperationGraph.ts @@ -716,10 +716,6 @@ export class OperationGraph implements IOperationGraph { return; } - for (const executionRecord of executionRecords.values()) { - eventSink?.onOperationRegistered?.(executionRecord.name, executionRecord.silent); - } - this._setScheduledIteration(iterationContext); // Notify listeners that an iteration has been scheduled with the planned operation records try { @@ -731,6 +727,9 @@ export class OperationGraph implements IOperationGraph { terminal.writeStderrLine(Colorize.red(errorMessage)); throw e; } + for (const executionRecord of executionRecords.values()) { + eventSink?.onOperationRegistered?.(executionRecord.name, executionRecord.silent); + } if (!this._currentIteration) { this._setIdleTimeout(); } else if (!this.pauseNextIteration) { @@ -878,12 +877,26 @@ export class OperationGraph implements IOperationGraph { terminal.writeStdoutLine(parallelismLine); eventSink?.onActivity?.(parallelismLine); - const bailStatus: OperationStatus | undefined | void = abortSignal.aborted - ? OperationStatus.Aborted - : await measureAsyncFn( - `${PERF_PREFIX}:beforeExecuteIterationAsync`, - async () => await hooks.beforeExecuteIterationAsync.promise(executionRecords, iterationOptions) - ); + let bailStatus: OperationStatus | undefined | void; + try { + bailStatus = abortSignal.aborted + ? OperationStatus.Aborted + : await measureAsyncFn( + `${PERF_PREFIX}:beforeExecuteIterationAsync`, + async () => await hooks.beforeExecuteIterationAsync.promise(executionRecords, iterationOptions) + ); + } catch (error) { + for (const record of executionRecords.values()) { + if (!record.isTerminal) { + record.status = OperationStatus.Aborted; + } + record.closeOperationStream(); + eventSink?.onOperationCompleted?.(record); + record.stdioSummarizer.close(); + record.problemCollector.close(); + } + throw error; + } if (bailStatus) { // A tap short-circuited the iteration. If it bailed with a successful status (e.g. the diff --git a/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts b/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts index 10ba5f15bd..87e347f764 100644 --- a/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts +++ b/libraries/rush-lib/src/logic/operations/ReporterOperationEventSink.ts @@ -245,11 +245,7 @@ class ReporterOperationEventSink implements IOperationGraphEventSink { Array.from(operation.completedResults.values(), ({ status: resultStatus }) => resultStatus) ); const durationMs: number | undefined = _getAggregateDurationMs(operation.completedResults); - operation.streamEmitter.completeOperation( - operation.operationId, - status, - durationMs - ); + operation.streamEmitter.completeOperation(operation.operationId, status, durationMs); } } diff --git a/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts b/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts index 92bec43e5a..e11aa7b767 100644 --- a/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts +++ b/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts @@ -302,8 +302,8 @@ describe('OperationGraph event sink (dual-emit)', () => { }); it('emits the opted-in canonical stream without duplicating or losing operation chunks', async () => { - const stdoutText: string = `${'a'.repeat(64 * 1024 + 7)}\n`; - const stderrText: string = 'stderr detail\n'; + const stdoutText: string = `${'a'.repeat(64 * 1024 + 7)}\rprogress`; + const stderrText: string = 'stderr detail\r'; const createOutputRunner = (): IOperationRunner => ({ name: '@scope/project (_phase:build)', reportTiming: true, @@ -402,6 +402,126 @@ describe('OperationGraph event sink (dual-emit)', () => { }); }); + it('combines sharded implementation records into one project x phase stream', async () => { + const reporterSink: CapturingReporterSink = new CapturingReporterSink(); + const rushSession: RushSession = new RushSession({ + terminalProvider: new StringBufferTerminalProvider(), + getIsDebugMode: () => false, + reporter: { + eventSink: reporterSink, + sessionId: 'sharded-operation-stream', + operationStreamEnabled: true + } + }); + const projectName: string = '@scope/sharded'; + const preShardRunner: IOperationRunner = { + name: `${projectName} (phase) - pre-shard`, + reportTiming: false, + silent: true, + cacheable: false, + warningsAreAllowed: false, + isNoOp: true, + executeAsync: async () => OperationStatus.NoOp, + getConfigHash: () => 'pre-shard' + }; + const shardRunner: IOperationRunner = { + name: `${projectName} (phase) - shard 1/1`, + reportTiming: true, + silent: false, + cacheable: false, + warningsAreAllowed: false, + isNoOp: false, + executeAsync: async (context: IOperationRunnerContext) => + await context.runWithTerminalAsync( + async (terminal) => { + terminal.write('shard output without newline'); + return OperationStatus.Failure; + }, + { createLogFile: false, logFileSuffix: '' } + ), + getConfigHash: () => 'shard' + }; + const collatorRunner: IOperationRunner = { + name: `${projectName} (phase) - collate`, + reportTiming: true, + silent: false, + cacheable: false, + warningsAreAllowed: false, + isNoOp: false, + executeAsync: async () => OperationStatus.Success, + getConfigHash: () => 'collate' + }; + const preShard: Operation = createOperation('pre-shard', preShardRunner, mockPhase, projectName); + const shard: Operation = createOperation('shard', shardRunner, mockPhase, projectName); + const collator: Operation = createOperation('collator', collatorRunner, mockPhase, projectName); + shard.addDependency(preShard); + collator.addDependency(shard); + const graph: OperationGraph = new OperationGraph( + new Set([collator, preShard, shard]), + createGraphOptions(mockWritable, false) + ); + + attachReporterOperationEventSink(graph, rushSession, 'build'); + await graph.executeAsync({}); + + const operationEvents: IReporterEmitEventInput[] = reporterSink.inputs.filter( + ({ scope }) => scope?.operationId === `${projectName}#phase` + ); + expect(operationEvents.filter(({ type }) => type === 'operationRegistered')).toHaveLength(1); + expect( + operationEvents + .filter(({ type }) => type === 'externalOutput') + .map(({ payload }) => (payload as { text: string }).text) + .join('') + ).toBe('shard output without newline'); + expect(operationEvents.filter(({ type }) => type === 'operationStreamClosed')).toHaveLength(1); + expect(operationEvents.filter(({ type }) => type === 'operationCompleted')).toEqual([ + expect.objectContaining({ + payload: expect.objectContaining({ operationId: `${projectName}#phase`, status: 'failure' }) + }) + ]); + expect( + operationEvents.filter( + ({ type, payload }) => + type === 'diagnosticEmitted' && (payload as { code?: string }).code === 'RUSH_OPERATION_FAILED' + ) + ).toHaveLength(1); + }); + + it('does not register operations when scheduling hooks reject the iteration', async () => { + const sink: RecordingSink = new RecordingSink(); + const graph: OperationGraph = new OperationGraph( + new Set([createOperation('hook failure', new MockOperationRunner('hook failure'))]), + createGraphOptions(mockWritable, false) + ); + graph.eventSink = sink; + graph.hooks.onIterationScheduled.tap('test', () => { + throw new Error('schedule rejected'); + }); + + await expect(graph.executeAsync({})).rejects.toThrow('schedule rejected'); + expect(sink.registered).toEqual([]); + expect(sink.closed).toEqual([]); + expect(sink.completed).toEqual([]); + }); + + it('finalizes registered operations when the pre-execution hook rejects', async () => { + const sink: RecordingSink = new RecordingSink(); + const graph: OperationGraph = new OperationGraph( + new Set([createOperation('hook failure', new MockOperationRunner('hook failure'))]), + createGraphOptions(mockWritable, false) + ); + graph.eventSink = sink; + graph.hooks.beforeExecuteIterationAsync.tapPromise('test', async () => { + throw new Error('pre-execution rejected'); + }); + + await expect(graph.executeAsync({})).rejects.toThrow('pre-execution rejected'); + expect(sink.registered).toEqual([['hook failure', false]]); + expect(sink.closed).toEqual(['hook failure']); + expect(sink.completed).toEqual([['hook failure', OperationStatus.Aborted]]); + }); + it('reports silent operation metadata and outcomes on the opted-in stream', async () => { const silentRunner: IOperationRunner = { name: 'silent synthetic', From 11a7f8462dd40b1d3c0d83ffafc5f0b748a80394 Mon Sep 17 00:00:00 2001 From: TheLarkInn Date: Fri, 28 Aug 2026 17:30:45 +0000 Subject: [PATCH 3/3] Fix operation stream review findings Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d6318e80-5da9-4858-a147-817e8692f10e --- common/reviews/api/rush-reporter.api.md | 1 + .../src/bootstrap/BootstrapEventBuffer.ts | 4 +- .../src/events/IReporterEventEnvelope.ts | 3 +- .../reporter/src/events/ReporterEventType.ts | 31 +++- .../reporter/src/frontend/ReporterHost.ts | 24 ++- .../reporter/src/test/HeftIntegration.test.ts | 51 ++++++ libraries/reporter/src/test/Manager.test.ts | 13 +- .../src/test/OperationStreamEmitter.test.ts | 2 + libraries/reporter/src/test/Protocol.test.ts | 7 + .../reporter/src/test/ReporterHost.test.ts | 12 +- .../operations/OperationExecutionRecord.ts | 24 +++ .../src/logic/operations/OperationGraph.ts | 38 +++-- .../test/OperationGraphEventSink.test.ts | 156 ++++++++---------- 13 files changed, 238 insertions(+), 128 deletions(-) diff --git a/common/reviews/api/rush-reporter.api.md b/common/reviews/api/rush-reporter.api.md index 029dea99d8..087a4ca45f 100644 --- a/common/reviews/api/rush-reporter.api.md +++ b/common/reviews/api/rush-reporter.api.md @@ -819,6 +819,7 @@ export interface IReporterHostOptions { readonly manager?: ReporterManager; readonly nowMs?: () => number; readonly retentionMs?: number; + readonly supportedProtocolVersion?: IReporterProtocolVersion; } // @beta diff --git a/libraries/reporter/src/bootstrap/BootstrapEventBuffer.ts b/libraries/reporter/src/bootstrap/BootstrapEventBuffer.ts index c3789cf878..6ef91a2aa1 100644 --- a/libraries/reporter/src/bootstrap/BootstrapEventBuffer.ts +++ b/libraries/reporter/src/bootstrap/BootstrapEventBuffer.ts @@ -7,7 +7,7 @@ import { BOOTSTRAP_BUFFER_TRUNCATED_EXTENSION_NAME, encodeBootstrapEnvelope } from './BootstrapProtocol'; -import type { ReporterEventType } from '../events/ReporterEventType'; +import { isReporterEventRequired, type ReporterEventType } from '../events/ReporterEventType'; import { chunkUtf8Text } from '../utilities/chunkUtf8Text'; const TRUNCATION_NOTICE_RESERVE_BYTES: number = 512; @@ -191,7 +191,7 @@ export class BootstrapEventBuffer { */ public emit(input: IBootstrapEventInput): string { const eventId: string = `boot_${this._nextEventId++}`; - const required: boolean = input.type !== 'activityChanged'; + const required: boolean = isReporterEventRequired(input.type); const line: string = encodeBootstrapEnvelope({ eventId, sessionId: this._sessionId, diff --git a/libraries/reporter/src/events/IReporterEventEnvelope.ts b/libraries/reporter/src/events/IReporterEventEnvelope.ts index b9a964fca4..f0cd5495f7 100644 --- a/libraries/reporter/src/events/IReporterEventEnvelope.ts +++ b/libraries/reporter/src/events/IReporterEventEnvelope.ts @@ -132,7 +132,8 @@ export interface IReporterEventEnvelope { readonly privacy: ReporterPrivacyClassification; /** - * Whether this event is correctness-critical and must never be dropped. + * Whether an older same-major consumer must reject the stream if it does not + * recognize this event. Event types added in a minor version are optional. */ readonly required: boolean; diff --git a/libraries/reporter/src/events/ReporterEventType.ts b/libraries/reporter/src/events/ReporterEventType.ts index 7e7010b953..c9e28830c2 100644 --- a/libraries/reporter/src/events/ReporterEventType.ts +++ b/libraries/reporter/src/events/ReporterEventType.ts @@ -12,7 +12,7 @@ * Per-type policy (the contract the manager, log-level filters, and reporters * implement): * - * | type | never dropped | minimum log level | + * | type | required on wire | minimum log level | * | --- | --- | --- | * | `sessionStarted` | yes | `normal` | * | `sessionCompleted` | yes | `quiet` | @@ -30,8 +30,8 @@ * | `artifactAvailable` | yes | `normal` | * | `commandResult` | yes | `quiet` | * | `extension` | yes | `normal` | - * | `operationStreamClosed` | yes | `debug` | - * | `operationCompleted` | yes | `normal` | + * | `operationStreamClosed` | additive optional | `debug` | + * | `operationCompleted` | additive optional | `normal` | * * Coalescing a replaceable `activityChanged` event under queue pressure leaves * gaps in the delivered `sequence` values; gaps are legal and are not a @@ -72,19 +72,38 @@ export const REPORTER_EVENT_TYPES = [ */ export type ReporterEventType = (typeof REPORTER_EVENT_TYPES)[number]; +const REQUIRED_REPORTER_EVENT_TYPES: ReadonlySet = new Set([ + 'sessionStarted', + 'sessionCompleted', + 'commandStarted', + 'commandCompleted', + 'operationRegistered', + 'operationStatusChanged', + 'watchCycleCompleted', + 'diagnosticEmitted', + 'messageEmitted', + 'externalProcessStarted', + 'externalOutput', + 'externalProcessCompleted', + 'artifactAvailable', + 'commandResult', + 'extension' +]); + /** * Returns `true` if events of this type are correctness-critical and must * never be dropped or coalesced. * * @remarks * The manager derives the envelope `required` flag from this policy; - * producers never set it. Only `activityChanged` is replaceable — every other - * type, including `extension`, must be delivered. + * producers never set it. The required set is frozen to the protocol 1.0 event + * types so a same-major older peer can skip event types introduced by a newer + * minor version without discarding the stream. * * @param type - the event type to check * * @beta */ export function isReporterEventRequired(type: ReporterEventType): boolean { - return type !== 'activityChanged'; + return REQUIRED_REPORTER_EVENT_TYPES.has(type); } diff --git a/libraries/reporter/src/frontend/ReporterHost.ts b/libraries/reporter/src/frontend/ReporterHost.ts index 0b6acdd378..be139470bf 100644 --- a/libraries/reporter/src/frontend/ReporterHost.ts +++ b/libraries/reporter/src/frontend/ReporterHost.ts @@ -11,10 +11,7 @@ import type { ReporterEventType } from '../events/ReporterEventType'; import type { IReporterEventSink } from '../producers/IReporterEventSink'; import { REPORTER_EVENT_TYPES } from '../events/ReporterEventType'; import { ReporterManager } from '../manager/ReporterManager'; -import { - REPORTER_PROTOCOL_VERSION, - isReporterProtocolCompatible -} from '../protocol/ReporterProtocol'; +import { REPORTER_PROTOCOL_VERSION, isReporterProtocolCompatible } from '../protocol/ReporterProtocol'; import { RUSH_REPORTER_BOOTSTRAP_HANDOFF_ENV_VAR, RUSH_REPORTER_BOOTSTRAP_NONCE_ENV_VAR @@ -64,6 +61,11 @@ export interface IReporterHostOptions { * Returns the current time in milliseconds. Injectable for testing. */ readonly nowMs?: () => number; + + /** + * The protocol version supported by this host. Defaults to the current version. + */ + readonly supportedProtocolVersion?: IReporterProtocolVersion; } /** @@ -101,7 +103,12 @@ export interface IBootstrapReplayResult { * The reason no events were replayed, when a handoff path was present. * `nonce-mismatch` means the file failed authentication and was rejected. */ - readonly skipReason?: 'unreadable' | 'invalid-path' | 'nonce-mismatch' | 'invalid-event' | 'incompatible-protocol'; + readonly skipReason?: + | 'unreadable' + | 'invalid-path' + | 'nonce-mismatch' + | 'invalid-event' + | 'incompatible-protocol'; } function isRecord(value: unknown): value is Record { @@ -161,6 +168,7 @@ export class ReporterHost { private readonly _handoffDirectory: string; private readonly _retentionMs: number; private readonly _nowMs: () => number; + private readonly _supportedProtocolVersion: IReporterProtocolVersion; public constructor(options: IReporterHostOptions = {}) { this._manager = options.manager ?? new ReporterManager(); @@ -168,6 +176,7 @@ export class ReporterHost { this._handoffDirectory = options.handoffDirectory ?? os.tmpdir(); this._retentionMs = options.retentionMs ?? DEFAULT_HANDOFF_RETENTION_MS; this._nowMs = options.nowMs ?? (() => Date.now()); + this._supportedProtocolVersion = options.supportedProtocolVersion ?? REPORTER_PROTOCOL_VERSION; } /** @@ -247,10 +256,7 @@ export class ReporterHost { let skippedEventCount: number = discardedRecordCount; for (const event of events) { const protocolVersion: IReporterProtocolVersion | undefined = getProtocolVersion(event); - if ( - protocolVersion && - !isReporterProtocolCompatible(REPORTER_PROTOCOL_VERSION, protocolVersion) - ) { + if (protocolVersion && !isReporterProtocolCompatible(this._supportedProtocolVersion, protocolVersion)) { await deleteBootstrapHandoffFileAsync(handoffPath); return { direct: false, diff --git a/libraries/reporter/src/test/HeftIntegration.test.ts b/libraries/reporter/src/test/HeftIntegration.test.ts index 5d2990e33a..bd1f0d0eed 100644 --- a/libraries/reporter/src/test/HeftIntegration.test.ts +++ b/libraries/reporter/src/test/HeftIntegration.test.ts @@ -269,6 +269,57 @@ describe('HeftDescriptorHost new descriptor path', () => { expect(result.diagnostic?.code).toBe('RUSH_PROTOCOL_UPDATE_REQUIRED'); }); + it('lets a 1.0 consumer skip an unknown optional 1.1 event and continue the stream', () => { + 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: 1 }, + producerVersion: '@rushstack/heft 1.2.19', + capabilities: [], + requiredFeatures: [] + }) + ).toBe(true); + expect( + host.processChildRecord({ + protocolVersion: { major: 1, minor: 1 }, + eventId: 'future_optional', + sessionId: 'child-sess', + sequence: 1, + timestamp: '2026-01-01T00:00:00.000Z', + source: SOURCE, + privacy: 'public', + required: false, + type: 'futureMinorEvent', + payload: {} + }) + ).toBe(true); + expect( + host.processChildRecord({ + protocolVersion: { major: 1, minor: 1 }, + eventId: 'known_after_future', + sessionId: 'child-sess', + sequence: 2, + timestamp: '2026-01-01T00:00:00.001Z', + source: SOURCE, + privacy: 'public', + required: true, + type: 'commandCompleted', + payload: { commandName: 'build', exitCode: 0 } + }) + ).toBe(true); + + const result: IHeftChildResult = host.processChildRecords([]); + expect(result).toMatchObject({ accepted: true, eventCount: 1 }); + expect(forwarded.map(({ eventId }) => eventId)).toEqual(['known_after_future']); + }); + it('rejects malformed records without throwing from the streaming drain', () => { const negotiationResults: boolean[] = []; const host: HeftDescriptorHost = new HeftDescriptorHost({ diff --git a/libraries/reporter/src/test/Manager.test.ts b/libraries/reporter/src/test/Manager.test.ts index 582f88a8c2..cd77f28ecd 100644 --- a/libraries/reporter/src/test/Manager.test.ts +++ b/libraries/reporter/src/test/Manager.test.ts @@ -117,12 +117,16 @@ describe('ReporterManager ordering and assignment', () => { manager.emit(makeInput('activityChanged')); manager.emit(makeInput('messageEmitted')); manager.emit(makeInput('commandStarted')); + manager.emit(makeInput('operationStreamClosed')); + manager.emit(makeInput('operationCompleted')); await manager.flushAsync(); expect(reporter.reported.map((e: IReporterEventEnvelope) => e.required)).toEqual([ false, true, - true + true, + false, + false ]); }); @@ -152,9 +156,10 @@ describe('ReporterManager ordering and assignment', () => { manager.ingestForeignEnvelope(foreign); await manager.flushAsync(); - const byIdentity: [string, string][] = reporter.reported.map( - (e: IReporterEventEnvelope) => [e.sessionId, e.eventId] - ); + const byIdentity: [string, string][] = reporter.reported.map((e: IReporterEventEnvelope) => [ + e.sessionId, + e.eventId + ]); expect(byIdentity).toEqual([ ['sess', 'evt_1'], ['child', 'evt_1'] diff --git a/libraries/reporter/src/test/OperationStreamEmitter.test.ts b/libraries/reporter/src/test/OperationStreamEmitter.test.ts index 680cdb49e1..adb5e6c6a9 100644 --- a/libraries/reporter/src/test/OperationStreamEmitter.test.ts +++ b/libraries/reporter/src/test/OperationStreamEmitter.test.ts @@ -89,6 +89,8 @@ describe('OperationStreamEmitter', () => { }); // externalOutput is protected (never coalesced/dropped); the manager derives `required`. expect(isReporterEventRequired('externalOutput')).toBe(true); + expect(isReporterEventRequired('operationStreamClosed')).toBe(false); + expect(isReporterEventRequired('operationCompleted')).toBe(false); }); it('records silent metadata and orders close before completion', () => { diff --git a/libraries/reporter/src/test/Protocol.test.ts b/libraries/reporter/src/test/Protocol.test.ts index 8efe5138ba..587de7ae8b 100644 --- a/libraries/reporter/src/test/Protocol.test.ts +++ b/libraries/reporter/src/test/Protocol.test.ts @@ -5,6 +5,7 @@ import { REPORTER_PROTOCOL_VERSION, REPORTER_PROTOCOL_LIMITS, isReporterProtocolCompatible, + isReporterEventRequired, encodeNdjsonRecord, NdjsonDecoder, NdjsonInvalidRecordError, @@ -28,6 +29,12 @@ describe('ReporterProtocol', () => { expect(isReporterProtocolCompatible({ major: 1, minor: 0 }, { major: 1, minor: 9 })).toBe(true); expect(isReporterProtocolCompatible({ major: 1, minor: 0 }, { major: 2, minor: 0 })).toBe(false); }); + + it('marks event types added in protocol 1.1 as optional for protocol 1.0 consumers', () => { + expect(isReporterEventRequired('operationStreamClosed')).toBe(false); + expect(isReporterEventRequired('operationCompleted')).toBe(false); + expect(isReporterEventRequired('commandResult')).toBe(true); + }); }); describe('NDJSON encode/decode', () => { diff --git a/libraries/reporter/src/test/ReporterHost.test.ts b/libraries/reporter/src/test/ReporterHost.test.ts index 208c458b76..2472ef5793 100644 --- a/libraries/reporter/src/test/ReporterHost.test.ts +++ b/libraries/reporter/src/test/ReporterHost.test.ts @@ -234,7 +234,7 @@ describe('ReporterHost handoff replay', () => { }); }); - it('skips an unknown additive event and replays known events', async () => { + it('lets a 1.0 consumer skip an unknown optional 1.1 event and replay the remaining stream', async () => { await withTempDir(async (directory: string) => { const buffer: BootstrapEventBuffer = makeBuffer(); buffer.emit({ type: 'sessionStarted', payload: {} }); @@ -251,12 +251,15 @@ describe('ReporterHost handoff replay', () => { lines.splice(2, 0, JSON.stringify(unknownEvent)); await fs.promises.writeFile(handoffPath, `${lines.join('\n')}\n`); - const manager: ReporterManager = new ReporterManager(); + const manager: ReporterManager = new ReporterManager({ + protocolVersion: { major: 1, minor: 0 } + }); const reporter: RecordingReporter = new RecordingReporter(); manager.addReporter(reporter); await manager.initializeAsync(); const host: ReporterHost = new ReporterHost({ manager, + supportedProtocolVersion: { major: 1, minor: 0 }, env: { [RUSH_REPORTER_BOOTSTRAP_HANDOFF_ENV_VAR]: handoffPath, [RUSH_REPORTER_BOOTSTRAP_NONCE_ENV_VAR]: nonce @@ -267,10 +270,7 @@ describe('ReporterHost handoff replay', () => { const result: IBootstrapReplayResult = await host.replayBootstrapHandoffAsync(); await manager.flushAsync(); expect(result).toMatchObject({ replayed: true, eventCount: 2, skippedEventCount: 1 }); - expect(reporter.reported.map((event) => event.type)).toEqual([ - 'sessionStarted', - 'diagnosticEmitted' - ]); + expect(reporter.reported.map((event) => event.type)).toEqual(['sessionStarted', 'diagnosticEmitted']); }); }); diff --git a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts index 61cfe514a4..3276ab4b91 100644 --- a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts +++ b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts @@ -174,6 +174,7 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera private _stateHash: string | undefined; private _stateHashComponents: IOperationStateHashComponents | undefined; private _operationStreamClosed: boolean = false; + private _operationCompleted: boolean = false; public constructor(operation: Operation, context: IOperationExecutionRecordContext) { const { runner, associatedPhase, associatedProject, enabled } = operation; @@ -296,6 +297,28 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera } } + /** + * Emits the ordered terminal stream events exactly once. + * + * @internal + */ + public finalizeOperation(): void { + this.closeOperationStream(); + if (!this._operationCompleted) { + this._operationCompleted = true; + this._context.eventSink?.onOperationCompleted?.(this); + } + } + + /** + * Whether this record has emitted its terminal completion event. + * + * @internal + */ + public get isOperationCompleted(): boolean { + return this._operationCompleted; + } + /** * Adds the reporter's lossless operation-output tap ahead of any legacy presentation transforms. * @@ -509,6 +532,7 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera } finally { if (this.isTerminal) { this._collatedWriter?.close(); + this.finalizeOperation(); this.stdioSummarizer.close(); this.problemCollector.close(); } diff --git a/libraries/rush-lib/src/logic/operations/OperationGraph.ts b/libraries/rush-lib/src/logic/operations/OperationGraph.ts index 22def845d9..3d68986253 100644 --- a/libraries/rush-lib/src/logic/operations/OperationGraph.ts +++ b/libraries/rush-lib/src/logic/operations/OperationGraph.ts @@ -677,7 +677,6 @@ export class OperationGraph implements IOperationGraph { iterationContext ); executionRecords.set(operation, executionRecord); - executionRecords.set(operation, executionRecord); } for (const [operation, record] of executionRecords) { @@ -800,6 +799,7 @@ export class OperationGraph implements IOperationGraph { this._setStatus(OperationStatus.Executing); const { hooks } = this; + const graph: OperationGraph = this; const { abortController, records: executionRecords, terminal, totalOperations } = iterationContext; @@ -886,12 +886,14 @@ export class OperationGraph implements IOperationGraph { async () => await hooks.beforeExecuteIterationAsync.promise(executionRecords, iterationOptions) ); } catch (error) { + await closeRunnersAndReportFailuresAsync( + [...executionRecords.values()].filter((record) => !record.shouldRunnerPersist) + ); for (const record of executionRecords.values()) { if (!record.isTerminal) { record.status = OperationStatus.Aborted; } - record.closeOperationStream(); - eventSink?.onOperationCompleted?.(record); + record.finalizeOperation(); record.stdioSummarizer.close(); record.problemCollector.close(); } @@ -944,21 +946,20 @@ export class OperationGraph implements IOperationGraph { }); } - const recordsToClose: OperationExecutionRecord[] = []; - for (const record of executionRecords.values()) { - if (!record.shouldRunnerPersist) { - recordsToClose.push(record); - } - } function reportRunnerCleanupFailure(record: OperationExecutionRecord, error: Error): void { record.error = error; record.status = OperationStatus.Failure; _reportOperationErrorIfAny(record); state.hasAnyFailures = true; } - if (recordsToClose.length > 0) { + async function closeRunnersAndReportFailuresAsync( + recordsToClose: readonly OperationExecutionRecord[] + ): Promise { + if (recordsToClose.length === 0) { + return; + } try { - await this.closeRunnersAsync(recordsToClose.map((record) => record.operation)); + await graph.closeRunnersAsync(recordsToClose.map((record) => record.operation)); } catch (e) { if (e instanceof AggregateError) { for (const error of e.errors) { @@ -976,9 +977,17 @@ export class OperationGraph implements IOperationGraph { } } } + const incompleteRecordsToClose: OperationExecutionRecord[] = []; + for (const record of executionRecords.values()) { + if (!record.shouldRunnerPersist && !record.isOperationCompleted) { + incompleteRecordsToClose.push(record); + } + } + await closeRunnersAndReportFailuresAsync(incompleteRecordsToClose); for (const record of executionRecords.values()) { - record.closeOperationStream(); - eventSink?.onOperationCompleted?.(record); + if (!record.isOperationCompleted) { + record.finalizeOperation(); + } record.stdioSummarizer.close(); record.problemCollector.close(); } @@ -1137,6 +1146,9 @@ export class OperationGraph implements IOperationGraph { record.error = e; record.status = OperationStatus.Failure; } + if (!record.shouldRunnerPersist) { + await closeRunnersAndReportFailuresAsync([record]); + } _onOperationComplete(record, state); } } diff --git a/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts b/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts index e11aa7b767..90e77890ce 100644 --- a/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts +++ b/libraries/rush-lib/src/logic/operations/test/OperationGraphEventSink.test.ts @@ -224,6 +224,73 @@ describe('OperationGraph event sink (dual-emit)', () => { expect(mockWritable.getAllOutput()).not.toContain('quiet-hidden-stdout'); }); + it('emits terminal events before a dependent operation starts', async () => { + const sink: RecordingSink = new RecordingSink(); + const first: Operation = createOperation( + 'first', + new MockOperationRunner('first', async () => OperationStatus.Success) + ); + let firstWasFinalized: boolean = false; + const second: Operation = createOperation( + 'second', + new MockOperationRunner('second', async () => { + firstWasFinalized = + sink.closed.filter((name) => name === 'first').length === 1 && + sink.completed.filter(([name]) => name === 'first').length === 1; + return OperationStatus.Success; + }) + ); + second.addDependency(first); + const graph: OperationGraph = new OperationGraph( + new Set([first, second]), + createGraphOptions(mockWritable, false) + ); + graph.eventSink = sink; + + await graph.executeAsync({}); + + expect(firstWasFinalized).toBe(true); + expect(sink.closed).toEqual(['first', 'second']); + expect(sink.completed).toEqual([ + ['first', OperationStatus.Success], + ['second', OperationStatus.Success] + ]); + }); + + it('emits the runner cleanup failure as the single authoritative completion', async () => { + const sink: RecordingSink = new RecordingSink(); + const runner: IOperationRunner = { + name: 'cleanup failure', + reportTiming: true, + silent: false, + cacheable: false, + warningsAreAllowed: false, + isNoOp: false, + executeAsync: async () => OperationStatus.Success, + closeAsync: async () => { + throw new Error('cleanup failed'); + }, + getConfigHash: () => 'cleanup-failure' + }; + const operation: Operation = createOperation('cleanup failure', runner); + const graph: OperationGraph = new OperationGraph( + new Set([operation]), + createGraphOptions(mockWritable, false) + ); + graph.eventSink = sink; + graph.hooks.configureIteration.tap('test', (records) => { + for (const record of records.values()) { + record.shouldRunnerPersist = false; + } + }); + + const result = await graph.executeAsync({}); + + expect(result.status).toBe(OperationStatus.Failure); + expect(sink.closed).toEqual(['cleanup failure']); + expect(sink.completed).toEqual([['cleanup failure', OperationStatus.Failure]]); + }); + it('leaves terminal output byte-identical whether or not a sink is attached', async () => { const makeRunner: () => MockOperationRunner = () => new MockOperationRunner('echo', async (terminal: CollatedTerminal) => { @@ -402,92 +469,6 @@ describe('OperationGraph event sink (dual-emit)', () => { }); }); - it('combines sharded implementation records into one project x phase stream', async () => { - const reporterSink: CapturingReporterSink = new CapturingReporterSink(); - const rushSession: RushSession = new RushSession({ - terminalProvider: new StringBufferTerminalProvider(), - getIsDebugMode: () => false, - reporter: { - eventSink: reporterSink, - sessionId: 'sharded-operation-stream', - operationStreamEnabled: true - } - }); - const projectName: string = '@scope/sharded'; - const preShardRunner: IOperationRunner = { - name: `${projectName} (phase) - pre-shard`, - reportTiming: false, - silent: true, - cacheable: false, - warningsAreAllowed: false, - isNoOp: true, - executeAsync: async () => OperationStatus.NoOp, - getConfigHash: () => 'pre-shard' - }; - const shardRunner: IOperationRunner = { - name: `${projectName} (phase) - shard 1/1`, - reportTiming: true, - silent: false, - cacheable: false, - warningsAreAllowed: false, - isNoOp: false, - executeAsync: async (context: IOperationRunnerContext) => - await context.runWithTerminalAsync( - async (terminal) => { - terminal.write('shard output without newline'); - return OperationStatus.Failure; - }, - { createLogFile: false, logFileSuffix: '' } - ), - getConfigHash: () => 'shard' - }; - const collatorRunner: IOperationRunner = { - name: `${projectName} (phase) - collate`, - reportTiming: true, - silent: false, - cacheable: false, - warningsAreAllowed: false, - isNoOp: false, - executeAsync: async () => OperationStatus.Success, - getConfigHash: () => 'collate' - }; - const preShard: Operation = createOperation('pre-shard', preShardRunner, mockPhase, projectName); - const shard: Operation = createOperation('shard', shardRunner, mockPhase, projectName); - const collator: Operation = createOperation('collator', collatorRunner, mockPhase, projectName); - shard.addDependency(preShard); - collator.addDependency(shard); - const graph: OperationGraph = new OperationGraph( - new Set([collator, preShard, shard]), - createGraphOptions(mockWritable, false) - ); - - attachReporterOperationEventSink(graph, rushSession, 'build'); - await graph.executeAsync({}); - - const operationEvents: IReporterEmitEventInput[] = reporterSink.inputs.filter( - ({ scope }) => scope?.operationId === `${projectName}#phase` - ); - expect(operationEvents.filter(({ type }) => type === 'operationRegistered')).toHaveLength(1); - expect( - operationEvents - .filter(({ type }) => type === 'externalOutput') - .map(({ payload }) => (payload as { text: string }).text) - .join('') - ).toBe('shard output without newline'); - expect(operationEvents.filter(({ type }) => type === 'operationStreamClosed')).toHaveLength(1); - expect(operationEvents.filter(({ type }) => type === 'operationCompleted')).toEqual([ - expect.objectContaining({ - payload: expect.objectContaining({ operationId: `${projectName}#phase`, status: 'failure' }) - }) - ]); - expect( - operationEvents.filter( - ({ type, payload }) => - type === 'diagnosticEmitted' && (payload as { code?: string }).code === 'RUSH_OPERATION_FAILED' - ) - ).toHaveLength(1); - }); - it('does not register operations when scheduling hooks reject the iteration', async () => { const sink: RecordingSink = new RecordingSink(); const graph: OperationGraph = new OperationGraph( @@ -802,10 +783,11 @@ describe('OperationGraph event sink (dual-emit)', () => { reporterSink.inputs .filter(({ type }) => type === 'operationCompleted') .map(({ scope }) => scope?.operationId) + .sort() ).toEqual([ '@scope/project#_phase:compile', - '@scope/project#_phase:test', '@scope/project#_phase:compile', + '@scope/project#_phase:test', '@scope/project#_phase:test' ]); for (const event of reporterSink.inputs.filter(({ type }) => type === 'operationStatusChanged')) {