From 6c7fe770e4b0afe346202b6b571141ec981a8e76 Mon Sep 17 00:00:00 2001 From: Dustin Healy <54083382+dustinhealy@users.noreply.github.com> Date: Sun, 7 Jun 2026 21:35:15 -0700 Subject: [PATCH] =?UTF-8?q?=F0=9F=8C=90=20fix:=20Persist=20pii=5Fmatches?= =?UTF-8?q?=20across=20replicas=20for=20multi-replica=20deployments?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 10 ++++++++++ api/server/controllers/agents/request.js | 6 ++++++ packages/api/src/stream/GenerationJobManager.ts | 17 +++++++++++++++++ .../src/stream/implementations/RedisJobStore.ts | 1 + packages/api/src/stream/interfaces/IJobStore.ts | 13 +++++++++++++ packages/api/src/types/stream.ts | 14 +++++++++++++- 6 files changed, 60 insertions(+), 1 deletion(-) diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index 79eae61105..7d1036a5c9 100644 --- a/api/server/controllers/agents/client.js +++ b/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, + ), + ); } }, }); diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index 519ea3fd25..c8369fa27e 100644 --- a/api/server/controllers/agents/request.js +++ b/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 diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index ad41a5e72a..300120c3e4 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/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); } diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index d49affdc28..1295314174 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/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, diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 17dca7b943..c122503472 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/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; diff --git a/packages/api/src/types/stream.ts b/packages/api/src/types/stream.ts index dd125a1aab..a0aad37113 100644 --- a/packages/api/src/types/stream.ts +++ b/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';