import {
	UNLIMITED_CREDITS,
	applyBranchReadOnlyOverrides,
	buildProxyHeaders,
	type InstanceAiAttachment,
	type InstanceAiAgentNode,
	type InstanceAiConfirmRequest,
	type InstanceAiEvent,
	type InstanceAiThreadStatusResponse,
	type InstanceAiGatewayCapabilities,
	type McpToolCallResult,
	type ToolCategory,
	type TaskList,
} from '@n8n/api-types';
import type { Message, Workspace } from '@n8n/agents';
import { Logger } from '@n8n/backend-common';
import { GlobalConfig, SsrfProtectionConfig, type InstanceAiConfig } from '@n8n/config';
import { OnLeaderStepdown, OnLeaderTakeover } from '@n8n/decorators';
import { ErrorReporter, InstanceSettings } from 'n8n-core';

import { SsrfProtectionService } from '@/services/ssrf/ssrf-protection.service';
import { AiBuilderTemporaryWorkflowRepository, UserRepository, type User } from '@n8n/db';
import { Service } from '@n8n/di';
import { UrlService } from '@/services/url.service';
import {
	MAX_STEPS,
	createInstanceAgent,
	createAllTools,
	createSandbox,
	createWorkspace,
	createLazyRuntimeWorkspace,
	createLazyWorkspaceRuntimeSkillSource,
	setupSandboxWorkspace,
	loadInstanceAiRuntimeSkillSource,
	createInstanceAiTraceContext,
	createInternalOperationTraceContext,
	continueInstanceAiTraceContext,
	createInstanceAiLivenessPolicyConfig,
	InstanceAiLivenessPolicy,
	McpClientManager,
	createDomainAccessTracker,
	BackgroundTaskManager,
	buildAgentTreeFromEvents,
	classifyAttachments,
	buildAttachmentManifest,
	isParseableAttachment,
	enrichMessageWithBackgroundTasks,
	InstanceAiTerminalResponseGuard,
	PlannedTaskCoordinator,
	PlannedTaskStorage,
	TerminalOutcomeStorage,
	applyPlannedTaskPermissions,
	PLANNED_TASK_PERMISSION_OVERRIDES,
	releaseTraceClient,
	submitLangsmithUserFeedback,
	resumeAgentRun,
	RunStateRegistry,
	startBuildWorkflowAgentTask,
	startDetachedDelegateTask,
	streamAgentRun,
	truncateToTitle,
	generateTitleForRun,
	patchThread,
	type ConfirmationData,
	type BuiltMemory,
	type DomainAccessTracker,
	type InstanceAiContext,
	type ManagedBackgroundTask,
	type McpServerConfig,
	type ModelConfig,
	type OrchestrationContext,
	type InstanceAiTraceContext,
	type PlannedTaskGraph,
	type PlannedTaskRecord,
	type SandboxConfig,
	type SpawnBackgroundTaskOptions,
	type SpawnBackgroundTaskResult,
	type ServiceProxyConfig,
	type StreamableAgent,
	type SuspendedRunState,
	type TerminalOutcome,
	type TerminalResponseDecision,
	type TerminalResponseStatus,
	type WorkSummary,
	WorkflowTaskCoordinator,
	WorkflowLoopStorage,
	ThreadTaskStorage,
} from '@n8n/instance-ai';
import { setSchemaBaseDirs } from '@n8n/workflow-sdk';
import { nanoid } from 'nanoid';
import { OperationalError, UnexpectedError, UserError } from 'n8n-workflow';
import type * as Undici from 'undici';
import { v5 as uuidv5 } from 'uuid';

import { N8N_VERSION, WORKFLOW_SDK_VERSION } from '@/constants';
import { EventService } from '@/events/event.service';
import { SourceControlPreferencesService } from '@/modules/source-control.ee/source-control-preferences.service.ee';
import { AiService } from '@/services/ai.service';
import { Push } from '@/push';
import { Telemetry } from '@/telemetry';
import { InProcessEventBus } from './event-bus/in-process-event-bus';
import type { LocalGateway } from './filesystem';
import { LocalGatewayRegistry } from './filesystem';
import { InstanceAiSettingsService } from './instance-ai-settings.service';
import { InstanceAiAdapterService } from './instance-ai.adapter.service';
import { AUTO_FOLLOW_UP_MESSAGE } from './internal-messages';
import { DbSnapshotStorage } from './storage/db-snapshot-storage';
import { DbIterationLogStorage } from './storage/db-iteration-log-storage';
import { TypeORMAgentCheckpointStore } from './storage/typeorm-agent-checkpoint-store';
import { TypeORMAgentMemory } from './storage/typeorm-agent-memory';
import { ProxyTokenManager } from '@/services/proxy-token-manager';
import { InstanceAiPendingConfirmationRepository } from './repositories/instance-ai-pending-confirmation.repository';
import { InstanceAiThreadRepository } from './repositories/instance-ai-thread.repository';
import { TraceReplayState } from './trace-replay-state';
import { INSTANCE_AI_RUN_TIMEOUT_REASON, InstanceAiLivenessService } from './liveness';
import {
	buildInstanceAiRunTraceMetadata,
	type InstanceAiRunTraceMetadataOptions,
} from './run-trace-metadata';

function getErrorMessage(error: unknown): string {
	return error instanceof Error ? error.message : String(error);
}

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;

/**
 * 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'
	);
}

const ORCHESTRATOR_AGENT_ID = 'agent-001';

type RuntimeSandboxEntry = {
	sandbox: NonNullable<Awaited<ReturnType<typeof createSandbox>>>;
	workspace: NonNullable<ReturnType<typeof createWorkspace>>;
	setupComplete: boolean;
	setupPromise: Promise<void> | undefined;
	expiresAt: number;
	cleanupTimer?: ReturnType<typeof setTimeout>;
};

const SANDBOX_NAME_MAX_LEN = 63;
const SANDBOX_LABEL_MAX_LEN = 63;
const NAME_PREFIX_SLUG_MAX_LEN = 24;
const SHORT_RUN_ID_LEN = 8;
const DEFAULT_SANDBOX_TTL_MS = 15 * 60 * 1000;

function slugifySandboxName(value: string, maxLen: number): string {
	const slug = value
		.toLowerCase()
		.replace(/[^a-z0-9]+/g, '-')
		.replace(/^-+|-+$/g, '');
	return slug.slice(0, maxLen).replace(/-+$/, '');
}

function slugifySandboxLabel(value: string, maxLen: number): string {
	return value
		.replace(/[^A-Za-z0-9_.-]+/g, '-')
		.replace(/^[-.]+|[-.]+$/g, '')
		.slice(0, maxLen)
		.replace(/[-.]+$/, '');
}

function getThreadScopedSandboxName(threadId: string): string {
	return `instance-ai-thread-${threadId}`;
}

function buildThreadScopedSandboxName(
	threadId: string,
	namePrefix: string | undefined,
	runId: string | undefined,
): string {
	const parts: string[] = [];
	if (namePrefix) {
		const prefixSlug = slugifySandboxName(namePrefix, NAME_PREFIX_SLUG_MAX_LEN);
		if (prefixSlug) parts.push(prefixSlug);
	}
	if (runId) {
		const runSlug = slugifySandboxName(runId, SHORT_RUN_ID_LEN);
		if (runSlug) parts.push(runSlug);
	}
	const threadSlug = slugifySandboxName(getThreadScopedSandboxName(threadId), SANDBOX_NAME_MAX_LEN);
	if (threadSlug) parts.push(threadSlug);
	const name = slugifySandboxName(parts.join('-'), SANDBOX_NAME_MAX_LEN);
	if (!name) throw new UnexpectedError('Failed to build thread-scoped sandbox name');
	return name;
}

function buildThreadScopedSandboxLabels(
	threadId: string,
	namePrefix: string | undefined,
	runId: string | undefined,
): Record<string, string> {
	const baseName = getThreadScopedSandboxName(threadId);
	const labels: Record<string, string> = {
		'n8n-builder': slugifySandboxLabel(baseName, SANDBOX_LABEL_MAX_LEN),
		thread_id: slugifySandboxLabel(threadId, SANDBOX_LABEL_MAX_LEN),
	};
	if (namePrefix) labels.name_prefix = slugifySandboxLabel(namePrefix, SANDBOX_LABEL_MAX_LEN);
	if (runId) labels.run_id = slugifySandboxLabel(runId, SANDBOX_LABEL_MAX_LEN);
	return labels;
}

function withThreadScopedSandboxIdentity(
	config: SandboxConfig,
	threadId: string,
	runId?: string,
): SandboxConfig {
	if (!config.enabled || config.provider !== 'daytona') return config;

	const name = buildThreadScopedSandboxName(threadId, config.namePrefix, runId);
	return {
		...config,
		id: name,
		name,
		labels: {
			...buildThreadScopedSandboxLabels(threadId, config.namePrefix, runId),
			...config.labels,
		},
	};
}

function getUserFacingErrorMessage(error: unknown): string {
	if (error instanceof UserError) {
		return error.message;
	}

	if (error instanceof OperationalError) {
		return 'I hit an operational error before I could finish that response. Please try again.';
	}

	if (error instanceof UnexpectedError) {
		return 'Something went wrong before I could finish that response. Please try again.';
	}

	return 'Something went wrong before I could finish that response. Please try again.';
}

function getBackgroundOutcomeResponseId(outcome: TerminalOutcome): string {
	return `background-outcome:${outcome.id}`;
}

function createTerminalOutcomeAgentTree(
	outcome: TerminalOutcome,
	responseId: string,
): InstanceAiAgentNode {
	return {
		agentId: ORCHESTRATOR_AGENT_ID,
		role: 'orchestrator',
		status:
			outcome.status === 'cancelled'
				? 'cancelled'
				: outcome.status === 'failed'
					? 'error'
					: 'completed',
		textContent: outcome.userFacingMessage,
		reasoning: '',
		toolCalls: [],
		children: [],
		timeline: [{ type: 'text', content: outcome.userFacingMessage, responseId }],
	};
}

function appendTerminalOutcomeToAgentTree(
	tree: InstanceAiAgentNode,
	outcome: TerminalOutcome,
	responseId: string,
): { tree: InstanceAiAgentNode; appended: boolean } {
	const text = outcome.userFacingMessage.trim();
	if (!text) return { tree, appended: false };

	const alreadyInTimeline = tree.timeline.some(
		(entry) => entry.type === 'text' && entry.responseId === responseId,
	);
	if (alreadyInTimeline) {
		return { tree, appended: false };
	}

	return {
		appended: true,
		tree: {
			...tree,
			textContent: tree.textContent ? `${tree.textContent}\n\n${outcome.userFacingMessage}` : text,
			timeline: [
				...tree.timeline,
				{ type: 'text', content: outcome.userFacingMessage, responseId },
			],
		},
	};
}

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';
}

// Stable UUID namespace for deterministic feedback IDs. Submitting the same
// (key, responseId) pair twice produces the same feedback UUID so LangSmith
// upserts the record (thumbs-down → later text comment = one record, not two).
const INSTANCE_AI_FEEDBACK_NAMESPACE = 'c5be4c87-5b6e-49ed-afe1-9c5c1f99a5c0';
const MAX_CONCURRENT_BACKGROUND_TASKS_PER_THREAD = 5;

function stringifyForContextValue(value: unknown): string {
	if (typeof value === 'string') return value;
	try {
		return JSON.stringify(value);
	} catch {
		return String(value);
	}
}

const PLANNED_TASK_CONTEXT_VALUE_LIMIT = 1_500;

function truncateContextValue(value: string): string {
	if (value.length <= PLANNED_TASK_CONTEXT_VALUE_LIMIT) return value;
	return `${value.slice(0, PLANNED_TASK_CONTEXT_VALUE_LIMIT)}...`;
}

function buildPlannedTaskConversationContext(
	task: PlannedTaskRecord,
	graph: PlannedTaskGraph | undefined,
): string | undefined {
	if (!graph) return undefined;

	const parts: string[] = [
		`Approved plan task: ${task.title}`,
		`Task id: ${task.id}`,
		`Task kind: ${task.kind}`,
		`Plan run id: ${graph.planRunId}`,
	];

	if (task.workflowId) {
		parts.push(`Target workflow id: ${task.workflowId}`);
	}

	const dependencies = graph.tasks.filter((candidate) => task.deps.includes(candidate.id));
	if (dependencies.length > 0) {
		parts.push('Completed dependency context:');
		for (const dependency of dependencies) {
			const dependencyParts = [
				`- ${dependency.id} (${dependency.kind}, ${dependency.status}): ${dependency.title}`,
			];
			if (dependency.result) {
				dependencyParts.push(`result=${truncateContextValue(dependency.result)}`);
			}
			if (dependency.error) {
				dependencyParts.push(`error=${truncateContextValue(dependency.error)}`);
			}
			if (dependency.outcome) {
				dependencyParts.push(
					`outcome=${truncateContextValue(stringifyForContextValue(dependency.outcome))}`,
				);
			}
			parts.push(dependencyParts.join(' '));
		}
	}

	return parts.join('\n');
}

/**
 * When HTTP_PROXY / HTTPS_PROXY is set (e.g. in e2e tests with MockServer),
 * return a fetch function that routes requests through the proxy. Node.js's
 * globalThis.fetch does not respect these env vars, so AI SDK providers would
 * bypass the proxy without this.
 */
function getProxyFetch(): typeof globalThis.fetch | undefined {
	const proxyUrl = process.env.HTTPS_PROXY || process.env.HTTP_PROXY;
	if (!proxyUrl) return undefined;

	// eslint-disable-next-line @typescript-eslint/no-require-imports
	const { ProxyAgent } = require('undici') as typeof Undici;
	const dispatcher = new ProxyAgent(proxyUrl);
	return (async (url: string | URL | Request, init?: RequestInit) =>
		await globalThis.fetch(url, {
			...init,
			// @ts-expect-error dispatcher is a valid undici option for Node.js fetch
			dispatcher,
		})) as typeof globalThis.fetch;
}

interface MessageTraceFinalization {
	status: 'completed' | 'cancelled' | 'error' | 'suspended';
	outputText?: string;
	reason?: string;
	modelId?: ModelConfig;
	outputs?: Record<string, unknown>;
	metadata?: Record<string, unknown>;
	error?: string;
}

type OrchestratorResumeReason =
	| 'approval'
	| 'background_task_completed'
	| 'planned_checkpoint'
	| 'replan';

/** A pending-confirmation orphan that's already passed the `canResumeOrphan`
 *  predicate — i.e. has the SDK-side pointers (`toolCallId`, `checkpointKey`)
 *  needed to rebuild a suspended run from the checkpoint blob. */
type ResumableOrphan = NonNullable<
	Awaited<ReturnType<InstanceAiPendingConfirmationRepository['claim']>>
> & {
	toolCallId: string;
	checkpointKey: string;
};

/** Result of `rebuildSuspendedRunFromCheckpoint`. Each failure variant
 *  corresponds to one rebuild step the caller can log distinctly; `ready`
 *  carries the fully-assembled state for `runState.suspendRun`. */
type RebuildSuspendedRunOutcome =
	| { kind: 'ready'; state: SuspendedRunState<User> }
	| { kind: 'no-user' }
	| { kind: 'no-checkpoint'; error?: unknown }
	| { kind: 'env-failure'; error: unknown }
	| { kind: 'agent-failure'; error: unknown };

/** Collapse the frontend's typed confirmation union into the flat payload
 *  consumed by native tool resume schemas and sub-agent HITL. Only the fields
 *  relevant to the submitted kind are populated — everything else stays undefined.
 *
 *  Most kinds carry implicit approval (you wouldn't be submitting answers,
 *  selected credentials, or a setup action otherwise) — only `approval`,
 *  `domainAccessDeny`, and `planDeny` carry a denial path. */
function toConfirmationData(request: InstanceAiConfirmRequest): ConfirmationData {
	switch (request.kind) {
		case 'approval':
			return { approved: request.approved, userInput: request.userInput };
		case 'domainAccessApprove':
			return { approved: true, domainAccessAction: request.domainAccessAction };
		case 'domainAccessDeny':
			return { approved: false };
		case 'planDeny':
			return { approved: false, denied: true };
		case 'questions':
			return { approved: true, answers: request.answers };
		case 'credentialSelection':
			return { approved: true, credentials: request.credentials };
		case 'resourceDecision':
			return { approved: true, resourceDecision: request.resourceDecision };
		case 'setupWorkflowApply':
			return {
				approved: true,
				action: 'apply',
				nodeCredentials: request.nodeCredentials,
				nodeParameters: request.nodeParameters,
			};
		case 'setupWorkflowTestTrigger':
			return {
				approved: true,
				action: 'test-trigger',
				testTriggerNode: request.testTriggerNode,
				nodeCredentials: request.nodeCredentials,
				nodeParameters: request.nodeParameters,
			};
	}
}

