From 60ca751a7f9d515097f998604a1fd5b2484cc75c Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 31 Jul 2026 14:06:13 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=A0=20fix:=20Preserve=20Deferred=20Too?= =?UTF-8?q?l=20Schemas=20Across=20HITL=20Resume=20(#14552)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * ๐Ÿง  fix: Preserve deferred tool schemas across HITL resume * ๐Ÿงช test: Harden deferred tool resume regression * ๐Ÿ“ฆ chore: bump @librechat/agents to v3.3.10 --- api/package.json | 2 +- api/server/controllers/agents/client.js | 51 ++--- api/server/controllers/agents/client.test.js | 59 +++++ api/server/controllers/agents/resume.js | 3 +- e2e/setup/fake-model.js | 210 +++++++++++++++++- e2e/specs/mock/deferred-tools-hitl.spec.ts | 175 +++++++++++++++ package-lock.json | 10 +- packages/api/package.json | 2 +- .../__tests__/run-summarization.test.ts | 5 +- packages/api/src/agents/run.spec.ts | 34 +++ packages/api/src/agents/run.ts | 30 ++- packages/api/src/stream/ApprovalLifecycle.ts | 32 ++- .../stream/__tests__/pendingAction.spec.ts | 34 +++ .../api/src/stream/interfaces/IJobStore.ts | 4 +- packages/api/src/types/stream.ts | 2 +- 15 files changed, 597 insertions(+), 56 deletions(-) create mode 100644 e2e/specs/mock/deferred-tools-hitl.spec.ts diff --git a/api/package.json b/api/package.json index ee4fc58872..b136d82cd2 100644 --- a/api/package.json +++ b/api/package.json @@ -46,7 +46,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "^4.3.3", - "@librechat/agents": "^3.3.9", + "@librechat/agents": "^3.3.10", "@librechat/api": "*", "@librechat/data-schemas": "*", "@microsoft/microsoft-graph-client": "^3.0.7", diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index 63e5060776..a01a74b839 100644 --- a/api/server/controllers/agents/client.js +++ b/api/server/controllers/agents/client.js @@ -36,7 +36,7 @@ const { buildPendingAction, toClientPendingAction, computeAgentRequestFingerprint, - extractDiscoveredToolsFromHistory, + getRunDiscoveredTools, captureResumeModelParameters, pickResumeContext, getApprovalTtlMs, @@ -2123,35 +2123,12 @@ class AgentClient extends BaseClient { } } - const paused = await GenerationJobManager.approvals.pause(streamId, pendingAction); - if (!paused) { - logger.debug( - `[AgentClient] Interrupt fired but job ${streamId} was not running; not pausing`, - ); - return; - } - - // Capture deferred tools discovered (via tool_search) earlier in THIS turn so resume - // can replay them into createRun. The resumed graph is rebuilt with `messages: []` - // (state comes from the checkpoint), so the in-turn tool_search results that mark a - // deferred tool discovered aren't present there โ€” without this the paused deferred - // tool would be missing from the rebuilt schema-only toolMap and resume would fail - // with "unknown tool". Inert for non-deferred turns (the set comes back empty). + // Snapshot deferred-tool discovery before exposing the pause. Tool-search results + // may live only in the interrupted SDK graph, so they must be committed atomically + // with requires_action for an immediate/cross-replica resume to retain the schemas. + let discoveredTools = []; try { - const runMessages = - typeof run.getRunMessages === 'function' ? run.getRunMessages() : undefined; - if (Array.isArray(runMessages) && runMessages.length > 0) { - const discovered = extractDiscoveredToolsFromHistory(runMessages); - if (discovered.size > 0) { - await GenerationJobManager.updateMetadata( - streamId, - { - discoveredTools: Array.from(discovered), - }, - this.jobCreatedAt, - ); - } - } + discoveredTools = getRunDiscoveredTools(run); } catch (err) { logger.warn( `[AgentClient] Failed to capture discovered tools for resume on ${streamId}`, @@ -2159,6 +2136,17 @@ class AgentClient extends BaseClient { ); } + const paused = await GenerationJobManager.approvals.pause(streamId, pendingAction, { + expectedCreatedAt: this.jobCreatedAt, + ...(discoveredTools.length > 0 ? { discoveredTools } : {}), + }); + if (!paused) { + logger.debug( + `[AgentClient] Interrupt fired but job ${streamId} was not running; not pausing`, + ); + return; + } + this.pendingApproval = pendingAction; // Release the concurrency slot this request held the MOMENT the turn is durably // paused โ€” before the approval card is emitted โ€” so the user's `/resume` can @@ -2838,9 +2826,8 @@ class AgentClient extends BaseClient { // batches keep claiming slots and generating group headers. activityLabel: this.buildActivityLabelWiring(streamId, abortController.signal), // Replay deferred tools discovered before the pause. With `messages: []` the - // discovery scan finds nothing, so a deferred tool the paused call targets - // would be absent from the rebuilt toolMap; these names (captured at pause) - // force it back in. Undefined/empty for non-deferred turns โ€” a harmless no-op. + // discovery scan finds nothing, so these names restore the schemas to the + // rebuilt model binding. Undefined/empty for non-deferred turns is a no-op. discoveredToolNames, initialSessions, runId: this.responseMessageId, diff --git a/api/server/controllers/agents/client.test.js b/api/server/controllers/agents/client.test.js index 954c45b747..72b663c5cf 100644 --- a/api/server/controllers/agents/client.test.js +++ b/api/server/controllers/agents/client.test.js @@ -13,6 +13,7 @@ const mockFormatAgentMessages = jest.fn(() => ({ const { Providers } = require('@librechat/agents'); const { Constants, ContentTypes, EModelEndpoint } = require('librechat-data-provider'); +const { GenerationJobManager, createStreamServices } = require('@librechat/api'); const AgentClient = require('./client'); const { resolveConfigServers } = require('~/server/services/MCP'); @@ -42,6 +43,7 @@ jest.mock('@librechat/api', () => ({ countTokens: jest.fn((text) => Math.ceil(String(text ?? '').length / 4)), createTokenCounter: jest.fn(() => jest.fn(() => 0)), deleteAgentCheckpoint: (...args) => mockDeleteAgentCheckpoint(...args), + decrementPendingRequest: jest.fn(async () => {}), initializeAgent: jest.fn(), isHITLEnabled: (...args) => mockIsHITLEnabled(...args), createMemoryProcessor: jest.fn(), @@ -55,6 +57,63 @@ jest.mock('@librechat/api', () => ({ maybePrewarmCodeSandbox: jest.fn(), })); +describe('AgentClient - interrupt discovery persistence', () => { + beforeEach(async () => { + await GenerationJobManager.destroy(); + GenerationJobManager.configure({ ...createStreamServices(), cleanupOnComplete: false }); + GenerationJobManager.initialize(); + }); + + afterEach(async () => { + await GenerationJobManager.destroy(); + }); + + it('makes the run discovery snapshot durable when the run pauses', async () => { + const streamId = 'conversation-discovered-pause'; + const job = await GenerationJobManager.createJob(streamId, 'user-123', streamId); + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: { endpoint: EModelEndpoint.agents, agent_id: 'agent-123' }, + config: { endpoints: { [EModelEndpoint.agents]: {} } }, + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + }, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = streamId; + client.responseMessageId = 'response-discovered-pause'; + client.jobCreatedAt = job.createdAt; + + await client.handleRunInterrupt( + { + getInterrupt: () => ({ + interruptId: 'ask-interrupt', + threadId: streamId, + payload: { + type: 'ask_user_question', + question: { question: 'Proceed?' }, + }, + }), + getDiscoveredTools: () => ['save_issue_mcp_linear'], + getRunMessages: () => [], + }, + streamId, + ); + + const paused = await GenerationJobManager.getJob(streamId); + expect(paused?.status).toBe('requires_action'); + expect(paused?.metadata.discoveredTools).toEqual(['save_issue_mcp_linear']); + }); +}); + jest.mock('~/server/services/Config', () => ({ getMCPServerTools: jest.fn(), })); diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 9819eaec2f..0d3e557acf 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -788,8 +788,7 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) // Carry the user's MCP auth so approved MCP tools run with their credentials. userMCPAuthMap: result.userMCPAuthMap, // Replay deferred tools discovered before the pause (captured at pause). The rebuilt - // graph passes `messages: []`, so without these an approved deferred tool would be - // absent from the schema-only toolMap and resume would fail with "unknown tool". + // graph passes `messages: []`, so without these the model would lose their schemas. discoveredToolNames: job.metadata?.discoveredTools, }); diff --git a/e2e/setup/fake-model.js b/e2e/setup/fake-model.js index d0cb5890f7..12f34a53de 100644 --- a/e2e/setup/fake-model.js +++ b/e2e/setup/fake-model.js @@ -42,6 +42,7 @@ const TOOL_APPROVAL_MARKER = 'E2E_TOOL_APPROVAL:'; const TOOL_APPROVAL_BATCH_MARKER = 'E2E_TOOL_APPROVAL_BATCH:'; const TOOL_APPROVAL_RESTRICTED_MARKER = 'E2E_TOOL_APPROVAL_RESTRICTED:'; const TOOL_APPROVAL_REWRITE_MARKER = 'E2E_TOOL_APPROVAL_REWRITE:'; +const DEFERRED_HITL_MARKER = 'E2E_DEFERRED_HITL:'; const HANDOFF_MARKER = 'E2E_HANDOFF:'; const HANDOFF_TOOL_PREFIX = 'lc_transfer_to_'; const CREATE_FILE_AUTHORING_FINAL_TEXT = 'E2E file authoring complete'; @@ -70,6 +71,10 @@ const SKILL_TOOL_NAME = 'skill'; const CREATE_SKILL_TOOL_CALL_ID = 'call_e2e_create_skill'; const EDIT_SKILL_TOOL_CALL_ID = 'call_e2e_edit_skill'; const BACKGROUND_TOOL_NAME = 'slow_echo_mcp_e2e-memory'; +const DEFERRED_HITL_TOOL_NAME = BACKGROUND_TOOL_NAME; +const DEFERRED_HITL_CONTROL_TOOL_NAME = 'recall_fact_mcp_e2e-memory'; +const TOOL_SEARCH_NAME = 'tool_search'; +const ASK_USER_QUESTION_NAME = 'ask_user_question'; const CHECK_BACKGROUND_TASK_TOOL_NAME = 'check_background_task'; const APPROVAL_TOOL_NAME = 'approval_probe_mcp_e2e-memory'; const APPROVAL_TOOL_CALL_PREFIX = 'call_e2e_approval_'; @@ -1348,6 +1353,201 @@ function getGraphTools(agentContext) { return result; } +function getInvocationAgentContext(graph, options, runManager) { + const directAgentId = runManager?.metadata?.agentId ?? options?.metadata?.agentId; + const agentId = + typeof directAgentId === 'string' + ? directAgentId + : getAgentIdFromInvocationOptions(options, runManager); + if (typeof agentId === 'string') { + const context = graph?.agentContexts?.get(agentId); + if (context) { + return context; + } + } + if (graph?.agentContexts?.size === 1) { + return graph.agentContexts.values().next().value; + } + return null; +} + +function findToolMessage(messages, toolCallId) { + return (messages ?? []).find( + (message) => messageType(message) === 'tool' && message?.tool_call_id === toolCallId, + ); +} + +function deferredHitlCallId(label, phase) { + return `call_e2e_deferred_hitl_${phase}_${label}`; +} + +function validateDeferredHitlSchema(agentContext, { expectBound }) { + const tools = getGraphTools(agentContext); + const tool = tools.get(DEFERRED_HITL_TOOL_NAME); + const failures = []; + if (tools.has(DEFERRED_HITL_CONTROL_TOOL_NAME)) { + failures.push( + `${DEFERRED_HITL_CONTROL_TOOL_NAME} negative control was provider-bound without discovery`, + ); + } + if (!expectBound) { + if (tool != null) { + failures.push(`${DEFERRED_HITL_TOOL_NAME} was bound before tool_search discovered it`); + } + return failures; + } + if (!tool) { + failures.push(`${DEFERRED_HITL_TOOL_NAME} was not provider-bound`); + return failures; + } + + const schema = tool.schema; + if (schema?.type !== 'object') { + failures.push(`${DEFERRED_HITL_TOOL_NAME} schema was not typed as object`); + } + const properties = + schema && + typeof schema === 'object' && + !Array.isArray(schema) && + schema.properties && + typeof schema.properties === 'object' && + !Array.isArray(schema.properties) + ? schema.properties + : null; + if (!properties) { + failures.push(`${DEFERRED_HITL_TOOL_NAME} did not expose an object properties schema`); + return failures; + } + + const propertyNames = Object.keys(properties).sort(); + if (JSON.stringify(propertyNames) !== JSON.stringify(['delay_ms', 'text'])) { + failures.push( + `${DEFERRED_HITL_TOOL_NAME} properties differed from delay_ms,text (${propertyNames.join(',')})`, + ); + } + if (properties.text?.type !== 'string') { + failures.push(`${DEFERRED_HITL_TOOL_NAME}.text was not typed as string`); + } + if (properties.delay_ms?.type !== 'number') { + failures.push(`${DEFERRED_HITL_TOOL_NAME}.delay_ms was not typed as number`); + } + const required = Array.isArray(schema.required) ? [...schema.required].sort() : null; + if (JSON.stringify(required) !== JSON.stringify(['text'])) { + failures.push( + `${DEFERRED_HITL_TOOL_NAME} required fields differed from text (${required?.join(',') ?? 'invalid'})`, + ); + } + return failures; +} + +/** + * Public-flow deferred-tool/HITL tracer. Every phase is inferred from message + * history because `/resume` rebuilds both the graph and this fake-model hook. + * Inspecting `getToolsForBinding()` mirrors the schemas a real provider sees; + * the registry alone would give a false positive for still-deferred tools. + */ +function deferredHitlInvocationResponse({ graph, messages, options, runManager }) { + const label = getMarkerValue(getLatestUserText(messages), DEFERRED_HITL_MARKER); + if (!label) { + return null; + } + + const searchCallId = deferredHitlCallId(label, 'search'); + const askCallId = deferredHitlCallId(label, 'ask'); + const probeCallId = deferredHitlCallId(label, 'probe'); + const searchResult = findToolMessage(messages, searchCallId); + const askResult = findToolMessage(messages, askCallId); + const probeResult = findToolMessage(messages, probeCallId); + const agentContext = getInvocationAgentContext(graph, options, runManager); + if (!agentContext) { + return { response: `E2E deferred HITL failed ${label}: active agent context was unavailable` }; + } + + if (probeResult) { + const expectedOutput = `E2E slow echo: resume-${label}`; + const output = getContentText(probeResult.content); + if (!output.includes(expectedOutput)) { + return { + response: `E2E deferred HITL failed ${label}: unexpected probe output ${output || '(empty)'}`, + }; + } + return { response: `E2E deferred HITL passed ${label}: ${expectedOutput}` }; + } + + if (askResult) { + const failures = validateDeferredHitlSchema(agentContext, { expectBound: true }); + const expectedAnswer = `continue-${label}`; + const answer = getContentText(askResult.content); + if (!answer.includes(expectedAnswer)) { + failures.push( + `ask answer mismatch (expected ${expectedAnswer}, received ${answer || '(empty)'})`, + ); + } + if (failures.length > 0) { + return { response: `E2E deferred HITL failed ${label}: ${failures.join('; ')}` }; + } + return { + response: '', + toolCalls: [ + { + id: probeCallId, + name: DEFERRED_HITL_TOOL_NAME, + args: { text: `resume-${label}` }, + type: 'tool_call', + }, + ], + }; + } + + if (searchResult) { + const failures = validateDeferredHitlSchema(agentContext, { expectBound: true }); + const searchOutput = getContentText(searchResult.content); + if (!searchOutput.includes(DEFERRED_HITL_TOOL_NAME)) { + failures.push(`${TOOL_SEARCH_NAME} output did not include ${DEFERRED_HITL_TOOL_NAME}`); + } + if (failures.length > 0) { + return { response: `E2E deferred HITL failed ${label}: ${failures.join('; ')}` }; + } + return { + response: '', + toolCalls: [ + { + id: askCallId, + name: ASK_USER_QUESTION_NAME, + args: { + question: `Continue deferred schema check ${label}?`, + options: [{ label: `Continue ${label}`, value: `continue-${label}` }], + }, + type: 'tool_call', + }, + ], + }; + } + + const failures = validateDeferredHitlSchema(agentContext, { expectBound: false }); + const boundTools = getGraphTools(agentContext); + if (!boundTools.has(TOOL_SEARCH_NAME)) { + failures.push(`${TOOL_SEARCH_NAME} was not provider-bound`); + } + if (!boundTools.has(ASK_USER_QUESTION_NAME)) { + failures.push(`${ASK_USER_QUESTION_NAME} was not provider-bound`); + } + if (failures.length > 0) { + return { response: `E2E deferred HITL failed ${label}: ${failures.join('; ')}` }; + } + return { + response: '', + toolCalls: [ + { + id: searchCallId, + name: TOOL_SEARCH_NAME, + args: { query: DEFERRED_HITL_TOOL_NAME, max_results: 1 }, + type: 'tool_call', + }, + ], + }; +} + function validateHandoffTool(route, tool, toolName) { const failures = []; const expectedDescription = route.description ?? `Transfer control to agent '${route.to}'`; @@ -1754,7 +1954,15 @@ module.exports = function fakeModelHook(run, context) { sleep, toolCalls, thrownError, - resolveInvocation, + resolveInvocation: async (streamMessages, streamOptions, runManager) => + deferredHitlInvocationResponse({ + graph, + messages: streamMessages, + options: streamOptions, + runManager, + }) ?? + resolveInvocation?.(streamMessages, streamOptions, runManager) ?? + null, resolveOnStream: (streamMessages, streamOptions, runManager) => approvalOutcomeResponses(streamMessages) ?? resolveOnStream?.(streamMessages, streamOptions, runManager) ?? diff --git a/e2e/specs/mock/deferred-tools-hitl.spec.ts b/e2e/specs/mock/deferred-tools-hitl.spec.ts new file mode 100644 index 0000000000..cdf58d4bda --- /dev/null +++ b/e2e/specs/mock/deferred-tools-hitl.spec.ts @@ -0,0 +1,175 @@ +import { expect, test } from '@playwright/test'; +import type { Page, Request } from '@playwright/test'; +import type { AgentDetail } from './agents.helpers'; +import { cleanupAgent, openAgentBuilder, uniqueAgentName } from './agents.helpers'; +import { + MOCK_ENDPOINTS, + NEW_CHAT_PATH, + escapeRegExp, + fetchJson, + getAccessToken, + messagesView, + requestJson, + sendMessage, +} from './helpers'; + +const MCP_SERVER_NAME = 'e2e-memory'; +const MCP_SERVER_TOOL_ID = `sys__server__sys_mcp_${MCP_SERVER_NAME}`; +const DEFERRED_TOOL_ID = `slow_echo_mcp_${MCP_SERVER_NAME}`; +const DEFERRED_CONTROL_TOOL_ID = `recall_fact_mcp_${MCP_SERVER_NAME}`; +const ASK_USER_QUESTION_TOOL_ID = 'ask_user_question'; +const PROMPT_MARKER = 'E2E_DEFERRED_HITL:'; +const DESCRIPTION = + 'Verifies deferred-tool discovery survives an ask_user_question pause and resume.'; + +type MCPToolsResponse = { + servers?: Record }>; +}; + +type AskResumeBody = { + actionId?: string; + agent_id?: string; + answer?: string; + conversationId?: string; + endpoint?: string; +}; + +function isResumeRequest(request: Request) { + return ( + request.method() === 'POST' && new URL(request.url()).pathname === '/api/agents/chat/resume' + ); +} + +async function waitForDeferredTools(page: Page) { + const token = await getAccessToken(page); + let latestTools: MCPToolsResponse | null = null; + + for (let attempt = 0; attempt < 20; attempt++) { + latestTools = await fetchJson(page, '/api/mcp/tools', token); + const tools = latestTools.servers?.[MCP_SERVER_NAME]?.tools ?? []; + const toolIds = new Set(tools.map((tool) => tool.pluginKey)); + if (toolIds.has(DEFERRED_TOOL_ID) && toolIds.has(DEFERRED_CONTROL_TOOL_ID)) { + return; + } + await page.waitForTimeout(500); + } + + expect( + latestTools?.servers?.[MCP_SERVER_NAME]?.tools, + `Expected ${MCP_SERVER_NAME} to expose both deferred test tools`, + ).toEqual( + expect.arrayContaining([ + expect.objectContaining({ pluginKey: DEFERRED_TOOL_ID }), + expect.objectContaining({ pluginKey: DEFERRED_CONTROL_TOOL_ID }), + ]), + ); +} + +async function createAgent(page: Page): Promise<{ id: string; name: string }> { + await page.goto(NEW_CHAT_PATH, { timeout: 10000 }); + await waitForDeferredTools(page); + + const token = await getAccessToken(page); + const agentName = uniqueAgentName('E2E Deferred HITL Agent'); + const agent = await requestJson(page, { + path: '/api/agents', + token, + method: 'POST', + body: { + name: agentName, + description: DESCRIPTION, + instructions: 'Discover the requested tool, ask the user, then run the discovered tool.', + provider: MOCK_ENDPOINTS[0].label, + model: MOCK_ENDPOINTS[0].model, + tools: [ + MCP_SERVER_TOOL_ID, + DEFERRED_TOOL_ID, + DEFERRED_CONTROL_TOOL_ID, + ASK_USER_QUESTION_TOOL_ID, + ], + tool_options: { + [DEFERRED_TOOL_ID]: { defer_loading: true }, + [DEFERRED_CONTROL_TOOL_ID]: { defer_loading: true }, + }, + }, + }); + expect(agent.tools).toEqual( + expect.arrayContaining([DEFERRED_TOOL_ID, DEFERRED_CONTROL_TOOL_ID, ASK_USER_QUESTION_TOOL_ID]), + ); + + return { id: agent.id, name: agentName }; +} + +async function selectAgent(page: Page, agentName: string) { + const form = await openAgentBuilder(page); + await form.getByRole('combobox', { name: 'Agent', exact: true }).click(); + await page.getByRole('option', { name: agentName }).click(); + await expect(form.getByLabel('Agent name')).toHaveValue(agentName); + await form.getByRole('button', { name: 'Select Agent' }).click(); +} + +test.describe('deferred tools across HITL resume', () => { + test('keeps a discovered tool provider-bound after ask_user_question resumes', async ({ + page, + }) => { + test.setTimeout(120000); + const label = `${Date.now()}-${Math.floor(Math.random() * 1e4)}`; + const question = `Continue deferred schema check ${label}?`; + const optionLabel = `Continue ${label}`; + const answer = `continue-${label}`; + let agentId: string | undefined; + + try { + const agent = await createAgent(page); + agentId = agent.id; + await selectAgent(page, agent.name); + + const response = await sendMessage(page, `${PROMPT_MARKER}${label}`); + expect(response.ok()).toBeTruthy(); + await expect(page).toHaveURL(/\/c\/(?!new)/, { timeout: 15000 }); + await expect(page.getByText(question, { exact: true })).toBeVisible({ timeout: 30000 }); + + /** Reload the public conversation route while the graph is paused. This + * proves the browser reconstructs the real persisted pending action, + * rather than resuming from transient state held by the original page. */ + const conversationPath = new URL(page.url()).pathname; + await page.reload({ waitUntil: 'domcontentloaded' }); + await expect(page).toHaveURL(conversationPath); + await expect(page.getByText(question, { exact: true })).toBeVisible({ timeout: 30000 }); + + const option = page.getByRole('button', { + name: new RegExp(`${escapeRegExp(optionLabel)}$`), + }); + await expect(option).toBeVisible(); + const [resumeRequest, resumeResponse] = await Promise.all([ + page.waitForRequest(isResumeRequest), + page.waitForResponse( + (candidate) => isResumeRequest(candidate.request()) && candidate.status() === 200, + ), + option.click(), + ]); + + const conversationId = conversationPath.replace('/c/', ''); + const body = resumeRequest.postDataJSON() as AskResumeBody; + expect(body.actionId).toBeTruthy(); + expect(body.agent_id).toBe(agentId); + expect(body.answer).toBe(answer); + expect(body.conversationId).toBe(conversationId); + expect(body.endpoint).toBe('agents'); + expect(resumeResponse.ok()).toBeTruthy(); + + /** The fake provider checks the deferred tool's exact JSON schema after + * tool_search and again after resume, while a second deferred tool stays + * unbound as a negative control. It invokes real MCP only if all pass. */ + const expectedFinal = + `E2E deferred HITL passed ${label}: ` + `E2E slow echo: resume-${label}`; + const terminal = messagesView(page).getByText( + new RegExp(`^E2E deferred HITL (?:passed|failed) ${escapeRegExp(label)}:`), + ); + await expect(terminal).toBeVisible({ timeout: 30000 }); + expect(await terminal.textContent()).toBe(expectedFinal); + } finally { + await cleanupAgent(page, agentId); + } + }); +}); diff --git a/package-lock.json b/package-lock.json index 2ecc1b488c..4767776f7c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -61,7 +61,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "^4.3.3", - "@librechat/agents": "^3.3.9", + "@librechat/agents": "^3.3.10", "@librechat/api": "*", "@librechat/data-schemas": "*", "@microsoft/microsoft-graph-client": "^3.0.7", @@ -10615,9 +10615,9 @@ } }, "node_modules/@librechat/agents": { - "version": "3.3.9", - "resolved": "https://registry.npmjs.org/@librechat/agents/-/agents-3.3.9.tgz", - "integrity": "sha512-4PJZR/jV8iS2PpLQZHxVw/PeNrkNBKgRaM1sKRhiV/72iofu4Ofm1wdaVZh0wlJ29mxkMyXxJGzsog+GflYsOA==", + "version": "3.3.10", + "resolved": "https://registry.npmjs.org/@librechat/agents/-/agents-3.3.10.tgz", + "integrity": "sha512-oiu1RPPKAcBNaRtYHiZVEjaPLyTaavJn1gm6/K62YkBQZ0ROUCp32zbbVajevUZT9GM6+9SB2FfaEzp5ApAb+A==", "license": "MIT", "dependencies": { "@anthropic-ai/sdk": "^0.115.0", @@ -42539,7 +42539,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "^4.3.3", - "@librechat/agents": "^3.3.9", + "@librechat/agents": "^3.3.10", "@librechat/data-schemas": "*", "@modelcontextprotocol/sdk": "^1.29.0", "@opentelemetry/api": "^1.9.0", diff --git a/packages/api/package.json b/packages/api/package.json index 99172e6045..083c2d30fb 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -108,7 +108,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "^4.3.3", - "@librechat/agents": "^3.3.9", + "@librechat/agents": "^3.3.10", "@librechat/data-schemas": "*", "@modelcontextprotocol/sdk": "^1.29.0", "@opentelemetry/api": "^1.9.0", diff --git a/packages/api/src/agents/__tests__/run-summarization.test.ts b/packages/api/src/agents/__tests__/run-summarization.test.ts index f2c8f1d21d..f85e2d1797 100644 --- a/packages/api/src/agents/__tests__/run-summarization.test.ts +++ b/packages/api/src/agents/__tests__/run-summarization.test.ts @@ -1885,9 +1885,8 @@ describe('toolOutputReferences gating', () => { // durable checkpoint), so the in-turn `tool_search` results that mark a deferred // tool discovered aren't on the critical path. createRun's `discoveredToolNames` // input replays those names โ€” captured at pause โ€” so the paused deferred tool is -// promoted back into `toolDefinitions` (and `defer_loading` flipped) and is present -// in the rebuilt schema-only toolMap. Without it, the approved tool would be missing -// and resume would fail with "unknown tool". +// promoted back into `toolDefinitions` (and `defer_loading` flipped) and its schema +// is restored to the rebuilt model binding. // --------------------------------------------------------------------------- describe('createRun deferred-tool replay (HITL resume)', () => { /** Agent whose discoverable `deep_tool` lives ONLY in the registry (deferred). */ diff --git a/packages/api/src/agents/run.spec.ts b/packages/api/src/agents/run.spec.ts index 0900e44518..d6577d275b 100644 --- a/packages/api/src/agents/run.spec.ts +++ b/packages/api/src/agents/run.spec.ts @@ -3,12 +3,46 @@ import { ReasoningResponseKey } from 'librechat-data-provider'; import { ToolMessage, AIMessage, HumanMessage } from '@librechat/agents/langchain/messages'; import { extractDiscoveredToolsFromHistory, + getRunDiscoveredTools, getReasoningKey, isDeepSeekReasoningProvider, shouldReplayReasoningContent, anyAgentReplaysReasoningContent, } from './run'; +describe('getRunDiscoveredTools', () => { + it('uses the run discovery snapshot instead of reconstructing it from messages', () => { + const messages = [ + new ToolMessage({ + content: JSON.stringify({ tools: [{ name: 'save_project_mcp_linear' }] }), + tool_call_id: 'call_1', + name: 'tool_search', + }), + ]; + + expect( + getRunDiscoveredTools({ + getDiscoveredTools: () => ['save_issue_mcp_linear'], + getRunMessages: () => messages, + }), + ).toEqual(['save_issue_mcp_linear']); + }); + + it('falls back to tool-search messages for agents releases without a snapshot API', () => { + const messages = [ + new ToolMessage({ + content: JSON.stringify({ tools: [{ name: 'save_issue_mcp_linear' }] }), + tool_call_id: 'call_1', + name: 'tool_search', + }), + ]; + + expect(getRunDiscoveredTools({ getRunMessages: () => messages })).toEqual([ + 'save_issue_mcp_linear', + ]); + }); +}); + describe('extractDiscoveredToolsFromHistory', () => { it('extracts tool names from tool_search JSON output', () => { const toolSearchOutput = JSON.stringify({ diff --git a/packages/api/src/agents/run.ts b/packages/api/src/agents/run.ts index fb94d680e8..4e71841e7d 100644 --- a/packages/api/src/agents/run.ts +++ b/packages/api/src/agents/run.ts @@ -154,6 +154,30 @@ export function extractDiscoveredToolsFromHistory(messages: BaseMessage[]): Set< return discoveredTools; } +export interface RunDiscoverySnapshot { + getDiscoveredTools?: () => string[]; + getRunMessages?: () => BaseMessage[] | undefined; +} + +/** Reads canonical run discovery state, with best-effort history parsing for older releases. */ +export function getRunDiscoveredTools(run: RunDiscoverySnapshot): string[] { + if (typeof run.getDiscoveredTools === 'function') { + const discoveredTools = run.getDiscoveredTools(); + if (Array.isArray(discoveredTools)) { + return Array.from(new Set(discoveredTools)); + } + } + + if (typeof run.getRunMessages !== 'function') { + return []; + } + const messages = run.getRunMessages(); + if (!Array.isArray(messages) || messages.length === 0) { + return []; + } + return Array.from(extractDiscoveredToolsFromHistory(messages)); +} + /** * Extracts skill names that were invoked in previous turns from raw message payload. * Scans assistant messages for tool_call content parts where name === 'skill'. @@ -1085,9 +1109,9 @@ export async function createRun({ * extraction. The HITL resume path rebuilds the graph with `messages: []` (state * comes from the durable checkpoint), so the in-turn `tool_search` results that * would normally mark a deferred tool discovered aren't present โ€” without this the - * paused tool would be absent from the rebuilt schema-only toolMap and resume would - * fail with "unknown tool". Captured at pause via `extractDiscoveredToolsFromHistory` - * and replayed here. Merged with (not replacing) any names extracted from `messages`. + * paused tool's schema would be absent from the rebuilt model binding. Captured at + * pause from canonical run state (with message parsing for older SDK releases) and + * replayed here. Merged with (not replacing) names extracted from `messages`. */ discoveredToolNames?: string[]; summarizationConfig?: SummarizationConfig; diff --git a/packages/api/src/stream/ApprovalLifecycle.ts b/packages/api/src/stream/ApprovalLifecycle.ts index 71ce0953e6..7e6a2396b6 100644 --- a/packages/api/src/stream/ApprovalLifecycle.ts +++ b/packages/api/src/stream/ApprovalLifecycle.ts @@ -9,6 +9,12 @@ export interface ApprovalLifecycleCallbacks { onExpired?: (streamId: string, createdAt: number) => void; } +export interface ApprovalPauseOptions { + discoveredTools?: string[]; + /** Generation identity observed by the interrupted run. */ + expectedCreatedAt?: number; +} + /** * The guarded lifecycle of a run paused for human review (`requires_action`). * @@ -42,20 +48,36 @@ export class ApprovalLifecycle { * Returns `false` when the job was not running (aborted mid-flight, gone), * so a late interrupt is dropped rather than pausing a dead job. */ - async pause(streamId: string, pendingAction: Agents.PendingAction): Promise { + async pause( + streamId: string, + pendingAction: Agents.PendingAction, + options?: ApprovalPauseOptions, + ): Promise { const job = await this.store.getJob(streamId); - if (!job || job.status !== 'running') { + if ( + !job || + job.status !== 'running' || + (options?.expectedCreatedAt != null && job.createdAt !== options.expectedCreatedAt) + ) { return false; } + const discoveredTools = options?.discoveredTools; + const expectedCreatedAt = options?.expectedCreatedAt ?? job.createdAt; const ok = await this.store.transitionStatus(streamId, { from: 'running', to: 'requires_action', // pendingActionId is the flat mirror the atomic resolve/expire guard on. - patch: { pendingAction, pendingActionId: pendingAction.actionId }, - expectCreatedAt: job.createdAt, + patch: { + pendingAction, + pendingActionId: pendingAction.actionId, + ...(discoveredTools != null && discoveredTools.length > 0 + ? { discoveredTools: [...discoveredTools] } + : {}), + }, + expectCreatedAt: expectedCreatedAt, }); if (ok) { - this.callbacks.onPaused?.(streamId, job.createdAt); + this.callbacks.onPaused?.(streamId, expectedCreatedAt); logger.debug( `[ApprovalLifecycle] paused for review: ${streamId} action=${pendingAction.actionId}`, ); diff --git a/packages/api/src/stream/__tests__/pendingAction.spec.ts b/packages/api/src/stream/__tests__/pendingAction.spec.ts index ee5dd3df92..deda3c9d75 100644 --- a/packages/api/src/stream/__tests__/pendingAction.spec.ts +++ b/packages/api/src/stream/__tests__/pendingAction.spec.ts @@ -59,6 +59,40 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () } }); + test('persists discovered tools in the same transition that makes the pause visible', async () => { + const streamId = 'stream-pause-discoveries'; + await manager.createJob(streamId, 'user-1'); + + const action = buildAction(streamId); + expect( + await manager.approvals.pause(streamId, action, { + discoveredTools: ['save_issue_mcp_linear'], + }), + ).toBe(true); + + const paused = await manager.getJob(streamId); + expect(paused?.status).toBe('requires_action'); + expect(paused?.metadata.discoveredTools).toEqual(['save_issue_mcp_linear']); + }); + + test('does not write a stale pause or discoveries onto a replacement job', async () => { + const streamId = 'stream-pause-replaced'; + const original = await manager.createJob(streamId, 'user-1'); + const replacement = await manager.createJob(streamId, 'user-1'); + + expect( + await manager.approvals.pause(streamId, buildAction(streamId), { + discoveredTools: ['stale_tool'], + expectedCreatedAt: original.createdAt, + }), + ).toBe(false); + + const liveJob = await manager.getJob(streamId); + expect(liveJob?.createdAt).toBe(replacement.createdAt); + expect(liveJob?.status).toBe('running'); + expect(liveJob?.metadata.discoveredTools).toBeUndefined(); + }); + test('returns false when the job is already terminal', async () => { const streamId = 'stream-pause-dead'; await manager.createJob(streamId, 'user-1'); diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index fe5313f50a..024c624faa 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -55,8 +55,8 @@ export interface SerializableJobData { /** * Deferred-tool names discovered (via `tool_search`) before a HITL pause, captured * so a resume can replay them into `createRun` โ€” the rebuilt graph uses `messages: []` - * (state comes from the checkpoint), so without these the paused deferred tool would be - * absent from the schema-only toolMap and resume would fail with "unknown tool". + * (state comes from the checkpoint), so without these the rebuilt model would lose + * the discovered tool schemas. */ discoveredTools?: string[]; /** diff --git a/packages/api/src/types/stream.ts b/packages/api/src/types/stream.ts index 30b8e6326b..2684bb83aa 100644 --- a/packages/api/src/types/stream.ts +++ b/packages/api/src/types/stream.ts @@ -27,7 +27,7 @@ export interface GenerationJobMetadata { /** * Deferred-tool names discovered (via `tool_search`) before a HITL pause. A resume * replays these into `createRun` because the rebuilt graph uses `messages: []`, so - * without them the paused deferred tool would be missing from the schema-only toolMap. + * without them the rebuilt model would lose the discovered tool schemas. */ discoveredTools?: string[]; /** See `SerializableJobData.preemptCapable`. */