fix: never let a failed preempt subscription reject into the void

registerPreemptSubscription is called detached at both sites, so a
rejected Redis SUBSCRIBE became an unhandled rejection — process-fatal
under Node's default --unhandled-rejections=throw. The comment already
promised this path merely degrades steering; it now does.

Swallowed and logged inside the registration rather than at each call
site, so a future third caller cannot reintroduce the trap. Losing the
channel costs this generation's cross-replica preempts, not the server:
same-replica arming is runtime state and still works, and remote arms
fall back to the next tool boundary.

Verified counterfactually — the new spec surfaces SUBSCRIBE failed as an
unhandled rejection against the unfixed registration.

packages/api/src/stream/GenerationJobManager.ts

@@ -658,6 +658,13 @@ class GenerationJobManagerClass {
    * cross-replica publish can never arm a replacement job on the same
    * streamId. Arm requests cap at {@link STEER_QUEUE_MAX_DEPTH}, matching
    * the durable queue they mirror.
+   *
+   * Never rejects. Every caller fires this without awaiting (see the
+   * createJob registration for why), so a propagating subscription error
+   * would be an unhandled rejection — fatal under Node's default
+   * `--unhandled-rejections=throw`. A failed subscription is not worth a
+   * process: it degrades this generation's preemptive steers to the next
+   * tool boundary, which is the documented fallback.
    */
   private async registerPreemptSubscription(
     streamId: string,
@@ -667,25 +674,33 @@ class GenerationJobManagerClass {
       return;
     }
 
-    const unsubscribe = await this.eventTransport.onPreempt(streamId, (msg) => {
-      const currentRuntime = this.runtimeState.get(streamId);
-      if (currentRuntime !== runtime || currentRuntime.createdAt !== msg.createdAt) {
-        return;
+    try {
+      const unsubscribe = await this.eventTransport.onPreempt(streamId, (msg) => {
+        const currentRuntime = this.runtimeState.get(streamId);
+        if (currentRuntime !== runtime || currentRuntime.createdAt !== msg.createdAt) {
+          return;
+        }
+
+        if (msg.op === 'clear') {
+          this.clearPreemptIds(currentRuntime, msg.createdAt, msg.steerIds);
+          return;
+        }
+
+        this.armPreemptIds(currentRuntime, msg.createdAt, msg.steerIds);
+      });
+
+      if (typeof unsubscribe === 'function') {
+        runtime.preemptUnsubscribe = unsubscribe;
       }
-
-      if (msg.op === 'clear') {
-        this.clearPreemptIds(currentRuntime, msg.createdAt, msg.steerIds);
-        return;
+      if (this.runtimeState.get(streamId) !== runtime || runtime.abortController.signal.aborted) {
+        this.releasePreemptSubscription(runtime);
       }
-
-      this.armPreemptIds(currentRuntime, msg.createdAt, msg.steerIds);
-    });
-
-    if (typeof unsubscribe === 'function') {
-      runtime.preemptUnsubscribe = unsubscribe;
-    }
-    if (this.runtimeState.get(streamId) !== runtime || runtime.abortController.signal.aborted) {
-      this.releasePreemptSubscription(runtime);
+    } catch (err) {
+      logger.error(
+        `[GenerationJobManager] Failed to subscribe to preempts for ${streamId}; ` +
+          'steers on this generation will apply at the next tool boundary:',
+        err,
+      );
     }
   }
 
@@ -914,7 +929,8 @@ class GenerationJobManagerClass {
        * documented fallback, so blocking job creation on a second channel
        * subscription would trade a real hang risk for a cosmetic guarantee.
        * The registration's own lost-race tail releases it if the runtime is
-       * retired before the subscription resolves.
+       * retired before the subscription resolves, and it swallows and logs
+       * its own failures, so this detached call cannot reject.
        */
       void this.registerPreemptSubscription(streamId, runtime);
       if (this.runtimeState.get(streamId) !== runtime) {

packages/api/src/stream/__tests__/steering.spec.ts

@@ -1,6 +1,7 @@
+import { logger } from '@librechat/data-schemas';
 import { SteerEvents } from 'librechat-data-provider';
 import type { TPendingSteer, Agents } from 'librechat-data-provider';
-import type { SteerQueueItem } from '~/stream/interfaces/IJobStore';
+import type { SteerQueueItem, IEventTransport } from '~/stream/interfaces/IJobStore';
 import type { ResumeState, ServerSentEvent } from '~/types';
 import {
   STEER_ENQUEUE_NOT_RUNNING,
@@ -916,6 +917,50 @@ describe('preempt request lifecycle (in-memory)', () => {
     expect((await manager.getJob(streamId))?.status).toBe('running');
   });
 
+  test('a failing preempt subscription degrades the job instead of crashing the process', async () => {
+    const transport: IEventTransport = new InMemoryEventTransport();
+    transport.onPreempt = jest.fn().mockRejectedValue(new Error('SUBSCRIBE failed'));
+
+    /**
+     * The registration is deliberately detached, so a rejection propagating out
+     * of it would be unhandled — fatal under Node's default settings.
+     */
+    const unhandled: unknown[] = [];
+    const collect = (reason: unknown): number => unhandled.push(reason);
+    process.on('unhandledRejection', collect);
+    const logged = jest.spyOn(logger, 'error').mockImplementation(() => logger);
+
+    const degraded = new GenerationJobManagerClass();
+    degraded.configure({
+      jobStore: new InMemoryJobStore({ ttlAfterComplete: 60000 }),
+      eventTransport: transport,
+      isRedis: false,
+      cleanupOnComplete: false,
+    });
+    degraded.initialize();
+
+    try {
+      const streamId = 'preempt-subscribe-fails';
+      const job = await degraded.createJob(streamId, 'user-1');
+      await new Promise((resolve) => setImmediate(resolve));
+
+      expect(unhandled).toEqual([]);
+      expect(job.status).toBe('running');
+      expect(logged).toHaveBeenCalledWith(
+        expect.stringContaining('Failed to subscribe to preempts'),
+        expect.any(Error),
+      );
+
+      /** Same-replica arming is runtime state, so it survives the lost channel. */
+      degraded.requestPreempt(streamId, 'steer-1', job.createdAt);
+      expect(degraded.isPreemptRequested(streamId)).toBe(true);
+    } finally {
+      process.off('unhandledRejection', collect);
+      logged.mockRestore();
+      await degraded.destroy();
+    }
+  });
+
   test('abortJob retires the armed set with the runtime', async () => {
     const streamId = 'preempt-abort';
     const job = await manager.createJob(streamId, 'user-1');