Replace StreamCollator with raw semantic operation events

Add the uncollated operation event stream for @rushstack/reporter (#5858).

- Add OperationStreamEmitter so the scheduler emits operation registration,
  status transitions, raw output chunks, and the aggregate command result
- Emit output chunks immediately in call order and never collate them, so the
  concise reporter derives activity without buffering while the detailed and
  file reporters own grouping
- Add iterateExternalOutput and regroupOperationOutput so problem matchers
  consume the uncollated stream and reporters reconstruct StreamCollator-parity
  grouping
- Cover emission, chunking, uncollated ordering, and reporter parity with tests

Assistant-model: GitHub Copilot
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 897dcf7e-e6e8-4a84-85ca-34b93fa29be3

Author: TheLarkInn

common/reviews/api/reporter.api.md

@@ -365,6 +365,13 @@ export interface IEngineSinkResolution {
     readonly sink: IReporterEventSink;
 }
 
+// @beta
+export interface IExternalOutputChunk {
+    readonly operationId?: string;
+    readonly stream: string;
+    readonly text: string;
+}
+
 // @beta
 export interface IFileReporterArtifact {
     readonly available: boolean;
@@ -454,6 +461,16 @@ export interface IOperationStatusChangedPayload {
     readonly status: OperationStatus;
 }
 
+// @beta
+export interface IOperationStreamEmitterOptions {
+    readonly maxChunkBytes?: number;
+    readonly protocolVersion?: IReporterProtocolVersion;
+    readonly scope?: IReporterEventScope;
+    readonly sessionId: string;
+    readonly sink: IReporterEventSink;
+    readonly source: IReporterEventSource;
+}
+
 // @beta
 export interface IPlaintextReporterOptions {
     readonly color?: boolean;
@@ -827,6 +844,9 @@ export interface ITelemetryAggregate {
     readonly result?: TelemetryResult;
 }
 
+// @beta
+export function iterateExternalOutput(events: readonly IReporterEventEnvelope<unknown>[]): IExternalOutputChunk[];
+
 // @beta
 export interface IWatchCycleCompletedPayload {
     readonly changedProjects?: readonly string[];
@@ -934,6 +954,17 @@ export class OldEngineOutputAdapter {
 // @beta
 export type OperationStatus = 'ready' | 'executing' | 'success' | 'successWithWarnings' | 'failure' | 'blocked' | 'skipped' | 'fromCache' | 'noOp';
 
+// @beta
+export class OperationStreamEmitter {
+    constructor(options: IOperationStreamEmitterOptions);
+    changeStatus(operationId: string, status: OperationStatus, durationMs?: number): string;
+    completeCommand(commandName: string, succeeded: boolean, exitCode: number, operationCounts?: {
+        readonly [status: string]: number;
+    }): string;
+    registerOperation(operationId: string, projectName?: string, phaseName?: string): string;
+    writeOutput(operationId: string, stream: 'stdout' | 'stderr', text: string): string[];
+}
+
 // @beta
 export function parseEarlyReporterControls(argv: readonly string[], env: Record<string, string | undefined>): IEarlyReporterControls;
 
@@ -965,6 +996,9 @@ export function planAutomaticReporters(selection: IReporterSelection): IAutomati
 // @beta
 export function readBootstrapHandoffFileAsync(filePath: string): Promise<unknown[]>;
 
+// @beta
+export function regroupOperationOutput(events: readonly IReporterEventEnvelope<unknown>[]): Map<string, string>;
+
 // @beta
 export function renderActiveProjectsRow(projects: readonly string[], width: number): string;
 

libraries/reporter/src/index.ts

@@ -241,6 +241,11 @@ export {
   isLegacyEmergencyFallbackRequested
 } from './reporters/LegacyReporter';
 
+export type { IOperationStreamEmitterOptions } from './scheduler/OperationStreamEmitter';
+export { OperationStreamEmitter } from './scheduler/OperationStreamEmitter';
+export type { IExternalOutputChunk } from './scheduler/OperationOutputGrouping';
+export { iterateExternalOutput, regroupOperationOutput } from './scheduler/OperationOutputGrouping';
+
 export type { IReporterEmitEventInput, IReporterEventSink } from './producers/IReporterEventSink';
 export type {
   ReporterMessageSeverity,

libraries/reporter/src/scheduler/OperationOutputGrouping.ts

@@ -0,0 +1,81 @@
+// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license.
+// See LICENSE in the project root for license information.
+
+import type { IReporterEventEnvelope } from '../events/IReporterEventEnvelope';
+
+/**
+ * A single raw external-output chunk in the uncollated stream.
+ *
+ * @beta
+ */
+export interface IExternalOutputChunk {
+  /**
+   * The operation the chunk belongs to, when scoped.
+   */
+  readonly operationId?: string;
+
+  /**
+   * The originating stream.
+   */
+  readonly stream: string;
+
+  /**
+   * The raw text.
+   */
+  readonly text: string;
+}
+
+/**
+ * Extracts the uncollated external-output chunks from an event stream, in order.
+ *
+ * @remarks
+ * Problem matchers consume this uncollated source stream directly.
+ *
+ * @param events - the event stream
+ *
+ * @beta
+ */
+export function iterateExternalOutput(
+  events: readonly IReporterEventEnvelope<unknown>[]
+): IExternalOutputChunk[] {
+  const chunks: IExternalOutputChunk[] = [];
+  for (const event of events) {
+    if (event.type === 'externalOutput') {
+      const payload: { stream?: string; text?: string } = event.payload as {
+        stream?: string;
+        text?: string;
+      };
+      chunks.push({
+        operationId: event.scope?.operationId,
+        stream: payload.stream ?? 'stdout',
+        text: payload.text ?? ''
+      });
+    }
+  }
+  return chunks;
+}
+
+/**
+ * Regroups the uncollated external output by operation, reconstructing the
+ * per-operation ordering that StreamCollator produced.
+ *
+ * @remarks
+ * The detailed and file reporters use this to own grouping and buffering,
+ * achieving parity with StreamCollator from the uncollated stream.
+ *
+ * @param events - the event stream
+ *
+ * @beta
+ */
+export function regroupOperationOutput(
+  events: readonly IReporterEventEnvelope<unknown>[]
+): Map<string, string> {
+  const groups: Map<string, string> = new Map();
+  for (const chunk of iterateExternalOutput(events)) {
+    if (chunk.operationId === undefined) {
+      continue;
+    }
+    groups.set(chunk.operationId, (groups.get(chunk.operationId) ?? '') + chunk.text);
+  }
+  return groups;
+}

libraries/reporter/src/scheduler/OperationStreamEmitter.ts

@@ -0,0 +1,165 @@
+// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license.
+// See LICENSE in the project root for license information.
+
+import type { IReporterProtocolVersion } from '../events/ReporterProtocolVersion';
+import type { IReporterEventScope, IReporterEventSource } from '../events/IReporterEventEnvelope';
+import type { IReporterEventSink } from '../producers/IReporterEventSink';
+import type { OperationStatus } from '../lifecycle/LifecycleEvents';
+import { REPORTER_PROTOCOL_VERSION, REPORTER_PROTOCOL_LIMITS } from '../protocol/ReporterProtocol';
+
+/**
+ * Options for constructing an {@link OperationStreamEmitter}.
+ *
+ * @beta
+ */
+export interface IOperationStreamEmitterOptions {
+  /**
+   * The sink events are emitted into.
+   */
+  readonly sink: IReporterEventSink;
+
+  /**
+   * The session id stamped onto emitted events.
+   */
+  readonly sessionId: string;
+
+  /**
+   * The producer identity stamped onto emitted events.
+   */
+  readonly source: IReporterEventSource;
+
+  /**
+   * The base command scope merged into every emitted event.
+   */
+  readonly scope?: IReporterEventScope;
+
+  /**
+   * The protocol version stamped onto emitted events.
+   */
+  readonly protocolVersion?: IReporterProtocolVersion;
+
+  /**
+   * The maximum external-output chunk size in bytes. Defaults to 64 KiB.
+   */
+  readonly maxChunkBytes?: number;
+}
+
+/**
+ * Emits the raw, uncollated semantic events that replace StreamCollator.
+ *
+ * @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.
+ *
+ * @beta
+ */
+export class OperationStreamEmitter {
+  private readonly _sink: IReporterEventSink;
+  private readonly _sessionId: string;
+  private readonly _source: IReporterEventSource;
+  private readonly _scope: IReporterEventScope | undefined;
+  private readonly _protocolVersion: IReporterProtocolVersion;
+  private readonly _maxChunkBytes: number;
+
+  public constructor(options: IOperationStreamEmitterOptions) {
+    this._sink = options.sink;
+    this._sessionId = options.sessionId;
+    this._source = options.source;
+    this._scope = options.scope;
+    this._protocolVersion = options.protocolVersion ?? REPORTER_PROTOCOL_VERSION;
+    this._maxChunkBytes = options.maxChunkBytes ?? REPORTER_PROTOCOL_LIMITS.externalOutputChunkBytes;
+  }
+
+  /**
+   * Emits an operation registration event.
+   */
+  public registerOperation(operationId: string, projectName?: string, phaseName?: string): string {
+    return this._emit(
+      'operationRegistered',
+      { operationId, projectName, phaseName },
+      { operationId, projectName, phaseName },
+      'public',
+      true
+    );
+  }
+
+  /**
+   * Emits an operation status transition.
+   */
+  public changeStatus(operationId: string, status: OperationStatus, durationMs?: number): string {
+    return this._emit(
+      'operationStatusChanged',
+      { operationId, status, durationMs },
+      { operationId },
+      'public',
+      true
+    );
+  }
+
+  /**
+   * Emits raw operation output as one or more uncollated `externalOutput` chunks.
+   *
+   * @param operationId - the originating operation
+   * @param stream - the originating stream
+   * @param text - the raw output text
+   * @returns the emitted event ids
+   */
+  public writeOutput(operationId: string, stream: 'stdout' | 'stderr', text: string): string[] {
+    const eventIds: string[] = [];
+    let offset: number = 0;
+    while (offset < text.length) {
+      let end: number = text.length;
+      while (Buffer.byteLength(text.slice(offset, end), 'utf8') > this._maxChunkBytes) {
+        end = offset + Math.floor((end - offset) / 2);
+      }
+      const chunk: string = text.slice(offset, end === offset ? offset + 1 : end);
+      eventIds.push(
+        this._emit('externalOutput', { stream, text: chunk }, { operationId }, 'local-sensitive', false)
+      );
+      offset += chunk.length;
+    }
+    return eventIds;
+  }
+
+  /**
+   * Emits the aggregate command result.
+   */
+  public completeCommand(
+    commandName: string,
+    succeeded: boolean,
+    exitCode: number,
+    operationCounts?: { readonly [status: string]: number }
+  ): string {
+    return this._emit(
+      'commandResult',
+      { commandName, succeeded, exitCode, operationCounts },
+      { commandName },
+      'public',
+      true
+    );
+  }
+
+  private _emit(
+    type: 'operationRegistered' | 'operationStatusChanged' | 'externalOutput' | 'commandResult',
+    payload: unknown,
+    scopeOverride: IReporterEventScope,
+    privacy: 'public' | 'local-sensitive' | 'secret',
+    required: boolean
+  ): string {
+    const scope: IReporterEventScope = { ...this._scope, ...scopeOverride };
+    return this._sink.emit({
+      protocolVersion: this._protocolVersion,
+      sessionId: this._sessionId,
+      source: this._source,
+      scope,
+      privacy,
+      required,
+      type,
+      payload
+    });
+  }
+}