[rush-daemon] Fix phased request event routing

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: mojazayeri

common/changes/@rushstack/rush-terminal-renderer/mojazayeri-authoritative-operation-headers_2026-08-25-13-53.json

@@ -0,0 +1,11 @@
+{
+  "changes": [
+    {
+      "packageName": "@rushstack/rush-terminal-renderer",
+      "comment": "Use authoritative daemon operation-header counters when collating partial warm iterations.",
+      "type": "patch"
+    }
+  ],
+  "packageName": "@rushstack/rush-terminal-renderer",
+  "email": "mojazayeri@users.noreply.github.com"
+}

common/reviews/api/rush-terminal-renderer.api.md

@@ -7,6 +7,7 @@
 import type { DaemonVerbosity } from '@rushstack/rush-daemon-protocol';
 import type { IDaemonClientCaps } from '@rushstack/rush-daemon-protocol';
 import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol';
+import type { IDaemonOperationHeaderPayload } from '@rushstack/rush-daemon-protocol';
 import { ITerminalChunk } from '@rushstack/terminal';
 import { TerminalWritable } from '@rushstack/terminal';
 
@@ -82,6 +83,7 @@ export class OperationStreamRegistry {
     constructor(options: IOperationStreamRegistryOptions);
     closeOperation(operationId: string): void;
     registerOperation(): void;
+    setOperationHeader(header: IDaemonOperationHeaderPayload): void;
     writeChunk(operationId: string, chunk: ITerminalChunk): void;
 }
 

libraries/rush-daemon/src/PhasedRequestEventSink.ts

@@ -30,7 +30,7 @@ const EVENT_SOURCE_COMPONENT: string = 'OperationGraph';
 const TEXT_ENCODER: InstanceType<typeof TextEncoder> = new TextEncoder();
 
 interface IObservedOperationResult {
-  readonly errorMessage: string | undefined;
+  readonly executionResult: IOperationExecutionResult;
   readonly status: string;
 }
 
@@ -129,7 +129,7 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
       return;
     }
     this.#observedResults.set(result.operation, {
-      errorMessage: result.error?.message,
+      executionResult: result,
       status: result.status
     });
     this.#emitEvent('operationStatusChanged', {
@@ -141,10 +141,14 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
 
   public onOperationHeader(operationId: string, completed: number, total: number): void {
     if (this.#activeOperationIds.has(operationId)) {
-      this.#emitEvent('extension', {
-        data: { completedOperations: completed, operationId, totalOperations: total },
-        name: RUSHD_OPERATION_HEADER
-      });
+      this.#emitEvent(
+        'extension',
+        {
+          data: { completedOperations: completed, operationId, totalOperations: total },
+          name: RUSHD_OPERATION_HEADER
+        },
+        { required: true }
+      );
     }
   }
 
@@ -172,13 +176,13 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
 
   public onActivity(text: string, options?: _IOperationActivityOptions): void {
     const operationId: string | undefined = options?.operationId;
-    if (!operationId || !this.#activeOperationIds.has(operationId)) {
+    if (operationId !== undefined && !this.#activeOperationIds.has(operationId)) {
       return;
     }
     this.#emitEvent(
       'activityChanged',
       { stream: options?.stderr === true ? 'stderr' : 'stdout', text },
-      { required: true, scope: { operationId } }
+      { required: true, scope: operationId === undefined ? undefined : { operationId } }
     );
   }
 

libraries/rush-daemon/src/PhasedRequestRouter.ts

@@ -358,7 +358,7 @@ function collectOperationResults(
       continue;
     }
     const errorMessage: string | undefined = observed
-      ? observed.errorMessage
+      ? observed.executionResult.error?.message
       : retained?.error?.message;
     results.push({ operationId: operation.name, status, errorMessage });
   }

libraries/rush-daemon/src/test/PhasedRequestEventSink.test.ts

