feat(core): add logging and progress message types to daemon

packages/nx/src/daemon/client/client.ts

@@ -28,6 +28,11 @@ import { preventRecursionInGraphConstruction } from '../../project-graph/project
 import { ConfigurationSourceMaps } from '../../project-graph/utils/project-configuration/source-maps';
 import { parseMessage } from '../../utils/consume-messages-from-socket';
 import { DelayedSpinner } from '../../utils/delayed-spinner';
+import { globalSpinner } from '../../utils/spinner';
+import {
+  isEmitLogMessage,
+  isUpdateProgressMessage,
+} from '../message-types/streaming-messages';
 import { handleImport } from '../../utils/handle-import';
 import { isCI } from '../../utils/is-ci';
 import { isSandbox } from '../../utils/is-sandbox';
@@ -1259,6 +1264,20 @@ export class DaemonClient {
         'result-parse-start-' + this.currentMessage.type,
         'result-parse-end-' + this.currentMessage.type
       );
+      // Streaming messages fire side-effects on the client but do not
+      // resolve the pending request promise — the daemon can push several
+      // of these before finally sending the real response.
+      if (isUpdateProgressMessage(parsedResult)) {
+        if (globalSpinner.isSpinning()) {
+          globalSpinner.updateText(parsedResult.message);
+        }
+        return;
+      }
+      if (isEmitLogMessage(parsedResult)) {
+        // eslint-disable-next-line no-console
+        console[parsedResult.level](parsedResult.message);
+        return;
+      }
       if (parsedResult.error) {
         this.currentReject(parsedResult.error);
       } else {

packages/nx/src/daemon/message-types/streaming-messages.ts

@@ -0,0 +1,36 @@
+export const UPDATE_PROGRESS_MESSAGE = 'UPDATE_PROGRESS_MESSAGE' as const;
+
+export type UpdateProgressMessage = {
+  type: typeof UPDATE_PROGRESS_MESSAGE;
+  message: string;
+};
+
+export function isUpdateProgressMessage(
+  message: unknown
+): message is UpdateProgressMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === UPDATE_PROGRESS_MESSAGE
+  );
+}
+
+export const EMIT_LOG = 'EMIT_LOG' as const;
+
+export type EmitLogLevel = 'log' | 'warn' | 'error';
+
+export type EmitLogMessage = {
+  type: typeof EMIT_LOG;
+  level: EmitLogLevel;
+  message: string;
+};
+
+export function isEmitLogMessage(message: unknown): message is EmitLogMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === EMIT_LOG
+  );
+}

packages/nx/src/daemon/server/client-socket-context.ts

@@ -0,0 +1,93 @@
+import { AsyncLocalStorage } from 'async_hooks';
+import type { Socket } from 'net';
+import { MESSAGE_END_SEQ } from '../../utils/consume-messages-from-socket';
+import {
+  EMIT_LOG,
+  EmitLogLevel,
+  UPDATE_PROGRESS_MESSAGE,
+} from '../message-types/streaming-messages';
+import { isOnDaemon } from '../is-on-daemon';
+import { serverLogger } from '../logger';
+
+// Messages that stream back to the client while a request is in flight
+// (progress updates, log forwarding) need to know which socket belongs to
+// the currently-handled request. Threading a socket through every layer
+// of the handler call tree would be invasive, so we stash it in
+// AsyncLocalStorage and surface it via the helpers below.
+const clientSocketStorage = new AsyncLocalStorage<Socket>();
+
+export function runWithClientSocket<T>(
+  socket: Socket,
+  fn: () => Promise<T> | T
+): Promise<T> | T {
+  return clientSocketStorage.run(socket, fn);
+}
+
+export function getActiveClientSocket(): Socket | undefined {
+  return clientSocketStorage.getStore();
+}
+
+function assertOnDaemon(helperName: string) {
+  if (!isOnDaemon()) {
+    throw new Error(
+      `${helperName} can only be called from the Nx daemon process.`
+    );
+  }
+}
+
+function writeStreamingMessage(socket: Socket, payload: unknown) {
+  try {
+    socket.write(JSON.stringify(payload) + MESSAGE_END_SEQ, (err) => {
+      if (err) {
+        serverLogger.log(
+          `Streaming message write error (client likely disconnected): ${err.message}`
+        );
+      }
+    });
+  } catch (e) {
+    serverLogger.log(
+      `Failed to send streaming message to client: ${
+        e instanceof Error ? e.message : String(e)
+      }`
+    );
+  }
+}
+
+/**
+ * Sends a progress message to the currently-connected client, which will
+ * update the in-flight spinner on the client side. No-op when called
+ * outside of a request-handling async context (no active client socket).
+ *
+ * Must only be invoked from inside the Nx daemon process.
+ */
+export function sendProgressMessageToClient(message: string): void {
+  assertOnDaemon('sendProgressMessageToClient');
+  const socket = getActiveClientSocket();
+  if (!socket) return;
+  writeStreamingMessage(socket, {
+    type: UPDATE_PROGRESS_MESSAGE,
+    message,
+  });
+}
+
+/**
+ * Emits a log line to the currently-connected client. If invoked outside
+ * of a request-handling async context (e.g. background daemon work with
+ * no subscriber), the message is written to the daemon logger instead so
+ * it is not silently dropped.
+ *
+ * Must only be invoked from inside the Nx daemon process.
+ */
+export function emitLogToClient(level: EmitLogLevel, message: string): void {
+  assertOnDaemon('emitLogToClient');
+  const socket = getActiveClientSocket();
+  if (!socket) {
+    serverLogger.log(`[emit-log:${level}] ${message}`);
+    return;
+  }
+  writeStreamingMessage(socket, {
+    type: EMIT_LOG,
+    level,
+    message,
+  });
+}

