mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
refactor(core): Bind integration context to agent turns (no-changelog) (#39043)
This commit is contained in:
Robin Braumann
GitHub
parent
a70938f228
commit
87fe3c02f6
|
|
@ -2206,17 +2206,17 @@ function createRuntimeWithCheckpointStore(
|
|||
}
|
||||
|
||||
function makeClaimingCheckpointStore(): ClaimingCheckpointStore {
|
||||
const checkpoints = new Map<string, SerializableAgentState>();
|
||||
const checkpoints = new Map<string, string>();
|
||||
|
||||
return {
|
||||
save: vi.fn(async (key: string, state: SerializableAgentState): Promise<void> => {
|
||||
await Promise.resolve();
|
||||
checkpoints.set(key, structuredClone(state));
|
||||
checkpoints.set(key, JSON.stringify(state));
|
||||
}),
|
||||
load: vi.fn(async (key: string): Promise<SerializableAgentState | undefined> => {
|
||||
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<void> => {
|
||||
await Promise.resolve();
|
||||
|
|
@ -2225,8 +2225,8 @@ function makeClaimingCheckpointStore(): ClaimingCheckpointStore {
|
|||
claimForResume: vi.fn<ClaimForResume>(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<ToolContext['persistence']> = [];
|
||||
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 () => {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<AiConfig>({
|
||||
|
|
@ -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<AgentExecutionService>();
|
||||
const telemetry = mock<Telemetry>();
|
||||
const runtimeCacheService = mock<AgentRuntimeCacheService>();
|
||||
const integrationMessageContextService = mock<IntegrationMessageContextService>();
|
||||
const contextService = new IntegrationMessageContextService(
|
||||
mock<AgentThreadRepository>(),
|
||||
mock<AgentResourceRepository>(),
|
||||
mockLogger(),
|
||||
);
|
||||
const integrationMessageContextService = Object.assign(contextService, {
|
||||
getLatest: vi.spyOn(contextService, 'getLatest'),
|
||||
setLatest: vi.spyOn(contextService, 'setLatest').mockResolvedValue(undefined),
|
||||
});
|
||||
const agentRunTracingService = mock<AgentRunTracingService>();
|
||||
const externalHooks = mock<ExternalHooks>();
|
||||
const agentSandboxRuntimeService = mock<AgentSandboxRuntimeService>({
|
||||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<AgentRunTracingService>();
|
||||
const executionLevelTracer = mock<ExecutionLevelTracer>();
|
||||
const nodeToolAiGatewayService = mock<NodeToolAiGatewayService>();
|
||||
const integrationMessageContextService = mock<IntegrationMessageContextService>();
|
||||
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 },
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<AgentBackgroundJobService>();
|
||||
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 () => {
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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 }),
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string | undefined> {
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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<IntegrationMessageContextService>();
|
||||
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<IntegrationMessageContextService>();
|
||||
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<IntegrationMessageContextService>();
|
||||
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<IntegrationMessageContextService>();
|
||||
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<IntegrationMessageContextService>();
|
||||
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 () => {
|
||||
|
|
|
|||
|
|
@ -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 () => ({}),
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -20,6 +20,8 @@ import { getIntegrationToolConnectionDescriptors } from '../../integration-tools
|
|||
|
||||
export type ReplayWebhookOptions = { waitUntil?: (task: Promise<unknown>) => void };
|
||||
|
||||
type AgentExecutor = ConstructorParameters<typeof AgentChatBridge>[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<StreamChunk> {
|
||||
|
|
@ -196,6 +190,8 @@ export interface ReplayContextSetup<TChat extends ChatInstance = ChatInstance> {
|
|||
descriptor: ReturnType<typeof getIntegrationToolConnectionDescriptors>[number];
|
||||
integration: AgentIntegrationConfig;
|
||||
messageContextStore: MemoryMessageContextStore;
|
||||
latestContext: () => IntegrationMessageContext | undefined;
|
||||
latestThreadId: () => string | undefined;
|
||||
nextStream: (chunks: StreamChunk[]) => void;
|
||||
shutdown: () => Promise<void>;
|
||||
}
|
||||
|
|
@ -215,9 +211,18 @@ export function createReplayContextSetup<TChat extends ChatInstance>(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<AgentExecutor['executeForChatPublished']>((config) => {
|
||||
selectedContext = config.messageContext ?? undefined;
|
||||
selectedThreadId = config.memory.threadId.id;
|
||||
return toStream(stream);
|
||||
}),
|
||||
resumeForChat: vi.fn<AgentExecutor['resumeForChat']>((config) => {
|
||||
selectedContext = config.messageContext ?? undefined;
|
||||
return toStream(stream);
|
||||
}),
|
||||
};
|
||||
const messageContextStore = new MemoryMessageContextStore();
|
||||
|
||||
|
|
@ -248,6 +253,8 @@ export function createReplayContextSetup<TChat extends ChatInstance>(params: {
|
|||
descriptor,
|
||||
integration: params.integration,
|
||||
messageContextStore,
|
||||
latestContext: () => selectedContext,
|
||||
latestThreadId: () => selectedThreadId,
|
||||
nextStream: (chunks: StreamChunk[]) => {
|
||||
stream = chunks;
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<typeof persistence>(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: [] },
|
||||
|
|
|
|||
|
|
@ -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<IntegrationMessageContextStore>();
|
||||
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<IntegrationContextQueryExecutor>();
|
||||
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<IntegrationActionExecutor>();
|
||||
actionExecutor.execute.mockResolvedValue({ ok: true, messageContext: sentContext });
|
||||
const messageContextStore = mock<IntegrationMessageContextStore>();
|
||||
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<IntegrationMessageContextStore>();
|
||||
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<IntegrationContextQueryExecutor>();
|
||||
|
||||
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<IntegrationMessageContextStore>();
|
||||
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<IntegrationContextQueryExecutor>();
|
||||
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<IntegrationMessageContextStore>(),
|
||||
queryExecutor: mock<IntegrationContextQueryExecutor>(),
|
||||
}).build();
|
||||
const schema = tool.inputSchema as z.ZodType;
|
||||
|
|
@ -240,7 +290,6 @@ describe('integration tools', () => {
|
|||
'search_channels',
|
||||
],
|
||||
}))[0],
|
||||
messageContextStore: mock<IntegrationMessageContextStore>(),
|
||||
queryExecutor: mock<IntegrationContextQueryExecutor>(),
|
||||
}).build();
|
||||
const schema = tool.inputSchema as z.ZodType;
|
||||
|
|
@ -279,7 +328,6 @@ describe('integration tools', () => {
|
|||
'search_issues',
|
||||
],
|
||||
}))[0],
|
||||
messageContextStore: mock<IntegrationMessageContextStore>(),
|
||||
queryExecutor: mock<IntegrationContextQueryExecutor>(),
|
||||
}).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<IntegrationMessageContextStore>(),
|
||||
queryExecutor,
|
||||
}).build();
|
||||
const schema = tool.inputSchema as z.ZodType;
|
||||
|
|
@ -417,7 +464,6 @@ describe('integration tools', () => {
|
|||
|
||||
const contextTool = createIntegrationContextTool({
|
||||
descriptor,
|
||||
messageContextStore: mock<IntegrationMessageContextStore>(),
|
||||
queryExecutor: mock<IntegrationContextQueryExecutor>(),
|
||||
}).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<IntegrationMessageContextStore>();
|
||||
messageContextStore.getLatest.mockResolvedValue(null);
|
||||
const messageContext: IntegrationMessageContext | null = null;
|
||||
const actionExecutor = mock<IntegrationActionExecutor>();
|
||||
|
||||
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<IntegrationMessageContextStore>();
|
||||
messageContextStore.getLatest.mockResolvedValue(inboundContext);
|
||||
const messageContext: IntegrationMessageContext | null = inboundContext;
|
||||
const actionExecutor = mock<IntegrationActionExecutor>();
|
||||
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<IntegrationMessageContextStore>();
|
||||
messageContextStore.getLatest.mockResolvedValue(inboundContext);
|
||||
const messageContext: IntegrationMessageContext | null = inboundContext;
|
||||
const actionExecutor = mock<IntegrationActionExecutor>();
|
||||
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<IntegrationMessageContextStore>();
|
||||
messageContextStore.setLatest.mockRejectedValue(new Error('database unavailable'));
|
||||
const actionExecutor = mock<IntegrationActionExecutor>();
|
||||
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<IntegrationMessageContextStore>();
|
||||
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<IntegrationActionExecutor>();
|
||||
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<IntegrationMessageContextStore>();
|
||||
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<IntegrationActionExecutor>();
|
||||
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<IntegrationMessageContextStore>();
|
||||
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<IntegrationActionExecutor>();
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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<StreamChunk>;
|
||||
|
||||
resumeForChat(config: {
|
||||
agentId: string;
|
||||
projectId: string;
|
||||
runId: string;
|
||||
toolCallId: string;
|
||||
resumeData: unknown;
|
||||
integrationType?: string;
|
||||
}): AsyncGenerator<StreamChunk>;
|
||||
interface AgentExecutor extends Pick<AgentExecutionOrchestratorService, 'resumeForChat'> {
|
||||
executeForChatPublished(
|
||||
config: Omit<ExecuteForChatPublishedConfig, 'memory'> & {
|
||||
memory: { threadId: InternalThread; resourceId: string };
|
||||
},
|
||||
): AsyncGenerator<StreamChunk>;
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<StreamChunk>;
|
||||
}
|
||||
type ResumeExecutor = Pick<AgentExecutionOrchestratorService, 'resumeForChat'>;
|
||||
|
||||
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<ResumeForChatConfig, 'messageContext' | 'contextConversation'> & {
|
||||
notifyOnDuplicate?: boolean;
|
||||
} = {},
|
||||
): Promise<void> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<unknown, unknown>,
|
||||
options: UpdateLatestMessageContextOptions = {},
|
||||
): Promise<IntegrationMessageContext | undefined> {
|
||||
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<unknown>): Promise<IntegrationMessageSubject | undefined> {
|
||||
|
|
@ -114,27 +98,6 @@ export class AgentChatMessageContextBridge {
|
|||
if (!this.messageContextStore) return;
|
||||
await this.messageContextStore.unbindSession(threadId);
|
||||
}
|
||||
|
||||
private async getPreviousContext(
|
||||
threadId: string,
|
||||
integrationConnectionId: string,
|
||||
): Promise<IntegrationMessageContext | undefined> {
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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<IntegrationMessageContext | null> {
|
||||
|
|
@ -30,6 +34,44 @@ export class IntegrationMessageContextService implements IntegrationMessageConte
|
|||
return isIntegrationMessageContext(value) ? value : null;
|
||||
}
|
||||
|
||||
async getForResume(
|
||||
persistence: NonNullable<ToolContext['persistence']>,
|
||||
): Promise<IntegrationMessageContext | null> {
|
||||
const context = readIntegrationMessageContext(persistence);
|
||||
return context === undefined ? await this.getLatest(persistence.threadId) : context;
|
||||
}
|
||||
|
||||
async getLatestForIncoming(threadId: string): Promise<IntegrationMessageContext | null> {
|
||||
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<void> {
|
||||
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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<JSONObject>(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<ToolContext['persistence']>,
|
||||
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;
|
||||
}
|
||||
|
|
@ -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<unknown> {
|
||||
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<unknown> {
|
||||
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<IntegrationMessageContext | undefined> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user