CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-10-25 03:32:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 14

🛂 test: Cover Tool Approval Workflows End to End (#14427)

Browse source 
* test: cover tool approval workflows end to end

* fix: preserve tool approval state across resume

* fix: preserve agent context in mock stream responses

* fix: preserve nested approvals in collapsed groups
This commit is contained in:
Danny Avila 2026-07-26 21:58:25 -04:00 • committed by GitHub
parent f3159f9891
commit d8427ffc5e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
42 changed files with 2960 additions and 145 deletions
Download patch file Download diff file Expand all files Collapse all files

58
api/server/controllers/agents/__tests__/callbacks.spec.js
View file

@ -7,6 +7,9 @@ jest.mock('nanoid', () => ({
jest.mock('@librechat/api', () => ({
sendEvent: jest.fn(),
GenerationJobManager: {
emitChunk: jest.fn(),
},
HOST_FILE_AUTHORING_ARTIFACT_KEY: '__librechat_file_authoring',
getToolInputValidationDetails: jest.fn((result, validationError) =>
validationError != null
@ -74,6 +77,61 @@ jest.mock('~/server/services/Files/process', () => ({
saveBase64Image: jest.fn(),
}));
describe('resumable event generation fencing', () => {
beforeEach(() => {
jest.clearAllMocks();
});
it('forwards the originating job epoch with run-step events', async () => {
const { GenerationJobManager } = require('@librechat/api');
const { GraphEvents } = jest.requireActual('@librechat/agents');
const { getDefaultHandlers } = require('../callbacks');
const data = {
id: 'step-1',
index: 0,
stepDetails: {
type: 'tool_calls',
tool_calls: [{ id: 'call-1', name: 'approval_probe', args: '{}' }],
},
};
const handlers = getDefaultHandlers({
res: { write: jest.fn() },
aggregateContent: jest.fn(),
toolEndCallback: jest.fn(),
collectedUsage: [],
streamId: 'conversation-1',
jobCreatedAt: 1234,
});
await handlers[GraphEvents.ON_RUN_STEP].handle(GraphEvents.ON_RUN_STEP, data);
expect(GenerationJobManager.emitChunk).toHaveBeenCalledWith(
'conversation-1',
{ event: GraphEvents.ON_RUN_STEP, data },
{ expectedCreatedAt: 1234 },
);
});
it('forwards the originating job epoch with deferred attachments', () => {
const { GenerationJobManager } = require('@librechat/api');
const { createAttachmentEmitter } = require('../callbacks');
const attachment = { file_id: 'file-1', status: 'ready' };
const emitAttachment = createAttachmentEmitter({
res: { write: jest.fn() },
streamId: 'conversation-1',
jobCreatedAt: 1234,
});
emitAttachment(attachment);
expect(GenerationJobManager.emitChunk).toHaveBeenCalledWith(
'conversation-1',
{ event: 'attachment', data: attachment },
{ expectedCreatedAt: 1234 },
);
});
});
describe('createToolEndCallback', () => {
let req, res, artifactPromises, createToolEndCallback;
let logger;

27
api/server/controllers/agents/__tests__/resume.spec.js
View file

@ -646,6 +646,33 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
);
});
it('passes persisted run steps into the rebuilt run for tool-result correlation', async () => {
const runSteps = [
{
id: 'step-approval',
index: 1,
type: 'tool_calls',
stepDetails: {
type: 'tool_calls',
tool_calls: [{ id: 'tc1', name: 'approval_probe', args: '{}' }],
},
usage: null,
},
];
mockGenerationJobManager.getJob.mockResolvedValue(makeToolApprovalJob());
mockGenerationJobManager.getResumeState.mockResolvedValue({
aggregatedContent: [],
runSteps,
});
await post(approveBody());
await settled;
await flush();
const client = await mockInitializeClient.mock.results[0].value.then((r) => r.client);
expect(client.resumeCompletion).toHaveBeenCalledWith(expect.objectContaining({ runSteps }));
});
it('restores the paused user message files before reconstruction (execute-code files)', async () => {
mockGenerationJobManager.getJob.mockResolvedValue(makeToolApprovalJob());
// The resume body carries no files; the controller must source them from the

106
api/server/controllers/agents/callbacks.js
View file

@ -222,11 +222,12 @@ function checkIfLastAgent(last_agent_id, langgraph_node) {
* @param {ServerResponse} res - The server response object
* @param {string | null} streamId - The stream ID for resumable mode, or null for standard mode
* @param {Object} eventData - The event data to send
* @param {number} [expectedCreatedAt] - The generation epoch that produced the event
* @returns {Promise<void>}
*/
async function emitEvent(res, streamId, eventData) {
async function emitEvent(res, streamId, eventData, expectedCreatedAt) {
if (streamId) {
await GenerationJobManager.emitChunk(streamId, eventData);
await GenerationJobManager.emitChunk(streamId, eventData, { expectedCreatedAt });
} else {
sendEvent(res, eventData);
}
@ -239,13 +240,12 @@ async function emitEvent(res, streamId, eventData) {
* running state. Only signals while a fired prewarm remains unresolved
* ({@link shouldSignalSandboxStart}); stateless deployments never fire one
* and completed boots clear the marker, so both stay on the generic label.
* @param {ServerResponse} res - The server response object
* @param {string | null} streamId - The stream ID for resumable mode, or null for standard mode
* @param {(eventData: Object) => Promise<void>} emitForJob - Generation-fenced event emitter
* @param {StreamEventData} data - The `on_run_step` event data
* @param {GraphRunnableConfig['configurable']} [metadata] The runnable metadata
* @returns {Promise<void>}
*/
async function maybeEmitSandboxStarting(res, streamId, data, metadata) {
async function maybeEmitSandboxStarting(emitForJob, data, metadata) {
const conversationId = metadata?.thread_id;
if (!conversationId || !(await shouldSignalSandboxStart(conversationId))) {
return;
@ -256,7 +256,7 @@ async function maybeEmitSandboxStarting(res, streamId, data, metadata) {
if (!toolCall?.id || name == null || !isCodeSessionToolName(name)) {
continue;
}
await emitEvent(res, streamId, {
await emitForJob({
event: StepEvents.ON_SANDBOX_STARTING,
data: { tool_call_id: toolCall.id, runId: metadata?.run_id },
});
@ -323,6 +323,7 @@ function feedSubagentAggregator(aggregator, event) {
* @param {ToolEndCallback} options.toolEndCallback - Callback to use when tool ends.
* @param {Array<UsageMetadata>} options.collectedUsage - The list of collected usage metadata.
* @param {string | null} [options.streamId] - The stream ID for resumable mode, or null for standard mode.
* @param {number} [options.jobCreatedAt] - The generation epoch that owns emitted events.
* @param {ToolExecuteOptions} [options.toolExecuteOptions] - Options for event-driven tool execution.
* @param {UsageCostDeps} [options.usageCost] - Pricing context for authoritative per-event cost.
* @param {{ latest: TContextUsageEvent | null, count: number }} [options.contextUsageSink] - Mutable
@ -343,6 +344,7 @@ function getDefaultHandlers({
collectedUsage,
collectedThoughtSignatures = null,
streamId = null,
jobCreatedAt,
toolExecuteOptions = null,
summarizationOptions = null,
subagentAggregatorsByToolCallId = null,
@ -355,6 +357,7 @@ function getDefaultHandlers({
`[getDefaultHandlers] Missing required options: res: ${!res}, aggregateContent: ${!aggregateContent}`,
);
}
const emitForJob = (eventData) => emitEvent(res, streamId, eventData, jobCreatedAt);
/**
* Emit a token-usage event, attaching the authoritative per-event USD cost
* when cost display is enabled. The backend is the single source of truth
@ -385,7 +388,7 @@ function getDefaultHandlers({
if (usageEmitSink) {
usageEmitSink.push(payload);
}
return emitEvent(res, streamId, { event: UsageEvents.ON_TOKEN_USAGE, data: payload });
return emitForJob({ event: UsageEvents.ON_TOKEN_USAGE, data: payload });
};
const handlers = {
[GraphEvents.CHAT_MODEL_END]: new ModelEndHandler(
@ -404,17 +407,17 @@ function getDefaultHandlers({
handle: async (event, data, metadata) => {
aggregateContent({ event, data });
if (data?.stepDetails.type === StepTypes.TOOL_CALLS) {
await emitEvent(res, streamId, { event, data });
await maybeEmitSandboxStarting(res, streamId, data, metadata);
await emitForJob({ event, data });
await maybeEmitSandboxStarting(emitForJob, data, metadata);
} else if (checkIfLastAgent(metadata?.last_agent_id, metadata?.langgraph_node)) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else if (!metadata?.hide_sequential_outputs) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else {
const agentName = metadata?.name ?? 'Agent';
const isToolCall = data?.stepDetails.type === StepTypes.TOOL_CALLS;
const action = isToolCall ? 'performing a task...' : 'thinking...';
await emitEvent(res, streamId, {
await emitForJob({
event: 'on_agent_update',
data: {
runId: metadata?.run_id,
@ -434,11 +437,11 @@ function getDefaultHandlers({
handle: async (event, data, metadata) => {
aggregateContent({ event, data });
if (data?.delta.type === StepTypes.TOOL_CALLS) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else if (checkIfLastAgent(metadata?.last_agent_id, metadata?.langgraph_node)) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else if (!metadata?.hide_sequential_outputs) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
}
},
},
@ -477,11 +480,11 @@ function getDefaultHandlers({
}
}
if (data?.result != null) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else if (checkIfLastAgent(metadata?.last_agent_id, metadata?.langgraph_node)) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else if (!metadata?.hide_sequential_outputs) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
}
},
},
@ -495,9 +498,9 @@ function getDefaultHandlers({
handle: async (event, data, metadata) => {
aggregateContent({ event, data });
if (checkIfLastAgent(metadata?.last_agent_id, metadata?.langgraph_node)) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else if (!metadata?.hide_sequential_outputs) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
}
},
},
@ -511,9 +514,9 @@ function getDefaultHandlers({
handle: async (event, data, metadata) => {
aggregateContent({ event, data });
if (checkIfLastAgent(metadata?.last_agent_id, metadata?.langgraph_node)) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
} else if (!metadata?.hide_sequential_outputs) {
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
}
},
},
@ -568,14 +571,14 @@ function getDefaultHandlers({
);
}
}
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
},
};
if (summarizationOptions?.enabled !== false) {
handlers[GraphEvents.ON_SUMMARIZE_START] = {
handle: async (_event, data) => {
await emitEvent(res, streamId, {
await emitForJob({
event: GraphEvents.ON_SUMMARIZE_START,
data,
});
@ -584,7 +587,7 @@ function getDefaultHandlers({
handlers[GraphEvents.ON_SUMMARIZE_DELTA] = {
handle: async (_event, data) => {
aggregateContent({ event: GraphEvents.ON_SUMMARIZE_DELTA, data });
await emitEvent(res, streamId, {
await emitForJob({
event: GraphEvents.ON_SUMMARIZE_DELTA,
data,
});
@ -593,7 +596,7 @@ function getDefaultHandlers({
handlers[GraphEvents.ON_SUMMARIZE_COMPLETE] = {
handle: async (_event, data) => {
aggregateContent({ event: GraphEvents.ON_SUMMARIZE_COMPLETE, data });
await emitEvent(res, streamId, {
await emitForJob({
event: GraphEvents.ON_SUMMARIZE_COMPLETE,
data,
});
@ -634,7 +637,7 @@ function getDefaultHandlers({
contextUsageSink.count = (contextUsageSink.count ?? 0) + 1;
contextUsageSink.latestUsageIndex = usageEmitSink?.length ?? 0;
}
await emitEvent(res, streamId, { event, data });
await emitForJob({ event, data });
}
},
};
@ -649,10 +652,15 @@ function getDefaultHandlers({
* @param {ServerResponse} res - The server response object
* @param {string | null} streamId - The stream ID for resumable mode, or null for standard mode
* @param {Object} attachment - The attachment data
* @param {number} [expectedCreatedAt] - The generation epoch that produced the attachment
*/
function writeAttachment(res, streamId, attachment) {
function writeAttachment(res, streamId, attachment, expectedCreatedAt) {
if (streamId) {
GenerationJobManager.emitChunk(streamId, { event: 'attachment', data: attachment });
GenerationJobManager.emitChunk(
streamId,
{ event: 'attachment', data: attachment },
{ expectedCreatedAt },
);
} else {
res.write(`event: attachment\ndata: ${JSON.stringify(attachment)}\n\n`);
}
@ -699,12 +707,13 @@ function isStreamWritable(res, streamId) {
* @param {ServerResponse} res
* @param {string | null} streamId
* @param {Object} attachment - Updated attachment payload (must carry `file_id`).
* @param {number} [expectedCreatedAt] - The generation epoch that produced the attachment
*/
function writeAttachmentUpdate(res, streamId, attachment) {
function writeAttachmentUpdate(res, streamId, attachment, expectedCreatedAt) {
if (!isStreamWritable(res, streamId)) {
return;
}
writeAttachment(res, streamId, attachment);
writeAttachment(res, streamId, attachment, expectedCreatedAt);
}
/**
@ -714,9 +723,10 @@ function writeAttachmentUpdate(res, streamId, attachment) {
* @param {ServerResponse} params.res
* @param {Promise<MongoFile | { filename: string; filepath: string; expires: number;} | null>[]} params.artifactPromises
* @param {string | null} [params.streamId] - The stream ID for resumable mode, or null for standard mode.
* @param {number} [params.jobCreatedAt] - The generation epoch that owns emitted attachments.
* @returns {ToolEndCallback} The tool end callback.
*/
function createToolEndCallback({ req, res, artifactPromises, streamId = null }) {
function createToolEndCallback({ req, res, artifactPromises, streamId = null, jobCreatedAt }) {
/**
* @type {ToolEndCallback}
*/
@ -747,7 +757,7 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
if (!streamId && !res.headersSent) {
return attachment;
}
writeAttachment(res, streamId, attachment);
writeAttachment(res, streamId, attachment, jobCreatedAt);
return attachment;
})().catch((error) => {
logger.error('Error processing file citations:', error);
@ -769,7 +779,7 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
if (!streamId && !res.headersSent) {
return attachment;
}
writeAttachment(res, streamId, attachment);
writeAttachment(res, streamId, attachment, jobCreatedAt);
return attachment;
})().catch((error) => {
logger.error('Error processing artifact content:', error);
@ -791,7 +801,7 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
if (!streamId && !res.headersSent) {
return attachment;
}
writeAttachment(res, streamId, attachment);
writeAttachment(res, streamId, attachment, jobCreatedAt);
return attachment;
})().catch((error) => {
logger.error('Error processing artifact content:', error);
@ -813,7 +823,7 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
if (!streamId && !res.headersSent) {
return attachment;
}
writeAttachment(res, streamId, attachment);
writeAttachment(res, streamId, attachment, jobCreatedAt);
return attachment;
})().catch((error) => {
logger.error('Error processing memory artifact content:', error);
@ -858,7 +868,7 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
return null;
}
writeAttachment(res, streamId, fileMetadata);
writeAttachment(res, streamId, fileMetadata, jobCreatedAt);
return fileMetadata;
})().catch((error) => {
logger.error('Error processing artifact content:', error);
@ -935,7 +945,7 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
* IIFE catch but logged as noise). Same gate the Responses
* path uses below. */
if (isStreamWritable(res, streamId)) {
writeAttachment(res, streamId, fileMetadata);
writeAttachment(res, streamId, fileMetadata, jobCreatedAt);
}
/* Deferred preview rendering: extraction continues running
* even after the HTTP response closes. If the stream is still
@ -958,11 +968,16 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
fileId: fileMetadata.file_id,
previewRevision: result?.previewRevision,
onResolved: (updated) => {
writeAttachmentUpdate(res, streamId, {
...updated,
messageId: metadata.run_id,
toolCallId,
});
writeAttachmentUpdate(
res,
streamId,
{
...updated,
messageId: metadata.run_id,
toolCallId,
},
jobCreatedAt,
);
},
});
return fileMetadata;
@ -983,14 +998,15 @@ function createToolEndCallback({ req, res, artifactPromises, streamId = null })
* @param {Object} params
* @param {ServerResponse} params.res
* @param {string | null} [params.streamId]
* @param {number} [params.jobCreatedAt]
* @returns {(attachment: Object) => void}
*/
function createAttachmentEmitter({ res, streamId = null }) {
function createAttachmentEmitter({ res, streamId = null, jobCreatedAt }) {
return (attachment) => {
if (!attachment || !isStreamWritable(res, streamId)) {
return;
}
writeAttachment(res, streamId, attachment);
writeAttachment(res, streamId, attachment, jobCreatedAt);
};
}

37
api/server/controllers/agents/client.js
View file

@ -44,6 +44,7 @@ const {
deleteAgentCheckpoint,
agentRequestsAskUserQuestion,
attachAskUserQuestionArgs,
hydrateResumeRunSteps,
createContentIndexOffsetHandlers,
createSteerIndexOffsetHandlers,
createSteerDrainHook,
@ -140,6 +141,7 @@ class AgentClient extends BaseClient {
const {
agentConfigs,
contentParts,
stepMap,
collectedUsage,
collectedThoughtSignatures,
artifactPromises,
@ -169,6 +171,10 @@ class AgentClient extends BaseClient {
this.toolInputValidationErrors = toolInputValidationErrors;
/** @type {MessageContentComplex[]} */
this.contentParts = contentParts;
/** Original run-step identity used by the content aggregator to attach
* completion events to their rendered content indices.
* @type {Map<string, import('@librechat/agents').RunStep | undefined> | undefined} */
this.stepMap = stepMap;
/** @type {Array<UsageMetadata>} */
this.collectedUsage = collectedUsage;
/** Vertex Gemini 3 thought signatures captured during the run, keyed by
@ -302,7 +308,7 @@ class AgentClient extends BaseClient {
conversationId: this.conversationId,
},
},
{ durable: true },
{ durable: true, expectedCreatedAt: this.jobCreatedAt },
);
}
@ -983,6 +989,7 @@ class AgentClient extends BaseClient {
config,
messageId,
streamId,
jobCreatedAt: this.jobCreatedAt,
conversationId,
memoryMethods: {
setMemory: db.setMemory,
@ -1333,10 +1340,14 @@ class AgentClient extends BaseClient {
const emit = (async () => {
try {
if (streamId) {
await GenerationJobManager.emitChunk(streamId, {
event: UsageEvents.ON_TOKEN_USAGE,
data,
});
await GenerationJobManager.emitChunk(
streamId,
{
event: UsageEvents.ON_TOKEN_USAGE,
data,
},
{ expectedCreatedAt: this.jobCreatedAt },
);
} else {
sendEvent(res, { event: UsageEvents.ON_TOKEN_USAGE, data });
}
@ -1533,10 +1544,14 @@ class AgentClient extends BaseClient {
logger.error(`[AgentClient] Failed to release request slot on pause ${streamId}`, err);
}
}
await GenerationJobManager.emitChunk(streamId, {
event: ApprovalEvents.ON_PENDING_ACTION,
data: toClientPendingAction(pendingAction),
});
await GenerationJobManager.emitChunk(
streamId,
{
event: ApprovalEvents.ON_PENDING_ACTION,
data: toClientPendingAction(pendingAction),
},
{ expectedCreatedAt: this.jobCreatedAt },
);
// Steers queued before this pause stay IN the store for the whole approval
// window: `resumeState.pendingSteers` re-seeds the client's chips on
// reload, and the resumed run drains them at its first tool boundary.
@ -2067,12 +2082,14 @@ class AgentClient extends BaseClient {
* @param {object} params
* @param {Agents.ToolApprovalDecisionMap | { answer: string }} params.resumeValue
* @param {Array} [params.seedContent] - content aggregated before the pause
* @param {Array<import('@librechat/agents').RunStep>} [params.runSteps] - run steps emitted before the pause
* @param {AbortController} [params.abortController]
* @param {Pick<import('@langchain/langgraph').Command, 'update' | 'goto'>} [params.commandOptions]
*/
async resumeCompletion({
resumeValue,
seedContent = [],
runSteps = [],
abortController = null,
commandOptions,
userMCPAuthMap,
@ -2203,6 +2220,8 @@ class AgentClient extends BaseClient {
throw new Error('Failed to create run for resume');
}
hydrateResumeRunSteps(runSteps, this.stepMap, run.Graph, seedContent);
this.run = run;
if (this._resolveRun) {
this._resolveRun(run);

23
api/server/controllers/agents/client.test.js
View file

@ -2922,6 +2922,29 @@ describe('AgentClient - titleConvo', () => {
);
});
it('should bind memory processing to the current generation epoch', async () => {
mockReq._resumableStreamId = 'convo-123';
mockCheckAccess.mockResolvedValue(true);
mockInitializeAgent.mockResolvedValue({
...mockAgent,
provider: EModelEndpoint.openAI,
});
mockCreateMemoryProcessor.mockResolvedValue([undefined, jest.fn()]);
client = new AgentClient({ ...mockOptions, jobCreatedAt: 1234 });
client.conversationId = 'convo-123';
client.responseMessageId = 'response-123';
await client.useMemory();
expect(mockCreateMemoryProcessor).toHaveBeenCalledWith(
expect.objectContaining({
streamId: 'convo-123',
jobCreatedAt: 1234,
}),
);
});
it('should load different agent when memory config agent.id differs from current agent id', async () => {
const differentAgentId = 'different-agent-456';
const differentAgent = {

60
api/server/controllers/agents/request.js
View file

@ -592,13 +592,17 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
if (titleAbortController.signal.aborted) {
return;
}
await GenerationJobManager.emitChunk(streamId, {
event: 'title',
data: {
conversationId: titleConversationId,
title,
await GenerationJobManager.emitChunk(
streamId,
{
event: 'title',
data: {
conversationId: titleConversationId,
title,
},
},
});
{ expectedCreatedAt: jobCreatedAt },
);
})().catch((err) => {
logger.error('[ResumableAgentController] Error emitting title event', err);
});
@ -643,27 +647,31 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
logger.error('[ResumableAgentController] Failed to persist start metadata', err);
});
GenerationJobManager.emitChunk(streamId, {
created: true,
// Skill selections aren't on `userMessage` yet at onStart (BaseClient adds
// them later), so attach them from the request — this is the message
// `trackUserMessage` persists as the authoritative job.metadata.userMessage,
// and it's what the live client renders the user bubble from.
message: {
...userMessage,
// Carry files so trackUserMessage (the authoritative writer) persists them on
// job.metadata.userMessage for a HITL resume (see the updateMetadata above).
...(Array.isArray(req.body?.files) &&
req.body.files.length > 0 && { files: req.body.files }),
...(Array.isArray(req.body?.manualSkills) &&
req.body.manualSkills.length > 0 && { manualSkills: req.body.manualSkills }),
...(Array.isArray(req.body?.alwaysAppliedSkills) &&
req.body.alwaysAppliedSkills.length > 0 && {
alwaysAppliedSkills: req.body.alwaysAppliedSkills,
}),
},
GenerationJobManager.emitChunk(
streamId,
}).catch((err) => {
{
created: true,
// Skill selections aren't on `userMessage` yet at onStart (BaseClient adds
// them later), so attach them from the request — this is the message
// `trackUserMessage` persists as the authoritative job.metadata.userMessage,
// and it's what the live client renders the user bubble from.
message: {
...userMessage,
// Carry files so trackUserMessage (the authoritative writer) persists them on
// job.metadata.userMessage for a HITL resume (see the updateMetadata above).
...(Array.isArray(req.body?.files) &&
req.body.files.length > 0 && { files: req.body.files }),
...(Array.isArray(req.body?.manualSkills) &&
req.body.manualSkills.length > 0 && { manualSkills: req.body.manualSkills }),
...(Array.isArray(req.body?.alwaysAppliedSkills) &&
req.body.alwaysAppliedSkills.length > 0 && {
alwaysAppliedSkills: req.body.alwaysAppliedSkills,
}),
},
streamId,
},
{ expectedCreatedAt: jobCreatedAt },
).catch((err) => {
logger.error('[ResumableAgentController] Failed to queue created event', err);
});
};

13
api/server/controllers/agents/resume.js
View file

@ -327,10 +327,14 @@ async function finalizeResumedTurn({
client,
onTitleGenerated: ({ conversationId: titleConvoId, title }) => {
conversation.title = title;
return GenerationJobManager.emitChunk(streamId, {
event: 'title',
data: { conversationId: titleConvoId, title },
});
return GenerationJobManager.emitChunk(
streamId,
{
event: 'title',
data: { conversationId: titleConvoId, title },
},
{ expectedCreatedAt: job.createdAt },
);
},
});
} catch (err) {
@ -695,6 +699,7 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
await client.resumeCompletion({
resumeValue: mapped.resumeValue,
seedContent,
runSteps: resumeState?.runSteps ?? [],
abortController: job.abortController,
// Carry the user's MCP auth so approved MCP tools run with their credentials.
userMCPAuthMap: result.userMCPAuthMap,