import { AgentEvent, createScopedWorkspace, filterRuntimeSkillSource } from '@n8n/agents'; import type { Message, Workspace, ScopedMemoryTaskEvent, AgentEventData, MemoryTaskUsageReport, } from '@n8n/agents'; import { applyBranchReadOnlyOverrides, buildProxyHeaders, mcpConnectRequestSchema, credentialSetupHintSchema, formatAttachmentSizeLimit, TEMPLATED_CUSTOM_AUTH_CREDENTIAL_TYPE, type InstanceAiAttachment, type InstanceAiHandoffContext, type InstanceAiAgentAttachment, type InstanceAiFileAttachment, type InstanceAiNodesAttachment, type InstanceAiResourceAttachment, type InstanceAiWorkflowAttachment, type InstanceAiConfirmRequest, type InstanceAiCredits, type InstanceAiConfirmResponse, type InstanceAiEvent, type InstanceAiThreadStatusResponse, } from '@n8n/api-types'; import { Logger, ModuleRegistry } from '@n8n/backend-common'; import { SsrfProtectionService } from '@n8n/backend-network'; import { GlobalConfig, SsrfProtectionConfig, type InstanceAiConfig } from '@n8n/config'; import { UserRepository, type User } from '@n8n/db'; import { OnLeaderStepdown, OnLeaderTakeover, OnPubSubEvent } from '@n8n/decorators'; import { Container, Service } from '@n8n/di'; import { MAX_STEPS, createInstanceAgent, createAllTools, createLazyRuntimeWorkspace, createLazyWorkspaceRuntimeSkillSource, getPromptWorkspaceRoot, getWorkspaceRoot, loadInstanceAiRuntimeSkillSource, disabledInstanceAiSkillIds, createInstanceAiTraceContext, threadProvenanceMetadata, createInternalOperationTraceContext, emitAgentSnapshotTraceEvent, createInstanceAiLivenessPolicyConfig, InstanceAiLivenessPolicy, McpClientManager, createDomainAccessTracker, BackgroundTaskManager, MemoryTaskRegistry, classifyAttachments, buildAttachmentManifest, getDateTimeSection, isParseableAttachment, enrichMessageWithBackgroundTasks, isQuotaExhaustedError, PlannedTaskCoordinator, PlannedTaskStorage, PLANNED_TASK_PERMISSION_OVERRIDES, releaseTraceClient, resumeAgentRun, RunStateRegistry, shutdownProductTelemetryProviders, tokenUsageToBuilderUsageItems, RunDebugBuffer, buildRunDebugLabel, createRunDebugStepHooks, streamAgentRun, truncateToTitle, generateTitleForRun, patchThread, createOrchestratorRunControl, createOrchestratorRunControlForState, orchestratorAgentId, resolveAgentPreviewSession, saveAgentBuilderTarget, type ConfirmationData, type DomainAccessTracker, type InstanceAiContext, type ManagedBackgroundTask, type McpServerConfig, type ModelConfig, type AgentSnapshotArtifact, type OrchestrationContext, type InstanceAiTraceContext, type PlannedTaskGraph, type PlannedTaskRecord, type PlannedTaskService, type PlannedWorkflowVerification, type OrchestratorRunHandoffState, type OrchestratorRunStopSignal, type SpawnBackgroundTaskOptions, type SpawnBackgroundTaskResult, type ServiceProxyConfig, type StreamableAgent, type SuspendedRunState, type SuspensionInfo, type WorkflowBuildOutcome, type WorkflowLoopWorkItemRecord, type WorkflowSetupRoutingClaim, type WorkflowTaskService, type WorkflowVerificationObligation, type WorkSummary, type RunTokenUsage, type RunDebugRecord, WorkflowTaskCoordinator, WorkflowLoopStorage, ThreadTaskStorage, } from '@n8n/instance-ai'; import { buildResumeData, toConfirmationData } from '@n8n/instance-ai/confirmation-payload'; import type { Scope } from '@n8n/permissions'; import { redactTelemetryProperties, redactTelemetryText, TELEMETRY_EVENT } from '@n8n/telemetry'; import { getErrorMessage } from '@n8n/utils/errors/get-error-message'; import { lazyImport } from '@n8n/utils/lazy-import'; import { setSchemaBaseDirs } from '@n8n/workflow-sdk'; import { ErrorReporter, InstanceSettings } from 'n8n-core'; import { OperationalError, UnexpectedError, UserError } from 'n8n-workflow'; import { nanoid } from 'nanoid'; import { N8N_VERSION, WORKFLOW_SDK_VERSION } from '@/constants'; import { ForbiddenError } from '@/errors/response-errors/forbidden.error'; import { EventService } from '@/events/event.service'; import { InstanceAiBuilderDelegateAdapterService } from '@/modules/agents/instance-ai-builder-delegate.adapter'; import { userHasScopes } from '@/permissions.ee/check-access'; import { Publisher } from '@/scaling/pubsub/publisher.service'; import type { PubSubCommandMap } from '@/scaling/pubsub/pubsub.event-map'; import { Push } from '@/push'; import { AiService } from '@/services/ai.service'; import { InstanceWriteAccessService } from '@/services/instance-write-access.service'; import { ProxyTokenManager } from '@/services/proxy-token-manager'; import { UrlService } from '@/services/url.service'; import { Telemetry } from '@/telemetry'; import { assertNever } from '@/utils'; import { resolveAgentPreviewHandoff } from './agent-preview-handoff'; import { composeLocalMcpServers } from './browser/composite-local-mcp-server'; import { InstanceAiBrowserSessionService } from './browser/instance-ai-browser-session.service'; import { CanvasNodeContextFlagGate } from './canvas-node-context-flag-gate'; import { dropRejectedAttachmentsFromHistory } from './drop-rejected-attachments'; import { EvalThreadCredentialAllowlistService } from './eval/thread-credential-allowlist.service'; import { DurableEventLog } from './event-bus/durable-event-log'; import { InProcessEventBus } from './event-bus/in-process-event-bus'; import { InterruptedRunSweeper } from './event-bus/interrupted-run-sweeper'; import { maskCreditsForDisplay } from './instance-ai-credit-display'; import { InstanceAiCreditService } from './instance-ai-credit.service'; import { getAgentErrorSeverity, InstanceAiErrorReporterService, } from './instance-ai-error-reporter.service'; import { BROWSER_TOOL_CATEGORY, InstanceAiGatewayService } from './instance-ai-gateway.service'; import { InstanceAiMemoryService } from './instance-ai-memory.service'; import { InstanceAiModelService } from './instance-ai-model.service'; import { InstanceAiRunLimitError } from './instance-ai-run-limit.error'; import { InstanceAiRunProbe } from './instance-ai-run-probe'; import { InstanceAiSettingsService } from './instance-ai-settings.service'; import { InstanceAiTemporaryWorkflowService } from './instance-ai-temporary-workflow.service'; import { InstanceAiTerminalOutcomeService } from './instance-ai-terminal-outcome.service'; import { InstanceAiAdapterService } from './instance-ai.adapter.service'; import { AUTO_FOLLOW_UP_MESSAGE, EDITOR_CONTEXT_OPEN_TAG, EDITOR_CONTEXT_CLOSE_TAG, CREDENTIAL_CONTEXT_OPEN_TAG, CREDENTIAL_CONTEXT_CLOSE_TAG, cleanStoredUserMessage, withCurrentDateTime, withProjectContext, getProjectContextSection, } from './internal-messages'; import { INSTANCE_AI_RUN_TIMEOUT_REASON, InstanceAiLivenessService } from './liveness'; import { InstanceAiMcpRegistryService } from './mcp'; import { buildInstanceAiObservabilityContext, runMetricsModelLabel, type InstanceAiObservabilityContext, } from './observability'; import { PlannedTaskActionRunner, type PlannedBuildFollowUp, type PlannedTaskDispatcher, type PlannedTaskFollowUpStarter, type PlannedTaskRunGate, type PlannedTaskRunScope, type PlannedTaskView, type PlannedWorkflowVerificationGate, type PlannedWorkflowVerificationTracker, } from './planned-task-action-runner'; import { InstanceAiEventLogRepository } from './repositories/instance-ai-event-log.repository'; import { InstanceAiPendingConfirmationRepository } from './repositories/instance-ai-pending-confirmation.repository'; import { InstanceAiThreadGrantRepository } from './repositories/instance-ai-thread-grant.repository'; import { InstanceAiSandboxService, type RuntimeSandboxEntry } from './sandbox'; import { DbIterationLogStorage } from './storage/db-iteration-log-storage'; import { TypeORMAgentCheckpointStore } from './storage/typeorm-agent-checkpoint-store'; import { TypeORMAgentMemory } from './storage/typeorm-agent-memory'; import { isStreamTransportError } from './stream-transport-error'; import { SuspendedRunRestorer, type RebuildSuspendedRunOutcome, type ResumableOrphan, } from './suspended-run-restorer.service'; import { SuspendedThreadPersistenceService } from './suspended-thread-persistence.service'; import { InstanceAiTracingService, type MessageTraceFinalization, type OrchestratorResumeReason, } from './tracing'; import { parseWorkflowBuildOutcome, WorkflowVerificationObligationService, } from './workflow-verification-obligation-service'; import { WorkflowVerificationTaskProjector } from './workflow-verification-task-projector'; import { AgentExecutionService } from '../agents/agent-execution.service'; import { formatPreviewSessionContext } from '../agents/builder/format-preview-context'; /** A resource attachment as the trace records it: the reference, not its contents. */ type TracedResourceAttachment = { type: InstanceAiResourceAttachment['type']; id: string; projectId?: string; executionId?: string; }; /** Root-run outputs for a suspended segment — keep the LangSmith turn readable (AGENT-371). */ function buildSuspensionTraceOutputs(runId: string, suspension: SuspensionInfo | undefined) { const rawMessage = suspension?.suspendPayload.message; const message = typeof rawMessage === 'string' && rawMessage ? rawMessage : undefined; return { status: 'suspended', runId, ...(suspension?.requestId ? { requestId: suspension.requestId } : {}), ...(suspension?.toolCallId ? { pendingToolCallId: suspension.toolCallId } : {}), ...(suspension?.toolName ? { toolName: suspension.toolName } : {}), ...(message ? { message } : {}), }; } /** * Renders a message's resource attachments (e.g. a workflow + execution, or an * agent, handed off from the editor) as a context block telling the agent what * the user is looking at. Informative only: the agent should greet the user and ask how it * can help rather than inspecting the resources up front. The ids stay in the * block so they're available once the user actually asks for something. * Returns an empty string when there are none. */ /** Renders one canvas node-selection attachment as one line per set. */ function buildNodesAttachmentLine(attachment: InstanceAiNodesAttachment): string { const setLines = attachment.sets.map((set) => { const names = set.nodes.map((node) => node.name ?? node.id); const label = names.length === 1 ? `Node "${names[0]}"` : `A chain of connected nodes: ${names.join(' → ')}`; const input = set.inputNode ? `, receiving input from "${set.inputNode.name ?? set.inputNode.id}"` : ''; const output = set.outputNode ? `, sending output to "${set.outputNode.name ?? set.outputNode.id}"` : ''; const group = set.canvasGroupName ? `, part of canvas group "${set.canvasGroupName}"` : set.canvasGroupId ? `, part of canvas group \`${set.canvasGroupId}\`` : ''; return ` - ${label}${input}${output}${group}.`; }); const hasBoundary = attachment.sets.some((set) => set.inputNode ?? set.outputNode); const boundaryNote = hasBoundary ? '\n The "receiving input from"/"sending output to" nodes show only where the selection connects; they are not part of the selection. Do not describe, inspect, or make claims about them — scope your answer to the selected nodes.' : ''; return `- Selected nodes in workflow \`${attachment.workflowId}\`:\n${setLines.join('\n')}${boundaryNote}`; } export function buildContextResourcesBlock( contextAttachments: InstanceAiResourceAttachment[], ): string { if (contextAttachments.length === 0) { return ''; } const lines = contextAttachments.map((attachment) => { if (attachment.type === 'nodes') { return buildNodesAttachmentLine(attachment); } const name = attachment.name ? ` "${attachment.name}"` : ''; if (attachment.type === 'agent') { if (attachment.pending) { return `- New unsaved Agent artifact${name} (pending id: \`${attachment.id}\`, in project \`${attachment.projectId}\`).`; } return `- Agent${name} (id: \`${attachment.id}\`, in project \`${attachment.projectId}\`).`; } // Attachment type must be workflow at this point // Only mention the execution when one was actually handed off. const execution = attachment.executionId ? `, currently viewing its execution \`${attachment.executionId}\`` : ''; return `- Workflow${name} (id: \`${attachment.id}\`)${execution}.`; }); const header = contextAttachments.some((attachment) => attachment.type === 'agent') ? 'The user opened this conversation from the agent editor, where they are looking at:' : 'The user opened this conversation from the workflow editor, where they are looking at:'; const pendingAgentGuidance = contextAttachments.some( (attachment) => attachment.type === 'agent' && attachment.pending, ) ? "Treat references such as “the agent” as this pending artifact. It has no persisted agent row yet. When the user asks to build or change it, use `build-agent`'s new-agent path with a name; do not pass its pending id as an existing `agentId`. The thread's pending target will make creation reuse that id." : ''; const prose = [ header, ...lines, pendingAgentGuidance, "Treat this purely as context. Until the user tells you what they need, don't read, inspect, run, or otherwise call tools on these resources, and don't make claims about their contents — just briefly acknowledge what they're working on and ask how you can help.", ] .filter(Boolean) .join('\n'); // Wrap in EDITOR_CONTEXT_BLOCK so the UI strips it from the visible message // (cleanStoredUserMessage) and the parser can reconstruct the attachments on // reload from the leading JSON line — keeping the resource durable without // persisting it as visible text. return `${EDITOR_CONTEXT_OPEN_TAG}\n${JSON.stringify(contextAttachments)}\n\n${prose}\n${EDITOR_CONTEXT_CLOSE_TAG}`; } /** Workflow/agent attachments carry a display name; a nodes attachment doesn't. */ function isNamedResourceAttachment( attachment: InstanceAiResourceAttachment, ): attachment is InstanceAiWorkflowAttachment | InstanceAiAgentAttachment { return attachment.type !== 'nodes' && Boolean(attachment.name); } function buildHandoffContextBlock(context: InstanceAiHandoffContext | undefined): string { if (!context || context.source !== 'credential-modal') return ''; const { credential } = context; const placeholderTitles = credential.placeholderTitles ?? []; const lines = [ `- Credential type: \`${credential.credentialType}\` (${credential.displayName}).`, credential.id ? `- Existing credential id: \`${credential.id}\`.` : '', credential.nodeName ? `- Node name: "${credential.nodeName}".` : '', credential.nodeType ? `- Node type: \`${credential.nodeType}\`.` : '', placeholderTitles.length ? `- The credential form is fully pre-filled from a recipe; the user only pastes: ${placeholderTitles.map((title) => `"${title}"`).join(', ')}.` : '', credential.docsUrl ? `- The provider page where the user creates/copies the secret (verified during recipe research): ${credential.docsUrl}` : '', credential.documentationUrl ? `- n8n documentation URL: ${credential.documentationUrl}` : '', credential.oauthRedirectUrl ? `- OAuth redirect/callback URL shown in the modal: ${credential.oauthRedirectUrl}` : '', ].filter(Boolean); const prose = [ 'The user opened this conversation from the credential setup modal and is asking for setup guidance.', ...lines, 'Use this metadata only as setup context. Never ask the user to paste credential secrets into chat. For credential setup docs, load `n8n-docs-assistant` and use `n8n-docs` with `intent: "credential-setup"`.', placeholderTitles.length ? `Because the form is pre-filled, give step-by-step guidance on where to obtain the listed value(s) on the provider side${credential.docsUrl ? ' — direct the user to the provider page above rather than re-researching' : ' (research the provider if needed)'} — and do NOT suggest editing the auth template, test URL, or any other credential field.` : '', ] .filter(Boolean) .join('\n'); return `${CREDENTIAL_CONTEXT_OPEN_TAG}\n${JSON.stringify(context)}\n\n${prose}\n${CREDENTIAL_CONTEXT_CLOSE_TAG}`; } function isTelemetryConfigurableAgent( agent: unknown, ): agent is { telemetry: (telemetry: unknown) => void } { return ( typeof agent === 'object' && agent !== null && typeof Reflect.get(agent, 'telemetry') === 'function' ); } const INSTANCE_AI_CHECKPOINT_PRUNE_RETRY_MS = 30 * 1000; const WORKFLOW_SETUP_ROUTING_CLAIM_TTL_MS = 15 * 60 * 1000; const CONFIRMATION_EXPIRED_MESSAGE = 'This confirmation has expired. Send a new message to continue.'; const RESUME_REJECTED_MESSAGE = 'I could not apply that confirmation. Send a new message to continue.'; /** * Upper bound on how long `shutdown()` will wait for in-flight executeRun / * processResumedStream promises to drain after their abortControllers fire. * Sized well below n8n's `gracefulShutdownTimeoutInS` (30s default) so a * stuck agent can't burn the whole budget here. */ const INSTANCE_AI_SHUTDOWN_DRAIN_TIMEOUT_MS = 5 * 1000; function isTextMessagePart(part: unknown): part is { type: 'text'; text: string } { return ( typeof part === 'object' && part !== null && 'type' in part && part.type === 'text' && 'text' in part && typeof part.text === 'string' ); } function isSandboxEndpointNotAllowedError(error: unknown): boolean { return getErrorMessage(error).toLowerCase().includes('endpoint not allowed'); } function isStaleResumeError(error: unknown): boolean { // Name check instead of instanceof: the class crosses the @n8n/agents package boundary. return error instanceof Error && error.name === 'StaleResumeError'; } /** Signals that the failure is about an attached file rather than the request as a whole. */ const ATTACHMENT_SUBJECT_PATTERN = /image|attachment/; /** Signals that the attachment was refused for its size/decodability, not merely mentioned. */ const ATTACHMENT_REFUSAL_PATTERN = /exceed|too large|maximum|max allowed|could not process/; /** * Every scrap of message text reachable from an error, following `cause` chains, * arrays (an aggregate of stream failures) and the message-bearing fields the AI * SDK hangs the provider's response on. * * Needed because the terminal error is the ai-sdk's `AI_NoOutputGeneratedError` * wrapper: the provider's actual refusal is nested underneath it, and a plain * `error.message` check sees only "No output generated". */ function collectErrorText(value: unknown, depth = 0): string[] { if (depth > 4 || value === null || value === undefined) return []; if (typeof value === 'string') return [value]; if (Array.isArray(value)) return value.flatMap((item) => collectErrorText(item, depth + 1)); if (value instanceof Error) { return [value.message, ...collectErrorText(Reflect.get(value, 'cause'), depth + 1)]; } if (typeof value === 'object') { const record = value as Record; return ['message', 'error', 'responseBody', 'data', 'cause', 'errors'].flatMap((key) => collectErrorText(record[key], depth + 1), ); } return []; } /** * True when the provider refused an attached file outright. Both signals must be * present: matching on the subject alone would misclassify unrelated failures that * happen to mention an image. * * Matched on the message text because the provider surfaces this as a plain 400 * with no machine-readable discriminator. `validateAttachmentSizes` rejects the * known cases before the request is built, so this is the net for provider-side * limits we don't model (pixel dimensions, per-provider ceilings, format quirks) — * without it, such a rejection sticks in thread history and fails every later turn. */ export function isAttachmentRejectedByProviderError(error: unknown): boolean { const haystack = [getErrorMessage(error), ...collectErrorText(error)].join(' ').toLowerCase(); return ATTACHMENT_SUBJECT_PATTERN.test(haystack) && ATTACHMENT_REFUSAL_PATTERN.test(haystack); } /** * Shown when the user has exhausted their AI credits/quota. Self-contained so it * still reads clearly on older clients that don't render the structured * `quota_exhausted` error state; kept in sync with the FE i18n copy. */ export const QUOTA_EXHAUSTED_USER_MESSAGE = "You've run out of AI credits. Upgrade your plan to continue using the AI assistant."; const OPERATIONAL_ERROR_USER_MESSAGE = 'I hit an operational error before I could finish that response. Please try again.'; const GENERIC_ERROR_USER_MESSAGE = 'Something went wrong before I could finish that response. Please try again.'; /** Structured error code for the UI when a run failed because credits ran out. */ function getUserFacingErrorCode(error: unknown): 'quota_exhausted' | undefined { return isQuotaExhaustedError(error) ? 'quota_exhausted' : undefined; } type TerminalErrorCode = NonNullable>; /** `fallback` lets a caller name what specifically failed when the error itself * carries no user-facing meaning. */ export function getUserFacingErrorMessage( error: unknown, fallback: string = GENERIC_ERROR_USER_MESSAGE, opts: { /** * Whether the refused attachment was actually removed from thread history. * `false` means the thread still replays it, so the guidance must send the user * to a new conversation rather than promise a clean slate. */ attachmentRemoved?: boolean; } = {}, ): string { if (isQuotaExhaustedError(error)) { return QUOTA_EXHAUSTED_USER_MESSAGE; } if (isStreamTransportError(error)) { return 'The connection to the AI provider dropped before I could finish that response. Please try again.'; } if (error instanceof UserError) { return error.message; } if (isStaleResumeError(error)) { return "This approval was already handled. If the assistant isn't responding, send a new message to continue."; } if (isSandboxEndpointNotAllowedError(error)) { return "I couldn't finish preparing the workspace sandbox. Please try again in a moment."; } // Deliberately no "try again": retrying replays the same attachment. Wording stays // generic because PDFs and spreadsheets get refused too, and the size hint is only // offered when the error actually says so — the provider's text is usually // unavailable here, and guessing "too large" would misdirect the user. if (opts.attachmentRemoved !== undefined || isAttachmentRejectedByProviderError(error)) { const sizeAdvice = isAttachmentRejectedByProviderError(error) ? ` Keep it under ${formatAttachmentSizeLimit()}, and for images no more than 8000x8000 pixels.` : ''; return opts.attachmentRemoved === false ? 'I could not read one of the attached files, and I could not remove it from this ' + `conversation either. Start a new chat and attach it again.${sizeAdvice}` : 'I could not read one of the attached files, so I left it out. Attach it again — ' + `ideally a smaller version.${sizeAdvice}`; } if (error instanceof OperationalError) { return OPERATIONAL_ERROR_USER_MESSAGE; } return fallback; } /** * Masked terminal failures whose real cause is unrecoverable from the error * object: the ai-sdk's flush wrapper when zero steps were recorded (created * without `cause`), and undici's `TypeError: terminated` when the upstream * connection is killed mid-stream. When the credit wall hits at the model call * instead of the token endpoint, the run dies with one of these rather than a * classified quota error — see `reclassifyMaskedStreamFailure`. */ export function isMaskedStreamFailure(error: unknown): error is Error { if (!(error instanceof Error)) return false; if (error.name === 'AI_NoOutputGeneratedError') return true; return error.name === 'TypeError' && error.message === 'terminated'; } /** Error codes reported by the browser tool wrapper for failures that happen * before credential persistence (capturing/resolving secrets in the browser). */ const GENERATION_STAGE_ERROR_CODES = new Set([ 'missing_captured_fields', 'unresolved_field', 'gateway_context_missing', ]); function failureStageForErrorCode( errorCode: string | undefined, ): 'generation' | 'persistence' | 'unknown' { if (errorCode && GENERATION_STAGE_ERROR_CODES.has(errorCode)) return 'generation'; if (errorCode === 'credential_create_failed') return 'persistence'; return 'unknown'; } /** * Substituted for a masked stream failure once the credit re-check confirms * the user is out of credits. Carries the machine-readable quota code so the * run is classified exactly like a quota error surfaced by the proxy (credits * message + structured `errorCode`, not reported to Sentry), and keeps the * masked original as `cause` for tracing and telemetry. */ export class QuotaExhaustedStreamError extends UserError { readonly errorCode = 'quota_exhausted'; constructor(maskedError: Error) { super(`AI credits exhausted (${maskedError.name}: ${maskedError.message})`, { cause: maskedError, }); } } function createInertAbortSignal(): AbortSignal { return new AbortController().signal; } function getAbortReason(signal: AbortSignal): string { const reason = (signal as AbortSignal & { reason?: unknown }).reason; if ( typeof reason === 'object' && reason !== null && 'name' in reason && reason.name === 'AbortError' ) { return 'user_cancelled'; } if (reason instanceof Error) return reason.message; return typeof reason === 'string' ? reason : 'user_cancelled'; } /** Error details for the 'Builder generation errored' telemetry event. */ type RunFinishErrorInfo = { /** Raw error message — the SSE run-finish payload carries the user-facing reason instead. */ errorMessage?: string; /** 'stream' = the run reported an error but terminated cleanly; 'exception' = the run loop threw. */ errorSource?: 'stream' | 'exception'; }; type UnclaimedResumeContext = { threadId: string; runId: string; user: User; signal: AbortSignal; tracing?: InstanceAiTraceContext; messageGroupId?: string; unregisteredResumeTracing?: InstanceAiTraceContext; }; type UnclaimedResumeOutcome = | { kind: 'preserve-hitl' } | { kind: 'stale' } | { kind: 'cancelled' } | { kind: 'errored'; reason: string; errorCode?: TerminalErrorCode; errorMessage: string }; /** * A resume that never claimed its checkpoint gets no terminal event from the run * loop, so one has to be issued for it — except when the checkpoint is stale, * which means another owner is driving the run and may still finish it * successfully. The cancellation reason is left to the caller because reading it * consumes the run's timeout, which must not happen on the silent outcomes. */ function classifyUnclaimedResume( error: unknown, flags: { aborted: boolean; preserveHitl: boolean }, ): UnclaimedResumeOutcome { if (isStaleResumeError(error)) return { kind: 'stale' }; if (flags.aborted) return flags.preserveHitl ? { kind: 'preserve-hitl' } : { kind: 'cancelled' }; return { kind: 'errored', reason: getUserFacingErrorMessage(error, RESUME_REJECTED_MESSAGE), errorCode: getUserFacingErrorCode(error), errorMessage: getErrorMessage(error), }; } const MAX_CONCURRENT_BACKGROUND_TASKS_PER_THREAD = 5; /** Sentinel for "no cap", matching the execution concurrency limits. */ const UNLIMITED_CONCURRENCY = -1; /** * Circuit breaker for machine-started follow-up runs (verification, synthesize, * replan, …). A follow-up that dies before the agent can settle its trigger * (e.g. sandbox setup fails on an exhausted quota) would otherwise be re-armed * by its own post-run scheduler tick, producing an unbounded error loop. */ const MAX_CONSECUTIVE_FAILED_INTERNAL_FOLLOW_UPS = 3; const TITLE_REFINE_HISTORY_LIMIT = 50; /** The built orchestrator agent type returned by `createInstanceAgent`. */ type InstanceAgent = Awaited>['agent']; @Service() export class InstanceAiService { private _mcpClientManager?: McpClientManager; private readonly _ssrfProtectionConfig: SsrfProtectionConfig; private readonly _ssrfProtectionService: SsrfProtectionService; private get mcpClientManager(): McpClientManager { if (!this._mcpClientManager) { this._mcpClientManager = new McpClientManager( this._ssrfProtectionConfig.enabled ? this._ssrfProtectionService : undefined, { onToolCallSettled: (event) => this.trackMcpToolCall(event) }, ); } return this._mcpClientManager; } private readonly instanceAiConfig: InstanceAiConfig; private readonly oauth2CallbackUrl: string; private readonly webhookBaseUrl: string; private readonly formBaseUrl: string; private readonly runState = new RunStateRegistry((user) => user.id); private readonly backgroundTasks: BackgroundTaskManager; private readonly memoryTaskRegistry = new MemoryTaskRegistry(); /** Owns the LangSmith trace-context lifecycle for orchestration runs. */ private readonly tracing: InstanceAiTracingService; /** Owns the per-thread runtime sandbox/workspace lifecycle. */ private readonly sandboxService: InstanceAiSandboxService; /** Domain-access trackers per thread — persists approvals across runs within a conversation. */ private readonly domainAccessTrackersByThread = new Map(); /** Tracks the iframe pushRef per thread for live execution push events. */ private readonly threadPushRef = new Map(); /** * Runs where the credentials tool handed off to browser-assisted credential * setup (`needsBrowserSetup`), keyed by runId with one record per setup * attempt (a run can attempt several). Resolved to one terminal * success/failure telemetry event per attempt in `publishRunFinish`. * Entries exist only between the hand-off and the run's terminal event. */ private readonly pendingBrowserCredentialSetups = new Map< string, { userId: string; attempts: Array<{ credentialType: string; setupMethod: 'setup_card' | 'conversation'; attemptId?: string; startedAt: number; created: boolean; errorCode?: string; }>; } >(); /** Counts plan-review confirmations per thread, to tell the first plan apart from later revisions. */ private readonly planRequestsByThread = new Map(); /** Per-thread promise chain that serializes schedulePlannedTasks calls. */ private readonly schedulerLocks = new Map>(); /** * Consecutive machine-started follow-up runs that errored, per thread. * Gates `startInternalFollowUpRun` so a follow-up whose run keeps failing * (its trigger left unsettled) cannot re-arm itself forever; reset by any * run that completes or suspends, i.e. proves the thread is healthy again. */ private readonly failedInternalFollowUpStreaks = new Map(); /** * Checkpoint re-entries that could not fire when their parent-tagged child * settled (an orchestrator run was live, or other parent siblings were * still running). Drained from the post-run cleanup path so the checkpoint * is never left orphaned. */ private readonly pendingCheckpointReentries = new Map>(); private readonly terminalOutcome: InstanceAiTerminalOutcomeService; private readonly liveness: InstanceAiLivenessService>; /** Owns DB persistence of suspended runs + orphan-confirmation restoration. */ private readonly suspendedThreads: SuspendedThreadPersistenceService; private readonly suspendedRunRestorer: SuspendedRunRestorer; private readonly runDebugBuffer = new RunDebugBuffer(); /** Default IANA timezone for the instance (from GENERIC_TIMEZONE env var). */ private readonly defaultTimeZone: string; private readonly logger: Logger; private readonly workflowObligations: WorkflowVerificationObligationService; private readonly taskProjector: WorkflowVerificationTaskProjector; private checkpointPruneTimer: NodeJS.Timeout | undefined; private checkpointPruningStopped = true; /** * In-flight `executeRun` / `processResumedStream` promises. Tracked so * `shutdown()` can drain them before n8n closes the DB connection — the * SDK's abort-driven `cleanupRun` and `executeRun`'s `finally` block * both write to the DB during teardown, and racing the connection close * surfaces as `DriverAlreadyReleasedError` for callers. */ private readonly inFlightExecutions = new Set>(); /** * Run IDs whose post-stream terminal handling should be skipped when their * abort fires. Populated by `shutdown()` for runs that were sitting on an * inline HITL confirmation, and drained by `shouldPreserveHitlOnShutdown(runId)`. */ private readonly preserveHitlOnShutdown = new Set(); constructor( logger: Logger, globalConfig: GlobalConfig, private readonly instanceSettings: InstanceSettings, private readonly adapterService: InstanceAiAdapterService, private readonly eventBus: InProcessEventBus, private readonly eventLog: DurableEventLog, private readonly interruptedRunSweeper: InterruptedRunSweeper, private readonly settingsService: InstanceAiSettingsService, private readonly gatewayService: InstanceAiGatewayService, private readonly browserSessionService: InstanceAiBrowserSessionService, private readonly memoryService: InstanceAiMemoryService, private readonly agentMemory: TypeORMAgentMemory, private readonly checkpointStore: TypeORMAgentCheckpointStore, private readonly aiService: AiService, private readonly threadGrantRepo: InstanceAiThreadGrantRepository, private readonly pendingConfirmationRepo: InstanceAiPendingConfirmationRepository, private readonly urlService: UrlService, private readonly eventLogRepository: InstanceAiEventLogRepository, private readonly dbIterationLogStorage: DbIterationLogStorage, private readonly instanceWriteAccess: InstanceWriteAccessService, private readonly telemetry: Telemetry, private readonly mcpRegistryService: InstanceAiMcpRegistryService, private readonly userRepository: UserRepository, private readonly temporaryWorkflowService: InstanceAiTemporaryWorkflowService, private readonly errorReporter: ErrorReporter, ssrfProtectionConfig: SsrfProtectionConfig, ssrfProtectionService: SsrfProtectionService, private readonly eventService: EventService, private readonly evalCredentialAllowlists: EvalThreadCredentialAllowlistService, runProbe: InstanceAiRunProbe, private readonly modelService: InstanceAiModelService, private readonly creditService: InstanceAiCreditService, private readonly publisher: Publisher, private readonly instanceAiErrorReporter: InstanceAiErrorReporterService, private readonly canvasNodeContextFlagGate: CanvasNodeContextFlagGate, private readonly push: Push, ) { this.logger = logger.scoped('instance-ai'); runProbe.registerActiveRunCountProvider(() => this.runState.activeRunCount()); this.workflowObligations = new WorkflowVerificationObligationService(this.agentMemory); this.taskProjector = new WorkflowVerificationTaskProjector( this.agentMemory, this.eventBus, this.logger, this.workflowObligations, ); this.instanceAiConfig = globalConfig.instanceAi; this.backgroundTasks = new BackgroundTaskManager( MAX_CONCURRENT_BACKGROUND_TASKS_PER_THREAD, this.instanceAiConfig.maxConcurrentSubAgents, ); this.suspendedThreads = new SuspendedThreadPersistenceService({ logger: this.logger, config: this.instanceAiConfig, pendingConfirmationRepo: this.pendingConfirmationRepo, }); this.suspendedRunRestorer = new SuspendedRunRestorer({ logger: this.logger, pendingConfirmationRepo: this.pendingConfirmationRepo, runState: this.runState, eventBus: this.eventBus, rebuilder: { rebuildSuspendedRun: this.rebuildSuspendedRunFromCheckpoint.bind(this), resumeSuspendedRun: this.resumeSuspendedRun.bind(this), }, }); const livenessPolicyConfig = createInstanceAiLivenessPolicyConfig({ confirmationTimeoutMs: this.instanceAiConfig.confirmationTimeout, }); this.liveness = new InstanceAiLivenessService>({ policy: new InstanceAiLivenessPolicy(livenessPolicyConfig), backgroundTaskIdleTimeoutMs: livenessPolicyConfig.backgroundTaskIdleTimeoutMs, runState: this.runState, backgroundTasks: this.backgroundTasks, eventBus: this.eventBus, logger: this.logger, finalizeCancelledSuspendedRun: (suspended, reason) => { void this.finalizeCancelledSuspendedRun(suspended, reason); }, onPendingConfirmationRejected: (requestId) => { void this.suspendedThreads.dropPendingConfirmation(requestId); }, }); this.tracing = new InstanceAiTracingService({ logger: this.logger, // `first_visible_state` has to see the run's streamed text, which lives // in the log as coalesced blocks — the bus retains nothing. eventReader: { getEventsForRun: async (threadId, runId) => await this.readRunEvents(threadId, [runId]), }, runState: this.runState, eventLog: this.eventLogRepository, aiService: this.aiService, }); this.sandboxService = new InstanceAiSandboxService({ config: this.instanceAiConfig, logger: this.logger, errorReporter: this.errorReporter, runState: this.runState, backgroundTasks: this.backgroundTasks, settingsService: this.settingsService, aiService: this.aiService, }); this.terminalOutcome = new InstanceAiTerminalOutcomeService({ // The terminal guard and outcome-replay dedup must see the run's events // after a restart too, which only the durable log can provide (the bus // cache is empty in a fresh process). eventBus: { publish: (threadId, event) => this.eventBus.publish(threadId, event), getEventsForRun: async (threadId, runId) => await this.readRunEvents(threadId, [runId]), getEventsForRuns: async (threadId, runIds) => await this.readRunEvents(threadId, runIds), }, agentMemory: this.agentMemory, telemetry: this.telemetry, errorReporter: this.instanceAiErrorReporter, logger: this.logger, runState: this.runState, suspendedThreads: this.suspendedThreads, tracing: this.tracing, publishRunFinish: (threadId, runId, status, reason) => { this.publishRunFinish(threadId, runId, status, reason); }, }); this.defaultTimeZone = globalConfig.generic.timezone; const restEndpoint = globalConfig.endpoints.rest; this.oauth2CallbackUrl = `${this.urlService.getInstanceBaseUrl()}/${restEndpoint}/oauth2-credential/callback`; this.webhookBaseUrl = `${this.urlService.getWebhookBaseUrl()}${globalConfig.endpoints.webhook}`; this.formBaseUrl = `${this.urlService.getWebhookBaseUrl()}${globalConfig.endpoints.form}`; this._ssrfProtectionConfig = ssrfProtectionConfig; this._ssrfProtectionService = ssrfProtectionService; // Runtime clients capture provider settings at creation, so rebuild them // after admin settings change. In-flight sandbox users retain their entry. this.eventService.on('instance-ai-settings-updated', ({ mcpSettingsChanged }) => { this.sandboxService.invalidateCachedWorkspaces(); if (!mcpSettingsChanged) return; if (!this._mcpClientManager) return; this._mcpClientManager.disconnect().catch((error: unknown) => { this.logger.warn('Failed to disconnect MCP clients after settings change', { error: getErrorMessage(error), }); }); }); this.liveness.start(); if (this.instanceSettings.isLeader) this.startCheckpointPruning(); } private async createProxyRunConfig(user: User): Promise<{ searchProxyConfig?: ServiceProxyConfig; tracingProxyConfig?: ServiceProxyConfig; tokenManager?: ProxyTokenManager; proxyBaseUrl?: string; }> { if (!this.aiService.isProxyEnabled()) return {}; const client = await this.aiService.getClient(); const proxyBaseUrl = client.getApiProxyBaseUrl(); const tokenManager = new ProxyTokenManager(async () => { return await client.getInstanceAiApiProxyToken({ id: user.id }, { userMessageId: nanoid() }); }); const featureHeaders = buildProxyHeaders({ feature: 'instance-ai', n8nVersion: N8N_VERSION, }); return { proxyBaseUrl, tokenManager, searchProxyConfig: { apiUrl: proxyBaseUrl + '/brave-search', getAuthHeaders: async () => ({ ...(await tokenManager.getAuthHeaders()), ...featureHeaders, }), }, tracingProxyConfig: { apiUrl: proxyBaseUrl + '/langsmith', getAuthHeaders: async () => ({ ...(await tokenManager.getAuthHeaders()), ...featureHeaders, }), }, }; } /** * Full model-resolver chain shared between chat and eval paths. Delegates to * the model service so the eval endpoint gets the same working model the chat * endpoint uses. */ async resolveAgentModelConfig(user: User): Promise { return await this.modelService.resolveAgentModelConfig(user); } /** * Read the user's persisted "always allow" grants for a thread (keys like `executions:run`). * Persisted in `instance_ai_thread_grants` so they survive reload/navigation and are visible * across mains. Returns an empty set on any read error — a missing grant just re-asks, which * is safe. */ private async loadThreadSessionGrants(threadId: string, userId: string): Promise> { try { return await this.threadGrantRepo.findKeys(threadId, userId); } catch (error) { this.logger.warn('Failed to load Instance AI session grants', { threadId, error: getErrorMessage(error), }); return new Set(); } } /** * Persist a per-user, thread-level "always allow" grant. Idempotent across mains via the * composite PK. Best-effort — a failed write just means the user is re-asked next run. */ private async persistThreadSessionGrant( threadId: string, userId: string, key: string, ): Promise { try { await this.threadGrantRepo.grant(threadId, userId, key); } catch (error) { this.logger.warn('Failed to persist Instance AI session grant', { threadId, key, error: getErrorMessage(error), }); } } /** * Drop a per-user, thread-level grant. Used by decisions that are meant to be reversible * within a thread — e.g. the user skipped a credential's setup, then later asks for it. * Best-effort: a failed delete leaves the decision in place until the next run. */ private async revokeThreadSessionGrant( threadId: string, userId: string, key: string, ): Promise { try { await this.threadGrantRepo.revoke(threadId, userId, key); } catch (error) { this.logger.warn('Failed to revoke Instance AI session grant', { threadId, key, error: getErrorMessage(error), }); } } /** Whether the AI service proxy is enabled for credit counting. */ isProxyEnabled(): boolean { return this.modelService.isProxyEnabled(); } /** * Get current credit usage from the AI service proxy. * * Doubles as the reconcile point for the activation lock: this runs on every page load, so a * lock call that was lost to a failed request, an evicted record or a process restart is * re-asserted here without any scheduled job. */ async getCredits(user: User): Promise { await this.creditService.ensureQuotaLockApplied(user); const credits = await this.modelService.getCredits(user); return maskCreditsForDisplay(credits, this.settingsService.isActivationCapped()); } /** * When the credit wall hits at the model call (rather than the token * endpoint at sandbox start), the proxy failure surfaces as a masked * generic error. For those signatures, re-check the user's remaining * credits and substitute a quota-exhausted error so the run resolves to * the credits message instead of a generic failure. Best-effort: any * re-check failure keeps the original error, so genuinely unexplained * stream deaths stay visible. */ /** * Whether this run died because the provider refused an attachment. * * The terminal error is frequently the ai-sdk's masked wrapper * (`AI_NoOutputGeneratedError`), which carries none of the provider's text — the * refusal reaches us as a stream `error` event instead. Checking the run's own * events is what makes the recovery fire on the path that actually occurs; * classifying the terminal error alone silently misses it. */ /** * Whether a failed turn's attachments should be dropped from thread history. * * Deliberately *not* based on the provider's error text. That text is * unreachable by the time a run ends: the terminal error is an * `AI_NoOutputGeneratedError` wrapper whose `cause` is undefined, the agent's own * error events carry the same wrapper, and the error event holding the provider * message is not published until after the terminal handler has run. Matching on * it looks correct in unit tests and silently never fires in production. * * So key off what is knowable: this turn carried files, and it ended without * producing output. That is exactly the shape that strands a thread — the * attachment is already persisted, so every later turn replays it and dies the * same way. Dropping it on an unrelated transient failure is the acceptable * trade: the user is told, and re-attaching costs one message. Leaving it risks * a conversation that can never recover. */ /** * Translate a cleanup outcome into the `attachmentRemoved` hint the message * formatter takes. `undefined` means "say nothing about attachments" — a thread * that held none must not be told its attachment could not be removed. */ private async dropTurnAttachments(args: { threadId: string; resourceId: string; }): Promise { const outcome = await dropRejectedAttachmentsFromHistory( this.agentMemory, { threadId: args.threadId, resourceId: args.resourceId }, this.logger, ); if (outcome === 'no-attachments') return undefined; return outcome === 'removed'; } private shouldDropTurnAttachments(args: { /** * Whether this turn introduced files. Unknown on a resumed run — it replays * history rather than accepting new input — so callers there pass `true` and * rely on `dropRejectedAttachmentsFromHistory` no-opping when the thread holds * no inline files. */ turnHadAttachments: boolean; producedNoOutput: boolean; }): boolean { return args.turnHadAttachments && args.producedNoOutput; } private async reclassifyMaskedStreamFailure( error: unknown, user: User, context: { threadId: string; runId: string }, ): Promise { if (!isMaskedStreamFailure(error)) return error; try { const { creditsQuota, creditsClaimed, quotaLocked } = await this.modelService.getCredits(user); // The activation lock refuses use while the quota still has credits left, so the numbers // alone wouldn't explain the failure. Read from the proxy, not from n8n's own trigger // state: that only says the lock *should* apply, so a lock call that failed would turn // any unrelated stream death into a spurious upgrade wall. if (!quotaLocked && (creditsQuota < 0 || creditsClaimed < creditsQuota)) return error; } catch (creditsError) { this.logger.debug('Masked stream failure credit re-check failed; keeping original error', { error: getErrorMessage(creditsError), ...context, }); return error; } this.logger.info('Reclassified masked stream failure as quota-exhausted', { maskedError: getErrorMessage(error), ...context, }); return new QuotaExhaustedStreamError(error); } isEnabled(): boolean { return this.settingsService.isAgentEnabled() && !!this.instanceAiConfig.model; } hasActiveRun(threadId: string): boolean { return this.runState.hasLiveRun(threadId); } /** * Whether this specific run is live (active or suspended) in this process. * The interrupted-run sweeper uses this so a newer run on the same thread * never shields an older crashed run from being swept. */ isRunLive(threadId: string, runId: string): boolean { return ( this.runState.getActiveRunId(threadId) === runId || this.runState.getSuspendedRun(threadId)?.runId === runId ); } getThreadStatus(threadId: string): InstanceAiThreadStatusResponse { const status = this.runState.getThreadStatus( threadId, this.backgroundTasks.getTaskSnapshots(threadId), ); const memoryTasks = this.memoryTaskRegistry.getTasks(threadId); return { ...status, memoryTasks }; } private memoryTaskObserverFor( threadId: string, tracing: InstanceAiTraceContext | undefined, ): (event: ScopedMemoryTaskEvent) => void { return (event) => { // Retains/releases the trace's telemetry provider lease so the // task's LLM span can still export after the root trace finalizes // (see InstanceAiTraceContext.onMemoryTaskEvent). tracing?.onMemoryTaskEvent?.(event); this.memoryTaskRegistry.handleEvent(threadId, event); const pendingTasks = this.memoryTaskRegistry.getTasks(threadId); const logContext = { threadId, taskId: event.task.id, taskKind: event.task.taskKind, pendingCount: pendingTasks.length, ...(event.type === 'skipped' ? { reason: event.reason } : {}), ...(event.type === 'failed' ? { error: getErrorMessage(event.error) } : {}), ...(event.type === 'completed' && event.value && typeof event.value === 'object' && 'status' in event.value && typeof event.value.status === 'string' ? { outcome: event.value.status } : {}), }; this.logger.info(`Observational memory task ${event.type}`, logContext); }; } /** * Surface the agent's background/best-effort failures to Sentry. The SDK emits * `AgentEvent.Error` for these but nothing consumed it, so they were lost: memory * observer/reflector/episodic indexing and the eager input / turn-on-suspend * persists all fail silently. Optional memory processing is warning-level; * persistence stays error-level because it can lose turn data. We report only * the `source`-tagged events — the main agentic-loop errors (no source) already * reach Sentry via the errored run/stream result. */ private subscribeToAgentErrors(agent: InstanceAgent, threadId: string, runId: string): void { agent.on(AgentEvent.Error, (event: AgentEventData) => { if (event.type !== AgentEvent.Error) return; if (!event.source) return; const severity = getAgentErrorSeverity(event.source); this.instanceAiErrorReporter.report(event.error, { component: `instance-ai-${event.source}`, ...(severity ? { severity } : {}), threadId, runId, }); }); } isRunDebugEnabled(): boolean { return this.instanceAiConfig.runDebugEnabled; } private buildOrchestratorAgentStreamOptions( user: User, threadId: string, runId: string, signal: AbortSignal, ): Record { if (this.isRunDebugEnabled()) { this.runDebugBuffer.ensure(runId, threadId); } return { maxIterations: MAX_STEPS.ORCHESTRATOR, abortSignal: signal, // Recover token usage from raw provider events so a stopped/errored run // is still billed for the tokens consumed before the stop. recoverUsageOnAbort: true, persistence: { resourceId: user.id, threadId, // Host run id, persisted with checkpoints so the interrupted-run // sweep can match a crashed run's checkpoint exactly. hostRunId: runId, }, providerOptions: { anthropic: { cacheControl: { type: 'ephemeral' } }, }, ...(this.isRunDebugEnabled() ? createRunDebugStepHooks(this.runDebugBuffer, { runId, threadId }) : {}), }; } private buildOrchestratorResumeAgentOptions( user: User, threadId: string, runId: string, agentRunId: string, toolCallId: string, signal: AbortSignal, ): Record { if (this.isRunDebugEnabled()) { this.runDebugBuffer.ensure(runId, threadId); } return { runId: agentRunId, toolCallId, abortSignal: signal, // Keep billing stopped/errored resumed runs (see stream-options builder). recoverUsageOnAbort: true, persistence: { resourceId: user.id, threadId, hostRunId: runId }, // Must mirror buildOrchestratorAgentStreamOptions: without this request-level // cache directive, resumed (HITL) turns send no cache_control, so Anthropic // reprocesses the whole conversation uncached on every resume (~100K tokens). providerOptions: { anthropic: { cacheControl: { type: 'ephemeral' } }, }, ...(this.isRunDebugEnabled() ? createRunDebugStepHooks(this.runDebugBuffer, { runId, threadId }) : {}), }; } getRunDebug(runId: string) { return this.runDebugBuffer.get(runId); } listThreadDebugRuns(threadId: string) { return this.runDebugBuffer.listByThread(threadId).map((record: RunDebugRecord) => ({ runId: record.runId, threadId: record.threadId, startedAt: record.startedAt, stepCount: record.steps.length, workflowCodeCount: record.workflowCode.length, label: record.label, })); } clearTraceContextsForTest(): void { this.tracing.clearTraceContextsForTest(); } async submitLangsmithFeedback( user: User, threadId: string, responseId: string, payload: { rating: 'up' | 'down'; comment?: string }, ): Promise { await this.tracing.submitLangsmithFeedback(user, threadId, responseId, payload); } /** * Refuse a new user turn when a concurrency cap is already full. * * Only new turns are gated. Resumes and internal follow-up runs are always admitted: a * refused resume strands a conversation mid-confirmation, and a refused follow-up trips * the consecutive-failure breaker and abandons the thread's planned-task graph for good. * The ceiling is therefore soft by design -- it bounds how much new work starts, not how * much can be in flight. * * Counts are per process. That is the right scope for the instance cap, because the * pressure it relieves is per process; for the per-user cap it means a multi-main * deployment allows the cap once per main. * * The instance cap bounds concurrent execution, not resident memory: a suspended run * releases its slot but keeps its agent in memory. Counting those here would be worse * than the gap, because a few abandoned approval cards would then wall the whole * instance for the confirmation timeout. To close it, run state must leave memory. * * Either cap is disabled by setting it to `-1` (unlimited), matching * `N8N_CONCURRENCY_PRODUCTION_LIMIT`. The config schema rejects `0`, so it can't reach * here and be mistaken for either reading. */ private assertRunAdmissible(user: User): void { const { maxConcurrentRuns, maxConcurrentRunsPerUser } = this.instanceAiConfig; // Per-user first: it is the more specific diagnosis, and the only one the user can // act on themselves. if (maxConcurrentRunsPerUser !== UNLIMITED_CONCURRENCY) { const running = this.runState.activeRunCountForUser(user.id); if (running >= maxConcurrentRunsPerUser) { this.logger.warn('Refused Instance AI run: per-user concurrency limit reached', { userId: user.id, running, limit: maxConcurrentRunsPerUser, }); this.eventService.emit('instance-ai-run-refused', { reason: 'user_run_limit' }); throw new InstanceAiRunLimitError( `You already have ${running} conversations running. Wait for one to finish, or stop it, before starting another.`, { reason: 'user_run_limit', limit: maxConcurrentRunsPerUser }, ); } } if (maxConcurrentRuns !== UNLIMITED_CONCURRENCY) { const running = this.runState.activeRunCount(); if (running >= maxConcurrentRuns) { this.logger.warn('Refused Instance AI run: instance concurrency limit reached', { userId: user.id, running, limit: maxConcurrentRuns, }); this.eventService.emit('instance-ai-run-refused', { reason: 'instance_run_limit' }); throw new InstanceAiRunLimitError( 'This n8n instance is already running the maximum number of assistant conversations. Try again in a moment.', { reason: 'instance_run_limit', limit: maxConcurrentRuns }, ); } } } startRun( user: User, threadId: string, message: string, attachments?: InstanceAiAttachment[], context?: InstanceAiHandoffContext, timeZone?: string, pushRef?: string, ): string { this.assertRunAdmissible(user); this.liveness.clearThreadState(threadId); const { runId, abortController, messageGroupId } = this.runState.startRun({ threadId, user, }); // Persist the user's time zone so checkpoint / replan / synthesize // follow-up runs can reinject it into the system prompt // instead of falling back to GENERIC_TIMEZONE. if (timeZone) { this.runState.setTimeZone(threadId, timeZone); } if (pushRef !== undefined) { this.threadPushRef.set(threadId, pushRef); } this.startExecuteRun( user, threadId, runId, message, abortController, attachments, context, messageGroupId, timeZone, ); return runId; } /** Get the current messageGroupId for a thread (used by SSE sync). */ getMessageGroupId(threadId: string): string | undefined { return this.runState.getMessageGroupId(threadId); } /** * Get the messageGroupId for the thread's live activity. * Prefers the active/suspended run's group, then falls back to the * most recent running background task's group (which was captured * at spawn time and may differ from the thread's current group * if the user started a new turn). */ getLiveMessageGroupId(threadId: string): string | undefined { return this.runState.getLiveMessageGroupId( threadId, this.backgroundTasks.getTaskSnapshots(threadId), ); } /** Get all runIds belonging to a messageGroupId. */ getRunIdsForMessageGroup(messageGroupId: string): string[] { return this.runState.getRunIdsForMessageGroup(messageGroupId); } /** Get the active runId for a thread. */ getActiveRunId(threadId: string): string | undefined { return this.runState.getActiveRunId(threadId); } cancelRun(threadId: string, reason = 'user_cancelled'): void { const cancelledTasks = this.backgroundTasks.cancelThread(threadId); const user = this.runState.getThreadUser(threadId); for (const task of cancelledTasks) { void this.tracing.finalizeBackgroundTaskTracing(task, 'cancelled'); this.eventBus.publish(threadId, { type: 'agent-completed', runId: task.runId, agentId: task.agentId, payload: { role: task.role, result: '', status: 'cancelled', }, }); void this.terminalOutcome.recordBackgroundTerminalOutcome(task); if (user) { void this.handlePlannedTaskSettlement(user, task, 'cancelled', { reschedule: false }); } } // Clean up any awaiting_approval plan graph for this thread. The user // cancelled before approving, so leaving the graph persisted would (a) // cause doSchedulePlannedTasks() to republish the stale checklist on // every later pass via syncPlannedTasksToUi(). Only target awaiting_approval — active and // awaiting_replan graphs have their own settlement logic via the // background-task cancellations above. void this.cancelAwaitingApprovalPlan(threadId); const { active, suspended } = this.runState.cancelThread(threadId); if (active) { if (reason === INSTANCE_AI_RUN_TIMEOUT_REASON) this.liveness.markRunTimedOut(active.runId); active.abortController.abort(); // inline-kind rows are dropped via the resolve-callback fired by // runState.cancelThread; suspended-kind rows (no in-memory // resolver) get cleaned up here so the index never outlives the run. void this.suspendedThreads.dropPendingConfirmationsForThread(threadId); return; } if (suspended) { if (reason === INSTANCE_AI_RUN_TIMEOUT_REASON) this.liveness.markRunTimedOut(suspended.runId); suspended.abortController.abort(); void this.finalizeCancelledSuspendedRun(suspended, reason); } void this.suspendedThreads.dropPendingConfirmationsForThread(threadId); } /** Send a correction message to a running background task. */ sendCorrectionToTask( threadId: string, taskId: string, correction: string, ): 'queued' | 'task-completed' | 'task-not-found' { return this.backgroundTasks.queueCorrection(threadId, taskId, correction); } /** Cancel a single background task by ID. */ cancelBackgroundTask(threadId: string, taskId: string): void { const task = this.backgroundTasks.cancelTask(threadId, taskId); if (!task) return; void this.tracing.finalizeBackgroundTaskTracing(task, 'cancelled'); this.eventBus.publish(threadId, { type: 'agent-completed', runId: task.runId, agentId: task.agentId, payload: { role: task.role, result: '', status: 'cancelled' }, }); void this.terminalOutcome.recordBackgroundTerminalOutcome(task); const user = this.runState.getThreadUser(threadId); if (user) { void this.handlePlannedTaskSettlement(user, task, 'cancelled'); } } // ── Cross-main task-control routing ─────────────────────────────────────── // User actions (correct/cancel/clear) can land on a different main than the // one running the task/run. Each `route*` method applies the action locally // and, when the target isn't local (or is thread-wide), broadcasts so the // owning main applies it. Broadcast + local-gate — no shared ownership store. // `applyTaskControlLocally` is the single action → local-method mapping, // shared by the route entry points and the `@OnPubSubEvent` relay handler // (which never re-broadcasts, so there's no loop). These wrappers are the // controller entry points; internal callers keep using the local methods // directly so they don't trigger cross-main broadcasts. private broadcastTaskControl(payload: PubSubCommandMap['relay-instance-ai-task-control']): void { if (!this.instanceSettings.isMultiMain) return; void this.publisher .publishCommand({ command: 'relay-instance-ai-task-control', payload }) .catch((error: unknown) => this.logger.error('Failed to relay Instance AI task-control to sibling mains', { threadId: payload.threadId, action: payload.action, error, }), ); } /** Apply a task-control action to this main's local slice of the thread. * Returns whether the action's target was found locally: task-scoped * actions report a local hit to gate re-broadcast, thread-wide actions * always report a miss so they fan out to every main. */ private async applyTaskControlLocally({ threadId, taskId, action, correction, }: PubSubCommandMap['relay-instance-ai-task-control']): Promise { switch (action) { case 'correct': // A relay without its correction text is malformed: nothing to apply. if (!taskId || correction === undefined) return true; return this.sendCorrectionToTask(threadId, taskId, correction) !== 'task-not-found'; case 'cancel-task': { if (!taskId) return true; const isLocal = this.backgroundTasks .getTaskSnapshots(threadId) .some((task) => task.taskId === taskId); this.cancelBackgroundTask(threadId, taskId); return isLocal; } case 'cancel-thread': // A thread's run + tasks can be spread across mains, so always fan out. this.cancelRun(threadId); return false; case 'clear-thread': await this.clearThreadState(threadId); return false; } } private async routeTaskControl( payload: PubSubCommandMap['relay-instance-ai-task-control'], ): Promise { const foundLocally = await this.applyTaskControlLocally(payload); if (!foundLocally) this.broadcastTaskControl(payload); } async routeCorrectionToTask(threadId: string, taskId: string, correction: string): Promise { await this.routeTaskControl({ threadId, taskId, action: 'correct', correction }); } async routeCancelBackgroundTask(threadId: string, taskId: string): Promise { await this.routeTaskControl({ threadId, taskId, action: 'cancel-task' }); } async routeCancelRun(threadId: string): Promise { const hadLiveLocalRun = this.runState.hasLiveRun(threadId); await this.routeTaskControl({ threadId, action: 'cancel-thread' }); // Emit-shaped fallback for dead runs: a cancel that found nothing live // locally still terminalizes crashed runs in the durable log, so every // client and main converges through the normal replay path — a dead run // has no run body left to emit its own run-finish. A run live in THIS // process is excluded (its abort above emits the terminal fact), and a // run live on a sibling is excluded by the sweeper's durable-activity // grace window while the broadcast cancel reaches it. if (!hadLiveLocalRun) { // Settle the drain first so a run that JUST finished cannot be // misread as unfinished and given a second, later terminal fact. await this.eventLog.flush(threadId); await this.interruptedRunSweeper.cancelUnfinishedRuns(threadId); } } async routeClearThreadState(threadId: string): Promise { await this.routeTaskControl({ threadId, action: 'clear-thread' }); } /** Apply a task-control action relayed from another main to this main's local * slice of the thread. Never re-broadcasts. Not self-sent, so this never * fires on the originating main. */ @OnPubSubEvent('relay-instance-ai-task-control', { instanceType: 'main' }) async handleRelayTaskControl( payload: PubSubCommandMap['relay-instance-ai-task-control'], ): Promise { // Guard the whole handler: it runs as a fire-and-forget pubsub listener, so // a throw (e.g. from the async clearThreadState) would surface as an // unhandled rejection on this sibling main instead of being contained. try { await this.applyTaskControlLocally(payload); } catch (error) { this.logger.error('Failed to apply relayed Instance AI task-control', { threadId: payload.threadId, taskId: payload.taskId, action: payload.action, error, }); } } /** Cancel all background tasks across all threads. Test-only. */ cancelAllBackgroundTasks(): number { const cancelled = this.backgroundTasks.cancelAll(); for (const task of cancelled) { void this.tracing.finalizeBackgroundTaskTracing(task, 'cancelled'); } return cancelled.length; } async startStuckBackgroundTaskForTest( user: User, threadId: string, ): Promise<{ threadId: string; runId: string; messageGroupId: string; taskId: string; agentId: string; timeoutAt: number; }> { const messageId = `msg_${nanoid()}`; const messageText = 'I started a background workflow-builder task.'; const { runId, messageGroupId } = this.runState.startRun({ threadId, user }); if (!messageGroupId) { throw new UnexpectedError('Failed to create message group for timeout simulation'); } const taskId = `task_${nanoid()}`; const agentId = `agent_${nanoid()}`; this.eventBus.publish(threadId, { type: 'run-start', runId, agentId: orchestratorAgentId(runId), userId: user.id, payload: { messageId, messageGroupId }, }); this.eventBus.publish(threadId, { type: 'text-delta', runId, agentId: orchestratorAgentId(runId), responseId: `test-background-start:${runId}`, payload: { text: messageText }, }); this.eventBus.publish(threadId, { type: 'agent-spawned', runId, agentId, payload: { parentId: orchestratorAgentId(runId), role: 'workflow-builder', tools: [], taskId, kind: 'builder', title: 'Building workflow', subtitle: 'Timeout simulation', goal: 'Simulate a stuck background task timeout', }, }); // Persist the assistant message so the UI can render it after navigation. await this.agentMemory.saveMessages({ threadId, resourceId: user.id, messages: [ { id: messageId, createdAt: new Date(), type: 'llm', role: 'assistant', content: [{ type: 'text', text: messageText }], }, ], }); const outcome = this.backgroundTasks.spawn({ taskId, threadId, runId, role: 'workflow-builder', agentId, messageGroupId, run: async (signal) => await new Promise((resolve) => { signal.addEventListener('abort', () => resolve('aborted'), { once: true }); }), onFailed: (task) => { this.eventBus.publish(threadId, { type: 'agent-completed', runId, agentId, payload: { role: task.role, result: '', error: task.error ?? 'Unknown error', }, }); }, onSettled: async (task) => { await this.terminalOutcome.recordBackgroundTerminalOutcome(task); }, }); if (outcome.status !== 'started') { throw new UnexpectedError('Failed to start stuck background task simulation'); } this.runState.clearActiveRun(threadId); this.eventBus.publish(threadId, { type: 'run-finish', runId, agentId: orchestratorAgentId(runId), userId: user.id, payload: { status: 'completed' }, }); return { threadId, runId, messageGroupId, taskId, agentId, timeoutAt: outcome.task.lastActivityAt + this.liveness.backgroundTaskIdleTimeoutMs + 1, }; } async runLivenessSweepForTest(now?: number): Promise { await this.liveness.sweepTimedOutWork(now); } // ── Gateway lifecycle (delegated to LocalGatewayRegistry) ─────────────── // ── Test-only trace replay API ─────────────────────────────────────────── loadTraceEvents(slug: string, events: unknown[]): void { this.tracing.loadTraceEvents(slug, events); } getTraceEvents(slug: string): unknown[] { return this.tracing.getTraceEvents(slug); } hasRunningWorkForTest(): boolean { const threadIds = new Set(this.tracing.getTrackedThreadIds()); for (const threadId of threadIds) { if (this.runState.getActiveRunId(threadId)) return true; if (this.backgroundTasks.getRunningTasks(threadId).length > 0) return true; } return false; } activateTraceSlug(slug: string): void { this.tracing.activateTraceSlug(slug); } clearTraceEvents(slug: string): void { this.tracing.clearTraceEvents(slug); } /** * Remove all in-memory state associated with a thread. * Must be called when a thread is deleted so the maps don't leak. */ async clearThreadState(threadId: string): Promise { this.liveness.clearThreadState(threadId); // Clear run-state registry entries (active/suspended runs, confirmations, // user, time zone, and message-group mappings). const { active, suspended } = this.runState.clearThread(threadId); if (active) { active.abortController.abort(); await this.tracing.finalizeRunTracing(active.runId, active.tracing, { status: 'cancelled', reason: 'thread_cleared', }); } if (suspended) { suspended.abortController.abort(); await this.tracing.finalizeRunTracing(suspended.runId, suspended.tracing, { status: 'cancelled', reason: 'thread_cleared', }); } // Cancel background tasks belonging to this thread for (const task of this.backgroundTasks.cancelThread(threadId)) { task.abortController.abort(); await this.tracing.finalizeBackgroundTaskTracing(task, 'cancelled'); } await this.tracing.finalizeRemainingMessageTraceRoots(threadId, { status: 'cancelled', reason: 'thread_cleared', metadata: { completion_source: 'service_cleanup' }, }); this.schedulerLocks.delete(threadId); this.failedInternalFollowUpStreaks.delete(threadId); this.domainAccessTrackersByThread.delete(threadId); this.evalCredentialAllowlists.clearThread(threadId); this.threadPushRef.delete(threadId); this.planRequestsByThread.delete(threadId); this.memoryTaskRegistry.clearThread(threadId); this.tracing.deleteTraceContextsForThread(threadId); await this.deleteAgentBuilderSessions(threadId); await this.sandboxService.destroySandbox(threadId); await this.temporaryWorkflowService.reapForThreadCleanup(threadId); await this.suspendedThreads.dropPendingConfirmationsForThread(threadId); this.eventBus.clearThread(threadId); } /** Builder sub-agent sessions (`ia-builder::*`) live in the agents * module's memory tables; instance-AI storage cleanup does not cover them. * Best-effort: a failure here must never block thread deletion. */ private async deleteAgentBuilderSessions(threadId: string): Promise { if (!Container.get(ModuleRegistry).isActive('agents')) return; try { await Container.get(InstanceAiBuilderDelegateAdapterService).deleteBuilderSessions(threadId); } catch (error) { this.logger.warn('Failed to clean up agent-builder sessions for thread', { threadId, error: error instanceof Error ? error.message : String(error), }); } } async shutdown(): Promise { this.stopCheckpointPruning(); this.liveness.shutdown(); const { activeRuns, suspendedRuns, pendingThreadIds } = this.runState.shutdown(); const threadsWithPendingHitl = new Set(pendingThreadIds); for (const run of activeRuns) { // Runs holding an inline HITL confirmation (`create-tasks`, // sub-agent `ask-user`) sit in `activeRuns` because the orchestrator // is alive — it's just awaiting the in-process Promise. Their // `instance_ai_pending_confirmations` row survives the restart and // `handleOrphanedConfirmation` will issue the user-visible // `restart_lost_confirmation` UserError + `run-finish` when (if) // the user clicks confirm. If we publish run-finish here, the fold // would render the plan/ask card as cancelled before the user has a // chance to see it on reload. if (threadsWithPendingHitl.has(run.threadId)) { await this.tracing.finalizeRunTracing(run.runId, run.tracing, { status: 'cancelled', reason: 'service_shutdown', }); // Record the policy *before* the abort fires so the run's catch // handler (which runs synchronously off the abort) sees the // flag. The catch path consults `shouldPreserveHitlOnShutdown` // and skips the terminal-fallback / run-finish writes that // would otherwise overwrite the plan/ask card. this.preserveHitlOnShutdown.add(run.runId); run.abortController.abort(); continue; } // Truly mid-stream run: the durable run-finish is all history needs — // the fold derives the terminal tree from the log after restart. this.publishRunFinish(run.threadId, run.runId, 'cancelled', 'service_shutdown'); await this.tracing.finalizeRunTracing(run.runId, run.tracing, { status: 'cancelled', reason: 'service_shutdown', }); run.abortController.abort(); } for (const run of suspendedRuns) { // Suspended runs are recoverable from the checkpoint store + pending // confirmation index, so leave the run-finish unpublished and the // snapshot untouched. We only need to abort the in-process stream; // the DB rows are intentionally preserved across restart. await this.tracing.finalizeRunTracing(run.runId, run.tracing, { status: 'cancelled', reason: 'service_shutdown', }); run.abortController.abort(); } for (const task of this.backgroundTasks.cancelAll()) { task.abortController.abort(); await this.tracing.finalizeBackgroundTaskTracing(task, 'cancelled'); } // Drain the now-aborted executeRun / processResumedStream promises // before returning. Each one's finally block does DB work // (`schedulePlannedTasks`, `dropPendingConfirmationsForThread`) and // the SDK's abort-driven `cleanupRun` issues `checkpointStore.delete`, // all of which would race the connection close in `exitSuccessFully` // and surface as `DriverAlreadyReleasedError` otherwise. Bounded so // a hung agent can't block n8n's own graceful-shutdown deadline. await this.drainInFlightExecutions(INSTANCE_AI_SHUTDOWN_DRAIN_TIMEOUT_MS); this.instanceAiErrorReporter.endAllRuns(); const threadsWithTraces = new Set(this.tracing.getTrackedThreadIds()); for (const threadId of threadsWithTraces) { await this.tracing.finalizeRemainingMessageTraceRoots(threadId, { status: 'cancelled', reason: 'service_shutdown', metadata: { completion_source: 'service_cleanup' }, }); } this.gatewayService.disconnectAll(); await this.browserSessionService.shutdown(); this.sandboxService.stopSandboxExpiryTimers(); // Thread-scoped sandboxes survive service shutdown so a restarted process // can reuse them. Explicit thread cleanup and idle TTL remain the // teardown paths. this.domainAccessTrackersByThread.clear(); this.tracing.clear(); // Flush in-flight drains + open coalesce buffers so the tail of every // streamed segment survives the restart. await this.eventLog.flushAll(); this.eventBus.clear(); await this._mcpClientManager?.disconnect(); // Final drain of every trace's LangSmith provider so spans still sitting // in the batch exporter (e.g. from late memory tasks) are not lost on // shutdown. Best-effort by contract — never throws. await shutdownProductTelemetryProviders(); this.logger.debug('Instance AI service shut down'); } @OnLeaderTakeover() startCheckpointPruning(): void { if (this.checkpointPruneTimer || this.instanceAiConfig.pruneInterval <= 0) return; this.checkpointPruningStopped = false; this.scheduleCheckpointPrune(0); } @OnLeaderStepdown() stopCheckpointPruning(): void { this.checkpointPruningStopped = true; clearTimeout(this.checkpointPruneTimer); this.checkpointPruneTimer = undefined; } private scheduleCheckpointPrune(delayMs = this.instanceAiConfig.pruneInterval): void { if (this.checkpointPruningStopped) return; this.checkpointPruneTimer = setTimeout(() => { void this.runScheduledPrune(); }, delayMs); this.checkpointPruneTimer.unref(); } /** * Track a fire-and-forget run so `shutdown()` can wait for its cleanup * (finally block + SDK `cleanupRun`) to finish before the DB closes. * The promise removes itself from the set on settle so the set doesn't * grow unbounded across long-running threads. * * Prefer `startExecuteRun` / `startProcessResumedStream` over calling this * directly — they make tracking unmissable by binding the spawn and the * tracking together in one method. */ private trackInFlightExecution(promise: Promise): void { const tracked = promise.finally(() => { this.inFlightExecutions.delete(tracked); }); this.inFlightExecutions.add(tracked); } /** * Single spawn point for the executor — guarantees shutdown can drain the * run before closing the DB. New call sites should always go through this * wrapper rather than invoking `executeRun` directly. */ private startExecuteRun(...args: Parameters): void { this.trackInFlightExecution(this.executeRun(...args)); } /** Same shutdown-drain contract as `startExecuteRun`, for the resume path. */ private startProcessResumedStream( ...args: Parameters ): void { this.trackInFlightExecution(this.processResumedStream(...args)); } /** * True when `shutdown()` aborted this run with the explicit policy that its * inline-HITL snapshot should be left intact. Used by `executeRun` and * `processResumedStream` to short-circuit the cancelled-path terminal * finalisation (run-finish event + cancelled tree snapshot) that would * otherwise overwrite the plan/ask card the user expects to see on reload. */ private shouldPreserveHitlOnShutdown(runId: string): boolean { return this.preserveHitlOnShutdown.has(runId); } private async drainInFlightExecutions(timeoutMs: number): Promise { if (this.inFlightExecutions.size === 0) return; const drain = Promise.allSettled([...this.inFlightExecutions]); const timeout = new Promise<'timeout'>((resolve) => { setTimeout(() => resolve('timeout'), timeoutMs).unref(); }); const outcome = await Promise.race([drain.then(() => 'drained' as const), timeout]); if (outcome === 'timeout') { this.logger.warn('Timed out waiting for in-flight Instance AI runs to drain', { timeoutMs, stillInFlight: this.inFlightExecutions.size, }); } } /** * One tick of the recurring leader prune cycle: expire stale checkpoints, * hard-delete tombstones past the GC horizon, drop expired pending * confirmations, and delete expired conversation threads, then schedule the * next run. A checkpoint failure reschedules with a shorter retry delay; the * confirmation and thread steps swallow their own errors so they never * disrupt the cycle. */ private async runScheduledPrune(now = Date.now()): Promise { const olderThan = new Date(now - this.instanceAiConfig.snapshotRetention); try { const count = await this.checkpointStore.markExpiredOlderThan(olderThan); if (count > 0) { this.logger.info('Expired stale Instance AI checkpoints', { count }); } else { this.logger.debug('No stale Instance AI checkpoints to expire'); } await this.hardDeleteExpiredCheckpoints(now); await this.suspendedThreads.pruneStalePendingConfirmations(now); await this.pruneExpiredThreads(); this.scheduleCheckpointPrune(); } catch (error: unknown) { this.logger.warn('Failed to expire stale Instance AI checkpoints', { error: getErrorMessage(error), }); this.scheduleCheckpointPrune(INSTANCE_AI_CHECKPOINT_PRUNE_RETRY_MS); } } /** * Hard-delete expired checkpoint tombstones (`markExpiredOlderThan` releases * the blob but keeps the row so a stale resume gets a clear "expired" error; * this drops the row entirely once it's past the GC horizon so tombstones * don't grow unbounded). Has its own try/catch so a GC failure never forces * the checkpoint prune into its short retry cadence. No-op when * `checkpointGcRetention` is 0. */ private async hardDeleteExpiredCheckpoints(now: number): Promise { const retention = this.instanceAiConfig.checkpointGcRetention; if (retention <= 0) return; try { const olderThan = new Date(now - retention); const count = await this.checkpointStore.hardDeleteExpiredOlderThan(olderThan); if (count > 0) { this.logger.info('Hard-deleted expired Instance AI checkpoint tombstones', { count }); } } catch (error: unknown) { this.logger.warn('Failed to hard-delete expired Instance AI checkpoint tombstones', { error: getErrorMessage(error), }); } } /** * Delete conversation threads older than the configured TTL as part of the * recurring leader prune. Has its own try/catch so a failure here never * disrupts checkpoint pruning or the next scheduled run. No-op when * `threadTtlDays` is 0 (handled inside `cleanupExpiredThreads`). */ private async pruneExpiredThreads(): Promise { try { await this.memoryService.cleanupExpiredThreads( async (threadId) => await this.clearThreadState(threadId), ); } catch (error: unknown) { this.logger.warn('Failed to clean up expired Instance AI conversation threads', { error: getErrorMessage(error), }); } } private createAgentMemoryOptions(user: User, threadId: string, runId: string) { return { observationalMemory: { observerThresholdTokens: this.instanceAiConfig.observerMessageTokens, reflectorThresholdTokens: this.instanceAiConfig.reflectorObservationTokens, // Observer/reflector calls run in the background outside the run's // finish-chunk usage, so they are claimed here per report. Best-effort: // a billing failure must never block observation persistence. onTaskUsage: async (report: MemoryTaskUsageReport) => { try { const items = tokenUsageToBuilderUsageItems(report.model, report.usage); if (items.length === 0) return; await this.creditService.claimRunUsage( user, threadId, `${runId}:memory:${report.task}:${report.reportId}`, items, 'completed', ); } catch (error) { this.logger.warn('Failed to claim observational-memory usage', { threadId, runId, task: report.task, error: getErrorMessage(error), }); } }, }, }; } private createWorkflowTaskServiceWithUiSync( threadId: string, runId: string, workflowTasks: WorkflowTaskCoordinator, ): WorkflowTaskService { const sync = async () => await this.taskProjector.syncFromWorkflowLoop(threadId, runId); return { reportBuildOutcome: async (outcome) => { const action = await workflowTasks.reportBuildOutcome(outcome); await sync(); return action; }, reportVerificationVerdict: async (verdict) => { const action = await workflowTasks.reportVerificationVerdict(verdict); await sync(); return action; }, updateBuildOutcome: async (workItemId, update) => { await workflowTasks.updateBuildOutcome(workItemId, update); await sync(); }, getBuildOutcome: async (workItemId) => await workflowTasks.getBuildOutcome(workItemId), getLatestBuildOutcomeForWorkflow: async (workflowId) => await workflowTasks.getLatestBuildOutcomeForWorkflow(workflowId), getWorkflowLoopState: async (workItemId) => await workflowTasks.getWorkflowLoopState(workItemId), }; } private trackWorkflowVerificationObligation( obligation: WorkflowVerificationObligation, event: string, extra: Record = {}, ): void { try { this.telemetry?.track('instance_ai_workflow_verification_obligation', { event, thread_id: obligation.threadId, run_id: obligation.runId, task_id: obligation.taskId, planned_task_id: obligation.plannedTaskId, work_item_id: obligation.workItemId, workflow_id: obligation.workflowId, source: obligation.source, policy: obligation.policy, status: obligation.status, readiness_status: obligation.readiness?.status, setup_status: obligation.setupRequirement?.status, has_evidence: obligation.evidence?.attempted === true, evidence_success: obligation.evidence?.success, blocking_reason: obligation.blockingReason, ...extra, }); } catch (error) { this.logger.warn('Failed to track workflow verification obligation telemetry', { threadId: obligation.threadId, workItemId: obligation.workItemId, error: getErrorMessage(error), }); } } private buildPlannedTaskFollowUpMessage( type: 'synthesize' | 'replan' | 'checkpoint' | 'build-workflow', graph: PlannedTaskGraph, options: { failedTask?: PlannedTaskRecord; checkpoint?: PlannedTaskRecord; buildTask?: PlannedTaskRecord; } = {}, ): string { const payload: Record = { tasks: graph.tasks.map((task) => ({ id: task.id, title: task.title, kind: task.kind, status: task.status, result: task.result, error: task.error, outcome: task.outcome, })), }; if (options.failedTask) { payload.failedTask = { id: options.failedTask.id, title: options.failedTask.title, kind: options.failedTask.kind, error: options.failedTask.error, result: options.failedTask.result, }; } if (options.checkpoint) { const depOutcomes = graph.tasks .filter((t) => options.checkpoint!.deps.includes(t.id)) .map((t) => ({ id: t.id, title: t.title, kind: t.kind, status: t.status, result: t.result, outcome: t.outcome, })); payload.checkpoint = { id: options.checkpoint.id, title: options.checkpoint.title, instructions: options.checkpoint.spec, dependsOn: depOutcomes, }; } if (options.buildTask) { payload.buildTask = { id: options.buildTask.id, title: options.buildTask.title, kind: options.buildTask.kind, spec: options.buildTask.spec, workflowId: options.buildTask.workflowId, isSupportingWorkflow: options.buildTask.isSupportingWorkflow, deps: options.buildTask.deps, }; } return `\n${JSON.stringify(payload, null, 2)}\n\n\n${AUTO_FOLLOW_UP_MESSAGE}`; } private buildWorkflowVerificationFollowUpMessage(input: { obligation: WorkflowVerificationObligation; outcome?: WorkflowBuildOutcome; sourceTask?: Pick< ManagedBackgroundTask, 'taskId' | 'role' | 'status' | 'result' | 'error' | 'plannedTaskId' | 'workItemId' >; }): string { const payload = { obligation: input.obligation, outcome: input.outcome, sourceTask: input.sourceTask, }; return `\n${JSON.stringify(payload, null, 2)}\n\n\n${AUTO_FOLLOW_UP_MESSAGE}`; } private async createPlannedTaskState() { const memory = this.agentMemory; const taskStorage = new ThreadTaskStorage(memory); const plannedTaskStorage = new PlannedTaskStorage(memory); const plannedTaskService = new PlannedTaskCoordinator(plannedTaskStorage); return { memory, taskStorage, plannedTaskService }; } /** * Replays any undelivered background-task outcomes for a thread so a * reconnecting client never misses a result that completed while its stream * was closed. Delegates to {@link InstanceAiTerminalOutcomeService}. */ async replayUndeliveredTerminalOutcomes(threadId: string): Promise { await this.terminalOutcome.replayUndeliveredTerminalOutcomes(threadId); } private async syncPlannedTasksToUi(threadId: string, graph: PlannedTaskGraph): Promise { const { taskStorage } = await this.createPlannedTaskState(); const tasks = await this.taskProjector.projectPlannedTaskList(threadId, graph); await taskStorage.save(threadId, tasks); this.eventBus.publish(threadId, { type: 'tasks-update', runId: graph.planRunId, agentId: orchestratorAgentId(graph.planRunId), payload: { tasks }, }); } /** * Drop any persisted planned-task graph that is still `awaiting_approval`, * and clear the UI checklist. Called on run cancellation and HITL timeout so * stale approval state doesn't linger. A graph in `active` / `awaiting_replan` * is already in-flight and has its own settlement logic. */ private async cancelAwaitingApprovalPlan(threadId: string): Promise { try { const { plannedTaskService, taskStorage } = await this.createPlannedTaskState(); const graph = await plannedTaskService.getGraph(threadId); if (!graph || graph.status !== 'awaiting_approval') return; await plannedTaskService.clear(threadId); await taskStorage.save(threadId, { tasks: [] }); this.eventBus.publish(threadId, { type: 'tasks-update', runId: graph.planRunId, agentId: orchestratorAgentId(graph.planRunId), payload: { tasks: { tasks: [] }, planItems: [] }, }); } catch (error) { this.logger.warn('Failed to clean up awaiting_approval plan on cancel', { threadId, error: error instanceof Error ? error.message : String(error), }); } } private async createExecutionEnvironment( user: User, threadId: string, runId: string, abortSignal: AbortSignal, messageGroupId?: string, pushRef?: string, proxyRunConfig?: Awaited>, ) { const memory = this.agentMemory; const boundProjectId = await memory.getThreadProjectId(threadId); if (!boundProjectId) { throw new UnexpectedError( `Instance AI thread "${threadId}" has no bound project; it must be created via POST /instance-ai/threads before a run can start`, ); } const adminSettings = await this.settingsService.getAdminSettings(); const localGatewayDisabledGlobally = adminSettings.localGatewayDisabled; const browserUseEnabledGlobally = adminSettings.browserUseEnabled; const localGatewayDisabledForUser = await this.settingsService.isLocalGatewayDisabledForUser( user.id, ); const userGateway = this.gatewayService.findGateway(user.id); // There's another ensure lock check at `getCredits`, which only fires when the frontend mounts. await this.creditService.ensureQuotaLockApplied(user); const { searchProxyConfig, tracingProxyConfig, tokenManager, proxyBaseUrl } = proxyRunConfig ?? (await this.createProxyRunConfig(user)); const proxyContext = { runId, threadId }; const modelId = proxyBaseUrl && tokenManager ? await this.modelService.resolveProxyModel(user, proxyBaseUrl, tokenManager, proxyContext) : await this.modelService.resolveAgentModelConfig(user, proxyContext); const configEvalsEnabled = await this.adapterService.isConfigEvalsEnabled(user); const mcpConnectionsEnabled = await this.adapterService.isMcpConnectionsEnabled(user); const nodeUsageEnabled = await this.adapterService.isNodeUsageEnabled(user); const context = this.adapterService.createContext(user, { searchProxyConfig, pushRef, threadId, projectId: boundProjectId, credentialIdAllowlist: this.evalCredentialAllowlists.get(threadId), shouldBypassCredentialTest: (credentialId: string) => this.evalCredentialAllowlists.shouldBypassTest(threadId, credentialId), configEvalsEnabled, mcpConnectionsEnabled, nodeUsageEnabled, modelId, }); // Merge both local gateway and direct browser-use into a single // composite server, since context has a single `localMcpServer`. // Perhaps a better solution would be to have multiple local MCP // servers? But that requires more changes where `localMcpServer` // is currently used // When Browser Use is disabled instance-wide, hide the gateway's browser tools // too so they are neither advertised to nor callable by the agent. this.gatewayService.applyToolPolicy(user.id); const gatewayMcpServer = !localGatewayDisabledForUser && userGateway?.isConnected ? userGateway : undefined; const browserMcpServer = browserUseEnabledGlobally ? this.browserSessionService.findMcpServer(user.id) : undefined; const localMcpServer = composeLocalMcpServers(gatewayMcpServer, browserMcpServer); if (localMcpServer) { context.localMcpServer = localMcpServer; } context.permissions = this.settingsService.getPermissions(); if (this.instanceWriteAccess.isReadOnly()) { context.permissions = applyBranchReadOnlyOverrides(context.permissions); context.branchReadOnly = true; } context.runId = runId; context.browserCredentialSetup = this.createBrowserCredentialSetupTracker(runId, user.id); // Per-user, thread-level "always allow" grants are persisted in the DB so they survive // reload/navigation and are visible across mains. Load once per run; a tool resuming // from a `scope: 'session'` approval persists new grants via `grantSessionToolApproval`. // Keep the mutable set so a grant approved mid-run is honored by later calls in the same // run — the next run reloads it from the DB anyway. const sessionGrants = await this.loadThreadSessionGrants(threadId, user.id); context.sessionApprovedToolKeys = sessionGrants; const grantSessionToolApproval = async (key: string) => { await this.persistThreadSessionGrant(threadId, user.id, key); sessionGrants.add(key); }; context.grantSessionToolApproval = grantSessionToolApproval; context.revokeSessionToolApproval = async (key: string) => { await this.revokeThreadSessionGrant(threadId, user.id, key); sessionGrants.delete(key); }; // Domain-access approvals are stored as grant keys in `instance_ai_thread_grants` (via // the same load/persist path as above), so they survive restart and are visible // cross-main. Recreate the tracker per run seeded from the freshly loaded grants; // transient (allow_once) approvals are run-scoped and don't carry across runs. const domainTracker = createDomainAccessTracker({ grantedKeys: sessionGrants, persistGrant: grantSessionToolApproval, }); this.domainAccessTrackersByThread.set(threadId, domainTracker); context.domainAccessTracker = domainTracker; if (this.isRunDebugEnabled()) { context.recordWorkflowCodeSnapshot = (snapshot) => { this.runDebugBuffer.ensure(runId, threadId); this.runDebugBuffer.recordWorkflowCode(runId, snapshot); }; } browserMcpServer?.setDomainGate({ tracker: domainTracker, runId, permissionMode: context.permissions?.fetchUrl, createCredentialPermissionMode: context.permissions?.createCredential, }); // Compute gateway status for the system prompt. The direct browser // session contributes a `browser` capability even without the daemon. if (gatewayMcpServer || browserMcpServer) { const capabilities = new Set(); if (gatewayMcpServer) { // getStatus() already drops excluded categories (e.g. browser when disabled). for (const { name, enabled } of gatewayMcpServer.getStatus().toolCategories) { if (enabled) { capabilities.add(name); } } } if (browserMcpServer) { capabilities.add(BROWSER_TOOL_CATEGORY); } context.localGatewayStatus = { status: 'connected', capabilities: [...capabilities], }; } else if (localGatewayDisabledGlobally && !browserUseEnabledGlobally) { context.localGatewayStatus = { status: 'disabledGlobally' }; } else { context.localGatewayStatus = { status: localGatewayDisabledForUser ? 'disabled' : 'disconnected', }; } const taskStorage = new ThreadTaskStorage(memory); const iterationLog = this.dbIterationLogStorage; const workflowLoopStorage = new WorkflowLoopStorage(memory); const workflowTasks = this.createWorkflowTaskServiceWithUiSync( threadId, runId, new WorkflowTaskCoordinator(threadId, workflowLoopStorage), ); const plannedTaskStorage = new PlannedTaskStorage(memory); const plannedTaskService = new PlannedTaskCoordinator(plannedTaskStorage); const nodeDefDirs = this.adapterService.getNodeDefinitionDirs(); if (nodeDefDirs.length > 0) { setSchemaBaseDirs(nodeDefDirs); } // Per-user skill gate: hide flag-gated skills (filtered copy, cache // preserved) so every derived skill source inherits the exclusion. const flagDisabledSkillIds = disabledInstanceAiSkillIds({ configEvalsEnabled }); const allRuntimeSkills = flagDisabledSkillIds.length > 0 ? filterRuntimeSkillSource(loadInstanceAiRuntimeSkillSource(), flagDisabledSkillIds) : loadInstanceAiRuntimeSkillSource(); let runtimeSkills = allRuntimeSkills; let runtimeWorkspace: Workspace | undefined; let workspaceRoot: string | undefined; const sandboxStatus = this.settingsService.getSandboxStatus(); if (sandboxStatus.workflowBuilderAvailable) { const sandboxConfig = await this.instanceAiErrorReporter.withBoundary( 'instance-ai-sandbox-setup', { threadId, runId, userId: user.id, messageGroupId }, async () => await this.sandboxService.resolveSandboxConfig(user), ); if (sandboxConfig.enabled) { workspaceRoot = getPromptWorkspaceRoot(sandboxConfig.provider); let sandboxEntryPromise: Promise | undefined; const getSandboxEntry = async () => { sandboxEntryPromise ??= this.sandboxService .getOrCreateWorkspaceEntry(threadId, user) .catch((error: unknown) => { sandboxEntryPromise = undefined; throw error; }); return await sandboxEntryPromise; }; const getSetupSandboxEntry = async () => { return await this.sandboxService.getOrCreateWorkspace(threadId, user, context); }; const scopeWorkspaceForAgent = async ( workspace: Workspace | undefined, ): Promise => { if (!workspace) return undefined; const root = await getWorkspaceRoot(workspace); return createScopedWorkspace(workspace, root); }; runtimeWorkspace = createLazyRuntimeWorkspace({ // Empty + stable across resumes: sandbox/filesystem guidance lives in // the system prompt's `## Sandbox workspace` section. Passing '' here // (instead of omitting) keeps the lazy workspace from falling back to // resolution-dependent live `getInstructions()` text, which would // shift the cached prompt prefix across rebuilds/resumes. sandboxInstructions: '', filesystemInstructions: '', ensureWorkspace: async () => await scopeWorkspaceForAgent((await getSetupSandboxEntry())?.workspace), }); const runtimeSkillWorkspace = createLazyRuntimeWorkspace({ id: 'instance-ai-runtime-skill-workspace', name: 'Instance AI runtime skill workspace', ensureWorkspace: async () => await scopeWorkspaceForAgent((await getSandboxEntry())?.workspace), }); runtimeSkills = createLazyWorkspaceRuntimeSkillSource({ source: allRuntimeSkills, workspace: runtimeSkillWorkspace, logger: this.logger, }); } } context.workspace = runtimeWorkspace; context.workspaceRoot = workspaceRoot; context.threadId = threadId; context.threadMemory = memory; // Tool-emitted telemetry is an open-ended property bag (search queries, // remediation reasons, node error strings) — scrub at the boundary so a // new call site can't leak by omission. context.trackTelemetry = (eventName, properties) => { this.telemetry.track(eventName, redactTelemetryProperties(properties)); }; const domainTools = createAllTools(context); const orchestrationContext: OrchestrationContext = { threadId, runId, messageGroupId, userId: user.id, projectId: boundProjectId, orchestratorAgentId: orchestratorAgentId(runId), modelId, checkpointStore: this.checkpointStore, eventBus: this.eventBus, logger: this.logger, trackTelemetry: (eventName, properties) => { this.telemetry.track(eventName, redactTelemetryProperties(properties)); }, // Aggregate on the instance-AI thread (not the `ia-builder:` session // thread) so the credit service's per-thread display total and FE push // attribute builder tokens to the conversation the user sees. The tool // awaits this before returning/cascading a terminal segment outcome, so // a billing failure must never propagate — best-effort by contract. claimSubAgentUsage: async (dedupeId, usage, status) => { try { await this.creditService.claimRunUsage(user, threadId, dedupeId, usage, status); } catch (error) { // claimRunUsage() handles ordinary claim failures (network, retries) // internally and only throws for exceptional contract violations // (e.g. a negative quota) — those must still reach centralized // Instance AI error reporting even though billing stays best-effort. this.instanceAiErrorReporter.report(error, { component: 'instance-ai-agent-builder-usage', threadId, runId, userId: user.id, ...(boundProjectId ? { projectId: boundProjectId } : {}), ...(messageGroupId ? { messageGroupId } : {}), }); this.logger.warn('Failed to claim agent-builder usage', { threadId, runId, dedupeId, error: getErrorMessage(error), }); } }, domainTools, abortSignal, taskStorage, timeZone: this.defaultTimeZone, localMcpServer: context.localMcpServer, runtimeSkills, runtimeSkillCatalog: allRuntimeSkills, oauth2CallbackUrl: this.oauth2CallbackUrl, webhookBaseUrl: this.webhookBaseUrl, formBaseUrl: this.formBaseUrl, waitForConfirmation: async (requestId: string) => { this.runState.touchActiveRun(threadId); return await new Promise((resolve) => { this.runState.registerPendingConfirmation(requestId, { resolve, threadId, userId: user.id, createdAt: Date.now(), }); void this.suspendedThreads.persistPendingConfirmation({ requestId, threadId, userId: user.id, runId, messageGroupId, kind: 'inline', }); }); }, cancelBackgroundTask: async (taskId) => this.cancelBackgroundTask(threadId, taskId), spawnBackgroundTask: (opts) => this.spawnBackgroundTask(runId, opts, messageGroupId), touchRun: () => this.runState.touchActiveRun(threadId), touchBackgroundTask: (taskId) => this.backgroundTasks.touchTask(threadId, taskId), plannedTaskService, schedulePlannedTasks: async () => await this.schedulePlannedTasks(user, threadId), iterationLog, sendCorrectionToTask: (taskId, correction) => this.sendCorrectionToTask(threadId, taskId, correction), workflowTaskService: workflowTasks, workspace: runtimeWorkspace, workspaceRoot, nodeDefinitionDirs: nodeDefDirs.length > 0 ? nodeDefDirs : undefined, domainContext: context, tracingProxyConfig, memory, }; return { context, memory, taskStorage, iterationLog, workflowTasks, plannedTaskService, modelId, orchestrationContext, }; } /** * Resolve the agents execution service only when the `agents` module is * active. Mirrors the builder-delegate gate so agent-preview handoff is * absent (rather than exploding) when agents are disabled. */ private getAgentExecutionService(): AgentExecutionService | null { if (!Container.get(ModuleRegistry).isActive('agents')) return null; try { return Container.get(AgentExecutionService); } catch { return null; } } /** * Hydrate the thread-persisted preview-session reference (if any) and wire * the on-demand transcript resolver. Must run before createInstanceAgent so * createOrchestrationTools can register get-session on follow-up turns. */ private async bindAgentPreviewSession( context: Awaited>['context'], user: User, ): Promise { await resolveAgentPreviewSession(context); const projectId = context.projectId; if (!context.agentPreviewSession || !projectId) return; if (!(await this.canAccessAgentPreviewHandoff(user, projectId))) { context.agentPreviewSession = undefined; return; } context.resolvePreviewSession = async (ref) => { const service = this.getAgentExecutionService(); if (!service) return null; const detail = await service.getThreadDetail(ref.threadId, projectId, ref.agentId); if (!detail) return null; const transcript = formatPreviewSessionContext( detail.thread, detail.executions, ref.executionId, ); if (transcript === null) return null; return { title: detail.thread.title?.trim() || `Session #${detail.thread.sessionNumber}`, sessionNumber: detail.thread.sessionNumber, transcript, }; }; } private async dispatchPlannedTask( task: PlannedTaskRecord, context: OrchestrationContext, _graph?: PlannedTaskGraph, ): Promise { if (task.kind === 'build-workflow' || task.kind === 'checkpoint') { this.logger.warn('dispatchPlannedTask called for a runtime planned-task kind', { threadId: context.threadId, taskId: task.id, kind: task.kind, }); return; } await context.plannedTaskService?.markFailed(context.threadId, task.id, { error: `Planned task kind "${task.kind}" is no longer supported`, }); const nextGraph = await context.plannedTaskService?.getGraph(context.threadId); if (nextGraph) { await this.syncPlannedTasksToUi(context.threadId, nextGraph); } } private collectWorkflowIds(value: unknown, workflowIds: Set): void { if (value === null || value === undefined || typeof value !== 'object') return; if (Array.isArray(value)) { for (const item of value) { this.collectWorkflowIds(item, workflowIds); } return; } for (const [key, child] of Object.entries(value as Record)) { if (key === 'workflowId' && typeof child === 'string' && child.length > 0) { workflowIds.add(child); continue; } if (key === 'supportingWorkflowIds' && Array.isArray(child)) { for (const workflowId of child) { if (typeof workflowId === 'string' && workflowId.length > 0) { workflowIds.add(workflowId); } } continue; } this.collectWorkflowIds(child, workflowIds); } } private getBuildTaskWorkflowName(task: PlannedTaskRecord): string | undefined { if (task.kind !== 'build-workflow') return undefined; const titleMatch = task.title.match(/^Build '(.+)' workflow$/) ?? task.title.match(/^Build "(.+)" workflow$/); return titleMatch?.[1]; } /** * Keep explicit user-requested runs approval-gated when they happen as checkpoint fallback. */ private checkpointRequiresRunApproval( graph: PlannedTaskGraph, _checkpoint: PlannedTaskRecord, ): boolean { return graph.postBuildRunApprovalRequired === true; } /** * Resolve the workflows the checkpoint task is verifying so the runWorkflow * permission override can be scoped. Workflow names are carried as an E2E replay * fallback because runtime workflow IDs can be remapped. */ private async getCheckpointRunPolicy( threadId: string, checkpointTaskId: string, ): Promise<{ allowedWorkflowIds: ReadonlySet; allowedWorkflowNames: ReadonlySet; requireApproval: boolean; }> { try { const { plannedTaskService } = await this.createPlannedTaskState(); const graph = await plannedTaskService.getGraph(threadId); const checkpoint = graph?.tasks.find((t) => t.id === checkpointTaskId); if (!graph || !checkpoint) { return { allowedWorkflowIds: new Set(), allowedWorkflowNames: new Set(), requireApproval: false, }; } const deps = new Set(checkpoint.deps); const ids = new Set(); const names = new Set(); for (const task of graph.tasks) { if (!deps.has(task.id)) continue; const workflowName = this.getBuildTaskWorkflowName(task); if (workflowName) { names.add(workflowName); } if (task.workflowId) { ids.add(task.workflowId); } this.collectWorkflowIds(task.outcome, ids); } const tracing = this.tracing.getTraceContextForContinuation(threadId); for (const workflowId of [...ids]) { const remappedWorkflowId = tracing?.idRemapper?.remapOutput(workflowId); if (typeof remappedWorkflowId === 'string' && remappedWorkflowId.length > 0) { ids.add(remappedWorkflowId); } } return { allowedWorkflowIds: ids, allowedWorkflowNames: names, requireApproval: this.checkpointRequiresRunApproval(graph, checkpoint), }; } catch (error) { this.logger.warn('Failed to resolve checkpoint allowed workflow IDs', { threadId, checkpointTaskId, error: error instanceof Error ? error.message : String(error), }); return { allowedWorkflowIds: new Set(), allowedWorkflowNames: new Set(), requireApproval: false, }; } } private async handlePlannedTaskSettlement( user: User, task: ManagedBackgroundTask, status: 'succeeded' | 'failed' | 'cancelled', { reschedule = true }: { reschedule?: boolean } = {}, ): Promise { if (!task.plannedTaskId) return; const { plannedTaskService } = await this.createPlannedTaskState(); let graph: PlannedTaskGraph | null = null; if (status === 'succeeded') { graph = await plannedTaskService.markSucceeded(task.threadId, task.plannedTaskId, { result: task.result, outcome: task.outcome, }); } else if (status === 'failed') { graph = await plannedTaskService.markFailed(task.threadId, task.plannedTaskId, { error: task.error, }); } else { graph = await plannedTaskService.markCancelled(task.threadId, task.plannedTaskId, { error: task.error, }); } if (graph) { await this.syncPlannedTasksToUi(task.threadId, graph); } // Callers that tore the whole thread down (user stop, run timeout, // shutdown) opt out: the scheduler must not start a follow-up run. The // graph stays intact so the next user message can pick it back up. if (!reschedule) return; await this.schedulePlannedTasks(user, task.threadId); } private async maybeStartWorkflowVerificationFollowUp( user: User, task: ManagedBackgroundTask, ): Promise { if (task.role !== 'workflow-builder' || !task.workItemId) return false; const obligation = await this.workflowObligations.getObligation( task.threadId, task.workItemId, { source: task.plannedTaskId ? 'planned' : 'direct', plannedTaskId: task.plannedTaskId, }, ); if (!obligation) return false; this.trackWorkflowVerificationObligation(obligation, 'background_task_settled'); // Only run a verification follow-up when there is something to verify. // Setup (mocked credentials, unresolved placeholders) is handled separately // and deterministically by `maybeStartWorkflowSetupFollowUp`. if (obligation.status !== 'ready_to_verify' && obligation.status !== 'verifying') { return false; } const outcome = parseWorkflowBuildOutcome(task.outcome); const startedRunId = await this.startInternalFollowUpRun( user, task.threadId, this.buildWorkflowVerificationFollowUpMessage({ obligation, outcome, sourceTask: { taskId: task.taskId, role: task.role, status: task.status, result: task.result, error: task.error, plannedTaskId: task.plannedTaskId, workItemId: task.workItemId, }, }), task.messageGroupId, false, undefined, 'workflow_verification', ); this.trackWorkflowVerificationObligation(obligation, 'follow_up_start_attempted', { follow_up_started: startedRunId.length > 0, }); return startedRunId.length > 0; } private buildWorkflowSetupFollowUpMessage(obligation: WorkflowVerificationObligation): string { const payload = { workflowId: obligation.workflowId, workItemId: obligation.workItemId, setupRequirement: obligation.setupRequirement, verificationReadiness: obligation.readiness, }; return `\n${JSON.stringify(payload, null, 2)}\n\n\n${AUTO_FOLLOW_UP_MESSAGE}`; } private getWorkflowSetupSuspensionWorkflowId( toolName: string | undefined, suspendPayload: Record | undefined, ): string | undefined { if (toolName !== 'workflows' || !suspendPayload) return undefined; if (!Array.isArray(suspendPayload.setupRequests)) return undefined; return typeof suspendPayload.workflowId === 'string' ? suspendPayload.workflowId : undefined; } private async markWorkflowSetupHandled( threadId: string, workflowId: string, runId?: string, ): Promise { const records = await this.listWorkflowLoopRecords(threadId); if (records.length === 0) return false; const candidates: Array<{ record: WorkflowLoopWorkItemRecord; obligation: WorkflowVerificationObligation; }> = []; for (const record of records) { if (record.state.setupRoutedAt) continue; if (this.workflowObligations.isPlannedRecord(record)) continue; const obligation = this.workflowObligations.obligationFromRecord(threadId, record, { source: 'direct', }); if (obligation.workflowId !== workflowId) continue; if (obligation.setupRequirement?.status !== 'required') continue; candidates.push({ record, obligation }); } const sameRunCandidates = runId ? candidates.filter( ({ obligation, record }) => obligation.runId === runId || record.state.runId === runId, ) : []; const fallbackCandidates = candidates.filter( (candidate) => !sameRunCandidates.includes(candidate), ); for (const { record, obligation } of [...sameRunCandidates, ...fallbackCandidates]) { const claim = await this.claimWorkItemSetupRouting(threadId, record); if (!claim) continue; const marked = await this.markWorkItemSetupRouted( threadId, record.state.workItemId, claim.claimId, ); if (!marked) { await this.releaseWorkItemSetupRoutingClaim( threadId, record.state.workItemId, claim.claimId, ); this.logger.warn('Workflow setup completed but routing marker was not saved', { threadId, workItemId: record.state.workItemId, workflowId, }); continue; } this.trackWorkflowVerificationObligation(obligation, 'setup_completed_by_tool'); return true; } return false; } /** * Deterministically route a settled direct build to setup when its saved * workflow still needs real credentials or values. Runs after verification so * setup does not depend on the orchestrator choosing to call it. The persisted * claim and `setupRoutedAt` marker make this fire at most once per build, so * concurrent finalization re-entries cannot start duplicate setup runs. */ private async maybeStartWorkflowSetupFollowUp(user: User, threadId: string): Promise { const records = await this.listWorkflowLoopRecords(threadId); if (records.length === 0) return false; for (const record of records) { if (record.state.setupRoutedAt) continue; if (this.workflowObligations.isPlannedRecord(record)) continue; const obligation = this.workflowObligations.obligationFromRecord(threadId, record, { source: 'direct', }); const verificationConcluded = obligation.status === 'verified' || obligation.status === 'needs_setup' || obligation.status === 'not_verifiable'; if (!verificationConcluded) continue; if (obligation.setupRequirement?.status !== 'required' || !obligation.workflowId) continue; const claim = await this.claimWorkItemSetupRouting(threadId, record); if (!claim) continue; const startedRunId = await this.startInternalFollowUpRun( user, threadId, this.buildWorkflowSetupFollowUpMessage(obligation), this.runState.getMessageGroupId(threadId), false, undefined, 'workflow_setup', ); if (startedRunId.length === 0) { await this.releaseWorkItemSetupRoutingClaim( threadId, record.state.workItemId, claim.claimId, ); return false; } const marked = await this.markWorkItemSetupRouted( threadId, record.state.workItemId, claim.claimId, ); if (!marked) { this.logger.warn('Workflow setup follow-up started but routing marker was not saved', { threadId, workItemId: record.state.workItemId, }); } this.trackWorkflowVerificationObligation(obligation, 'setup_follow_up_started'); return true; } return false; } private async listWorkflowLoopRecords(threadId: string): Promise { return await new WorkflowLoopStorage(this.agentMemory).listWorkItems(threadId); } private createWorkflowSetupRoutingClaim(): WorkflowSetupRoutingClaim { const claimedAt = new Date(); const expiresAt = new Date(claimedAt.getTime() + WORKFLOW_SETUP_ROUTING_CLAIM_TTL_MS); return { claimId: `setup:${nanoid()}`, claimedAt: claimedAt.toISOString(), expiresAt: expiresAt.toISOString(), }; } private async claimWorkItemSetupRouting( threadId: string, record: WorkflowLoopWorkItemRecord, ): Promise { const claim = this.createWorkflowSetupRoutingClaim(); const claimed = await new WorkflowLoopStorage(this.agentMemory).claimSetupRouting( threadId, record.state.workItemId, claim, ); return claimed ? claim : null; } private async markWorkItemSetupRouted( threadId: string, workItemId: string, claimId: string, ): Promise { return await new WorkflowLoopStorage(this.agentMemory).markSetupRouted( threadId, workItemId, claimId, new Date().toISOString(), ); } private async releaseWorkItemSetupRoutingClaim( threadId: string, workItemId: string, claimId: string, ): Promise { await new WorkflowLoopStorage(this.agentMemory).releaseSetupRoutingClaim( threadId, workItemId, claimId, ); } /** * Feed the follow-up circuit breaker from a run's terminal status. Errored * machine-started follow-ups (`isInternalFollowUp`) extend the streak; any * run that completes or suspends proves the thread executes again and * resets it. Cancelled runs carry no signal either way. */ private updateInternalFollowUpFailureStreak( threadId: string, status: MessageTraceFinalization['status'] | undefined, isInternalFollowUp: boolean, ): void { if (status === 'completed' || status === 'suspended') { this.failedInternalFollowUpStreaks.delete(threadId); return; } if (status === 'error' && isInternalFollowUp) { this.failedInternalFollowUpStreaks.set( threadId, (this.failedInternalFollowUpStreaks.get(threadId) ?? 0) + 1, ); } } private async startInternalFollowUpRun( user: User, threadId: string, message: string, messageGroupId?: string, isReplanFollowUp: boolean = false, checkpoint?: { isCheckpointFollowUp: true; checkpointTaskId: string }, resumeReasonOverride?: OrchestratorResumeReason, plannedBuild?: PlannedBuildFollowUp, ): Promise { if (this.runState.hasLiveRun(threadId)) { this.logger.warn('Skipping internal follow-up: active run exists', { threadId }); return ''; } const failedStreak = this.failedInternalFollowUpStreaks.get(threadId) ?? 0; if (failedStreak >= MAX_CONSECUTIVE_FAILED_INTERNAL_FOLLOW_UPS) { this.logger.warn('Skipping internal follow-up: consecutive follow-up runs keep failing', { threadId, failedStreak, resumeReason: resumeReasonOverride, }); return ''; } const { runId, abortController } = this.runState.startRun({ threadId, user, messageGroupId, }); // Resolve user time zone from the thread's run-state snapshot (captured on the // initial user-facing run) before falling back to the instance default. Follow-up // runs (checkpoint / replan / synthesize) used to drop this context, which made // user-local schedules fall back to "instance default timezone". const timeZone = this.runState.getTimeZone(threadId) ?? this.defaultTimeZone; const resumeReason: OrchestratorResumeReason = resumeReasonOverride ?? (checkpoint ? 'planned_checkpoint' : isReplanFollowUp ? 'replan' : 'background_task_completed'); this.startExecuteRun( user, threadId, runId, message, abortController, undefined, undefined, messageGroupId, timeZone, isReplanFollowUp, checkpoint, resumeReason, plannedBuild, ); return runId; } private async schedulePlannedTasks(user: User, threadId: string): Promise { const prev = this.schedulerLocks.get(threadId) ?? Promise.resolve(); // eslint-disable-next-line @typescript-eslint/promise-function-async const current = prev.then(() => this.doSchedulePlannedTasks(user, threadId)).catch(() => {}); this.schedulerLocks.set(threadId, current); await current; } private createPlannedTaskActionRunner( activeUser: User, threadId: string, plannedTaskService: PlannedTaskService, ): PlannedTaskActionRunner { const scope: PlannedTaskRunScope = { user: activeUser, threadId }; return new PlannedTaskActionRunner({ scope, plannedTaskService, logger: this.logger, view: this.createPlannedTaskView(), runGate: this.createPlannedTaskRunGate(), dispatcher: this.createPlannedTaskDispatcher(), followUps: this.createPlannedTaskFollowUps(), workflowVerificationGate: this.createPlannedWorkflowVerificationGate(threadId), workflowVerificationTracker: this.createPlannedWorkflowVerificationTracker(), }); } private createPlannedTaskView(): PlannedTaskView { return { sync: async (scope, graph) => await this.syncPlannedTasksToUi(scope.threadId, graph), }; } private createPlannedTaskRunGate(): PlannedTaskRunGate { return { hasLiveRun: (threadId) => this.runState.hasLiveRun(threadId), }; } private createPlannedTaskDispatcher(): PlannedTaskDispatcher { return { dispatch: async ({ scope, graph, tasks }) => { const context = await this.createPlannedTaskDispatchContext( scope.user, scope.threadId, graph, ); for (const task of tasks) { await this.dispatchPlannedTask(task, context, graph); } }, }; } private createPlannedTaskFollowUps(): PlannedTaskFollowUpStarter { return { startReplan: async ({ scope, graph, failedTask }) => await this.startInternalFollowUpRun( scope.user, scope.threadId, this.buildPlannedTaskFollowUpMessage('replan', graph, { failedTask }), graph.messageGroupId, true, undefined, undefined, undefined, ), startWorkflowVerification: async ({ scope, graph, verification }) => await this.startInternalFollowUpRun( scope.user, scope.threadId, this.buildWorkflowVerificationFollowUpMessage({ obligation: verification.obligation, outcome: verification.outcome, sourceTask: this.toWorkflowVerificationSourceTask(verification), }), graph.messageGroupId, false, undefined, 'workflow_verification', undefined, ), startSynthesis: async ({ scope, graph }) => await this.startInternalFollowUpRun( scope.user, scope.threadId, this.buildPlannedTaskFollowUpMessage('synthesize', graph), graph.messageGroupId, false, undefined, 'synthesize', undefined, ), startWorkflowBuild: async ({ scope, graph, task, workItemId }) => { const plannedBuild: PlannedBuildFollowUp = { isPlannedBuildFollowUp: true, buildTaskId: task.id, workItemId, isSupportingWorkflowTask: task.isSupportingWorkflow === true, }; return await this.startInternalFollowUpRun( scope.user, scope.threadId, this.buildPlannedTaskFollowUpMessage('build-workflow', graph, { buildTask: task }), graph.messageGroupId, false, undefined, undefined, plannedBuild, ); }, startCheckpoint: async ({ scope, graph, task }) => await this.startInternalFollowUpRun( scope.user, scope.threadId, this.buildPlannedTaskFollowUpMessage('checkpoint', graph, { checkpoint: task }), graph.messageGroupId, false, { isCheckpointFollowUp: true, checkpointTaskId: task.id }, undefined, undefined, ), }; } private createPlannedWorkflowVerificationGate(threadId: string): PlannedWorkflowVerificationGate { return { revalidate: async (verification) => await this.workflowObligations.revalidatePlannedWorkflowVerification( threadId, verification, ), }; } private toWorkflowVerificationSourceTask( verification: PlannedWorkflowVerification, ): Pick< ManagedBackgroundTask, 'taskId' | 'role' | 'status' | 'result' | 'error' | 'plannedTaskId' | 'workItemId' > { return { taskId: verification.task.backgroundTaskId ?? verification.task.id, role: 'workflow-builder', status: 'completed', result: verification.task.result, error: verification.task.error, plannedTaskId: verification.task.id, workItemId: verification.obligation.workItemId, }; } private createPlannedWorkflowVerificationTracker(): PlannedWorkflowVerificationTracker { return { scheduled: ({ obligation }) => this.trackWorkflowVerificationObligation(obligation, 'planned_verification_scheduled'), followUpStartAttempted: ({ obligation }, started) => this.trackWorkflowVerificationObligation(obligation, 'follow_up_start_attempted', { follow_up_started: started, }), }; } private async createPlannedTaskDispatchContext( user: User, threadId: string, graph: PlannedTaskGraph, ): Promise { const environment = await this.createExecutionEnvironment( user, threadId, graph.planRunId, createInertAbortSignal(), graph.messageGroupId, this.threadPushRef.get(threadId), ); environment.orchestrationContext.tracing = this.tracing.getTraceContext(graph.planRunId); return environment.orchestrationContext; } private async doSchedulePlannedTasks(user: User, threadId: string): Promise { const revalidated = await this.revalidateActiveUser(user.id); if (!revalidated) { this.logger.warn('Cancelling run: user no longer authorized for AI Assistant', { userId: user.id, threadId, }); this.cancelRun(threadId); return; } const activeUser = revalidated; const { plannedTaskService } = await this.createPlannedTaskState(); const actionRunner = this.createPlannedTaskActionRunner( activeUser, threadId, plannedTaskService, ); while (true) { const graph = await plannedTaskService.getGraph(threadId); if (!graph) return; await this.syncPlannedTasksToUi(threadId, graph); const availableSlots = Math.max( 0, MAX_CONCURRENT_BACKGROUND_TASKS_PER_THREAD - this.backgroundTasks.getRunningTasks(threadId).length, ); const pendingWorkflowVerification = await this.workflowObligations.findPendingPlannedWorkflowVerification(threadId, graph); const action = await plannedTaskService.tick(threadId, { availableSlots, pendingWorkflowVerification, }); if (action.type === 'none') return; const result = await actionRunner.run(action); if (result.type !== 'continue-scheduling') return; } } /** Save the user's prompt when Stop is hit before the stream starts; the SDK only persists it once the stream is invoked. */ private async persistInterruptedUserMessage( threadId: string, userId: string, message: string, createdAt: Date, ): Promise { if (!message) return; try { await this.agentMemory.saveMessages({ threadId, resourceId: userId, messages: [ { id: nanoid(), createdAt, type: 'llm', role: 'user', content: [{ type: 'text', text: message }], }, ], }); } catch (error) { this.logger.warn('Failed to persist user message on cancel', { threadId, error: error instanceof Error ? error.message : String(error), }); } } /** * Splits a message's attachments into the resource references that feed the * context block, gating canvas node-selection attachments behind * CANVAS_NODE_CONTEXT_FLAG per user. Workflow and agent references always pass * through — only `nodes` attachments are conditional. */ private async resolveContextAttachments( attachments: InstanceAiAttachment[] | undefined, user: User, ): Promise { const attachmentsOrEmpty = attachments ?? []; const workflowAttachments = attachmentsOrEmpty.filter( (attachment): attachment is InstanceAiWorkflowAttachment => attachment.type === 'workflow', ); const agentAttachments = attachmentsOrEmpty.filter( (attachment): attachment is InstanceAiAgentAttachment => attachment.type === 'agent', ); const nodeAttachments = attachmentsOrEmpty.filter( (attachment): attachment is InstanceAiNodesAttachment => attachment.type === 'nodes', ); const canvasNodeContextEnabled = nodeAttachments.length > 0 && (await this.canvasNodeContextFlagGate.isEnabled(user)); return [ ...workflowAttachments, ...agentAttachments, ...(canvasNodeContextEnabled ? nodeAttachments : []), ]; } /** * Run body for a fresh orchestrator turn. Never call directly — go through * `startExecuteRun` so the promise is registered with `inFlightExecutions` * and shutdown can drain it before the DB closes. */ // eslint-disable-next-line complexity /** Thread provenance for the trace. Best-effort by construction: a failed * metadata read must not take the run with it — the trace just loses a * label it would have been nice to have. */ private async readThreadProvenance( userId: string, threadId: string, ): Promise> { try { return threadProvenanceMetadata(await this.memoryService.getThreadMetadata(userId, threadId)); } catch (error) { this.logger.debug('Could not read thread provenance for tracing', { threadId, error: error instanceof Error ? error.message : String(error), }); return {}; } } private async executeRun( user: User, threadId: string, runId: string, message: string, abortController: AbortController, attachments?: InstanceAiAttachment[], handoffContext?: InstanceAiHandoffContext, messageGroupId?: string, timeZone?: string, isReplanFollowUp: boolean = false, checkpoint?: { isCheckpointFollowUp: true; checkpointTaskId: string }, resumeReason?: OrchestratorResumeReason, plannedBuild?: PlannedBuildFollowUp, ): Promise { // Split the message's attachments by kind once, here at the agent // boundary: files feed the parse-file / content-block path, workflow // references feed a context block the agent resolves with its tools. // Downstream logic stays single-kind. const fileAttachments = (attachments ?? []).filter( (attachment): attachment is InstanceAiFileAttachment => attachment.type === 'file', ); const contextAttachments = await this.resolveContextAttachments(attachments, user); const signal = abortController.signal; let tracing: InstanceAiTraceContext | undefined; let messageTraceFinalization: MessageTraceFinalization | undefined; let aiCreatedWorkflowIds: Set | undefined; let messageId = ''; let streamReached = false; /** Declared out here so the terminal handlers below can see it. */ let turnHadFileAttachments = false; const turnStartedAt = new Date(); let errorReporterExecutionToken: symbol | undefined; try { errorReporterExecutionToken = this.instanceAiErrorReporter.beginRun(runId); messageId = nanoid(); const traceInput: Record = { message }; if (fileAttachments.length) { traceInput.attachments = fileAttachments.map((attachment) => ({ mimeType: attachment.mimeType, size: attachment.data.length, })); } // `message` is the user's raw text, so without this the trace has no // record that the editor handed the agent a resource. if (contextAttachments.length) { traceInput.resourceAttachments = contextAttachments.map((attachment) => { if (attachment.type === 'nodes') { return { type: attachment.type, id: attachment.workflowId }; } const resource: TracedResourceAttachment = { type: attachment.type, id: attachment.id, }; if (attachment.type === 'agent') { resource.projectId = attachment.projectId; } if (attachment.type === 'workflow' && attachment.executionId) { resource.executionId = attachment.executionId; } return resource; }); } if (messageGroupId) { traceInput.messageGroupId = messageGroupId; } // Shared with createExecutionEnvironment so one ProxyTokenManager backs tracing + the run. const proxyRunConfig = await this.createProxyRunConfig(user); // Read per run: Chrome auto-updates extensions silently. const browserExtension = this.browserSessionService.getExtensionTraceContext(user.id); // Where this thread came from (entry point + the opener's own context), // stamped on every run of it. Both trace paths get it: a build finishes // on the RESUME beat, so stamping only the message turn would leave the // spans that actually contain the work unattributable. const threadProvenance = await this.readThreadProvenance(user.id, threadId); // Create the trace before run-start so the SSE event carries traceId (modelId lands at finalization). if (resumeReason) { tracing = await this.tracing.createOrchestratorResumeTraceContext({ threadId, messageId, messageGroupId, runId, userId: user.id, input: traceInput, resumeReason, metadata: { ...threadProvenance, ...(checkpoint?.isCheckpointFollowUp ? { checkpoint_task_id: checkpoint.checkpointTaskId } : {}), ...(plannedBuild?.isPlannedBuildFollowUp ? { build_task_id: plannedBuild.buildTaskId } : {}), }, browserExtension, }); } else { tracing = await createInstanceAiTraceContext({ threadId, messageId, messageGroupId, runId, userId: user.id, input: traceInput, metadata: threadProvenance, proxyConfig: proxyRunConfig.tracingProxyConfig, n8nVersion: N8N_VERSION, workflowSdkVersion: WORKFLOW_SDK_VERSION, browserExtension, }); } if (this.isRunDebugEnabled()) { this.runDebugBuffer.ensure(runId, threadId, buildRunDebugLabel({ message, resumeReason })); } // Publish run-start (includes userId for audit trail attribution). The // LangSmith ids ride here so user feedback can annotate the trace after // a restart — the durable log is their only home. const traceId = tracing?.rootRun.otelTraceId; const langsmithRunId = tracing?.rootRun.id; const langsmithTraceId = tracing?.rootRun.traceId; this.eventBus.publish(threadId, { type: 'run-start', runId, agentId: orchestratorAgentId(runId), userId: user.id, payload: { messageId, messageGroupId, ...(traceId ? { traceId } : {}), ...(langsmithRunId ? { langsmithRunId } : {}), ...(langsmithTraceId ? { langsmithTraceId } : {}), }, }); // Check if already cancelled before starting agent work if (signal.aborted) { await this.persistInterruptedUserMessage(threadId, user.id, message, turnStartedAt); await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'cancelled', { messageGroupId, correlationId: messageId, }); this.eventBus.publish(threadId, { type: 'run-finish', runId, agentId: orchestratorAgentId(runId), payload: { status: 'cancelled', reason: 'user_cancelled' }, }); return; } const executionPushRef = this.threadPushRef.get(threadId); const environment = await this.createExecutionEnvironment( user, threadId, runId, signal, messageGroupId, executionPushRef, proxyRunConfig, ); const { context, memory, taskStorage, workflowTasks, plannedTaskService, modelId, orchestrationContext, } = environment; aiCreatedWorkflowIds = context.aiCreatedWorkflowIds ??= new Set(); const isPostPlanFollowUp = isReplanFollowUp || checkpoint?.isCheckpointFollowUp === true; // Make the current user message available since memory history only // returns previously-saved messages. orchestrationContext.currentUserMessage = message; orchestrationContext.isReplanFollowUp = isReplanFollowUp; orchestrationContext.timeZone = timeZone ?? this.defaultTimeZone; if (checkpoint?.isCheckpointFollowUp) { orchestrationContext.isCheckpointFollowUp = true; orchestrationContext.checkpointTaskId = checkpoint.checkpointTaskId; // Plan approval authorizes verification; grant runWorkflow on the adapter context // because createInstanceAgent builds domain tools from `context`, not `orchestrationContext.domainContext`. context.permissions = { ...context.permissions, ...(PLANNED_TASK_PERMISSION_OVERRIDES.checkpoint ?? {}), } as typeof context.permissions; // Scope the runWorkflow override to the workflows this checkpoint is verifying: // the orchestrator can call `executions(action="run")` on a depended-on workflow // without HITL, but any other workflow id still requires user approval. const runPolicy = await this.getCheckpointRunPolicy(threadId, checkpoint.checkpointTaskId); context.allowedRunWorkflowIds = runPolicy.allowedWorkflowIds; context.allowedRunWorkflowNames = runPolicy.allowedWorkflowNames; context.requireRunWorkflowApproval = runPolicy.requireApproval; } if (plannedBuild?.isPlannedBuildFollowUp) { context.permissions = { ...context.permissions, ...(PLANNED_TASK_PERMISSION_OVERRIDES['build-workflow'] ?? {}), } as typeof context.permissions; context.workflowBuildContext = { threadId, runId, taskId: plannedBuild.buildTaskId, workItemId: plannedBuild.workItemId, allowPostPlanWorkflowCreate: true, isSupportingWorkflowTask: plannedBuild.isSupportingWorkflowTask, plannedTaskService, workflowTaskService: workflowTasks, onBuildOutcome: (outcome) => { plannedBuild.savedOutcome = outcome; }, }; } else { context.workflowBuildContext = { threadId, runId, taskId: `build-${runId}`, workItemId: `wi_${nanoid(8)}`, allowPostPlanWorkflowCreate: isPostPlanFollowUp, workflowTaskService: workflowTasks, }; } // Thread file attachments into the domain context so parse-file can access them if (fileAttachments.length > 0) { context.currentUserAttachments = fileAttachments; } // When trace replay is enabled but LangSmith isn't configured, // create a minimal context that only supports replay/record wrapping. if (!tracing && process.env.E2E_TESTS === 'true') { const { createTraceReplayOnlyContext } = await lazyImport< typeof import('@n8n/instance-ai') >(async () => await import('@n8n/instance-ai')); tracing = createTraceReplayOnlyContext(); } if (tracing) { orchestrationContext.tracing = tracing; if (this.tracing.getTraceContext(runId) !== tracing) { await this.tracing.configureTraceReplayMode(tracing); this.runState.attachTracing(threadId, tracing); this.tracing.storeTraceContext(runId, threadId, tracing, messageGroupId); } } // The builder never sees an attach-only turn, so nothing else records what // the attached agent looked like when the conversation opened. await this.snapshotAttachedAgents(contextAttachments, orchestrationContext, tracing); const enrichedMessage = await this.buildMessageWithRunningTasks(threadId, message); const contextResourcesBlock = buildContextResourcesBlock(contextAttachments); let handoffContextBlock = ''; let agentPreviewTitleFallback: string | undefined; if (handoffContext?.source === 'agent-preview') { const projectId = context.projectId; if (!projectId) { throw new UnexpectedError( `Instance AI thread "${threadId}" has no bound project; agent-preview handoff requires a project`, ); } await this.assertAgentPreviewHandoffScopes(user, projectId); const agentExecutionService = this.getAgentExecutionService(); if (!agentExecutionService) { throw new UserError('Agent preview handoff is not available'); } const resolved = await resolveAgentPreviewHandoff(handoffContext, { projectId, getThreadDetail: agentExecutionService.getThreadDetail.bind(agentExecutionService), }); handoffContextBlock = resolved.block; agentPreviewTitleFallback = resolved.titleFallback; context.agentBuilderTarget = resolved.target; context.agentPreviewSession = { agentId: handoffContext.agentId, threadId: handoffContext.threadId, ...(handoffContext.executionId ? { executionId: handoffContext.executionId } : {}), }; await saveAgentBuilderTarget(context, resolved.target, { previewSession: context.agentPreviewSession, }); } else { handoffContextBlock = buildHandoffContextBlock(handoffContext); } // Set heuristic title before agent starts — thread always has a title. // For an editor hand-off the user text is empty (the workflow is the // message), so title it with the workflow name and mark it refined so // the LLM title pass doesn't summarize the internal context block. const thread = await memory.getThread(threadId); if (thread && !thread.title) { const handoffTitle = contextAttachments.find(isNamedResourceAttachment)?.name ?? agentPreviewTitleFallback; await patchThread(memory, { threadId, update: ({ metadata }) => handoffTitle ? { title: truncateToTitle(handoffTitle), metadata: { ...metadata, titleRefined: true }, } : { title: truncateToTitle(message) }, }); } const existingTasks = await taskStorage.get(threadId); if (existingTasks) { this.eventBus.publish(threadId, { type: 'tasks-update', runId, agentId: orchestratorAgentId(runId), payload: { tasks: existingTasks }, }); } let nonStructuredAttachments: InstanceAiFileAttachment[] = []; let attachmentManifest = ''; let hasParseableAttachment = false; if (fileAttachments.length > 0) { const classifiedAttachments = classifyAttachments(fileAttachments); nonStructuredAttachments = fileAttachments.filter( (attachment) => !isParseableAttachment(attachment), ); turnHadFileAttachments = nonStructuredAttachments.length > 0; hasParseableAttachment = classifiedAttachments.some( (attachment: { parseable: boolean }) => attachment.parseable, ); attachmentManifest = buildAttachmentManifest(classifiedAttachments); } const messageBody = !message && hasParseableAttachment ? `The user attached file(s) without a message. Inspect the first parseable attachment with parse-file and provide a concise summary.\n\n${attachmentManifest}` : attachmentManifest ? `${enrichedMessage}\n\n${attachmentManifest}` : enrichedMessage; // The context block (an editor hand-off) leads the message so the agent // knows what the user is looking at. On an empty-text hand-off it is the // entire prompt, and the agent greets rather than investigating. const messageWithContext = [contextResourcesBlock, handoffContextBlock, messageBody] .filter(Boolean) .join('\n\n'); // The bound project's NAME rides turn for the same reason as the clock: it is per-thread, // so putting it in the cached system prefix would break caching. const projectSection = await this.resolveProjectContextSection(context); const messageWithProject = projectSection ? withProjectContext(messageWithContext, projectSection) : messageWithContext; // Carry "now" on the per-turn input, not the cached system prefix, so the prefix stays cacheable. // Wrapped so the parser strips it from the displayed user message on history reload. const fullMessage = withCurrentDateTime( messageWithProject, getDateTimeSection(timeZone ?? this.defaultTimeZone), ); const promptBuildRun = tracing ? await tracing.startChildRun(tracing.messageRun, { name: 'prepare: prompt', canonicalName: 'instance-ai.prompt_build', tags: ['prompt'], metadata: { agent_role: 'prompt_build' }, inputs: { message, attachmentCount: attachments?.length ?? 0, }, }) : undefined; let streamInput: string | Message[]; try { // Attachments need the explicit message-object shape (text + file blocks); // a plain prompt goes through as a string. The SDK assigns the message id // and persists the input on receipt. if (nonStructuredAttachments.length > 0) { const baseContent = [ { type: 'text' as const, text: fullMessage }, ...nonStructuredAttachments.map((attachment) => ({ type: 'file' as const, data: attachment.data, mediaType: attachment.mimeType, })), ]; streamInput = [ { role: 'user' as const, content: baseContent, }, ]; } else { streamInput = fullMessage; } if (promptBuildRun && tracing) { // Redact raw attachment data from trace output — log metadata only const traceOutput = typeof streamInput === 'string' ? { fullMessage: streamInput } : { fullMessage, attachmentCount: attachments?.length ?? 0, nonStructuredAttachmentCount: nonStructuredAttachments.length, }; await tracing.finishRun(promptBuildRun, { outputs: traceOutput, metadata: { final_status: 'completed' }, }); } } catch (error) { if (promptBuildRun && tracing) { await tracing.failRun(promptBuildRun, error, { final_status: 'error', }); } throw error; } if (tracing && tracing.actorRun.id === tracing.rootRun.id) { const actorRun = await tracing.startChildRun(tracing.rootRun, { name: 'agent: orchestrator', canonicalName: 'instance-ai.agent.orchestrator', tags: ['orchestrator'], metadata: { agent_role: 'orchestrator', agent_id: orchestratorAgentId(runId), execution_mode: 'foreground', trace_kind: tracing.traceKind, }, inputs: traceInput, }); tracing.actorRun = actorRun; tracing.orchestratorRun = actorRun; } const runControl = createOrchestratorRunControl(orchestrationContext); const stopSignal = (): OrchestratorRunStopSignal | undefined => runControl.getStopSignal(); const agent = await this.createAgentFromEnvironment( environment, threadId, runId, user, tracing, ); const streamOptions = this.buildOrchestratorAgentStreamOptions(user, threadId, runId, signal); streamReached = true; const result = tracing ? await tracing.withActiveSpan(tracing.actorRun, async () => { return await streamAgentRun(agent as StreamableAgent, streamInput, streamOptions, { threadId, runId, agentId: orchestratorAgentId(runId), signal, eventBus: this.eventBus, logger: this.logger, onActivity: () => this.runState.touchActiveRun(threadId), stopSignal, }); }) : await streamAgentRun(agent as StreamableAgent, streamInput, streamOptions, { threadId, runId, agentId: orchestratorAgentId(runId), signal, eventBus: this.eventBus, logger: this.logger, onActivity: () => this.runState.touchActiveRun(threadId), stopSignal, }); if (result.status === 'suspended') { // finalizeRun only fires on terminal outcomes; record suspended-segment usage here. this.emitRunMetrics(threadId, 'suspended', { modelId, workSummary: result.workSummary, usage: result.usage, }); if (result.suspension) { this.runState.suspendRun(threadId, { runId, agentRunId: result.agentRunId, agent, orchestrationContext, threadId, user, toolCallId: result.suspension.toolCallId, ...(result.suspension.toolName ? { toolName: result.suspension.toolName } : {}), ...(result.suspension.suspendPayload ? { suspendPayload: result.suspension.suspendPayload } : {}), requestId: result.suspension.requestId, abortController, messageGroupId, createdAt: Date.now(), tracing, modelId, checkpoint, plannedBuild, runHandoff: runControl.state, }); void this.suspendedThreads.persistPendingConfirmation({ requestId: result.suspension.requestId, threadId, userId: user.id, runId, messageGroupId, kind: 'suspended', toolCallId: result.suspension.toolCallId, checkpointKey: result.agentRunId, checkpointTaskId: checkpoint?.checkpointTaskId, }); // Bill the tokens this segment consumed to reach the suspension. Every // segment of a HITL run shares one agentRunId (resume reuses it), so the // terminal claim would otherwise be the only claim and swallow this usage // — worse, a stop while suspended never reaches a terminal claim at all. // Key on the per-suspension requestId so this never collides with the // terminal claim (bare agentRunId) or another suspension. void this.creditService.claimRunUsage( user, threadId, `${result.agentRunId || runId}:${result.suspension.requestId}`, result.usage?.usage ?? [], 'suspended', ); } // Track intermediate message (text streamed before suspension) const intermediateText = await (result.text ?? Promise.resolve('')); if (intermediateText) { this.telemetry.track('Builder sent message', { thread_id: threadId, message: redactTelemetryText(intermediateText), is_intermediate: true, }); } const waitingDecision = await this.terminalOutcome.evaluateWaitingResponse( threadId, runId, result.confirmationEvent, { messageGroupId, correlationId: messageId, }, ); if (waitingDecision?.reason === 'confirmation-invalid') { messageTraceFinalization = await this.terminalOutcome.finishInvalidConfirmationRun({ threadId, runId, abortController, tracing, }); return; } if (result.confirmationEvent) { this.trackConfirmationRequest(user.id, threadId, result.confirmationEvent); this.eventBus.publish(threadId, result.confirmationEvent); } const suspensionOutputs = buildSuspensionTraceOutputs(runId, result.suspension); await this.tracing.finalizeRunTracing(runId, tracing, { status: 'suspended', outputs: suspensionOutputs, metadata: { completion_source: 'orchestrator', ...(result.suspension?.requestId ? { request_id: result.suspension.requestId } : {}), ...(result.suspension?.toolCallId ? { pending_tool_call_id: result.suspension.toolCallId } : {}), ...(result.suspension?.toolName ? { pending_tool_name: result.suspension.toolName } : {}), }, }); messageTraceFinalization = { status: 'suspended', outputs: suspensionOutputs, metadata: { completion_source: 'orchestrator', ...(result.suspension?.requestId ? { request_id: result.suspension.requestId } : {}), ...(result.suspension?.toolCallId ? { pending_tool_call_id: result.suspension.toolCallId } : {}), ...(result.suspension?.toolName ? { pending_tool_name: result.suspension.toolName } : {}), }, }; return; } // `streamAgentRun` doesn't throw on abort — it returns // `status: 'cancelled'`. When the abort came from shutdown's // preserve-HITL path, falling through into the normal terminal // handling would still call `evaluateTerminalResponse` and // `finalizeRun`, both of which rewrite the snapshot. Bail out // before either fires so the plan/ask card stays on disk. if (result.status === 'cancelled' && this.shouldPreserveHitlOnShutdown(runId)) { return; } const outputText = await (result.text ?? Promise.resolve('')); const terminalError = result.status === 'errored' ? await this.reclassifyMaskedStreamFailure(result.error, user, { threadId, runId }) : undefined; if (result.status === 'errored') { this.instanceAiErrorReporter.report( terminalError ?? new Error('Instance AI stream errored'), { component: 'instance-ai-stream', providerStream: true, threadId, runId, tracing, agentId: orchestratorAgentId(runId), userId: user.id, messageGroupId, messageId, }, ); } // A refused attachment usually lands here rather than in the catch below: // the stream reports the failure instead of throwing. Recovering only on // thrown errors would leave the attachment in history on the common path. const attachmentRemoved = this.shouldDropTurnAttachments({ turnHadAttachments: turnHadFileAttachments, producedNoOutput: result.status === 'errored' && outputText.length === 0, }) ? await this.dropTurnAttachments({ threadId, resourceId: user.id }) : undefined; const userFacingErrorMessage = result.status === 'errored' ? getUserFacingErrorMessage(terminalError, undefined, { attachmentRemoved }) : undefined; const userFacingErrorCode = result.status === 'errored' ? getUserFacingErrorCode(terminalError) : undefined; if (runControl.shouldEmitTerminalOutcome(result.stopReason)) { await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, result.status, { messageGroupId, correlationId: messageId, workSummary: result.workSummary, errorMessage: userFacingErrorMessage, errorCode: userFacingErrorCode, suppressCompletedFallback: checkpoint?.isCheckpointFollowUp === true || plannedBuild?.isPlannedBuildFollowUp === true, }); } const finalStatus = result.status === 'errored' ? 'error' : result.status; await this.tracing.finalizeRunTracing(runId, tracing, { status: finalStatus, outputText, modelId, }); messageTraceFinalization = { status: finalStatus, outputText, modelId, metadata: await this.tracing.buildMessageTraceMetadata(threadId, runId, { status: finalStatus, }), }; const archivedWorkflowIds = await this.temporaryWorkflowService.reapForRun( threadId, user, aiCreatedWorkflowIds, this.backgroundTasks.getRunningTasks(threadId).length, ); await this.finalizeRun(threadId, runId, result.status, { userId: user.id, modelId, archivedWorkflowIds, workSummary: result.workSummary, usage: result.usage, errorReason: userFacingErrorMessage, ...(result.status === 'errored' ? { errorInfo: { errorMessage: terminalError ? getErrorMessage(terminalError) : 'Instance AI stream errored', errorSource: 'stream' as const, }, } : {}), }); // Bill token usage for every terminal outcome (completed / cancelled / errored), // deduped per run segment. `result.agentRunId` is the segment id; fall back to runId. await this.creditService.claimRunUsage( user, threadId, result.agentRunId || runId, result.usage?.usage ?? [], result.status, ); if (result.status === 'completed') { this.telemetry.track('Builder sent message', { thread_id: threadId, message: redactTelemetryText(outputText), }); this.telemetry.track('Builder satisfied user intent', { thread_id: threadId, }); } } catch (error) { if (signal.aborted) { // Shutdown asked us to preserve the HITL card on disk: the // service has already finalised tracing and the per-run DB // rows (pending_confirmation, snapshot) are the durable // signal. Emitting the terminal-fallback text + run-finish // here would clobber the plan/ask snapshot the user expects // to see on reload, so just bail out. if (this.shouldPreserveHitlOnShutdown(runId)) { return; } if (!streamReached) { await this.persistInterruptedUserMessage(threadId, user.id, message, turnStartedAt); } const runTimeout = this.liveness.consumeRunTimeout(runId); const cancellationReason = runTimeout.timedOut ? INSTANCE_AI_RUN_TIMEOUT_REASON : getAbortReason(signal); if (cancellationReason === INSTANCE_AI_RUN_TIMEOUT_REASON) { this.liveness.publishRunTimeoutNotice(threadId, runId); } await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'cancelled', { messageGroupId, correlationId: messageId, }); await this.tracing.finalizeRunTracing(runId, tracing, { status: 'cancelled', reason: cancellationReason, }); messageTraceFinalization = { status: 'cancelled', reason: cancellationReason, metadata: await this.tracing.buildMessageTraceMetadata(threadId, runId, { status: 'cancelled', cancellationReason, runTimeout, }), }; const archivedWorkflowIds = await this.temporaryWorkflowService.reapForRun( threadId, user, aiCreatedWorkflowIds, this.backgroundTasks.getRunningTasks(threadId).length, ); this.publishRunFinish( threadId, runId, 'cancelled', cancellationReason, archivedWorkflowIds, user.id, ); return; } const terminalError = await this.reclassifyMaskedStreamFailure(error, user, { threadId, runId, }); // The attachment is persisted in history before the model call is known to // have succeeded, so a refused file would fail every later turn as well. // Drop it here to keep the thread usable, and tell the user what really // happened — a failed cleanup leaves the thread poisoned. const attachmentRemoved = this.shouldDropTurnAttachments({ turnHadAttachments: turnHadFileAttachments, producedNoOutput: true, }) ? await this.dropTurnAttachments({ threadId, resourceId: user.id }) : undefined; const errorMessage = getErrorMessage(terminalError); const userFacingErrorMessage = getUserFacingErrorMessage(terminalError, undefined, { attachmentRemoved, }); const userFacingErrorCode = getUserFacingErrorCode(terminalError); const errCtx: InstanceAiObservabilityContext = { threadId, runId, tracing, agentId: orchestratorAgentId(runId), userId: user.id, messageGroupId, messageId, }; this.logger.error(`Instance AI run error: ${errorMessage}`, { error: errorMessage, ...buildInstanceAiObservabilityContext(errCtx), }); this.instanceAiErrorReporter.report(terminalError, { component: 'instance-ai-run', ...errCtx, }); await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'errored', { messageGroupId, correlationId: messageId, errorMessage: userFacingErrorMessage, errorCode: userFacingErrorCode, }); await this.tracing.finalizeRunTracing(runId, tracing, { status: 'error', reason: errorMessage, }); messageTraceFinalization = { status: 'error', reason: errorMessage, metadata: await this.tracing.buildMessageTraceMetadata(threadId, runId, { status: 'error', }), }; const archivedWorkflowIds = await this.temporaryWorkflowService.reapForRun( threadId, user, aiCreatedWorkflowIds, this.backgroundTasks.getRunningTasks(threadId).length, ); this.publishRunFinish( threadId, runId, 'errored', userFacingErrorMessage, archivedWorkflowIds, user.id, { errorMessage, errorSource: 'exception' }, ); } finally { this.runState.clearActiveRun(threadId); const segmentSuspended = messageTraceFinalization?.status === 'suspended'; // Note: don't delete threadPushRef here. Planned tasks (build agent, // checkpoint verifications) dispatch later in this same finally and // later still in the post-run scheduler — they need the pushRef to // route execution events to the user's iframe session. The next // startRun overwrites it; thread-cleanup deletes it on dispose. this.domainAccessTrackersByThread.get(threadId)?.clearRun(runId); if (messageTraceFinalization) { if (tracing) { await this.tracing.finalizeMessageTraceRoot(runId, tracing, messageTraceFinalization); } if (messageTraceFinalization.status !== 'cancelled' && !segmentSuspended) { this.liveness.consumeRunTimeout(runId); } } // Must precede the reschedule below: the next tick consults the streak // before re-arming another follow-up for the same unsettled trigger. this.updateInternalFollowUpFailureStreak( threadId, messageTraceFinalization?.status, resumeReason !== undefined, ); // Post-run planned-task wiring (only when the run is actually ending, // not when it merely suspended for HITL): // 1. Checkpoint deadlock fallback — if this run was a checkpoint // follow-up and the orchestrator exited without calling // complete-checkpoint, mark the task failed so the scheduler // can transition to awaiting_replan. Runs even on a stop: the // cancelled run's context is the only thing that knows about // this follow-up, so skipping it strands the task at `running`. // 2. Reschedule — drive the next tick, unless the run was stopped // (a stop must not start another run). This covers the case // where a background task settled during an ordinary chat run: // its schedulePlannedTasks call may have skipped the checkpoint // branch because hasLiveRun was true. Ticking again now (with // no live run) picks it up. schedulerLocks serializes this // call, and tick() is a no-op when no graph exists. // 3. UI projection — always, so a stopped run's task states reach // the client. if (!segmentSuspended && !this.runState.hasSuspendedRun(threadId)) { const reschedule = !signal.aborted; if (checkpoint?.isCheckpointFollowUp) { await this.finalizeCheckpointFollowUp(user, threadId, checkpoint.checkpointTaskId, { reschedule, }); } else if (plannedBuild?.isPlannedBuildFollowUp) { await this.finalizePlannedBuildFollowUp(user, threadId, plannedBuild, { reschedule }); } else if (reschedule) { await this.schedulePlannedTasks(user, threadId); } if (reschedule) { await this.drainPendingCheckpointReentries(user, threadId); } await this.taskProjector.syncFromWorkflowLoop(threadId, runId); if (reschedule) { await this.maybeStartWorkflowSetupFollowUp(user, threadId); } } if (errorReporterExecutionToken) { this.instanceAiErrorReporter.endRun(runId, errorReporterExecutionToken); } } } /** * Post-run cleanup for a checkpoint follow-up. Ensures the checkpoint task is * terminal (marking it failed if the orchestrator abandoned it) and re-ticks * the scheduler so the next planned action can fire. */ private queuePendingCheckpointReentry(threadId: string, checkpointTaskId: string): void { let set = this.pendingCheckpointReentries.get(threadId); if (!set) { set = new Set(); this.pendingCheckpointReentries.set(threadId, set); } set.add(checkpointTaskId); } /** * Drain any checkpoint re-entries whose parent-tagged children settled while * an orchestrator run was live (or while other siblings were still running). * Called from the post-run cleanup path in every run-ending `finally` block, * so the checkpoint is never left orphaned when the settlement path could * not fire immediately. */ private async drainPendingCheckpointReentries(user: User, threadId: string): Promise { const set = this.pendingCheckpointReentries.get(threadId); if (!set || set.size === 0) return; const snapshot = [...set]; for (const checkpointTaskId of snapshot) { // If a new run started while we were draining, stop — the next run's // cleanup will pick up the remaining markers. if (this.runState.getActiveRunId(threadId) || this.runState.hasSuspendedRun(threadId)) { return; } // A new parent-tagged child is running — let its settlement drive the // checkpoint instead of racing another re-entry. const siblings = this.backgroundTasks.getRunningTasksByParentCheckpoint( threadId, checkpointTaskId, ); if (siblings.length > 0) continue; set.delete(checkpointTaskId); await this.reenterCheckpointById(user, threadId, checkpointTaskId); } if (set.size === 0) this.pendingCheckpointReentries.delete(threadId); } /** * Fire a synthetic `` for the * given checkpoint task id when the parent-tagged children that drove it * are no longer running and no new orchestrator run is live. Used by both * the immediate re-entry path (via `maybeReenterParentCheckpoint`) and the * deferred drain (via `drainPendingCheckpointReentries`). */ private async reenterCheckpointById( user: User, threadId: string, checkpointTaskId: string, messageGroupId?: string, ): Promise { try { const { plannedTaskService } = await this.createPlannedTaskState(); const graph = await plannedTaskService.getGraph(threadId); const checkpoint = graph?.tasks.find((t) => t.id === checkpointTaskId); if (!graph || !checkpoint || checkpoint.kind !== 'checkpoint') return false; if (checkpoint.status !== 'running') return false; const startedRunId = await this.startInternalFollowUpRun( user, threadId, this.buildPlannedTaskFollowUpMessage('checkpoint', graph, { checkpoint }), messageGroupId, false, { isCheckpointFollowUp: true, checkpointTaskId }, ); if (!startedRunId) return false; this.logger.debug('Re-entered checkpoint follow-up', { threadId, checkpointTaskId, messageGroupId, }); return true; } catch (error) { this.logger.error('Failed to re-enter checkpoint follow-up', { threadId, checkpointTaskId, error: error instanceof Error ? error.message : String(error), }); return false; } } /** * When a direct background task (builder/research/data-table) * settles and was spawned inside a checkpoint follow-up, try to re-enter * that checkpoint so the orchestrator can call `complete-checkpoint`. * * Returns `true` only when a follow-up was actually started. Returns * `false` in every other case (checkpoint no longer running, siblings * still in-flight, an orchestrator run is active or suspended, or the * graph no longer has the checkpoint). The caller is responsible for * queuing a deferred re-entry in the false case — never falling through * to a generic `` shell, which would re-open * the orphan bug. */ private async maybeReenterParentCheckpoint( user: User, threadId: string, task: ManagedBackgroundTask, ): Promise { const parentCheckpointId = task.parentCheckpointId; if (!parentCheckpointId) return false; // If other parent-tagged children are still running, let the LAST one // re-drive the checkpoint; emitting multiple re-dispatches would race. const siblings = this.backgroundTasks .getRunningTasksByParentCheckpoint(threadId, parentCheckpointId) .filter((t) => t.taskId !== task.taskId); if (siblings.length > 0) return false; // If a run is live, defer — startInternalFollowUpRun would be rejected // and we must not fall through to the shell path. if (this.runState.getActiveRunId(threadId) || this.runState.hasSuspendedRun(threadId)) { return false; } return await this.reenterCheckpointById( user, threadId, parentCheckpointId, task.messageGroupId, ); } private async finalizeCheckpointFollowUp( user: User, threadId: string, checkpointTaskId: string, { reschedule = true }: { reschedule?: boolean } = {}, ): Promise { try { const { plannedTaskService } = await this.createPlannedTaskState(); const graph = await plannedTaskService.getGraph(threadId); const task = graph?.tasks.find((t) => t.id === checkpointTaskId); if (task && task.status === 'running') { // If the orchestrator spawned a detached sub-agent inside this // checkpoint's turn (builder, research, data-table) and // that child is still running, leave the checkpoint running. The // child's settlement path re-emits `orchestrate-checkpoint` so the // orchestrator re-enters the same checkpoint context and can then // call `complete-checkpoint`. const inflightChildren = this.backgroundTasks.getRunningTasksByParentCheckpoint( threadId, checkpointTaskId, ); if (inflightChildren.length > 0) { this.logger.debug( 'Checkpoint run ended with in-flight child tasks — deferring finalization', { threadId, checkpointTaskId, inflightTaskIds: inflightChildren.map((t) => t.taskId), }, ); } else { this.logger.warn('Checkpoint run ended without reporting completion — marking failed', { threadId, checkpointTaskId, }); await plannedTaskService.markCheckpointFailed(threadId, checkpointTaskId, { error: 'Checkpoint run ended without reporting completion', }); const nextGraph = await plannedTaskService.getGraph(threadId); if (nextGraph) { await this.syncPlannedTasksToUi(threadId, nextGraph); } } } } catch (error) { this.logger.error('Checkpoint finalization failed', { threadId, checkpointTaskId, error: error instanceof Error ? error.message : String(error), }); } if (!reschedule) return; await this.schedulePlannedTasks(user, threadId); } private async finalizePlannedBuildFollowUp( user: User, threadId: string, plannedBuild: PlannedBuildFollowUp, { reschedule = true }: { reschedule?: boolean } = {}, ): Promise { try { const { plannedTaskService } = await this.createPlannedTaskState(); const graph = await plannedTaskService.getGraph(threadId); const task = graph?.tasks.find((t) => t.id === plannedBuild.buildTaskId); if (task && task.status === 'running') { if (plannedBuild.savedOutcome?.submitted === true) { await plannedTaskService.markSucceeded(threadId, plannedBuild.buildTaskId, { result: plannedBuild.savedOutcome.summary, outcome: plannedBuild.savedOutcome, }); } else { this.logger.warn('Build workflow follow-up ended without saving — marking failed', { threadId, buildTaskId: plannedBuild.buildTaskId, }); await plannedTaskService.markFailed(threadId, plannedBuild.buildTaskId, { error: 'Workflow build run ended without saving a workflow', }); } const nextGraph = await plannedTaskService.getGraph(threadId); if (nextGraph) { await this.syncPlannedTasksToUi(threadId, nextGraph); } } } catch (error) { this.logger.error('Build workflow finalization failed', { threadId, buildTaskId: plannedBuild.buildTaskId, error: error instanceof Error ? error.message : String(error), }); } if (!reschedule) return; await this.schedulePlannedTasks(user, threadId); } async resolveConfirmation( requestingUserId: string, requestId: string, request: InstanceAiConfirmRequest, ): Promise { const data = toConfirmationData(request); const freshUser = await this.revalidateActiveUser(requestingUserId); if (!freshUser) { this.runState.rejectPendingConfirmation(requestId); const suspended = this.runState.findSuspendedByRequestId(requestId); if (suspended?.user.id === requestingUserId) { this.cancelRun(suspended.threadId); } this.logger.warn('Rejecting confirmation: user no longer authorized for AI Assistant', { userId: requestingUserId, requestId, }); return null; } // Close the same-process window: a click on a pre-rendered card can still // reach the in-memory fast path after the row's `expiresAt` but before the // prune/liveness sweep drops it. The persisted row is the source of truth // for expiry, so consult it first and refuse a stale click; the sweep still // owns releasing the suspended run. Scoped by `freshUser.id` so another // user's request ID falls through to the existing not-found handling // rather than leaking an "expired" signal. One extra SELECT per click — // the click path isn't hot. if (await this.pendingConfirmationRepo.isPastExpiry(requestId, freshUser.id, new Date())) { this.logger.debug('Rejecting expired confirmation', { requestId }); throw new UserError(CONFIRMATION_EXPIRED_MESSAGE); } const pending = this.runState.getPendingConfirmation(requestId); if ( pending && pending.userId === freshUser.id && this.runState.resolvePendingConfirmation(freshUser.id, requestId, data) ) { void this.suspendedThreads.dropPendingConfirmation(requestId); this.logger.debug('Resolved pending confirmation (sub-agent HITL)', { requestId, approved: data.approved, }); const runId = this.runState.getActiveRunId(pending.threadId); return { ok: true, ...(runId ? { runId } : {}), }; } this.logger.debug('Pending confirmation not found, trying suspended run resume', { requestId, approved: data.approved, }); const resumed = await this.resumeSuspendedRun(requestingUserId, requestId, data); if (resumed) { return resumed; } // Last resort: the in-memory state is gone, but a persisted index row // may still exist from before a process restart. For `suspended`-kind // rows we try to rebuild the agent from the DB-backed checkpoint and // resume; for `inline`-kind (no checkpoint, just an in-process Promise // that died with the previous process) or any rebuild failure we // publish a terminal `run-finish` and surface a clear UserError so the // client doesn't sit on a stale confirmation card. return await this.suspendedRunRestorer.resolveOrphanedConfirmation( requestingUserId, requestId, data, ); } private async buildMcpServers( user: User, threadId: string, runId: string, tracing: InstanceAiTraceContext | undefined, messageGroupId?: string, ): Promise { const staticMcpServers = this.parseMcpServers(this.instanceAiConfig.mcpServers); const registryMcpServers = this.settingsService.isMcpAccessEnabled() ? await this.instanceAiErrorReporter.withBoundary( 'instance-ai-mcp-setup', { threadId, runId, tracing, agentId: orchestratorAgentId(runId), userId: user.id, messageGroupId, }, async () => await this.mcpRegistryService.getRegistryMcpServers(user), ) : []; return [...staticMcpServers, ...registryMcpServers]; } private async createAgentFromEnvironment( environment: Awaited>, threadId: string, runId: string, user: User, tracing: InstanceAiTraceContext | undefined, ): Promise { if (tracing) { environment.orchestrationContext.tracing = tracing; } await this.bindAgentPreviewSession(environment.context, user); const mcpServers = await this.buildMcpServers( user, threadId, runId, tracing, environment.orchestrationContext.messageGroupId, ); const { agent, mcpConnectionFailures } = await createInstanceAgent({ modelId: environment.modelId, context: environment.context, orchestrationContext: environment.orchestrationContext, mcpServers, mcpManager: this.mcpClientManager, memoryConfig: this.createAgentMemoryOptions(user, threadId, runId), memory: environment.memory, checkpointStore: this.checkpointStore, onMemoryTaskEvent: this.memoryTaskObserverFor(threadId, tracing), thinkingEnabled: this.instanceAiConfig.thinkingEnabled, }); // Surface MCP connection failures as a non-fatal status event. Publishing // here (rather than at each call site) covers the foreground run and both // resume paths (auto-setup rebuild + process-restart resume), so the user // is informed whenever a server is skipped — including after a resume. // Failures come from createInstanceAgent's result (threaded from the MCP // manager's getRegularTools call), not a shared manager field, so they // reflect this run's config and can't leak another run's server names. if (mcpConnectionFailures.length > 0) { const names = mcpConnectionFailures.map((f) => f.server).join(', '); for (const failure of mcpConnectionFailures) { this.errorReporter.error( new Error(`MCP server "${failure.server}" failed to connect: ${failure.error}`), { level: 'warning', tags: { component: 'instance-ai-mcp', server: failure.server }, extra: { runId, threadId, server: failure.server, error: failure.error }, shouldIsolate: true, }, ); } this.eventBus.publish(threadId, { type: 'status', runId, agentId: orchestratorAgentId(runId), payload: { message: `Couldn't reach MCP server${mcpConnectionFailures.length > 1 ? 's' : ''} ${names}; continuing without their tools.`, }, }); } this.subscribeToAgentErrors(agent, threadId, runId); return agent; } private async buildFreshInstanceAgent( user: User, threadId: string, runId: string, abortSignal: AbortSignal, tracing: InstanceAiTraceContext | undefined, messageGroupId?: string, pushRef?: string, ): Promise<{ agent: InstanceAgent; modelId: ModelConfig; orchestrationContext: OrchestrationContext; }> { const environment = await this.createExecutionEnvironment( user, threadId, runId, abortSignal, messageGroupId, pushRef, ); const agent = await this.createAgentFromEnvironment( environment, threadId, runId, user, tracing, ); return { agent, modelId: environment.modelId, orchestrationContext: environment.orchestrationContext, }; } /** * Rebuild the in-memory pieces a suspended run needs (user, agent, * execution environment) from a persisted orphan row + checkpoint, and * package them into a `SuspendedRunState`. The caller wraps the result in * `runState.suspendRun` and hands off to `resumeSuspendedRun`. * * Returns a discriminated union so the caller can log a precise reason * per failure mode without inlining each step's try/catch. Logging here * would be premature — the helper has the failure context but not the * routing decision (`tryResumeFromOrphan` decides whether the failure is * worth surfacing to the user vs. just retrying). */ private async rebuildSuspendedRunFromCheckpoint( orphan: ResumableOrphan, ): Promise { const user = await this.revalidateActiveUser(orphan.userId); if (!user) return { kind: 'no-user' }; // Bail early if the checkpoint store doesn't have a usable snapshot — // `load()` throws UserError for expired tombstones and returns // undefined when the row is gone entirely. Either way there's nothing // to resume. try { const state = await this.checkpointStore.load(orphan.checkpointKey); if (!state) return { kind: 'no-checkpoint' }; } catch (error: unknown) { return { kind: 'no-checkpoint', error }; } const abortController = new AbortController(); let environment; try { environment = await this.createExecutionEnvironment( user, orphan.threadId, orphan.runId, abortController.signal, orphan.messageGroupId ?? undefined, this.threadPushRef.get(orphan.threadId), ); } catch (error: unknown) { return { kind: 'env-failure', error }; } const runControl = createOrchestratorRunControl(environment.orchestrationContext); let agent; try { agent = await this.createAgentFromEnvironment( environment, orphan.threadId, orphan.runId, user, undefined, ); } catch (error: unknown) { return { kind: 'agent-failure', error }; } return { kind: 'ready', state: { runId: orphan.runId, agentRunId: orphan.checkpointKey, agent, orchestrationContext: environment.orchestrationContext, threadId: orphan.threadId, user, toolCallId: orphan.toolCallId, requestId: orphan.requestId, abortController, messageGroupId: orphan.messageGroupId ?? undefined, createdAt: Date.now(), tracing: undefined, modelId: environment.modelId, checkpoint: orphan.checkpointTaskId ? { isCheckpointFollowUp: true, checkpointTaskId: orphan.checkpointTaskId } : undefined, runHandoff: runControl.state, }, }; } private async revalidateActiveUser(userId: string): Promise { try { const user = await this.userRepository.findOne({ where: { id: userId }, relations: ['role'], }); if (!user || user.disabled) return null; const hasInstanceAiMessageScope = user.role?.scopes?.some((scope) => scope.slug === 'instanceAi:message') ?? false; return hasInstanceAiMessageScope ? user : null; } catch (error: unknown) { this.logger.warn('Failed to revalidate user', { userId, error: getErrorMessage(error), }); return null; } } /** * The one-line "you are in project X" fact for the per-turn block, or undefined * when there is nothing useful to say (no bound project, no workspace adapter, a * project we can't read). * * Best-effort by design: this is a guardrail, not a precondition. A run that cannot * name its project should be a less-informed run, not a failed one - the write access is * locked to the bound project either way. */ private async resolveProjectContextSection( context: InstanceAiContext, ): Promise { const projectId = context.projectId; if (!projectId) return undefined; // Read per turn, deliberately NOT cached. A cache keyed by project id has no // invalidation path here, so a renamed project would have the agent naming the // old name for the rest of the process's life — and naming the wrong project is // the failure this block exists to prevent. try { const project = await context.workspaceService?.getProject?.(projectId); if (project) return getProjectContextSection({ name: project.name, type: project.type }); this.logger.warn('Instance AI could not name the bound project for this turn', { projectId, reason: context.workspaceService?.getProject ? 'not-readable' : 'no-workspace-adapter', }); return undefined; } catch (error) { this.logger.warn('Instance AI failed to resolve the bound project for this turn', { projectId, error: error instanceof Error ? error.message : String(error), }); this.errorReporter.error(error, { level: 'warning', tags: { component: 'instance-ai-project-context' }, extra: { projectId }, shouldIsolate: true, }); return undefined; } } private async canAccessAgentPreviewHandoff(user: User, projectId: string): Promise { const requiredScopes: Scope[] = ['agent:read', 'agent:update']; return await userHasScopes(user, requiredScopes, false, { projectId }); } private async assertAgentPreviewHandoffScopes(user: User, projectId: string): Promise { if (!(await this.canAccessAgentPreviewHandoff(user, projectId))) { throw new ForbiddenError( 'You do not have permission to load or edit agent previews in this project.', ); } } private async rebuildAgentForResume( user: User, threadId: string, runId: string, abortController: AbortController, tracing: InstanceAiTraceContext | undefined, runHandoff: OrchestratorRunHandoffState | undefined, messageGroupId?: string, ): Promise< | { agent: InstanceAgent; modelId: ModelConfig; orchestrationContext: OrchestrationContext; } | undefined > { try { const rebuilt = await this.buildFreshInstanceAgent( user, threadId, runId, abortController.signal, tracing, messageGroupId, this.threadPushRef.get(threadId), ); createOrchestratorRunControl(rebuilt.orchestrationContext, runHandoff ?? {}); return { agent: rebuilt.agent, modelId: rebuilt.modelId, orchestrationContext: rebuilt.orchestrationContext, }; } catch (error: unknown) { this.logger.warn('Failed to rebuild agent for resume', { threadId, runId, error: getErrorMessage(error), }); return undefined; } } private async resumeSuspendedRun( requestingUserId: string, requestId: string, data: ConfirmationData, ): Promise { const suspended = this.runState.findSuspendedByRequestId(requestId); if (!suspended) { this.logger.warn('Confirmation target not found: no pending confirmation or suspended run', { requestId, approved: data.approved, }); return null; } const { agent, runId, agentRunId, threadId, user, toolCallId, toolName, suspendPayload, abortController, tracing, modelId, messageGroupId, checkpoint, plannedBuild, runHandoff, orchestrationContext, } = suspended; if (user.id !== requestingUserId) return null; const activeRun = this.runState.getActiveRun(threadId); if (activeRun) { this.logger.warn('Rejecting suspended-run confirmation: thread already has an active run', { requestId, threadId, suspendedRunId: runId, activeRunId: activeRun.runId, }); return null; } const activeUser = await this.revalidateActiveUser(user.id); if (!activeUser) { this.logger.warn('Cancelling suspended run: user no longer authorized for AI Assistant', { userId: user.id, threadId, requestId, }); this.cancelRun(threadId); return null; } const resumeExecutionToken = Symbol('instance-ai-resume-execution'); if (!this.runState.activateSuspendedRun(threadId, resumeExecutionToken)) { return { ok: true, runId }; } // The in-memory `suspendedRuns` map carries no resolver callback, so // the suspended-kind DB row has to be dropped explicitly here. The // inline-kind drop is wired into the Promise resolver in // `waitForConfirmation` and fires whether the resolution came from the // user, from `cancelThread`, or from a liveness timeout. void this.suspendedThreads.dropPendingConfirmation(requestId); const resumeData = buildResumeData(data); const resumeTracing = await this.tracing.createOrchestratorResumeTraceContext({ baseTracing: tracing, threadId, messageId: nanoid(), messageGroupId, runId, userId: activeUser.id, modelId, input: { requestId, toolCallId, approved: data.approved, resumeFields: Object.keys(resumeData), ...(data.userInput ? { userInput: data.userInput } : {}), ...(data.action ? { action: data.action } : {}), ...(data.resourceDecision ? { resourceDecision: data.resourceDecision } : {}), ...(data.answers ? { answers: data.answers } : {}), }, resumeReason: 'approval', metadata: { ...(await this.readThreadProvenance(activeUser.id, threadId)), request_id: requestId, pending_tool_call_id: toolCallId, approved: data.approved, ...(checkpoint?.isCheckpointFollowUp ? { checkpoint_task_id: checkpoint.checkpointTaskId } : {}), ...(plannedBuild?.isPlannedBuildFollowUp ? { build_task_id: plannedBuild.buildTaskId } : {}), }, browserExtension: this.browserSessionService.getExtensionTraceContext(activeUser.id), register: false, }); const effectiveTracing = resumeTracing ?? tracing; const unregisteredResumeTracing = resumeTracing && resumeTracing !== tracing ? resumeTracing : undefined; let resumeAgent = agent; let resumeModelId = modelId; let resumeOrchestrationContext = orchestrationContext; if (data.autoSetup || data.connectedSlugs?.length) { const rebuilt = await this.rebuildAgentForResume( activeUser, threadId, runId, abortController, effectiveTracing, runHandoff, messageGroupId, ); if (!rebuilt) { const rebuildFailure = 'Agent rebuild failed'; await this.tracing.finalizeDetachedTraceRun( `resume-rebuild:${runId}`, unregisteredResumeTracing, { status: 'failed', error: rebuildFailure, metadata: { completion_source: 'resume_rebuild' }, }, ); this.cancelRun(threadId, 'agent_rebuild_failed'); // `activateSuspendedRun` above already promoted this run to active, so // `cancelRun` leaves the terminal event to a run loop that was never // started — and the pending confirmation is already dropped, so the // card can't be retried either. await this.emitTerminalRun({ threadId, runId, status: 'errored', reason: OPERATIONAL_ERROR_USER_MESSAGE, errorInfo: { errorMessage: rebuildFailure, errorSource: 'exception' }, messageGroupId, user: activeUser, }); this.runState.clearActiveRun(threadId, resumeExecutionToken); return null; } resumeAgent = rebuilt.agent; resumeModelId = rebuilt.modelId; resumeOrchestrationContext = rebuilt.orchestrationContext; } this.startProcessResumedStream(resumeAgent, resumeData, { runId, agentRunId, threadId, user: activeUser, toolCallId, toolName, suspendPayload, signal: abortController.signal, abortController, tracing: effectiveTracing, orchestrationContext: resumeOrchestrationContext, modelId: resumeModelId, checkpoint, plannedBuild, runHandoff, resumeExecutionToken, messageGroupId, resumeTracing, unregisteredResumeTracing, }); return { ok: true, runId }; } /** * Run body for a resumed suspended orchestrator turn. Never call directly * — go through `startProcessResumedStream` so the promise is registered * with `inFlightExecutions` and shutdown can drain it before the DB * closes. */ private async processResumedStream( agent: unknown, resumeData: Record, opts: { runId: string; agentRunId: string; threadId: string; user: User; toolCallId: string; toolName?: string; suspendPayload?: Record; signal: AbortSignal; abortController: AbortController; tracing?: InstanceAiTraceContext; orchestrationContext?: OrchestrationContext; modelId?: ModelConfig; checkpoint?: { isCheckpointFollowUp: true; checkpointTaskId: string }; plannedBuild?: PlannedBuildFollowUp; runHandoff?: OrchestratorRunHandoffState; resumeExecutionToken?: symbol; messageGroupId?: string; resumeTracing?: InstanceAiTraceContext; unregisteredResumeTracing?: InstanceAiTraceContext; }, ): Promise { let messageTraceFinalization: MessageTraceFinalization | undefined; let completedSetupWorkflowId: string | undefined; let skipPostRunCleanup = false; let resumeClaimed = false; let resumeTraceRegistered = false; let errorReporterExecutionToken: symbol | undefined; /** * Set once the model run has yielded output. The catch below also wraps * post-result finalization, so without this a finalization failure after a * perfectly good turn would be read as "the run produced nothing" and would * strip that turn's attachments out of history. */ let resumedRunProducedOutput = false; const onResumeClaimed = async () => { if (resumeClaimed) return; resumeClaimed = true; if (opts.resumeTracing && !resumeTraceRegistered) { this.tracing.registerTraceContext( opts.runId, opts.threadId, opts.resumeTracing, opts.messageGroupId, ); resumeTraceRegistered = true; } // Orchestration tools read this shared context at call time. Keep the // suspended trace attached until this process owns the durable resume. if (opts.orchestrationContext && opts.tracing) { opts.orchestrationContext.tracing = opts.tracing; } errorReporterExecutionToken = this.instanceAiErrorReporter.beginRun(opts.runId); }; try { // Built agents snapshot telemetry while creating the resume runtime, so // configure it before calling resume. Registration of the new trace // context itself remains gated by the durable checkpoint claim above. if (opts.tracing?.getTelemetry && isTelemetryConfigurableAgent(agent)) { try { agent.telemetry( opts.tracing.getTelemetry({ agentRole: 'orchestrator', functionId: 'instance-ai.orchestrator', executionMode: opts.tracing.traceKind === 'orchestrator_resume' ? 'resume' : 'foreground', }), ); } catch (error) { this.logger.warn('Failed to configure Instance AI resume tracing', { error: getErrorMessage(error), threadId: opts.threadId, runId: opts.runId, }); } } const resumeOptions = { ...this.buildOrchestratorResumeAgentOptions( opts.user, opts.threadId, opts.runId, opts.agentRunId, opts.toolCallId, opts.signal, ), onResumeClaimed, }; const runControl = createOrchestratorRunControlForState(opts.runHandoff); const stopSignal = (): OrchestratorRunStopSignal | undefined => runControl.getStopSignal(); const result = opts.tracing ? await opts.tracing.withActiveSpan(opts.tracing.actorRun, async () => { return await resumeAgentRun(agent, resumeData, resumeOptions, { threadId: opts.threadId, runId: opts.runId, agentId: orchestratorAgentId(opts.runId), signal: opts.signal, eventBus: this.eventBus, logger: this.logger, agentRunId: opts.agentRunId, onActivity: () => this.runState.touchActiveRun(opts.threadId), stopSignal, }); }) : await resumeAgentRun(agent, resumeData, resumeOptions, { threadId: opts.threadId, runId: opts.runId, agentId: orchestratorAgentId(opts.runId), signal: opts.signal, eventBus: this.eventBus, logger: this.logger, agentRunId: opts.agentRunId, onActivity: () => this.runState.touchActiveRun(opts.threadId), stopSignal, }); if (!resumeClaimed) { skipPostRunCleanup = true; const claimError = result.error ?? new Error('Resume checkpoint claim did not complete'); await this.settleUnclaimedResume(opts, claimError, 'stream'); return; } if (result.status === 'suspended') { // As in the initial-run path, record suspended-segment usage here. this.emitRunMetrics(opts.threadId, 'suspended', { modelId: opts.modelId, workSummary: result.workSummary, usage: result.usage, }); if (result.suspension) { const resumeMessageGroupId = this.tracing.getMessageGroupId(opts.runId); this.runState.suspendRun(opts.threadId, { runId: opts.runId, agentRunId: result.agentRunId, agent, orchestrationContext: opts.orchestrationContext, threadId: opts.threadId, user: opts.user, toolCallId: result.suspension.toolCallId, ...(result.suspension.toolName ? { toolName: result.suspension.toolName } : {}), ...(result.suspension.suspendPayload ? { suspendPayload: result.suspension.suspendPayload } : {}), requestId: result.suspension.requestId, abortController: opts.abortController, messageGroupId: resumeMessageGroupId, createdAt: Date.now(), tracing: opts.tracing, ...(opts.modelId !== undefined ? { modelId: opts.modelId } : {}), checkpoint: opts.checkpoint, plannedBuild: opts.plannedBuild, runHandoff: runControl.state, }); void this.suspendedThreads.persistPendingConfirmation({ requestId: result.suspension.requestId, threadId: opts.threadId, userId: opts.user.id, runId: opts.runId, messageGroupId: resumeMessageGroupId, kind: 'suspended', toolCallId: result.suspension.toolCallId, checkpointKey: result.agentRunId, checkpointTaskId: opts.checkpoint?.checkpointTaskId, }); // Bill the tokens this segment consumed to reach the suspension. Every // segment of a HITL run shares one agentRunId (resume reuses it), so the // terminal claim would otherwise be the only claim and swallow this usage // — worse, a stop while suspended never reaches a terminal claim at all. // Key on the per-suspension requestId so this never collides with the // terminal claim (bare agentRunId) or another suspension. void this.creditService.claimRunUsage( opts.user, opts.threadId, `${result.agentRunId || opts.runId}:${result.suspension.requestId}`, result.usage?.usage ?? [], 'suspended', ); } // Track intermediate message (text streamed before suspension) const intermediateText = await (result.text ?? Promise.resolve('')); if (intermediateText) { this.telemetry.track('Builder sent message', { thread_id: opts.threadId, message: redactTelemetryText(intermediateText), is_intermediate: true, }); } const messageGroupId = this.tracing.getMessageGroupId(opts.runId); const waitingDecision = await this.terminalOutcome.evaluateWaitingResponse( opts.threadId, opts.runId, result.confirmationEvent, { messageGroupId }, ); if (waitingDecision?.reason === 'confirmation-invalid') { messageTraceFinalization = await this.terminalOutcome.finishInvalidConfirmationRun({ threadId: opts.threadId, runId: opts.runId, abortController: opts.abortController, tracing: opts.tracing, }); return; } if (result.confirmationEvent) { this.trackConfirmationRequest(opts.user.id, opts.threadId, result.confirmationEvent); this.eventBus.publish(opts.threadId, result.confirmationEvent); } const suspensionOutputs = buildSuspensionTraceOutputs(opts.runId, result.suspension); await this.tracing.finalizeRunTracing(opts.runId, opts.tracing, { status: 'suspended', outputs: suspensionOutputs, metadata: { completion_source: 'orchestrator', ...(result.suspension?.requestId ? { request_id: result.suspension.requestId } : {}), ...(result.suspension?.toolCallId ? { pending_tool_call_id: result.suspension.toolCallId } : {}), ...(result.suspension?.toolName ? { pending_tool_name: result.suspension.toolName } : {}), }, }); messageTraceFinalization = { status: 'suspended', outputs: suspensionOutputs, metadata: { completion_source: 'orchestrator', ...(result.suspension?.requestId ? { request_id: result.suspension.requestId } : {}), ...(result.suspension?.toolCallId ? { pending_tool_call_id: result.suspension.toolCallId } : {}), ...(result.suspension?.toolName ? { pending_tool_name: result.suspension.toolName } : {}), }, }; return; } if (result.status === 'cancelled' && this.shouldPreserveHitlOnShutdown(opts.runId)) { return; } const outputText = await (result.text ?? Promise.resolve('')); resumedRunProducedOutput = outputText.length > 0; const messageGroupId = this.tracing.getMessageGroupId(opts.runId); const terminalError = result.status === 'errored' ? await this.reclassifyMaskedStreamFailure(result.error, opts.user, { threadId: opts.threadId, runId: opts.runId, }) : undefined; if (result.status === 'errored') { this.instanceAiErrorReporter.report( terminalError ?? new Error('Instance AI resumed stream errored'), { component: 'instance-ai-stream', providerStream: true, threadId: opts.threadId, runId: opts.runId, tracing: opts.tracing, agentId: orchestratorAgentId(opts.runId), userId: opts.user.id, messageGroupId, }, ); } // A turn that suspended for a confirmation replays its inline files on resume, // so a refusal here strands the thread exactly as it would on the first run. const attachmentRemoved = this.shouldDropTurnAttachments({ turnHadAttachments: true, producedNoOutput: result.status === 'errored' && outputText.length === 0, }) ? await this.dropTurnAttachments({ threadId: opts.threadId, resourceId: opts.user.id, }) : undefined; const userFacingErrorMessage = result.status === 'errored' ? getUserFacingErrorMessage(terminalError, undefined, { attachmentRemoved }) : undefined; const userFacingErrorCode = result.status === 'errored' ? getUserFacingErrorCode(terminalError) : undefined; if (runControl.shouldEmitTerminalOutcome(result.stopReason)) { await this.terminalOutcome.evaluateTerminalResponse( opts.threadId, opts.runId, result.status, { messageGroupId, workSummary: result.workSummary, errorMessage: userFacingErrorMessage, errorCode: userFacingErrorCode, suppressCompletedFallback: opts.checkpoint?.isCheckpointFollowUp === true || opts.plannedBuild?.isPlannedBuildFollowUp === true, }, ); } const finalStatus = result.status === 'errored' ? 'error' : result.status; await this.tracing.finalizeRunTracing(opts.runId, opts.tracing, { status: finalStatus, outputText, }); messageTraceFinalization = { status: finalStatus, outputText, metadata: await this.tracing.buildMessageTraceMetadata(opts.threadId, opts.runId, { status: finalStatus, }), }; const archivedWorkflowIds = await this.temporaryWorkflowService.reapForRun( opts.threadId, opts.user, undefined, this.backgroundTasks.getRunningTasks(opts.threadId).length, ); await this.finalizeRun(opts.threadId, opts.runId, result.status, { userId: opts.user.id, // Forward modelId so title refinement fires on the resume path too — a run // that suspends for HITL and completes here would otherwise never be titled. ...(opts.modelId !== undefined ? { modelId: opts.modelId } : {}), archivedWorkflowIds, workSummary: result.workSummary, usage: result.usage, errorReason: userFacingErrorMessage, ...(result.status === 'errored' ? { errorInfo: { errorMessage: terminalError ? getErrorMessage(terminalError) : 'Instance AI resumed stream errored', errorSource: 'stream' as const, }, } : {}), }); // Bill token usage for every terminal outcome, deduped per run segment. await this.creditService.claimRunUsage( opts.user, opts.threadId, result.agentRunId || opts.runId, result.usage?.usage ?? [], result.status, ); if (result.status === 'completed') { completedSetupWorkflowId = this.getWorkflowSetupSuspensionWorkflowId( opts.toolName, opts.suspendPayload, ); this.telemetry.track('Builder sent message', { thread_id: opts.threadId, message: redactTelemetryText(outputText), }); this.telemetry.track('Builder satisfied user intent', { thread_id: opts.threadId, }); } } catch (error) { if (!resumeClaimed) { skipPostRunCleanup = true; await this.settleUnclaimedResume(opts, error, 'exception'); return; } if (opts.signal.aborted) { if (this.shouldPreserveHitlOnShutdown(opts.runId)) { return; } const messageGroupId = this.tracing.getMessageGroupId(opts.runId); const runTimeout = this.liveness.consumeRunTimeout(opts.runId); const cancellationReason = runTimeout.timedOut ? INSTANCE_AI_RUN_TIMEOUT_REASON : getAbortReason(opts.signal); if (cancellationReason === INSTANCE_AI_RUN_TIMEOUT_REASON) { this.liveness.publishRunTimeoutNotice(opts.threadId, opts.runId); } await this.terminalOutcome.evaluateTerminalResponse( opts.threadId, opts.runId, 'cancelled', { messageGroupId, }, ); await this.tracing.finalizeRunTracing(opts.runId, opts.tracing, { status: 'cancelled', reason: cancellationReason, }); messageTraceFinalization = { status: 'cancelled', reason: cancellationReason, metadata: await this.tracing.buildMessageTraceMetadata(opts.threadId, opts.runId, { status: 'cancelled', cancellationReason, runTimeout, }), }; const archivedWorkflowIds = await this.temporaryWorkflowService.reapForRun( opts.threadId, opts.user, undefined, this.backgroundTasks.getRunningTasks(opts.threadId).length, ); this.publishRunFinish( opts.threadId, opts.runId, 'cancelled', cancellationReason, archivedWorkflowIds, opts.user.id, ); return; } const terminalError = await this.reclassifyMaskedStreamFailure(error, opts.user, { threadId: opts.threadId, runId: opts.runId, }); // Same reasoning as the resumed errored-result path above: a suspended // file-bearing turn replays its attachments, so a thrown refusal here would // leave them in history too. Gated on the run having produced nothing, // because this catch also covers post-result finalization — a failure there // follows a good turn whose attachments must be left alone. const attachmentRemoved = this.shouldDropTurnAttachments({ turnHadAttachments: true, producedNoOutput: !resumedRunProducedOutput, }) ? await this.dropTurnAttachments({ threadId: opts.threadId, resourceId: opts.user.id, }) : undefined; const errorMessage = getErrorMessage(terminalError); const userFacingErrorMessage = getUserFacingErrorMessage(terminalError, undefined, { attachmentRemoved, }); const userFacingErrorCode = getUserFacingErrorCode(terminalError); const messageGroupId = this.tracing.getMessageGroupId(opts.runId); const errCtx: InstanceAiObservabilityContext = { threadId: opts.threadId, runId: opts.runId, tracing: opts.tracing, agentId: orchestratorAgentId(opts.runId), userId: opts.user.id, messageGroupId, }; this.logger.error(`Instance AI resumed run error: ${errorMessage}`, { error: errorMessage, ...buildInstanceAiObservabilityContext(errCtx), }); this.instanceAiErrorReporter.report(terminalError, { component: 'instance-ai-run', ...errCtx, }); await this.terminalOutcome.evaluateTerminalResponse(opts.threadId, opts.runId, 'errored', { messageGroupId, errorMessage: userFacingErrorMessage, errorCode: userFacingErrorCode, }); await this.tracing.finalizeRunTracing(opts.runId, opts.tracing, { status: 'error', reason: errorMessage, }); messageTraceFinalization = { status: 'error', reason: errorMessage, metadata: await this.tracing.buildMessageTraceMetadata(opts.threadId, opts.runId, { status: 'error', }), }; const archivedWorkflowIds = await this.temporaryWorkflowService.reapForRun( opts.threadId, opts.user, undefined, this.backgroundTasks.getRunningTasks(opts.threadId).length, ); this.publishRunFinish( opts.threadId, opts.runId, 'errored', userFacingErrorMessage, archivedWorkflowIds, opts.user.id, { errorMessage, errorSource: 'exception' }, ); } finally { this.runState.clearActiveRun(opts.threadId, opts.resumeExecutionToken); const segmentSuspended = messageTraceFinalization?.status === 'suspended'; // See note in executeRun's finally — keep threadPushRef alive for // post-run planned-task dispatch. if (!skipPostRunCleanup && messageTraceFinalization) { if (opts.tracing) { await this.tracing.finalizeMessageTraceRoot( opts.runId, opts.tracing, messageTraceFinalization, ); } if (messageTraceFinalization.status !== 'cancelled' && !segmentSuspended) { this.liveness.consumeRunTimeout(opts.runId); } } // Resumed runs are user-driven, so they never extend the failure // streak — but a healthy one resets it before the reschedule below. if (!skipPostRunCleanup) { this.updateInternalFollowUpFailureStreak( opts.threadId, messageTraceFinalization?.status, false, ); } // Post-run planned-task wiring — mirror the executeRun finally. // Resumed ordinary-chat runs also need to drive the scheduler in case // a background task settled while they were active or suspended and // the orchestrate-checkpoint branch was skipped because of hasLiveRun. if ( !skipPostRunCleanup && !segmentSuspended && !this.runState.hasSuspendedRun(opts.threadId) ) { const reschedule = !opts.signal.aborted; if (opts.checkpoint?.isCheckpointFollowUp) { await this.finalizeCheckpointFollowUp( opts.user, opts.threadId, opts.checkpoint.checkpointTaskId, { reschedule }, ); } else if (opts.plannedBuild?.isPlannedBuildFollowUp) { await this.finalizePlannedBuildFollowUp(opts.user, opts.threadId, opts.plannedBuild, { reschedule, }); } else if (reschedule) { await this.schedulePlannedTasks(opts.user, opts.threadId); } if (reschedule) { await this.drainPendingCheckpointReentries(opts.user, opts.threadId); } // The setup claim must land even on a stop: it is what stops a later // run from re-routing setup for a workflow the user already set up. if (completedSetupWorkflowId) { await this.markWorkflowSetupHandled(opts.threadId, completedSetupWorkflowId, opts.runId); } await this.taskProjector.syncFromWorkflowLoop(opts.threadId, opts.runId); if (reschedule) { await this.maybeStartWorkflowSetupFollowUp(opts.user, opts.threadId); } } if (errorReporterExecutionToken) { this.instanceAiErrorReporter.endRun(opts.runId, errorReporterExecutionToken); } } } private async settleUnclaimedResume( opts: UnclaimedResumeContext, error: unknown, errorSource: NonNullable, ): Promise { const outcome = classifyUnclaimedResume(error, { aborted: opts.signal.aborted, preserveHitl: this.shouldPreserveHitlOnShutdown(opts.runId), }); switch (outcome.kind) { case 'preserve-hitl': return; case 'stale': await this.tracing.finalizeDetachedTraceRun( `stale-resume:${opts.runId}`, opts.unregisteredResumeTracing, { status: 'cancelled', outputs: { runId: opts.runId }, metadata: { completion_source: 'stale_resume' }, }, ); return; case 'cancelled': { const reason = this.liveness.consumeRunTimeout(opts.runId).timedOut ? INSTANCE_AI_RUN_TIMEOUT_REASON : getAbortReason(opts.signal); await this.tracing.finalizeDetachedTraceRun( `cancelled-resume:${opts.runId}`, opts.unregisteredResumeTracing, { status: 'cancelled', outputs: { runId: opts.runId }, metadata: { completion_source: 'resume_cancelled', cancellation_reason: reason }, }, ); await this.emitTerminalRun({ threadId: opts.threadId, runId: opts.runId, status: 'cancelled', reason, messageGroupId: opts.messageGroupId, user: opts.user, }); return; } case 'errored': this.instanceAiErrorReporter.report(error, { component: 'instance-ai-resume-claim', threadId: opts.threadId, runId: opts.runId, tracing: opts.tracing, agentId: orchestratorAgentId(opts.runId), userId: opts.user.id, messageGroupId: opts.messageGroupId, }); await this.tracing.finalizeDetachedTraceRun( `unclaimed-resume:${opts.runId}`, opts.unregisteredResumeTracing, { status: 'failed', error: outcome.errorMessage, metadata: { completion_source: 'resume_claim' }, }, ); await this.emitTerminalRun({ threadId: opts.threadId, runId: opts.runId, status: 'errored', reason: outcome.reason, errorCode: outcome.errorCode, errorInfo: { errorMessage: outcome.errorMessage, errorSource }, // The resume trace context is never registered before the claim, so the // message group has to come from the suspended run, not the trace registry. messageGroupId: opts.messageGroupId, user: opts.user, }); return; default: assertNever(outcome); } } /** * Terminalizes a run that ended in a stop or a failure. `publishRunFinish` is * the one step that has to land — without it the chat hangs forever — so every * DB-touching step around it is best-effort. */ private async emitTerminalRun(args: { threadId: string; runId: string; status: 'cancelled' | 'errored'; reason: string; errorCode?: TerminalErrorCode; errorInfo?: RunFinishErrorInfo; messageGroupId?: string; user: User; }): Promise { const { threadId, runId, status } = args; const context = { threadId, runId }; await this.bestEffort( 'Failed to evaluate the terminal response for a settling run', context, async () => await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, status, { messageGroupId: args.messageGroupId, ...(status === 'errored' ? { errorMessage: args.reason, errorCode: args.errorCode } : {}), }), ); const archivedWorkflowIds = (await this.bestEffort( 'Failed to reap temporary workflows for a settling run', context, async () => await this.temporaryWorkflowService.reapForRun( threadId, args.user, undefined, this.backgroundTasks.getRunningTasks(threadId).length, ), )) ?? []; this.publishRunFinish( threadId, runId, status, args.reason, archivedWorkflowIds, args.user.id, args.errorInfo, ); } private async bestEffort( failureMessage: string, context: { threadId: string; runId: string }, step: () => T | Promise, ): Promise | undefined> { try { return await step(); } catch (error) { this.logger.warn(failureMessage, { ...context, error: getErrorMessage(error) }); return undefined; } } // ── Background task management ────────────────────────────────────────── private spawnBackgroundTask( runId: string, opts: SpawnBackgroundTaskOptions, messageGroupIdOverride?: string, ): SpawnBackgroundTaskResult { const outcome = this.backgroundTasks.spawn({ taskId: opts.taskId, threadId: opts.threadId, runId, role: opts.role, agentId: opts.agentId, messageGroupId: messageGroupIdOverride ?? this.runState.getMessageGroupId(opts.threadId), plannedTaskId: opts.plannedTaskId, workItemId: opts.workItemId, traceContext: opts.traceContext, createTraceContext: opts.createTraceContext, dedupeKey: opts.dedupeKey, parentCheckpointId: opts.parentCheckpointId, run: opts.run, onLimitReached: async (errorMessage) => { await this.tracing.finalizeDetachedTraceRun(opts.taskId, opts.traceContext, { status: 'failed', outputs: { taskId: opts.taskId, agentId: opts.agentId, role: opts.role, }, error: errorMessage, metadata: { ...(opts.plannedTaskId ? { planned_task_id: opts.plannedTaskId } : {}), ...(opts.workItemId ? { work_item_id: opts.workItemId } : {}), }, }); this.eventBus.publish(opts.threadId, { type: 'agent-completed', runId, agentId: opts.agentId, payload: { role: opts.role, result: '', error: errorMessage, }, }); }, onCompleted: async (task) => { await this.tracing.finalizeBackgroundTaskTracing(task, 'completed'); this.eventBus.publish(opts.threadId, { type: 'agent-completed', runId, agentId: opts.agentId, payload: { role: opts.role, result: task.result ?? '' }, }); const user = this.runState.getThreadUser(opts.threadId); if (user) { await this.handlePlannedTaskSettlement(user, task, 'succeeded'); } }, onFailed: async (task) => { await this.tracing.finalizeBackgroundTaskTracing(task, 'failed'); this.instanceAiErrorReporter.report( new Error(task.error ?? 'Instance AI background task failed'), { component: 'instance-ai-background-task', threadId: opts.threadId, runId, tracing: task.traceContext, agentId: opts.agentId, messageGroupId: task.messageGroupId, taskId: task.taskId, role: task.role, }, ); this.eventBus.publish(opts.threadId, { type: 'agent-completed', runId, agentId: opts.agentId, payload: { role: opts.role, result: '', error: task.error ?? 'Unknown error' }, }); const user = this.runState.getThreadUser(opts.threadId); if (user) { await this.handlePlannedTaskSettlement(user, task, 'failed'); } }, onSettled: async (task) => { await this.terminalOutcome.recordBackgroundTerminalOutcome(task); // Auto-follow-up: when the last background task finishes and no // orchestrator run is active, resume the orchestrator so it can // synthesize results for the user. Planned tasks handle this via // schedulePlannedTasks(); this covers direct detached background calls. if (task.plannedTaskId) return; await this.taskProjector.syncFromBackgroundTask(task); // Parent-tagged children (patch-builder etc. spawned inside a // checkpoint follow-up) must NEVER emit a generic // `` shell — the orchestrator would // land outside the checkpoint context and the checkpoint would // be orphaned. Try immediate re-entry; if the run state or // still-running siblings block it, queue a deferred marker that // the post-run drain hook will pick up. const parentCheckpointId = task.parentCheckpointId; if (parentCheckpointId) { const user = this.runState.getThreadUser(opts.threadId); if (!user) { this.queuePendingCheckpointReentry(opts.threadId, parentCheckpointId); return; } const reentered = await this.maybeReenterParentCheckpoint(user, opts.threadId, task); if (!reentered) { this.queuePendingCheckpointReentry(opts.threadId, parentCheckpointId); } return; } const remaining = this.backgroundTasks.getRunningTasks(opts.threadId); const hasActiveRun = !!this.runState.getActiveRunId(opts.threadId); const hasSuspendedRun = this.runState.hasSuspendedRun(opts.threadId); if (remaining.length === 0 && !hasActiveRun && !hasSuspendedRun) { if (this.liveness.hasTimedOutActiveRunThread(opts.threadId)) { this.logger.debug('Skipping background auto-follow-up after active run timeout', { threadId: opts.threadId, taskId: task.taskId, }); return; } // Don't auto-respawn a task that timed out — hand back to the user instead. if (task.timeoutReason) { this.logger.debug('Skipping background auto-follow-up after task timeout', { threadId: opts.threadId, taskId: task.taskId, timeoutReason: task.timeoutReason, }); return; } const user = this.runState.getThreadUser(opts.threadId); if (user) { const verificationFollowUpStarted = await this.maybeStartWorkflowVerificationFollowUp( user, task, ); if (verificationFollowUpStarted) return; // Builder already verified (or setup is needed without a verify step): // route directly to setup so an unrunnable workflow is never presented // as done. const setupFollowUpStarted = await this.maybeStartWorkflowSetupFollowUp( user, task.threadId, ); if (setupFollowUpStarted) return; const payload = JSON.stringify( { role: opts.role, status: task.result ? 'completed' : task.error ? 'failed' : 'finished', result: task.result ?? undefined, outcome: task.outcome ?? undefined, error: task.error ?? undefined, }, null, 2, ); await this.startInternalFollowUpRun( user, opts.threadId, `\n${payload}\n\n\n${AUTO_FOLLOW_UP_MESSAGE}`, task.messageGroupId, ); } } }, }); if (outcome.status === 'started') { void this.taskProjector.syncFromBackgroundTask(outcome.task); return { status: 'started', taskId: outcome.task.taskId, agentId: outcome.task.agentId }; } if (outcome.status === 'duplicate') { this.logger.warn('Background task dispatch deduped — task already in flight', { threadId: opts.threadId, requestedTaskId: opts.taskId, existingTaskId: outcome.existing.taskId, plannedTaskId: opts.dedupeKey?.plannedTaskId, workflowId: opts.dedupeKey?.workflowId, role: opts.role, }); // The sub-agent dispatch tools publish `agent-spawned` and allocate a // detached LangSmith trace root BEFORE calling spawnBackgroundTask, so // the freshly-generated subAgentId for this deduped attempt already has // a phantom sub-agent node in the event stream and an unfinished trace // root. Compensate the same way `onLimitReached` does so the agent tree // snapshot doesn't keep a ghost child and the trace client is released. void this.tracing.finalizeDetachedTraceRun(opts.taskId, opts.traceContext, { status: 'cancelled', outputs: { taskId: opts.taskId, agentId: opts.agentId, role: opts.role, deduped_to: outcome.existing.taskId, }, metadata: { deduped: true, existing_task_id: outcome.existing.taskId, ...(opts.plannedTaskId ? { planned_task_id: opts.plannedTaskId } : {}), ...(opts.workItemId ? { work_item_id: opts.workItemId } : {}), }, }); this.eventBus.publish(opts.threadId, { type: 'agent-completed', runId, agentId: opts.agentId, payload: { role: opts.role, result: '', error: `Deduped: task already in flight as ${outcome.existing.taskId}`, }, }); return { status: 'duplicate', existing: { taskId: outcome.existing.taskId, agentId: outcome.existing.agentId, role: outcome.existing.role, plannedTaskId: outcome.existing.plannedTaskId, workItemId: outcome.existing.workItemId, }, }; } return { status: 'limit-reached' }; } /** Snapshot every Agent the editor attached, as it stands before this turn * acts on it. Best-effort: it only feeds eval-seed authoring. */ private async snapshotAttachedAgents( attachments: InstanceAiResourceAttachment[], orchestrationContext: OrchestrationContext, tracing: InstanceAiTraceContext | undefined, ): Promise { const delegate = orchestrationContext.domainContext?.builderDelegate; if (!delegate || !tracing) return; for (const attachment of attachments) { if (attachment.type !== 'agent') continue; let artifact: AgentSnapshotArtifact | null = null; try { artifact = (await delegate.readAgentArtifact?.(attachment.id)) ?? null; } catch (error) { this.logger.debug( `[agent-snapshot] attached read for ${attachment.id} failed: ${error instanceof Error ? error.message : String(error)}`, ); continue; } if (!artifact) continue; await emitAgentSnapshotTraceEvent(tracing, { agentId: attachment.id, projectId: attachment.projectId, reason: 'attached', artifact, logger: this.logger, }); } } private async buildMessageWithRunningTasks(threadId: string, message: string): Promise { return await enrichMessageWithBackgroundTasks( message, this.backgroundTasks.getRunningTasks(threadId), { formatTask: async (task: ManagedBackgroundTask) => `[Running task — ${task.role}]: taskId=${task.taskId}`, }, ); } private trackConfirmationRequest( userId: string, threadId: string, confirmationEvent: { payload: Record }, ): void { const payload = confirmationEvent.payload; const inputThreadId = nanoid(); payload.inputThreadId = inputThreadId; const inputType = payload.inputType as string | undefined; const mcpConnectServers = mcpConnectRequestSchema.safeParse(payload.mcpConnectRequest).data ?.servers; let type: string; if (inputType) { type = inputType; } else if (Array.isArray(payload.setupRequests) && payload.setupRequests.length > 0) { type = 'setup'; } else if (Array.isArray(payload.credentialRequests) && payload.credentialRequests.length > 0) { type = 'credential-setup'; } else if (mcpConnectServers) { type = 'mcp-connect'; } else { type = 'approval'; } let numSteps = 1; if (Array.isArray(payload.questions)) { numSteps = payload.questions.length; } else if (Array.isArray(payload.setupRequests)) { numSteps = payload.setupRequests.length; } else if (Array.isArray(payload.credentialRequests)) { numSteps = payload.credentialRequests.length; } else if (mcpConnectServers) { numSteps = mcpConnectServers.length; } if (inputType === 'plan-review') { // Tell the first plan in a thread apart from later revisions the user asked // for. The per-thread counter is cleared on thread cleanup. const planCount = (this.planRequestsByThread.get(threadId) ?? 0) + 1; this.planRequestsByThread.set(threadId, planCount); type = planCount === 1 ? 'first_plan' : 'revised_plan'; if (Array.isArray(payload.planItems)) { numSteps = payload.planItems.length; } } const credentialRequests = ( Array.isArray(payload.credentialRequests) ? payload.credentialRequests : [] ).filter( (request): request is { credentialType?: unknown; setupHint?: unknown } => typeof request === 'object' && request !== null, ); // Whether any requested credential is a recipe-driven Templated Custom Auth // one (recipe-seeded modal), to compare completion against plain types. const containsTemplatedCred = credentialRequests.some( (request) => request.credentialType === TEMPLATED_CUSTOM_AUTH_CREDENTIAL_TYPE || (request.setupHint !== null && request.setupHint !== undefined), ); this.telemetry.track('Builder asked for input', { user_id: userId, thread_id: threadId, input_thread_id: inputThreadId, type, num_steps: numSteps, contains_templated_cred: containsTemplatedCred, }); // Recipe content at spec time — production visibility into template and // link quality (the offline eval suite grades the same fields). One event // per recipe; secret-free by construction: recipes are agent-authored // before any user input. for (const request of credentialRequests) { const parsedHint = credentialSetupHintSchema.safeParse(request.setupHint); if (!parsedHint.success) continue; const hint = parsedHint.data; this.telemetry.track(TELEMETRY_EVENT.INSTANCE_AI.BUILDER_SPECCED_TEMPLATED_CRED, { thread_id: threadId, input_thread_id: inputThreadId, template: hint.template, placeholders: hint.placeholders, test_url: hint.testUrl, docs_url: hint.docsUrl, service_host: hint.serviceHost, accepted_status_codes: hint.acceptedStatusCodes, }); } } private async finalizeCancelledSuspendedRun( suspended: SuspendedRunState, reason = 'user_cancelled', ): Promise { const runTimeout = reason === INSTANCE_AI_RUN_TIMEOUT_REASON ? this.liveness.consumeRunTimeout(suspended.runId) : undefined; if (reason === INSTANCE_AI_RUN_TIMEOUT_REASON) { this.liveness.publishRunTimeoutNotice(suspended.threadId, suspended.runId); } await this.tracing.finalizeRunTracing(suspended.runId, suspended.tracing, { status: 'cancelled', reason, }); const archivedWorkflowIds = await this.temporaryWorkflowService.reapForRun( suspended.threadId, suspended.user, undefined, this.backgroundTasks.getRunningTasks(suspended.threadId).length, ); this.publishRunFinish( suspended.threadId, suspended.runId, 'cancelled', reason, archivedWorkflowIds, suspended.user.id, ); await this.tracing.maybeFinalizeRunTraceRoot(suspended.runId, { status: 'cancelled', reason, metadata: await this.tracing.buildMessageTraceMetadata(suspended.threadId, suspended.runId, { status: 'cancelled', cancellationReason: reason, ...(runTimeout ? { runTimeout } : {}), }), }); void this.suspendedThreads.dropPendingConfirmation(suspended.requestId); } private publishRunFinish( threadId: string, runId: string, status: 'completed' | 'cancelled' | 'errored', reason?: string, archivedWorkflowIds?: string[], userId?: string, errorInfo?: RunFinishErrorInfo, ): void { const effectiveStatus = status === 'errored' ? 'error' : status; const hasArchived = archivedWorkflowIds && archivedWorkflowIds.length > 0; this.eventBus.publish(threadId, { type: 'run-finish', runId, agentId: orchestratorAgentId(runId), payload: { status: effectiveStatus, ...(status === 'cancelled' ? { reason: reason ?? 'user_cancelled' } : status === 'errored' && reason ? { reason } : {}), ...(hasArchived ? { archivedWorkflowIds } : {}), }, }); // success-drop heartbeat; user_id required or PostHog drops instance-only events this.telemetry.track('instance_ai_run_finished', { thread_id: threadId, run_id: runId, status: effectiveStatus, ...(userId ? { user_id: userId } : {}), }); this.emitBrowserCredentialSetupOutcomes(threadId, runId, status, reason); if (status === 'errored') { this.telemetry.track('Builder generation errored', { thread_id: threadId, run_id: runId, error_message: redactTelemetryText(errorInfo?.errorMessage ?? reason ?? 'unknown'), ...(errorInfo?.errorSource ? { error_source: errorInfo.errorSource } : {}), ...(userId ? { user_id: userId } : {}), }); } } /** * Per-run bookkeeping behind `context.browserCredentialSetup` (NODE-5511). * `markPending` opens an attempt when the user clicks auto-setup on the * credential card; `markCreated`/`markCreateFailed` resolve the latest open * attempt for the type — or open an implicit `'conversation'` attempt when * there is none, so LLM-initiated creations (user asked directly in chat, * no setup card involved) still produce a terminal telemetry event. */ private createBrowserCredentialSetupTracker( runId: string, userId: string, ): NonNullable { const getOrCreateAttempts = () => { let pending = this.pendingBrowserCredentialSetups.get(runId); if (!pending) { pending = { userId, attempts: [] }; this.pendingBrowserCredentialSetups.set(runId, pending); } return pending.attempts; }; const resolveOpenAttempt = (credentialType: string) => { const attempts = getOrCreateAttempts(); let attempt = attempts.findLast((a) => a.credentialType === credentialType && !a.created); if (!attempt) { attempt = { credentialType, setupMethod: 'conversation', startedAt: Date.now(), created: false, }; attempts.push(attempt); } return attempt; }; return { markPending: (credentialType: string, attemptId?: string) => { const attempts = getOrCreateAttempts(); if (attemptId && attempts.some((attempt) => attempt.attemptId === attemptId)) { return; } attempts.push({ credentialType, setupMethod: 'setup_card', attemptId, startedAt: Date.now(), created: false, }); }, markCreated: (credentialType: string) => { const attempt = resolveOpenAttempt(credentialType); attempt.created = true; attempt.errorCode = undefined; }, markCreateFailed: (credentialType: string, errorCode: string) => { resolveOpenAttempt(credentialType).errorCode = errorCode; }, }; } /** * Emit one terminal telemetry event per browser-assisted credential setup * attempt of the finished run (NODE-5511). Consumes the pending record so * every attempt yields exactly one success or failure event. When the flow * itself never failed, the run's own termination (user stop, timeout, * stream error) is reported as the error code so aborts aren't counted as * flow failures. */ private emitBrowserCredentialSetupOutcomes( threadId: string, runId: string, runStatus: 'completed' | 'cancelled' | 'errored', runFinishReason?: string, ): void { const browserSetup = this.pendingBrowserCredentialSetups.get(runId); if (!browserSetup) return; this.pendingBrowserCredentialSetups.delete(runId); const terminalErrorCode = runStatus === 'completed' ? 'not_attempted' : runStatus === 'errored' ? 'run_errored' : runFinishReason === INSTANCE_AI_RUN_TIMEOUT_REASON ? 'run_timed_out' : 'run_cancelled'; for (const attempt of browserSetup.attempts) { this.telemetry.track('Instance AI Browser Use credential setup completed', { user_id: browserSetup.userId, credential_type: attempt.credentialType, status: attempt.created ? 'success' : 'failure', ...(attempt.created ? {} : { failure_stage: failureStageForErrorCode(attempt.errorCode), error_code: attempt.errorCode ?? terminalErrorCode, }), // The flow never runs a credential test, so validation support is unknown. is_valid: null, is_new: true, setup_method: attempt.setupMethod, thread_id: threadId, run_id: runId, ...(attempt.attemptId ? { credential_setup_attempt_id: attempt.attemptId } : {}), duration_ms: Date.now() - attempt.startedAt, }); } } private async finalizeRun( threadId: string, runId: string, status: 'completed' | 'cancelled' | 'errored', options?: { userId?: string; modelId?: ModelConfig; archivedWorkflowIds?: string[]; workSummary?: WorkSummary; usage?: RunTokenUsage; errorReason?: string; errorInfo?: RunFinishErrorInfo; }, ): Promise { this.publishRunFinish( threadId, runId, status, options?.errorReason, options?.archivedWorkflowIds, options?.userId, options?.errorInfo, ); this.emitRunMetrics(threadId, status, options); if (status === 'completed' && options?.userId && options?.modelId) { void this.refineTitleIfNeeded(threadId, options.userId, options.modelId); } } /** Emit a typed event consumed by the Prometheus Instance AI metrics collector. */ private emitRunMetrics( threadId: string, status: 'completed' | 'cancelled' | 'errored' | 'suspended', options?: { modelId?: ModelConfig; workSummary?: WorkSummary; usage?: RunTokenUsage }, ): void { const startedAt = this.runState.getActiveRun(threadId)?.startedAt; this.eventService.emit('instance-ai-run-finished', { status: status === 'errored' ? 'error' : status, // Duration is reported once, by the run's terminal event. durationMs: status !== 'suspended' && startedAt !== undefined ? Date.now() - startedAt : undefined, model: runMetricsModelLabel(options?.modelId), toolCalls: options?.workSummary?.totalToolCalls ?? 0, toolErrors: options?.workSummary?.totalToolErrors ?? 0, ...(options?.usage ? { usage: options.usage } : {}), }); } /** * Refine the thread title with an LLM-generated version after a run completes. * Fires asynchronously and is best-effort — the heuristic title remains if this fails. */ private async refineTitleIfNeeded( threadId: string, userId: string, modelId: ModelConfig, ): Promise { try { const memory = this.agentMemory; const thread = await memory.getThread(threadId); if (!thread?.title) return; // Skip if thread already has an LLM-refined title if (thread.metadata?.titleRefined) return; // Title the conversation from its opening user turns. const history = await memory.getMessages(threadId, { limit: TITLE_REFINE_HISTORY_LIMIT }); const userTexts: string[] = []; for (const m of history) { if (!('role' in m) || m.role !== 'user') continue; // Stored user messages carry service-injected blocks (, // task context). Strip them or a trivial "hey" looks substantial enough to // title, and the injected blocks leak into the title prompt. const text = cleanStoredUserMessage(this.extractStoredMessageText(m.content)); if (text && text.length > 0) userTexts.push(text); if (userTexts.length >= 5) break; } if (userTexts.length === 0) return; const userText = userTexts.join('\n'); const baseTracing = this.tracing.getTraceContextForContinuation(threadId); const titleTracing = await createInternalOperationTraceContext({ threadId, conversationId: threadId, messageId: `internal:title:${threadId}`, runId: `title-${nanoid()}`, userId, modelId, operationName: 'thread_title', input: { message_count: userTexts.length, source: 'thread_title_refinement', }, proxyConfig: baseTracing?.proxyConfig, metadata: { n8n_version: N8N_VERSION || undefined, operation_name: 'thread_title', trigger: 'run_completed', }, }); const titleTelemetry = titleTracing?.getTelemetry?.({ agentRole: 'thread_title', functionId: 'instance-ai.thread_title', executionMode: 'internal', metadata: { operation_name: 'thread_title', }, }); let llmTitle: string | null; if (titleTracing) { try { llmTitle = await titleTracing.withActiveSpan(titleTracing.rootRun, async () => { const title = await generateTitleForRun(modelId, userText, { ...(titleTelemetry ? { telemetry: titleTelemetry } : {}), }); if (title) { await titleTracing.finishRun(titleTracing.rootRun, { outputs: { title }, metadata: { final_status: 'completed' }, }); } else { await titleTracing.finishRun(titleTracing.rootRun, { outputs: { title: null }, metadata: { final_status: 'skipped' }, }); } return title; }); } finally { releaseTraceClient(titleTracing.rootRun.traceId); } } else { llmTitle = await generateTitleForRun(modelId, userText); } if (!llmTitle) return; await patchThread(memory, { threadId, update: ({ metadata }) => ({ title: llmTitle, metadata: { ...metadata, titleRefined: true }, }), }); // Push SSE event so frontend updates immediately this.eventBus.publish(threadId, { type: 'thread-title-updated', runId: '', agentId: 'orchestrator', payload: { title: llmTitle }, }); } catch (error) { this.logger.warn('Failed to refine thread title', { threadId, error: getErrorMessage(error), }); // Non-fatal — heuristic title remains } } private extractStoredMessageText(content: unknown): string { if (typeof content === 'string') return content; if (Array.isArray(content)) { return content.flatMap((part) => (isTextMessagePart(part) ? [part.text] : [])).join('\n'); } return ''; } /** * Read-own-writes barrier for run-scoped reads: settle the thread's drain * (including open coalesce buffers) so everything published before this call * is visible, then read the log. Every caller is a run boundary — * terminal-guard inputs, trace metadata, snapshot builds — where closing the * open segment early is correct anyway. */ private async readRunEvents(threadId: string, runIds: string[]): Promise { await this.eventLog.flush(threadId); return await this.eventLog.getEventsForRuns(threadId, runIds); } private parseMcpServers(raw: string): McpServerConfig[] { if (!raw.trim()) return []; return raw.split(',').map((entry) => { const [name, url] = entry.trim().split('='); return { name: name.trim(), url: url?.trim() }; }); } private trackMcpToolCall({ server, toolName, success, }: { server: McpServerConfig; toolName: string; success: boolean; }): void { const serverSlug = server.metadata?.serverSlug; const userId = server.metadata?.userId; const connectionId = server.metadata?.connectionId; if (serverSlug && userId) { this.telemetry.track('Instance AI mcp tool called', { user_id: userId, server_slug: serverSlug, tool_name: toolName, success, }); } if (!success && connectionId && userId) { this.push.sendToUsers({ type: 'instanceAiMcpToolCallFailed', data: { connectionId } }, [ userId, ]); } } }