@@ -0,0 +1,46 @@
+// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license.
+// See LICENSE in the project root for license information.
+
+import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol';
+
+import { PhasedRequestEventSink } from '../PhasedRequestEventSink';
+import { TestPhasedRequestClient } from './PhasedRequestRouterTestUtilities';
+
+const ACTIVE_OPERATION: string = 'project-a (_phase:test)';
+const OTHER_OPERATION: string = 'project-b (_phase:test)';
+
+function createSink(client: TestPhasedRequestClient): PhasedRequestEventSink {
+  return new PhasedRequestEventSink({
+    activeOperationIds: new Set([ACTIVE_OPERATION]),
+    client,
+    getNextSequence: () => client.getNextEventSequence(),
+    onWriteFailure: () => undefined,
+    rushVersion: '5.178.1'
+  });
+}
+
+it('forwards unscoped and active activity while filtering other operation activity', async () => {
+  const client: TestPhasedRequestClient = new TestPhasedRequestClient();
+  const sink: PhasedRequestEventSink = createSink(client);
+  sink.onActivity('request summary');
+  sink.onActivity('active detail', { operationId: ACTIVE_OPERATION });
+  sink.onActivity('other detail', { operationId: OTHER_OPERATION });
+
+  await sink.flushAsync();
+
+  const activities: IDaemonEventEnvelope[] = client.writes
+    .map(({ event }) => event)
+    .filter(
+      (event: IDaemonEventEnvelope | undefined): event is IDaemonEventEnvelope =>
+        event?.type === 'activityChanged'
+    );
+  expect(activities.map(({ payload }) => payload)).toEqual([
+    { stream: 'stdout', text: 'request summary' },
+    { stream: 'stdout', text: 'active detail' }
+  ]);
+  expect(activities.map(({ scope }) => scope)).toEqual([
+    undefined,
+    { operationId: ACTIVE_OPERATION }
+  ]);
+  expect(activities.every(({ required }) => required)).toBe(true);
+});

libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts

@@ -7,7 +7,10 @@ import type {
   IDaemonPhasedOperationSelection,
   IDaemonPhasedRequest
 } from '@rushstack/rush-daemon-protocol';
-import { RUSHD_OPERATION_STREAM_CLOSED } from '@rushstack/rush-daemon-protocol';
+import {
+  RUSHD_OPERATION_HEADER,
+  RUSHD_OPERATION_STREAM_CLOSED
+} from '@rushstack/rush-daemon-protocol';
 import { OperationStatus } from '@microsoft/rush-lib';
 
 import { PhasedRequestRouter } from '../PhasedRequestRouter';
@@ -204,7 +207,7 @@ describe(PhasedRequestRouter.name, () => {
     const fixture: ITestRoutingFixture = createThreeOperationFixture({
       actionAAsync: async (terminal: ITerminal): Promise<void> => {
         terminal.writeLine('stdout-a');
-        terminal.writeErrorLine('stderr-a');
+        terminal.writeErrorLine('stderr-a', { doNotOverrideSgrCodes: true });
       }
     });
     const client: TestPhasedRequestClient = new TestPhasedRequestClient();
@@ -528,4 +531,40 @@ describe(PhasedRequestRouter.name, () => {
     expect(firstSequences.length).toBeGreaterThan(0);
     expect(secondSequences[0]).toBeGreaterThan(firstSequences[firstSequences.length - 1]);
   });
+
+  it('uses authoritative header totals after a partial warm iteration', async () => {
+    const fixture: ITestRoutingFixture = createThreeOperationFixture();
+    let iteration: number = 0;
+    fixture.graph.hooks.configureIteration.tap('partial warm iteration', (records, previousResults) => {
+      if (iteration++ === 0) {
+        return;
+      }
+      for (const record of records.values()) {
+        if (record.operation.name === OPERATION_A && previousResults.has(record.operation)) {
+          record.enabled = false;
+        }
+      }
+    });
+    const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session);
+    await router.executeAsync(createRequest([select(OPERATION_B)]), new TestPhasedRequestClient());
+    const client: TestPhasedRequestClient = new TestPhasedRequestClient();
+
+    await router.executeAsync(
+      { ...createRequest([select(OPERATION_B)]), requestId: 'request-2' },
+      client
+    );
+
+    const headers: IDaemonEventEnvelope[] = client.writes
+      .map(({ event }) => event)
+      .filter(
+        (event: IDaemonEventEnvelope | undefined): event is IDaemonEventEnvelope =>
+          (event?.payload as { name?: unknown } | undefined)?.name === RUSHD_OPERATION_HEADER
+      );
+    expect(headers).toHaveLength(1);
+    expect(headers[0]?.required).toBe(true);
+    expect(headers[0]?.payload).toEqual({
+      data: { completedOperations: 1, operationId: OPERATION_B, totalOperations: 1 },
+      name: RUSHD_OPERATION_HEADER
+    });
+  });
 });

