diff --git a/CONTEXT.md b/CONTEXT.md index 3880719db0..db384f8b3e 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -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. diff --git a/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts b/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts index d1ad527497..3d22ce6274 100644 --- a/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts +++ b/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts @@ -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', diff --git a/client/src/hooks/SSE/useResumableSSE.ts b/client/src/hooks/SSE/useResumableSSE.ts index 10379900ca..503c4fa8e5 100644 --- a/client/src/hooks/SSE/useResumableSSE.ts +++ b/client/src/hooks/SSE/useResumableSSE.ts @@ -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) { diff --git a/packages/api/src/mcp/oauth/index.ts b/packages/api/src/mcp/oauth/index.ts index 7e48d8a5c0..eb3d65397f 100644 --- a/packages/api/src/mcp/oauth/index.ts +++ b/packages/api/src/mcp/oauth/index.ts @@ -6,3 +6,4 @@ export * from './methods'; export * from './obo'; export * from './pending'; export * from './events'; +export * from './resume'; diff --git a/packages/api/src/mcp/oauth/resume.spec.ts b/packages/api/src/mcp/oauth/resume.spec.ts new file mode 100644 index 0000000000..2615257570 --- /dev/null +++ b/packages/api/src/mcp/oauth/resume.spec.ts @@ -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[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', + }), + ]); + }); +}); diff --git a/packages/api/src/mcp/oauth/resume.ts b/packages/api/src/mcp/oauth/resume.ts new file mode 100644 index 0000000000..672a28bb42 --- /dev/null +++ b/packages/api/src/mcp/oauth/resume.ts @@ -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 | undefined { + return value != null && typeof value === 'object' + ? (value as Record) + : undefined; +} + +function getStepId(event: NonNullable[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 | 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 | 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(); + 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(); + 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; +} diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 9b7f979062..df3d1663bb 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -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 / diff --git a/packages/api/src/stream/__tests__/GenerationJobManager.resumeReplay.spec.ts b/packages/api/src/stream/__tests__/GenerationJobManager.resumeReplay.spec.ts index dee6a9e117..7e67fec122 100644 --- a/packages/api/src/stream/__tests__/GenerationJobManager.resumeReplay.spec.ts +++ b/packages/api/src/stream/__tests__/GenerationJobManager.resumeReplay.spec.ts @@ -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 () => { diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 4e4f981994..0f93d54668 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -860,6 +860,8 @@ export interface ResumeState { data?: unknown; [key: string]: unknown; }>; + /** Pending MCP authorization prompts projected from durable stream state. */ + pendingOAuthPrompts?: Agents.PendingMCPOAuthPrompt[]; } /** diff --git a/packages/data-provider/src/types/agents.ts b/packages/data-provider/src/types/agents.ts index 867cd7c9c6..a987504293 100644 --- a/packages/data-provider/src/types/agents.ts +++ b/packages/data-provider/src/types/agents.ts @@ -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 */