🌐 fix: Persist pii_matches across replicas for multi-replica deployments

The warn-mode toast was emitted via GenerationJobManager.emitChunk
during pre-redaction in api/server/controllers/agents/request.js and
again from the agents-side hook callback in client.js. In a Redis
multi-replica deployment the emitting replica buffers the event in
its earlyEventBuffer; a GET /chat/stream/:streamId subscriber that
lands on a different replica has no local buffer to replay, so the
toast is dropped. Server-side redaction still scrubs the credential
from the persisted message and the LLM prompt, but the user loses
the heads-up that their input was modified.

The created event already has a parallel reconstruction path in
GenerationJobManager.subscribe for exactly this case. Mirrored it for
pii_matches: added a piiMatches field to SerializableJobData (Redis
deserialization included; serialization is generic) and to
GenerationJobMetadata; wired updateMetadata to persist it; and added
a replay branch to the cross-replica subscribe fallback that emits
pii_matches from jobData.piiMatches alongside the reconstructed
created event. The two emit sites now dual-write: live emitChunk for
same-replica subscribers, updateMetadata for late or cross-replica
subscribers.

api/server/controllers/agents/client.js

@@ -1064,9 +1064,19 @@ class AgentClient extends BaseClient {
             // bytes flow through a separate GET endpoint fed by
             // GenerationJobManager. Writing to this.options.res here
             // is silently swallowed because it's already ended.
+            // Persisting on the job in parallel with the live emit
+            // lets the cross-replica subscribe fallback replay the
+            // toast for a subscriber that lands on a different
+            // replica than this one.
             const streamId = this.options.req?._resumableStreamId;
             if (streamId != null) {
               GenerationJobManager.emitChunk(streamId, { type: 'pii_matches', matches });
+              GenerationJobManager.updateMetadata(streamId, { piiMatches: matches }).catch((err) =>
+                logger.warn(
+                  `[messagePiiFilter] Failed to persist piiMatches on job ${streamId}:`,
+                  err,
+                ),
+              );
             }
           },
         });

api/server/controllers/agents/request.js

@@ -184,6 +184,12 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
         type: 'pii_matches',
         matches: piiPreRedactMatches,
       });
+      // Persist on the job so a subscriber landing on a different
+      // replica than this one gets the toast via the cross-replica
+      // reconstruction in GenerationJobManager.subscribe.
+      await GenerationJobManager.updateMetadata(streamId, {
+        piiMatches: piiPreRedactMatches,
+      });
     }
 
     // Send JSON response IMMEDIATELY so client can connect to SSE stream

packages/api/src/stream/GenerationJobManager.ts

@@ -948,6 +948,20 @@ class GenerationJobManagerClass {
           streamId,
         };
         onChunk(createdEvent);
+
+        /**
+         * Replay pii_matches from persisted metadata so the warn-mode
+         * toast still fires for subscribers that connect on a
+         * different replica than the one that emitted the live event.
+         * The same-replica path is already covered by earlyEventBuffer
+         * replay above; this branch only runs when no buffer exists.
+         */
+        if (jobData.piiMatches && jobData.piiMatches.length > 0) {
+          logger.debug(
+            `[GenerationJobManager] Cross-replica subscribe: emitting pii_matches from metadata for ${streamId}`,
+          );
+          onChunk({ type: 'pii_matches', matches: jobData.piiMatches });
+        }
       }
 
       try {
@@ -1269,6 +1283,9 @@ class GenerationJobManagerClass {
     if (metadata.promptTokens !== undefined) {
       updates.promptTokens = metadata.promptTokens;
     }
+    if (metadata.piiMatches) {
+      updates.piiMatches = metadata.piiMatches;
+    }
     await this.jobStore.updateJob(streamId, updates);
   }
 

packages/api/src/stream/implementations/RedisJobStore.ts

@@ -912,6 +912,7 @@ export class RedisJobStore implements IJobStore {
       conversationId: data.conversationId || undefined,
       error: data.error || undefined,
       userMessage: data.userMessage ? JSON.parse(data.userMessage) : undefined,
+      piiMatches: data.piiMatches ? JSON.parse(data.piiMatches) : undefined,
       responseMessageId: data.responseMessageId || undefined,
       createdEventEmitted: data.createdEventEmitted === '1',
       sender: data.sender || undefined,

packages/api/src/stream/interfaces/IJobStore.ts

@@ -33,6 +33,19 @@ export interface SerializableJobData {
   /** Whether the user-message created event has been emitted */
   createdEventEmitted?: boolean;
 
+  /**
+   * PII matches detected by the messagePiiFilter pre-redaction step.
+   * Persisted so a subscriber that connects on a different replica
+   * than the one that emitted the live `pii_matches` SSE event can
+   * still receive the warn-mode toast via the cross-replica
+   * reconstruction path in `GenerationJobManager.subscribe`.
+   */
+  piiMatches?: Array<{
+    patternId: string;
+    patternLabel: string;
+    count: number;
+  }>;
+
   /** Sender name for UI display */
   sender?: string;
 

packages/api/src/types/stream.ts

@@ -1,5 +1,5 @@
-import type { EventEmitter } from 'events';
 import type { Agents } from 'librechat-data-provider';
+import type { EventEmitter } from 'events';
 import type { ServerSentEvent } from '~/types';
 
 export interface GenerationJobMetadata {
@@ -20,6 +20,18 @@ export interface GenerationJobMetadata {
   model?: string;
   /** Prompt token count for abort token spending */
   promptTokens?: number;
+  /**
+   * PII matches detected by messagePiiFilter pre-redaction. When set
+   * via `updateMetadata`, the cross-replica subscribe fallback emits
+   * a `pii_matches` SSE event from this payload (parallel to the
+   * `created` event reconstruction) so the warn-mode toast survives
+   * a subscriber landing on a different replica than the emitter.
+   */
+  piiMatches?: Array<{
+    patternId: string;
+    patternLabel: string;
+    count: number;
+  }>;
 }
 
 export type GenerationJobStatus = 'running' | 'complete' | 'error' | 'aborted';