diff --git a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts index e7d796a72a5..d608d04ec9e 100644 --- a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts +++ b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts @@ -2206,17 +2206,17 @@ function createRuntimeWithCheckpointStore( } function makeClaimingCheckpointStore(): ClaimingCheckpointStore { - const checkpoints = new Map(); + const checkpoints = new Map(); return { save: vi.fn(async (key: string, state: SerializableAgentState): Promise => { await Promise.resolve(); - checkpoints.set(key, structuredClone(state)); + checkpoints.set(key, JSON.stringify(state)); }), load: vi.fn(async (key: string): Promise => { await Promise.resolve(); const state = checkpoints.get(key); - return state ? structuredClone(state) : undefined; + return state ? (JSON.parse(state) as SerializableAgentState) : undefined; }), delete: vi.fn(async (key: string): Promise => { await Promise.resolve(); @@ -2225,8 +2225,8 @@ function makeClaimingCheckpointStore(): ClaimingCheckpointStore { claimForResume: vi.fn(async (key, state) => { await Promise.resolve(); const current = checkpoints.get(key); - if (!current || JSON.stringify(current) !== JSON.stringify(state)) return false; - checkpoints.set(key, { ...state, status: 'running' }); + if (!current || current !== JSON.stringify(state)) return false; + checkpoints.set(key, JSON.stringify({ ...state, status: 'running' })); return true; }), }; @@ -5732,6 +5732,86 @@ describe('AgentRuntime.resume() — checkpoint lifecycle', () => { ).rejects.toThrow(`No suspended run found for runId: ${runId}`); }); + it.each(['generate', 'stream'] as const)( + 'restores and updates host metadata across repeated %s resumes', + async (method) => { + const checkpointStore = makeClaimingCheckpointStore(); + const observed: Array = []; + const tool = makeSuspendingTool('suspend_tool', async (_input, ctx) => { + if (!ctx.resumeData) return await ctx.suspend({ reason: 'needs approval' }); + observed.push(structuredClone(ctx.persistence)); + ctx.persistence!.hostMetadata!.actor = 'tool-update'; + return { approved: true }; + }); + const persistence = { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: { scope: { tenant: 'tenant-1' }, actor: 'original' }, + }; + generateText.mockResolvedValueOnce( + makeGenerateWithToolCalls([ + { toolCallId: 'tc-1', toolName: 'suspend_tool', args: {} }, + { toolCallId: 'tc-2', toolName: 'suspend_tool', args: {} }, + ]), + ); + const initial = createRuntimeWithCheckpointStore([tool], checkpointStore); + const { runId } = await initial.generate('run tools', { persistence }); + const checkpoint = await checkpointStore.load(runId); + const onResumeClaimed = vi.fn(async () => { + expect(await checkpointStore.load(runId)).toEqual({ ...checkpoint, status: 'running' }); + }); + const resumed = createRuntimeWithCheckpointStore([tool], checkpointStore); + const options = { + runId, + toolCallId: 'tc-1', + hostMetadata: { actor: 'selected' }, + onResumeClaimed, + }; + if (method === 'stream') { + await collectChunks((await resumed.resume('stream', { approved: true }, options)).stream); + } else { + await resumed.resume('generate', { approved: true }, options); + } + expect(observed).toEqual([ + { ...persistence, hostMetadata: { scope: { tenant: 'tenant-1' }, actor: 'selected' } }, + ]); + expect(onResumeClaimed).toHaveBeenCalledOnce(); + expect((await checkpointStore.load(runId))?.persistence).toEqual({ + ...persistence, + hostMetadata: { scope: { tenant: 'tenant-1' }, actor: 'tool-update' }, + }); + expect(persistence.hostMetadata.actor).toBe('original'); + + generateText.mockResolvedValueOnce(makeGenerateSuccess('done')); + const next = createRuntimeWithCheckpointStore([tool], checkpointStore); + await next.resume('generate', { approved: true }, { runId, toolCallId: 'tc-2' }); + expect(observed[1]?.hostMetadata?.actor).toBe('tool-update'); + expect(await checkpointStore.load(runId)).toBeUndefined(); + }, + ); + + it('rejects resume metadata without persistence before claiming the checkpoint', async () => { + const checkpointStore = makeClaimingCheckpointStore(); + const runtime = createRuntimeWithCheckpointStore([makeApprovalTool()], checkpointStore); + generateText.mockResolvedValueOnce( + makeGenerateWithToolCalls([{ toolCallId: 'tc-1', toolName: 'suspend_tool', args: {} }]), + ); + const { runId } = await runtime.generate('run tool'); + await expect( + runtime.resume( + 'generate', + { approved: true }, + { + runId, + toolCallId: 'tc-1', + hostMetadata: { actor: 'selected' }, + }, + ), + ).rejects.toThrow('Cannot update host metadata without persistence'); + expect(checkpointStore.claimForResume).not.toHaveBeenCalled(); + expect((await checkpointStore.load(runId))?.status).toBe('suspended'); + }); + it('claims the checkpoint after resume validation passes', async () => { const checkpointStore = makeClaimingCheckpointStore(); const runtime = createRuntimeWithCheckpointStore([makeApprovalTool()], checkpointStore); @@ -5777,16 +5857,28 @@ describe('AgentRuntime.resume() — checkpoint lifecycle', () => { generateText.mockResolvedValueOnce( makeGenerateWithToolCalls([{ toolCallId: 'tc-1', toolName: 'suspend_tool', args: {} }]), ); - const first = await runtime.generate('run tool'); + const first = await runtime.generate('run tool', { + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: { actor: 'original' }, + }, + }); const { runId, toolCallId } = first.pendingSuspend![0]; + const checkpoint = await checkpointStore.load(runId); checkpointStore.claimForResume.mockResolvedValueOnce(false); await expect( - runtime.resume('stream', { approved: true }, { runId, toolCallId, onResumeClaimed }), + runtime.resume( + 'stream', + { approved: true }, + { runId, toolCallId, onResumeClaimed, hostMetadata: { actor: 'selected' } }, + ), ).rejects.toBeInstanceOf(StaleResumeError); expect(errorEvents).toEqual([]); expect(streamText).not.toHaveBeenCalled(); expect(onResumeClaimed).not.toHaveBeenCalled(); + expect(await checkpointStore.load(runId)).toEqual(checkpoint); }); it('does not invoke the claim hook when checkpoint claiming fails', async () => { diff --git a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts index 9eb6ab33c80..00dae91f95e 100644 --- a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts +++ b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts @@ -399,6 +399,9 @@ export class AgentRuntime { if (!toolCall) { throw new StaleResumeError(`No tool call found for toolCallId: ${options.toolCallId}`); } + if (options.hostMetadata !== undefined && !state.persistence) { + throw new Error('Cannot update host metadata without persistence'); + } const list = AgentMessageList.deserialize(state.messageList); this.context.hydrateDeferredToolsFromList(list); @@ -430,6 +433,7 @@ export class AgentRuntime { runId: _rid, toolCallId: _tcid, onResumeClaimed: _onResumeClaimed, + hostMetadata, ...callerExecOptions } = options; const persisted = state.executionOptions ?? {}; @@ -452,16 +456,22 @@ export class AgentRuntime { ...(state.iterationCount !== undefined ? { iterationCount: state.iterationCount } : {}), }; - const resumeOptions: RuntimeExecutionOptions = { - persistence: state.persistence, - ...mergedExecOptions, - }; - const claimed = await this.runState.claimResume(this.runId, state); if (!claimed) { throw new StaleResumeError(`Run ${this.runId} is not suspended. Cannot resume.`); } resumeClaimed = true; + const resumeOptions: RuntimeExecutionOptions = { + persistence: state.persistence + ? { + ...state.persistence, + ...(state.persistence.hostMetadata || hostMetadata + ? { hostMetadata: { ...state.persistence.hostMetadata, ...hostMetadata } } + : {}), + } + : undefined, + ...mergedExecOptions, + }; await options.onResumeClaimed?.(); abortScope = this.eventBus.createAbortScope(resumeOptions.abortSignal); diff --git a/packages/@n8n/agents/src/types/sdk/agent.ts b/packages/@n8n/agents/src/types/sdk/agent.ts index 4a6a9eb7847..32762cea5c4 100644 --- a/packages/@n8n/agents/src/types/sdk/agent.ts +++ b/packages/@n8n/agents/src/types/sdk/agent.ts @@ -336,6 +336,8 @@ export interface StreamResult { export interface ResumeOptions { runId: string; toolCallId: string; + /** Merge these host metadata keys after the resume claim succeeds. Requires persistence. */ + hostMetadata?: JSONObject; /** @internal Host lifecycle hook invoked after the checkpoint claim succeeds. */ onResumeClaimed?: () => void | Promise; } diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index ba979ca84aa..118a3f37bf8 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -1,5 +1,6 @@ import type { Agent as RuntimeAgent, + ResumeOptions, JSONValue, SerializableAgentState, StreamChunk, @@ -36,8 +37,15 @@ import type { AgentSandboxRuntimeService } from '../agent-sandbox-runtime.servic import { AgentWakeService } from '../background/agent-wake.service'; import type { AgentChatBridge } from '../integrations/agent-chat-bridge'; import { ChatIntegrationService } from '../integrations/chat-integration.service'; -import type { IntegrationMessageContextService } from '../integrations/integration-message-context.service'; +import { IntegrationMessageContextService } from '../integrations/integration-message-context.service'; +import type { IntegrationMessageContext } from '../integrations/integration-tool-types'; import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage'; +import type { AgentThreadRepository } from '../repositories/agent-thread.repository'; +import type { AgentResourceRepository } from '../repositories/agent-resource.repository'; +import { + encodeIntegrationMessageContext, + readIntegrationMessageContext, +} from '../integrations/integration-message-context'; import type { ToolRegistry } from '../tool-registry'; const aiConfigMock = mock({ @@ -62,6 +70,15 @@ const integrationPrincipalHash = hashAgentSandboxPrincipal({ }); const taskPrincipalHash = hashAgentSandboxPrincipal({ type: 'scheduled-task', taskId: 'task-1' }); +const selectedContext: IntegrationMessageContext = { + integrationConnectionId: 'slack:credential-1', + platform: 'slack', + target: { type: 'thread', threadId: 'slack:channel:1' }, + interactingUserId: 'selected-user', + messageId: 'selected-message', + updatedAt: '2026-09-18T10:00:00.000Z', +}; + const schema: AgentJsonConfig = { name: 'Support Agent', model: 'anthropic/claude-sonnet-4-5', @@ -145,7 +162,15 @@ function makeService(sandboxEnabled = false) { const executionService = mock(); const telemetry = mock(); const runtimeCacheService = mock(); - const integrationMessageContextService = mock(); + const contextService = new IntegrationMessageContextService( + mock(), + mock(), + mockLogger(), + ); + const integrationMessageContextService = Object.assign(contextService, { + getLatest: vi.spyOn(contextService, 'getLatest'), + setLatest: vi.spyOn(contextService, 'setLatest').mockResolvedValue(undefined), + }); const agentRunTracingService = mock(); const externalHooks = mock(); const agentSandboxRuntimeService = mock({ @@ -708,10 +733,12 @@ describe('AgentExecutionOrchestratorService', () => { persistence: { threadId: 'thread-1', resourceId: 'resource-1', - hostMetadata: encodeAgentSandboxHostMetadata({ - projectId, - principalHash: userPrincipalHash, - }), + hostMetadata: expect.objectContaining( + encodeAgentSandboxHostMetadata({ + projectId, + principalHash: userPrincipalHash, + }), + ), }, executionCounter: expect.any(Object), abortSignal: abortController.signal, @@ -973,6 +1000,58 @@ describe('AgentExecutionOrchestratorService', () => { ); }); + it.each([false, true])( + 'installs a captured context before starting, including when metadata writes fail: %s', + async (writeFails) => { + const { service, runtimeCacheService, integrationMessageContextService } = makeService(); + const runtime = makeRuntime(); + runtimeCacheService.getRuntime.mockResolvedValue(runtime); + if (writeFails) + integrationMessageContextService.setLatest.mockRejectedValue( + new Error('database unavailable'), + ); + const memory = { threadId: 'task-run-1', resourceId: 'task:task-1' }; + const contextConversation = { threadId: 'platform-thread', resourceId: 'selected-user' }; + await collect( + service.executeForChatPublished({ + agentId, + projectId, + message: 'hello', + memory, + contextConversation, + messageContext: selectedContext, + integrationType: 'slack', + sandboxPrincipalHash: taskPrincipalHash, + }), + ); + expect(runtime.agent.stream).toHaveBeenCalledWith( + 'hello', + expect.objectContaining({ + persistence: { + ...memory, + hostMetadata: { + ...encodeAgentSandboxHostMetadata({ projectId, principalHash: taskPrincipalHash }), + ...encodeIntegrationMessageContext(selectedContext), + }, + }, + }), + ); + expect(integrationMessageContextService.setLatest).toHaveBeenCalledWith( + 'platform-thread', + 'selected-user', + selectedContext, + ); + expect(integrationMessageContextService.setLatest).toHaveBeenCalledWith( + 'task-run-1', + 'task:task-1', + selectedContext, + ); + expect(integrationMessageContextService.setLatest.mock.invocationCallOrder[1]).toBeLessThan( + runtime.agent.stream.mock.invocationCallOrder[0], + ); + }, + ); + it('records a failed session and rethrows when the published runtime cannot be built', async () => { const { service, runtimeCacheService, executionService, agentRepository } = makeService(); const buildError = new UserError('Credential "OpenAI" not found'); @@ -1331,8 +1410,21 @@ describe('AgentExecutionOrchestratorService', () => { ); it('delivers a published wake through the stored connection and reply thread', async () => { - const { service, runtimeCacheService, externalHooks, chatIntegrationService, bridge } = - makeService(); + const { + service, + runtimeCacheService, + externalHooks, + chatIntegrationService, + bridge, + integrationMessageContextService, + } = makeService(); + const messageContext = { + ...selectedContext, + replyTarget: { type: 'thread' as const, threadId: 'slack:channel-1:1' }, + }; + integrationMessageContextService.getLatest + .mockResolvedValueOnce(messageContext) + .mockRejectedValueOnce(new Error('Unexpected second context read')); const chunks: StreamChunk[] = [ { type: 'text-delta', id: 'text-1', delta: 'The job is done.' }, { type: 'finish', finishReason: 'stop' }, @@ -1364,6 +1456,9 @@ describe('AgentExecutionOrchestratorService', () => { expect(externalHooks.run).toHaveBeenCalledWith('agent.preExecute', [agentId]); expect(chatIntegrationService.getBridge).toHaveBeenCalledWith(agentId, 'slack', 'credential-1'); expect(bridge.deliverWakeResponse).toHaveBeenCalledWith('slack:channel-1:1', chunks); + expect( + readIntegrationMessageContext(runtime.agent.stream.mock.calls[0][1].persistence), + ).toEqual(messageContext); }); it('rejects a wake when chat delivery fails and releases the runtime', async () => { @@ -1661,6 +1756,77 @@ describe('AgentExecutionOrchestratorService', () => { ); }); + it.each([false, true])( + 'installs an interaction on the checkpoint memory only after a successful claim: %s', + async (claimed) => { + const { service, runtimeCacheService, checkpointStorage, integrationMessageContextService } = + makeService(); + const previous: IntegrationMessageContext = { + ...selectedContext, + interactingUserId: 'original-user', + messageId: 'original-message', + agentUserId: 'bot', + subject: { type: 'issue', id: 'issue-1' }, + }; + const memory = { threadId: 'task-run-1', resourceId: 'task:task-1' }; + checkpointStorage.getStatus.mockResolvedValue({ + status: 'active', + checkpoint: makeCheckpoint( + {}, + { + ...memory, + hostMetadata: encodeIntegrationMessageContext(previous), + }, + ), + }); + const runtime = makeRuntime(); + runtimeCacheService.getRuntime.mockResolvedValue(runtime); + runtime.agent.resume.mockImplementation(async (_method, _data, options: ResumeOptions) => { + expect(integrationMessageContextService.setLatest).not.toHaveBeenCalled(); + expect( + readIntegrationMessageContext({ ...memory, hostMetadata: options.hostMetadata }), + ).toEqual({ + ...selectedContext, + agentUserId: 'bot', + subject: { type: 'issue', id: 'issue-1' }, + }); + if (!claimed) throw new Error('Already handled'); + await options.onResumeClaimed?.(); + return { + runId: 'run-1', + stream: makeReadableStream([{ type: 'finish', finishReason: 'stop' }]), + }; + }); + const result = collect( + service.resumeForChat({ + agentId, + projectId, + runId: 'run-1', + toolCallId: 'tc-1', + resumeData: { approved: true }, + messageContext: selectedContext, + contextConversation: { threadId: 'platform-thread', resourceId: 'selected-user' }, + }), + ); + if (claimed) { + await result; + expect(integrationMessageContextService.setLatest).toHaveBeenCalledWith( + 'task-run-1', + 'task:task-1', + expect.objectContaining({ + interactingUserId: 'selected-user', + subject: previous.subject, + }), + ); + } else { + await expect(result).rejects.toThrow('Already handled'); + expect(integrationMessageContextService.setLatest).not.toHaveBeenCalled(); + } + expect(integrationMessageContextService.getLatest).not.toHaveBeenCalled(); + expect(runtimeCacheService.releaseRuntimeLease).toHaveBeenCalledWith(runtime.agent); + }, + ); + it('reconstructs a resumed runtime from the persisted sandbox scope', async () => { const { service, checkpointStorage, runtimeCacheService } = makeService(true); const runtime = makeRuntime(); diff --git a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts index df4764aa9e2..62a6bb4b9c8 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts @@ -24,6 +24,8 @@ import { AgentWorkflowExecutionService } from '../agent-workflow-execution.servi import type { Agent } from '../entities/agent.entity'; import type { NodeToolAiGatewayService } from '../json-config/node-tool-ai-gateway.service'; import type { AgentRepository } from '../repositories/agent.repository'; +import type { IntegrationMessageContextService } from '../integrations/integration-message-context.service'; +import type { IntegrationMessageContext } from '../integrations/integration-tool-types'; import type { ToolRegistry } from '../tool-registry'; import type { WorkflowAgentStreamObserver } from '../workflow-agent-stream'; @@ -123,6 +125,8 @@ function makeService() { const agentRunTracingService = mock(); const executionLevelTracer = mock(); const nodeToolAiGatewayService = mock(); + const integrationMessageContextService = mock(); + integrationMessageContextService.getLatest.mockResolvedValue(null); executionService.startExecutionRecording.mockResolvedValue('execution-1'); executionService.finalizeExecution.mockResolvedValue('execution-1'); @@ -144,6 +148,7 @@ function makeService() { executionLevelTracer, nodeToolAiGatewayService, aiConfigMock, + integrationMessageContextService, ); return { @@ -155,6 +160,7 @@ function makeService() { agentRunTracingService, executionLevelTracer, nodeToolAiGatewayService, + integrationMessageContextService, }; } @@ -171,7 +177,16 @@ describe('AgentWorkflowExecutionService', () => { executionService, telemetry, agentRunTracingService, + integrationMessageContextService, } = makeService(); + const messageContext: IntegrationMessageContext = { + integrationConnectionId: 'slack:credential-1', + platform: 'slack', + target: { type: 'thread', threadId: 'slack:channel-1:message-1' }, + interactingUserId: 'platform-user-1', + updatedAt: '2026-01-01T00:00:00.000Z', + }; + integrationMessageContextService.getLatest.mockResolvedValue(messageContext); const runtime = makeRuntime([ { type: 'tool-call', toolCallId: 'tc-1', toolName: 'lookup', input: { id: 1 } }, { type: 'tool-result', toolCallId: 'tc-1', toolName: 'lookup', output: { ok: true } }, @@ -198,9 +213,14 @@ describe('AgentWorkflowExecutionService', () => { // resourceId is the memory store's read scope: it must be stable // across executions (NOT the execution id) or a reused session id // would never see its prior messages. - persistence: { resourceId: 'thread-1', threadId: 'thread-1' }, + persistence: { + resourceId: 'thread-1', + threadId: 'thread-1', + hostMetadata: { n8nIntegrationMessageContext: messageContext }, + }, }), ); + expect(integrationMessageContextService.getLatest).toHaveBeenCalledExactlyOnceWith('thread-1'); expect(result).toEqual( expect.objectContaining({ response: '', @@ -283,21 +303,32 @@ describe('AgentWorkflowExecutionService', () => { 'hello', expect.objectContaining({ persistence: expect.objectContaining({ - hostMetadata: encodeAgentSandboxHostMetadata({ projectId, principalHash }), + hostMetadata: { + ...encodeAgentSandboxHostMetadata({ projectId, principalHash }), + n8nIntegrationMessageContext: null, + }, }), }), ); }); - it('records workflow stream setup failures', async () => { - const { service, agentRepository, reconstructionService, executionService } = makeService(); + it.each(['stream', 'context'] as const)('records workflow %s setup failures', async (step) => { + const { + service, + agentRepository, + reconstructionService, + executionService, + integrationMessageContextService, + } = makeService(); const runtime = makeRuntime(); const principalHash = hashAgentSandboxPrincipal({ type: 'workflow-execution', workflowId: 'workflow-1', executionId: 'execution-1', }); - runtime.agent.stream.mockRejectedValue(new Error('stream setup failed')); + const error = new Error(`${step} setup failed`); + if (step === 'stream') runtime.agent.stream.mockRejectedValue(error); + else integrationMessageContextService.getLatest.mockRejectedValue(error); agentRepository.findByIdAndProjectId.mockResolvedValue(makeAgent()); reconstructionService.reconstructFromAgentEntity.mockResolvedValue(runtime); executionService.startExecutionRecording.mockResolvedValue('fallback-execution-1'); @@ -315,12 +346,12 @@ describe('AgentWorkflowExecutionService', () => { undefined, { principalHash }, ), - ).rejects.toThrow('stream setup failed'); + ).rejects.toThrow(error.message); expect(executionService.finalizeExecution).toHaveBeenCalledWith( 'fallback-execution-1', expect.objectContaining({ - record: expect.objectContaining({ error: 'stream setup failed', finishReason: 'error' }), + record: expect.objectContaining({ error: error.message, finishReason: 'error' }), }), ); }); @@ -692,6 +723,7 @@ describe('AgentWorkflowExecutionService', () => { executionService, telemetry, agentRunTracingService, + integrationMessageContextService, } = makeService(); const runtime = makeRuntime([ { type: 'text-start', id: 'text-1' }, @@ -760,13 +792,18 @@ describe('AgentWorkflowExecutionService', () => { // Inline runs have no persisted session. expect(result.session).toBeNull(); expect(executionService.startExecutionRecording).not.toHaveBeenCalled(); + expect(integrationMessageContextService.getLatest).not.toHaveBeenCalled(); // Thread-scoped persistence: stable across executions, so a reused // session id continues the same conversation. expect(runtime.agent.stream).toHaveBeenCalledWith( 'hello', expect.objectContaining({ - persistence: { resourceId: 'thread-1', threadId: 'thread-1' }, + persistence: { + resourceId: 'thread-1', + threadId: 'thread-1', + hostMetadata: { n8nIntegrationMessageContext: null }, + }, }), ); diff --git a/packages/cli/src/modules/agents/__tests__/agent-workflow-tool-resume.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-workflow-tool-resume.service.test.ts index 95b48f3b35e..18b7ba80cfc 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-workflow-tool-resume.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-workflow-tool-resume.service.test.ts @@ -17,6 +17,7 @@ import type { AgentBackgroundJobService } from '../background/agent-background-j import type { AgentChatBridge } from '../integrations/agent-chat-bridge'; import type { ChatIntegrationService } from '../integrations/chat-integration.service'; import type { IntegrationMessageContextService } from '../integrations/integration-message-context.service'; +import { encodeIntegrationMessageContext } from '../integrations/integration-message-context'; import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage'; const agentRun: RelatedAgentRun = { @@ -50,7 +51,10 @@ function setup() { // Default to a genuinely parked run; tests that care override it. checkpointStorage.getStatus.mockResolvedValue({ status: 'active', - checkpoint: { status: 'suspended' }, + checkpoint: { + status: 'suspended', + persistence: { threadId: agentRun.threadId, resourceId: 'user-1' }, + }, } as never); const backgroundJobService = mock(); const service = new AgentWorkflowToolResumeService( @@ -110,6 +114,7 @@ describe('AgentWorkflowToolResumeService → lifecycle wiring', () => { 'run-1', 'call-1', { type: 'workflow_finished', value: 'success' }, + { messageContext: null, allowLegacyThreadId: true }, ); }); @@ -191,11 +196,81 @@ describe('AgentWorkflowToolResumeService → chat platforms', () => { expect.any(String), expect.any(String), { type: 'workflow_finished', value: 'error' }, + { messageContext: null, allowLegacyThreadId: true }, ); }); - // An agent with two connections on one platform must reply through the one the - // thread came in on, not whichever is found first. + // A successful action can move the turn to another platform. The continuation + // must use the platform and connection saved in the checkpoint. + it('uses the checkpoint platform, connection, and destination for a task continuation', async () => { + const { service, bridge, chatIntegrationService, messageContextService, checkpointStorage } = + setup(); + const messageContext = { + integrationConnectionId: 'discord:cred-2', + platform: 'discord', + target: { type: 'thread' as const, threadId: 'discord:outbound:1' }, + replyTarget: { type: 'thread' as const, threadId: 'discord:inbound:2' }, + updatedAt: '2026-09-18T10:00:00.000Z', + }; + checkpointStorage.getStatus.mockResolvedValue({ + status: 'active', + checkpoint: { + status: 'suspended', + persistence: { + threadId: 'task-run-1', + resourceId: 'task:task-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }, + } as never); + messageContextService.getLatest.mockResolvedValue({ + ...messageContext, + integrationConnectionId: 'slack:other', + }); + chatIntegrationService.getBridge.mockReturnValue(bridge); + + await service.resume({ ...agentRun, threadId: 'task-run-1' }, 'success'); + + expect(messageContextService.getLatest).not.toHaveBeenCalled(); + expect(chatIntegrationService.getBridge).toHaveBeenCalledWith('agent-1', 'discord', 'cred-2'); + expect(bridge.resumeInAgentThread).toHaveBeenCalledWith( + 'task-run-1', + 'run-1', + 'call-1', + { type: 'workflow_finished', value: 'success' }, + { messageContext, allowLegacyThreadId: false }, + ); + }); + + it.each([ + null, + { platform: 'slack' }, + { + integrationConnectionId: 'slack', + platform: 'slack', + target: { type: 'thread', threadId: 'slack:C1:1' }, + updatedAt: '2026-09-18T10:00:00.000Z', + }, + ])('keeps the run suspended when its saved reply context is incomplete: %j', async (snapshot) => { + const { service, chatIntegrationService, messageContextService, checkpointStorage } = setup(); + checkpointStorage.getStatus.mockResolvedValue({ + status: 'active', + checkpoint: { + status: 'suspended', + persistence: { + threadId: 'task-run-1', + resourceId: 'task:task-1', + hostMetadata: { n8nIntegrationMessageContext: snapshot }, + }, + }, + } as never); + + await service.resume({ ...agentRun, threadId: 'task-run-1' }, 'success'); + + expect(messageContextService.getLatest).not.toHaveBeenCalled(); + expect(chatIntegrationService.getBridge).not.toHaveBeenCalled(); + }); + it('resolves the bridge by the credential the thread came in on', async () => { const { service, bridge, chatIntegrationService, messageContextService } = setup(); messageContextService.getLatest.mockResolvedValue({ @@ -210,13 +285,18 @@ describe('AgentWorkflowToolResumeService → chat platforms', () => { }); it.each([ - ['there is no message context', null], + ['there is no message context', null, null], [ 'the context belongs to another platform', { integrationConnectionId: 'discord:c', platform: 'discord' }, + null, ], - ['no credential is bound yet', { integrationConnectionId: 'slack', platform: 'slack' }], - ])('falls back to any ingress bridge when %s', async (_label, context) => { + [ + 'no credential is bound yet', + { integrationConnectionId: 'slack', platform: 'slack' }, + { integrationConnectionId: 'slack', platform: 'slack' }, + ], + ])('falls back to any ingress bridge when %s', async (_label, context, expectedContext) => { const { service, bridge, chatIntegrationService, messageContextService } = setup(); messageContextService.getLatest.mockResolvedValue(context as never); chatIntegrationService.getBridge.mockReturnValue(bridge); @@ -224,6 +304,13 @@ describe('AgentWorkflowToolResumeService → chat platforms', () => { await service.resume(agentRun, 'success'); expect(chatIntegrationService.getBridge).toHaveBeenCalledWith('agent-1', 'slack', undefined); + expect(bridge.resumeInAgentThread).toHaveBeenCalledWith( + 'agent-1:slack:C123', + 'run-1', + 'call-1', + { type: 'workflow_finished', value: 'success' }, + { messageContext: expectedContext, allowLegacyThreadId: true }, + ); }); it('does nothing for a run with no chat surface at all', async () => { diff --git a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts index d45c278005d..ea17c3c8892 100644 --- a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts +++ b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts @@ -39,6 +39,14 @@ import { AgentSandboxRuntimeService } from './agent-sandbox-runtime.service'; import { buildAgentConfigurationTelemetry } from './agent-telemetry'; import { AgentTurnExecutionService } from './agent-turn-execution.service'; import type { AgentChatBridge } from './integrations/agent-chat-bridge'; +import { + encodeIntegrationMessageContext, + inheritIntegrationMessageContext, +} from './integrations/integration-message-context'; +import type { + IntegrationMessageContext, + SessionBinding, +} from './integrations/integration-tool-types'; import { IntegrationMessageContextService } from './integrations/integration-message-context.service'; import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage'; import { modelStreamStallOptions } from './model-stream-stall-options'; @@ -84,6 +92,9 @@ export interface ExecuteForChatConfig { } export interface ExecuteForChatPublishedConfig { + messageContext?: IntegrationMessageContext | null; + /** Platform conversation metadata scope. Execution memory can belong to a task. */ + contextConversation?: SessionBinding; agentId: string; projectId: string; /** What the user wrote; recorded in the execution transcript. */ @@ -106,6 +117,9 @@ export interface ExecuteForChatPublishedConfig { } export interface ResumeForChatConfig { + messageContext?: IntegrationMessageContext | null; + /** Platform conversation metadata scope. Execution memory can belong to a task. */ + contextConversation?: SessionBinding; agentId: string; projectId: string; runId: string; @@ -188,6 +202,7 @@ export interface ExecuteForWakeConfig { } export interface StreamChatResponseConfig { + messageContext?: IntegrationMessageContext | null; agentInstance: RuntimeAgent; toolRegistry: ToolRegistry; /** See `AgentRuntime.mcpServerAttributions`. */ @@ -451,6 +466,18 @@ export class AgentExecutionOrchestratorService { source: executionSource ?? 'unknown', modelId: modelIdFromSnapshot(runtime.agent.snapshot.model), }); + let messageContext = config.messageContext; + if (messageContext === undefined) { + messageContext = + await this.integrationMessageContextService.getForResume(memoryScope); + } else if (messageContext && config.contextConversation) { + messageContext = inheritIntegrationMessageContext( + messageContext, + await this.integrationMessageContextService.getForResume(memoryScope), + ); + } + const selectedContext = messageContext; + const conversation = config.contextConversation; return { type: 'resume', @@ -458,6 +485,17 @@ export class AgentExecutionOrchestratorService { options: { runId, toolCallId, + hostMetadata: encodeIntegrationMessageContext(selectedContext), + ...(selectedContext && conversation + ? { + onResumeClaimed: async () => + await this.integrationMessageContextService.installIncoming( + selectedContext, + memoryScope, + conversation, + ), + } + : {}), executionCounter: createAgentExecutionCounter(this.telemetry, { agentId, userId: user?.id, @@ -520,15 +558,21 @@ export class AgentExecutionOrchestratorService { previewChat, }), async (runtime) => { - await this.integrationMessageContextService.setLatest(memory.threadId, memory.resourceId, { + const messageContext: IntegrationMessageContext = { integrationConnectionId: N8N_CHAT_INTEGRATION_TYPE, platform: N8N_CHAT_INTEGRATION_TYPE, target: { type: 'dm', userId: user.id, threadId: memory.threadId }, interactingUserId: user.id, updatedAt: new Date().toISOString(), - }); + }; + await this.integrationMessageContextService.setLatest( + memory.threadId, + memory.resourceId, + messageContext, + ); return this.streamChatResponse({ + messageContext, agentInstance: runtime.agent, toolRegistry: runtime.toolRegistry, mcpServerAttributions: runtime.mcpServerAttributions, @@ -593,8 +637,21 @@ export class AgentExecutionOrchestratorService { source: integrationType, }, ), - (runtime) => - this.streamChatResponse({ + async (runtime) => { + let messageContext = config.messageContext; + if (messageContext && config.contextConversation) { + messageContext = inheritIntegrationMessageContext( + messageContext, + await this.integrationMessageContextService.getLatestForIncoming(memory.threadId), + ); + await this.integrationMessageContextService.installIncoming( + messageContext, + memory, + config.contextConversation, + ); + } + return this.streamChatResponse({ + messageContext, agentInstance: runtime.agent, toolRegistry: runtime.toolRegistry, mcpServerAttributions: runtime.mcpServerAttributions, @@ -611,7 +668,8 @@ export class AgentExecutionOrchestratorService { configuration: runtime.telemetryConfiguration, }, sandboxPrincipalHash, - }), + }); + }, ); } @@ -719,9 +777,10 @@ export class AgentExecutionOrchestratorService { if (!isDraft) await this.externalHooks.run('agent.preExecute', [agentId]); const integrationType = isDraft ? N8N_CHAT_INTEGRATION_TYPE : identity.integrationType; + const messageContext = await this.integrationMessageContextService.getLatest(memory.threadId); const delivery = isDraft ? undefined - : await this.getWakeDelivery(agentId, integrationType, memory.threadId); + : await this.getWakeDelivery(agentId, integrationType, messageContext); const stream = this.withRuntimeLease( async () => await this.runtimeCacheService.getRuntime({ @@ -735,6 +794,7 @@ export class AgentExecutionOrchestratorService { (runtime) => this.streamWakeResponse( this.streamChatResponse({ + messageContext, agentInstance: runtime.agent, toolRegistry: runtime.toolRegistry, mcpServerAttributions: runtime.mcpServerAttributions, @@ -785,8 +845,11 @@ export class AgentExecutionOrchestratorService { if (delivery) await delivery.bridge.deliverWakeResponse(delivery.threadId, chunks); } - private async getWakeDelivery(agentId: string, integrationType: string, threadId: string) { - const context = await this.integrationMessageContextService.getLatest(threadId); + private async getWakeDelivery( + agentId: string, + integrationType: string, + context: IntegrationMessageContext | null, + ) { const target = context?.replyTarget ?? context?.target; const [platform, credentialId] = context?.integrationConnectionId.split(':') ?? []; if ( @@ -864,10 +927,17 @@ export class AgentExecutionOrchestratorService { const input = attachments?.length ? buildInboundUserMessage(modelMessage, attachments) : modelMessage; - const hostMetadata = encodeAgentSandboxHostMetadata({ - projectId, - principalHash: sandboxPrincipalHash, - }); + const messageContext = + config.messageContext === undefined + ? await this.integrationMessageContextService.getLatest(threadId) + : config.messageContext; + const hostMetadata = { + ...encodeAgentSandboxHostMetadata({ + projectId, + principalHash: sandboxPrincipalHash, + }), + ...encodeIntegrationMessageContext(messageContext), + }; return { type: 'start', diff --git a/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts b/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts index 3e189bf6ab7..e2a757c6900 100644 --- a/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts +++ b/packages/cli/src/modules/agents/agent-runtime-reconstruction.service.ts @@ -965,9 +965,7 @@ export class AgentRuntimeReconstructionService { } for (const descriptor of descriptors) { - agent.tool( - createIntegrationContextTool({ descriptor, messageContextStore, queryExecutor }), - ); + agent.tool(createIntegrationContextTool({ descriptor, queryExecutor })); agent.tool( createIntegrationActionTool({ descriptor, messageContextStore, actionExecutor }), ); diff --git a/packages/cli/src/modules/agents/agent-workflow-execution.service.ts b/packages/cli/src/modules/agents/agent-workflow-execution.service.ts index e2219f1e042..7a4d44b3cf2 100644 --- a/packages/cli/src/modules/agents/agent-workflow-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-workflow-execution.service.ts @@ -39,6 +39,8 @@ import { import { AgentTurnExecutionService } from './agent-turn-execution.service'; import type { Agent } from './entities/agent.entity'; import type { ExecutionRecorder, MessageRecord } from './execution-recorder'; +import { encodeIntegrationMessageContext } from './integrations/integration-message-context'; +import { IntegrationMessageContextService } from './integrations/integration-message-context.service'; import { NodeToolAiGatewayService } from './json-config/node-tool-ai-gateway.service'; import { modelStreamStallOptions } from './model-stream-stall-options'; import { AgentRepository } from './repositories/agent.repository'; @@ -102,6 +104,7 @@ export class AgentWorkflowExecutionService { private readonly executionLevelTracer: ExecutionLevelTracer, private readonly nodeToolAiGatewayService: NodeToolAiGatewayService, private readonly aiConfig: AiConfig, + private readonly integrationMessageContextService: IntegrationMessageContextService, ) {} private normalizeWorkflowStreamError(error: unknown, outputSchema?: JSONSchema7): Error { @@ -331,6 +334,10 @@ export class AgentWorkflowExecutionService { hasParentContext: parentCtx !== undefined, }); + // Only stored agents have integration tools. + const messageContext = recordingParams + ? await this.integrationMessageContextService.getLatest(threadId) + : null; const resultStream = await agentInstance.stream(message, { // The memory store scopes message reads by `resourceId` (the // "per-user scope"; chat integrations pass the chat user id there). @@ -342,7 +349,10 @@ export class AgentWorkflowExecutionService { persistence: { resourceId: threadId, threadId, - ...(sandboxScope ? { hostMetadata: encodeAgentSandboxHostMetadata(sandboxScope) } : {}), + hostMetadata: { + ...(sandboxScope ? encodeAgentSandboxHostMetadata(sandboxScope) : {}), + ...encodeIntegrationMessageContext(messageContext), + }, }, executionCounter: createAgentExecutionCounter(this.telemetry, { agentId: telemetryAgentId, diff --git a/packages/cli/src/modules/agents/agent-workflow-tool-resume.service.ts b/packages/cli/src/modules/agents/agent-workflow-tool-resume.service.ts index 836e488ae78..a1bd3571d93 100644 --- a/packages/cli/src/modules/agents/agent-workflow-tool-resume.service.ts +++ b/packages/cli/src/modules/agents/agent-workflow-tool-resume.service.ts @@ -18,6 +18,7 @@ import { settlementStatusForExecution, } from './background/agent-background-job.service'; import { ChatIntegrationService } from './integrations/chat-integration.service'; +import { readIntegrationMessageContext } from './integrations/integration-message-context'; import { IntegrationMessageContextService } from './integrations/integration-message-context.service'; import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage'; @@ -152,13 +153,46 @@ export class AgentWorkflowToolResumeService { return; } - // Reply through the connection the thread actually came in on. An agent with - // two connections on one platform would otherwise post into whichever - // workspace happened to be found first. - const credentialId = await this.originatingCredentialId(agentRun); + const checkpoint = await this.checkpointStorage.getStatus(agentRun.runId, agentRun.agentId); + if (checkpoint.status !== 'active' || checkpoint.checkpoint.status !== 'suspended') return; + const persistence = checkpoint.checkpoint.persistence; + if (!persistence) return; + let messageContext = readIntegrationMessageContext(persistence); + const allowLegacyThreadId = messageContext === undefined; + if (messageContext === undefined) { + try { + messageContext = await this.messageContextService.getLatest(persistence.threadId); + } catch (error) { + this.logger.warn('Could not read the thread message context for an agent resume', { + runId: agentRun.runId, + error: error instanceof Error ? error.message : String(error), + }); + messageContext = null; + } + } + const [platform, storedCredentialId] = messageContext?.integrationConnectionId.split(':') ?? []; + let integrationType = agentRun.integrationType; + let credentialId: string | undefined; + if (allowLegacyThreadId) { + if (messageContext?.platform === integrationType && platform === integrationType) { + credentialId = storedCredentialId; + } else { + messageContext = null; + } + } else { + if (!messageContext || messageContext.platform !== platform || !storedCredentialId) { + this.logger.warn('Agent resume has no integration reply context', { + agentId: agentRun.agentId, + runId: agentRun.runId, + }); + return; + } + integrationType = messageContext.platform; + credentialId = storedCredentialId; + } const bridge = this.chatIntegrationService.getBridge( agentRun.agentId, - agentRun.integrationType, + integrationType, credentialId, ); if (!bridge) { @@ -166,7 +200,7 @@ export class AgentWorkflowToolResumeService { // so the run is recoverable once the integration reconnects. this.logger.warn('No live chat bridge to resume the agent run into', { agentId: agentRun.agentId, - integrationType: agentRun.integrationType, + integrationType, credentialId, runId: agentRun.runId, }); @@ -178,31 +212,10 @@ export class AgentWorkflowToolResumeService { agentRun.runId, agentRun.toolCallId, resumeData, + { messageContext, allowLegacyThreadId }, ); } - /** - * The credential of the connection this thread last exchanged a message on, as - * recorded in the thread's message context. Undefined when unknown, which - * leaves the lookup to fall back to any ingress bridge for the platform. - */ - private async originatingCredentialId(agentRun: RelatedAgentRun): Promise { - try { - const context = await this.messageContextService.getLatest(agentRun.threadId); - if (!context || context.platform !== agentRun.integrationType) return undefined; - // `integrationConnectionId` is `type` alone for a single-connection agent, - // or `type:credentialId` once a credential is bound. - const [, credentialId] = context.integrationConnectionId.split(':'); - return credentialId; - } catch (error) { - this.logger.warn('Could not read the thread message context for an agent resume', { - runId: agentRun.runId, - error: error instanceof Error ? error.message : String(error), - }); - return undefined; - } - } - /** * The preview's SSE stream closed when the run suspended, so there is nothing to * stream into: draining headlessly is what records the turn, and the push then diff --git a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-bridge.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-bridge.test.ts index bcb80d7d1ce..653cf95afe5 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-bridge.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-bridge.test.ts @@ -1522,7 +1522,7 @@ describe('AgentChatBridge — consumeStream', () => { 'project-1', integration, ); - return { bot, bridge }; + return { bot, bridge, agentExecutor }; } it('strips a rotation suffix before reconstructing the platform thread id', async () => { @@ -1538,6 +1538,31 @@ describe('AgentChatBridge — consumeStream', () => { expect(bot.thread).toHaveBeenCalledWith('1001'); }); + it('routes a task continuation to its saved reply destination', async () => { + const { bot, bridge, agentExecutor } = makeResumeBridge(streamingIntegration); + const messageContext = { + integrationConnectionId: 'test-streaming:cred-1', + platform: 'test-streaming', + target: { type: 'thread' as const, threadId: 'outbound-thread' }, + replyTarget: { type: 'thread' as const, threadId: 'reply-thread' }, + updatedAt: '2026-09-18T10:00:00.000Z', + }; + await bridge.resumeInAgentThread( + 'task-run-1', + 'run-1', + 'tool-1', + {}, + { + messageContext, + allowLegacyThreadId: false, + }, + ); + expect(bot.thread).toHaveBeenCalledWith('reply-thread'); + expect(agentExecutor.resumeForChat).toHaveBeenCalledWith( + expect.objectContaining({ messageContext }), + ); + }); + it('reconstructs an unrotated thread id unchanged', async () => { const { bot, bridge } = makeResumeBridge({ type: 'test-formatted-buffered', @@ -2480,12 +2505,11 @@ describe('AgentChatBridge — consumeStream', () => { }); describe('message context', () => { - it('strips the Slack bot mention before executing and stores the Slack bot user ID', async () => { + it('strips the Slack bot mention before executing and captures the Slack bot user ID', async () => { const { bot, handlers } = makeBot(); bot.getAdapter.mockReturnValue({ botUserId: 'U_BOT' }); const thread = makeThread(); const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue(null); const agentExecutor = makeAgentExecutor([{ type: 'finish', finishReason: 'stop' }]); new AgentChatBridge( @@ -2515,23 +2539,24 @@ describe('AgentChatBridge — consumeStream', () => { author: { id: 'u1', name: 'user1' }, }), ); - expect(messageContextStore.setLatest).toHaveBeenCalledWith( - 'agent-1:thread-1', - 'u1', + expect(agentExecutor.executeForChatPublished).toHaveBeenCalledWith( expect.objectContaining({ - agentUserId: 'U_BOT', - interactingUserId: 'u1', - messageId: 'message-1', + contextConversation: { threadId: 'agent-1:thread-1', resourceId: 'u1' }, + messageContext: expect.objectContaining({ + agentUserId: 'U_BOT', + interactingUserId: 'u1', + messageId: 'message-1', + }), }), ); + expect(messageContextStore.setLatest).not.toHaveBeenCalled(); }); - it('stores the platform reply expectation for subscribed Slack channel messages', async () => { + it('captures the platform reply expectation for subscribed Slack channel messages', async () => { const { bot, handlers } = makeBot(); bot.getAdapter.mockReturnValue({ botUserId: 'U_BOT' }); const thread = makeThread(); const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue(null); const agentExecutor = makeAgentExecutor([{ type: 'finish', finishReason: 'stop' }]); new AgentChatBridge( @@ -2555,26 +2580,27 @@ describe('AgentChatBridge — consumeStream', () => { raw: { channel: 'C123', channel_type: 'channel' }, }); - expect(messageContextStore.setLatest).toHaveBeenCalledWith( - 'agent-1:thread-1', - 'u1', + expect(agentExecutor.executeForChatPublished).toHaveBeenCalledWith( expect.objectContaining({ - replyExpectation: 'optional', - replyTarget: { - type: 'thread', - threadId: 'thread-1', - channelId: 'channel-1', - }, - replyMessageId: 'message-2', + contextConversation: { threadId: 'agent-1:thread-1', resourceId: 'u1' }, + messageContext: expect.objectContaining({ + replyExpectation: 'optional', + replyTarget: { + type: 'thread', + threadId: 'thread-1', + channelId: 'channel-1', + }, + replyMessageId: 'message-2', + }), }), ); + expect(messageContextStore.setLatest).not.toHaveBeenCalled(); }); it('defaults the reply expectation to required for platforms without a reply policy', async () => { const { bot, handlers } = makeBot(); const thread = makeThread(); const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue(null); const agentExecutor = makeAgentExecutor([{ type: 'finish', finishReason: 'stop' }]); new AgentChatBridge( @@ -2594,18 +2620,19 @@ describe('AgentChatBridge — consumeStream', () => { author: { userId: 'u1', userName: 'user1' }, }); - expect(messageContextStore.setLatest).toHaveBeenCalledWith( - 'agent-1:thread-1', - 'u1', - expect.objectContaining({ replyExpectation: 'required' }), + expect(agentExecutor.executeForChatPublished).toHaveBeenCalledWith( + expect.objectContaining({ + contextConversation: { threadId: 'agent-1:thread-1', resourceId: 'u1' }, + messageContext: expect.objectContaining({ replyExpectation: 'required' }), + }), ); + expect(messageContextStore.setLatest).not.toHaveBeenCalled(); }); - it('stores a sanitized message subject from the inbound message', async () => { + it('captures a sanitized message subject from the inbound message', async () => { const { bot, handlers } = makeBot(); const thread = makeThread(); const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue(null); const agentExecutor = makeAgentExecutor([{ type: 'finish', finishReason: 'stop' }]); new AgentChatBridge( @@ -2639,48 +2666,36 @@ describe('AgentChatBridge — consumeStream', () => { }, }); - expect(messageContextStore.setLatest).toHaveBeenCalledWith( - 'agent-1:thread-1', - 'u1', + expect(agentExecutor.executeForChatPublished).toHaveBeenCalledWith( expect.objectContaining({ - integrationConnectionId: 'test-streaming:cred-1', - platform: 'test-streaming', - target: { type: 'thread', threadId: 'thread-1', channelId: 'channel-1' }, - messageId: 'message-1', - interactingUserId: 'u1', - subject: { - type: 'issue', - id: 'ENG-123', - title: 'Fix signup', - description: 'Signup fails for invited users', - status: 'In Progress', - url: 'https://linear.app/n8n/issue/ENG-123/fix-signup', - labels: ['Bug'], - assignee: { id: 'user-2', name: 'Michael Drury' }, - author: { id: 'user-3', name: 'Ada Lovelace' }, - }, + contextConversation: { threadId: 'agent-1:thread-1', resourceId: 'u1' }, + messageContext: expect.objectContaining({ + integrationConnectionId: 'test-streaming:cred-1', + platform: 'test-streaming', + target: { type: 'thread', threadId: 'thread-1', channelId: 'channel-1' }, + messageId: 'message-1', + interactingUserId: 'u1', + subject: { + type: 'issue', + id: 'ENG-123', + title: 'Fix signup', + description: 'Signup fails for invited users', + status: 'In Progress', + url: 'https://linear.app/n8n/issue/ENG-123/fix-signup', + labels: ['Bug'], + assignee: { id: 'user-2', name: 'Michael Drury' }, + author: { id: 'user-3', name: 'Ada Lovelace' }, + }, + }), }), ); + expect(messageContextStore.setLatest).not.toHaveBeenCalled(); }); - it('keeps the previous subject when an action click updates the latest context', async () => { + it('captures the selected interaction without reading or writing thread context', async () => { const { bot, handlers } = makeBot(); const thread = makeThread(); const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue({ - integrationConnectionId: 'test-streaming:cred-1', - platform: 'test-streaming', - target: { type: 'thread', threadId: 'thread-1', channelId: 'channel-1' }, - messageId: 'message-1', - interactingUserId: 'u1', - agentUserId: 'U_BOT', - subject: { - type: 'issue', - id: 'ENG-123', - title: 'Fix signup', - }, - updatedAt: '2026-05-18T10:00:00.000Z', - }); const agentExecutor = makeAgentExecutor([{ type: 'finish', finishReason: 'stop' }]); new AgentChatBridge( @@ -2704,21 +2719,17 @@ describe('AgentChatBridge — consumeStream', () => { adapter: { deleteMessage: vi.fn().mockResolvedValue(undefined) }, }); - expect(messageContextStore.setLatest).toHaveBeenCalledWith( - 'agent-1:thread-1', - 'u2', + expect(agentExecutor.resumeForChat).toHaveBeenCalledWith( expect.objectContaining({ - messageId: 'card-message-1', - interactingUserId: 'u2', - agentUserId: 'U_BOT', - subject: { - type: 'issue', - id: 'ENG-123', - title: 'Fix signup', - }, - replyExpectation: 'required', + contextConversation: { threadId: 'agent-1:thread-1', resourceId: 'u2' }, + messageContext: expect.objectContaining({ + messageId: 'card-message-1', + interactingUserId: 'u2', + replyExpectation: 'required', + }), }), ); + expect(messageContextStore.setLatest).not.toHaveBeenCalled(); }); }); @@ -2763,20 +2774,17 @@ describe('AgentChatBridge — consumeStream', () => { }), }), ); - expect(messageContextStore.setLatest).toHaveBeenCalledWith( - 'agent-1:slack:D123:1001', - 'u1', - expect.objectContaining({ messageId: '1002', interactingUserId: 'u1' }), - ); - expect(messageContextStore.setLatest).toHaveBeenCalledWith( - 'task-1-uuid', - 'task:task-1', + expect(agentExecutor.executeForChatPublished).toHaveBeenCalledWith( expect.objectContaining({ - messageId: '1002', - interactingUserId: 'u1', - target: expect.objectContaining({ threadId: 'slack:D123:1001' }), + contextConversation: { threadId: 'agent-1:slack:D123:1001', resourceId: 'u1' }, + messageContext: expect.objectContaining({ + messageId: '1002', + interactingUserId: 'u1', + target: expect.objectContaining({ threadId: 'slack:D123:1001' }), + }), }), ); + expect(messageContextStore.setLatest).not.toHaveBeenCalled(); }); it('stores inbound attachments on the bound task thread', async () => { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts index 98ef3493239..6b68f1841ce 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-hitl-resume-handler.test.ts @@ -42,7 +42,7 @@ it.each([ 'discord:800000000000000001:700000000000000001:600000000000000001', toAgentThreadId: () => ({ id: 'agent-thread-1' }) as never, getPlatformAgentContext: () => ({}), - messageContextBridge: { updateLatest: vi.fn().mockResolvedValue(undefined) } as never, + messageContextBridge: { capture: vi.fn().mockReturnValue(undefined) } as never, streamConsumer: { consume: vi.fn().mockResolvedValue(undefined) } as never, createResumeExecutionContext: async () => ({}), }); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/channel-integration-contract.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/channel-integration-contract.ts index c8037a70382..644585cd42e 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/channel-integration-contract.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/channel-integration-contract.ts @@ -2,6 +2,7 @@ import type { AgentMessageAuthor } from '@n8n/api-types'; import type { Mock } from 'vitest'; import type { ChatIntegrationActionExecutor } from '../../integration-action-executor'; +import { encodeIntegrationMessageContext } from '../../integration-message-context'; import { createIntegrationContextTool } from '../../integration-tools'; import type { IntegrationMessageContext, @@ -77,7 +78,7 @@ export function runSharedChannelIntegrationContract(scenario: ChannelIntegration ); }); - it('persists current message context for the integration context tool', async () => { + it('captures current message context for the integration context tool', async () => { ctx = await scenario.createContext(); await ctx.sendWebhook(scenario.fixtures.mention); @@ -91,12 +92,17 @@ export function runSharedChannelIntegrationContract(scenario: ChannelIntegration queryExecutor: { execute: vi.fn(), }, - messageContextStore: ctx.messageContextStore, }).build(); const result = await contextTool.handler!( { query: 'get_current_message_context', input: {} }, - { persistence: { threadId, resourceId: scenario.expected.resourceId } }, + { + persistence: { + threadId, + resourceId: scenario.expected.resourceId, + hostMetadata: encodeIntegrationMessageContext(context ?? null), + }, + }, ); expect(result).toEqual({ ok: true, context }); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts index 4ffb7bbfccb..ad6bf7d9114 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/linear/replay-test-context.ts @@ -300,8 +300,8 @@ export async function createLinearReplayContext( chat: chat as unknown as ChatInstance, apiCalls: stub.apiCalls, sendWebhook, - latestContext: () => setup.messageContextStore.latest(), - latestThreadId: () => setup.messageContextStore.latestThreadId(), + latestContext: setup.latestContext, + latestThreadId: setup.latestThreadId, lastPost: () => stub.apiCalls.filter((call) => POST_METHODS.has(call.method)).at(-1), shutdown: async () => { try { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts index 76ecc6169e6..b2d9aca714e 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts @@ -20,6 +20,8 @@ import { getIntegrationToolConnectionDescriptors } from '../../integration-tools export type ReplayWebhookOptions = { waitUntil?: (task: Promise) => void }; +type AgentExecutor = ConstructorParameters[2]; + export type ReplayWebhookHandler = ( request: Request, options?: ReplayWebhookOptions, @@ -74,14 +76,6 @@ export class MemoryMessageContextStore implements IntegrationMessageContextStore // that need unbind semantics call unbindSession directly. await Promise.resolve(); } - - latest(): IntegrationMessageContext | undefined { - return [...this.contexts.values()].at(-1); - } - - latestThreadId(): string | undefined { - return [...this.contexts.keys()].at(-1); - } } export function toStream(chunks: StreamChunk[]): AsyncGenerator { @@ -196,6 +190,8 @@ export interface ReplayContextSetup { descriptor: ReturnType[number]; integration: AgentIntegrationConfig; messageContextStore: MemoryMessageContextStore; + latestContext: () => IntegrationMessageContext | undefined; + latestThreadId: () => string | undefined; nextStream: (chunks: StreamChunk[]) => void; shutdown: () => Promise; } @@ -215,9 +211,18 @@ export function createReplayContextSetup(params: { { type: 'text-delta', id: 'text-1', delta: 'Got it' }, { type: 'finish', finishReason: 'stop' }, ]; + let selectedContext: IntegrationMessageContext | undefined; + let selectedThreadId: string | undefined; const agentExecutor = { - executeForChatPublished: vi.fn(() => toStream(stream)), - resumeForChat: vi.fn(() => toStream(stream)), + executeForChatPublished: vi.fn((config) => { + selectedContext = config.messageContext ?? undefined; + selectedThreadId = config.memory.threadId.id; + return toStream(stream); + }), + resumeForChat: vi.fn((config) => { + selectedContext = config.messageContext ?? undefined; + return toStream(stream); + }), }; const messageContextStore = new MemoryMessageContextStore(); @@ -248,6 +253,8 @@ export function createReplayContextSetup(params: { descriptor, integration: params.integration, messageContextStore, + latestContext: () => selectedContext, + latestThreadId: () => selectedThreadId, nextStream: (chunks: StreamChunk[]) => { stream = chunks; }, diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/slack/replay-test-context.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/slack/replay-test-context.ts index 6d9e95cc13c..17ff8eaed5b 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/slack/replay-test-context.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/slack/replay-test-context.ts @@ -239,8 +239,8 @@ export async function createSlackReplayContext( chat: chat as unknown as ChatInstance, apiCalls: stub.apiCalls, sendWebhook, - latestContext: () => setup.messageContextStore.latest(), - latestThreadId: () => setup.messageContextStore.latestThreadId(), + latestContext: setup.latestContext, + latestThreadId: setup.latestThreadId, lastPost: () => stub.apiCalls.filter((call) => call.method === 'chat.postMessage').at(-1), shutdown: async () => { try { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts index b63e3669f6e..52427372e94 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/telegram/replay-test-context.ts @@ -258,8 +258,8 @@ export async function createTelegramReplayContext( apiCalls: stub.apiCalls, sendTelegramWebhook, sendWebhook: sendTelegramWebhook, - latestContext: () => setup.messageContextStore.latest(), - latestThreadId: () => setup.messageContextStore.latestThreadId(), + latestContext: setup.latestContext, + latestThreadId: setup.latestThreadId, lastApiCall: (method: string) => stub.apiCalls.filter((call) => call.method === method).at(-1), lastPost: () => stub.apiCalls diff --git a/packages/cli/src/modules/agents/integrations/__tests__/integration-message-context.service.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/integration-message-context.service.test.ts index 049705569e0..dfeb330d58c 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/integration-message-context.service.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/integration-message-context.service.test.ts @@ -1,7 +1,10 @@ +import { mockLogger } from '@n8n/backend-test-utils'; +import { jsonParse } from 'n8n-workflow'; import { mock } from 'vitest-mock-extended'; import type { AgentResourceRepository } from '../../repositories/agent-resource.repository'; import type { AgentThreadRepository } from '../../repositories/agent-thread.repository'; +import { encodeIntegrationMessageContext } from '../integration-message-context'; import { IntegrationMessageContextService } from '../integration-message-context.service'; import type { IntegrationMessageContext } from '../integration-tools'; @@ -34,10 +37,61 @@ describe('IntegrationMessageContextService — session binding', () => { return undefined as never; }, ); - const service = new IntegrationMessageContextService(threadRepository, resourceRepository); + const service = new IntegrationMessageContextService( + threadRepository, + resourceRepository, + mockLogger(), + ); return { service, threadRepository, threads }; } + it.each([ + ['older checkpoint', {}, true], + ['empty snapshot', { n8nIntegrationMessageContext: null }, false], + ['malformed snapshot', { n8nIntegrationMessageContext: { platform: 'slack' } }, false], + ] as const)( + 'restores context for an %s without replacing explicit snapshots', + async (_name, hostMetadata, legacy) => { + const { service } = setup({ thread: { currentMessageContext: context } }); + const snapshot = await service.getForResume({ + threadId: 'thread', + resourceId: 'user', + hostMetadata, + }); + expect(snapshot).toEqual(legacy ? context : null); + }, + ); + + it('restores the serialized turn snapshot when the thread has a later context', async () => { + const persistence = { + threadId: 'thread', + resourceId: 'user', + hostMetadata: encodeIntegrationMessageContext(context), + }; + const serialized = JSON.stringify(persistence); + const { service } = setup({ + thread: { currentMessageContext: { ...context, messageId: 'later' } }, + }); + expect(await service.getForResume(jsonParse(serialized))).toEqual(context); + }); + + it('updates execution metadata when the platform metadata write fails', async () => { + const { service, threadRepository } = setup({ + task: { continueAs: { threadId: 'origin', resourceId: 'owner' } }, + }); + threadRepository.save.mockRejectedValueOnce(new Error('database unavailable')); + await service.installIncoming( + context, + { threadId: 'task', resourceId: 'task:task-1' }, + { threadId: 'platform', resourceId: 'actor' }, + ); + expect(await service.getLatest('task')).toEqual(context); + expect(await service.resolveSession('task')).toEqual({ + threadId: 'origin', + resourceId: 'owner', + }); + }); + it('binds a derived thread to an origin and resolves it back', async () => { const { service } = setup({ task: { boundThreads: [] }, diff --git a/packages/cli/src/modules/agents/integrations/__tests__/integration-tools.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/integration-tools.test.ts index 10e5dfb32e9..038d31bdb1a 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/integration-tools.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/integration-tools.test.ts @@ -2,6 +2,10 @@ import type { InterruptibleToolContext } from '@n8n/agents'; import { zodToJsonSchema } from '@n8n/ai-utilities/json-schema'; import type { AgentIntegrationConfig } from '@n8n/api-types'; import { mock } from 'vitest-mock-extended'; +import { + encodeIntegrationMessageContext, + readIntegrationMessageContext, +} from '../integration-message-context'; import type { z } from 'zod'; import { @@ -11,6 +15,7 @@ import { type IntegrationActionExecutor, type IntegrationContextQueryExecutor, type IntegrationMessageContextStore, + type IntegrationMessageContext, } from '../integration-tools'; const slackA: AgentIntegrationConfig = { @@ -73,44 +78,77 @@ describe('integration tools', () => { ); }); - it('context tool returns the latest message context for its integration connection', async () => { - const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue({ + it('keeps selected turn contexts separate when the thread context changes', async () => { + const first = { integrationConnectionId: 'slack:cred-a', platform: 'slack', - target: { type: 'thread', threadId: 'slack:C123:123.456' }, + target: { type: 'thread' as const, threadId: 'slack:C123:123.456' }, messageId: '123.456', + interactingUserId: 'user-1', + subject: { type: 'issue', id: 'issue-1' }, + replyTarget: { type: 'thread' as const, threadId: 'slack:C123:123.456' }, updatedAt: '2026-05-18T10:00:00.000Z', - }); + }; + const second = { + ...first, + messageId: '123.789', + interactingUserId: 'user-2', + subject: { type: 'issue', id: 'issue-2' }, + replyTarget: { type: 'thread' as const, threadId: 'slack:C456:123.789' }, + }; + const turns = [first, second].map((context, index) => + makeInterruptibleCtx({ + runId: `run-${index}`, + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(context), + }, + }), + ); const queryExecutor = mock(); + const descriptor = getIntegrationToolConnectionDescriptors([slackA])[0]; const tool = createIntegrationContextTool({ - descriptor: getIntegrationToolConnectionDescriptors([slackA])[0], - messageContextStore, + descriptor, queryExecutor, }).build(); - const result = await tool.handler!( - { query: 'get_current_message_context', input: {} }, - { persistence: { threadId: 'thread-1', resourceId: 'resource-1' } }, + for (const index of [0, 1, 0]) { + const result = await tool.handler!( + { query: 'get_current_message_context', input: {} }, + turns[index], + ); + expect(result).toEqual({ ok: true, context: [first, second][index] }); + } + const sentContext = { ...second, messageId: 'sent-message' }; + const actionExecutor = mock(); + actionExecutor.execute.mockResolvedValue({ ok: true, messageContext: sentContext }); + const messageContextStore = mock(); + const actionTool = createIntegrationActionTool({ + descriptor, + messageContextStore, + actionExecutor, + }).build(); + await actionTool.handler!( + { action: 'respond', input: { message: { text: 'Done' } } }, + turns[1], ); - - expect(result).toEqual({ - ok: true, - context: { - integrationConnectionId: 'slack:cred-a', - platform: 'slack', - target: { type: 'thread', threadId: 'slack:C123:123.456' }, - messageId: '123.456', - updatedAt: '2026-05-18T10:00:00.000Z', - }, - }); + expect(messageContextStore.setLatest).toHaveBeenCalledWith( + 'thread-1', + 'resource-1', + sentContext, + ); + for (const index of [1, 0]) { + expect( + await tool.handler!({ query: 'get_current_message_context', input: {} }, turns[index]), + ).toEqual({ ok: true, context: [first, sentContext][index] }); + } expect(queryExecutor.execute).not.toHaveBeenCalled(); }); it('context tool returns the current message subject when available', async () => { - const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue({ + const messageContext: IntegrationMessageContext | null = { integrationConnectionId: 'linear:cred-c', platform: 'linear', target: { type: 'thread', threadId: 'linear:issue-comment-1' }, @@ -127,12 +165,11 @@ describe('integration tools', () => { author: { id: 'user-2', name: 'Ada Lovelace' }, }, updatedAt: '2026-05-18T10:00:00.000Z', - }); + }; const queryExecutor = mock(); const tool = createIntegrationContextTool({ descriptor: getIntegrationToolConnectionDescriptors([linear])[0], - messageContextStore, queryExecutor, }).build(); const schema = tool.inputSchema as z.ZodType; @@ -142,7 +179,13 @@ describe('integration tools', () => { const result = await tool.handler!( { query: 'get_current_subject', input: {} }, - { persistence: { threadId: 'thread-1', resourceId: 'resource-1' } }, + { + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }, ); expect(result).toEqual({ @@ -163,15 +206,14 @@ describe('integration tools', () => { }); it('context tool accepts an argument-free query without an input object', async () => { - const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue({ + const messageContext: IntegrationMessageContext | null = { integrationConnectionId: 'slack:cred-a', platform: 'slack', target: { type: 'thread', threadId: 'slack:C123:123.456' }, messageId: '123.456', interactingUserId: 'U123', updatedAt: '2026-05-18T10:00:00.000Z', - }); + }; const queryExecutor = mock(); queryExecutor.execute.mockResolvedValue({ ok: true, @@ -180,7 +222,6 @@ describe('integration tools', () => { const tool = createIntegrationContextTool({ descriptor: getIntegrationToolConnectionDescriptors([slackA])[0], - messageContextStore, queryExecutor, }).build(); const schema = tool.inputSchema as z.ZodType; @@ -189,7 +230,13 @@ describe('integration tools', () => { const result = await tool.handler!( { query: 'get_current_user' }, - { persistence: { threadId: 'thread-1', resourceId: 'resource-1' } }, + { + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }, ); expect(result).toEqual({ @@ -200,14 +247,17 @@ describe('integration tools', () => { descriptor: expect.any(Object), query: 'get_user', input: { userId: 'U123' }, - persistence: { threadId: 'thread-1', resourceId: 'resource-1' }, + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, }); }); it('context tool schema requires platform IDs for user and channel lookups', () => { const tool = createIntegrationContextTool({ descriptor: getIntegrationToolConnectionDescriptors([slackA])[0], - messageContextStore: mock(), queryExecutor: mock(), }).build(); const schema = tool.inputSchema as z.ZodType; @@ -240,7 +290,6 @@ describe('integration tools', () => { 'search_channels', ], }))[0], - messageContextStore: mock(), queryExecutor: mock(), }).build(); const schema = tool.inputSchema as z.ZodType; @@ -279,7 +328,6 @@ describe('integration tools', () => { 'search_issues', ], }))[0], - messageContextStore: mock(), queryExecutor: mock(), }).build(); const schema = tool.inputSchema as z.ZodType; @@ -357,7 +405,6 @@ describe('integration tools', () => { descriptor: getIntegrationToolConnectionDescriptors([linear], 'agent-1', () => ({ contextQueries: ['search_teams', 'search_labels'], }))[0], - messageContextStore: mock(), queryExecutor, }).build(); const schema = tool.inputSchema as z.ZodType; @@ -417,7 +464,6 @@ describe('integration tools', () => { const contextTool = createIntegrationContextTool({ descriptor, - messageContextStore: mock(), queryExecutor: mock(), }).build(); const actionTool = createIntegrationActionTool({ @@ -432,7 +478,7 @@ describe('integration tools', () => { it('respond returns a structured error when no latest message context exists', async () => { const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue(null); + const messageContext: IntegrationMessageContext | null = null; const actionExecutor = mock(); const tool = createIntegrationActionTool({ @@ -443,7 +489,13 @@ describe('integration tools', () => { const result = await tool.handler!( { action: 'respond', input: { message: { text: 'Hello' } } }, - makeInterruptibleCtx(), + makeInterruptibleCtx({ + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }), ); expect(result).toEqual({ @@ -472,7 +524,7 @@ describe('integration tools', () => { updatedAt: '2026-07-31T10:00:01.000Z', }; const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue(inboundContext); + const messageContext: IntegrationMessageContext | null = inboundContext; const actionExecutor = mock(); actionExecutor.execute.mockImplementation(async ({ action }) => action === 'respond' @@ -495,7 +547,13 @@ describe('integration tools', () => { { action: 'do_not_respond' }, ], }, - { persistence: { threadId: 'thread-1', resourceId: 'resource-1' } }, + { + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }, ); expect(actionExecutor.execute).toHaveBeenLastCalledWith( @@ -521,7 +579,7 @@ describe('integration tools', () => { updatedAt: '2026-07-31T10:00:00.000Z', }; const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue(inboundContext); + const messageContext: IntegrationMessageContext | null = inboundContext; const actionExecutor = mock(); actionExecutor.execute .mockResolvedValueOnce({ @@ -551,7 +609,13 @@ describe('integration tools', () => { { action: 'do_not_respond' }, ], }, - { persistence: { threadId: 'thread-1', resourceId: 'resource-1' } }, + { + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }, ); expect(actionExecutor.execute).toHaveBeenLastCalledWith( @@ -1158,6 +1222,7 @@ describe('integration tools', () => { messageId: '123.456', }), ); + expect(readIntegrationMessageContext(ctx.persistence)?.messageId).toBe('123.456'); expect(ctx.suspend).toHaveBeenCalledWith({ type: 'integration_action', action: 'send_channel_message', @@ -1170,15 +1235,49 @@ describe('integration tools', () => { }); }); + it('keeps the turn context and propagates an action context write failure', async () => { + const original: IntegrationMessageContext = { + integrationConnectionId: 'slack:cred-a', + platform: 'slack', + target: { type: 'thread', threadId: 'slack:C1:1' }, + messageId: 'original', + updatedAt: '2026-09-18T10:00:00.000Z', + }; + const ctx = makeInterruptibleCtx({ + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(original), + }, + }); + const messageContextStore = mock(); + messageContextStore.setLatest.mockRejectedValue(new Error('database unavailable')); + const actionExecutor = mock(); + actionExecutor.execute.mockResolvedValue({ + ok: true, + messageContext: { ...original, messageId: 'sent' }, + }); + const tool = createIntegrationActionTool({ + descriptor: getIntegrationToolConnectionDescriptors([slackA])[0], + messageContextStore, + actionExecutor, + }).build(); + await expect( + tool.handler!({ action: 'respond', input: { message: { text: 'hello' } } }, ctx), + ).rejects.toThrow('database unavailable'); + expect(readIntegrationMessageContext(ctx.persistence)).toEqual(original); + expect(ctx.suspend).not.toHaveBeenCalled(); + }); + it('section accessory button sends first, updates message context, then suspends', async () => { const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue({ + const messageContext: IntegrationMessageContext | null = { integrationConnectionId: 'slack:cred-a', platform: 'slack', target: { type: 'thread', threadId: 'slack:C123:123.456', channelId: 'slack:C123' }, messageId: '123.456', updatedAt: '2026-05-18T10:00:00.000Z', - }); + }; const actionExecutor = mock(); actionExecutor.execute.mockResolvedValue({ ok: true, @@ -1190,7 +1289,13 @@ describe('integration tools', () => { updatedAt: '2026-05-18T10:01:00.000Z', }, }); - const ctx = makeInterruptibleCtx(); + const ctx = makeInterruptibleCtx({ + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }); const tool = createIntegrationActionTool({ descriptor: getIntegrationToolConnectionDescriptors([slackA])[0], @@ -1325,13 +1430,13 @@ describe('integration tools', () => { it('does not bind for respond (operates on an existing thread)', async () => { const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue({ + const messageContext: IntegrationMessageContext | null = { integrationConnectionId: 'slack:cred-a', platform: 'slack', target: { type: 'thread' as const, threadId: 'slack:C1:1.1', channelId: 'slack:C1' }, messageId: '1.1', updatedAt: '2026-08-20T10:00:00.000Z', - }); + }; const actionExecutor = mock(); actionExecutor.execute.mockResolvedValue({ ok: true, messageContext: sentContext }); @@ -1343,7 +1448,13 @@ describe('integration tools', () => { await tool.handler!( { action: 'respond', input: { message: { text: 'hi' } } }, - makeInterruptibleCtx(), + makeInterruptibleCtx({ + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }), ); expect(messageContextStore.bindSession).not.toHaveBeenCalled(); @@ -1395,7 +1506,7 @@ describe('integration tools', () => { it('action tool preserves the current subject when updating message context', async () => { const messageContextStore = mock(); - messageContextStore.getLatest.mockResolvedValue({ + const messageContext: IntegrationMessageContext | null = { integrationConnectionId: 'slack:cred-a', platform: 'slack', target: { type: 'thread', threadId: 'slack:C123:123.456', channelId: 'slack:C123' }, @@ -1407,7 +1518,7 @@ describe('integration tools', () => { title: 'Fix signup', }, updatedAt: '2026-05-18T10:00:00.000Z', - }); + }; const actionExecutor = mock(); actionExecutor.execute.mockResolvedValue({ ok: true, @@ -1428,7 +1539,13 @@ describe('integration tools', () => { const result = await tool.handler!( { action: 'send_dm', input: { userId: 'slack:U123', message: { text: 'Hello' } } }, - makeInterruptibleCtx(), + makeInterruptibleCtx({ + persistence: { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: encodeIntegrationMessageContext(messageContext), + }, + }), ); expect(messageContextStore.setLatest).toHaveBeenCalledWith( diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts b/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts index d324aa64391..64814368a1b 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts @@ -20,12 +20,12 @@ import { AgentChatAttachmentService, type StoredAttachmentRef, } from '../agent-chat-attachment.service'; -import type { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service'; +import type { + AgentExecutionOrchestratorService, + ExecuteForChatPublishedConfig, +} from '../agent-execution-orchestrator.service'; import { AgentExecutionService } from '../agent-execution.service'; -import { - hashAgentSandboxPrincipal, - type AgentSandboxPrincipalHash, -} from '../agent-sandbox-principal'; +import { hashAgentSandboxPrincipal } from '../agent-sandbox-principal'; import { integrationMemoryResourceId } from '../utils/agent-memory-scope'; import { resolveInboundMimeType } from '../utils/inbound-attachments'; import type { @@ -45,7 +45,7 @@ import { CallbackStore, type CallbackMetadata } from './callback-store'; import type { ComponentMapper, ShortenCallback } from './component-mapper'; import { loadChatSdk } from './esm-loader'; import { IntegrationMessageContextService } from './integration-message-context.service'; -import type { ReplyExpectation } from './integration-tools'; +import type { ReplyExpectation, IntegrationMessageContext } from './integration-tools'; import { N8NCheckpointStorage } from './n8n-checkpoint-storage'; import { downloadDiscordAttachment } from './platforms/discord-operations'; @@ -90,27 +90,12 @@ function stillWaitingNotice(suspendPayload: unknown): string { return `⏳ ${title} — use the buttons on that card and I'll continue from there.`; } -interface AgentExecutor { - executeForChatPublished(config: { - agentId: string; - projectId: string; - message: string; - modelMessage?: string; - author?: AgentMessageAuthor; - attachments?: StoredAttachmentRef[]; - memory: { threadId: InternalThread; resourceId: string }; - integrationType?: string; - sandboxPrincipalHash: AgentSandboxPrincipalHash; - }): AsyncGenerator; - - resumeForChat(config: { - agentId: string; - projectId: string; - runId: string; - toolCallId: string; - resumeData: unknown; - integrationType?: string; - }): AsyncGenerator; +interface AgentExecutor extends Pick { + executeForChatPublished( + config: Omit & { + memory: { threadId: InternalThread; resourceId: string }; + }, + ): AsyncGenerator; /** * The thread's still-open suspension, if the run is parked on one right now. @@ -258,6 +243,8 @@ export class AgentChatBridge { attachments, integrationType, sandboxPrincipalHash, + messageContext, + contextConversation, }) { yield* agentService.executeForChatPublished({ agentId: aid, @@ -275,6 +262,8 @@ export class AgentChatBridge { }, integrationType, sandboxPrincipalHash, + messageContext, + contextConversation, }); }, async *resumeForChat(config) { @@ -398,16 +387,21 @@ export class AgentChatBridge { // --------------------------------------------------------------------------- /** - * Resume from a server-side trigger rather than a user action. Rebuilds the - * platform thread from the stored agent thread id, so the continuation streams - * back into the conversation the suspension was posted to. + * Resume a server-side trigger in the checkpoint's reply destination. + * Older checkpoints can fall back to the stored agent thread ID. */ async resumeInAgentThread( agentThreadId: string, runId: string, toolCallId: string, resumeData: unknown, + context?: { messageContext: IntegrationMessageContext | null; allowLegacyThreadId: boolean }, ): Promise { + const target = context?.messageContext?.replyTarget ?? context?.messageContext?.target; + if (context && !target?.threadId && !context.allowLegacyThreadId) { + this.logger.warn('Agent resume has no reply destination', { agentId: this.agentId, runId }); + return; + } const prefix = `${this.agentId}:`; const withoutAgentPrefix = agentThreadId.startsWith(prefix) ? agentThreadId.slice(prefix.length) @@ -419,14 +413,16 @@ export class AgentChatBridge { // expects the real platform id only. const platformThreadId = withoutAgentPrefix.replace(SESSION_GENERATION_SUFFIX_RE, ''); const sdkThreadId = - this.integrationImpl?.formatThreadId?.toSdk(platformThreadId) ?? platformThreadId; + target?.threadId ?? + this.integrationImpl?.formatThreadId?.toSdk(platformThreadId) ?? + platformThreadId; await this.hitlResumeHandler.executeResume( this.chat.thread(sdkThreadId), runId, toolCallId, resumeData, - false, + { notifyOnDuplicate: false, ...(context ? { messageContext: context.messageContext } : {}) }, ); } @@ -670,30 +666,13 @@ export class AgentChatBridge { this.messageContextBridge.resolveSubject(message), ]); statusHandle = onceStatusHandle(bridgeExecutionContext.statusHandle); - const latestContextOptions = { + const messageContext = this.messageContextBridge.capture(thread, { messageId: message.id, interactingUserId: message.author.userId, ...bridgeExecutionContext.platformAgentContext, subject, replyExpectation, - }; - await this.messageContextBridge.updateLatest( - threadId.id, - message.author.userId, - thread, - latestContextOptions, - ); - // Tools look up context on persistence.threadId (the execution - // session). When a bound reply continues a task, that is the origin - // thread, not the Slack thread — store this turn there too. - if (memoryThreadId.id !== threadId.id) { - await this.messageContextBridge.updateLatest( - memoryThreadId.id, - memoryResourceId, - thread, - latestContextOptions, - ); - } + }); // threadId.id is agent-prefixed for shared conversation history; // resourceId keeps the author identity so episodic recall follows them. // Always run the published snapshot — integrations are production traffic. @@ -706,6 +685,8 @@ export class AgentChatBridge { ? `${bridgeExecutionContext.historyContext}\n\n${labelledText}` : labelledText; const stream = this.agentService.executeForChatPublished({ + messageContext, + contextConversation: { threadId: threadId.id, resourceId: message.author.userId }, agentId: this.agentId, projectId: this.n8nProjectId, message: textWithNotes, diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts b/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts index 3c7ee311d79..0313e12bd61 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-hitl-resume-handler.ts @@ -1,4 +1,3 @@ -import type { StreamChunk } from '@n8n/agents'; import type { AgentIntegrationConfig } from '@n8n/api-types'; import type { ActionEvent, Thread } from 'chat'; import type { Logger } from 'n8n-workflow'; @@ -14,17 +13,12 @@ import type { AgentChatMessageContextBridge } from './agent-chat-message-context import type { AgentChatStreamConsumer } from './agent-chat-stream-consumer'; import type { CallbackStore } from './callback-store'; import type { InternalThread } from './types'; +import type { + ResumeForChatConfig, + AgentExecutionOrchestratorService, +} from '../agent-execution-orchestrator.service'; -interface ResumeExecutor { - resumeForChat(config: { - agentId: string; - projectId: string; - runId: string; - toolCallId: string; - resumeData: unknown; - integrationType?: string; - }): AsyncGenerator; -} +type ResumeExecutor = Pick; interface AgentChatHitlResumeHandlerOptions { agentId: string; @@ -73,22 +67,20 @@ export class AgentChatHitlResumeHandler { const parsed = this.parseActionId(callbackData.actionId, callbackData.value); if (!parsed) return; - // Persist the interacting user / messageId into the thread's message - // context so tools running on resume can read it via the message - // context store — no need to bolt a duplicate copy onto resumeData. const platformThreadId = this.options.resolvePlatformThreadId(thread); const threadId = this.options.toAgentThreadId(platformThreadId); - await this.options.messageContextBridge.updateLatest(threadId.id, event.user.userId, thread, { + const messageContext = this.options.messageContextBridge.capture(thread, { messageId: event.messageId, interactingUserId: event.user.userId, ...this.options.getPlatformAgentContext(), - // The resume response streams back to this thread like any chat turn, - // so the same reply-delivery rules apply. replyExpectation: 'required', }); await this.cleanUpBeforeResume(event, parsed.resumeData, callbackData); - await this.executeResume(thread, parsed.runId, parsed.toolCallId, parsed.resumeData); + await this.executeResume(thread, parsed.runId, parsed.toolCallId, parsed.resumeData, { + messageContext, + contextConversation: { threadId: threadId.id, resourceId: event.user.userId }, + }); } /** Parsed result from an action ID. */ @@ -230,9 +222,11 @@ export class AgentChatHitlResumeHandler { runId: string, toolCallId: string, resumeData: unknown, - /** Tell the user the action was already handled. Only for their own clicks. */ - notifyOnDuplicate = true, + options: Pick & { + notifyOnDuplicate?: boolean; + } = {}, ): Promise { + const { notifyOnDuplicate = true, ...context } = options; if (this.activeResumedRuns.has(runId)) { this.options.logger.warn('[AgentChatBridge] Run is already active', { runId, toolCallId }); if (notifyOnDuplicate) await thread.post('This action has already been handled'); @@ -245,6 +239,7 @@ export class AgentChatHitlResumeHandler { const statusHandle = onceStatusHandle(resumeExecutionContext.statusHandle); try { const stream = this.options.agentService.resumeForChat({ + ...context, agentId: this.options.agentId, projectId: this.options.projectId, runId, diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-message-context.ts b/packages/cli/src/modules/agents/integrations/agent-chat-message-context.ts index c0847aea0bd..ae2d57aeb32 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-message-context.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-message-context.ts @@ -11,7 +11,7 @@ import { type SessionBinding, } from './integration-tools'; -interface UpdateLatestMessageContextOptions { +interface CaptureMessageContextOptions { messageId?: string; interactingUserId?: string; agentUserId?: string; @@ -27,17 +27,12 @@ export class AgentChatMessageContextBridge { private readonly logger: Logger, ) {} - async updateLatest( - threadId: string, - resourceId: string, + capture( thread: Thread, - options: UpdateLatestMessageContextOptions = {}, - ): Promise { - if (!this.messageContextStore) return undefined; - + options: CaptureMessageContextOptions = {}, + ): IntegrationMessageContext { const integrationConnectionId = buildIntegrationConnectionId(this.integration); - const previousContext = await this.getPreviousContext(threadId, integrationConnectionId); - const agentUserId = options.agentUserId ?? previousContext?.agentUserId; + const agentUserId = options.agentUserId; const target: IntegrationMessageContext['target'] = { type: 'thread', threadId: thread.id, @@ -51,7 +46,6 @@ export class AgentChatMessageContextBridge { ...(options.interactingUserId ? { interactingUserId: options.interactingUserId } : {}), ...(agentUserId ? { agentUserId } : {}), ...(options.subject ? { subject: options.subject } : {}), - ...(!options.subject && previousContext?.subject ? { subject: previousContext.subject } : {}), ...(options.replyExpectation ? { replyExpectation: options.replyExpectation, @@ -62,17 +56,7 @@ export class AgentChatMessageContextBridge { updatedAt: new Date().toISOString(), }; - try { - await this.messageContextStore.setLatest(threadId, resourceId, context); - return context; - } catch (error) { - this.logger.warn('[AgentChatBridge] Failed to update latest message context', { - agentId: this.agentId, - threadId, - error: error instanceof Error ? error.message : String(error), - }); - return undefined; - } + return context; } async resolveSubject(message: Message): Promise { @@ -114,27 +98,6 @@ export class AgentChatMessageContextBridge { if (!this.messageContextStore) return; await this.messageContextStore.unbindSession(threadId); } - - private async getPreviousContext( - threadId: string, - integrationConnectionId: string, - ): Promise { - if (!this.messageContextStore) return undefined; - try { - const previousContext = await this.messageContextStore.getLatest(threadId); - if (previousContext?.integrationConnectionId !== integrationConnectionId) { - return undefined; - } - return previousContext; - } catch (error) { - this.logger.warn('[AgentChatBridge] Failed to read previous message context', { - agentId: this.agentId, - threadId, - error: error instanceof Error ? error.message : String(error), - }); - return undefined; - } - } } function toIntegrationMessageSubject( diff --git a/packages/cli/src/modules/agents/integrations/integration-message-context.service.ts b/packages/cli/src/modules/agents/integrations/integration-message-context.service.ts index 2ab48f3c27b..775ca8a7ac1 100644 --- a/packages/cli/src/modules/agents/integrations/integration-message-context.service.ts +++ b/packages/cli/src/modules/agents/integrations/integration-message-context.service.ts @@ -1,15 +1,18 @@ +import type { ToolContext } from '@n8n/agents'; +import { Logger } from '@n8n/backend-common'; import { Service } from '@n8n/di'; import { isRecord } from '@n8n/utils/is-record'; import { jsonParse } from 'n8n-workflow'; import type { IntegrationMessageContext, - IntegrationMessageSubject, - IntegrationSubjectPerson, IntegrationMessageContextStore, - IntegrationMessageTarget, SessionBinding, -} from './integration-tools'; +} from './integration-tool-types'; +import { + isIntegrationMessageContext, + readIntegrationMessageContext, +} from './integration-message-context'; import { AgentResourceRepository } from '../repositories/agent-resource.repository'; import { AgentThreadRepository } from '../repositories/agent-thread.repository'; @@ -22,6 +25,7 @@ export class IntegrationMessageContextService implements IntegrationMessageConte constructor( private readonly threadRepository: AgentThreadRepository, private readonly resourceRepository: AgentResourceRepository, + private readonly logger: Logger, ) {} async getLatest(threadId: string): Promise { @@ -30,6 +34,44 @@ export class IntegrationMessageContextService implements IntegrationMessageConte return isIntegrationMessageContext(value) ? value : null; } + async getForResume( + persistence: NonNullable, + ): Promise { + const context = readIntegrationMessageContext(persistence); + return context === undefined ? await this.getLatest(persistence.threadId) : context; + } + + async getLatestForIncoming(threadId: string): Promise { + try { + return await this.getLatest(threadId); + } catch (error) { + this.logger.warn('Failed to read previous integration message context', { + threadId, + error: error instanceof Error ? error.message : String(error), + }); + return null; + } + } + + async installIncoming( + context: IntegrationMessageContext, + memory: SessionBinding, + conversation: SessionBinding, + ): Promise { + const scopes = + conversation.threadId === memory.threadId ? [conversation] : [conversation, memory]; + for (const { threadId, resourceId } of scopes) { + try { + await this.setLatest(threadId, resourceId, context); + } catch (error) { + this.logger.warn('Failed to update integration message context', { + threadId, + error: error instanceof Error ? error.message : String(error), + }); + } + } + } + async setLatest( threadId: string, resourceId: string, @@ -171,72 +213,8 @@ export class IntegrationMessageContextService implements IntegrationMessageConte } } -export function isIntegrationMessageContext(value: unknown): value is IntegrationMessageContext { - if (!value || typeof value !== 'object') return false; - const context = value as Record; - return ( - typeof context.integrationConnectionId === 'string' && - typeof context.platform === 'string' && - isIntegrationMessageTarget(context.target) && - (context.messageId === undefined || typeof context.messageId === 'string') && - (context.interactingUserId === undefined || typeof context.interactingUserId === 'string') && - (context.agentUserId === undefined || typeof context.agentUserId === 'string') && - (context.subject === undefined || isIntegrationMessageSubject(context.subject)) && - typeof context.updatedAt === 'string' - ); -} - function isSessionBinding(value: unknown): value is SessionBinding { return ( isRecord(value) && typeof value.threadId === 'string' && typeof value.resourceId === 'string' ); } - -function isIntegrationMessageSubject(value: unknown): value is IntegrationMessageSubject { - if (!value || typeof value !== 'object') return false; - const subject = value as Record; - return ( - typeof subject.type === 'string' && - typeof subject.id === 'string' && - (subject.title === undefined || typeof subject.title === 'string') && - (subject.description === undefined || typeof subject.description === 'string') && - (subject.url === undefined || typeof subject.url === 'string') && - (subject.status === undefined || typeof subject.status === 'string') && - (subject.labels === undefined || - (Array.isArray(subject.labels) && - subject.labels.every((label) => typeof label === 'string'))) && - (subject.assignee === undefined || isIntegrationSubjectPerson(subject.assignee)) && - (subject.author === undefined || isIntegrationSubjectPerson(subject.author)) - ); -} - -function isIntegrationSubjectPerson(value: unknown): value is IntegrationSubjectPerson { - if (!value || typeof value !== 'object') return false; - const person = value as Record; - return typeof person.id === 'string' && typeof person.name === 'string'; -} - -function isIntegrationMessageTarget(value: unknown): value is IntegrationMessageTarget { - if (!value || typeof value !== 'object') return false; - const target = value as Record; - if (target.type === 'thread') { - return ( - typeof target.threadId === 'string' && - (target.channelId === undefined || typeof target.channelId === 'string') && - (target.userId === undefined || typeof target.userId === 'string') - ); - } - if (target.type === 'channel') { - return ( - typeof target.channelId === 'string' && - (target.threadId === undefined || typeof target.threadId === 'string') - ); - } - if (target.type === 'dm') { - return ( - typeof target.userId === 'string' && - (target.threadId === undefined || typeof target.threadId === 'string') - ); - } - return false; -} diff --git a/packages/cli/src/modules/agents/integrations/integration-message-context.ts b/packages/cli/src/modules/agents/integrations/integration-message-context.ts new file mode 100644 index 00000000000..273e9f4fe8c --- /dev/null +++ b/packages/cli/src/modules/agents/integrations/integration-message-context.ts @@ -0,0 +1,119 @@ +import type { JSONObject, ToolContext } from '@n8n/agents'; +import { isRecord } from '@n8n/utils/is-record'; +import { jsonParse } from 'n8n-workflow'; + +import type { + IntegrationMessageContext, + IntegrationMessageSubject, + IntegrationSubjectPerson, + IntegrationMessageTarget, +} from './integration-tool-types'; + +const TURN_CONTEXT_KEY = 'n8nIntegrationMessageContext'; + +export function encodeIntegrationMessageContext( + context: IntegrationMessageContext | null, +): JSONObject { + return { [TURN_CONTEXT_KEY]: context ? jsonParse(JSON.stringify(context)) : null }; +} + +/** Undefined identifies an older checkpoint. Null must not use newer thread context. */ +export function readIntegrationMessageContext( + persistence: ToolContext['persistence'], +): IntegrationMessageContext | null | undefined { + const metadata = persistence?.hostMetadata; + if (!metadata || !Object.hasOwn(metadata, TURN_CONTEXT_KEY)) return undefined; + const value = metadata[TURN_CONTEXT_KEY]; + return isIntegrationMessageContext(value) ? value : null; +} + +export function replaceIntegrationMessageContext( + persistence: NonNullable, + context: IntegrationMessageContext, +): void { + persistence.hostMetadata = { + ...persistence.hostMetadata, + ...encodeIntegrationMessageContext(context), + }; +} + +export function inheritIntegrationMessageContext( + context: IntegrationMessageContext, + previous: IntegrationMessageContext | null, +): IntegrationMessageContext { + if (previous?.integrationConnectionId !== context.integrationConnectionId) return context; + return { + ...context, + ...(previous.subject && !context.subject ? { subject: previous.subject } : {}), + ...(previous.agentUserId && !context.agentUserId ? { agentUserId: previous.agentUserId } : {}), + }; +} + +export function isIntegrationMessageContext(value: unknown): value is IntegrationMessageContext { + if (!isRecord(value)) return false; + const context = value; + return ( + typeof context.integrationConnectionId === 'string' && + typeof context.platform === 'string' && + isIntegrationMessageTarget(context.target) && + (context.messageId === undefined || typeof context.messageId === 'string') && + (context.interactingUserId === undefined || typeof context.interactingUserId === 'string') && + (context.agentUserId === undefined || typeof context.agentUserId === 'string') && + (context.subject === undefined || isIntegrationMessageSubject(context.subject)) && + (context.replyExpectation === undefined || + context.replyExpectation === 'required' || + context.replyExpectation === 'optional') && + (context.replyTarget === undefined || isIntegrationMessageTarget(context.replyTarget)) && + (context.replyMessageId === undefined || typeof context.replyMessageId === 'string') && + typeof context.updatedAt === 'string' + ); +} + +function isIntegrationMessageSubject(value: unknown): value is IntegrationMessageSubject { + if (!isRecord(value)) return false; + const subject = value; + return ( + typeof subject.type === 'string' && + typeof subject.id === 'string' && + (subject.title === undefined || typeof subject.title === 'string') && + (subject.description === undefined || typeof subject.description === 'string') && + (subject.url === undefined || typeof subject.url === 'string') && + (subject.status === undefined || typeof subject.status === 'string') && + (subject.labels === undefined || + (Array.isArray(subject.labels) && + subject.labels.every((label) => typeof label === 'string'))) && + (subject.assignee === undefined || isIntegrationSubjectPerson(subject.assignee)) && + (subject.author === undefined || isIntegrationSubjectPerson(subject.author)) + ); +} + +function isIntegrationSubjectPerson(value: unknown): value is IntegrationSubjectPerson { + if (!isRecord(value)) return false; + const person = value; + return typeof person.id === 'string' && typeof person.name === 'string'; +} + +function isIntegrationMessageTarget(value: unknown): value is IntegrationMessageTarget { + if (!isRecord(value)) return false; + const target = value; + if (target.type === 'thread') { + return ( + typeof target.threadId === 'string' && + (target.channelId === undefined || typeof target.channelId === 'string') && + (target.userId === undefined || typeof target.userId === 'string') + ); + } + if (target.type === 'channel') { + return ( + typeof target.channelId === 'string' && + (target.threadId === undefined || typeof target.threadId === 'string') + ); + } + if (target.type === 'dm') { + return ( + typeof target.userId === 'string' && + (target.threadId === undefined || typeof target.threadId === 'string') + ); + } + return false; +} diff --git a/packages/cli/src/modules/agents/integrations/integration-tool-execution.ts b/packages/cli/src/modules/agents/integrations/integration-tool-execution.ts index c674d5e1d36..0c6607c069f 100644 --- a/packages/cli/src/modules/agents/integrations/integration-tool-execution.ts +++ b/packages/cli/src/modules/agents/integrations/integration-tool-execution.ts @@ -5,6 +5,11 @@ import type { z } from 'zod'; import { isTaskRunMemoryResourceId } from '../utils/agent-memory-scope'; import { messageSchema, type IntegrationCardComponent } from './integration-tool-definitions'; import { INTEGRATION_ERROR_CODES } from './integration-error-codes'; +import { + isIntegrationMessageContext, + readIntegrationMessageContext, + replaceIntegrationMessageContext, +} from './integration-message-context'; import type { IntegrationAction, IntegrationActionExecutor, @@ -20,11 +25,10 @@ import type { RawActionToolOperation, RawContextToolOperation } from './integrat export async function executeContextToolOperation(params: { operation: RawContextToolOperation; descriptor: IntegrationToolConnectionDescriptor; - messageContextStore: IntegrationMessageContextStore; queryExecutor: IntegrationContextQueryExecutor; persistence: ToolContext['persistence']; }): Promise { - const { operation, descriptor, messageContextStore, queryExecutor, persistence } = params; + const { operation, descriptor, queryExecutor, persistence } = params; if (isCurrentContextQuery(operation.query)) { if (!persistence) { @@ -36,7 +40,7 @@ export async function executeContextToolOperation(params: { }, }; } - const context = await messageContextStore.getLatest(persistence.threadId); + const context = readIntegrationMessageContext(persistence); if (!context || context.integrationConnectionId !== descriptor.integrationConnectionId) { if (operation.query === 'get_current_message_context') { return { ok: true, context: null }; @@ -110,14 +114,10 @@ export async function executeActionToolBatch(params: { ctx: ToolContext; }): Promise { const { operations, descriptor, messageContextStore, actionExecutor, ctx } = params; - let currentMessageContext = - ctx.persistence !== undefined - ? await getOptionalCurrentContext({ - descriptor, - messageContextStore, - threadId: ctx.persistence.threadId, - }) - : undefined; + let currentMessageContext = getOptionalCurrentContext({ + descriptor, + persistence: ctx.persistence, + }); const results: Array<{ action: IntegrationAction; result: unknown }> = []; for (const operation of operations) { @@ -178,22 +178,14 @@ export async function executeActionToolOperation(params: { let currentMessageContext = params.currentMessageContext; if (operation.action === 'respond') { if (!currentMessageContext) { - const contextResult = await getRespondContext({ - descriptor, - messageContextStore, - persistence, - }); + const contextResult = getRespondContext({ descriptor, persistence }); if (!contextResult.ok) { return contextResult; } currentMessageContext = contextResult.context; } } else if (!currentMessageContext && persistence) { - currentMessageContext = await getOptionalCurrentContext({ - descriptor, - messageContextStore, - threadId: persistence.threadId, - }); + currentMessageContext = getOptionalCurrentContext({ descriptor, persistence }); } const result = await actionExecutor.execute({ @@ -216,6 +208,7 @@ export async function executeActionToolOperation(params: { persistence.resourceId, messageContext, ); + replaceIntegrationMessageContext(persistence, messageContext); // Task-run sends bind the outbound thread so inbound replies continue // that task session. Chat mentions stay on their own thread. // respond/edit/reaction operate on existing threads and do not bind. @@ -310,42 +303,23 @@ function extractSuccessfulMessageContext(result: unknown): IntegrationMessageCon return isIntegrationMessageContext(messageContext) ? messageContext : undefined; } -function isIntegrationMessageContext(value: unknown): value is IntegrationMessageContext { - return ( - isRecord(value) && - typeof value.integrationConnectionId === 'string' && - typeof value.platform === 'string' && - isRecord(value.target) && - (value.agentUserId === undefined || typeof value.agentUserId === 'string') && - typeof value.updatedAt === 'string' - ); -} - -async function getOptionalCurrentContext(params: { +function getOptionalCurrentContext(params: { descriptor: IntegrationToolConnectionDescriptor; - messageContextStore: IntegrationMessageContextStore; - threadId: string; -}): Promise { - try { - const context = await params.messageContextStore.getLatest(params.threadId); - if (context?.integrationConnectionId !== params.descriptor.integrationConnectionId) { - return undefined; - } - return context; - } catch { - return undefined; - } -} - -async function getRespondContext(params: { - descriptor: IntegrationToolConnectionDescriptor; - messageContextStore: IntegrationMessageContextStore; persistence: ToolContext['persistence']; -}): Promise< +}): IntegrationMessageContext | undefined { + const context = readIntegrationMessageContext(params.persistence); + return context?.integrationConnectionId === params.descriptor.integrationConnectionId + ? context + : undefined; +} + +function getRespondContext(params: { + descriptor: IntegrationToolConnectionDescriptor; + persistence: ToolContext['persistence']; +}): | { ok: true; context: IntegrationMessageContext } - | { ok: false; error: { code: string; message: string } } -> { - const { descriptor, messageContextStore, persistence } = params; + | { ok: false; error: { code: string; message: string } } { + const { descriptor, persistence } = params; if (!persistence) { return { ok: false, @@ -356,7 +330,7 @@ async function getRespondContext(params: { }; } - const context = await messageContextStore.getLatest(persistence.threadId); + const context = readIntegrationMessageContext(persistence); if (!context) { return { ok: false, diff --git a/packages/cli/src/modules/agents/integrations/integration-tool-factory.ts b/packages/cli/src/modules/agents/integrations/integration-tool-factory.ts index 04ab6c8080a..c9fd4049735 100644 --- a/packages/cli/src/modules/agents/integrations/integration-tool-factory.ts +++ b/packages/cli/src/modules/agents/integrations/integration-tool-factory.ts @@ -105,10 +105,9 @@ export function buildIntegrationConnectionId(integration: IntegrationToolConnect export function createIntegrationContextTool(params: { descriptor: IntegrationToolConnectionDescriptor; - messageContextStore: IntegrationMessageContextStore; queryExecutor: IntegrationContextQueryExecutor; }) { - const { descriptor, messageContextStore, queryExecutor } = params; + const { descriptor, queryExecutor } = params; return new Tool(descriptor.contextToolName) .description(buildContextToolDescription(descriptor)) @@ -124,7 +123,6 @@ export function createIntegrationContextTool(params: { result: await executeContextToolOperation({ operation, descriptor, - messageContextStore, queryExecutor, persistence: ctx.persistence, }), @@ -138,7 +136,6 @@ export function createIntegrationContextTool(params: { return await executeContextToolOperation({ operation: toSingleContextOperation(toolInput), descriptor, - messageContextStore, queryExecutor, persistence: ctx.persistence, });