libraries/rush-terminal-renderer/src/HostEventRouter.ts

@@ -5,8 +5,10 @@ import {
   type DaemonVerbosity,
   type IDaemonEventEnvelope,
   type IDaemonExtensionEventPayload,
+  type IDaemonOperationHeaderPayload,
   type IDaemonOperationRegisteredPayload,
   type IDaemonOperationStreamClosedPayload,
+  RUSHD_OPERATION_HEADER,
   RUSHD_OPERATION_STREAM_CLOSED,
   shouldSerializeDaemonEvent
 } from '@rushstack/rush-daemon-protocol';
@@ -20,11 +22,7 @@ function readScopeOperationId(envelope: IDaemonEventEnvelope): string | undefine
   return scope === undefined ? undefined : scope.operationId;
 }
 
-/**
- * Routes decoded event envelopes between the collator (stream-affecting and
- * operation-scoped events) and the verbosity-filtered renderer.
- * @internal
- */
+/** Routes decoded events between the operation collator and renderer. @internal */
 export class HostEventRouter {
   private readonly _streams: OperationStreamRegistry;
   private readonly _renderer: IDaemonRenderer;
@@ -40,7 +38,6 @@ export class HostEventRouter {
     this._verbosity = verbosity;
   }
 
-  /** Routes one decoded `0x05` event envelope. */
   public routeEvent(envelope: IDaemonEventEnvelope): void {
     this._trackOperationLifecycle(envelope);
     if (this._routeScopedActivity(envelope)) {
@@ -67,16 +64,17 @@ export class HostEventRouter {
   }
 
   private _trackExtension(payload: IDaemonExtensionEventPayload): void {
+    if (payload.name === RUSHD_OPERATION_HEADER) {
+      this._streams.setOperationHeader(payload.data as IDaemonOperationHeaderPayload);
+      return;
+    }
     if (payload.name === RUSHD_OPERATION_STREAM_CLOSED) {
       const data: IDaemonOperationStreamClosedPayload =
         payload.data as IDaemonOperationStreamClosedPayload;
       this._streams.closeOperation(data.operationId);
     }
   }
 
-  // Operation-scoped activity lines are part of the operation's output block
-  // (legacy writes them to the operation's collated stream, bypassing the
-  // quiet-mode stdout discard), so they route to the collator, not the renderer.
   private _routeScopedActivity(envelope: IDaemonEventEnvelope): boolean {
     const operationId: string | undefined = readScopeOperationId(envelope);
     if (envelope.type !== 'activityChanged' || operationId === undefined) {

libraries/rush-terminal-renderer/src/OperationHeaderTracker.ts

@@ -0,0 +1,38 @@
+// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license.
+// See LICENSE in the project root for license information.
+
+import type { IDaemonOperationHeaderPayload } from '@rushstack/rush-daemon-protocol';
+
+const INITIAL_OPERATION_COUNT: number = 0;
+const OPERATION_COUNT_INCREMENT: number = 1;
+
+export class OperationHeaderTracker {
+  private readonly _headerByOperation: Map<string, IDaemonOperationHeaderPayload> = new Map();
+  private _completedOperations: number = INITIAL_OPERATION_COUNT;
+  private _totalOperations: number = INITIAL_OPERATION_COUNT;
+
+  public registerOperation(): void {
+    this._totalOperations += OPERATION_COUNT_INCREMENT;
+  }
+
+  public setOperationHeader(header: IDaemonOperationHeaderPayload): void {
+    this._headerByOperation.set(header.operationId, header);
+  }
+
+  public takeOperationHeader(operationId: string): IDaemonOperationHeaderPayload {
+    const header: IDaemonOperationHeaderPayload | undefined =
+      this._headerByOperation.get(operationId);
+    if (header !== undefined) {
+      this._headerByOperation.delete(operationId);
+      this._completedOperations = header.completedOperations;
+      this._totalOperations = header.totalOperations;
+      return header;
+    }
+    this._completedOperations += OPERATION_COUNT_INCREMENT;
+    return {
+      completedOperations: this._completedOperations,
+      operationId,
+      totalOperations: this._totalOperations
+    };
+  }
+}

libraries/rush-terminal-renderer/src/OperationStreamRegistry.ts

@@ -2,11 +2,13 @@
 // See LICENSE in the project root for license information.
 
 import { NewlineKind } from '@rushstack/node-core-library';
+import type { IDaemonOperationHeaderPayload } from '@rushstack/rush-daemon-protocol';
 import { CollatedTerminal, StreamCollator } from '@rushstack/stream-collator';
 import type { CollatedWriter } from '@rushstack/stream-collator';
 import { TextRewriterTransform } from '@rushstack/terminal';
 import type { ITerminalChunk, TerminalWritable } from '@rushstack/terminal';
 
+import { OperationHeaderTracker } from './OperationHeaderTracker';
 import { formatDaemonOperationHeader } from './RendererHeader';
 
 /** Options for {@link OperationStreamRegistry}. @beta */
@@ -29,16 +31,12 @@ export interface IOperationStreamRegistryOptions {
 export class OperationStreamRegistry {
   private readonly _collator: StreamCollator;
   private readonly _collatedTerminal: CollatedTerminal;
-  private readonly _writers: Map<string, CollatedWriter>;
+  private readonly _headers: OperationHeaderTracker = new OperationHeaderTracker();
+  private readonly _writers: Map<string, CollatedWriter> = new Map();
   private readonly _quiet: boolean;
-  private _completedOperations: number;
-  private _totalOperations: number;
 
   public constructor(options: IOperationStreamRegistryOptions) {
-    this._writers = new Map();
     this._quiet = options.quiet;
-    this._completedOperations = 0;
-    this._totalOperations = 0;
     const transform: TextRewriterTransform = new TextRewriterTransform({
       destination: options.destination,
       normalizeNewlines: NewlineKind.OsDefault,
@@ -53,7 +51,12 @@ export class OperationStreamRegistry {
 
   /** Increments the total-operation count shown in headers. */
   public registerOperation(): void {
-    this._totalOperations += 1;
+    this._headers.registerOperation();
+  }
+
+  /** Records engine-authoritative counters before an operation's stream activates. */
+  public setOperationHeader(header: IDaemonOperationHeaderPayload): void {
+    this._headers.setOperationHeader(header);
   }
 
   /** Writes one raw chunk to the operation's collated stream. */
@@ -78,11 +81,13 @@ export class OperationStreamRegistry {
     if (writer === undefined) {
       return;
     }
-    this._completedOperations += 1;
+    const counters: IDaemonOperationHeaderPayload = this._headers.takeOperationHeader(
+      writer.taskName
+    );
     const header: string = formatDaemonOperationHeader(
       writer.taskName,
-      this._completedOperations,
-      this._totalOperations
+      counters.completedOperations,
+      counters.totalOperations
     );
     this._collatedTerminal.writeStdoutLine(`\n${header}`);
     if (!this._quiet) {