Add feature-flagged operation event adapter

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
   };

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<string> = 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<ReporterEventType> = 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<void> {
+    return this._reporter.initializeAsync(context);
+  }
+
+  public report(event: IReporterEventEnvelope<unknown>): void {
+    if (!DEFERRED_OPERATION_EVENT_TYPES.has(event.type)) {
+      this._reporter.report(event);
+    }
+  }
+
+  public flushAsync(): Promise<void> {
+    return this._reporter.flushAsync();
+  }
+
+  public closeAsync(): Promise<void> {
+    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'
       });
     }

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']);

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<void> = 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<string, unknown>[] = stdoutText
+        .trim()
+        .split('\n')
+        .map((line: string) => JSON.parse(line) as Record<string, unknown>);
+      const fileEvents: Record<string, unknown>[] = (await fs.promises.readFile(outputPath, 'utf8'))
+        .trim()
+        .split('\n')
+        .map((line: string) => JSON.parse(line) as Record<string, unknown>);
+      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 });
     }

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"
+}

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"
+}

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;
 }
 

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;
@@ -805,6 +819,7 @@ export interface IReporterHostOptions {
     readonly manager?: ReporterManager;
     readonly nowMs?: () => number;
     readonly retentionMs?: number;
+    readonly supportedProtocolVersion?: IReporterProtocolVersion;
 }
 
 // @beta
@@ -1251,11 +1266,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 +1345,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 [];

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,

libraries/reporter/src/config/LogLevelFilter.ts

@@ -59,6 +59,7 @@ export function getEventMinimumLogLevel(event: IReporterEventEnvelope<unknown>):
     case 'sessionStarted':
     case 'commandStarted':
     case 'operationStatusChanged':
+    case 'operationCompleted':
     case 'watchCycleCompleted':
     case 'artifactAvailable':
       return 'normal';
@@ -79,6 +80,7 @@ export function getEventMinimumLogLevel(event: IReporterEventEnvelope<unknown>):
     case 'externalProcessCompleted':
       return 'verbose';
     case 'externalOutput':
+    case 'operationStreamClosed':
       return 'debug';
     case 'extension':
       return 'normal';

libraries/reporter/src/events/IReporterEventEnvelope.ts

@@ -132,7 +132,8 @@ export interface IReporterEventEnvelope<TPayload = unknown> {
   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;