mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-04 14:57:42 +00:00
🌐 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.
This commit is contained in:
parent
eddb31853e
commit
6c7fe770e4
6 changed files with 60 additions and 1 deletions
|
|
@ -1064,9 +1064,19 @@ class AgentClient extends BaseClient {
|
||||||
// bytes flow through a separate GET endpoint fed by
|
// bytes flow through a separate GET endpoint fed by
|
||||||
// GenerationJobManager. Writing to this.options.res here
|
// GenerationJobManager. Writing to this.options.res here
|
||||||
// is silently swallowed because it's already ended.
|
// 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;
|
const streamId = this.options.req?._resumableStreamId;
|
||||||
if (streamId != null) {
|
if (streamId != null) {
|
||||||
GenerationJobManager.emitChunk(streamId, { type: 'pii_matches', matches });
|
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,
|
||||||
|
),
|
||||||
|
);
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
|
||||||
|
|
@ -184,6 +184,12 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
||||||
type: 'pii_matches',
|
type: 'pii_matches',
|
||||||
matches: piiPreRedactMatches,
|
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
|
// Send JSON response IMMEDIATELY so client can connect to SSE stream
|
||||||
|
|
|
||||||
|
|
@ -948,6 +948,20 @@ class GenerationJobManagerClass {
|
||||||
streamId,
|
streamId,
|
||||||
};
|
};
|
||||||
onChunk(createdEvent);
|
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 {
|
try {
|
||||||
|
|
@ -1269,6 +1283,9 @@ class GenerationJobManagerClass {
|
||||||
if (metadata.promptTokens !== undefined) {
|
if (metadata.promptTokens !== undefined) {
|
||||||
updates.promptTokens = metadata.promptTokens;
|
updates.promptTokens = metadata.promptTokens;
|
||||||
}
|
}
|
||||||
|
if (metadata.piiMatches) {
|
||||||
|
updates.piiMatches = metadata.piiMatches;
|
||||||
|
}
|
||||||
await this.jobStore.updateJob(streamId, updates);
|
await this.jobStore.updateJob(streamId, updates);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -912,6 +912,7 @@ export class RedisJobStore implements IJobStore {
|
||||||
conversationId: data.conversationId || undefined,
|
conversationId: data.conversationId || undefined,
|
||||||
error: data.error || undefined,
|
error: data.error || undefined,
|
||||||
userMessage: data.userMessage ? JSON.parse(data.userMessage) : undefined,
|
userMessage: data.userMessage ? JSON.parse(data.userMessage) : undefined,
|
||||||
|
piiMatches: data.piiMatches ? JSON.parse(data.piiMatches) : undefined,
|
||||||
responseMessageId: data.responseMessageId || undefined,
|
responseMessageId: data.responseMessageId || undefined,
|
||||||
createdEventEmitted: data.createdEventEmitted === '1',
|
createdEventEmitted: data.createdEventEmitted === '1',
|
||||||
sender: data.sender || undefined,
|
sender: data.sender || undefined,
|
||||||
|
|
|
||||||
|
|
@ -33,6 +33,19 @@ export interface SerializableJobData {
|
||||||
/** Whether the user-message created event has been emitted */
|
/** Whether the user-message created event has been emitted */
|
||||||
createdEventEmitted?: boolean;
|
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 name for UI display */
|
||||||
sender?: string;
|
sender?: string;
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,5 @@
|
||||||
import type { EventEmitter } from 'events';
|
|
||||||
import type { Agents } from 'librechat-data-provider';
|
import type { Agents } from 'librechat-data-provider';
|
||||||
|
import type { EventEmitter } from 'events';
|
||||||
import type { ServerSentEvent } from '~/types';
|
import type { ServerSentEvent } from '~/types';
|
||||||
|
|
||||||
export interface GenerationJobMetadata {
|
export interface GenerationJobMetadata {
|
||||||
|
|
@ -20,6 +20,18 @@ export interface GenerationJobMetadata {
|
||||||
model?: string;
|
model?: string;
|
||||||
/** Prompt token count for abort token spending */
|
/** Prompt token count for abort token spending */
|
||||||
promptTokens?: number;
|
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';
|
export type GenerationJobStatus = 'running' | 'complete' | 'error' | 'aborted';
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue