From 115f2f677d7faf95ae13048456cee51fa6bd2a4d Mon Sep 17 00:00:00 2001 From: Lia Date: Mon, 28 Sep 2026 16:32:14 +0000 Subject: [PATCH 1/2] fix: Resolve Persisted Responses by Their IDs --- .../agents/__tests__/responses.unit.spec.js | 367 +++++++++++++++++- api/server/controllers/agents/responses.js | 256 +++++++----- packages/api/src/agents/responses/index.ts | 3 + packages/api/src/agents/responses/service.ts | 16 +- .../agents/responses/storage.mongo.spec.ts | 74 ++++ .../api/src/agents/responses/storage.spec.ts | 80 ++++ packages/api/src/agents/responses/storage.ts | 71 ++++ 7 files changed, 753 insertions(+), 114 deletions(-) create mode 100644 packages/api/src/agents/responses/storage.mongo.spec.ts create mode 100644 packages/api/src/agents/responses/storage.spec.ts create mode 100644 packages/api/src/agents/responses/storage.ts diff --git a/api/server/controllers/agents/__tests__/responses.unit.spec.js b/api/server/controllers/agents/__tests__/responses.unit.spec.js index e8b1f2e7abe..13abb101bae 100644 --- a/api/server/controllers/agents/__tests__/responses.unit.spec.js +++ b/api/server/controllers/agents/__tests__/responses.unit.spec.js @@ -311,6 +311,7 @@ jest.mock('@librechat/api', () => ({ createSubagentUsageSink: jest.fn().mockReturnValue(jest.fn()), CHILD_THREAD_READ_ONLY_ERROR: 'This subagent thread is view-only. Continue it from its parent agent or create a separate chat.', + announceReply: jest.fn().mockResolvedValue(undefined), getLangfuseTraceMessageFields: jest.fn().mockResolvedValue({ langfuseSampled: true, langfuseDestinationIds: ['destination-1'], @@ -333,6 +334,10 @@ jest.mock('@librechat/api', () => ({ writeDone: jest.fn(), buildResponse: jest.fn().mockReturnValue({ id: 'resp_123', output: [] }), generateResponseId: jest.fn().mockReturnValue('resp_mock-123'), + resolveStoredResponse: (...args) => + jest.requireActual('@librechat/api').resolveStoredResponse(...args), + selectStoredResponseHistory: (...args) => + jest.requireActual('@librechat/api').selectStoredResponseHistory(...args), isValidationFailure: jest.fn().mockReturnValue(false), inspectContent: mockInspectContent, extractConversationTitleContent: jest.fn(({ title }) => [ @@ -402,7 +407,10 @@ jest.mock('@librechat/api', () => ({ on_chat_model_end: { handle: jest.fn() }, }, finalizeStream: jest.fn(), + prepareStream: jest.fn(), + completeStream: jest.fn(), }), + emitResponseFailed: jest.fn(), createAggregatorEventHandlers: jest.fn().mockReturnValue({ on_message_delta: { handle: jest.fn() }, on_reasoning_delta: { handle: jest.fn() }, @@ -543,6 +551,8 @@ jest.mock('~/models', () => ({ getFiles: jest.fn(), getUserKey: jest.fn(), getMessages: jest.fn().mockResolvedValue([]), + getMessage: jest.fn().mockResolvedValue(null), + updateMessage: jest.fn().mockResolvedValue({}), saveMessage: jest.fn().mockResolvedValue({}), updateFilesUsage: jest.fn(), getUserKeyValues: jest.fn(), @@ -1036,7 +1046,7 @@ describe('createResponse controller', () => { expect(api.getLangfuseTraceMessageFields).toHaveBeenCalledWith(req.config, 'resp_mock-123'); expect(saveMessage).toHaveBeenCalledWith( - req, + expect.objectContaining({ userId: 'user-123' }), expect.objectContaining({ messageId: 'resp_mock-123', isCreatedByUser: false, @@ -1470,10 +1480,10 @@ describe('createResponse controller', () => { model: 'agent-123', input: 'Hello', stream: false, - previous_response_id: 'resp_imported', + previous_response_id: 'previous-imported', }, }); - db.getConvo.mockResolvedValueOnce({ conversationId: 'resp_imported', user: 'user-123' }); + db.getConvo.mockResolvedValueOnce({ conversationId: 'previous-imported', user: 'user-123' }); db.getMessages.mockResolvedValueOnce([storedMessage]); api.assertModelBoundContent.mockImplementationOnce(({ storedMessages }) => { expect(storedMessages).toEqual([ @@ -1527,11 +1537,11 @@ describe('createResponse controller', () => { model: 'agent-123', input: 'Hello', stream: false, - previous_response_id: 'resp_path_marked', + previous_response_id: 'previous-path-marked', }, }); db.getConvo.mockResolvedValueOnce({ - conversationId: 'resp_path_marked', + conversationId: 'previous-path-marked', user: 'user-123', }); db.getMessages.mockResolvedValueOnce([ @@ -1583,11 +1593,11 @@ describe('createResponse controller', () => { model: 'agent-123', input: 'Hello', stream: false, - previous_response_id: 'resp_neighboring_model_output', + previous_response_id: 'previous-neighboring-model-output', }, }); db.getConvo.mockResolvedValueOnce({ - conversationId: 'resp_neighboring_model_output', + conversationId: 'previous-neighboring-model-output', user: 'user-123', }); db.getMessages.mockResolvedValueOnce([ @@ -2040,6 +2050,328 @@ describe('createResponse controller', () => { }); }); + describe('stored Responses round trip', () => { + it.each([false, true])( + 'POST store:true → GET response ID → POST previous_response_id (stream=%s)', + async (stream) => { + const api = require('@librechat/api'); + const db = require('~/models'); + const controller = require('../responses'); + const messages = []; + const conversations = new Map(); + const order = []; + const originals = [ + db.getMessage, + db.getConvo, + db.getMessages, + db.saveMessage, + db.saveConvo, + db.updateMessage, + ].map((mock) => [mock, mock.getMockImplementation()]); + const originalHandlers = api.createResponsesEventHandlers.getMockImplementation(); + db.getMessage.mockImplementation( + async ({ user, messageId }) => + messages.find((message) => message.user === user && message.messageId === messageId) ?? + null, + ); + db.getConvo.mockImplementation(async (user, id) => { + const convo = conversations.get(id); + return convo?.user === user ? convo : null; + }); + db.getMessages.mockImplementation(async ({ conversationId, user }) => + messages.filter( + (message) => message.conversationId === conversationId && message.user === user, + ), + ); + db.saveMessage.mockImplementation(async (context, message) => { + if (!context.userId) throw new Error('User not authenticated'); + const saved = { ...message, user: context.userId }; + messages.push(saved); + order.push(message.isCreatedByUser ? 'input' : 'output'); + return saved; + }); + db.updateMessage.mockImplementation(async (user, update) => { + const message = messages.find( + (row) => row.user === user && row.messageId === update.messageId, + ); + if (!message) throw new Error('Message not found'); + Object.assign(message, update); + order.push('commit'); + return message; + }); + db.saveConvo.mockImplementation(async (context, convo) => { + if (!context.userId) throw new Error('User not authenticated'); + const saved = { ...convo, user: context.userId, messages: [...messages] }; + conversations.set(convo.conversationId, saved); + order.push('conversation'); + return saved; + }); + api.createResponsesEventHandlers.mockImplementation(() => ({ + ...originalHandlers(), + prepareStream: () => order.push('prepare'), + completeStream: () => order.push('complete'), + })); + try { + for (const [id, input, previous] of [ + ['resp_first', 'first', undefined], + ['resp_second', 'second', 'resp_first'], + ]) { + api.generateResponseId.mockReturnValueOnce(id); + api.createResponseContext.mockReturnValueOnce({ responseId: id }); + api.convertInputToMessages.mockReturnValueOnce([{ role: 'user', content: input }]); + api.validateResponseRequest.mockReturnValueOnce({ + request: { + model: 'agent-123', + input, + stream, + store: true, + ...(previous && { previous_response_id: previous }), + }, + }); + const response = { + id, + status: 'completed', + output: [ + { type: 'message', content: [{ type: 'output_text', text: `answer to ${input}` }] }, + ], + usage: { output_tokens: 5 }, + }; + if (stream) api.buildResponse.mockReturnValueOnce(response); + else api.buildAggregatedResponse.mockReturnValueOnce(response); + await createResponse(req, res); + expect(api.sendResponsesErrorResponse).not.toHaveBeenCalled(); + if (id === 'resp_first') { + req.params = { id }; + await controller.getResponse(req, res); + expect(res.json.mock.calls.at(-1)[0]).toMatchObject({ + id, + status: 'completed', + output: [{ id, content: [{ text: 'answer to first' }] }], + }); + expect(order).toEqual( + stream + ? ['prepare', 'input', 'output', 'conversation', 'commit', 'complete'] + : ['input', 'output', 'conversation', 'commit'], + ); + } + } + expect(db.getMessage).toHaveBeenCalledWith({ user: 'user-123', messageId: 'resp_first' }); + expect(db.getConvo).toHaveBeenCalledWith('user-123', 'mock-uuid-456'); + expect(db.getMessages).toHaveBeenCalledWith({ + conversationId: 'mock-uuid-456', + user: 'user-123', + }); + expect(messages).toHaveLength(4); + expect(conversations.get('mock-uuid-456').messages).toHaveLength(4); + req.params = { id: 'resp_first' }; + await controller.getResponse(req, res); + expect(res.json.mock.calls.at(-1)[0].output).toEqual([ + expect.objectContaining({ + id: 'resp_first', + content: [expect.objectContaining({ text: 'answer to first' })], + }), + ]); + const callsBeforeDenial = api.createRun.mock.calls.length; + req.user = { id: 'another-user' }; + req.params = { id: 'resp_first' }; + await controller.getResponse(req, res); + expect(api.sendResponsesErrorResponse).toHaveBeenLastCalledWith( + res, + 404, + 'Response not found: resp_first', + 'not_found', + 'response_not_found', + ); + api.validateResponseRequest.mockReturnValueOnce({ + request: { + model: 'agent-123', + input: 'intruder', + previous_response_id: 'resp_first', + }, + }); + await createResponse(req, res); + expect(api.createRun).toHaveBeenCalledTimes(callsBeforeDenial); + expect(api.sendResponsesErrorResponse).toHaveBeenLastCalledWith( + res, + 404, + 'Response not found', + 'not_found', + ); + } finally { + for (const [mock, original] of originals) mock.mockImplementation(original); + api.createResponsesEventHandlers.mockImplementation(originalHandlers); + } + }, + ); + + it('rejects an unstored response ID for retrieval and continuation', async () => { + const api = require('@librechat/api'); + const db = require('~/models'); + const controller = require('../responses'); + await createResponse(req, res); + expect(db.saveMessage).not.toHaveBeenCalled(); + req.params = { id: 'resp_mock-123' }; + await controller.getResponse(req, res); + expect(api.sendResponsesErrorResponse).toHaveBeenLastCalledWith( + res, + 404, + 'Response not found: resp_mock-123', + 'not_found', + 'response_not_found', + ); + api.validateResponseRequest.mockReturnValueOnce({ + request: { + model: 'agent-123', + input: 'continue', + previous_response_id: 'resp_mock-123', + }, + }); + const priorRuns = api.createRun.mock.calls.length; + await createResponse(req, res); + expect(api.createRun).toHaveBeenCalledTimes(priorRuns); + expect(api.sendResponsesErrorResponse).toHaveBeenLastCalledWith( + res, + 404, + 'Response not found', + 'not_found', + ); + }); + + it.each([false, true])( + 'never reports completed when a conversation or commit write fails (stream=%s)', + async (stream) => { + const api = require('@librechat/api'); + const db = require('~/models'); + const controller = require('../responses'); + for (const method of ['saveConvo', 'updateMessage']) { + jest.clearAllMocks(); + api.validateResponseRequest.mockReturnValueOnce({ + request: { + model: 'agent-123', + input: 'hello', + store: true, + stream, + }, + }); + db[method].mockRejectedValueOnce(new Error('PRIVATE STORAGE ERROR')); + await createResponse(req, res); + if (stream) { + expect(api.emitResponseFailed).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ code: 'response_storage_failed' }), + ); + expect( + api.createResponsesEventHandlers.mock.results.at(-1).value.completeStream, + ).not.toHaveBeenCalled(); + } else { + expect(api.sendResponsesErrorResponse).toHaveBeenCalledWith( + res, + 500, + 'Failed to store response', + 'server_error', + 'response_storage_failed', + ); + expect(res.json).not.toHaveBeenCalled(); + } + db.getMessage.mockResolvedValueOnce({ + user: 'user-123', + messageId: 'resp_mock-123', + conversationId: 'mock-uuid-456', + isCreatedByUser: false, + endpoint: 'agents', + sender: 'Agent', + finish_reason: 'pending_storage', + }); + req.params = { id: 'resp_mock-123' }; + await controller.getResponse(req, res); + expect(api.sendResponsesErrorResponse).toHaveBeenLastCalledWith( + res, + 404, + 'Response not found: resp_mock-123', + 'not_found', + 'response_not_found', + ); + expect(JSON.stringify(api.sendResponsesErrorResponse.mock.calls)).not.toContain( + 'PRIVATE STORAGE ERROR', + ); + } + }, + ); + + it('does not use empty history after an indexed message read fails', async () => { + const api = require('@librechat/api'); + const db = require('~/models'); + api.validateResponseRequest.mockReturnValueOnce({ + request: { + model: 'agent-123', + input: 'continue', + previous_response_id: 'resp_existing', + }, + }); + db.getMessage.mockResolvedValueOnce({ + messageId: 'resp_existing', + conversationId: 'conversation-one', + endpoint: 'agents', + sender: 'Agent', + isCreatedByUser: false, + finish_reason: 'stop', + }); + db.getConvo.mockResolvedValueOnce({ conversationId: 'conversation-one', user: 'user-123' }); + db.getMessages.mockRejectedValueOnce(new Error('PRIVATE DB ERROR')); + await createResponse(req, res); + expect(api.createRun).not.toHaveBeenCalled(); + expect(api.sendResponsesErrorResponse).toHaveBeenCalledWith( + res, + 500, + 'Failed to load previous response history', + 'server_error', + ); + expect(JSON.stringify(api.sendResponsesErrorResponse.mock.calls)).not.toContain( + 'PRIVATE DB ERROR', + ); + }); + + it.each([false, true])( + 'never reports completed when a required save fails (stream=%s)', + async (stream) => { + const api = require('@librechat/api'); + const db = require('~/models'); + api.validateResponseRequest.mockReturnValueOnce({ + request: { + model: 'agent-123', + input: 'hello', + store: true, + stream, + }, + }); + db.saveMessage.mockRejectedValueOnce(new Error('PRIVATE STORAGE ERROR')); + await createResponse(req, res); + if (stream) { + expect(api.emitResponseFailed).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ code: 'response_storage_failed' }), + ); + expect( + api.createResponsesEventHandlers.mock.results.at(-1).value.completeStream, + ).not.toHaveBeenCalled(); + } else { + expect(api.sendResponsesErrorResponse).toHaveBeenCalledWith( + res, + 500, + 'Failed to store response', + 'server_error', + 'response_storage_failed', + ); + expect(res.json).not.toHaveBeenCalled(); + } + expect(JSON.stringify(res.write.mock.calls)).not.toContain('PRIVATE STORAGE ERROR'); + expect(JSON.stringify(api.sendResponsesErrorResponse.mock.calls)).not.toContain( + 'PRIVATE STORAGE ERROR', + ); + }, + ); + }); + describe('conversation ownership validation', () => { it('should skip ownership check when previous_response_id is not provided', async () => { const { getConvo } = require('~/models'); @@ -2075,17 +2407,17 @@ describe('createResponse controller', () => { model: 'agent-123', input: 'Hello', stream: false, - previous_response_id: 'resp_abc', + previous_response_id: 'previous', }, }); getConvo.mockResolvedValueOnce(null); await createResponse(req, res); - expect(getConvo).toHaveBeenCalledWith('user-123', 'resp_abc'); + expect(getConvo).toHaveBeenCalledWith('user-123', 'previous'); expect(sendResponsesErrorResponse).toHaveBeenCalledWith( res, 404, - 'Conversation not found', + 'Response not found', 'not_found', ); }); @@ -2098,13 +2430,13 @@ describe('createResponse controller', () => { model: 'agent-123', input: 'Hello', stream: false, - previous_response_id: 'resp_abc', + previous_response_id: 'previous', }, }); - getConvo.mockResolvedValueOnce({ conversationId: 'resp_abc', user: 'user-123' }); + getConvo.mockResolvedValueOnce({ conversationId: 'previous', user: 'user-123' }); await createResponse(req, res); - expect(getConvo).toHaveBeenCalledWith('user-123', 'resp_abc'); + expect(getConvo).toHaveBeenCalledWith('user-123', 'previous'); expect(sendResponsesErrorResponse).not.toHaveBeenCalledWith( res, 404, @@ -2156,7 +2488,7 @@ describe('createResponse controller', () => { model: 'agent-123', input: 'Hello', stream: false, - previous_response_id: 'resp_abc', + previous_response_id: 'previous', }, }); getConvo.mockRejectedValueOnce(new Error('DB connection failed')); @@ -2484,9 +2816,10 @@ describe('createResponse controller', () => { await createResponse(req, res); - const finalizeStream = - api.createResponsesEventHandlers.mock.results.at(-1).value.finalizeStream; - expect(finalizeStream).toHaveBeenCalledWith(mockResponsesUsage); + const { prepareStream, completeStream } = + api.createResponsesEventHandlers.mock.results.at(-1).value; + expect(prepareStream).toHaveBeenCalledTimes(1); + expect(completeStream).toHaveBeenCalledWith(mockResponsesUsage); }); }); diff --git a/api/server/controllers/agents/responses.js b/api/server/controllers/agents/responses.js index 016ee1d10ad..1ddc27fbfb2 100644 --- a/api/server/controllers/agents/responses.js +++ b/api/server/controllers/agents/responses.js @@ -63,6 +63,9 @@ const { writeDone, buildResponse, generateResponseId, + resolveStoredResponse, + selectStoredResponseHistory, + emitResponseFailed, isValidationFailure, emitResponseCreated, createResponseContext, @@ -319,17 +322,16 @@ function extractResponseRequestContent(request, messageFragments) { /** * Load messages from a previous response/conversation * @param {string} conversationId - The conversation/response ID - * @param {string} userId - The user ID + * @param {string} userId - The authenticated user ID + * @param {string} [responseId] - Bound the history to this stored reply * @returns {Promise} Messages from the conversation */ -async function loadPreviousMessages(conversationId, userId) { +async function loadPreviousMessages(conversationId, userId, responseId) { try { - const messages = await db.getMessages({ conversationId, user: userId }); - if (!messages || messages.length === 0) { - return []; - } - - // Convert stored messages to internal format + const messages = selectStoredResponseHistory( + (await db.getMessages({ conversationId, user: userId })) ?? [], + responseId, + ); return messages.map((msg) => { let text; if (typeof msg.text === 'string') { @@ -358,23 +360,23 @@ async function loadPreviousMessages(conversationId, userId) { }); } catch (error) { logger.error('[Responses API] Error loading previous messages:', getSafeErrorMetadata(error)); - return []; + throw new Error('Failed to load previous response history'); } } /** * Save input messages to database - * @param {import('express').Request} req + * @param {{userId: string}} messageContext * @param {string} conversationId * @param {Array} inputMessages - Internal format messages * @param {string} agentId * @returns {Promise} */ -async function saveInputMessages(req, conversationId, inputMessages, agentId) { +async function saveInputMessages(messageContext, conversationId, inputMessages, agentId) { for (const msg of inputMessages) { if (msg.role === 'user') { - await db.saveMessage( - req, + const saved = await db.saveMessage( + messageContext, { messageId: msg.messageId || nanoid(), conversationId, @@ -387,6 +389,7 @@ async function saveInputMessages(req, conversationId, inputMessages, agentId) { }, { context: 'Responses API - save user input' }, ); + if (!saved) throw new Error('Failed to store response input'); } } } @@ -394,6 +397,7 @@ async function saveInputMessages(req, conversationId, inputMessages, agentId) { /** * Save response output to database * @param {import('express').Request} req + * @param {{userId: string}} messageContext * @param {string} conversationId * @param {string} responseId * @param {import('@librechat/api').Response} response @@ -403,6 +407,7 @@ async function saveInputMessages(req, conversationId, inputMessages, agentId) { */ async function saveResponseOutput( req, + messageContext, conversationId, responseId, response, @@ -424,8 +429,8 @@ async function saveResponseOutput( const langfuseTraceFields = await getLangfuseTraceMessageFields(req.config, responseId); // Save the assistant message - return db.saveMessage( - req, + const saved = await db.saveMessage( + messageContext, { messageId: responseId, conversationId, @@ -436,11 +441,13 @@ async function saveResponseOutput( sender: 'Agent', endpoint: EModelEndpoint.agents, model: agentId, - finish_reason: response.status === 'completed' ? 'stop' : response.status, + finish_reason: response.status === 'completed' ? 'pending_storage' : response.status, tokenCount: visibleOutputTokens ?? response.usage?.output_tokens, }, { context: 'Responses API - save assistant response' }, ); + if (!saved) throw new Error('Failed to store response output'); + return saved; } /** @@ -454,7 +461,7 @@ async function saveResponseOutput( */ async function saveConversation(req, conversationId, agentId, agent, codeEnvironmentDecision) { const title = resolveConversationTitle(req, agent?.name || 'Open Responses Conversation'); - await db.saveConvo( + const saved = await db.saveConvo( { userId: req?.user?.id, isTemporary: req?.resolvedConversation?.isTemporary ?? req?.body?.isTemporary, @@ -478,6 +485,7 @@ async function saveConversation(req, conversationId, agentId, agent, codeEnviron initialAgentId: agent?.id === agentId ? agentId : null, }, ); + if (!saved) throw new Error('Failed to store response conversation'); } /** @@ -645,7 +653,39 @@ const executeResponse = async (envelope, { req, res }) => { `[Responses API] Request ${responseId} started for agent ${agentId}, stream: ${isStreaming}`, ); - const conversationId = request.previous_response_id ?? uuidv4(); + let previousResponse; + if (request.previous_response_id != null) { + if (typeof request.previous_response_id !== 'string') { + return sendResponsesErrorResponse( + res, + 400, + 'previous_response_id must be a string', + 'invalid_request', + ); + } + try { + previousResponse = await resolveStoredResponse( + principal.userId, + request.previous_response_id, + db, + ); + } catch (error) { + logger.error( + '[Responses API] Error resolving previous response:', + getSafeErrorMetadata(error), + ); + return sendResponsesErrorResponse( + res, + 500, + 'Failed to load previous response', + 'server_error', + ); + } + if (!previousResponse) { + return sendResponsesErrorResponse(res, 404, 'Response not found', 'not_found'); + } + } + const conversationId = previousResponse?.conversation.conversationId ?? uuidv4(); /** @type {Promise[]} */ const artifactPromises = []; let artifactWritesCovered = false; @@ -689,24 +729,9 @@ const executeResponse = async (envelope, { req, res }) => { return handleExecutionError({ error, res, appConfig }); }, execute: async (execution) => { - if (request.previous_response_id != null) { - if (typeof request.previous_response_id !== 'string') { - return sendResponsesErrorResponse( - res, - 400, - 'previous_response_id must be a string', - 'invalid_request', - ); - } - const previousConversation = await db.getConvo( - principal.userId, - request.previous_response_id, - ); - if (!previousConversation) { - return sendResponsesErrorResponse(res, 404, 'Conversation not found', 'not_found'); - } - req.resolvedConversation = previousConversation; - if (previousConversation.subagentThread != null) { + if (previousResponse) { + req.resolvedConversation = previousResponse.conversation; + if (previousResponse.conversation.subagentThread != null) { return sendResponsesErrorResponse( res, 409, @@ -728,6 +753,12 @@ const executeResponse = async (envelope, { req, res }) => { }); req.resolvedConversation = admittedConversation; const parentMessageId = null; + const messageContext = { + userId: principal.userId, + isTemporary: req?.resolvedConversation?.isTemporary ?? req?.body?.isTemporary, + expiredAt: req?.resolvedConversation?.expiredAt, + interfaceConfig: req?.config?.interfaceConfig, + }; const mcpRequestBody = createMCPRuntimeRequestBody({ messageId: responseId, conversationId, @@ -739,10 +770,17 @@ const executeResponse = async (envelope, { req, res }) => { agentsEConfig?.backgroundTasks?.ordinaryToolCancellation === true; const backgroundCompletionResultMaxChars = agentsEConfig?.backgroundTasks?.completionResultMaxChars; - const previousMessages = request.previous_response_id - ? await loadPreviousMessages(request.previous_response_id, principal.userId) + const previousMessages = previousResponse + ? await loadPreviousMessages( + conversationId, + principal.userId, + previousResponse.message?.messageId, + ) : []; - if (request.previous_response_id) { + if (previousResponse?.message && previousMessages.length === 0) { + return sendResponsesErrorResponse(res, 404, 'Response not found', 'not_found'); + } + if (previousResponse) { assertModelBoundContent({ onTraversalFailure: reportLocatorTraversalFailure, filters: appConfig?.filters, @@ -1205,7 +1243,8 @@ const executeResponse = async (envelope, { req, res }) => { // Create event handlers const { handlers: responsesHandlers, - finalizeStream, + prepareStream, + completeStream, emitClientToolDeferral, } = createResponsesEventHandlers(handlerConfig); @@ -1396,52 +1435,66 @@ const executeResponse = async (envelope, { req, res }) => { const usage = buildResponsesUsage(collectedUsage); - // Finalize the stream - finalizeStream(usage); - res.end(); - - const duration = Date.now() - requestStartTime; - logger.debug( - `[Responses API] Request ${responseId} completed in ${duration}ms (streaming)`, - ); - - // Save to database if store: true + // Seal output items before storing their final text; do not publish success until durable. + prepareStream(); if (request.store === true) { try { - // Save conversation - await saveConversation(req, conversationId, agentId, agent, codeEnvironmentDecision); - - // Save input messages - await saveInputMessages(req, conversationId, inputMessages, agentId); - - // Build response for saving (use tracker with buildResponse for streaming) + await saveInputMessages(messageContext, conversationId, inputMessages, agentId); const finalResponse = buildResponse(context, tracker, 'completed'); const savedResponse = await saveResponseOutput( req, + messageContext, conversationId, responseId, finalResponse, agentId, tracker.usage.outputTokens, ); - await announceReply(db, { - userId: req?.user?.id, - conversationId, - reply: { - ...savedResponse, - isTemporary: req?.resolvedConversation?.isTemporary ?? req?.body?.isTemporary, - }, - context: 'Responses API - announce stored reply', + await saveConversation(req, conversationId, agentId, agent, codeEnvironmentDecision); + await db.updateMessage(principal.userId, { + messageId: responseId, + finish_reason: 'stop', }); + try { + await announceReply(db, { + userId: req?.user?.id, + conversationId, + reply: { + ...savedResponse, + finish_reason: 'stop', + isTemporary: req?.resolvedConversation?.isTemporary ?? req?.body?.isTemporary, + }, + context: 'Responses API - announce stored reply', + }); + } catch (announceError) { + logger.warn( + '[Responses API] Failed to announce stored reply:', + getSafeErrorMetadata(announceError), + ); + } logger.debug( `[Responses API] Stored response ${responseId} in conversation ${conversationId}`, ); } catch (saveError) { logger.error('[Responses API] Error saving response:', getSafeErrorMetadata(saveError)); - // Don't fail the request if saving fails + emitResponseFailed(handlerConfig, { + type: 'server_error', + message: 'Failed to store response', + code: 'response_storage_failed', + }); + writeDone(res); + res.end(); + return; } } + completeStream(usage); + res.end(); + + const duration = Date.now() - requestStartTime; + logger.debug( + `[Responses API] Request ${responseId} completed in ${duration}ms (streaming)`, + ); // The HTTP response is complete, while destructive cleanup still waits for artifacts. if (artifactPromises.length > 0) { @@ -1655,34 +1708,51 @@ const executeResponse = async (envelope, { req, res }) => { if (request.store === true) { try { - await saveConversation(req, conversationId, agentId, agent, codeEnvironmentDecision); - - await saveInputMessages(req, conversationId, inputMessages, agentId); - + await saveInputMessages(messageContext, conversationId, inputMessages, agentId); const savedResponse = await saveResponseOutput( req, + messageContext, conversationId, responseId, response, agentId, aggregator.usage.outputTokens, ); - await announceReply(db, { - userId: req?.user?.id, - conversationId, - reply: { - ...savedResponse, - isTemporary: req?.resolvedConversation?.isTemporary ?? req?.body?.isTemporary, - }, - context: 'Responses API - announce stored reply', + await saveConversation(req, conversationId, agentId, agent, codeEnvironmentDecision); + await db.updateMessage(principal.userId, { + messageId: responseId, + finish_reason: 'stop', }); + try { + await announceReply(db, { + userId: req?.user?.id, + conversationId, + reply: { + ...savedResponse, + finish_reason: 'stop', + isTemporary: req?.resolvedConversation?.isTemporary ?? req?.body?.isTemporary, + }, + context: 'Responses API - announce stored reply', + }); + } catch (announceError) { + logger.warn( + '[Responses API] Failed to announce stored reply:', + getSafeErrorMetadata(announceError), + ); + } logger.debug( `[Responses API] Stored response ${responseId} in conversation ${conversationId}`, ); } catch (saveError) { logger.error('[Responses API] Error saving response:', getSafeErrorMetadata(saveError)); - // Don't fail the request if saving fails + return sendResponsesErrorResponse( + res, + 500, + 'Failed to store response', + 'server_error', + 'response_storage_failed', + ); } } @@ -1794,7 +1864,7 @@ const listModels = async (req, res) => { * Get Response - GET /v1/responses/:id * * Retrieves a stored response by its ID. - * The response ID maps to a conversationId in LibreChat's storage. + * The response ID identifies the stored assistant message in its owning conversation. * * @param {import('express').Request} req * @param {import('express').Response} res @@ -1808,11 +1878,8 @@ const getResponse = async (req, res) => { return sendResponsesErrorResponse(res, 400, 'Response ID is required'); } - // The responseId could be either the response ID or the conversation ID - // Try to find a conversation with this ID - const conversation = await db.getConvo(userId, responseId); - - if (!conversation) { + const stored = await resolveStoredResponse(userId, responseId, db); + if (!stored) { return sendResponsesErrorResponse( res, 404, @@ -1822,8 +1889,14 @@ const getResponse = async (req, res) => { ); } - // Load messages for this conversation - const messages = await db.getMessages({ conversationId: responseId, user: userId }); + const { conversation, message } = stored; + // A response ID returns that one reply; conversation-ID reads retain the legacy view. + const messages = message + ? [message] + : selectStoredResponseHistory( + (await db.getMessages({ conversationId: conversation.conversationId, user: userId })) ?? + [], + ); if (!messages || messages.length === 0) { return sendResponsesErrorResponse( @@ -1839,7 +1912,7 @@ const getResponse = async (req, res) => { const output = convertMessagesToOutputItems(messages); // Find the last assistant message for usage info - const lastAssistantMessage = messages.filter((m) => !m.isCreatedByUser).pop(); + const lastAssistantMessage = message ?? messages.filter((m) => !m.isCreatedByUser).pop(); // Build the response object const response = { @@ -1886,12 +1959,7 @@ const getResponse = async (req, res) => { res.json(response); } catch (error) { logger.error('[Responses API] Error getting response:', getSafeErrorMetadata(error)); - sendResponsesErrorResponse( - res, - 500, - error instanceof Error ? error.message : 'Failed to get response', - 'server_error', - ); + sendResponsesErrorResponse(res, 500, 'Failed to get response', 'server_error'); } }; diff --git a/packages/api/src/agents/responses/index.ts b/packages/api/src/agents/responses/index.ts index 692174a4c7e..77912707593 100644 --- a/packages/api/src/agents/responses/index.ts +++ b/packages/api/src/agents/responses/index.ts @@ -157,6 +157,8 @@ export { updateTrackerUsage, } from './handlers'; +export { resolveStoredResponse, selectStoredResponseHistory } from './storage'; + // Service export { // Validation @@ -170,6 +172,7 @@ export { sendResponsesErrorResponse, // Context generateResponseId, + RESPONSE_ID_PREFIX, createResponseContext, // Streaming setup setupStreamingResponse, diff --git a/packages/api/src/agents/responses/service.ts b/packages/api/src/agents/responses/service.ts index ccd44ba4bff..be005c95f7b 100644 --- a/packages/api/src/agents/responses/service.ts +++ b/packages/api/src/agents/responses/service.ts @@ -482,8 +482,10 @@ export function sendResponsesErrorResponse( /** * Generate a unique response ID */ +export const RESPONSE_ID_PREFIX = 'resp_'; + export function generateResponseId(): string { - return `resp_${Date.now().toString(36)}${Math.random().toString(36).substring(2, 8)}`; + return `${RESPONSE_ID_PREFIX}${Date.now().toString(36)}${Math.random().toString(36).substring(2, 8)}`; } /** @@ -625,6 +627,8 @@ export function createResponsesEventHandlers(config: StreamHandlerConfig): { handlers: Record void }>; state: StreamState; finalizeStream: (usage?: Usage) => void; + prepareStream: () => void; + completeStream: (usage?: Usage) => void; emitClientToolDeferral: (callId: string, output: string) => void; } { const state: StreamState = { @@ -955,7 +959,7 @@ export function createResponsesEventHandlers(config: StreamHandlerConfig): { /** * Finalize the stream - close open items and emit completed */ - const finalizeStream = (usage?: Usage): void => { + const prepareStream = (): void => { closeOpenStreams(); for (const [callId, output] of pendingClientToolDeferrals ?? []) { deliverClientToolDeferral(callId, output); @@ -963,11 +967,17 @@ export function createResponsesEventHandlers(config: StreamHandlerConfig): { // A later step can announce a sibling before its arguments or deferral // are settled. Only terminate handoffs after the run has completed. closeOpenClientToolCalls(); + }; + const completeStream = (usage?: Usage): void => { emitResponseCompleted(config, usage); writeDone(config.res); }; + const finalizeStream = (usage?: Usage): void => { + prepareStream(); + completeStream(usage); + }; - return { handlers, state, finalizeStream, emitClientToolDeferral }; + return { handlers, state, finalizeStream, prepareStream, completeStream, emitClientToolDeferral }; } /* ============================================================================= diff --git a/packages/api/src/agents/responses/storage.mongo.spec.ts b/packages/api/src/agents/responses/storage.mongo.spec.ts new file mode 100644 index 00000000000..e3b59228c18 --- /dev/null +++ b/packages/api/src/agents/responses/storage.mongo.spec.ts @@ -0,0 +1,74 @@ +import mongoose from 'mongoose'; +import { EModelEndpoint } from 'librechat-data-provider'; +import { MongoMemoryServer } from 'mongodb-memory-server'; +import { createMethods, createModels } from '@librechat/data-schemas'; +import { resolveStoredResponse } from './storage'; + +const conversationId = '10000000-0000-4000-8000-000000000001'; +const owner = 'owner-user'; + +let mongod: MongoMemoryServer; +let db: ReturnType; + +beforeAll(async () => { + mongod = await MongoMemoryServer.create({ instance: { args: ['--nounixsocket'] } }); + await mongoose.connect(mongod.getUri()); + createModels(mongoose); + db = createMethods(mongoose); +}); + +afterAll(async () => { + await mongoose.disconnect(); + if (mongod) await mongod.stop(); +}); + +it('stores a response under an independent message ID and admits it only after the conversation commit', async () => { + const context = { userId: owner }; + const input = await db.saveMessage(context, { + messageId: 'input-one', + conversationId, + isCreatedByUser: true, + text: 'first', + sender: 'User', + endpoint: EModelEndpoint.agents, + }); + expect(input).toBeTruthy(); + const pending = await db.saveMessage(context, { + messageId: 'resp_one', + conversationId, + isCreatedByUser: false, + text: 'first answer', + sender: 'Agent', + endpoint: EModelEndpoint.agents, + finish_reason: 'pending_storage', + }); + expect(pending).toBeTruthy(); + expect(await resolveStoredResponse(owner, 'resp_one', db)).toBeNull(); + + const conversation = await db.saveConvo(context, { + conversationId, + agent_id: 'agent-one', + endpoint: EModelEndpoint.agents, + }); + expect(conversation).toBeTruthy(); + expect((await db.getConvo(owner, conversationId))?.messages).toHaveLength(2); + expect(await resolveStoredResponse(owner, 'resp_one', db)).toBeNull(); + + await db.updateMessage(owner, { messageId: 'resp_one', finish_reason: 'stop' }); + const stored = await resolveStoredResponse(owner, 'resp_one', db); + expect(stored?.conversation.conversationId).toBe(conversationId); + expect(stored?.message?.text).toBe('first answer'); + expect(await resolveStoredResponse('another-user', 'resp_one', db)).toBeNull(); + expect(await resolveStoredResponse(owner, 'resp_missing', db)).toBeNull(); + + await db.saveMessage(context, { + messageId: 'resp_failed', + conversationId, + isCreatedByUser: false, + text: 'uncommitted', + sender: 'Agent', + endpoint: EModelEndpoint.agents, + finish_reason: 'pending_storage', + }); + expect(await resolveStoredResponse(owner, 'resp_failed', db)).toBeNull(); +}); diff --git a/packages/api/src/agents/responses/storage.spec.ts b/packages/api/src/agents/responses/storage.spec.ts new file mode 100644 index 00000000000..5c22436039b --- /dev/null +++ b/packages/api/src/agents/responses/storage.spec.ts @@ -0,0 +1,80 @@ +import { EModelEndpoint } from 'librechat-data-provider'; +import { resolveStoredResponse, selectStoredResponseHistory } from './storage'; + +const responseMessage = { + messageId: 'resp_one', + conversationId: 'conversation-one', + endpoint: EModelEndpoint.agents, + sender: 'Agent', + isCreatedByUser: false, + finish_reason: 'stop', +}; +const conversation = { conversationId: 'conversation-one', user: 'owner' }; + +function reads() { + return { + getMessage: jest.fn().mockResolvedValue(responseMessage), + getConvo: jest.fn().mockResolvedValue(conversation), + }; +} + +describe('stored Responses ID resolution', () => { + it('looks up a response by its owner-scoped indexed message, then verifies its conversation', async () => { + const db = reads(); + expect(await resolveStoredResponse('owner', 'resp_one', db)).toEqual({ + message: responseMessage, + conversation, + }); + expect(db.getMessage).toHaveBeenCalledWith({ user: 'owner', messageId: 'resp_one' }); + expect(db.getConvo).toHaveBeenCalledWith('owner', 'conversation-one'); + }); + + it('does not accept unstored, foreign, non-assistant, or deleted-conversation response IDs', async () => { + const db = reads(); + for (const message of [ + null, + { ...responseMessage, isCreatedByUser: true }, + { ...responseMessage, endpoint: 'openAI' }, + { ...responseMessage, sender: 'User' }, + { ...responseMessage, finish_reason: 'pending_storage' }, + { ...responseMessage, conversationId: '' }, + ]) { + db.getMessage.mockResolvedValueOnce(message); + expect(await resolveStoredResponse('owner', 'resp_one', db)).toBeNull(); + } + db.getMessage.mockResolvedValueOnce(responseMessage); + db.getConvo.mockResolvedValueOnce(null); + expect(await resolveStoredResponse('owner', 'resp_one', db)).toBeNull(); + db.getMessage.mockResolvedValueOnce(responseMessage); + db.getConvo.mockResolvedValueOnce({ conversationId: 'different' }); + expect(await resolveStoredResponse('owner', 'resp_one', db)).toBeNull(); + }); + + it('replays only through the referenced response and excludes incomplete writes', () => { + const messages = [ + { messageId: 'user-one' }, + { messageId: 'resp_one', finish_reason: 'stop' }, + { messageId: 'resp_failed', finish_reason: 'pending_storage' }, + { messageId: 'user-two' }, + { messageId: 'resp_two', finish_reason: 'stop' }, + ]; + expect(selectStoredResponseHistory(messages, 'resp_one')).toEqual(messages.slice(0, 2)); + expect(selectStoredResponseHistory(messages, 'resp_two')).toEqual([ + messages[0], + messages[1], + messages[3], + messages[4], + ]); + expect(selectStoredResponseHistory(messages, 'resp_missing')).toEqual([]); + }); + + it('keeps the existing authorized conversation-ID form without a guaranteed-miss message read', async () => { + const db = reads(); + expect(await resolveStoredResponse('owner', 'conversation-one', db)).toEqual({ + message: null, + conversation, + }); + expect(db.getMessage).not.toHaveBeenCalled(); + expect(db.getConvo).toHaveBeenCalledWith('owner', 'conversation-one'); + }); +}); diff --git a/packages/api/src/agents/responses/storage.ts b/packages/api/src/agents/responses/storage.ts new file mode 100644 index 00000000000..18516755ef8 --- /dev/null +++ b/packages/api/src/agents/responses/storage.ts @@ -0,0 +1,71 @@ +import { EModelEndpoint } from 'librechat-data-provider'; +import { RESPONSE_ID_PREFIX } from './service'; + +/** Only persisted Responses assistant messages can name a continued run. */ +export interface StoredResponseMessage { + messageId: string; + conversationId: string; + endpoint?: string; + isCreatedByUser?: boolean; + sender?: string; + finish_reason?: string; + text?: string; + tokenCount?: number; + createdAt?: Date; + updatedAt?: Date; +} + +export interface StoredResponseConversation { + conversationId: string; +} + +/** The referenced response bounds history; an incomplete store is never model input. */ +export function selectStoredResponseHistory< + Message extends { messageId?: string; finish_reason?: string }, +>(messages: Message[], responseId?: string): Message[] { + const lastIndex = responseId + ? messages.findIndex((message) => message.messageId === responseId) + : -1; + if (responseId && lastIndex < 0) return []; + return (responseId ? messages.slice(0, lastIndex + 1) : messages).filter( + (message) => + !message.messageId?.startsWith(RESPONSE_ID_PREFIX) || message.finish_reason === 'stop', + ); +} + +export interface StoredResponseReads< + Message extends StoredResponseMessage, + Conversation extends StoredResponseConversation, +> { + getMessage: (params: { user: string; messageId: string }) => Promise; + getConvo: (user: string, conversationId: string) => Promise; +} + +/** A response ID is the assistant message ID; legacy conversation IDs remain supported. */ +export async function resolveStoredResponse< + Message extends StoredResponseMessage, + Conversation extends StoredResponseConversation, +>( + user: string, + id: string, + reads: StoredResponseReads, +): Promise<{ conversation: Conversation; message: Message | null } | null> { + if (!id.startsWith(RESPONSE_ID_PREFIX)) { + const conversation = await reads.getConvo(user, id); + return conversation ? { conversation, message: null } : null; + } + + const message = await reads.getMessage({ user, messageId: id }); + if ( + message?.messageId !== id || + message.isCreatedByUser !== false || + message.endpoint !== EModelEndpoint.agents || + message.sender !== 'Agent' || + message.finish_reason !== 'stop' || + !message.conversationId + ) { + return null; + } + const conversation = await reads.getConvo(user, message.conversationId); + return conversation?.conversationId === message.conversationId ? { conversation, message } : null; +} From 385b15018af7c523b519802c57465cab9c8c9e08 Mon Sep 17 00:00:00 2001 From: Lia Date: Mon, 28 Sep 2026 16:36:55 +0000 Subject: [PATCH 2/2] test: Prove Responses Complete Only After Stream Preparation --- .../responses/__tests__/service.test.ts | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/packages/api/src/agents/responses/__tests__/service.test.ts b/packages/api/src/agents/responses/__tests__/service.test.ts index 12c3c687a58..ba347341ace 100644 --- a/packages/api/src/agents/responses/__tests__/service.test.ts +++ b/packages/api/src/agents/responses/__tests__/service.test.ts @@ -81,6 +81,32 @@ describe('response usage aggregation', () => { expect(response.usage).toEqual(usage); }); + it('prepares final text without publishing completion until storage succeeds', () => { + const writes: string[] = []; + const res = { + write: (chunk: string) => { + writes.push(chunk); + }, + } as unknown as ServerResponse; + const tracker = createResponseTracker(); + const { handlers, prepareStream, completeStream } = createResponsesEventHandlers({ + res, + context, + tracker, + }); + handlers.on_message_delta.handle('on_message_delta', { + delta: { content: [{ type: 'text', text: 'saved answer' }] }, + }); + prepareStream(); + expect(buildResponse(context, tracker, 'completed').output).toEqual([ + expect.objectContaining({ content: [expect.objectContaining({ text: 'saved answer' })] }), + ]); + expect(writes.join('')).not.toContain('response.completed'); + completeStream(); + expect(writes.join('')).toContain('response.completed'); + expect(writes[writes.length - 1]).toBe('data: [DONE]\n\n'); + }); + it('uses the normalized override in the completed streaming event', () => { const writes: string[] = []; const res = {