packages/nx/src/daemon/server/server.ts

@@ -93,6 +93,7 @@ import {
   isWindows,
   killSocketOrPath,
 } from '../socket-utils';
+import { runWithClientSocket } from './client-socket-context';
 import { registerFileChangeListener } from './file-watching/file-change-events';
 import {
   hasRegisteredFileWatcherSockets,
@@ -494,7 +495,11 @@ export async function handleResult(
   let hr: HandlerResult;
   const startMark = new Date();
   try {
-    hr = await hrFn();
+    // Run the handler inside an AsyncLocalStorage context keyed on the
+    // requesting client socket. Any code path reached from here can call
+    // sendProgressMessageToClient / emitLogToClient to stream messages
+    // back to this specific client while the request is in flight.
+    hr = await runWithClientSocket(socket, () => hrFn());
   } catch (error) {
     hr = { description: `[${type}]`, error };
   }

packages/nx/src/project-graph/plugins/isolation/isolated-plugin.ts

@@ -29,9 +29,18 @@ import type {
   MessageResult,
   PluginWorkerLoadResult,
   PluginWorkerMessage,
+  PluginWorkerNotification,
   PluginWorkerResult,
 } from './messaging';
-import { isPluginWorkerResult, sendMessageOverSocket } from './messaging';
+import {
+  isPluginWorkerNotification,
+  isPluginWorkerResult,
+  sendMessageOverSocket,
+} from './messaging';
+import {
+  emitLogToClient,
+  sendProgressMessageToClient,
+} from '../../../daemon/server/client-socket-context';
 import {
   Hook,
   Phase,
@@ -213,6 +222,10 @@ export class IsolatedPlugin implements LoadedNxPlugin {
 
   private handleSocketData = (raw: string) => {
     const message = parseMessage<any>(raw);
+    if (isPluginWorkerNotification(message)) {
+      handlePluginWorkerNotification(message);
+      return;
+    }
     if (!isPluginWorkerResult(message)) {
       return;
     }
@@ -696,3 +709,28 @@ type Falsy = false | 0 | '' | null | undefined | 0n;
 function hooks(...array: Array<Hook | Falsy>): Array<Hook> {
   return array.filter((v): v is Hook => !!v);
 }
+
+// When the host process is the daemon, forward notifications to the
+// currently-connected client so log lines surface in the user's
+// terminal and progress updates reach the client-side spinner. When the
+// host is the direct CLI there is no client socket, so log lines go
+// straight to stdout/stderr; progress updates from workers have no
+// destination in that case and are dropped (the CLI's own in-process
+// plugin-loading loop is responsible for spinner updates there).
+function handlePluginWorkerNotification(
+  notification: PluginWorkerNotification
+): void {
+  if ((global as any).NX_DAEMON) {
+    if (notification.type === 'emitLog') {
+      emitLogToClient(notification.level, notification.message);
+    } else if (notification.type === 'updateProgress') {
+      sendProgressMessageToClient(notification.message);
+    }
+    return;
+  }
+  if (notification.type === 'emitLog') {
+    const stream =
+      notification.level === 'error' ? process.stderr : process.stdout;
+    stream.write(notification.message + '\n');
+  }
+}

packages/nx/src/project-graph/plugins/isolation/messaging.ts

@@ -173,6 +173,42 @@ export type MessageResult<T extends PluginWorkerMessage['type']> = ResultOf<
   T & WithResult<PluginMessageDefs>
 >;
 
+// =============================================================================
+// NOTIFICATIONS (worker -> host, unsolicited, no response expected)
+// =============================================================================
+
+export type PluginWorkerEmitLogNotification = {
+  type: 'emitLog';
+  level: 'log' | 'warn' | 'error';
+  message: string;
+};
+
+export type PluginWorkerUpdateProgressNotification = {
+  type: 'updateProgress';
+  message: string;
+};
+
+export type PluginWorkerNotification =
+  | PluginWorkerEmitLogNotification
+  | PluginWorkerUpdateProgressNotification;
+
+const NOTIFICATION_TYPES: ReadonlyArray<PluginWorkerNotification['type']> = [
+  'emitLog',
+  'updateProgress',
+];
+
+export function isPluginWorkerNotification(
+  message: Serializable
+): message is PluginWorkerNotification {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    typeof message.type === 'string' &&
+    (NOTIFICATION_TYPES as readonly string[]).includes(message.type)
+  );
+}
+
 // =============================================================================
 // TYPE GUARDS
 // =============================================================================
@@ -264,7 +300,7 @@ export async function consumeMessage(
  */
 export function sendMessageOverSocket(
   socket: Socket,
-  message: PluginWorkerMessage | PluginWorkerResult
+  message: PluginWorkerMessage | PluginWorkerResult | PluginWorkerNotification
 ): void {
   socket.write(serialize(message));
   socket.write(MESSAGE_END_SEQ);

packages/nx/src/project-graph/plugins/isolation/plugin-worker.ts

@@ -10,6 +10,7 @@ import { logger } from '../../../utils/logger';
 import { createSerializableError } from '../../../utils/serializable-error';
 import type { LoadedNxPlugin } from '../loaded-nx-plugin';
 import { consumeMessage, isPluginWorkerMessage } from './messaging';
+import { setPluginWorkerHostSocket } from './worker-streaming';
 
 import { unlinkSync } from 'fs';
 import { createServer } from 'net';
@@ -51,6 +52,11 @@ let connectErrorTimeout = setErrorTimeout(
 const server = createServer((socket) => {
   connectErrorTimeout?.clear();
 
+  // Make the host-facing socket available to plugin code running in this
+  // worker so it can emit log / progress notifications without having
+  // the socket threaded through every call site.
+  setPluginWorkerHostSocket(socket);
+
   logger.verbose(
     `[plugin-worker] "${expectedPluginName}" (pid: ${process.pid}) connected`
   );

packages/nx/src/project-graph/plugins/isolation/worker-streaming.ts

@@ -0,0 +1,54 @@
+import type { Socket } from 'net';
+import {
+  PluginWorkerEmitLogNotification,
+  PluginWorkerUpdateProgressNotification,
+  sendMessageOverSocket,
+} from './messaging';
+
+// Plugin workers talk to their host process over a single socket that is
+// established when the host connects. Plugin code running anywhere in the
+// worker process needs a way to emit log lines / progress updates without
+// having that socket threaded through every call frame, so we stash a
+// module-level reference here when the host connects.
+let hostSocket: Socket | null = null;
+
+export function setPluginWorkerHostSocket(socket: Socket): void {
+  hostSocket = socket;
+  socket.once('close', () => {
+    if (hostSocket === socket) hostSocket = null;
+  });
+}
+
+/**
+ * Emits a log line from the plugin worker up to its host. The host
+ * decides where it ends up (direct stdout/stderr when running under the
+ * CLI, forwarded to the active daemon client when running under the
+ * daemon).
+ *
+ * No-op when called outside of a plugin worker (i.e. no host connected).
+ */
+export function emitPluginWorkerLog(
+  level: PluginWorkerEmitLogNotification['level'],
+  message: string
+): void {
+  if (!hostSocket) return;
+  sendMessageOverSocket(hostSocket, {
+    type: 'emitLog',
+    level,
+    message,
+  });
+}
+
+/**
+ * Emits a progress message from the plugin worker up to its host. The
+ * host forwards it to the active daemon client (if running under the
+ * daemon) so an in-flight spinner can reflect the update. No-op when
+ * called outside of a plugin worker.
+ */
+export function emitPluginWorkerProgress(message: string): void {
+  if (!hostSocket) return;
+  sendMessageOverSocket(hostSocket, {
+    type: 'updateProgress',
+    message,
+  } satisfies PluginWorkerUpdateProgressNotification);
+}