🛄 fix: Reclaim Pending MCP OAuth Prompts on Resume (#15471)

This commit is contained in:
Danny Avila 2026-09-01 16:52:18 -04:00 • committed by GitHub
parent 74535662fe
commit 6122d210da
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 502 additions and 1 deletions

View file

@ -7,6 +7,7 @@
- **Agent turn execution plan**: the immutable, request-local decision compiled once after authentication, agent resolution, and tool initialization. It records the trusted turn origin, conversation lineage, pause capability, binding/action context, and the preferred checkpoint, history, or fresh state-loading strategy without executing the model or owning persistence. Checkpoint failure falls back to durable history within the same Agents lifecycle.
- **Effective agent selection**: the resolved endpoint and agent identity after an enforced model spec is applied. Authorization and agent loading must consume this same identity before the Agent run envelope is initialized.
- **MCP runtime request body**: trusted chat identifiers supplied only while an MCP server handles an agent request. It enables request-scoped header placeholders without retaining user-specific request data on a shared server definition.
- **MCP OAuth prompt projection**: the client-safe, generation-scoped view of authorization prompts that remain actionable in a resumable Agent stream. It is derived from durable OAuth step state, carries no tokens or flow internals, and lets reconnecting clients render current state without interpreting the replay log.
- **Caller Capability Projection**: the versioned, SDK-owned classification of currently active tools by direct and programmatic callers. Event-driven execution transports this projection as data; LibreChat intersects it with its trusted registry and never recomputes deferred-tool discovery policy or treats the projection as authorization.
- **Subagent thread**: a durable, view-only child conversation owned by one parent conversation and subagent identity. A parent agent may continue it by stable `threadId`; each continuation uses a fresh execution lease restored from the canonical child transcript. It is not an ordinary human-writable chat.
- **Live subagent task owner**: the one API process holding a detached child execution, its abort controller, and its bounded control queue. Redis may route trusted poll/control envelopes to that owner, but it does not migrate or persist the executor; Mongo persists only the logical child thread and its continuation fence.

View file

@ -5,6 +5,7 @@ import {
LocalStorageKeys,
QueryKeys,
StepEvents,
StepTypes,
request,
} from 'librechat-data-provider';
import type { TMessage, TSubmission } from 'librechat-data-provider';
@ -3260,6 +3261,97 @@ describe('useResumableSSE', () => {
unmount();
});
it('hydrates a projected pending OAuth prompt once against the resumed response', async () => {
const submission = buildSubmission();
const chatHelpers = buildChatHelpers();
const { unmount } = renderHook(() => useResumableSSE(submission, chatHelpers));
await act(async () => {
await Promise.resolve();
});
const runStep = {
id: 'step-oauth',
runId: Constants.USE_PRELIM_RESPONSE_MESSAGE_ID,
index: 0,
type: StepTypes.TOOL_CALLS,
stepDetails: {
type: StepTypes.TOOL_CALLS,
tool_calls: [{ id: 'call-oauth', name: 'oauth_mcp_Google-Workspace', args: '' }],
},
};
const replayEvent = {
event: StepEvents.ON_RUN_STEP_DELTA,
data: {
id: 'step-oauth',
delta: {
type: StepTypes.TOOL_CALLS,
tool_calls: [{ id: 'call-oauth', name: 'oauth_mcp_Google-Workspace', args: '' }],
auth: 'https://auth.example.com/oauth',
},
},
};
await act(async () => {
getLastSSE()._emit('message', {
data: JSON.stringify({
sync: true,
resumeState: {
runSteps: [runStep],
replayEvents: [replayEvent],
responseMessageId: 'resumed-response',
userMessage: {
messageId: 'resumed-user',
conversationId: CONV_ID,
text: 'Use Google Workspace',
},
pendingOAuthPrompts: [
{
stepId: 'step-oauth',
runId: Constants.USE_PRELIM_RESPONSE_MESSAGE_ID,
index: 0,
toolCallId: 'call-oauth',
toolName: 'oauth_mcp_Google-Workspace',
authURL: 'https://auth.example.com/oauth',
},
],
},
}),
});
});
expect(mockStepHandler).toHaveBeenCalledTimes(2);
expect(mockStepHandler).toHaveBeenNthCalledWith(
1,
{
event: StepEvents.ON_RUN_STEP,
data: expect.objectContaining({
id: 'step-oauth',
runId: Constants.USE_PRELIM_RESPONSE_MESSAGE_ID,
index: 0,
}),
},
expect.objectContaining({
initialResponse: expect.objectContaining({ messageId: 'resumed-response' }),
}),
);
expect(mockStepHandler).toHaveBeenNthCalledWith(
2,
{
event: StepEvents.ON_RUN_STEP_DELTA,
data: expect.objectContaining({
id: 'step-oauth',
delta: expect.objectContaining({ auth: 'https://auth.example.com/oauth' }),
}),
},
expect.objectContaining({
initialResponse: expect.objectContaining({ messageId: 'resumed-response' }),
}),
);
unmount();
});
it('merges resumed user and response messages into loaded conversation history', async () => {
const originalUser = {
messageId: 'original-user',

View file

@ -9,10 +9,12 @@ import {
QueryKeys,
ErrorTypes,
StepEvents,
StepTypes,
apiBaseUrl,
SteerEvents,
dataService,
ContentTypes,
ToolCallTypes,
ActivityLabelEvents,
ReasoningLabelEvents,
UsageEvents,
@ -456,6 +458,26 @@ const isOAuthStepEvent = (data: unknown) => {
return false;
};
const getStepEventId = (event: unknown): string | undefined => {
if (event == null || typeof event !== 'object' || !('data' in event)) {
return undefined;
}
const data = event.data;
if (data == null || typeof data !== 'object') {
return undefined;
}
if ('id' in data && typeof data.id === 'string') {
return data.id;
}
const result = 'result' in data ? data.result : undefined;
return result != null &&
typeof result === 'object' &&
'id' in result &&
typeof result.id === 'string'
? result.id
: undefined;
};
const replaceNewConversationUrl = (conversationId: string) => {
if (window.location.pathname !== `/c/${Constants.NEW_CONVO}`) {
return;
@ -1351,6 +1373,53 @@ export default function useResumableSSE(
stepHandler(event, submission);
}
};
const applyPendingOAuthPrompt = (
prompt: Agents.PendingMCPOAuthPrompt,
submission: EventSubmission,
) => {
const toolCall: Agents.ToolCall = {
id: prompt.toolCallId,
name: prompt.toolName,
args: '',
type: ToolCallTypes.TOOL_CALL,
};
const toolCallDelta: Agents.ToolCallChunk = {
id: prompt.toolCallId,
name: prompt.toolName,
args: '',
};
stepHandler(
{
event: StepEvents.ON_RUN_STEP,
data: {
id: prompt.stepId,
runId: prompt.runId,
index: prompt.index,
type: StepTypes.TOOL_CALLS,
stepDetails: {
type: StepTypes.TOOL_CALLS,
tool_calls: [toolCall],
},
},
},
submission,
);
stepHandler(
{
event: StepEvents.ON_RUN_STEP_DELTA,
data: {
id: prompt.stepId,
delta: {
type: StepTypes.TOOL_CALLS,
tool_calls: [toolCallDelta],
auth: prompt.authURL,
expires_at: prompt.expiresAt,
},
},
},
submission,
);
};
/**
* Maps a pending action onto the in-flight response message so the
@ -2111,8 +2180,13 @@ export default function useResumableSSE(
resumeSubmission,
);
const pendingOAuthPrompts = data.resumeState?.pendingOAuthPrompts ?? [];
const pendingOAuthStepIds = new Set(pendingOAuthPrompts.map((prompt) => prompt.stepId));
if (data.resumeState?.runSteps) {
for (const runStep of data.resumeState.runSteps) {
if (pendingOAuthStepIds.has(runStep.id)) {
continue;
}
stepHandler({ event: StepEvents.ON_RUN_STEP, data: runStep }, resumeSubmission);
}
}
@ -2217,6 +2291,10 @@ export default function useResumableSSE(
}
}
for (const prompt of pendingOAuthPrompts) {
applyPendingOAuthPrompt(prompt, resumeSubmission);
}
/**
* Re-pause on reconnect: the run is parked on a human-review
* interrupt. Re-apply the pending action so the approval / ask-user
@ -2268,6 +2346,14 @@ export default function useResumableSSE(
`Replaying ${data.resumeState.replayEvents.length} resume events`,
);
for (const replayEvent of data.resumeState.replayEvents) {
const replayStepId = getStepEventId(replayEvent);
if (
replayStepId != null &&
pendingOAuthStepIds.has(replayStepId) &&
isOAuthStepEvent(replayEvent)
) {
continue;
}
if (replayEvent.event === UsageEvents.ON_CONTEXT_USAGE) {
contextHandler(replayEvent.data, resumeSubmission);
} else if (replayEvent.event === UsageEvents.ON_TOKEN_USAGE) {

View file

@ -6,3 +6,4 @@ export * from './methods';
export * from './obo';
export * from './pending';
export * from './events';
export * from './resume';

View file

@ -0,0 +1,140 @@
import { StepTypes } from 'librechat-data-provider';
import type { Agents } from 'librechat-data-provider';
import type * as t from '~/types';
import { projectPendingMCPOAuthPrompts } from './resume';
const NOW = 1_800_000_000_000;
function runStep(id: string, index: number, serverName: string): Agents.RunStep {
return {
id,
runId: 'USE_PRELIM_RESPONSE_MESSAGE_ID',
index,
type: StepTypes.TOOL_CALLS,
stepDetails: {
type: StepTypes.TOOL_CALLS,
tool_calls: [{ id: `call-${id}`, name: `oauth_mcp_${serverName}`, args: '' }],
},
};
}
function authDelta(
id: string,
serverName: string,
authURL: string,
expiresAt = NOW + 60_000,
): NonNullable<t.ResumeState['replayEvents']>[number] {
return {
event: 'on_run_step_delta',
data: {
id,
delta: {
type: 'tool_calls',
tool_calls: [{ name: `oauth_mcp_${serverName}`, args: '' }],
auth: authURL,
expires_at: expiresAt,
},
},
};
}
describe('projectPendingMCPOAuthPrompts', () => {
test('projects the latest active prompt for each OAuth step in content order', () => {
const steps = [runStep('step-slack', 2, 'Slack'), runStep('step-drive', 1, 'Drive')];
const replayEvents = [
authDelta('step-slack', 'Slack', 'https://old.example.com/oauth'),
authDelta('step-drive', 'Drive', 'https://drive.example.com/oauth'),
authDelta('step-slack', 'Slack', 'https://slack.example.com/oauth'),
];
expect(projectPendingMCPOAuthPrompts(replayEvents, steps, NOW)).toEqual([
{
stepId: 'step-drive',
runId: 'USE_PRELIM_RESPONSE_MESSAGE_ID',
index: 1,
toolCallId: 'call-step-drive',
toolName: 'oauth_mcp_Drive',
authURL: 'https://drive.example.com/oauth',
expiresAt: NOW + 60_000,
},
{
stepId: 'step-slack',
runId: 'USE_PRELIM_RESPONSE_MESSAGE_ID',
index: 2,
toolCallId: 'call-step-slack',
toolName: 'oauth_mcp_Slack',
authURL: 'https://slack.example.com/oauth',
expiresAt: NOW + 60_000,
},
]);
});
test('omits completed and expired prompts', () => {
const steps = [runStep('step-complete', 0, 'Slack'), runStep('step-expired', 1, 'Drive')];
const replayEvents = [
authDelta('step-complete', 'Slack', 'https://slack.example.com/oauth'),
authDelta('step-expired', 'Drive', 'https://drive.example.com/oauth', NOW - 1),
{
event: 'on_run_step_completed',
data: {
result: {
id: 'step-complete',
index: 0,
tool_call: {
id: 'call-step-complete',
name: 'oauth_mcp_Slack',
output: 'OAuth authentication completed',
},
},
},
},
];
expect(projectPendingMCPOAuthPrompts(replayEvents, steps, NOW)).toBeUndefined();
});
test('ignores an auth delta without durable run-step identity', () => {
expect(
projectPendingMCPOAuthPrompts(
[authDelta('missing-step', 'Slack', 'https://slack.example.com/oauth')],
[],
NOW,
),
).toBeUndefined();
});
test('does not let an earlier completed flow hide a newer prompt on the same step', () => {
const nextStep = runStep('step-slack', 0, 'Slack');
nextStep.stepDetails = {
type: StepTypes.TOOL_CALLS,
tool_calls: [{ id: 'call-new', name: 'oauth_mcp_Slack', args: '' }],
};
const replayEvents = [
authDelta('step-slack', 'Slack', 'https://slack.example.com/oauth'),
{
event: 'on_run_step_completed',
data: {
result: {
id: 'step-slack',
index: 0,
tool_call: {
id: 'call-old',
name: 'oauth_mcp_Slack',
output: 'OAuth authentication completed',
},
},
},
},
];
expect(projectPendingMCPOAuthPrompts(replayEvents, [nextStep], NOW)).toEqual([
expect.objectContaining({
stepId: 'step-slack',
toolCallId: 'call-new',
authURL: 'https://slack.example.com/oauth',
}),
]);
});
});

View file

@ -0,0 +1,133 @@
import { Constants } from 'librechat-data-provider';
import type { Agents } from 'librechat-data-provider';
import type * as t from '~/types';
const OAUTH_TOOL_PREFIX = `oauth${Constants.mcp_delimiter}`;
function getRecord(value: unknown): Record<string, unknown> | undefined {
return value != null && typeof value === 'object'
? (value as Record<string, unknown>)
: undefined;
}
function getStepId(event: NonNullable<t.ResumeState['replayEvents']>[number]): string | undefined {
const data = getRecord(event.data);
if (!data) {
return undefined;
}
if (event.event === 'on_run_step_completed') {
const result = getRecord(data.result);
return typeof result?.id === 'string' ? result.id : undefined;
}
return typeof data.id === 'string' ? data.id : undefined;
}
function getOAuthToolCall(toolCalls: unknown): Record<string, unknown> | undefined {
if (!Array.isArray(toolCalls)) {
return undefined;
}
return toolCalls.find((toolCall) => {
const candidate = getRecord(toolCall);
return typeof candidate?.name === 'string' && candidate.name.startsWith(OAUTH_TOOL_PREFIX);
}) as Record<string, unknown> | undefined;
}
function getPreferredString(primary: unknown, fallback: unknown): string | undefined {
if (typeof primary === 'string') {
return primary;
}
return typeof fallback === 'string' ? fallback : undefined;
}
/**
* Projects the current client-safe MCP OAuth prompts from the durable event
* representation retained for mixed-version resume compatibility.
*/
export function projectPendingMCPOAuthPrompts(
replayEvents: t.ResumeState['replayEvents'],
runSteps: readonly Agents.RunStep[],
now: number = Date.now(),
): Agents.PendingMCPOAuthPrompt[] | undefined {
if (!replayEvents?.length) {
return undefined;
}
const runStepsById = new Map(runSteps.map((runStep) => [runStep.id, runStep]));
const terminalToolCallIds = new Map<string, string | undefined>();
for (const event of replayEvents) {
if (event.event !== 'on_run_step_completed') {
continue;
}
const stepId = getStepId(event);
if (stepId) {
const data = getRecord(event.data);
const result = getRecord(data?.result);
const toolCall = getRecord(result?.tool_call);
terminalToolCallIds.set(stepId, typeof toolCall?.id === 'string' ? toolCall.id : undefined);
}
}
const prompts = new Map<string, Agents.PendingMCPOAuthPrompt>();
for (const event of replayEvents) {
if (event.event !== 'on_run_step_delta') {
continue;
}
const data = getRecord(event.data);
const delta = getRecord(data?.delta);
const stepId = typeof data?.id === 'string' ? data.id : undefined;
const authURL = typeof delta?.auth === 'string' ? delta.auth : undefined;
if (!delta || !stepId || !authURL) {
continue;
}
const expiresAt =
typeof delta.expires_at === 'number' && Number.isFinite(delta.expires_at)
? delta.expires_at
: undefined;
if (expiresAt != null && expiresAt <= now) {
continue;
}
const runStep = runStepsById.get(stepId);
if (
!runStep ||
runStep.status === 'completed' ||
runStep.status === 'cancelled' ||
runStep.status === 'failed'
) {
continue;
}
const runStepToolCalls =
runStep.stepDetails.type === 'tool_calls' ? runStep.stepDetails.tool_calls : undefined;
const deltaToolCall = getOAuthToolCall(delta.tool_calls);
const runStepToolCall = getOAuthToolCall(runStepToolCalls);
const toolName = getPreferredString(deltaToolCall?.name, runStepToolCall?.name);
if (!toolName) {
continue;
}
const deltaToolCallId = deltaToolCall?.id;
const runStepToolCallId = runStepToolCall?.id;
const toolCallId = getPreferredString(deltaToolCallId, runStepToolCallId);
if (terminalToolCallIds.has(stepId)) {
const terminalToolCallId = terminalToolCallIds.get(stepId);
if (!terminalToolCallId || !toolCallId || terminalToolCallId === toolCallId) {
continue;
}
}
prompts.set(stepId, {
stepId,
runId: runStep.runId,
index: runStep.index,
toolCallId,
toolName,
authURL,
expiresAt,
});
}
return prompts.size > 0
? [...prompts.values()].sort((left, right) => left.index - right.index)
: undefined;
}

View file

@ -82,6 +82,7 @@ import {
import { filterPersistableAbortContent } from './abortContent';
import { toClientPendingAction } from '~/agents/hitl/policy';
import { ApprovalLifecycle, pausePersistenceActionId } from './ApprovalLifecycle';
import { projectPendingMCPOAuthPrompts } from '~/mcp/oauth/resume';
import { sanitizeJobMetadata } from './metadata';
/** Terminal error surfaced to a client still attached when its approval window lapses. */
@ -7567,6 +7568,7 @@ class GenerationJobManagerClass {
model: jobData.model,
titleEvent,
replayEvents,
pendingOAuthPrompts: projectPendingMCPOAuthPrompts(replayEvents, effectiveRunSteps),
collectedUsage,
contextUsage,
// Carry the live pending approval in the resume contract so a reloading /

View file

@ -86,6 +86,7 @@ describe('GenerationJobManager resume replay events', () => {
test('includes OAuth run step and delta replay events in resume state', async () => {
manager = createInMemoryManager();
const streamId = `oauth-delta-resume-${Date.now()}`;
const expiresAt = Date.now() + 60_000;
await manager.createJob(streamId, 'user-1', streamId);
const runStepEvent = {
@ -108,7 +109,7 @@ describe('GenerationJobManager resume replay events', () => {
type: 'tool_calls',
tool_calls: [{ name: 'oauth_mcp_Google-Workspace', args: '' }],
auth: 'https://auth.example.com/oauth',
expires_at: 1780791946,
expires_at: expiresAt,
},
},
} satisfies ServerSentEvent;
@ -141,6 +142,36 @@ describe('GenerationJobManager resume replay events', () => {
const resumeState = await manager.getResumeState(streamId);
expect(resumeState?.replayEvents).toEqual([runStepEvent, authEvent]);
expect(resumeState?.pendingOAuthPrompts).toEqual([
{
stepId: 'step-oauth',
runId: 'USE_PRELIM_RESPONSE_MESSAGE_ID',
index: 0,
toolCallId: 'call-oauth',
toolName: 'oauth_mcp_Google-Workspace',
authURL: 'https://auth.example.com/oauth',
expiresAt,
},
]);
await manager.emitChunk(streamId, {
event: 'on_run_step_completed',
data: {
result: {
id: 'step-oauth',
index: 0,
tool_call: {
id: 'call-oauth',
name: 'oauth_mcp_Google-Workspace',
output: 'OAuth authentication completed',
},
},
},
});
await expect(manager.getResumeState(streamId)).resolves.toMatchObject({
pendingOAuthPrompts: undefined,
});
});
test('retains emitted run steps when the live graph is unavailable during resume', async () => {

View file

@ -860,6 +860,8 @@ export interface ResumeState {
data?: unknown;
[key: string]: unknown;
}>;
/** Pending MCP authorization prompts projected from durable stream state. */
pendingOAuthPrompts?: Agents.PendingMCPOAuthPrompt[];
}
/**

View file

@ -275,6 +275,17 @@ export namespace Agents {
alwaysAppliedSkills?: string[];
}
/** Client-safe state for one MCP authorization prompt that remains actionable. */
export interface PendingMCPOAuthPrompt {
stepId: string;
runId?: string;
index: number;
toolCallId?: string;
toolName: string;
authURL: string;
expiresAt?: number;
}
/** State data sent to reconnecting clients */
export interface ResumeState {
runSteps: RunStep[];
@ -300,6 +311,8 @@ export namespace Agents {
data?: unknown;
[key: string]: unknown;
}>;
/** Pending MCP authorization prompts projected from durable stream state. */
pendingOAuthPrompts?: PendingMCPOAuthPrompt[];
/** Cumulative provider-reported usage for the run; backfills usage totals on resume */
collectedUsage?: TTokenUsageEvent[];
/** Latest context window snapshot; restores the usage gauge on resume */