@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,
			);
		}
		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>();

	private readonly backgroundTasks = new BackgroundTaskManager(
		MAX_CONCURRENT_BACKGROUND_TASKS_PER_THREAD,
	);

	/** Trace contexts keyed by the n8n run ID that started the orchestration turn. */
	private readonly traceContextsByRunId = new Map<
		string,
		{
			threadId: string;
			messageGroupId?: string;
			tracing: InstanceAiTraceContext;
			traceSlug?: string;
		}
	>();
	/**
	 * Shared runtime workspaces keyed by thread ID. This is only an in-process
	 * cache; deterministic sandbox names let providers reconnect after restart
	 * or from another main when the thread uses the workspace again.
	 */
	private readonly sandboxes = new Map<string, RuntimeSandboxEntry>();

	/** In-flight runtime workspace creations keyed by thread ID. */
	private readonly sandboxCreations = new Map<string, Promise<RuntimeSandboxEntry | undefined>>();

	/** Per-user Local Gateway connections. Handles pairing tokens, session keys, and tool dispatch. */
	private readonly gatewayRegistry = new LocalGatewayRegistry();

	/** Domain-access trackers per thread — persists approvals across runs within a conversation. */
	private readonly domainAccessTrackersByThread = new Map<string, DomainAccessTracker>();

	/** Tracks the iframe pushRef per thread for live execution push events. */
	private readonly threadPushRef = new Map<string, string>();

	/** Per-thread promise chain that serializes schedulePlannedTasks calls. */
	private readonly schedulerLocks = new Map<string, Promise<void>>();

	/**
	 * 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<string, Set<string>>();

	private readonly pendingTerminalOutcomes = new Map<string, TerminalOutcome>();

	private terminalOutcomeStorage?: TerminalOutcomeStorage;

	private readonly liveness: InstanceAiLivenessService<SuspendedRunState<User>>;

	/** In-memory guard to prevent double credit counting within the same process. */
	private readonly creditedThreads = new Set<string>();

	/** Test-only trace replay state (slugs, events, shared TraceIndex/IdRemapper). */
	private readonly traceReplay = new TraceReplayState();

	/** Default IANA timezone for the instance (from GENERIC_TIMEZONE env var). */
	private readonly defaultTimeZone: string;

	private readonly logger: Logger;

	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<Promise<unknown>>();

	/**
	 * 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<string>();

	/**
	 * Restart-recovery metadata for runs that haven't yet been persisted by the
	 * SDK's end-of-turn save. Registered by the run entry point with the chosen
	 * `messageId` + raw text; consumed by:
	 *   1. `executeRun`'s streamInput builder, to tag the SDK's message with
	 *      our id so its eventual end-of-turn save upserts the same row.
	 *   2. `persistUserMessageOnFirstSuspend`, called from `waitForConfirmation`
	 *      so an inline HITL that pauses the turn before the SDK can save
	 *      still leaves the user's prompt in `instance_ai_messages` for
	 *      restart recovery.
	 *
	 * Entries are removed on confirmed DB save and cleared unconditionally in
	 * `executeRun`'s finally — the SDK's end-of-turn save handles the success
	 * path, and a mid-turn error means we deliberately don't want a
	 * half-saved row.
	 */
	private readonly userMessagePersistenceByRun = new Map<
		string,
		{ userId: string; message: { id: string; text: string } }
	>();

	constructor(
		logger: Logger,
		globalConfig: GlobalConfig,
		private readonly instanceSettings: InstanceSettings,
		private readonly adapterService: InstanceAiAdapterService,
		private readonly eventBus: InProcessEventBus,
		private readonly settingsService: InstanceAiSettingsService,
		private readonly agentMemory: TypeORMAgentMemory,
		private readonly checkpointStore: TypeORMAgentCheckpointStore,
		private readonly aiService: AiService,
		private readonly push: Push,
		private readonly threadRepo: InstanceAiThreadRepository,
		private readonly pendingConfirmationRepo: InstanceAiPendingConfirmationRepository,
		private readonly urlService: UrlService,
		private readonly dbSnapshotStorage: DbSnapshotStorage,
		private readonly dbIterationLogStorage: DbIterationLogStorage,
		private readonly sourceControlPreferencesService: SourceControlPreferencesService,
		private readonly telemetry: Telemetry,
		private readonly userRepository: UserRepository,
		private readonly aiBuilderTemporaryWorkflowRepository: AiBuilderTemporaryWorkflowRepository,
		private readonly errorReporter: ErrorReporter,
		ssrfProtectionConfig: SsrfProtectionConfig,
		ssrfProtectionService: SsrfProtectionService,
		private readonly eventService: EventService,
	) {
		this.logger = logger.scoped('instance-ai');
		this.instanceAiConfig = globalConfig.instanceAi;
		const livenessPolicyConfig = createInstanceAiLivenessPolicyConfig({
			confirmationTimeoutMs: this.instanceAiConfig.confirmationTimeout,
		});
		this.liveness = new InstanceAiLivenessService<SuspendedRunState<User>>({
			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.dropPendingConfirmation(requestId);
			},
		});
		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;

		// When the admin changes MCP settings, tear down existing clients so the
		// next agent run rebuilds them against the new config. In-flight tool
		// calls on disconnected clients will fail — that's accepted: the
		// alternative is leaking clients keyed by stale config until shutdown.
		// We only listen for the MCP-changed flag so unrelated settings saves
		// don't churn live MCP connections.
		this.eventService.on('instance-ai-settings-updated', ({ mcpSettingsChanged }) => {
			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 getSandboxConfigFromEnv(): SandboxConfig {
		const {
			sandboxEnabled,
			sandboxProvider,
			daytonaApiUrl,
			daytonaApiKey,
			n8nSandboxServiceUrl,
			n8nSandboxServiceApiKey,
			sandboxImage,
			sandboxTimeout,
			sandboxNamePrefix,
			daytonaTokenRefreshSkewMs,
		} = this.instanceAiConfig;
		if (!sandboxEnabled) {
			return {
				enabled: false,
				provider:
					sandboxProvider === 'n8n-sandbox'
						? 'n8n-sandbox'
						: sandboxProvider === 'daytona'
							? 'daytona'
							: 'local',
				timeout: sandboxTimeout,
			};
		}

		if (sandboxProvider === 'daytona') {
			return {
				enabled: true,
				provider: 'daytona',
				daytonaApiUrl: daytonaApiUrl || undefined,
				daytonaApiKey: daytonaApiKey || undefined,
				image: sandboxImage || undefined,
				n8nVersion: N8N_VERSION || undefined,
				timeout: sandboxTimeout,
				namePrefix: sandboxNamePrefix || undefined,
				refreshSkewMs: daytonaTokenRefreshSkewMs,
			};
		}

		if (sandboxProvider === 'n8n-sandbox') {
			return {
				enabled: true,
				provider: 'n8n-sandbox',
				serviceUrl: n8nSandboxServiceUrl || undefined,
				apiKey: n8nSandboxServiceApiKey || undefined,
				timeout: sandboxTimeout,
			};
		}

		return {
			enabled: true,
			provider: 'local',
			timeout: sandboxTimeout,
		};
	}

	private async resolveSandboxConfig(user: User): Promise<SandboxConfig> {
		const base = this.getSandboxConfigFromEnv();
		if (!base.enabled) return base;
		if (base.provider === 'daytona') {
			// If AI assistant service is available, route Daytona calls through its sandbox proxy
			if (this.aiService.isProxyEnabled()) {
				const client = await this.aiService.getClient();
				const proxyConfig = await client.getSandboxProxyConfig();
				return {
					...base,
					daytonaApiUrl: client.getSandboxProxyBaseUrl(),
					image: proxyConfig.image,
					logger: this.logger,
					getAuthToken: async () => {
						const token = await client.getBuilderApiProxyToken(
							{ id: user.id },
							{ userMessageId: nanoid() },
						);

						return token.accessToken;
					},
				};
			}

			// Direct mode: Daytona credentials from env vars or admin credential
			const daytona = await this.settingsService.resolveDaytonaConfig(user);
			return {
				...base,
				daytonaApiUrl: daytona.apiUrl ?? base.daytonaApiUrl,
				daytonaApiKey: daytona.apiKey ?? base.daytonaApiKey,
			};
		}
		if (base.provider === 'n8n-sandbox') {
			const sandbox = await this.settingsService.resolveN8nSandboxConfig(user);
			return {
				...base,
				serviceUrl: sandbox.serviceUrl ?? base.serviceUrl,
				apiKey: sandbox.apiKey ?? base.apiKey,
			};
		}
		return base;
	}

	private async getOrCreateWorkspaceEntry(
		threadId: string,
		user: User,
		runId?: string,
	): Promise<RuntimeSandboxEntry | undefined> {
		const existing = this.sandboxes.get(threadId);
		if (existing) {
			if (this.isSandboxEntryExpired(existing) && !this.isSandboxInUse(threadId)) {
				this.evictSandboxEntry(threadId, existing);
			} else {
				this.touchSandboxEntry(threadId, existing);
				return existing;
			}
		}

		const pending = this.sandboxCreations.get(threadId);
		if (pending) return await pending;

		const creation = this.createWorkspaceEntry(threadId, user, runId);
		this.sandboxCreations.set(threadId, creation);
		try {
			return await creation;
		} finally {
			this.sandboxCreations.delete(threadId);
		}
	}

	/** Get or create the shared runtime sandbox + workspace for a thread. */
	private async getOrCreateWorkspace(
		threadId: string,
		user: User,
		context: InstanceAiContext,
		runId?: string,
	): Promise<RuntimeSandboxEntry | undefined> {
		const entry = await this.getOrCreateWorkspaceEntry(threadId, user, runId);
		if (entry) await this.ensureWorkspaceSetup(entry, context);
		return entry;
	}

	private async ensureWorkspaceSetup(
		entry: RuntimeSandboxEntry,
		context: InstanceAiContext,
	): Promise<void> {
		if (entry.setupComplete) return;

		entry.setupPromise ??= setupSandboxWorkspace(entry.workspace, context)
			.then(() => {
				entry.setupComplete = true;
			})
			.finally(() => {
				entry.setupPromise = undefined;
			});

		await entry.setupPromise;
	}

	private async createWorkspaceEntry(
		threadId: string,
		user: User,
		runId?: string,
	): Promise<RuntimeSandboxEntry | undefined> {
		const config = withThreadScopedSandboxIdentity(
			await this.resolveSandboxConfig(user),
			threadId,
			runId,
		);
		if (!config.enabled) return undefined;

		const sandbox = await createSandbox(config, {
			logger: this.logger,
			errorReporter: this.errorReporter,
			useSnapshotFallback: true,
		});
		const workspace = createWorkspace(sandbox);
		if (!sandbox || !workspace) return undefined;
		try {
			await workspace.init();
		} catch (error) {
			try {
				await workspace.destroy();
			} catch {
				// Best-effort cleanup when the sandbox cannot start
			}
			throw error;
		}

		const entry: RuntimeSandboxEntry = {
			sandbox,
			workspace,
			setupComplete: false,
			setupPromise: undefined,
			expiresAt: this.nextSandboxExpiry(),
		};
		this.sandboxes.set(threadId, entry);
		this.scheduleSandboxExpiry(threadId, entry);
		return entry;
	}

	private evictSandboxEntry(threadId: string, entry: RuntimeSandboxEntry): void {
		if (this.sandboxes.get(threadId) !== entry) return;

		this.sandboxes.delete(threadId);
		if (entry.cleanupTimer) {
			clearTimeout(entry.cleanupTimer);
			entry.cleanupTimer = undefined;
		}
	}

	/** Destroy and remove the shared runtime workspace for a thread. */
	private async destroySandbox(threadId: string, reason = 'thread_cleanup'): Promise<void> {
		const entry = this.sandboxes.get(threadId);
		if (!entry?.sandbox) return;

		this.evictSandboxEntry(threadId, entry);
		try {
			await entry.workspace?.destroy();
		} catch (error) {
			this.logger.warn('Failed to destroy sandbox', {
				threadId,
				reason,
				error: error instanceof Error ? error.message : String(error),
			});
		}
	}

	private get sandboxTtlMs(): number {
		return this.instanceAiConfig?.builderSandboxTtlMs ?? DEFAULT_SANDBOX_TTL_MS;
	}

	private nextSandboxExpiry(): number {
		return Date.now() + this.sandboxTtlMs;
	}

	private isSandboxEntryExpired(entry: RuntimeSandboxEntry): boolean {
		return this.sandboxTtlMs > 0 && entry.expiresAt <= Date.now();
	}

	private touchSandboxEntry(threadId: string, entry: RuntimeSandboxEntry): void {
		if (this.sandboxTtlMs <= 0) return;
		entry.expiresAt = this.nextSandboxExpiry();
		this.scheduleSandboxExpiry(threadId, entry);
	}

	private isSandboxInUse(threadId: string): boolean {
		return Boolean(
			this.runState.getActiveRunId(threadId) ||
				this.runState.hasSuspendedRun(threadId) ||
				this.backgroundTasks.getRunningTasks(threadId).length > 0,
		);
	}

	private scheduleSandboxExpiry(threadId: string, entry: RuntimeSandboxEntry): void {
		if (this.sandboxTtlMs <= 0) return;
		if (entry.cleanupTimer) clearTimeout(entry.cleanupTimer);

		// Provider auto-stop handles remote Daytona sandboxes. This timer only
		// drops our in-process cache entry so the map cannot grow indefinitely.
		const delay = Math.max(0, entry.expiresAt - Date.now());
		entry.cleanupTimer = setTimeout(() => {
			const current = this.sandboxes.get(threadId);
			if (current !== entry) return;
			if (this.isSandboxInUse(threadId)) {
				this.touchSandboxEntry(threadId, entry);
				return;
			}
			this.evictSandboxEntry(threadId, entry);
		}, delay);
		entry.cleanupTimer.unref();
	}

	private stopSandboxExpiryTimers(): void {
		for (const entry of this.sandboxes.values()) {
			if (!entry.cleanupTimer) continue;
			clearTimeout(entry.cleanupTimer);
			entry.cleanupTimer = undefined;
		}
	}

	/**
	 * Fetch a fresh proxy auth token and return the client + Authorization headers.
	 * Each caller gets a unique token (separate nanoid) for audit tracking.
	 */
	private async getProxyAuth(user: User) {
		const client = await this.aiService.getClient();
		const token = await client.getBuilderApiProxyToken(
			{ id: user.id },
			{ userMessageId: nanoid() },
		);
		return {
			client,
			headers: { Authorization: `${token.tokenType} ${token.accessToken}` },
		};
	}

	/**
	 * Full model-resolver chain shared between chat and eval paths.
	 *
	 * Mirrors the resolution used in `processMessage`:
	 *   1. AI service proxy (when enabled) — wraps with proxy auth.
	 *   2. HTTP_PROXY (when set, e.g. e2e tests) — wraps with proxy fetch.
	 *   3. Env vars / user credential — raw settings resolution.
	 *
	 * Call this instead of `settingsService.resolveModelConfig` directly so
	 * the eval endpoint gets the same working model the chat endpoint uses.
	 */
	async resolveAgentModelConfig(user: User): Promise<ModelConfig> {
		if (this.aiService.isProxyEnabled()) {
			const client = await this.aiService.getClient();
			const proxyBaseUrl = client.getApiProxyBaseUrl();
			const tokenManager = new ProxyTokenManager(async () => {
				return await client.getBuilderApiProxyToken({ id: user.id }, { userMessageId: nanoid() });
			});
			return await this.resolveProxyModel(user, proxyBaseUrl, tokenManager);
		}
		const httpProxyModel = await this.resolveHttpProxyModel(user);
		if (httpProxyModel) return httpProxyModel;
		return await this.settingsService.resolveModelConfig(user);
	}

	/**
	 * Build model config. When the AI service proxy is enabled, returns a native
	 * Anthropic LanguageModelV2 instance pointing at the proxy.
	 *
	 * We use `@ai-sdk/anthropic` directly instead of returning a `{ url }` config
	 * object because this proxy route needs the native Anthropic transport.
	 * The proxy may forward to Vertex AI, which only supports the native Anthropic
	 * Messages API (`/v1/messages`), not the OpenAI-compatible endpoint.
	 *
	 * Auth headers are injected via a custom `fetch` wrapper so that each
	 * request gets a fresh-or-cached token from the ProxyTokenManager,
	 * avoiding 401s on long-running agent turns.
	 */
	private async resolveProxyModel(
		user: User,
		proxyBaseUrl: string,
		tokenManager: ProxyTokenManager,
	): Promise<ModelConfig> {
		const modelName = this.settingsService.resolveModelName(user);
		const { createAnthropic } = await import('@ai-sdk/anthropic');
		const provider = createAnthropic({
			baseURL: proxyBaseUrl + '/anthropic/v1',
			apiKey: 'proxy-managed',
			fetch: async (input, init) => {
				const headers = new Headers(init?.headers);
				const auth = await tokenManager.getAuthHeaders();
				for (const [k, v] of Object.entries(auth)) {
					headers.set(k, v);
				}
				for (const [k, v] of Object.entries(
					buildProxyHeaders({ feature: 'instance-ai', n8nVersion: N8N_VERSION }),
				)) {
					headers.set(k, v);
				}
				return await globalThis.fetch(input, { ...init, headers });
			},
		});
		return provider(modelName);
	}

	/**
	 * When HTTP_PROXY is set (e.g. e2e tests with MockServer), build the model
	 * with a proxy-aware fetch so the AI SDK routes through the proxy.
	 * Returns undefined if no HTTP_PROXY is set or the model isn't anthropic.
	 */
	private async resolveHttpProxyModel(user: User): Promise<ModelConfig | undefined> {
		const proxyFetch = getProxyFetch();
		if (!proxyFetch) return undefined;

		const config = await this.settingsService.resolveModelConfig(user);
		const modelId = typeof config === 'string' ? config : 'id' in config ? config.id : null;
		if (!modelId) return undefined;

		const [provider, ...rest] = modelId.split('/');
		const modelName = rest.join('/');
		const apiKey = typeof config === 'object' && 'apiKey' in config ? config.apiKey : undefined;
		const baseURL = typeof config === 'object' && 'url' in config ? config.url : undefined;
		if (provider !== 'anthropic') return undefined;

		const { createAnthropic } = await import('@ai-sdk/anthropic');
		return createAnthropic({
			apiKey,
			baseURL: baseURL || undefined,
			fetch: proxyFetch,
		})(modelName);
	}

	/**
	 * Count one credit for the first completed orchestrator run in a thread.
	 * Subsequent messages in the same thread are free.
	 *
	 * Race-condition mitigation strategy:
	 * - In-memory Set (`creditedThreads`) prevents concurrent calls within
	 *   the same process from both passing the check.
	 * - DB metadata (`creditCounted: true`) survives process restarts.
	 * - markBuilderSuccess is idempotent on the proxy side, so a theoretical
	 *   double-count after a crash mid-save is harmless.
	 */
	private async countCreditsIfFirst(user: User, threadId: string, runId: string): Promise<void> {
		if (!this.aiService.isProxyEnabled()) return;

		// Fast in-memory check — prevents the read-then-write race within a single process.
		if (this.creditedThreads.has(threadId)) return;

		let thread: Awaited<ReturnType<InstanceAiThreadRepository['findOneBy']>>;
		try {
			thread = await this.threadRepo.findOneBy({ id: threadId });
		} catch (error) {
			this.logger.warn('Failed to check Instance AI credit status', {
				threadId,
				runId,
				error: getErrorMessage(error),
			});
			return;
		}
		if (!thread) return;
		if (thread.metadata?.creditCounted) {
			this.creditedThreads.add(threadId); // Sync in-memory with DB state
			return;
		}

		try {
			this.creditedThreads.add(threadId); // Claim before async work
			const { client, headers: authHeaders } = await this.getProxyAuth(user);
			const info = await client.markBuilderSuccess({ id: user.id }, authHeaders);
			if (info) {
				thread.metadata = { ...thread.metadata, creditCounted: true };
				await this.threadRepo.save(thread);
				this.push.sendToUsers(
					{
						type: 'updateInstanceAiCredits',
						data: { creditsQuota: info.creditsQuota, creditsClaimed: info.creditsClaimed },
					},
					[user.id],
				);
			}
		} catch (error) {
			this.creditedThreads.delete(threadId); // Allow retry on failure
			this.logger.warn('Failed to count Instance AI credits', {
				error: getErrorMessage(error),
				threadId,
				runId,
			});
		}
	}

	/** Whether the AI service proxy is enabled for credit counting. */
	isProxyEnabled(): boolean {
		return this.aiService.isProxyEnabled();
	}

	/** Get current credit usage from the AI service proxy. */
	async getCredits(user: User): Promise<{ creditsQuota: number; creditsClaimed: number }> {
		if (!this.aiService.isProxyEnabled()) {
			return { creditsQuota: UNLIMITED_CREDITS, creditsClaimed: 0 };
		}
		const client = await this.aiService.getClient();
		return await client.getBuilderInstanceCredits({ id: user.id });
	}

	isEnabled(): boolean {
		return this.settingsService.isAgentEnabled() && !!this.instanceAiConfig.model;
	}

	hasActiveRun(threadId: string): boolean {
		return this.runState.hasLiveRun(threadId);
	}

	getThreadStatus(threadId: string): InstanceAiThreadStatusResponse {
		return this.runState.getThreadStatus(threadId, this.backgroundTasks.getTaskSnapshots(threadId));
	}

	private storeTraceContext(
		runId: string,
		threadId: string,
		tracing: InstanceAiTraceContext,
		messageGroupId?: string,
	): void {
		this.traceContextsByRunId.set(runId, {
			threadId,
			messageGroupId,
			tracing,
			traceSlug: this.traceReplay.getActiveSlug(),
		});
	}

	private getTraceContext(runId: string): InstanceAiTraceContext | undefined {
		return this.traceContextsByRunId.get(runId)?.tracing;
	}

	private getTraceContextForContinuation(
		threadId: string,
		messageGroupId?: string,
	): InstanceAiTraceContext | undefined {
		const entries = [...this.traceContextsByRunId.values()].reverse();
		const sameGroup =
			messageGroupId === undefined
				? undefined
				: entries.find(
						(entry) => entry.threadId === threadId && entry.messageGroupId === messageGroupId,
					)?.tracing;
		return sameGroup ?? entries.find((entry) => entry.threadId === threadId)?.tracing;
	}

	private async createOrchestratorResumeTraceContext(options: {
		baseTracing?: InstanceAiTraceContext;
		threadId: string;
		messageId: string;
		messageGroupId?: string;
		runId: string;
		userId: string;
		modelId?: ModelConfig;
		input: Record<string, unknown>;
		proxyConfig?: ServiceProxyConfig;
		resumeReason: OrchestratorResumeReason;
		metadata?: Record<string, unknown>;
	}): Promise<InstanceAiTraceContext | undefined> {
		const baseTracing =
			options.baseTracing ??
			this.getTraceContextForContinuation(options.threadId, options.messageGroupId);
		if (!baseTracing) return undefined;

		const tracing = await continueInstanceAiTraceContext(baseTracing, {
			threadId: options.threadId,
			messageId: options.messageId,
			messageGroupId: options.messageGroupId,
			runId: options.runId,
			userId: options.userId,
			modelId: options.modelId,
			input: options.input,
			proxyConfig: options.proxyConfig ?? baseTracing?.proxyConfig,
			metadata: {
				resume_reason: options.resumeReason,
				agent_id: ORCHESTRATOR_AGENT_ID,
				...options.metadata,
			},
			n8nVersion: N8N_VERSION,
			workflowSdkVersion: WORKFLOW_SDK_VERSION,
		});

		if (tracing) {
			await this.configureTraceReplayMode(tracing);
			this.storeTraceContext(options.runId, options.threadId, tracing, options.messageGroupId);
			this.runState.attachTracing(options.threadId, tracing);
		}

		return tracing;
	}

	private async configureTraceReplayMode(tracing: InstanceAiTraceContext): Promise<void> {
		await this.traceReplay.configureReplayMode(tracing);
	}

	private async finalizeMessageTraceRoot(
		runId: string,
		tracing: InstanceAiTraceContext,
		options: MessageTraceFinalization,
	): Promise<void> {
		if (tracing.rootRun.endTime) return;

		const outputs = options.outputs ?? {
			status: options.status,
			runId,
			...(options.outputText ? { response: options.outputText } : {}),
			...(options.reason ? { reason: options.reason } : {}),
		};
		const metadata = {
			final_status: options.status,
			...(options.modelId !== undefined ? { model_id: options.modelId } : {}),
			...options.metadata,
		};

		try {
			await tracing.finishRun(tracing.rootRun, {
				outputs,
				metadata,
				...(options.error
					? { error: options.error }
					: options.status === 'error' && options.reason
						? { error: options.reason }
						: {}),
			});
		} catch (error) {
			this.logger.warn('Failed to finalize Instance AI message trace root', {
				runId,
				threadId: tracing.rootRun.metadata?.thread_id,
				error: getErrorMessage(error),
			});
		} finally {
			releaseTraceClient(tracing.rootRun.traceId);
		}
	}

	private async maybeFinalizeRunTraceRoot(
		runId: string,
		options: MessageTraceFinalization,
	): Promise<void> {
		const tracing = this.getTraceContext(runId);
		if (!tracing) return;
		await this.finalizeMessageTraceRoot(runId, tracing, options);
	}

	private buildMessageTraceMetadata(
		threadId: string,
		runId: string,
		options: InstanceAiRunTraceMetadataOptions,
	): Record<string, unknown> {
		const traceOptions = {
			status: options.status,
			...(options.cancellationReason !== undefined
				? { cancellationReason: options.cancellationReason }
				: {}),
			...(options.runTimeout !== undefined ? { runTimeout: options.runTimeout } : {}),
		};

		return {
			completion_source: 'orchestrator',
			...buildInstanceAiRunTraceMetadata(
				this.eventBus.getEventsForRun(threadId, runId),
				traceOptions,
			),
		};
	}

	private async finalizeRemainingMessageTraceRoots(
		threadId: string,
		options: MessageTraceFinalization,
	): Promise<void> {
		const finalizedMessageRuns = new Set<string>();

		for (const [runId, entry] of this.traceContextsByRunId) {
			if (entry.threadId !== threadId) continue;
			if (finalizedMessageRuns.has(entry.tracing.rootRun.id)) continue;

			finalizedMessageRuns.add(entry.tracing.rootRun.id);
			await this.finalizeMessageTraceRoot(runId, entry.tracing, options);
		}
	}

	private deleteTraceContextsForThread(threadId: string): void {
		for (const [runId, entry] of this.traceContextsByRunId) {
			if (entry.threadId === threadId) {
				releaseTraceClient(entry.tracing.rootRun.traceId);
				// Preserve recorded trace events in the slug-scoped store
				// so the test fixture teardown can still retrieve them via GET.
				if (entry.tracing.traceWriter && entry.traceSlug) {
					this.traceReplay.preserveWriterEvents(
						entry.traceSlug,
						entry.tracing.traceWriter.getEvents(),
					);
				}
				this.traceContextsByRunId.delete(runId);
			}
		}
	}

	private async finalizeDetachedTraceRun(
		taskId: string,
		traceContext: InstanceAiTraceContext | undefined,
		options: {
			status: 'completed' | 'failed' | 'cancelled';
			outputs?: Record<string, unknown>;
			error?: string;
			metadata?: Record<string, unknown>;
		},
	): Promise<void> {
		if (!traceContext) return;

		try {
			if (
				traceContext.actorRun.id !== traceContext.rootRun.id &&
				traceContext.actorRun.endTime === undefined
			) {
				await traceContext.finishRun(traceContext.actorRun, {
					outputs: {
						status: options.status,
						...options.outputs,
					},
					metadata: {
						final_status: options.status,
						...options.metadata,
					},
					...(options.error ? { error: options.error } : {}),
				});
			}
			await traceContext.finishRun(traceContext.rootRun, {
				outputs: {
					status: options.status,
					...options.outputs,
				},
				metadata: {
					final_status: options.status,
					...options.metadata,
				},
				...(options.error ? { error: options.error } : {}),
			});
		} catch (error) {
			this.logger.warn('Failed to finalize Instance AI detached trace run', {
				taskId,
				traceRunId: traceContext.rootRun.id,
				error: getErrorMessage(error),
			});
		} finally {
			releaseTraceClient(traceContext.rootRun.traceId);
		}
	}

	private async finalizeRunTracing(
		runId: string,
		tracing: InstanceAiTraceContext | undefined,
		options: MessageTraceFinalization,
	): Promise<void> {
		if (!tracing) return;
		if (tracing.actorRun.endTime) return;

		const outputs = options.outputs ?? {
			status: options.status,
			runId,
			...(options.outputText ? { response: options.outputText } : {}),
			...(options.reason ? { reason: options.reason } : {}),
		};

		const metadata = {
			final_status: options.status,
			...(options.modelId !== undefined ? { model_id: options.modelId } : {}),
			...options.metadata,
		};

		try {
			await tracing.finishRun(tracing.actorRun, {
				outputs,
				metadata,
				...(options.status === 'error' && options.reason ? { error: options.reason } : {}),
			});
		} catch (error) {
			this.logger.warn('Failed to finalize Instance AI run tracing', {
				runId,
				threadId: tracing.actorRun.metadata?.thread_id,
				error: getErrorMessage(error),
			});
		}
	}

	private async finalizeBackgroundTaskTracing(
		task: ManagedBackgroundTask,
		status: 'completed' | 'failed' | 'cancelled',
	): Promise<void> {
		await this.finalizeDetachedTraceRun(task.taskId, task.traceContext, {
			status,
			outputs: {
				taskId: task.taskId,
				agentId: task.agentId,
				role: task.role,
				...(task.result ? { result: task.result } : {}),
			},
			...(status === 'failed' && task.error ? { error: task.error } : {}),
			metadata: {
				...(task.plannedTaskId ? { planned_task_id: task.plannedTaskId } : {}),
				...(task.workItemId ? { work_item_id: task.workItemId } : {}),
			},
		});
	}

	async submitLangsmithFeedback(
		user: User,
		threadId: string,
		responseId: string,
		payload: { rating: 'up' | 'down'; comment?: string },
	): Promise<void> {
		const anchor = await this.dbSnapshotStorage.findLangsmithAnchor(threadId, responseId);
		if (!anchor) {
			this.logger.debug('No LangSmith anchor for feedback; skipping annotation', {
				threadId,
				responseId,
			});
			return;
		}

		let tracingProxyConfig: ServiceProxyConfig | undefined;
		if (this.aiService.isProxyEnabled()) {
			try {
				const client = await this.aiService.getClient();
				const baseUrl = client.getApiProxyBaseUrl();
				const manager = new ProxyTokenManager(
					async () =>
						await client.getBuilderApiProxyToken({ id: user.id }, { userMessageId: nanoid() }),
				);
				tracingProxyConfig = {
					apiUrl: baseUrl + '/langsmith',
					getAuthHeaders: async () => await manager.getAuthHeaders(),
				};
			} catch (error) {
				this.logger.warn('Failed to build LangSmith proxy config for feedback', {
					threadId,
					responseId,
					error: getErrorMessage(error),
				});
				return;
			}
		}

		const key = 'user_score';
		const feedbackId = uuidv5(`${key}:${responseId}`, INSTANCE_AI_FEEDBACK_NAMESPACE);

		try {
			await submitLangsmithUserFeedback({
				langsmithRunId: anchor.langsmithRunId,
				langsmithTraceId: anchor.langsmithTraceId,
				key,
				score: payload.rating === 'up' ? 1 : 0,
				value: payload.rating,
				comment: payload.comment,
				feedbackId,
				sourceInfo: {
					thread_id: threadId,
					response_id: responseId,
					user_id: user.id,
					rating: payload.rating,
				},
				proxyConfig: tracingProxyConfig,
			});
		} catch (error) {
			this.logger.warn('Failed to submit LangSmith feedback', {
				threadId,
				responseId,
				error: getErrorMessage(error),
			});
		}
	}

	startRun(
		user: User,
		threadId: string,
		message: string,
		attachments?: InstanceAiAttachment[],
		timeZone?: string,
		pushRef?: string,
	): string {
		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 planner and system prompt
		// instead of falling back to GENERIC_TIMEZONE.
		if (timeZone) {
			this.runState.setTimeZone(threadId, timeZone);
		}

		if (pushRef !== undefined) {
			this.threadPushRef.set(threadId, pushRef);
		}

		// Stable id for the user-typed message. Coordinated with the SDK's
		// end-of-turn save (see executeRun's streamInput construction) so we
		// end up with exactly one user-message row per turn — and so a save
		// triggered by `waitForConfirmation` on a suspended HITL turn uses the
		// same id the SDK would have used for completion.
		const userMessageId = nanoid();
		this.userMessagePersistenceByRun.set(runId, {
			userId: user.id,
			message: { id: userMessageId, text: message },
		});

		this.startExecuteRun(
			user,
			threadId,
			runId,
			message,
			abortController,
			attachments,
			messageGroupId,
			timeZone,
			false,
			undefined,
			undefined,
		);

		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.finalizeBackgroundTaskTracing(task, 'cancelled');
			this.eventBus.publish(threadId, {
				type: 'agent-completed',
				runId: task.runId,
				agentId: task.agentId,
				payload: {
					role: task.role,
					result: '',
					error: reason === INSTANCE_AI_RUN_TIMEOUT_REASON ? 'Timed out' : 'Cancelled by user',
				},
			});
			void this.recordBackgroundTerminalOutcome(task).finally(() => {
				void this.saveAgentTreeSnapshot(
					threadId,
					task.runId,
					this.dbSnapshotStorage,
					true,
					task.messageGroupId,
				);
			});
			if (user) {
				void this.handlePlannedTaskSettlement(user, task, 'cancelled');
			}
		}

		// 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(), and (b) incorrectly let a
		// future unrelated create-tasks call bypass the replan-only guard via
		// threadHasExistingPlan(). 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.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.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.finalizeBackgroundTaskTracing(task, 'cancelled');
		this.eventBus.publish(threadId, {
			type: 'agent-completed',
			runId: task.runId,
			agentId: task.agentId,
			payload: { role: task.role, result: '', error: 'Cancelled by user' },
		});

		// Persist the updated agent tree so cancelled status survives page reload.
		// The onSettled callback in executeTask is skipped for aborted tasks,
		// so we must save the snapshot explicitly here.
		void this.recordBackgroundTerminalOutcome(task).finally(() => {
			void this.saveAgentTreeSnapshot(
				threadId,
				task.runId,
				this.dbSnapshotStorage,
				true,
				task.messageGroupId,
			);
		});

		const user = this.runState.getThreadUser(threadId);
		if (user) {
			void this.handlePlannedTaskSettlement(user, task, 'cancelled');
		}
	}

	/** Cancel all background tasks across all threads. Test-only. */
	cancelAllBackgroundTasks(): number {
		const cancelled = this.backgroundTasks.cancelAll();
		for (const task of cancelled) {
			void this.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: ORCHESTRATOR_AGENT_ID,
			userId: user.id,
			payload: { messageId, messageGroupId },
		});
		this.eventBus.publish(threadId, {
			type: 'text-delta',
			runId,
			agentId: ORCHESTRATOR_AGENT_ID,
			responseId: `test-background-start:${runId}`,
			payload: { text: messageText },
		});
		this.eventBus.publish(threadId, {
			type: 'agent-spawned',
			runId,
			agentId,
			payload: {
				parentId: ORCHESTRATOR_AGENT_ID,
				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<string>((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.recordBackgroundTerminalOutcome(task);
				await this.saveAgentTreeSnapshot(
					threadId,
					runId,
					this.dbSnapshotStorage,
					true,
					messageGroupId,
				);
			},
		});

		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: ORCHESTRATOR_AGENT_ID,
			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<void> {
		await this.liveness.sweepTimedOutWork(now);
	}

	// ── Gateway lifecycle (delegated to LocalGatewayRegistry) ───────────────

	// ── Test-only trace replay API ───────────────────────────────────────────

	loadTraceEvents(slug: string, events: unknown[]): void {
		this.traceReplay.loadEvents(slug, events);
	}

	getTraceEvents(slug: string): unknown[] {
		return this.traceReplay.getEventsWithWriterFallback(slug, this.traceContextsByRunId.values());
	}

	activateTraceSlug(slug: string): void {
		this.traceReplay.activateSlug(slug);
	}

	clearTraceEvents(slug: string): void {
		this.traceReplay.clearEvents(slug);
	}

	getUserIdForApiKey(key: string): string | undefined {
		return this.gatewayRegistry.getUserIdForApiKey(key);
	}

	generatePairingToken(userId: string): string {
		return this.gatewayRegistry.generatePairingToken(userId);
	}

	getGatewayApiKeyExpiresAt(userId: string, key: string): Date | null {
		return this.gatewayRegistry.getApiKeyExpiresAt(userId, key);
	}

	getPairingToken(userId: string): string | null {
		return this.gatewayRegistry.getPairingToken(userId);
	}

	consumePairingToken(userId: string, token: string): string | null {
		return this.gatewayRegistry.consumePairingToken(userId, token);
	}

	getActiveSessionKey(userId: string): string | null {
		return this.gatewayRegistry.getActiveSessionKey(userId);
	}

	clearActiveSessionKey(userId: string): void {
		this.gatewayRegistry.clearActiveSessionKey(userId);
	}

	getLocalGateway(userId: string): LocalGateway {
		return this.gatewayRegistry.getGateway(userId);
	}

	initGateway(userId: string, data: InstanceAiGatewayCapabilities): void {
		this.gatewayRegistry.initGateway(userId, data);
		this.telemetry.track('User connected to Computer Use', {
			user_id: userId,
			tool_groups: data.toolCategories.filter((c) => c.enabled).map((c) => c.name),
		});
	}

	resolveGatewayRequest(
		userId: string,
		requestId: string,
		result?: McpToolCallResult,
		error?: string,
	): boolean {
		return this.gatewayRegistry.resolveGatewayRequest(userId, requestId, result, error);
	}

	disconnectGateway(userId: string): void {
		this.gatewayRegistry.disconnectGateway(userId);
	}

	/** Disconnect all connected gateways and return the user IDs that were connected. */
	disconnectAllGateways(): string[] {
		const connectedUserIds = this.gatewayRegistry.getConnectedUserIds();
		this.gatewayRegistry.disconnectAll();
		return connectedUserIds;
	}

	isLocalGatewayDisabled(): boolean {
		return this.settingsService.isLocalGatewayDisabled();
	}

	getGatewayStatus(userId: string): {
		connected: boolean;
		connectedAt: string | null;
		directory: string | null;
		hostIdentifier: string | null;
		toolCategories: ToolCategory[];
	} {
		return this.gatewayRegistry.getGatewayStatus(userId);
	}

	startDisconnectTimer(userId: string, onDisconnect: () => void): void {
		this.gatewayRegistry.startDisconnectTimer(userId, onDisconnect);
	}

	clearDisconnectTimer(userId: string): void {
		this.gatewayRegistry.clearDisconnectTimer(userId);
	}

	/**
	 * 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<void> {
		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.finalizeRunTracing(active.runId, active.tracing, {
				status: 'cancelled',
				reason: 'thread_cleared',
			});
		}
		if (suspended) {
			suspended.abortController.abort();
			await this.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.finalizeBackgroundTaskTracing(task, 'cancelled');
		}
		await this.finalizeRemainingMessageTraceRoots(threadId, {
			status: 'cancelled',
			reason: 'thread_cleared',
			metadata: { completion_source: 'service_cleanup' },
		});

		this.creditedThreads.delete(threadId);
		this.schedulerLocks.delete(threadId);
		this.domainAccessTrackersByThread.delete(threadId);
		this.threadPushRef.delete(threadId);
		this.deleteTraceContextsForThread(threadId);
		await this.destroySandbox(threadId);
		await this.reapAiTemporaryForThreadCleanup(threadId);
		await this.dropPendingConfirmationsForThread(threadId);
		this.eventBus.clearThread(threadId);
	}

	async shutdown(): Promise<void> {
		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 (planner `submit-plan`,
			// 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 + re-save the
			// snapshot here, we'd permanently overwrite the plan/ask card with
			// a `status: 'cancelled'` tree before the user has a chance to see
			// it on reload.
			if (threadsWithPendingHitl.has(run.threadId)) {
				await this.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 / snapshot
				// writes that would otherwise overwrite the plan/ask card.
				this.preserveHitlOnShutdown.add(run.runId);
				run.abortController.abort();
				continue;
			}

			// Truly mid-stream run: publish run-finish first so the terminal
			// event lands in the event bus before the snapshot reads it;
			// saveAgentTreeSnapshot rebuilds the tree from the bus, so without
			// this order the persisted tree would still look mid-stream after
			// the process is gone.
			this.publishRunFinish(run.threadId, run.runId, 'cancelled', 'service_shutdown');
			await this.persistShutdownSnapshot(run.threadId, run.runId, run.messageGroupId);
			await this.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.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.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);

		const threadsWithTraces = new Set(
			[...this.traceContextsByRunId.values()].map((entry) => entry.threadId),
		);
		for (const threadId of threadsWithTraces) {
			await this.finalizeRemainingMessageTraceRoots(threadId, {
				status: 'cancelled',
				reason: 'service_shutdown',
				metadata: { completion_source: 'service_cleanup' },
			});
		}

		this.gatewayRegistry.disconnectAll();

		this.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.traceContextsByRunId.clear();

		this.eventBus.clear();
		await this._mcpClientManager?.disconnect();
		this.logger.debug('Instance AI service shut down');
	}

	@OnLeaderTakeover()
	startCheckpointPruning(): void {
		if (this.checkpointPruneTimer || this.instanceAiConfig.snapshotPruneInterval <= 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.snapshotPruneInterval): void {
		if (this.checkpointPruningStopped) return;
		this.checkpointPruneTimer = setTimeout(() => {
			void this.pruneStaleCheckpoints();
		}, 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<unknown>): 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<InstanceAiService['executeRun']>): void {
		this.trackInFlightExecution(this.executeRun(...args));
	}

	/** Same shutdown-drain contract as `startExecuteRun`, for the resume path. */
	private startProcessResumedStream(
		...args: Parameters<InstanceAiService['processResumedStream']>
	): 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);
	}

	/**
	 * Idempotent first-suspend persistence. The first `waitForConfirmation`
	 * for a run consumes the pending entry; later HITLs are no-ops because
	 * the entry has been removed. If the DB write fails the entry stays in
	 * the map so the next HITL retries — see `persistUserMessageOnSuspend`'s
	 * success-boolean return for the retry contract.
	 */
	private async persistUserMessageOnFirstSuspend(threadId: string, runId: string): Promise<void> {
		const entry = this.userMessagePersistenceByRun.get(runId);
		if (!entry) return;
		const ok = await this.persistUserMessageOnSuspend(threadId, entry.userId, entry.message);
		if (ok) this.userMessagePersistenceByRun.delete(runId);
	}

	private async drainInFlightExecutions(timeoutMs: number): Promise<void> {
		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,
			});
		}
	}

	private async pruneStaleCheckpoints(now = Date.now()): Promise<void> {
		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.pruneStalePendingConfirmations(now);
			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);
		}
	}

	private async pruneStalePendingConfirmations(now: number): Promise<void> {
		try {
			const count = await this.pendingConfirmationRepo.deleteExpired(new Date(now));
			if (count === 0) {
				this.logger.debug('No stale Instance AI pending confirmations to drop');
				return;
			}
			this.logger.info('Dropped stale Instance AI pending confirmations', { count });
		} catch (error: unknown) {
			this.logger.warn('Failed to drop stale Instance AI pending confirmations', {
				error: getErrorMessage(error),
			});
		}
	}

	private computePendingConfirmationExpiresAt(): Date | null {
		const timeoutMs = this.instanceAiConfig.confirmationTimeout;
		return timeoutMs > 0 ? new Date(Date.now() + timeoutMs) : null;
	}

	/**
	 * Persist the index for a HITL confirmation so a fresh process can find it
	 * after the in-memory `pendingConfirmations` / `suspendedRuns` maps are gone.
	 * Fire-and-forget: a DB write failure must not block the agent flow, which
	 * still operates correctly via the in-memory state on this main.
	 */
	private async persistPendingConfirmation(params: {
		requestId: string;
		threadId: string;
		userId: string;
		runId: string;
		messageGroupId?: string;
		kind: 'inline' | 'suspended';
		toolCallId?: string;
		checkpointKey?: string;
		checkpointTaskId?: string;
	}): Promise<void> {
		try {
			await this.pendingConfirmationRepo.save(
				this.pendingConfirmationRepo.create({
					requestId: params.requestId,
					threadId: params.threadId,
					userId: params.userId,
					kind: params.kind,
					runId: params.runId,
					messageGroupId: params.messageGroupId ?? null,
					toolCallId: params.toolCallId ?? null,
					checkpointKey: params.checkpointKey ?? null,
					checkpointTaskId: params.checkpointTaskId ?? null,
					expiresAt: this.computePendingConfirmationExpiresAt(),
				}),
			);
		} catch (error: unknown) {
			this.logger.warn('Failed to persist pending confirmation', {
				requestId: params.requestId,
				threadId: params.threadId,
				kind: params.kind,
				error: getErrorMessage(error),
			});
		}
	}

	private async dropPendingConfirmation(requestId: string): Promise<void> {
		try {
			await this.pendingConfirmationRepo.deleteByRequestId(requestId);
		} catch (error: unknown) {
			this.logger.warn('Failed to drop pending confirmation', {
				requestId,
				error: getErrorMessage(error),
			});
		}
	}

	private async dropPendingConfirmationsForThread(threadId: string): Promise<void> {
		try {
			await this.pendingConfirmationRepo.deleteByThreadId(threadId);
		} catch (error: unknown) {
			this.logger.warn('Failed to drop pending confirmations for thread', {
				threadId,
				error: getErrorMessage(error),
			});
		}
	}

	/**
	 * Persist the original user-typed message to thread memory the first time
	 * a run hits an inline HITL confirmation, so the message bubble survives
	 * a restart that happens while the run is suspended. The `id` matches the
	 * one we pass to the SDK's streamInput, so the SDK's eventual end-of-turn
	 * save (if the run does resume and complete) upserts the same row instead
	 * of creating a duplicate.
	 */
	private async persistUserMessageOnSuspend(
		threadId: string,
		userId: string,
		message: { id: string; text: string },
	): Promise<boolean> {
		try {
			await this.agentMemory.saveMessages({
				threadId,
				resourceId: userId,
				messages: [
					{
						id: message.id,
						role: 'user',
						content: [{ type: 'text', text: message.text }],
						createdAt: new Date(),
					},
				],
			});
			return true;
		} catch (error: unknown) {
			this.logger.warn('Failed to persist user message on HITL suspend', {
				threadId,
				userId,
				error: getErrorMessage(error),
			});
			return false;
		}
	}

	/**
	 * Save the in-flight agent tree as a terminal snapshot so the UI doesn't
	 * sit on a half-rendered turn after the process restarts. Best-effort: a
	 * DB write failure here must not block the rest of shutdown.
	 */
	private async persistShutdownSnapshot(
		threadId: string,
		runId: string,
		messageGroupId: string | undefined,
	): Promise<void> {
		try {
			await this.saveAgentTreeSnapshot(
				threadId,
				runId,
				this.dbSnapshotStorage,
				true,
				messageGroupId,
			);
		} catch (error: unknown) {
			this.logger.warn('Failed to persist shutdown snapshot', {
				threadId,
				runId,
				error: getErrorMessage(error),
			});
		}
	}

	private createAgentMemoryOptions() {
		return {
			lastMessages: this.instanceAiConfig.lastMessages,
			observationalMemory: {
				observerThresholdTokens: this.instanceAiConfig.observerMessageTokens,
				reflectorThresholdTokens: this.instanceAiConfig.reflectorObservationTokens,
			},
		};
	}

	private async ensureThreadExists(
		memory: BuiltMemory,
		threadId: string,
		resourceId: string,
	): Promise<void> {
		const existingThread = await memory.getThread(threadId);
		if (existingThread) return;

		await memory.saveThread({
			id: threadId,
			resourceId,
			title: '',
		});
	}

	private projectPlannedTaskList(graph: PlannedTaskGraph): TaskList {
		return {
			tasks: graph.tasks.map((task) => ({
				id: task.id,
				description: task.title,
				status:
					task.status === 'planned'
						? 'todo'
						: task.status === 'running'
							? 'in_progress'
							: task.status === 'succeeded'
								? 'done'
								: task.status,
			})),
		};
	}

	private buildPlannedTaskFollowUpMessage(
		type: 'synthesize' | 'replan' | 'checkpoint',
		graph: PlannedTaskGraph,
		options: { failedTask?: PlannedTaskRecord; checkpoint?: PlannedTaskRecord } = {},
	): string {
		const payload: Record<string, unknown> = {
			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,
			};
		}

		return `<planned-task-follow-up type="${type}">\n${JSON.stringify(payload, null, 2)}\n</planned-task-follow-up>\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 };
	}

	private evaluateTerminalResponse(
		threadId: string,
		runId: string,
		status: Exclude<TerminalResponseStatus, 'waiting'>,
		options: {
			messageGroupId?: string;
			correlationId?: string;
			workSummary?: WorkSummary;
			errorMessage?: string;
		} = {},
	): TerminalResponseDecision | undefined {
		const guard = new InstanceAiTerminalResponseGuard({
			runId,
			rootAgentId: ORCHESTRATOR_AGENT_ID,
			messageGroupId: options.messageGroupId,
			correlationId: options.correlationId,
		});
		const decision = guard.evaluateTerminal(
			this.getTerminalGuardEvents(threadId, runId, options.messageGroupId),
			status,
			{
				workSummary: options.workSummary,
				errorMessage: options.errorMessage,
			},
		);
		this.handleTerminalResponseDecision(threadId, runId, decision, options.messageGroupId);
		return decision;
	}

	private evaluateWaitingResponse(
		threadId: string,
		runId: string,
		confirmationEvent: Extract<InstanceAiEvent, { type: 'confirmation-request' }> | undefined,
		options: { messageGroupId?: string; correlationId?: string } = {},
	): TerminalResponseDecision | undefined {
		const guard = new InstanceAiTerminalResponseGuard({
			runId,
			rootAgentId: ORCHESTRATOR_AGENT_ID,
			messageGroupId: options.messageGroupId,
			correlationId: options.correlationId,
		});
		const decision = guard.evaluateWaiting(
			this.getTerminalGuardEvents(threadId, runId, options.messageGroupId),
			confirmationEvent,
		);
		this.handleTerminalResponseDecision(threadId, runId, decision, options.messageGroupId);
		return decision;
	}

	private getTerminalGuardEvents(
		threadId: string,
		runId: string,
		messageGroupId?: string,
	): InstanceAiEvent[] {
		if (!messageGroupId) return this.eventBus.getEventsForRun(threadId, runId);

		const groupRunIds = this.getRunIdsForMessageGroup(messageGroupId);
		return groupRunIds.length > 0
			? this.eventBus.getEventsForRuns(threadId, groupRunIds)
			: this.eventBus.getEventsForRun(threadId, runId);
	}

	private handleTerminalResponseDecision(
		threadId: string,
		runId: string,
		decision: TerminalResponseDecision,
		messageGroupId?: string,
	): void {
		this.telemetry.track('instance_ai_terminal_response_decision', {
			thread_id: threadId,
			run_id: runId,
			message_group_id: messageGroupId,
			source: 'terminal_guard',
			status: decision.status,
			action: decision.action,
			reason: decision.reason,
			visibility_source: decision.visibilitySource,
		});

		if (decision.reason === 'completed-after-error') {
			this.logger.warn('completed_after_error_event', {
				threadId,
				runId,
				messageGroupId,
			});
		}

		if (decision.reason === 'confirmation-invalid') {
			this.logger.warn('invalid_confirmation_payload', {
				threadId,
				runId,
				messageGroupId,
			});
		}

		if (decision.action === 'emit' && decision.event) {
			this.eventBus.publish(threadId, decision.event);
		}
	}

	private createTerminalOutcomeStorage(): TerminalOutcomeStorage {
		this.terminalOutcomeStorage ??= new TerminalOutcomeStorage(this.agentMemory);
		return this.terminalOutcomeStorage;
	}

	private async finishInvalidConfirmationRun(args: {
		threadId: string;
		runId: string;
		abortController: AbortController;
		snapshotStorage: DbSnapshotStorage;
		tracing?: InstanceAiTraceContext;
	}): Promise<MessageTraceFinalization> {
		this.runState.cancelThread(args.threadId);
		void this.dropPendingConfirmationsForThread(args.threadId);
		args.abortController.abort();
		await this.finalizeRunTracing(args.runId, args.tracing, {
			status: 'error',
			reason: 'invalid_confirmation_payload',
		});
		this.publishRunFinish(
			args.threadId,
			args.runId,
			'errored',
			'I need your input to continue, but I could not display the prompt. Please try again.',
		);
		await this.saveAgentTreeSnapshot(args.threadId, args.runId, args.snapshotStorage);
		return {
			status: 'error',
			reason: 'invalid_confirmation_payload',
			metadata: this.buildMessageTraceMetadata(args.threadId, args.runId, {
				status: 'error',
			}),
		};
	}

	private buildBackgroundTerminalOutcome(task: ManagedBackgroundTask): TerminalOutcome {
		const status =
			task.status === 'failed' ? 'failed' : task.status === 'cancelled' ? 'cancelled' : 'completed';
		const userFacingMessage =
			status === 'completed'
				? `The background ${task.role} task finished.`
				: status === 'cancelled'
					? `The background ${task.role} task was cancelled.`
					: `The background ${task.role} task failed before I could complete that part.`;

		return {
			id: `${task.messageGroupId ?? task.runId}:${task.taskId}:${status}`,
			threadId: task.threadId,
			runId: task.runId,
			messageGroupId: task.messageGroupId,
			correlationId: task.messageGroupId,
			taskId: task.taskId,
			agentId: task.agentId,
			status,
			userFacingMessage,
			createdAt: new Date().toISOString(),
		};
	}

	async replayUndeliveredTerminalOutcomes(
		threadId: string,
		options: { delivery?: 'snapshot' | 'event' } = {},
	): Promise<void> {
		const storage = this.createTerminalOutcomeStorage();
		const persistedOutcomes = await storage.getUndelivered(threadId).catch((error) => {
			this.logger.warn('Failed to load undelivered Instance AI terminal outcomes', {
				threadId,
				error: getErrorMessage(error),
			});
			return [] as TerminalOutcome[];
		});
		const inMemoryOutcomes = [...this.pendingTerminalOutcomes.values()].filter(
			(outcome) => outcome.threadId === threadId,
		);
		const outcomes = new Map<string, TerminalOutcome>();
		for (const outcome of [...persistedOutcomes, ...inMemoryOutcomes]) {
			outcomes.set(outcome.id, outcome);
		}
		const persistedOutcomeIds = new Set(persistedOutcomes.map((outcome) => outcome.id));
		const delivery = options.delivery ?? 'snapshot';

		for (const outcome of outcomes.values()) {
			const responseId = getBackgroundOutcomeResponseId(outcome);
			let snapshotDelivered = false;
			try {
				snapshotDelivered = await this.persistTerminalOutcomeLineToSnapshot(outcome, responseId);
			} catch (error) {
				this.logger.warn('Failed to replay Instance AI terminal outcome', {
					threadId,
					runId: outcome.runId,
					taskId: outcome.taskId,
					error: getErrorMessage(error),
				});
				if (delivery === 'event') {
					const published = this.publishTerminalOutcomeLine(outcome, responseId);
					this.telemetry.track('instance_ai_terminal_response_decision', {
						thread_id: threadId,
						run_id: outcome.runId,
						message_group_id: outcome.messageGroupId,
						task_id: outcome.taskId,
						source: 'terminal_outcome_replay',
						status: outcome.status,
						action: published ? 'replay_event' : 'already-emitted',
						visibility_source: 'background-outcome',
					});
				}
				continue;
			}

			if (!snapshotDelivered) continue;

			let action = 'replay_snapshot';
			if (delivery === 'event') {
				const published = this.publishTerminalOutcomeLine(outcome, responseId);
				action = published ? 'replay_event' : 'already-emitted';
			}

			if (persistedOutcomeIds.has(outcome.id)) {
				await storage
					.markDelivered(threadId, outcome.id, new Date().toISOString())
					.catch((error) => {
						this.logger.warn('Failed to mark Instance AI terminal outcome as delivered', {
							threadId,
							runId: outcome.runId,
							taskId: outcome.taskId,
							error: getErrorMessage(error),
						});
					});
			}
			this.pendingTerminalOutcomes.delete(outcome.id);
			this.telemetry.track('instance_ai_terminal_response_decision', {
				thread_id: threadId,
				run_id: outcome.runId,
				message_group_id: outcome.messageGroupId,
				task_id: outcome.taskId,
				source: 'terminal_outcome_replay',
				status: outcome.status,
				action,
				visibility_source: 'background-outcome',
			});
		}
	}

	private async persistTerminalOutcomeLineToSnapshot(
		outcome: TerminalOutcome,
		responseId: string,
	): Promise<boolean> {
		const snapshot = await this.dbSnapshotStorage.getLatest(outcome.threadId, {
			messageGroupId: outcome.messageGroupId,
			runId: outcome.runId,
		});
		if (!snapshot) {
			await this.dbSnapshotStorage.save(
				outcome.threadId,
				createTerminalOutcomeAgentTree(outcome, responseId),
				outcome.runId,
				{
					messageGroupId: outcome.messageGroupId,
					runIds: [outcome.runId],
				},
			);
			return true;
		}

		const { tree } = appendTerminalOutcomeToAgentTree(snapshot.tree, outcome, responseId);
		const runIds = new Set(snapshot.runIds ?? [snapshot.runId]);
		runIds.add(outcome.runId);
		await this.dbSnapshotStorage.updateLast(outcome.threadId, tree, snapshot.runId, {
			messageGroupId: snapshot.messageGroupId ?? outcome.messageGroupId,
			runIds: [...runIds],
			langsmithRunId: snapshot.langsmithRunId,
			langsmithTraceId: snapshot.langsmithTraceId,
		});
		return true;
	}

	private publishTerminalOutcomeLine(outcome: TerminalOutcome, responseId: string): boolean {
		const alreadyPublished = this.eventBus
			.getEventsForRun(outcome.threadId, outcome.runId)
			.some((event) => event.responseId === responseId);
		if (alreadyPublished) return false;

		this.eventBus.publish(outcome.threadId, {
			type: 'text-delta',
			runId: outcome.runId,
			agentId: ORCHESTRATOR_AGENT_ID,
			responseId,
			payload: { text: outcome.userFacingMessage },
		});
		return true;
	}

	private async recordBackgroundTerminalOutcome(task: ManagedBackgroundTask): Promise<void> {
		const outcome = this.buildBackgroundTerminalOutcome(task);
		let persisted = false;
		try {
			await this.createTerminalOutcomeStorage().upsert(task.threadId, outcome);
			persisted = true;
		} catch (error) {
			this.pendingTerminalOutcomes.set(outcome.id, outcome);
			this.logger.warn('Failed to persist Instance AI terminal outcome', {
				threadId: task.threadId,
				runId: task.runId,
				taskId: task.taskId,
				error: getErrorMessage(error),
			});
			this.telemetry.track('instance_ai_terminal_outcome_persistence_failure', {
				thread_id: task.threadId,
				run_id: task.runId,
				task_id: task.taskId,
				status: outcome.status,
				phase: 'metadata',
			});
		}

		const responseId = getBackgroundOutcomeResponseId(outcome);
		const published = this.publishTerminalOutcomeLine(outcome, responseId);

		this.telemetry.track('instance_ai_terminal_response_decision', {
			thread_id: task.threadId,
			run_id: task.runId,
			message_group_id: task.messageGroupId,
			task_id: task.taskId,
			source: 'background_outcome',
			status: outcome.status,
			action: published ? 'emit' : 'already-emitted',
			visibility_source: 'background-outcome',
		});

		let snapshotDelivered = false;
		try {
			snapshotDelivered = await this.persistTerminalOutcomeLineToSnapshot(outcome, responseId);
		} catch (error) {
			this.logger.warn('Failed to persist Instance AI terminal outcome line to snapshot', {
				threadId: task.threadId,
				runId: task.runId,
				taskId: task.taskId,
				error: getErrorMessage(error),
			});
			this.telemetry.track('instance_ai_terminal_outcome_persistence_failure', {
				thread_id: task.threadId,
				run_id: task.runId,
				task_id: task.taskId,
				status: outcome.status,
				phase: 'snapshot',
			});
		}

		if (!persisted || !snapshotDelivered) return;

		try {
			await this.createTerminalOutcomeStorage().markDelivered(
				task.threadId,
				outcome.id,
				new Date().toISOString(),
			);
			this.pendingTerminalOutcomes.delete(outcome.id);
		} catch (error) {
			this.logger.warn('Failed to mark Instance AI terminal outcome as delivered', {
				threadId: task.threadId,
				runId: task.runId,
				taskId: task.taskId,
				error: getErrorMessage(error),
			});
		}
	}

	private async syncPlannedTasksToUi(threadId: string, graph: PlannedTaskGraph): Promise<void> {
		const { taskStorage } = await this.createPlannedTaskState();
		const tasks = this.projectPlannedTaskList(graph);
		await taskStorage.save(threadId, tasks);
		this.eventBus.publish(threadId, {
			type: 'tasks-update',
			runId: graph.planRunId,
			agentId: ORCHESTRATOR_AGENT_ID,
			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<void> {
		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: ORCHESTRATOR_AGENT_ID,
				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,
	) {
		const adminSettings = this.settingsService.getAdminSettings();
		const localGatewayDisabledGlobally = adminSettings.localGatewayDisabled;
		const localGatewayDisabledForUser = await this.settingsService.isLocalGatewayDisabledForUser(
			user.id,
		);
		const userGateway = this.gatewayRegistry.findGateway(user.id);

		// When the proxy is enabled, create a single ProxyTokenManager and
		// AiAssistantClient that are shared across model, search, and tracing
		// configs.  The token manager caches the JWT and refreshes it
		// transparently before it expires.
		let searchProxyConfig: ServiceProxyConfig | undefined;
		let tracingProxyConfig: ServiceProxyConfig | undefined;
		let tokenManager: ProxyTokenManager | undefined;
		let proxyBaseUrl: string | undefined;
		if (this.aiService.isProxyEnabled()) {
			const client = await this.aiService.getClient();
			proxyBaseUrl = client.getApiProxyBaseUrl();
			const manager = new ProxyTokenManager(async () => {
				return await client.getBuilderApiProxyToken({ id: user.id }, { userMessageId: nanoid() });
			});
			tokenManager = manager;
			const featureHeaders = buildProxyHeaders({
				feature: 'instance-ai',
				n8nVersion: N8N_VERSION,
			});
			searchProxyConfig = {
				apiUrl: proxyBaseUrl + '/brave-search',
				getAuthHeaders: async () => ({
					...(await manager.getAuthHeaders()),
					...featureHeaders,
				}),
			};
			tracingProxyConfig = {
				apiUrl: proxyBaseUrl + '/langsmith',
				getAuthHeaders: async () => ({
					...(await manager.getAuthHeaders()),
					...featureHeaders,
				}),
			};
		}

		const context = this.adapterService.createContext(user, {
			searchProxyConfig,
			pushRef,
			threadId,
		});
		if (!localGatewayDisabledForUser && userGateway?.isConnected) {
			context.localMcpServer = userGateway;
		}
		context.permissions = this.settingsService.getPermissions();
		if (this.sourceControlPreferencesService.getPreferences().branchReadOnly) {
			context.permissions = applyBranchReadOnlyOverrides(context.permissions);
			context.branchReadOnly = true;
		}

		let domainTracker = this.domainAccessTrackersByThread.get(threadId);
		if (!domainTracker) {
			domainTracker = createDomainAccessTracker();
			this.domainAccessTrackersByThread.set(threadId, domainTracker);
		}
		context.domainAccessTracker = domainTracker;
		context.runId = runId;

		// Compute gateway status for the system prompt
		if (localGatewayDisabledGlobally) {
			context.localGatewayStatus = { status: 'disabledGlobally' };
		} else if (!localGatewayDisabledForUser && userGateway?.isConnected) {
			context.localGatewayStatus = {
				status: 'connected',
				capabilities: userGateway
					.getStatus()
					.toolCategories.filter(({ enabled }) => enabled)
					.map(({ name }) => name),
			};
		} else {
			context.localGatewayStatus = {
				status: localGatewayDisabledForUser ? 'disabled' : 'disconnected',
			};
		}

		const modelId =
			proxyBaseUrl && tokenManager
				? await this.resolveProxyModel(user, proxyBaseUrl, tokenManager)
				: await this.resolveAgentModelConfig(user);
		const memory = this.agentMemory;
		await this.ensureThreadExists(memory, threadId, user.id);

		const taskStorage = new ThreadTaskStorage(memory);
		const iterationLog = this.dbIterationLogStorage;
		const snapshotStorage = this.dbSnapshotStorage;
		const workflowLoopStorage = new WorkflowLoopStorage(memory);
		const workflowTasks = 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);
		}

		const domainTools = createAllTools(context);
		const baseRuntimeSkills = loadInstanceAiRuntimeSkillSource();
		let runtimeSkills = baseRuntimeSkills;
		let runtimeWorkspace: Workspace | undefined;
		if (adminSettings.sandboxEnabled) {
			let sandboxEntryPromise: Promise<RuntimeSandboxEntry | undefined> | undefined;
			const getSandboxEntry = async () => {
				sandboxEntryPromise ??= this.getOrCreateWorkspaceEntry(threadId, user, runId).catch(
					(error: unknown) => {
						sandboxEntryPromise = undefined;
						throw error;
					},
				);

				return await sandboxEntryPromise;
			};
			const getSetupSandboxEntry = async () => {
				return await this.getOrCreateWorkspace(threadId, user, context, runId);
			};

			runtimeWorkspace = createLazyRuntimeWorkspace({
				ensureWorkspace: async () => (await getSetupSandboxEntry())?.workspace,
			});
			const runtimeSkillWorkspace = createLazyRuntimeWorkspace({
				id: 'instance-ai-runtime-skill-workspace',
				name: 'Instance AI runtime skill workspace',
				ensureWorkspace: async () => (await getSandboxEntry())?.workspace,
			});
			runtimeSkills = createLazyWorkspaceRuntimeSkillSource({
				source: baseRuntimeSkills,
				workspace: runtimeSkillWorkspace,
				logger: this.logger,
			});
		}

		const orchestrationContext: OrchestrationContext = {
			threadId,
			runId,
			messageGroupId,
			userId: user.id,
			orchestratorAgentId: ORCHESTRATOR_AGENT_ID,
			modelId,
			checkpointStore: this.checkpointStore,
			subAgentMaxSteps: this.instanceAiConfig.subAgentMaxSteps,
			eventBus: this.eventBus,
			logger: this.logger,
			trackTelemetry: (eventName, properties) => {
				this.telemetry.track(eventName, properties);
			},
			domainTools,
			abortSignal,
			taskStorage,
			timeZone: this.defaultTimeZone,
			localMcpServer: context.localMcpServer,
			runtimeSkills,
			runtimeSkillCatalog: baseRuntimeSkills,
			oauth2CallbackUrl: this.oauth2CallbackUrl,
			webhookBaseUrl: this.webhookBaseUrl,
			formBaseUrl: this.formBaseUrl,
			waitForConfirmation: async (requestId: string) => {
				this.runState.touchActiveRun(threadId);
				// First HITL on this run: persist the original user message to
				// memory so it survives a restart-while-suspended. The SDK only
				// commits the turn delta on a clean loop completion, and inline
				// HITL never reaches that point. Doing this *now* (rather than
				// at executeRun start) avoids polluting the agent's prompt —
				// the SDK has already loaded its memory snapshot for this run.
				void this.persistUserMessageOnFirstSuspend(threadId, runId);

				return await new Promise<ConfirmationData>((resolve) => {
					this.runState.registerPendingConfirmation(requestId, {
						resolve,
						threadId,
						userId: user.id,
						createdAt: Date.now(),
					});

					void this.persistPendingConfirmation({
						requestId,
						threadId,
						userId: user.id,
						runId,
						messageGroupId,
						kind: 'inline',
					});

					// Inline HITL (planner questions / plan approval / sub-agent asks)
					// keeps the orchestrator run active, so the normal suspended/completed
					// snapshot paths do not execute. Queue a snapshot after the current
					// confirmation-request event is published to preserve refresh recovery.
					queueMicrotask(() => {
						void this.saveAgentTreeSnapshot(threadId, runId, snapshotStorage);
					});
				});
			},
			cancelBackgroundTask: async (taskId) => this.cancelBackgroundTask(threadId, taskId),
			spawnBackgroundTask: (opts) =>
				this.spawnBackgroundTask(runId, opts, snapshotStorage, 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,
			nodeDefinitionDirs: nodeDefDirs.length > 0 ? nodeDefDirs : undefined,
			domainContext: context,
			tracingProxyConfig,
			memory,
		};

		return {
			context,
			memory,
			taskStorage,
			iterationLog,
			snapshotStorage,
			workflowTasks,
			plannedTaskService,
			modelId,
			orchestrationContext,
		};
	}

	private async dispatchPlannedTask(
		task: PlannedTaskRecord,
		context: OrchestrationContext,
		graph?: PlannedTaskGraph,
	): Promise<void> {
		// Plan approval authorizes the task-family's non-destructive tools,
		// so the sub-agent can execute without a redundant second confirmation.
		const taskContext = this.createPlannedTaskContext(task.kind, context);
		const conversationContext = buildPlannedTaskConversationContext(task, graph);

		let started: { taskId: string; agentId: string; result: string } | null = null;

		switch (task.kind) {
			case 'build-workflow':
				started = await startBuildWorkflowAgentTask(taskContext, {
					task: task.spec,
					workflowId: task.workflowId,
					plannedTaskId: task.id,
					conversationContext,
				});
				break;
			case 'delegate':
				started = await startDetachedDelegateTask(taskContext, {
					title: task.title,
					spec: task.spec,
					tools: task.tools ?? [],
					plannedTaskId: task.id,
					conversationContext,
				});
				break;
		}

		if (!started?.taskId) {
			await context.plannedTaskService?.markFailed(context.threadId, task.id, {
				error: started?.result || `Failed to start planned task "${task.title}"`,
			});
			return;
		}

		await context.plannedTaskService?.markRunning(context.threadId, task.id, {
			agentId: started.agentId,
			backgroundTaskId: started.taskId,
		});

		const nextGraph = await context.plannedTaskService?.getGraph(context.threadId);
		if (nextGraph) {
			await this.syncPlannedTasksToUi(context.threadId, nextGraph);
		}
	}

	/**
	 * Creates a task-scoped OrchestrationContext with plan-approved permission
	 * overrides. Rebuilds domain tools so each sub-agent gets its own closure
	 * with the correct permissions, preventing cross-task leakage.
	 */
	private createPlannedTaskContext(
		kind: PlannedTaskRecord['kind'],
		context: OrchestrationContext,
	): OrchestrationContext {
		if (!context.domainContext) return context;

		const taskDomainContext = applyPlannedTaskPermissions(context.domainContext, kind);
		if (taskDomainContext === context.domainContext) return context;

		return {
			...context,
			domainContext: taskDomainContext,
			domainTools: createAllTools(taskDomainContext),
		};
	}

	/**
	 * Resolve the workflow IDs the checkpoint task is verifying so the runWorkflow
	 * permission override can be scoped. Walks the checkpoint's `dependsOn` to find
	 * the build-workflow tasks it depends on and reads their `outcome.workflowId`.
	 * Returns an empty set when the graph is missing or the checkpoint has no
	 * resolved workflow deps (in which case the override applies broadly via the
	 * `allowList === undefined` short-circuit only if we don't set the field).
	 */
	private async getCheckpointAllowedWorkflowIds(
		threadId: string,
		checkpointTaskId: string,
	): Promise<ReadonlySet<string>> {
		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 new Set();
			const deps = new Set(checkpoint.deps);
			const allowed = new Set<string>();
			for (const task of graph.tasks) {
				if (!deps.has(task.id)) continue;
				const workflowId = task.outcome?.workflowId;
				if (typeof workflowId === 'string' && workflowId.length > 0) {
					allowed.add(workflowId);
				}
			}
			return allowed;
		} catch (error) {
			this.logger.warn('Failed to resolve checkpoint allowed workflow IDs', {
				threadId,
				checkpointTaskId,
				error: error instanceof Error ? error.message : String(error),
			});
			return new Set();
		}
	}

	private async handlePlannedTaskSettlement(
		user: User,
		task: ManagedBackgroundTask,
		status: 'succeeded' | 'failed' | 'cancelled',
	): Promise<void> {
		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);
		}

		await this.schedulePlannedTasks(user, task.threadId);
	}

	private async startInternalFollowUpRun(
		user: User,
		threadId: string,
		message: string,
		messageGroupId?: string,
		isReplanFollowUp: boolean = false,
		checkpoint?: { isCheckpointFollowUp: true; checkpointTaskId: string },
	): Promise<string> {
		if (this.runState.hasLiveRun(threadId)) {
			this.logger.warn('Skipping internal follow-up: active run exists', { threadId });
			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
		// the planner emit "instance default timezone" for user-local schedules.
		const timeZone = this.runState.getTimeZone(threadId) ?? this.defaultTimeZone;
		const resumeReason: OrchestratorResumeReason = checkpoint
			? 'planned_checkpoint'
			: isReplanFollowUp
				? 'replan'
				: 'background_task_completed';

		this.startExecuteRun(
			user,
			threadId,
			runId,
			message,
			abortController,
			undefined,
			messageGroupId,
			timeZone,
			isReplanFollowUp,
			checkpoint,
			resumeReason,
		);

		return runId;
	}

	private async schedulePlannedTasks(user: User, threadId: string): Promise<void> {
		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 async doSchedulePlannedTasks(user: User, threadId: string): Promise<void> {
		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 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 action = await plannedTaskService.tick(threadId, { availableSlots });
		if (action.type === 'none') return;

		if (action.type === 'replan') {
			await this.syncPlannedTasksToUi(threadId, action.graph);
			const startedRunId = await this.startInternalFollowUpRun(
				activeUser,
				threadId,
				this.buildPlannedTaskFollowUpMessage('replan', action.graph, {
					failedTask: action.failedTask,
				}),
				action.graph.messageGroupId,
				true,
			);
			// tick() already transitioned the graph to `awaiting_replan`. If the
			// follow-up run couldn't start (live run present), revert the status
			// so the next schedulePlannedTasks() pass can re-emit this action.
			// Without this, tick() returns `none` for non-active graphs and the
			// replan is silently lost.
			if (!startedRunId) {
				await plannedTaskService.revertToActive(threadId);
			}
			return;
		}

		if (action.type === 'synthesize') {
			await this.syncPlannedTasksToUi(threadId, action.graph);
			const startedRunId = await this.startInternalFollowUpRun(
				activeUser,
				threadId,
				this.buildPlannedTaskFollowUpMessage('synthesize', action.graph),
				action.graph.messageGroupId,
			);
			// Same rollback as replan: tick() transitioned to `completed`, but if
			// the synthesize follow-up didn't actually start, revert so the next
			// tick can emit it again.
			if (!startedRunId) {
				await plannedTaskService.revertToActive(threadId);
			}
			return;
		}

		if (action.type === 'orchestrate-checkpoint') {
			// Defer if a run is already active or suspended. The currently-live
			// run's post-finally reschedule hook will pick this checkpoint up.
			if (this.runState.hasLiveRun(threadId)) {
				return;
			}

			const checkpoint = action.tasks[0];

			// Mark running before starting the follow-up so complete-checkpoint
			// (which requires status === 'running') always sees the correct state.
			// If startInternalFollowUpRun no-ops below (tight race), we roll back
			// the transition to avoid leaving the task in a phantom 'running' state.
			await plannedTaskService.markRunning(threadId, checkpoint.id, {
				agentId: ORCHESTRATOR_AGENT_ID,
			});
			const graphAfterMark = (await plannedTaskService.getGraph(threadId)) ?? action.graph;
			await this.syncPlannedTasksToUi(threadId, graphAfterMark);

			const checkpointRecord =
				graphAfterMark.tasks.find((t) => t.id === checkpoint.id) ?? checkpoint;

			const startedRunId = await this.startInternalFollowUpRun(
				activeUser,
				threadId,
				this.buildPlannedTaskFollowUpMessage('checkpoint', graphAfterMark, {
					checkpoint: checkpointRecord,
				}),
				action.graph.messageGroupId,
				false,
				{ isCheckpointFollowUp: true, checkpointTaskId: checkpoint.id },
			);

			if (!startedRunId) {
				// Rare race: the outer hasLiveRun check passed but the inner guard
				// in startInternalFollowUpRun did not (another path started a run
				// between our two checks). Revert the checkpoint back to `planned`
				// so the next scheduler tick re-emits `orchestrate-checkpoint` —
				// marking it `failed` here would cascade cancel to every dependent
				// and destroy downstream work even though nothing actually failed.
				this.logger.warn(
					'Checkpoint follow-up run did not start — reverting checkpoint to planned for retry',
					{ threadId, checkpointTaskId: checkpoint.id },
				);
				await plannedTaskService.revertCheckpointToPlanned(threadId, checkpoint.id);
			}
			return;
		}

		const environment = await this.createExecutionEnvironment(
			activeUser,
			threadId,
			action.graph.planRunId,
			createInertAbortSignal(),
			action.graph.messageGroupId,
			// Route planned-task workflow runs (build agent, checkpoint verifications)
			// to the user's iframe session so live execution push events reach the
			// frontend, matching the orchestrator main-run path.
			this.threadPushRef.get(threadId),
		);
		environment.orchestrationContext.tracing = this.getTraceContext(action.graph.planRunId);

		for (const task of action.tasks) {
			await this.dispatchPlannedTask(task, environment.orchestrationContext, action.graph);
		}

		await this.doSchedulePlannedTasks(activeUser, threadId);
	}

	/**
	 * 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.
	 */
	private async executeRun(
		user: User,
		threadId: string,
		runId: string,
		message: string,
		abortController: AbortController,
		attachments?: InstanceAiAttachment[],
		messageGroupId?: string,
		timeZone?: string,
		isReplanFollowUp: boolean = false,
		checkpoint?: { isCheckpointFollowUp: true; checkpointTaskId: string },
		resumeReason?: OrchestratorResumeReason,
	): Promise<void> {
		// Read once at the top so the streamInput builder + (if any later
		// retry) see the same view of restart-recovery metadata.
		const userMessagePersistence = this.userMessagePersistenceByRun.get(runId)?.message;
		const signal = abortController.signal;
		let tracing: InstanceAiTraceContext | undefined;
		let messageTraceFinalization: MessageTraceFinalization | undefined;
		let aiCreatedWorkflowIds: Set<string> | undefined;
		let activeSnapshotStorage: DbSnapshotStorage | undefined;
		let messageId = '';

		try {
			messageId = nanoid();

			// Publish run-start (includes userId for audit trail attribution)
			this.eventBus.publish(threadId, {
				type: 'run-start',
				runId,
				agentId: ORCHESTRATOR_AGENT_ID,
				userId: user.id,
				payload: { messageId, messageGroupId },
			});

			// Check if already cancelled before starting agent work
			if (signal.aborted) {
				this.evaluateTerminalResponse(threadId, runId, 'cancelled', {
					messageGroupId,
					correlationId: messageId,
				});
				this.eventBus.publish(threadId, {
					type: 'run-finish',
					runId,
					agentId: ORCHESTRATOR_AGENT_ID,
					payload: { status: 'cancelled', reason: 'user_cancelled' },
				});
				return;
			}

			const mcpServers = this.parseMcpServers(this.instanceAiConfig.mcpServers);

			const executionPushRef = this.threadPushRef.get(threadId);
			const environment = await this.createExecutionEnvironment(
				user,
				threadId,
				runId,
				signal,
				messageGroupId,
				executionPushRef,
			);
			activeSnapshotStorage = environment.snapshotStorage;
			const { context, memory, taskStorage, snapshotStorage, modelId, orchestrationContext } =
				environment;
			aiCreatedWorkflowIds = context.aiCreatedWorkflowIds ??= new Set<string>();
			// Make the current user message available to sub-agents (e.g. planner)
			// 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.
				context.allowedRunWorkflowIds = await this.getCheckpointAllowedWorkflowIds(
					threadId,
					checkpoint.checkpointTaskId,
				);
			}

			// Thread attachments into the domain context so parse-file can access them
			if (attachments && attachments.length > 0) {
				context.currentUserAttachments = attachments;
			}
			const memoryConfig = this.createAgentMemoryOptions();
			const traceInput = {
				message,
				...(attachments?.length
					? {
							attachments: attachments.map((attachment) => ({
								mimeType: attachment.mimeType,
								size: attachment.data.length,
							})),
						}
					: {}),
				...(messageGroupId ? { messageGroupId } : {}),
			};
			tracing = resumeReason
				? await this.createOrchestratorResumeTraceContext({
						threadId,
						messageId,
						messageGroupId,
						runId,
						userId: user.id,
						modelId,
						input: traceInput,
						proxyConfig: orchestrationContext.tracingProxyConfig,
						resumeReason,
						metadata: {
							...(checkpoint?.isCheckpointFollowUp
								? { checkpoint_task_id: checkpoint.checkpointTaskId }
								: {}),
						},
					})
				: await createInstanceAiTraceContext({
						threadId,
						messageId,
						messageGroupId,
						runId,
						userId: user.id,
						modelId,
						input: traceInput,
						proxyConfig: orchestrationContext.tracingProxyConfig,
						n8nVersion: N8N_VERSION,
						workflowSdkVersion: WORKFLOW_SDK_VERSION,
					});

			// 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 import('@n8n/instance-ai');
				tracing = createTraceReplayOnlyContext();
			}

			if (tracing) {
				orchestrationContext.tracing = tracing;
				if (this.getTraceContext(runId) !== tracing) {
					await this.configureTraceReplayMode(tracing);
					this.runState.attachTracing(threadId, tracing);
					this.storeTraceContext(runId, threadId, tracing, messageGroupId);
				}
			}

			// Set heuristic title before agent starts — thread always has a title
			const thread = await memory.getThread(threadId);
			if (thread && !thread.title) {
				await patchThread(memory, {
					threadId,
					update: () => ({ title: truncateToTitle(message) }),
				});
			}

			const existingTasks = await taskStorage.get(threadId);
			if (existingTasks) {
				this.eventBus.publish(threadId, {
					type: 'tasks-update',
					runId,
					agentId: ORCHESTRATOR_AGENT_ID,
					payload: { tasks: existingTasks },
				});
			}

			const enrichedMessage = await this.buildMessageWithRunningTasks(threadId, message);
			let nonStructuredAttachments: InstanceAiAttachment[] = [];
			let attachmentManifest = '';
			let hasParseableAttachment = false;

			if (attachments && attachments.length > 0) {
				const classifiedAttachments = classifyAttachments(attachments);
				nonStructuredAttachments = attachments.filter(
					(attachment) => !isParseableAttachment(attachment),
				);
				hasParseableAttachment = classifiedAttachments.some(
					(attachment: { parseable: boolean }) => attachment.parseable,
				);
				attachmentManifest = buildAttachmentManifest(classifiedAttachments);
			}

			const fullMessage =
				!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;

			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 {
				// When this is a user-initiated turn, build the input as an
				// explicit message object carrying our chosen id. The SDK
				// preserves the id on its end-of-turn save so the row matches
				// any user-message row we may have written from inside
				// `waitForConfirmation`.
				if (nonStructuredAttachments.length > 0 || userMessagePersistence) {
					const baseContent = [
						{ type: 'text' as const, text: fullMessage },
						...nonStructuredAttachments.map((attachment) => ({
							type: 'file' as const,
							data: attachment.data,
							mediaType: attachment.mimeType,
						})),
					];
					streamInput = [
						{
							...(userMessagePersistence ? { id: userMessagePersistence.id } : {}),
							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: ORCHESTRATOR_AGENT_ID,
						execution_mode: 'foreground',
						trace_kind: tracing.traceKind,
					},
					inputs: traceInput,
				});
				tracing.actorRun = actorRun;
				tracing.orchestratorRun = actorRun;
			}

			const agent = await createInstanceAgent({
				modelId,
				context,
				orchestrationContext,
				mcpServers,
				mcpManager: this.mcpClientManager,
				memoryConfig,
				memory,
				checkpointStore: this.checkpointStore,
				timeZone: timeZone ?? this.defaultTimeZone,
			});

			const result = tracing
				? await tracing.withActiveSpan(tracing.actorRun, async () => {
						return await streamAgentRun(
							agent as StreamableAgent,
							streamInput,
							{
								maxIterations: MAX_STEPS.ORCHESTRATOR,
								abortSignal: signal,
								persistence: {
									resourceId: user.id,
									threadId,
								},
								providerOptions: {
									anthropic: { cacheControl: { type: 'ephemeral' } },
								},
							},
							{
								threadId,
								runId,
								agentId: ORCHESTRATOR_AGENT_ID,
								signal,
								eventBus: this.eventBus,
								logger: this.logger,
								onActivity: () => this.runState.touchActiveRun(threadId),
							},
						);
					})
				: await streamAgentRun(
						agent as StreamableAgent,
						streamInput,
						{
							maxIterations: MAX_STEPS.ORCHESTRATOR,
							abortSignal: signal,
							persistence: {
								resourceId: user.id,
								threadId,
							},
							providerOptions: {
								anthropic: { cacheControl: { type: 'ephemeral' } },
							},
						},
						{
							threadId,
							runId,
							agentId: ORCHESTRATOR_AGENT_ID,
							signal,
							eventBus: this.eventBus,
							logger: this.logger,
							onActivity: () => this.runState.touchActiveRun(threadId),
						},
					);
			if (result.status === 'suspended') {
				if (result.suspension) {
					this.runState.suspendRun(threadId, {
						runId,
						agentRunId: result.agentRunId,
						agent,
						threadId,
						user,
						toolCallId: result.suspension.toolCallId,
						requestId: result.suspension.requestId,
						abortController,
						messageGroupId,
						createdAt: Date.now(),
						tracing,
						modelId,
						checkpoint,
					});
					void this.persistPendingConfirmation({
						requestId: result.suspension.requestId,
						threadId,
						userId: user.id,
						runId,
						messageGroupId,
						kind: 'suspended',
						toolCallId: result.suspension.toolCallId,
						checkpointKey: result.agentRunId,
						checkpointTaskId: checkpoint?.checkpointTaskId,
					});
				}

				// 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: intermediateText,
						is_intermediate: true,
					});
				}

				const waitingDecision = this.evaluateWaitingResponse(
					threadId,
					runId,
					result.confirmationEvent,
					{
						messageGroupId,
						correlationId: messageId,
					},
				);

				if (waitingDecision?.reason === 'confirmation-invalid') {
					messageTraceFinalization = await this.finishInvalidConfirmationRun({
						threadId,
						runId,
						abortController,
						snapshotStorage,
						tracing,
					});
					return;
				}

				if (result.confirmationEvent) {
					this.trackConfirmationRequest(threadId, result.confirmationEvent);
					this.eventBus.publish(threadId, result.confirmationEvent);
				}

				// Persist the agent tree so the confirmation UI survives page refresh.
				// The tree is rebuilt from in-memory events and includes the
				// confirmation-request data that the frontend needs.
				await this.saveAgentTreeSnapshot(threadId, runId, snapshotStorage);
				const suspensionOutputs = {
					status: 'suspended',
					runId,
					...(result.suspension?.requestId ? { requestId: result.suspension.requestId } : {}),
					...(result.suspension?.toolCallId
						? { pendingToolCallId: result.suspension.toolCallId }
						: {}),
					...(result.suspension?.toolName ? { toolName: result.suspension.toolName } : {}),
				};
				await this.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(''));
			this.evaluateTerminalResponse(threadId, runId, result.status, {
				messageGroupId,
				correlationId: messageId,
				workSummary: result.workSummary,
			});
			const finalStatus = result.status === 'errored' ? 'error' : result.status;
			await this.finalizeRunTracing(runId, tracing, {
				status: finalStatus,
				outputText,
				modelId,
			});
			messageTraceFinalization = {
				status: finalStatus,
				outputText,
				modelId,
				metadata: this.buildMessageTraceMetadata(threadId, runId, { status: finalStatus }),
			};
			const archivedWorkflowIds = await this.reapAiTemporaryFromRun(
				threadId,
				user,
				aiCreatedWorkflowIds,
			);
			await this.finalizeRun(threadId, runId, result.status, snapshotStorage, {
				userId: user.id,
				modelId,
				archivedWorkflowIds,
			});

			// Count credits on first completed run per thread
			if (result.status === 'completed') {
				await this.countCreditsIfFirst(user, threadId, runId);
				this.telemetry.track('Builder sent message', {
					thread_id: threadId,
					message: 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;
				}
				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);
				}
				this.evaluateTerminalResponse(threadId, runId, 'cancelled', {
					messageGroupId,
					correlationId: messageId,
				});
				await this.finalizeRunTracing(runId, tracing, {
					status: 'cancelled',
					reason: cancellationReason,
				});
				messageTraceFinalization = {
					status: 'cancelled',
					reason: cancellationReason,
					metadata: this.buildMessageTraceMetadata(threadId, runId, {
						status: 'cancelled',
						cancellationReason,
						runTimeout,
					}),
				};
				const archivedWorkflowIds = await this.reapAiTemporaryFromRun(
					threadId,
					user,
					aiCreatedWorkflowIds,
				);
				this.publishRunFinish(
					threadId,
					runId,
					'cancelled',
					cancellationReason,
					archivedWorkflowIds,
				);
				if (activeSnapshotStorage) {
					await this.saveAgentTreeSnapshot(threadId, runId, activeSnapshotStorage);
				}
				return;
			}

			const errorMessage = getErrorMessage(error);
			const userFacingErrorMessage = getUserFacingErrorMessage(error);

			this.logger.error('Instance AI run error', {
				error: errorMessage,
				threadId,
				runId,
			});
			this.evaluateTerminalResponse(threadId, runId, 'errored', {
				messageGroupId,
				correlationId: messageId,
				errorMessage: userFacingErrorMessage,
			});
			await this.finalizeRunTracing(runId, tracing, {
				status: 'error',
				reason: errorMessage,
			});
			messageTraceFinalization = {
				status: 'error',
				reason: errorMessage,
				metadata: this.buildMessageTraceMetadata(threadId, runId, { status: 'error' }),
			};

			const archivedWorkflowIds = await this.reapAiTemporaryFromRun(
				threadId,
				user,
				aiCreatedWorkflowIds,
			);
			this.eventBus.publish(threadId, {
				type: 'run-finish',
				runId,
				agentId: ORCHESTRATOR_AGENT_ID,
				payload: {
					status: 'error',
					reason: userFacingErrorMessage,
					...(archivedWorkflowIds.length > 0 ? { archivedWorkflowIds } : {}),
				},
			});
			if (activeSnapshotStorage) {
				await this.saveAgentTreeSnapshot(threadId, runId, activeSnapshotStorage);
			}
		} finally {
			this.runState.clearActiveRun(threadId);
			// Drop any unconsumed first-suspend persistence intent. SDK
			// end-of-turn save handles the success path; a mid-turn error
			// means we deliberately don't want a half-saved row.
			this.userMessagePersistenceByRun.delete(runId);
			// 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) {
				await this.maybeFinalizeRunTraceRoot(runId, messageTraceFinalization);
				if (messageTraceFinalization.status !== 'cancelled') {
					this.liveness.consumeRunTimeout(runId);
				}
			}
			// 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.
			//   2. Unconditional reschedule — drive the next tick. 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.
			if (!this.runState.hasSuspendedRun(threadId)) {
				if (checkpoint?.isCheckpointFollowUp) {
					await this.finalizeCheckpointFollowUp(user, threadId, checkpoint.checkpointTaskId);
				} else {
					await this.schedulePlannedTasks(user, threadId);
				}
				await this.drainPendingCheckpointReentries(user, threadId);
			}
		}
	}

	/**
	 * 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<void> {
		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 `<planned-task-follow-up type="checkpoint">` 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<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 || 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/delegate)
	 * 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 `<background-task-completed>` shell, which would re-open
	 * the orphan bug.
	 */
	private async maybeReenterParentCheckpoint(
		user: User,
		threadId: string,
		task: ManagedBackgroundTask,
	): Promise<boolean> {
		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,
	): Promise<void> {
		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, delegate) 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),
			});
		}

		await this.schedulePlannedTasks(user, threadId);
	}

	async resolveConfirmation(
		requestingUserId: string,
		requestId: string,
		request: InstanceAiConfirmRequest,
	): Promise<boolean> {
		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 false;
		}

		if (this.runState.resolvePendingConfirmation(freshUser.id, requestId, data)) {
			void this.dropPendingConfirmation(requestId);
			this.logger.debug('Resolved pending confirmation (sub-agent HITL)', {
				requestId,
				approved: data.approved,
			});
			return true;
		}

		this.logger.debug('Pending confirmation not found, trying suspended run resume', {
			requestId,
			approved: data.approved,
		});

		if (await this.resumeSuspendedRun(requestingUserId, requestId, data)) {
			return true;
		}

		// 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.resolveOrphanedConfirmation(requestingUserId, requestId, data);
	}

	private async resolveOrphanedConfirmation(
		userId: string,
		requestId: string,
		data: ConfirmationData,
	): Promise<boolean> {
		let orphan: Awaited<ReturnType<typeof this.pendingConfirmationRepo.claim>>;
		try {
			orphan = await this.pendingConfirmationRepo.claim(requestId, userId);
		} catch (error: unknown) {
			this.logger.warn('Failed to claim orphaned pending confirmation', {
				requestId,
				error: getErrorMessage(error),
			});
			return false;
		}
		if (!orphan) return false;

		this.logger.info('Reclaiming pending confirmation orphaned by a process restart', {
			requestId,
			threadId: orphan.threadId,
			runId: orphan.runId,
			kind: orphan.kind,
			hasCheckpoint: Boolean(orphan.checkpointKey),
		});

		if (orphan.kind === 'suspended' && this.canResumeOrphan(orphan)) {
			const resumed = await this.tryResumeFromOrphan(orphan, data);
			if (resumed) return true;
		}

		this.finalizeUnresumableOrphan(orphan);
		throw new UserError(
			'This confirmation was lost when the assistant restarted. Send a new message to continue.',
		);
	}

	private canResumeOrphan(
		orphan: NonNullable<Awaited<ReturnType<typeof this.pendingConfirmationRepo.claim>>>,
	): orphan is ResumableOrphan {
		return Boolean(orphan.toolCallId && orphan.checkpointKey);
	}

	private finalizeUnresumableOrphan(
		orphan: NonNullable<Awaited<ReturnType<typeof this.pendingConfirmationRepo.claim>>>,
	): void {
		try {
			// Live SSE clients use this to drop their interactive card.
			this.publishRunFinish(
				orphan.threadId,
				orphan.runId,
				'cancelled',
				'restart_lost_confirmation',
			);
			// Terminalise the existing snapshot in place instead of rebuilding
			// the tree from the in-memory event bus. After a restart the bus
			// only carries the run-finish we just emitted, so a rebuild would
			// replace the saved plan/ask card with an empty cancelled tree;
			// `markRunCancelled` keeps the plan content intact while flipping
			// all in-flight nodes and confirmation buttons off.
			void this.dbSnapshotStorage
				.markRunCancelled(orphan.threadId, orphan.runId)
				.catch((error: unknown) => {
					this.logger.warn('Failed to mark orphan snapshot as cancelled', {
						requestId: orphan.requestId,
						threadId: orphan.threadId,
						runId: orphan.runId,
						error: getErrorMessage(error),
					});
				});
		} catch (error: unknown) {
			this.logger.warn('Failed to finalize orphaned confirmation snapshot', {
				requestId: orphan.requestId,
				error: getErrorMessage(error),
			});
		}
	}

	/**
	 * Rebuild the orchestration environment + agent for a checkpoint-backed
	 * suspended run that survived a process restart, register it as a
	 * `SuspendedRunState` in the in-memory registry, and hand off to the
	 * existing `resumeSuspendedRun` path. The original `runId` /
	 * `messageGroupId` are reused so the frontend's SSE correlation
	 * (`groupIdByRunId`) keeps working.
	 */
	private async tryResumeFromOrphan(
		orphan: ResumableOrphan,
		data: ConfirmationData,
	): Promise<boolean> {
		const outcome = await this.rebuildSuspendedRunFromCheckpoint(orphan);
		switch (outcome.kind) {
			case 'ready':
				// Re-seed the in-memory runState so `resumeSuspendedRun` can find
				// this confirmation by requestId and the rest of the cancel /
				// liveness / shutdown paths see the run as live. We deliberately
				// do NOT call `persistPendingConfirmation` again — the DB row
				// was already consumed by `claim()` above.
				this.runState.suspendRun(orphan.threadId, outcome.state);
				return await this.resumeSuspendedRun(orphan.userId, orphan.requestId, data);
			case 'no-user':
				this.logger.warn('Cannot resume orphaned run: user no longer authorized', {
					requestId: orphan.requestId,
					userId: orphan.userId,
				});
				return false;
			case 'no-checkpoint':
				this.logger.warn('Cannot resume orphaned run: checkpoint missing or unavailable', {
					requestId: orphan.requestId,
					checkpointKey: orphan.checkpointKey,
					...(outcome.error ? { error: getErrorMessage(outcome.error) } : {}),
				});
				return false;
			case 'env-failure':
				this.logger.warn('Cannot resume orphaned run: failed to build execution environment', {
					requestId: orphan.requestId,
					threadId: orphan.threadId,
					error: getErrorMessage(outcome.error),
				});
				return false;
			case 'agent-failure':
				this.logger.warn('Cannot resume orphaned run: failed to build agent', {
					requestId: orphan.requestId,
					threadId: orphan.threadId,
					error: getErrorMessage(outcome.error),
				});
				return false;
		}
	}

	/**
	 * 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<RebuildSuspendedRunOutcome> {
		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 mcpServers = this.parseMcpServers(this.instanceAiConfig.mcpServers);
		let agent;
		try {
			agent = await createInstanceAgent({
				modelId: environment.modelId,
				context: environment.context,
				orchestrationContext: environment.orchestrationContext,
				mcpServers,
				mcpManager: this.mcpClientManager,
				memoryConfig: this.createAgentMemoryOptions(),
				memory: environment.memory,
				checkpointStore: this.checkpointStore,
				timeZone: this.runState.getTimeZone(orphan.threadId) ?? this.defaultTimeZone,
			});
		} catch (error: unknown) {
			return { kind: 'agent-failure', error };
		}

		return {
			kind: 'ready',
			state: {
				runId: orphan.runId,
				agentRunId: orphan.checkpointKey,
				agent,
				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,
			},
		};
	}

	private async revalidateActiveUser(userId: string): Promise<User | null> {
		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;
		}
	}

	private async resumeSuspendedRun(
		requestingUserId: string,
		requestId: string,
		data: ConfirmationData,
	): Promise<boolean> {
		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 false;
		}

		const {
			agent,
			runId,
			agentRunId,
			threadId,
			user,
			toolCallId,
			abortController,
			tracing,
			modelId,
			messageGroupId,
			checkpoint,
		} = suspended;
		if (user.id !== requestingUserId) return false;

		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 false;
		}

		this.runState.activateSuspendedRun(threadId);

		// 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.dropPendingConfirmation(requestId);

		// setup-workflow uses nodeCredentials (per-node) format for its credentials field;
		// other tools use the flat credentials map. Prefer nodeCredentials when present.
		const credentialsPayload = data.nodeCredentials ?? data.credentials;
		const resumeData = {
			approved: data.approved,
			...(credentialsPayload ? { credentials: credentialsPayload } : {}),
			...(data.userInput !== undefined ? { userInput: data.userInput } : {}),
			...(data.domainAccessAction ? { domainAccessAction: data.domainAccessAction } : {}),
			...(data.action ? { action: data.action } : {}),
			...(data.nodeParameters ? { nodeParameters: data.nodeParameters } : {}),
			...(data.testTriggerNode ? { testTriggerNode: data.testTriggerNode } : {}),
			...(data.answers ? { answers: data.answers } : {}),
			...(data.resourceDecision ? { resourceDecision: data.resourceDecision } : {}),
		};

		const resumeTracing = await this.createOrchestratorResumeTraceContext({
			baseTracing: tracing,
			threadId,
			messageId: nanoid(),
			messageGroupId,
			runId,
			userId: activeUser.id,
			modelId,
			input: {
				requestId,
				toolCallId,
				approved: data.approved,
				resumeFields: Object.keys(resumeData),
			},
			resumeReason: 'approval',
			metadata: {
				request_id: requestId,
				pending_tool_call_id: toolCallId,
				approved: data.approved,
				...(checkpoint?.isCheckpointFollowUp
					? { checkpoint_task_id: checkpoint.checkpointTaskId }
					: {}),
			},
		});

		this.startProcessResumedStream(agent, resumeData, {
			runId,
			agentRunId,
			threadId,
			user: activeUser,
			toolCallId,
			signal: abortController.signal,
			abortController,
			snapshotStorage: this.dbSnapshotStorage,
			tracing: resumeTracing ?? tracing,
			modelId,
			checkpoint,
		});
		return true;
	}

	/**
	 * 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<string, unknown>,
		opts: {
			runId: string;
			agentRunId: string;
			threadId: string;
			user: User;
			toolCallId: string;
			signal: AbortSignal;
			abortController: AbortController;
			snapshotStorage: DbSnapshotStorage;
			tracing?: InstanceAiTraceContext;
			modelId?: ModelConfig;
			checkpoint?: { isCheckpointFollowUp: true; checkpointTaskId: string };
		},
	): Promise<void> {
		let messageTraceFinalization: MessageTraceFinalization | undefined;

		try {
			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 result = opts.tracing
				? await opts.tracing.withActiveSpan(opts.tracing.actorRun, async () => {
						return await resumeAgentRun(
							agent,
							resumeData,
							{
								runId: opts.agentRunId,
								toolCallId: opts.toolCallId,
								persistence: { resourceId: opts.user.id, threadId: opts.threadId },
							},
							{
								threadId: opts.threadId,
								runId: opts.runId,
								agentId: ORCHESTRATOR_AGENT_ID,
								signal: opts.signal,
								eventBus: this.eventBus,
								logger: this.logger,
								agentRunId: opts.agentRunId,
								onActivity: () => this.runState.touchActiveRun(opts.threadId),
							},
						);
					})
				: await resumeAgentRun(
						agent,
						resumeData,
						{
							runId: opts.agentRunId,
							toolCallId: opts.toolCallId,
							persistence: { resourceId: opts.user.id, threadId: opts.threadId },
						},
						{
							threadId: opts.threadId,
							runId: opts.runId,
							agentId: ORCHESTRATOR_AGENT_ID,
							signal: opts.signal,
							eventBus: this.eventBus,
							logger: this.logger,
							agentRunId: opts.agentRunId,
							onActivity: () => this.runState.touchActiveRun(opts.threadId),
						},
					);

			if (result.status === 'suspended') {
				if (result.suspension) {
					const resumeMessageGroupId = this.traceContextsByRunId.get(opts.runId)?.messageGroupId;
					this.runState.suspendRun(opts.threadId, {
						runId: opts.runId,
						agentRunId: result.agentRunId,
						agent,
						threadId: opts.threadId,
						user: opts.user,
						toolCallId: result.suspension.toolCallId,
						requestId: result.suspension.requestId,
						abortController: opts.abortController,
						messageGroupId: resumeMessageGroupId,
						createdAt: Date.now(),
						tracing: opts.tracing,
						...(opts.modelId !== undefined ? { modelId: opts.modelId } : {}),
						checkpoint: opts.checkpoint,
					});
					void this.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,
					});
				}

				// 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: intermediateText,
						is_intermediate: true,
					});
				}

				const messageGroupId = this.traceContextsByRunId.get(opts.runId)?.messageGroupId;
				const waitingDecision = this.evaluateWaitingResponse(
					opts.threadId,
					opts.runId,
					result.confirmationEvent,
					{ messageGroupId },
				);

				if (waitingDecision?.reason === 'confirmation-invalid') {
					messageTraceFinalization = await this.finishInvalidConfirmationRun({
						threadId: opts.threadId,
						runId: opts.runId,
						abortController: opts.abortController,
						snapshotStorage: opts.snapshotStorage,
						tracing: opts.tracing,
					});
					return;
				}

				if (result.confirmationEvent) {
					this.trackConfirmationRequest(opts.threadId, result.confirmationEvent);
					this.eventBus.publish(opts.threadId, result.confirmationEvent);
				}

				// Persist the refreshed agent tree so repeated HITL waits
				// survive page refresh after a resume as well.
				await this.saveAgentTreeSnapshot(opts.threadId, opts.runId, opts.snapshotStorage);
				const suspensionOutputs = {
					status: 'suspended',
					runId: opts.runId,
					...(result.suspension?.requestId ? { requestId: result.suspension.requestId } : {}),
					...(result.suspension?.toolCallId
						? { pendingToolCallId: result.suspension.toolCallId }
						: {}),
					...(result.suspension?.toolName ? { toolName: result.suspension.toolName } : {}),
				};
				await this.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(''));
			const messageGroupId = this.traceContextsByRunId.get(opts.runId)?.messageGroupId;
			this.evaluateTerminalResponse(opts.threadId, opts.runId, result.status, {
				messageGroupId,
				workSummary: result.workSummary,
			});
			const finalStatus = result.status === 'errored' ? 'error' : result.status;
			await this.finalizeRunTracing(opts.runId, opts.tracing, {
				status: finalStatus,
				outputText,
			});
			messageTraceFinalization = {
				status: finalStatus,
				outputText,
				metadata: this.buildMessageTraceMetadata(opts.threadId, opts.runId, {
					status: finalStatus,
				}),
			};
			const archivedWorkflowIds = await this.reapAiTemporaryFromRun(
				opts.threadId,
				opts.user,
				undefined,
			);
			await this.finalizeRun(opts.threadId, opts.runId, result.status, opts.snapshotStorage, {
				archivedWorkflowIds,
			});

			if (result.status === 'completed') {
				await this.countCreditsIfFirst(opts.user, opts.threadId, opts.runId);
				this.telemetry.track('Builder sent message', {
					thread_id: opts.threadId,
					message: outputText,
				});
				this.telemetry.track('Builder satisfied user intent', {
					thread_id: opts.threadId,
				});
			}
		} catch (error) {
			if (opts.signal.aborted) {
				if (this.shouldPreserveHitlOnShutdown(opts.runId)) {
					return;
				}
				const messageGroupId = this.traceContextsByRunId.get(opts.runId)?.messageGroupId;
				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);
				}
				this.evaluateTerminalResponse(opts.threadId, opts.runId, 'cancelled', {
					messageGroupId,
				});
				await this.finalizeRunTracing(opts.runId, opts.tracing, {
					status: 'cancelled',
					reason: cancellationReason,
				});
				messageTraceFinalization = {
					status: 'cancelled',
					reason: cancellationReason,
					metadata: this.buildMessageTraceMetadata(opts.threadId, opts.runId, {
						status: 'cancelled',
						cancellationReason,
						runTimeout,
					}),
				};
				const archivedWorkflowIds = await this.reapAiTemporaryFromRun(
					opts.threadId,
					opts.user,
					undefined,
				);
				this.publishRunFinish(
					opts.threadId,
					opts.runId,
					'cancelled',
					cancellationReason,
					archivedWorkflowIds,
				);
				await this.saveAgentTreeSnapshot(opts.threadId, opts.runId, opts.snapshotStorage);
				return;
			}

			const errorMessage = getErrorMessage(error);
			const userFacingErrorMessage = getUserFacingErrorMessage(error);

			this.logger.error('Instance AI resumed run error', {
				error: errorMessage,
				threadId: opts.threadId,
				runId: opts.runId,
			});
			const messageGroupId = this.traceContextsByRunId.get(opts.runId)?.messageGroupId;
			this.evaluateTerminalResponse(opts.threadId, opts.runId, 'errored', {
				messageGroupId,
				errorMessage: userFacingErrorMessage,
			});
			await this.finalizeRunTracing(opts.runId, opts.tracing, {
				status: 'error',
				reason: errorMessage,
			});
			messageTraceFinalization = {
				status: 'error',
				reason: errorMessage,
				metadata: this.buildMessageTraceMetadata(opts.threadId, opts.runId, {
					status: 'error',
				}),
			};

			const archivedWorkflowIds = await this.reapAiTemporaryFromRun(
				opts.threadId,
				opts.user,
				undefined,
			);
			this.eventBus.publish(opts.threadId, {
				type: 'run-finish',
				runId: opts.runId,
				agentId: ORCHESTRATOR_AGENT_ID,
				payload: {
					status: 'error',
					reason: userFacingErrorMessage,
					...(archivedWorkflowIds.length > 0 ? { archivedWorkflowIds } : {}),
				},
			});
			await this.saveAgentTreeSnapshot(opts.threadId, opts.runId, opts.snapshotStorage);
		} finally {
			this.runState.clearActiveRun(opts.threadId);
			// See note in executeRun's finally — keep threadPushRef alive for
			// post-run planned-task dispatch.
			if (messageTraceFinalization) {
				await this.maybeFinalizeRunTraceRoot(opts.runId, messageTraceFinalization);
				if (messageTraceFinalization.status !== 'cancelled') {
					this.liveness.consumeRunTimeout(opts.runId);
				}
			}
			// 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 (!this.runState.hasSuspendedRun(opts.threadId)) {
				if (opts.checkpoint?.isCheckpointFollowUp) {
					await this.finalizeCheckpointFollowUp(
						opts.user,
						opts.threadId,
						opts.checkpoint.checkpointTaskId,
					);
				} else {
					await this.schedulePlannedTasks(opts.user, opts.threadId);
				}
				await this.drainPendingCheckpointReentries(opts.user, opts.threadId);
			}
		}
	}

	// ── Background task management ──────────────────────────────────────────

	private spawnBackgroundTask(
		runId: string,
		opts: SpawnBackgroundTaskOptions,
		snapshotStorage: DbSnapshotStorage,
		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.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.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.finalizeBackgroundTaskTracing(task, 'failed');
				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.recordBackgroundTerminalOutcome(task);
				await this.saveAgentTreeSnapshot(
					opts.threadId,
					runId,
					snapshotStorage,
					true,
					task.messageGroupId,
				);

				// 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 build-workflow-with-agent calls.
				if (task.plannedTaskId) return;

				// Parent-tagged children (patch-builder etc. spawned inside a
				// checkpoint follow-up) must NEVER emit a generic
				// `<background-task-completed>` 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;
					}

					const user = this.runState.getThreadUser(opts.threadId);
					if (user) {
						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,
							`<background-task-completed>\n${payload}\n</background-task-completed>\n\n${AUTO_FOLLOW_UP_MESSAGE}`,
							task.messageGroupId,
						);
					}
				}
			},
		});

		if (outcome.status === 'started') {
			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.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' };
	}

	private async buildMessageWithRunningTasks(threadId: string, message: string): Promise<string> {
		return await enrichMessageWithBackgroundTasks(
			message,
			this.backgroundTasks.getRunningTasks(threadId),
			{
				formatTask: async (task: ManagedBackgroundTask) =>
					`[Running task — ${task.role}]: taskId=${task.taskId}`,
			},
		);
	}

	private trackConfirmationRequest(
		threadId: string,
		confirmationEvent: { payload: Record<string, unknown> },
	): void {
		const payload = confirmationEvent.payload;
		const inputThreadId = nanoid();
		payload.inputThreadId = inputThreadId;

		const inputType = payload.inputType as string | undefined;
		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 {
			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;
		}

		this.telemetry.track('Builder asked for input', {
			thread_id: threadId,
			input_thread_id: inputThreadId,
			type,
			num_steps: numSteps,
		});
	}

	/**
	 * Archive any workflow the agent created for this thread that still carries
	 * the AI-builder temporary marker. The orchestrator clears the marker on the
	 * main deliverable before run-finish, so anything still marked is a
	 * stepping-stone — chunk, scratch, or sub-workflow the user never sees in
	 * the workflows list. Soft delete: a mistaken reap is recoverable from the
	 * archive view.
	 *
	 * Best-effort. Individual archive failures are logged but do not block
	 * the run-finish emit.
	 */
	private async reapAiTemporaryFromRun(
		threadId: string,
		user: User,
		createdWorkflowIds: Set<string> | undefined,
	): Promise<string[]> {
		const runningTaskCount = this.backgroundTasks.getRunningTasks(threadId).length;
		if (runningTaskCount > 0) {
			this.logger.debug('Deferring AI-builder temporary workflow cleanup until tasks settle', {
				threadId,
				runningTaskCount,
			});
			return [];
		}

		let markedWorkflows: Array<{ workflowId: string }> = [];
		try {
			markedWorkflows = await this.aiBuilderTemporaryWorkflowRepository.findByThread(threadId);
		} catch (error) {
			this.logger.warn('Failed to inspect AI-builder temporary workflows during run finish', {
				threadId,
				error: getErrorMessage(error),
			});
		}
		const workflowIds = new Set([
			...markedWorkflows.map(({ workflowId }) => workflowId),
			...(createdWorkflowIds ?? []),
		]);
		if (workflowIds.size === 0) return [];

		return await this.archiveAiTemporaryWorkflows(threadId, user, workflowIds);
	}

	private async archiveAiTemporaryWorkflows(
		threadId: string,
		user: User,
		workflowIds: Set<string>,
	): Promise<string[]> {
		const adapter = this.adapterService.createContext(user, { threadId });
		const archived: string[] = [];
		for (const workflowId of workflowIds) {
			try {
				const didArchive = await adapter.workflowService.archiveIfAiTemporary(workflowId);
				if (didArchive) archived.push(workflowId);
			} catch (error) {
				this.logger.warn('Failed to reap AI-builder temporary workflow', {
					threadId,
					workflowId,
					error: getErrorMessage(error),
				});
			}
		}
		return archived;
	}

	private async finalizeCancelledSuspendedRun(
		suspended: SuspendedRunState<User>,
		reason = 'user_cancelled',
	): Promise<void> {
		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.finalizeRunTracing(suspended.runId, suspended.tracing, {
			status: 'cancelled',
			reason,
		});

		const archivedWorkflowIds = await this.reapAiTemporaryFromRun(
			suspended.threadId,
			suspended.user,
			undefined,
		);
		this.publishRunFinish(
			suspended.threadId,
			suspended.runId,
			'cancelled',
			reason,
			archivedWorkflowIds,
		);

		// Persist the snapshot so the run-finish event (which clears
		// in-flight tool calls) is reflected in the stored tree.
		await this.saveAgentTreeSnapshot(
			suspended.threadId,
			suspended.runId,
			this.dbSnapshotStorage,
			true,
		);
		await this.maybeFinalizeRunTraceRoot(suspended.runId, {
			status: 'cancelled',
			reason,
			metadata: this.buildMessageTraceMetadata(suspended.threadId, suspended.runId, {
				status: 'cancelled',
				cancellationReason: reason,
				...(runTimeout ? { runTimeout } : {}),
			}),
		});

		void this.dropPendingConfirmation(suspended.requestId);
	}

	private async reapAiTemporaryForThreadCleanup(threadId: string): Promise<void> {
		let markedWorkflows: Array<{ workflowId: string }>;
		try {
			markedWorkflows = await this.aiBuilderTemporaryWorkflowRepository.findByThread(threadId);
		} catch (error) {
			this.logger.warn('Failed to inspect AI-builder temporary workflows during thread cleanup', {
				threadId,
				error: getErrorMessage(error),
			});
			return;
		}

		if (markedWorkflows.length === 0) return;

		let thread: Awaited<ReturnType<InstanceAiThreadRepository['findOneBy']>>;
		try {
			thread = await this.threadRepo.findOneBy({ id: threadId });
		} catch (error) {
			this.logger.warn('Failed to load thread owner for AI-builder temporary workflow cleanup', {
				threadId,
				markedWorkflowCount: markedWorkflows.length,
				error: getErrorMessage(error),
			});
			return;
		}
		if (!thread?.resourceId) {
			this.logger.warn('Skipping AI-builder temporary workflow cleanup for thread without owner', {
				threadId,
				markedWorkflowCount: markedWorkflows.length,
			});
			return;
		}

		let user: User | null;
		try {
			user = await this.userRepository.findOneBy({ id: thread.resourceId });
		} catch (error) {
			this.logger.warn('Failed to load user for AI-builder temporary workflow cleanup', {
				threadId,
				userId: thread.resourceId,
				markedWorkflowCount: markedWorkflows.length,
				error: getErrorMessage(error),
			});
			return;
		}
		if (!user) {
			this.logger.warn('Skipping AI-builder temporary workflow cleanup for missing thread owner', {
				threadId,
				userId: thread.resourceId,
				markedWorkflowCount: markedWorkflows.length,
			});
			return;
		}

		await this.archiveAiTemporaryWorkflows(
			threadId,
			user,
			new Set(markedWorkflows.map(({ workflowId }) => workflowId)),
		);
	}

	private publishRunFinish(
		threadId: string,
		runId: string,
		status: 'completed' | 'cancelled' | 'errored',
		reason?: string,
		archivedWorkflowIds?: string[],
	): void {
		const effectiveStatus = status === 'errored' ? 'error' : status;
		const hasArchived = archivedWorkflowIds && archivedWorkflowIds.length > 0;
		this.eventBus.publish(threadId, {
			type: 'run-finish',
			runId,
			agentId: ORCHESTRATOR_AGENT_ID,
			payload: {
				status: effectiveStatus,
				...(status === 'cancelled' ? { reason: reason ?? 'user_cancelled' } : {}),
				...(hasArchived ? { archivedWorkflowIds } : {}),
			},
		});
	}

	private async finalizeRun(
		threadId: string,
		runId: string,
		status: 'completed' | 'cancelled' | 'errored',
		snapshotStorage: DbSnapshotStorage,
		options?: { userId?: string; modelId?: ModelConfig; archivedWorkflowIds?: string[] },
	): Promise<void> {
		this.publishRunFinish(threadId, runId, status, undefined, options?.archivedWorkflowIds);
		await this.saveAgentTreeSnapshot(threadId, runId, snapshotStorage);
		if (status === 'completed' && options?.userId && options?.modelId) {
			void this.refineTitleIfNeeded(threadId, options.userId, options.modelId);
		}
	}

	/**
	 * 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<void> {
		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;

			// Concat recent user messages so retries after a trivial first message
			// (e.g. "hey") have enough signal to produce a good title.
			const history = await memory.getMessages(threadId, { limit: 5 });
			const userTexts = history.flatMap((m) => {
				if (!('role' in m) || m.role !== 'user') return [];
				const text = this.extractStoredMessageText(m.content);
				return text.length > 0 ? [text] : [];
			});
			if (userTexts.length === 0) return;
			const userText = userTexts.join('\n');

			const baseTracing = this.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_AGENT_ID,
				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 '';
	}

	/**
	 * Build an agent tree from in-memory events and persist it as a thread metadata snapshot.
	 * @param isUpdate If true, updates the existing snapshot for this runId (background task completion).
	 */
	private async saveAgentTreeSnapshot(
		threadId: string,
		runId: string,
		snapshotStorage: DbSnapshotStorage,
		isUpdate = false,
		overrideMessageGroupId?: string,
	): Promise<void> {
		try {
			const messageGroupId = overrideMessageGroupId ?? this.runState.getMessageGroupId(threadId);

			let events: InstanceAiEvent[];
			let groupRunIds: string[] | undefined;
			if (messageGroupId) {
				groupRunIds = this.getRunIdsForMessageGroup(messageGroupId);
				if (groupRunIds.length === 0) {
					const snapshot = await snapshotStorage.getLatest(threadId, { messageGroupId, runId });
					groupRunIds = snapshot?.runIds?.length ? snapshot.runIds : [runId];
				}
				events = this.eventBus.getEventsForRuns(threadId, groupRunIds);
			} else {
				events = this.eventBus.getEventsForRun(threadId, runId);
			}
			if (isUpdate && events.length === 0) {
				this.logger.warn('Skipped updating empty Instance AI agent tree snapshot', {
					threadId,
					runId,
					messageGroupId,
				});
				return;
			}
			const agentTree = buildAgentTreeFromEvents(events);

			const tracing = this.traceContextsByRunId.get(runId)?.tracing;
			const saveOptions = {
				messageGroupId,
				runIds: groupRunIds,
				traceId: tracing?.rootRun.otelTraceId,
				spanId: tracing?.rootRun.otelSpanId,
				langsmithRunId: tracing?.rootRun.id,
				langsmithTraceId: tracing?.rootRun.traceId,
			};

			if (isUpdate) {
				await snapshotStorage.updateLast(threadId, agentTree, runId, saveOptions);
			} else {
				await snapshotStorage.save(threadId, agentTree, runId, saveOptions);
			}
		} catch (error) {
			this.logger.warn('Failed to save agent tree snapshot', {
				threadId,
				runId,
				error: error instanceof Error ? error.message : String(error),
			});
		}
	}

	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() };
		});
	}
}