Add feature-flagged operation event adapter

apps/rush/src/RushFrontend.ts

@@ -162,7 +162,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';
@@ -77,6 +78,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[];
@@ -118,6 +126,42 @@ class LogLevelReporter implements IReporter {
   }
 }
 
+/**
+ * Keeps operation presentation on the legacy collator until R5B transfers terminal ownership.
+ * Output without an operation scope remains owned by the primary reporter.
+ */
+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) ||
+      (event.type === 'externalOutput' && event.scope?.operationId === undefined)
+    ) {
+      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;
 
@@ -675,7 +719,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

@@ -179,7 +179,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();
@@ -208,7 +208,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');
@@ -250,7 +251,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

@@ -5,7 +5,11 @@ import * as fs from 'node:fs';
 import * as os from 'node:os';
 import * as path from 'node:path';
 
-import type { IReporterEventSink } from '@rushstack/rush-reporter';
+import {
+  OldEngineOutputAdapter,
+  type IReporterEventEnvelope,
+  type IReporterEventSink
+} from '@rushstack/rush-reporter';
 
 import {
   initializeRushReporterHostAsync,
@@ -44,6 +48,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 [
@@ -648,7 +691,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,
@@ -660,14 +708,86 @@ 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 });
     }
   });
+
+  it.each(['json', 'plaintext'])(
+    'preserves unscoped and command-scoped output while deferring collated operations: %s',
+    async (reporter) => {
+      let output: string = '';
+      const initialized = await initializeRushReporterHostAsync({
+        argv: ['build', `--reporter=${reporter}`, '--log-level=debug'],
+        env: { CI: 'true' },
+        stdout: { isTTY: false, write: (text: string) => (output += text) },
+        includeDefaultFileReporter: false
+      });
+      const adapter: OldEngineOutputAdapter = new OldEngineOutputAdapter({
+        sink: initialized.sink,
+        sessionId: 'session',
+        source: { packageName: '@microsoft/rush-lib', packageVersion: '5.178.1' }
+      });
+      try {
+        emitCommandStarted(initialized.sink);
+        adapter.capture('stdout', 'bootstrap stdout\n');
+        emitOperationEvents(initialized.sink);
+        adapter.capture('stderr', 'bootstrap stderr\n');
+        initialized.sink.emit({
+          protocolVersion: { major: 1, minor: 1 },
+          sessionId: 'session',
+          source: { packageName: '@microsoft/rush-lib', packageVersion: '5.178.1' },
+          scope: { commandName: 'build' },
+          privacy: 'local-sensitive',
+          type: 'externalOutput',
+          payload: { stream: 'stdout', text: 'command output\n' }
+        });
+        await initialized.closeAsync();
+
+        if (reporter === 'json') {
+          const events: IReporterEventEnvelope<{ stream?: string; text?: string }>[] = output
+            .trim()
+            .split('\n')
+            .map((line) => JSON.parse(line));
+          expect(events.map((event) => event.type)).toEqual([
+            'commandStarted',
+            'externalOutput',
+            'externalOutput',
+            'externalOutput'
+          ]);
+          expect(events.slice(1).map((event) => event.payload)).toEqual([
+            { stream: 'stdout', text: 'bootstrap stdout\n' },
+            { stream: 'stderr', text: 'bootstrap stderr\n' },
+            { stream: 'stdout', text: 'command output\n' }
+          ]);
+        } else {
+          expect(output).toBe('Starting "rush build"\nbootstrap stdout\nbootstrap stderr\ncommand output\n');
+        }
+      } finally {
+        await initialized.closeAsync();
+      }
+    }
+  );
 });

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/@microsoft/rush/reporter-unscoped-output_2026-09-10.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@microsoft/rush",
+      "comment": "Preserve unscoped and command-scoped external output in the primary reporter while operation output remains owned by the legacy collator.",
+      "type": "none"
+    }
+  ],
+  "packageName": "@microsoft/rush"
+}

common/changes/@rushstack/rush-daemon/reporter-operation-forwarding_2026-09-07.json

@@ -0,0 +1,11 @@
+{
+  "changes": [
+    {
+      "packageName": "@rushstack/rush-daemon",
+      "comment": "Forward reporter operation completion and iteration identity through phased request event multiplexing.",
+      "type": "patch"
+    }
+  ],
+  "packageName": "@rushstack/rush-daemon",
+  "email": "223556219+Copilot@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

@@ -683,11 +683,12 @@ export interface IOperationGraphContext extends ICreateOperationsContext {
 // @internal
 export interface _IOperationGraphEventSink {
     onActivity?(text: string, options?: _IOperationActivityOptions): void;
-    onOperationChunk?(operationId: string, chunk: ITerminalChunk): void;
+    onOperationChunk?(operationId: string, chunk: ITerminalChunk, result?: IOperationExecutionResult): void;
+    onOperationCompleted?(result: IOperationExecutionResult): void;
     onOperationHeader?(operationId: string, completedOperations: number, totalOperations: number): void;
     onOperationRegistered?(operationId: string, silent: boolean, result?: IOperationExecutionResult): void;
     onOperationStatusChanged?(result: IOperationExecutionResult, previousStatus: OperationStatus): void;
-    onOperationStreamClosed?(operationId: string): void;
+    onOperationStreamClosed?(operationId: string, result?: IOperationExecutionResult): void;
 }
 
 // @alpha
@@ -1016,6 +1017,8 @@ export interface IRushSessionOptions {
 // @beta
 export interface IRushSessionReporterOptions {
     readonly eventSink: IReporterEventSink;
+    // @internal
+    readonly operationStreamEnabled?: boolean;
     readonly sessionId: string;
 }