feat(core): Wire the Instance AI durable event log behind N8N_INSTANCE_AI_DURABLE_LOG (no-changelog) (#33984)

packages/@n8n/config/src/configs/instance-ai.config.ts

@@ -172,6 +172,14 @@ export class InstanceAiConfig {
 	@Env('N8N_INSTANCE_AI_RUN_DEBUG_ENABLED')
 	runDebugEnabled: boolean = false;
 
+	/**
+	 * EXPERIMENTAL: persist Instance AI events to a durable DB log
+	 * (`instance_ai_events`) and serve SSE replay + history from it. Off =
+	 * today's in-memory-only behavior. See RFC: instance-ai durable event log.
+	 */
+	@Env('N8N_INSTANCE_AI_DURABLE_LOG')
+	durableLog: boolean = false;
+
 	/** Enable extended thinking / reasoning for the orchestrator agent. */
 	@Env('N8N_INSTANCE_AI_THINKING_ENABLED')
 	thinkingEnabled: boolean = true;

packages/@n8n/config/test/config.test.ts

@@ -356,6 +356,7 @@ describe('GlobalConfig', () => {
 			outputRedactionPlaceholder: '[REDACTED]',
 			runDebugEnabled: false,
 			thinkingEnabled: true,
+			durableLog: false,
 		},
 		queue: {
 			health: {

packages/@n8n/instance-ai/docs/architecture.md

@@ -275,7 +275,11 @@ The event bus decouples agent execution from event delivery:
 - All events carry `runId` (correlates to triggering message) and `agentId`
 - SSE events use monotonically increasing per-thread `id` values for replay
 - SSE supports both `Last-Event-ID` header and `?lastEventId` query parameter
-- Events are persisted to thread storage regardless of transport
+- Event storage depends on `N8N_INSTANCE_AI_DURABLE_LOG`: off (default), events
+  live only in a bounded in-memory buffer (500 events / 2 MB per thread,
+  FIFO-evicted, ids reset on restart); on, coalesced step-level facts are
+  appended to the `instance_ai_events` table (the durable replay source, ids
+  survive restarts) while token deltas stay memory-only
 - No need to pipe sub-agent streams through orchestrator tool execution
 - One active run per thread (additional `POST /chat` is rejected while active)
 - Cancellation via `POST /instance-ai/chat/:threadId/cancel` (idempotent)

packages/@n8n/instance-ai/docs/configuration.md

@@ -166,7 +166,14 @@ The event bus transport is selected automatically:
 - **Single instance**: In-process `EventEmitter` — zero infrastructure
 - **Queue mode**: Redis Pub/Sub — uses n8n's existing Redis connection
 
-Event persistence always uses thread storage regardless of transport.
+Event persistence is controlled by `N8N_INSTANCE_AI_DURABLE_LOG` (default
+`false`). Off, events live only in a bounded in-memory buffer per thread
+(500 events / 2 MB, FIFO-evicted; ids reset on restart, so replay does not
+survive a restart). On, coalesced step-level facts (completed text/reasoning
+blocks, tool calls and results, run lifecycle) are appended to the
+`instance_ai_events` table and replay reads the database; token deltas are
+never persisted. Rows cascade-delete with their thread
+(`N8N_INSTANCE_AI_THREAD_TTL_DAYS`).
 
 Runtime behavior:
 - One active run per thread. Additional `POST /instance-ai/chat/:threadId`

packages/@n8n/instance-ai/docs/streaming-protocol.md

@@ -388,8 +388,12 @@ simultaneously persisted to thread storage and delivered to connected SSE client
 | Single instance | In-process `EventEmitter` | Zero infrastructure |
 | Queue mode | Redis Pub/Sub | n8n already uses Redis |
 
-Event persistence uses thread storage regardless of transport — this provides
-replay capability for reconnection.
+Replay storage depends on `N8N_INSTANCE_AI_DURABLE_LOG`. Off (default),
+replay serves from a bounded in-memory buffer per thread (500 events / 2 MB,
+FIFO-evicted; ids reset on restart). On, the durable event log
+(`instance_ai_events`) is the replay source: coalesced step-level facts are
+appended with a per-thread `seq` assigned by the writer's drain, so cursors
+stay valid across restarts and across mains sharing one database.
 
 ### Reconnection & Replay (Canonical Rule)
 
@@ -417,6 +421,20 @@ connection may occasionally deliver a lower id after a higher one. The
 frontend therefore tracks its reconnect cursor as the max id seen and drops
 already-seen ids on replay overlap.
 
+With the durable log enabled (`N8N_INSTANCE_AI_DURABLE_LOG`), ids are
+database-assigned sequence numbers and only DURABLE facts carry an `id:`
+line. Ephemeral frames (`text-delta`, `reasoning-delta`, `status`,
+`filesystem-request`) are live-only: their SSE frames have no `id:` line, so
+the browser's replay cursor never points at them (the same mechanism as the
+`run-sync` control frames). On replay, the deltas a client missed are covered
+by coalesced `text-block` / `reasoning-block` facts, which the shared run
+reducer applies with REPLACE semantics keyed on the segment's `responseId` —
+a client that reconnects mid-block never renders partial text twice. The
+writer persists a batch before emitting it live, so a fact is never in
+neither store, and same-connection replay does not require dedup; the
+endpoint's replay-and-subscribe handoff dedups by `seq` across its one async
+read.
+
 ## Abort Support
 
 The frontend can abort a running agent by sending:

packages/@n8n/instance-ai/src/event-bus/event-bus.interface.ts

@@ -1,8 +1,13 @@
 import type { InstanceAiEvent } from '@n8n/api-types';
 
-/** Stored event with a per-thread monotonic ID for SSE replay. */
+/**
+ * Stored event with a per-thread monotonic ID for SSE replay.
+ * `id` is absent on ephemeral events (text/reasoning deltas, status): they are
+ * live-delivered but never persisted, and their SSE frames carry no `id:` line
+ * so the browser replay cursor only advances on durable facts.
+ */
 export interface StoredEvent {
-	id: number; // monotonically increasing per thread, 1-based
+	id?: number; // monotonically increasing per thread, 1-based, durable facts only
 	event: InstanceAiEvent;
 }
 

packages/@n8n/instance-ai/src/runtime/__tests__/terminal-response-guard.test.ts

@@ -87,6 +87,20 @@ describe('InstanceAiTerminalResponseGuard', () => {
 		expect(decision.visibilitySource).toBe('root-text');
 	});
 
+	it('counts coalesced text-block facts as root text (durable-log reads carry no deltas)', () => {
+		const rootBlock: InstanceAiEvent = {
+			type: 'text-block',
+			runId,
+			agentId: rootAgentId,
+			responseId: 'resp-1',
+			payload: { text: 'hello' },
+		};
+		const decision = guard().evaluateTerminal([runStart(), rootBlock], 'completed');
+
+		expect(decision.action).toBe('none');
+		expect(decision.visibilitySource).toBe('root-text');
+	});
+
 	it('emits text fallback for silent completed runs with structured work counts only', () => {
 		const decision = guard().evaluateTerminal([runStart()], 'completed', {
 			workSummary: { totalToolCalls: 3, totalToolErrors: 1, toolCalls: [] },

packages/@n8n/instance-ai/src/runtime/terminal-response-guard.ts

@@ -62,7 +62,12 @@ function formatWorkSummaryCounts(workSummary?: WorkSummary): string {
 }
 
 function hasText(event: InstanceAiEvent): boolean {
-	return event.type === 'text-delta' && event.payload.text.trim().length > 0;
+	// The durable log stores coalesced text-block facts, never deltas, so
+	// flag-on guard reads must recognize both shapes of streamed text.
+	return (
+		(event.type === 'text-delta' || event.type === 'text-block') &&
+		event.payload.text.trim().length > 0
+	);
 }
 
 export class InstanceAiTerminalResponseGuard {

packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts

@@ -46,6 +46,8 @@ function createService(options: { threadTtlDays?: number } = {}): InstanceAiMemo
 		},
 	};
 	const mockLogger = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() };
+	// Flag off: the durable-log metrics forwarder only fires on parser fallbacks.
+	const mockDurableLogMetrics = { notifyParserFallbacks: vi.fn() };
 	return new InstanceAiMemoryService(
 		mockLogger as never,
 		mockConfig as never,
@@ -53,6 +55,7 @@ function createService(options: { threadTtlDays?: number } = {}): InstanceAiMemo
 		mockDbSnapshotStorage as never,
 		mockCheckpointRepository as never,
 		mockPendingConfirmationRepository as never,
+		mockDurableLogMetrics as never,
 	);
 }
 

packages/cli/src/modules/instance-ai/__tests__/instance-ai-terminal-outcome.service.test.ts

@@ -216,6 +216,7 @@ function createService(snapshotTree?: InstanceAiAgentNode): {
 	};
 
 	const options = {
+		durableLog: false,
 		eventBus: deps.eventBus,
 		dbSnapshotStorage: deps.dbSnapshotStorage,
 		agentMemory: {},
@@ -411,11 +412,51 @@ describe('InstanceAiTerminalOutcomeService — background outcome recording', ()
 	});
 });
 
+describe('InstanceAiTerminalOutcomeService — durable-log outcome lines', () => {
+	it('publishes the outcome line as a text-block when the durable log is on', async () => {
+		// A trailing delta would race the coalescer's idle flush on an immediate
+		// page reload; a text-block is persisted before it is emitted live.
+		const { deps } = createService(makeAgentTree());
+		const service = new InstanceAiTerminalOutcomeService({
+			durableLog: true,
+			eventBus: deps.eventBus,
+			dbSnapshotStorage: deps.dbSnapshotStorage,
+			agentMemory: {},
+			telemetry: deps.telemetry,
+			logger: deps.logger,
+			runState: deps.runState,
+			suspendedThreads: deps.suspendedThreads,
+			tracing: deps.tracing,
+			publishRunFinish: deps.publishRunFinish,
+			saveAgentTreeSnapshot: vi.fn(async () => {}),
+		} as never);
+
+		await service.recordBackgroundTerminalOutcome({
+			taskId: 'task-dl',
+			threadId: 'thread-dl',
+			runId: 'run-dl',
+			role: 'workflow-builder',
+			agentId: 'agent-builder',
+			status: 'cancelled',
+			result: undefined,
+			startedAt: 0,
+			lastActivityAt: 0,
+			abortController: new AbortController(),
+			corrections: [],
+		} as never);
+
+		const line = deps.eventBus.events.find(
+			(event) => event.type === 'text-block' || event.type === 'text-delta',
+		);
+		expect(line?.type).toBe('text-block');
+	});
+});
+
 describe('InstanceAiTerminalOutcomeService — terminal response guard wiring', () => {
-	it('publishes fallback output before run-finish on a silent completed run', () => {
+	it('publishes fallback output before run-finish on a silent completed run', async () => {
 		const { service, deps } = createService();
 
-		service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
+		await service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
 			messageGroupId: 'group-1',
 		});
 		deps.publishRunFinish('thread-a', 'run-1', 'completed');
@@ -423,10 +464,10 @@ describe('InstanceAiTerminalOutcomeService — terminal response guard wiring',
 		expect(deps.eventBus.events.map((event) => event.type)).toEqual(['text-delta', 'run-finish']);
 	});
 
-	it('does not publish completed fallback output when silence is expected', () => {
+	it('does not publish completed fallback output when silence is expected', async () => {
 		const { service, deps } = createService();
 
-		const decision = service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
+		const decision = await service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
 			messageGroupId: 'group-1',
 			suppressCompletedFallback: true,
 		});
@@ -438,10 +479,10 @@ describe('InstanceAiTerminalOutcomeService — terminal response guard wiring',
 		expect(deps.eventBus.events).toEqual([]);
 	});
 
-	it('publishes fallback error before run-finish on a silent failed run', () => {
+	it('publishes fallback error before run-finish on a silent failed run', async () => {
 		const { service, deps } = createService();
 
-		service.evaluateTerminalResponse('thread-a', 'run-1', 'errored', {
+		await service.evaluateTerminalResponse('thread-a', 'run-1', 'errored', {
 			messageGroupId: 'group-1',
 			errorMessage: 'Safe user-facing error',
 		});
@@ -450,10 +491,10 @@ describe('InstanceAiTerminalOutcomeService — terminal response guard wiring',
 		expect(deps.eventBus.events.map((event) => event.type)).toEqual(['error', 'run-finish']);
 	});
 
-	it('forwards a structured error code onto the emitted error event', () => {
+	it('forwards a structured error code onto the emitted error event', async () => {
 		const { service, deps } = createService();
 
-		service.evaluateTerminalResponse('thread-a', 'run-1', 'errored', {
+		await service.evaluateTerminalResponse('thread-a', 'run-1', 'errored', {
 			messageGroupId: 'group-1',
 			errorMessage: "You've run out of AI credits.",
 			errorCode: 'quota_exhausted',
@@ -467,7 +508,7 @@ describe('InstanceAiTerminalOutcomeService — terminal response guard wiring',
 		const { service, deps } = createService();
 		const abortController = new AbortController();
 
-		const decision = service.evaluateWaitingResponse('thread-a', 'run-1', undefined, {
+		const decision = await service.evaluateWaitingResponse('thread-a', 'run-1', undefined, {
 			messageGroupId: 'group-1',
 		});
 		expect(decision?.reason).toBe('confirmation-invalid');
@@ -493,10 +534,10 @@ describe('InstanceAiTerminalOutcomeService — terminal response guard wiring',
 		});
 	});
 
-	it('reads events across the message group when a group id is provided', () => {
+	it('reads events across the message group when a group id is provided', async () => {
 		const { service, deps } = createService();
 
-		service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
+		await service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
 			messageGroupId: 'group-1',
 		});
 
@@ -504,11 +545,11 @@ describe('InstanceAiTerminalOutcomeService — terminal response guard wiring',
 		expect(deps.eventBus.getEventsForRuns).toHaveBeenCalledWith('thread-a', ['run-1']);
 	});
 
-	it('falls back to the single run when the message group has no runs', () => {
+	it('falls back to the single run when the message group has no runs', async () => {
 		const { service, deps } = createService();
 		deps.runState.getRunIdsForMessageGroup.mockReturnValue([]);
 
-		service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
+		await service.evaluateTerminalResponse('thread-a', 'run-1', 'completed', {
 			messageGroupId: 'group-1',
 		});
 

packages/cli/src/modules/instance-ai/__tests__/instance-ai.controller.test.ts

@@ -66,6 +66,8 @@ import type { InstanceAiBrowserSessionService } from '../browser/instance-ai-bro
 import type { EvalExecutionService } from '../eval/execution.service';
 import { EvalThreadCredentialAllowlistService } from '../eval/thread-credential-allowlist.service';
 import type { EvalThreadRestoreService } from '../eval/thread-restore.service';
+import type { DurableEventLog } from '../event-bus/durable-event-log';
+import type { DurableLogMetrics } from '../event-bus/durable-log-metrics';
 import type { InProcessEventBus } from '../event-bus/in-process-event-bus';
 import type { LocalGateway } from '../filesystem/local-gateway';
 import type { InstanceAiGatewayService } from '../instance-ai-gateway.service';
@@ -95,11 +97,13 @@ describe('InstanceAiController', () => {
 	const memoryService = mock<InstanceAiMemoryService>();
 	const settingsService = mock<InstanceAiSettingsService>();
 	const eventBus = mock<InProcessEventBus>();
+	const eventLog = mock<DurableEventLog>();
+	const durableLogMetrics = mock<DurableLogMetrics>();
 	const moduleRegistry = mock<ModuleRegistry>();
 	const push = mock<Push>();
 	const urlService = mock<UrlService>();
 	const globalConfig = mock<GlobalConfig>({
-		instanceAi: { gatewayApiKey: 'static-key' },
+		instanceAi: { gatewayApiKey: 'static-key', durableLog: false },
 		editorBaseUrl: 'http://localhost:5678',
 		port: 5678,
 	});
@@ -122,6 +126,8 @@ describe('InstanceAiController', () => {
 		evalCredentialAllowlists,
 		evalThreadRestore,
 		eventBus,
+		eventLog,
+		durableLogMetrics,
 		moduleRegistry,
 		push,
 		urlService,
@@ -1625,3 +1631,279 @@ describe('InstanceAiController', () => {
 		});
 	});
 });
+
+describe('InstanceAiController — durable-log SSE replay (flag on)', () => {
+	const instanceAiService = mock<InstanceAiService>();
+	const memoryService = mock<InstanceAiMemoryService>();
+	const settingsService = mock<InstanceAiSettingsService>();
+	const eventBus = mock<InProcessEventBus>();
+	const eventLog = mock<DurableEventLog>();
+	const durableLogMetrics = mock<DurableLogMetrics>();
+	const globalConfig = mock<GlobalConfig>({
+		instanceAi: { gatewayApiKey: 'static-key', durableLog: true },
+		editorBaseUrl: 'http://localhost:5678',
+		port: 5678,
+	});
+
+	const controller = new InstanceAiController(
+		instanceAiService,
+		mock<InstanceAiGatewayService>(),
+		mock<InstanceAiBrowserSessionService>(),
+		memoryService,
+		settingsService,
+		mock<EvalExecutionService>(),
+		new EvalThreadCredentialAllowlistService(),
+		mock<EvalThreadRestoreService>(),
+		eventBus,
+		eventLog,
+		durableLogMetrics,
+		mock<ModuleRegistry>(),
+		mock<Push>(),
+		mock<UrlService>(),
+		mock<UserRepository>(),
+		mock<CredentialsService>(),
+		mock<ProjectService>(),
+		mock<InstanceAiErrorReporterService>(),
+		globalConfig,
+	);
+
+	beforeEach(() => {
+		vi.clearAllMocks();
+		settingsService.isInstanceAiEnabled.mockReturnValue(true);
+	});
+
+	it('replays from the durable log and dedups events that land during the async read', async () => {
+		memoryService.checkThreadOwnership.mockResolvedValue('owned');
+		instanceAiService.getThreadStatus.mockReturnValue({
+			hasActiveRun: false,
+			isSuspended: false,
+			backgroundTasks: [],
+		} as never);
+
+		const handlers: Array<(stored: unknown) => void> = [];
+		eventBus.subscribe.mockImplementation((_threadId, handler) => {
+			handlers.push(handler as never);
+			return vi.fn();
+		});
+
+		const factA = {
+			id: 6,
+			event: {
+				type: 'tool-call',
+				runId: 'run-1',
+				agentId: 'a1',
+				payload: { toolCallId: 'tc', toolName: 't', args: {} },
+			},
+		};
+		const factB = {
+			id: 7,
+			event: {
+				type: 'run-finish',
+				runId: 'run-1',
+				agentId: 'a1',
+				payload: { status: 'completed' },
+			},
+		};
+		eventLog.getEventsAfter.mockImplementation(async () => {
+			// While the DB read is in flight, the drain emits fact 7 live (already
+			// part of the replay result: must dedupe) and an id-less delta (must
+			// pass through: the cursor never points at it).
+			const buffering = handlers.at(-1)!;
+			buffering(factB);
+			buffering({
+				event: { type: 'text-delta', runId: 'run-1', agentId: 'a1', payload: { text: 'x' } },
+			});
+			return [factA, factB] as never;
+		});
+
+		const sseRes = mock<Response & { flush?: () => void }>({
+			setHeader: vi.fn(),
+			flushHeaders: vi.fn(),
+			write: vi.fn(),
+			end: vi.fn(),
+			flush: vi.fn(),
+		});
+		const sseReq = mock<AuthenticatedRequest>({
+			user: { id: USER_ID },
+			headers: {},
+			once: vi.fn(),
+		});
+
+		await controller.events(sseReq, sseRes, THREAD_ID, { lastEventId: 5 } as never);
+
+		expect(eventLog.getEventsAfter).toHaveBeenCalledWith(THREAD_ID, 5);
+		const frames = (sseRes.write as Mock).mock.calls.map(([frame]) => String(frame));
+		// Fact 7 was both replayed and buffered live: delivered exactly once.
+		expect(frames.filter((f) => f.includes('run-finish'))).toHaveLength(1);
+		expect(frames.filter((f) => f.includes('tool-call'))).toHaveLength(1);
+		// The id-less delta passes through with NO id: line.
+		const deltaFrame = frames.find((f) => f.includes('text-delta'));
+		expect(deltaFrame).toBeDefined();
+		expect(deltaFrame!.startsWith('data: ')).toBe(true);
+		// Replay instrumentation recorded events served + cursor age.
+		expect(durableLogMetrics.recordReplay).toHaveBeenCalledWith(2, 2);
+	});
+
+	it('delivers events that land during the run-sync tree reads', async () => {
+		memoryService.checkThreadOwnership.mockResolvedValue('owned');
+		memoryService.getLatestRunSnapshot.mockResolvedValue(undefined);
+		instanceAiService.getThreadStatus.mockReturnValue({
+			hasActiveRun: true,
+			isSuspended: false,
+			backgroundTasks: [],
+		} as never);
+		instanceAiService.getMessageGroupId.mockReturnValue('group-1');
+		instanceAiService.getRunIdsForMessageGroup.mockReturnValue(['run-1']);
+
+		const handlers: Array<(stored: unknown) => void> = [];
+		eventBus.subscribe.mockImplementation((_threadId, handler) => {
+			handlers.push(handler as never);
+			return vi.fn();
+		});
+
+		eventLog.getEventsAfter.mockResolvedValue([]);
+		const treeEvent = {
+			type: 'tool-call',
+			runId: 'run-1',
+			agentId: 'a1',
+			payload: { toolCallId: 'tc', toolName: 't', args: {} },
+		};
+		eventLog.getEventsForRuns.mockImplementation(async () => {
+			// A fact lands while the bootstrap tree is being read from the DB:
+			// the buffering subscription must still be active here, or the event
+			// is lost for good (the cursor advances past it on the next event).
+			handlers.at(-1)!({
+				id: 8,
+				event: {
+					type: 'run-finish',
+					runId: 'run-1',
+					agentId: 'a1',
+					payload: { status: 'completed' },
+				},
+			});
+			return [treeEvent] as never;
+		});
+
+		const sseRes = mock<Response & { flush?: () => void }>({
+			setHeader: vi.fn(),
+			flushHeaders: vi.fn(),
+			write: vi.fn(),
+			end: vi.fn(),
+			flush: vi.fn(),
+		});
+		const sseReq = mock<AuthenticatedRequest>({
+			user: { id: USER_ID },
+			headers: {},
+			once: vi.fn(),
+		});
+
+		await controller.events(sseReq, sseRes, THREAD_ID, { lastEventId: 5 } as never);
+
+		const frames = (sseRes.write as Mock).mock.calls.map(([frame]) => String(frame));
+		const syncIndex = frames.findIndex((f) => f.startsWith('event: run-sync'));
+		const finishIndex = frames.findIndex((f) => f.includes('run-finish'));
+		expect(syncIndex).toBeGreaterThanOrEqual(0);
+		expect(finishIndex).toBeGreaterThanOrEqual(0);
+		// Delivered exactly once, after the frame whose tree may already fold it
+		// (the shared reducer applies it idempotently, like any post-frame event).
+		expect(frames.filter((f) => f.includes('run-finish'))).toHaveLength(1);
+		expect(finishIndex).toBeGreaterThan(syncIndex);
+	});
+
+	it('removes the buffering subscription when a durable read throws', async () => {
+		memoryService.checkThreadOwnership.mockResolvedValue('owned');
+		instanceAiService.getThreadStatus.mockReturnValue({
+			hasActiveRun: false,
+			isSuspended: false,
+			backgroundTasks: [],
+		} as never);
+
+		const unsubscribers: Array<ReturnType<typeof vi.fn>> = [];
+		eventBus.subscribe.mockImplementation(() => {
+			const unsubscribe = vi.fn();
+			unsubscribers.push(unsubscribe);
+			return unsubscribe;
+		});
+		eventLog.getEventsAfter.mockRejectedValue(new Error('db down'));
+
+		const sseRes = mock<Response & { flush?: () => void }>({
+			setHeader: vi.fn(),
+			flushHeaders: vi.fn(),
+			write: vi.fn(),
+			end: vi.fn(),
+			flush: vi.fn(),
+		});
+		const sseReq = mock<AuthenticatedRequest>({
+			user: { id: USER_ID },
+			headers: {},
+			once: vi.fn(),
+		});
+
+		await expect(
+			controller.events(sseReq, sseRes, THREAD_ID, { lastEventId: 5 } as never),
+		).rejects.toThrow('db down');
+
+		// Two subscriptions exist: the step-1 live one (cleaned up on connection
+		// close) and the temporary replay buffer, which must be removed on the
+		// error path rather than lingering on the thread emitter.
+		expect(unsubscribers).toHaveLength(2);
+		expect(unsubscribers[1]).toHaveBeenCalledTimes(1);
+	});
+
+	it('stops the bootstrap when the client disconnects during a durable read', async () => {
+		memoryService.checkThreadOwnership.mockResolvedValue('owned');
+		memoryService.getLatestRunSnapshot.mockResolvedValue(undefined);
+		instanceAiService.getThreadStatus.mockReturnValue({
+			hasActiveRun: true,
+			isSuspended: false,
+			backgroundTasks: [],
+		} as never);
+		instanceAiService.getMessageGroupId.mockReturnValue('group-1');
+		instanceAiService.getRunIdsForMessageGroup.mockReturnValue(['run-1']);
+
+		eventBus.subscribe.mockReturnValue(vi.fn());
+
+		const onceHandlers = new Map<string, () => void>();
+		const sseReq = mock<AuthenticatedRequest>({
+			user: { id: USER_ID },
+			headers: {},
+			once: vi.fn(((eventName: string, handler: () => void) => {
+				onceHandlers.set(eventName, handler);
+			}) as never),
+		});
+		const sseRes = mock<Response & { flush?: () => void }>({
+			setHeader: vi.fn(),
+			flushHeaders: vi.fn(),
+			write: vi.fn(),
+			end: vi.fn(),
+			flush: vi.fn(),
+		});
+
+		eventLog.getEventsAfter.mockImplementation(async () => {
+			// The browser goes away while the replay read is in flight. Without a
+			// post-await check, the frames below would be written to a dead
+			// response and the keep-alive interval would be created after cleanup
+			// already ran, leaking it for good.
+			onceHandlers.get('close')!();
+			return [
+				{
+					id: 6,
+					event: {
+						type: 'tool-call',
+						runId: 'run-1',
+						agentId: 'a1',
+						payload: { toolCallId: 'tc', toolName: 't', args: {} },
+					},
+				},
+			] as never;
+		});
+
+		await controller.events(sseReq, sseRes, THREAD_ID, { lastEventId: 5 } as never);
+
+		const frames = (sseRes.write as Mock).mock.calls.map(([frame]) => String(frame));
+		expect(frames.some((f) => f.includes('tool-call'))).toBe(false);
+		expect(frames.some((f) => f.startsWith('event: run-sync'))).toBe(false);
+		expect(eventLog.getEventsForRuns).not.toHaveBeenCalled();
+		expect(durableLogMetrics.recordReplay).not.toHaveBeenCalled();
+	});
+});

packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts

@@ -555,6 +555,7 @@ type ShutdownServiceInternals = {
 	browserSessionService: { shutdown: MockedFunction<() => Promise<void>> };
 	domainAccessTrackersByThread: Map<string, unknown>;
 	eventBus: { clear: MockedFunction<() => void> };
+	eventLog: { flushAll: MockedFunction<() => Promise<void>> };
 	_mcpClientManager?: { disconnect: MockedFunction<() => Promise<void>> };
 	inFlightExecutions: Set<Promise<unknown>>;
 	logger: { debug: Mock; warn: Mock };
@@ -643,6 +644,8 @@ type SnapshotServiceInternals = {
 		getEventsForRun: Mock;
 		getEventsForRuns: Mock;
 	};
+	eventLog: { flush: Mock; getEventsForRuns: Mock };
+	instanceAiConfig: { durableLog: boolean };
 	tracing: { getTraceContext: Mock };
 	logger: { warn: Mock };
 };
@@ -699,6 +702,7 @@ function createTerminalGuardOrderService(): TerminalGuardOrderServiceInternals {
 	service.preserveHitlOnShutdown = new Set();
 
 	service.terminalOutcome = new InstanceAiTerminalOutcomeService({
+		durableLog: false,
 		eventBus: service.eventBus,
 		dbSnapshotStorage: {},
 		agentMemory: {},
@@ -736,6 +740,8 @@ function createSnapshotService(): SnapshotServiceInternals {
 		getEventsForRun: vi.fn(() => []),
 		getEventsForRuns: vi.fn(() => []),
 	};
+	service.eventLog = { flush: vi.fn(async () => {}), getEventsForRuns: vi.fn(async () => []) };
+	service.instanceAiConfig = { durableLog: false };
 	service.tracing = { getTraceContext: vi.fn(() => undefined) };
 	service.logger = { warn: vi.fn() };
 	return service;
@@ -1024,6 +1030,7 @@ describe('InstanceAiService — shutdown', () => {
 		service.browserSessionService = { shutdown: vi.fn(async () => {}) };
 		service.domainAccessTrackersByThread = new Map();
 		service.eventBus = { clear: vi.fn() };
+		service.eventLog = { flushAll: vi.fn(async () => {}) };
 		service._mcpClientManager = { disconnect: vi.fn(async () => {}) };
 		service.inFlightExecutions = new Set();
 		service.logger = { debug: vi.fn(), warn: vi.fn() };
@@ -1035,6 +1042,13 @@ describe('InstanceAiService — shutdown', () => {
 		// are left intact (via the delegated sandboxService) so a restarted
 		// process can reconnect to them.
 		expect(service.sandboxService.stopSandboxExpiryTimers).toHaveBeenCalledTimes(1);
+
+		// The durable-log flush must precede eventBus.clear(): clear() invalidates
+		// drain lifecycles, so the reverse order would drop unflushed segment tails.
+		expect(service.eventLog.flushAll).toHaveBeenCalledTimes(1);
+		expect(service.eventLog.flushAll.mock.invocationCallOrder[0]).toBeLessThan(
+			service.eventBus.clear.mock.invocationCallOrder[0],
+		);
 	});
 });
 
@@ -2450,6 +2464,40 @@ describe('InstanceAiService — agent tree snapshots', () => {
 			}),
 		);
 	});
+
+	it('reads snapshot input from the durable log instead of the bus when the flag is on', async () => {
+		const service = createSnapshotService();
+		service.instanceAiConfig.durableLog = true;
+		const logEvent: InstanceAiEvent = {
+			type: 'text-delta',
+			runId: 'run-1',
+			agentId: 'agent-001',
+			payload: { text: 'from the log' },
+		};
+		service.eventLog.getEventsForRuns.mockResolvedValue([logEvent]);
+		const snapshotStorage = {
+			getLatest: vi.fn(async () => undefined),
+			save: vi.fn(async () => {}),
+			updateLast: vi.fn(async () => {}),
+		};
+
+		await service.saveAgentTreeSnapshot('thread-a', 'run-1', snapshotStorage);
+
+		expect(service.eventLog.getEventsForRuns).toHaveBeenCalledWith('thread-a', ['run-1']);
+		expect(service.eventBus.getEventsForRun).not.toHaveBeenCalled();
+		// Read-own-writes barrier: the drain settles before the snapshot input is
+		// read, so a just-published terminal fact can't be missing from the tree.
+		expect(service.eventLog.flush).toHaveBeenCalledWith('thread-a');
+		expect(service.eventLog.flush.mock.invocationCallOrder[0]).toBeLessThan(
+			service.eventLog.getEventsForRuns.mock.invocationCallOrder[0],
+		);
+		expect(snapshotStorage.save).toHaveBeenCalledWith(
+			'thread-a',
+			expect.objectContaining({ textContent: 'from the log' }),
+			'run-1',
+			expect.any(Object),
+		);
+	});
 });
 
 describe('InstanceAiService — terminal response guard wiring', () => {
@@ -2576,7 +2624,7 @@ describe('InstanceAiService — terminal response guard wiring', () => {
 
 	it('claims credits for the consumed segment when a resumed run suspends again', async () => {
 		const service = createTerminalGuardOrderService();
-		vi.spyOn(service.terminalOutcome, 'evaluateWaitingResponse').mockReturnValue(undefined);
+		vi.spyOn(service.terminalOutcome, 'evaluateWaitingResponse').mockResolvedValue(undefined);
 		const abortController = new AbortController();
 		const usageItem = {
 			type: 'llmTokens' as const,
@@ -2630,7 +2678,7 @@ describe('InstanceAiService — terminal response guard wiring', () => {
 
 	it('bills each segment once under disjoint keys across suspend -> resume -> continue', async () => {
 		const service = createTerminalGuardOrderService();
-		vi.spyOn(service.terminalOutcome, 'evaluateWaitingResponse').mockReturnValue(undefined);
+		vi.spyOn(service.terminalOutcome, 'evaluateWaitingResponse').mockResolvedValue(undefined);
 		const abortController = new AbortController();
 		const segmentOneUsage = {
 			type: 'llmTokens' as const,
@@ -2716,7 +2764,7 @@ describe('InstanceAiService — terminal response guard wiring', () => {
 
 	it('bills each segment once under disjoint keys across suspend -> resume -> abort', async () => {
 		const service = createTerminalGuardOrderService();
-		vi.spyOn(service.terminalOutcome, 'evaluateWaitingResponse').mockReturnValue(undefined);
+		vi.spyOn(service.terminalOutcome, 'evaluateWaitingResponse').mockResolvedValue(undefined);
 		const abortController = new AbortController();
 		const segmentOneUsage = {
 			type: 'llmTokens' as const,

packages/cli/src/modules/instance-ai/__tests__/output-redaction-config.test.ts

@@ -17,6 +17,10 @@ describe('resolveOutputRedaction', () => {
 		expect(resolveOutputRedaction(config({ outputRedactionEnabled: false }))).toBe(false);
 	});
 
+	it('returns false when the durable log is on, even with redaction enabled (raw-at-rest)', () => {
+		expect(resolveOutputRedaction(config({ durableLog: true }))).toBe(false);
+	});
+
 	it('maps secrets, PII categories, and placeholder from config', () => {
 		expect(resolveOutputRedaction(config())).toEqual({
 			secrets: true,

packages/cli/src/modules/instance-ai/event-bus/__tests__/in-process-event-bus.test.ts

@@ -6,6 +6,7 @@ import type { InstanceSettings } from 'n8n-core';
 
 import type { Publisher } from '@/scaling/pubsub/publisher.service';
 
+import type { DurableEventLog } from '../durable-event-log';
 import { InProcessEventBus } from '../in-process-event-bus';
 
 function makeEvent(type: string, runId: string): InstanceAiEvent {
@@ -25,6 +26,7 @@ async function flushDrain() {
 describe('InProcessEventBus', () => {
 	let bus: InProcessEventBus;
 	let publisher: ReturnType<typeof mock<Publisher>>;
+	let eventLog: ReturnType<typeof mock<DurableEventLog>>;
 	let instanceSettings: { isMultiMain: boolean };
 
 	/** Shared fake Redis sequence — one Map plays the role of the Redis server,
@@ -71,17 +73,23 @@ describe('InProcessEventBus', () => {
 		},
 	};
 
-	function buildBus() {
+	function buildBus({ durableLog = false } = {}) {
 		const logger = mock<Logger>();
 		logger.scoped.mockReturnValue(logger);
 		publisher = mock<Publisher>();
 		publisher.publishCommand.mockResolvedValue(undefined);
 		publisher.getClient.mockReturnValue(redisClient as never);
-		const globalConfig = mock<GlobalConfig>({ redis: { prefix: 'n8n' } });
+		// Flag off: the durable log is never touched, so a bare mock suffices.
+		eventLog = mock<DurableEventLog>();
+		const globalConfig = mock<GlobalConfig>({
+			redis: { prefix: 'n8n' },
+			instanceAi: { durableLog },
+		});
 		return new InProcessEventBus(
 			logger,
 			instanceSettings as InstanceSettings,
 			publisher,
+			eventLog,
 			globalConfig,
 		);
 	}
@@ -206,7 +214,7 @@ describe('InProcessEventBus', () => {
 
 	describe('subscribe', () => {
 		it('should receive events published after subscription', () => {
-			const received: Array<{ id: number; event: InstanceAiEvent }> = [];
+			const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
 			bus.subscribe('thread-1', (stored) => received.push(stored));
 
 			bus.publish('thread-1', makeEvent('a', 'run_1'));
@@ -218,7 +226,7 @@ describe('InProcessEventBus', () => {
 		});
 
 		it('should not receive events from other threads', () => {
-			const received: Array<{ id: number; event: InstanceAiEvent }> = [];
+			const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
 			bus.subscribe('thread-1', (stored) => received.push(stored));
 
 			bus.publish('thread-2', makeEvent('a', 'run_2'));
@@ -227,7 +235,7 @@ describe('InProcessEventBus', () => {
 		});
 
 		it('should stop delivery after unsubscribe', () => {
-			const received: Array<{ id: number; event: InstanceAiEvent }> = [];
+			const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
 			const unsubscribe = bus.subscribe('thread-1', (stored) => received.push(stored));
 
 			bus.publish('thread-1', makeEvent('a', 'run_1'));
@@ -361,7 +369,7 @@ describe('InProcessEventBus', () => {
 
 	describe('clear', () => {
 		it('should remove all stored events and listeners', async () => {
-			const received: Array<{ id: number; event: InstanceAiEvent }> = [];
+			const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
 			bus.subscribe('thread-1', (stored) => received.push(stored));
 
 			bus.publish('thread-1', makeEvent('a', 'run_1'));
@@ -427,7 +435,7 @@ describe('InProcessEventBus', () => {
 			instanceSettings = { isMultiMain: true };
 			bus = buildBus();
 			const received: number[] = [];
-			bus.subscribe('thread-1', (e) => received.push(e.id));
+			bus.subscribe('thread-1', (e) => received.push(e.id!));
 
 			bus.publish('thread-1', makeEvent('a', 'run_1'));
 			await flushDrain();
@@ -440,7 +448,7 @@ describe('InProcessEventBus', () => {
 			instanceSettings = { isMultiMain: true };
 			bus = buildBus();
 			const received: number[] = [];
-			bus.subscribe('thread-1', (e) => received.push(e.id));
+			bus.subscribe('thread-1', (e) => received.push(e.id!));
 			const huge = makeEvent('a', 'run_1');
 			(huge.payload as { text: string }).text = 'x'.repeat(6 * 1024 * 1024);
 
@@ -457,7 +465,7 @@ describe('InProcessEventBus', () => {
 	describe('handleRelayInstanceAiEvent', () => {
 		it('stores and re-emits a relayed event under its producer-assigned id', () => {
 			const received: number[] = [];
-			bus.subscribe('thread-1', (e) => received.push(e.id));
+			bus.subscribe('thread-1', (e) => received.push(e.id!));
 
 			bus.handleRelayInstanceAiEvent({
 				threadId: 'thread-1',
@@ -497,7 +505,7 @@ describe('InProcessEventBus', () => {
 
 		it('drops a duplicate id instead of storing or emitting it twice', () => {
 			const received: number[] = [];
-			bus.subscribe('thread-1', (e) => received.push(e.id));
+			bus.subscribe('thread-1', (e) => received.push(e.id!));
 			const storedEvent = { id: 5, event: makeEvent('a', 'run_1') };
 
 			bus.handleRelayInstanceAiEvent({ threadId: 'thread-1', storedEvent });
@@ -517,4 +525,111 @@ describe('InProcessEventBus', () => {
 			expect(bus.hasSubscribers('thread-1')).toBe(false);
 		});
 	});
+
+	describe('durable log (flag on)', () => {
+		type EmitFn = (drained: { id?: number; event: InstanceAiEvent; live: boolean }) => void;
+
+		/** Route publish through the mocked drain and hand back its emit callback. */
+		function publishAndCaptureEmit(threadId: string, event: InstanceAiEvent): EmitFn {
+			bus.publish(threadId, event);
+			const call = eventLog.publish.mock.calls.at(-1)!;
+			expect(call[0]).toBe(threadId);
+			expect(call[1]).toBe(event);
+			return call[2] as EmitFn;
+		}
+
+		beforeEach(() => {
+			bus = buildBus({ durableLog: true });
+		});
+
+		it('routes publishes into the durable log instead of the Redis sequence', () => {
+			instanceSettings.isMultiMain = true;
+			bus = buildBus({ durableLog: true });
+
+			bus.publish('thread-1', makeEvent('a', 'run_1'));
+
+			expect(eventLog.publish).toHaveBeenCalledTimes(1);
+			expect(incrbyCalls).toHaveLength(0);
+		});
+
+		it('caches drained durable facts and emits live ones with their DB seq', () => {
+			const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
+			bus.subscribe('thread-1', (stored) => received.push(stored));
+			const event = makeEvent('a', 'run_1');
+
+			const emit = publishAndCaptureEmit('thread-1', event);
+			emit({ id: 7, event, live: true });
+
+			expect(received).toEqual([{ id: 7, event }]);
+			expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([7]);
+		});
+
+		it('emits ephemeral events live without an id and never caches them', () => {
+			const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
+			bus.subscribe('thread-1', (stored) => received.push(stored));
+			const event = makeEvent('a', 'run_1');
+
+			const emit = publishAndCaptureEmit('thread-1', event);
+			emit({ event, live: true });
+
+			expect(received).toEqual([{ event }]);
+			expect(received[0]).not.toHaveProperty('id');
+			expect(bus.getEventsAfter('thread-1', 0)).toHaveLength(0);
+		});
+
+		it('caches a coalesced block without live-emitting it (subscribers saw its deltas)', () => {
+			const received: unknown[] = [];
+			bus.subscribe('thread-1', (stored) => received.push(stored));
+			const event = makeEvent('a', 'run_1');
+
+			const emit = publishAndCaptureEmit('thread-1', event);
+			emit({ id: 3, event, live: false });
+
+			expect(received).toHaveLength(0);
+			expect(bus.getEventsAfter('thread-1', 0).map((e) => e.id)).toEqual([3]);
+		});
+
+		it('relays live drained events to siblings with the DB seq passed through', () => {
+			instanceSettings.isMultiMain = true;
+			bus = buildBus({ durableLog: true });
+			const event = makeEvent('a', 'run_1');
+
+			const emit = publishAndCaptureEmit('thread-1', event);
+			emit({ id: 9, event, live: true });
+			emit({ event, live: true });
+
+			expect(publisher.publishCommand).toHaveBeenCalledTimes(2);
+			expect(publisher.publishCommand).toHaveBeenNthCalledWith(1, {
+				command: 'relay-instance-ai-event',
+				payload: { threadId: 'thread-1', storedEvent: { id: 9, event } },
+			});
+			// The ephemeral relay frame carries no id.
+			expect(publisher.publishCommand).toHaveBeenNthCalledWith(2, {
+				command: 'relay-instance-ai-event',
+				payload: { threadId: 'thread-1', storedEvent: { event } },
+			});
+		});
+
+		it('re-emits a relayed id-less frame to subscribers without storing it', () => {
+			const received: Array<{ id?: number; event: InstanceAiEvent }> = [];
+			bus.subscribe('thread-1', (stored) => received.push(stored));
+
+			bus.handleRelayInstanceAiEvent({
+				threadId: 'thread-1',
+				storedEvent: { event: makeEvent('a', 'run_1') },
+			});
+
+			expect(received).toHaveLength(1);
+			expect(received[0]).not.toHaveProperty('id');
+			expect(bus.getEventsAfter('thread-1', 0)).toHaveLength(0);
+		});
+
+		it('clearThread and clear drop the durable log drain state too', () => {
+			bus.clearThread('thread-1');
+			expect(eventLog.clearThread).toHaveBeenCalledWith('thread-1');
+
+			bus.clear();
+			expect(eventLog.clear).toHaveBeenCalledTimes(1);
+		});
+	});
 });

packages/cli/src/modules/instance-ai/event-bus/in-process-event-bus.ts

@@ -10,6 +10,12 @@ import { InstanceSettings } from 'n8n-core';
 import { MAX_PUBSUB_PAYLOAD_BYTES } from '@/scaling/constants';
 import { Publisher } from '@/scaling/pubsub/publisher.service';
 
+import { DurableEventLog, type DrainedEvent } from './durable-event-log';
+
+// With the durable log ON, the in-memory store is a live-delivery CACHE over
+// instance_ai_events, not the source of truth: eviction bounds the cache and
+// can no longer lose data (replay reads the DB through DurableEventLog).
+// With the flag OFF it is today's only store, and eviction is data loss.
 const MAX_EVENTS_PER_THREAD = 500;
 const MAX_BYTES_PER_THREAD = 2 * 1024 * 1024; // 2 MB
 
@@ -22,11 +28,14 @@ const MAX_BYTES_PER_THREAD = 2 * 1024 * 1024; // 2 MB
  */
 const SEQ_KEY_TTL_SECONDS = 14 * 24 * 60 * 60;
 
+/** Only id-bearing events enter the store; the id is the replay cursor. */
+type SequencedEvent = StoredEvent & { id: number };
+
 @Service()
 export class InProcessEventBus implements InstanceAiEventBus {
 	private readonly emitter = new EventEmitter();
 
-	private readonly store = new Map<string, StoredEvent[]>();
+	private readonly store = new Map<string, SequencedEvent[]>();
 
 	/** Approximate serialized size per thread for eviction. */
 	private readonly sizeBytes = new Map<string, number>();
@@ -56,14 +65,18 @@ export class InProcessEventBus implements InstanceAiEventBus {
 
 	private readonly seqKeyPrefix: string;
 
+	private readonly durableLogEnabled: boolean;
+
 	constructor(
 		private readonly logger: Logger,
 		private readonly instanceSettings: InstanceSettings,
 		private readonly publisher: Publisher,
+		private readonly eventLog: DurableEventLog,
 		globalConfig: GlobalConfig,
 	) {
 		this.logger = this.logger.scoped('instance-ai');
 		this.seqKeyPrefix = `${globalConfig.redis.prefix}:instance-ai:event-seq:`;
+		this.durableLogEnabled = globalConfig.instanceAi.durableLog;
 		// Avoid warnings when many SSE clients connect (each adds a listener per thread)
 		this.emitter.setMaxListeners(0);
 	}
@@ -71,15 +84,29 @@ export class InProcessEventBus implements InstanceAiEventBus {
 	/**
 	 * Publish an event for a thread.
 	 *
-	 * Single-main: assign the next local id and deliver in the same tick.
+	 * Durable log ON: synchronous enqueue into the durable log's per-thread
+	 * drain, which assigns `seq` from the DB, persists durable facts, and hands
+	 * each event back here (`onDrained`): durable ones enter the cache; live
+	 * ones go to local SSE subscribers and — in multi-main — to sibling mains
+	 * via the pubsub relay. Ephemeral events (deltas, status) carry NO id, so
+	 * their SSE frames have no `id:` line and the browser's replay cursor only
+	 * ever points at durable facts. The Redis sequence machinery below is never
+	 * touched; INS-844 composes the two drains into one.
 	 *
-	 * Multi-main: enqueue and drain asynchronously — event ids come from a
-	 * shared per-thread Redis sequence, so every main agrees on them and the
-	 * frontend's replay cursor is valid against any main. The queue preserves
-	 * publish order; each sequenced event is stored, delivered to local SSE
-	 * subscribers, and relayed to sibling mains with its id.
+	 * Flag OFF, single-main: assign the next local id and deliver in the same tick.
+	 *
+	 * Flag OFF, multi-main: enqueue and drain asynchronously — event ids come
+	 * from a shared per-thread Redis sequence, so every main agrees on them and
+	 * the frontend's replay cursor is valid against any main. The queue
+	 * preserves publish order; each sequenced event is stored, delivered to
+	 * local SSE subscribers, and relayed to sibling mains with its id.
 	 */
 	publish(threadId: string, event: InstanceAiEvent): void {
+		if (this.durableLogEnabled) {
+			this.eventLog.publish(threadId, event, (drained) => this.onDrained(threadId, drained));
+			return;
+		}
+
 		if (!this.instanceSettings.isMultiMain) {
 			const id = (this.lastLocalId.get(threadId) ?? 0) + 1;
 			this.lastLocalId.set(threadId, id);
@@ -96,6 +123,39 @@ export class InProcessEventBus implements InstanceAiEventBus {
 		void this.drainQueue(threadId);
 	}
 
+	/**
+	 * An event handed back by the durable log's drain (flag on). Durable events
+	 * (id = DB-assigned seq) enter the live-delivery cache; live ones are
+	 * emitted to local SSE subscribers and relayed to sibling mains. Coalesced
+	 * blocks are durable but NOT live (subscribers already saw their deltas).
+	 */
+	private onDrained(threadId: string, drained: DrainedEvent): void {
+		const sizeBytes = Buffer.byteLength(JSON.stringify(drained.event), 'utf8');
+		if (drained.id !== undefined) {
+			this.cacheSequencedEvent(threadId, { id: drained.id, event: drained.event }, sizeBytes);
+		}
+		if (drained.live) {
+			const stored: StoredEvent = {
+				...(drained.id !== undefined ? { id: drained.id } : {}),
+				event: drained.event,
+			};
+			this.emitter.emit(threadId, stored);
+			this.relayToSiblings(threadId, stored, sizeBytes);
+		}
+	}
+
+	/** Insert into the bounded cache without emitting (durable-log path). */
+	private cacheSequencedEvent(
+		threadId: string,
+		sequenced: SequencedEvent,
+		sizeBytes: number,
+	): void {
+		const events = this.getOrCreateStore(threadId);
+		if (!this.insertById(events, sequenced)) return;
+		this.sizeBytes.set(threadId, (this.sizeBytes.get(threadId) ?? 0) + sizeBytes);
+		this.evictIfNeeded(threadId, events);
+	}
+
 	/**
 	 * Assign sequence ids to queued events and dispatch them, preserving
 	 * publish order. Only one drain runs per thread; events queued while a
@@ -112,7 +172,7 @@ export class InProcessEventBus implements InstanceAiEventBus {
 				// Never throws — falls back to local ids on Redis failure.
 				const firstId = await this.assignSequenceBlock(threadId, batch.length);
 				for (let i = 0; i < batch.length; i++) {
-					const stored: StoredEvent = { id: firstId + i, event: batch[i] };
+					const stored: SequencedEvent = { id: firstId + i, event: batch[i] };
 					// Serialize once: reused for the store's size accounting and the relay guard.
 					const sizeBytes = Buffer.byteLength(JSON.stringify(batch[i]), 'utf8');
 					this.storeAndEmit(threadId, stored, sizeBytes);
@@ -199,7 +259,7 @@ export class InProcessEventBus implements InstanceAiEventBus {
 		}
 	}
 
-	private storeAndEmit(threadId: string, stored: StoredEvent, eventSizeBytes?: number): void {
+	private storeAndEmit(threadId: string, stored: SequencedEvent, eventSizeBytes?: number): void {
 		const size = eventSizeBytes ?? Buffer.byteLength(JSON.stringify(stored.event), 'utf8');
 		const events = this.getOrCreateStore(threadId);
 
@@ -220,7 +280,7 @@ export class InProcessEventBus implements InstanceAiEventBus {
 	 * background task while the orchestrator runs elsewhere) can arrive with a
 	 * lower id than the latest stored one. Returns false for a duplicate id.
 	 */
-	private insertById(events: StoredEvent[], stored: StoredEvent): boolean {
+	private insertById(events: SequencedEvent[], stored: SequencedEvent): boolean {
 		if (events.length === 0 || events[events.length - 1].id < stored.id) {
 			events.push(stored);
 			return true;
@@ -257,8 +317,10 @@ export class InProcessEventBus implements InstanceAiEventBus {
 	}
 
 	/** A relayed event from another main, carrying its producer-assigned id
-	 *  from the shared sequence. Stored/re-emitted only if this main holds a
-	 *  subscription for the thread (avoids every main buffering every thread). */
+	 *  from the shared sequence (or the DB-assigned seq with the durable log
+	 *  on; id-less = ephemeral, live-only). Stored/re-emitted only if this main
+	 *  holds a subscription for the thread (avoids every main buffering every
+	 *  thread). */
 	@OnPubSubEvent('relay-instance-ai-event', { instanceType: 'main' })
 	handleRelayInstanceAiEvent({
 		threadId,
@@ -266,9 +328,17 @@ export class InProcessEventBus implements InstanceAiEventBus {
 	}: { threadId: string; storedEvent: StoredEvent }): void {
 		// Track the shared-sequence high-water mark even without subscribers, so
 		// a Redis-outage fallback keeps assigning ids above what siblings used.
-		this.bumpLocalHighWaterMark(threadId, storedEvent.id);
+		if (storedEvent.id !== undefined) {
+			this.bumpLocalHighWaterMark(threadId, storedEvent.id);
+		}
 		if (!this.hasSubscribers(threadId)) return;
-		this.storeAndEmit(threadId, storedEvent);
+		if (storedEvent.id === undefined) {
+			// Ephemeral durable-log frame: deliver live, never store (the DB seq
+			// is the shared replay authority, so the cache doesn't need it).
+			this.emitter.emit(threadId, storedEvent);
+			return;
+		}
+		this.storeAndEmit(threadId, { id: storedEvent.id, event: storedEvent.event });
 	}
 
 	subscribe(threadId: string, handler: (storedEvent: StoredEvent) => void): () => void {
@@ -285,6 +355,9 @@ export class InProcessEventBus implements InstanceAiEventBus {
 	 * Events still awaiting a sequence number are intentionally excluded: they
 	 * have no id yet, and once sequenced they reach subscribers live — the SSE
 	 * bootstrap subscribes before calling this, so nothing is missed.
+	 *
+	 * Durable log ON: cache-scoped read for same-process consumers;
+	 * cross-restart/cross-main replay must use DurableEventLog.getEventsAfter.
 	 */
 	getEventsAfter(threadId: string, afterId: number): StoredEvent[] {
 		const events = this.store.get(threadId);
@@ -314,6 +387,12 @@ export class InProcessEventBus implements InstanceAiEventBus {
 	}
 
 	async getNextEventId(threadId: string): Promise<number> {
+		if (this.durableLogEnabled) {
+			// Cache-scoped; the durable authority is DurableEventLog.getNextEventId.
+			const events = this.store.get(threadId);
+			const last = events?.length ? events[events.length - 1].id : undefined;
+			return (last ?? 0) + 1;
+		}
 		if (this.instanceSettings.isMultiMain) {
 			try {
 				const value = await this.getRedisClient().get(this.seqKey(threadId));
@@ -335,6 +414,7 @@ export class InProcessEventBus implements InstanceAiEventBus {
 		this.lastLocalId.delete(threadId);
 		this.pendingByThread.delete(threadId);
 		this.inFlightByThread.delete(threadId);
+		this.eventLog.clearThread(threadId);
 		this.emitter.removeAllListeners(threadId);
 		if (this.instanceSettings.isMultiMain) {
 			// Every main clears on thread deletion (task-control broadcast), so the
@@ -358,10 +438,11 @@ export class InProcessEventBus implements InstanceAiEventBus {
 		this.lastLocalId.clear();
 		this.pendingByThread.clear();
 		this.inFlightByThread.clear();
+		this.eventLog.clear();
 		this.emitter.removeAllListeners();
 	}
 
-	private evictIfNeeded(threadId: string, events: StoredEvent[]): void {
+	private evictIfNeeded(threadId: string, events: SequencedEvent[]): void {
 		let totalSize = this.sizeBytes.get(threadId) ?? 0;
 
 		while (events.length > MAX_EVENTS_PER_THREAD || totalSize > MAX_BYTES_PER_THREAD) {
@@ -373,7 +454,7 @@ export class InProcessEventBus implements InstanceAiEventBus {
 		this.sizeBytes.set(threadId, Math.max(0, totalSize));
 	}
 
-	private getOrCreateStore(threadId: string): StoredEvent[] {
+	private getOrCreateStore(threadId: string): SequencedEvent[] {
 		let events = this.store.get(threadId);
 		if (!events) {
 			events = [];

packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts

@@ -23,9 +23,11 @@ import { DbSnapshotStorage } from './storage/db-snapshot-storage';
 import { BadRequestError } from '@/errors/response-errors/bad-request.error';
 import { NotFoundError } from '@/errors/response-errors/not-found.error';
 
+import { DurableLogMetrics } from './event-bus/durable-log-metrics';
 import {
 	collectConfirmationRequestIds,
 	markExpiredConfirmations,
+	messageParserStats,
 	parseStoredMessages,
 } from './message-parser';
 import { InstanceAiCheckpointRepository } from './repositories/instance-ai-checkpoint.repository';
@@ -96,6 +98,7 @@ export class InstanceAiMemoryService {
 		private readonly dbSnapshotStorage: DbSnapshotStorage,
 		private readonly checkpointRepository: InstanceAiCheckpointRepository,
 		private readonly pendingConfirmationRepository: InstanceAiPendingConfirmationRepository,
+		private readonly durableLogMetrics: DurableLogMetrics,
 	) {
 		this.instanceAiConfig = globalConfig.instanceAi;
 	}
@@ -244,7 +247,11 @@ export class InstanceAiMemoryService {
 		const checkpointMessages = await this.loadInFlightCheckpointMessages(threadId);
 		const storedMessages = mergeMessagesById(result.messages, checkpointMessages);
 
+		const fallbacksBefore = messageParserStats.fallbackActivations;
 		const messages = parseStoredMessages(storedMessages, snapshots);
+		this.durableLogMetrics.notifyParserFallbacks(
+			messageParserStats.fallbackActivations - fallbacksBefore,
+		);
 		await this.flagExpiredConfirmations(messages);
 
 		const projectId = await this.agentMemory.getThreadProjectId(threadId);

packages/cli/src/modules/instance-ai/instance-ai-terminal-outcome.service.ts

@@ -87,10 +87,15 @@ function appendTerminalOutcomeToAgentTree(
 // The slice of each collaborator the terminal-outcome coordinator actually
 // uses. Anchored to the concrete types via `Pick` so the signatures stay in
 // sync with the source.
-export type InstanceAiTerminalOutcomeEventBus = Pick<
-	InProcessEventBus,
-	'getEventsForRun' | 'getEventsForRuns' | 'publish'
->;
+// Reads may be sync (in-memory bus, flag off) or async (durable log, flag on);
+// the host injects a flag-resolved adapter.
+export type InstanceAiTerminalOutcomeEventBus = Pick<InProcessEventBus, 'publish'> & {
+	getEventsForRun(threadId: string, runId: string): InstanceAiEvent[] | Promise<InstanceAiEvent[]>;
+	getEventsForRuns(
+		threadId: string,
+		runIds: string[],
+	): InstanceAiEvent[] | Promise<InstanceAiEvent[]>;
+};
 
 export type InstanceAiTerminalOutcomeSnapshotStorage = Pick<
 	DbSnapshotStorage,
@@ -116,6 +121,13 @@ export type InstanceAiTerminalOutcomeTracing = Pick<
 
 export interface InstanceAiTerminalOutcomeServiceOptions {
 	eventBus: InstanceAiTerminalOutcomeEventBus;
+	/**
+	 * Durable-log flag: outcome lines publish as `text-block` (a structural
+	 * fact, persisted before it is emitted live) instead of a trailing
+	 * `text-delta`, so a page reload right after a background outcome folds
+	 * the line from the log instead of racing the coalescer's idle flush.
+	 */
+	durableLog: boolean;
 	dbSnapshotStorage: InstanceAiTerminalOutcomeSnapshotStorage;
 	agentMemory: PatchableThreadMemory;
 	telemetry: InstanceAiTerminalOutcomeTelemetry;
@@ -168,6 +180,8 @@ export class InstanceAiTerminalOutcomeService {
 
 	private readonly eventBus: InstanceAiTerminalOutcomeEventBus;
 
+	private readonly durableLog: boolean;
+
 	private readonly dbSnapshotStorage: InstanceAiTerminalOutcomeSnapshotStorage;
 
 	private readonly agentMemory: PatchableThreadMemory;
@@ -188,6 +202,7 @@ export class InstanceAiTerminalOutcomeService {
 
 	constructor(options: InstanceAiTerminalOutcomeServiceOptions) {
 		this.eventBus = options.eventBus;
+		this.durableLog = options.durableLog;
 		this.dbSnapshotStorage = options.dbSnapshotStorage;
 		this.agentMemory = options.agentMemory;
 		this.telemetry = options.telemetry;
@@ -199,7 +214,7 @@ export class InstanceAiTerminalOutcomeService {
 		this.saveAgentTreeSnapshot = options.saveAgentTreeSnapshot;
 	}
 
-	evaluateTerminalResponse(
+	async evaluateTerminalResponse(
 		threadId: string,
 		runId: string,
 		status: Exclude<TerminalResponseStatus, 'waiting'>,
@@ -211,7 +226,7 @@ export class InstanceAiTerminalOutcomeService {
 			errorCode?: InstanceAiErrorCode;
 			suppressCompletedFallback?: boolean;
 		} = {},
-	): TerminalResponseDecision | undefined {
+	): Promise<TerminalResponseDecision | undefined> {
 		const guard = new InstanceAiTerminalResponseGuard({
 			runId,
 			rootAgentId: orchestratorAgentId(runId),
@@ -219,7 +234,7 @@ export class InstanceAiTerminalOutcomeService {
 			correlationId: options.correlationId,
 		});
 		const decision = guard.evaluateTerminal(
-			this.getTerminalGuardEvents(threadId, runId, options.messageGroupId),
+			await this.getTerminalGuardEvents(threadId, runId, options.messageGroupId),
 			status,
 			{
 				workSummary: options.workSummary,
@@ -232,12 +247,12 @@ export class InstanceAiTerminalOutcomeService {
 		return decision;
 	}
 
-	evaluateWaitingResponse(
+	async evaluateWaitingResponse(
 		threadId: string,
 		runId: string,
 		confirmationEvent: Extract<InstanceAiEvent, { type: 'confirmation-request' }> | undefined,
 		options: { messageGroupId?: string; correlationId?: string } = {},
-	): TerminalResponseDecision | undefined {
+	): Promise<TerminalResponseDecision | undefined> {
 		const guard = new InstanceAiTerminalResponseGuard({
 			runId,
 			rootAgentId: orchestratorAgentId(runId),
@@ -245,24 +260,24 @@ export class InstanceAiTerminalOutcomeService {
 			correlationId: options.correlationId,
 		});
 		const decision = guard.evaluateWaiting(
-			this.getTerminalGuardEvents(threadId, runId, options.messageGroupId),
+			await this.getTerminalGuardEvents(threadId, runId, options.messageGroupId),
 			confirmationEvent,
 		);
 		this.handleTerminalResponseDecision(threadId, runId, decision, options.messageGroupId);
 		return decision;
 	}
 
-	private getTerminalGuardEvents(
+	private async getTerminalGuardEvents(
 		threadId: string,
 		runId: string,
 		messageGroupId?: string,
-	): InstanceAiEvent[] {
-		if (!messageGroupId) return this.eventBus.getEventsForRun(threadId, runId);
+	): Promise<InstanceAiEvent[]> {
+		if (!messageGroupId) return await this.eventBus.getEventsForRun(threadId, runId);
 
 		const groupRunIds = this.runState.getRunIdsForMessageGroup(messageGroupId);
 		return groupRunIds.length > 0
-			? this.eventBus.getEventsForRuns(threadId, groupRunIds)
-			: this.eventBus.getEventsForRun(threadId, runId);
+			? await this.eventBus.getEventsForRuns(threadId, groupRunIds)
+			: await this.eventBus.getEventsForRun(threadId, runId);
 	}
 
 	private handleTerminalResponseDecision(
@@ -398,7 +413,7 @@ export class InstanceAiTerminalOutcomeService {
 					error: getErrorMessage(error),
 				});
 				if (delivery === 'event') {
-					const published = this.publishTerminalOutcomeLine(outcome, responseId);
+					const published = await this.publishTerminalOutcomeLine(outcome, responseId);
 					this.telemetry.track('instance_ai_terminal_response_decision', {
 						thread_id: threadId,
 						run_id: outcome.runId,
@@ -417,7 +432,7 @@ export class InstanceAiTerminalOutcomeService {
 
 			let action = 'replay_snapshot';
 			if (delivery === 'event') {
-				const published = this.publishTerminalOutcomeLine(outcome, responseId);
+				const published = await this.publishTerminalOutcomeLine(outcome, responseId);
 				action = published ? 'replay_event' : 'already-emitted';
 			}
 
@@ -480,14 +495,17 @@ export class InstanceAiTerminalOutcomeService {
 		return true;
 	}
 
-	private publishTerminalOutcomeLine(outcome: TerminalOutcome, responseId: string): boolean {
-		const alreadyPublished = this.eventBus
-			.getEventsForRun(outcome.threadId, outcome.runId)
-			.some((event) => event.responseId === responseId);
+	private async publishTerminalOutcomeLine(
+		outcome: TerminalOutcome,
+		responseId: string,
+	): Promise<boolean> {
+		const alreadyPublished = (
+			await this.eventBus.getEventsForRun(outcome.threadId, outcome.runId)
+		).some((event) => event.responseId === responseId);
 		if (alreadyPublished) return false;
 
 		this.eventBus.publish(outcome.threadId, {
-			type: 'text-delta',
+			type: this.durableLog ? 'text-block' : 'text-delta',
 			runId: outcome.runId,
 			agentId: orchestratorAgentId(outcome.runId),
 			responseId,
@@ -520,7 +538,7 @@ export class InstanceAiTerminalOutcomeService {
 		}
 
 		const responseId = getBackgroundOutcomeResponseId(outcome);
-		const published = this.publishTerminalOutcomeLine(outcome, responseId);
+		const published = await this.publishTerminalOutcomeLine(outcome, responseId);
 
 		this.telemetry.track('instance_ai_terminal_response_decision', {
 			thread_id: task.threadId,

packages/cli/src/modules/instance-ai/instance-ai.controller.ts

@@ -18,7 +18,7 @@ import {
 	InstanceAiEvalRestoreThreadRequest,
 	normalizeInstanceAiThreadSource,
 } from '@n8n/api-types';
-import type { InstanceAiAgentNode } from '@n8n/api-types';
+import type { InstanceAiAgentNode, InstanceAiEvent } from '@n8n/api-types';
 import { ModuleRegistry } from '@n8n/backend-common';
 import { GlobalConfig } from '@n8n/config';
 import { AuthenticatedRequest, User, UserRepository } from '@n8n/db';
@@ -44,6 +44,8 @@ import { InstanceAiBrowserSessionService } from './browser/instance-ai-browser-s
 import { EvalExecutionService } from './eval/execution.service';
 import { EvalThreadCredentialAllowlistService } from './eval/thread-credential-allowlist.service';
 import { EvalThreadRestoreService } from './eval/thread-restore.service';
+import { DurableEventLog } from './event-bus/durable-event-log';
+import { DurableLogMetrics } from './event-bus/durable-log-metrics';
 import { InProcessEventBus } from './event-bus/in-process-event-bus';
 import { InstanceAiErrorReporterService } from './instance-ai-error-reporter.service';
 import { InstanceAiGatewayService } from './instance-ai-gateway.service';
@@ -68,6 +70,10 @@ const KEEP_ALIVE_INTERVAL_MS = 15_000;
 export class InstanceAiController {
 	private readonly gatewayApiKey: string;
 
+	/** Durable-log prototype flag (N8N_INSTANCE_AI_DURABLE_LOG): replay and
+	 *  cursors come from the DB-backed log instead of the in-memory bus. */
+	private readonly durableLogEnabled: boolean;
+
 	private static getTreeRichnessScore(tree: InstanceAiAgentNode): number {
 		let score = 0;
 		const stack = [tree];
@@ -108,6 +114,8 @@ export class InstanceAiController {
 		private readonly evalCredentialAllowlists: EvalThreadCredentialAllowlistService,
 		private readonly evalThreadRestore: EvalThreadRestoreService,
 		private readonly eventBus: InProcessEventBus,
+		private readonly eventLog: DurableEventLog,
+		private readonly durableLogMetrics: DurableLogMetrics,
 		private readonly moduleRegistry: ModuleRegistry,
 		private readonly push: Push,
 		private readonly urlService: UrlService,
@@ -118,6 +126,7 @@ export class InstanceAiController {
 		globalConfig: GlobalConfig,
 	) {
 		this.gatewayApiKey = globalConfig.instanceAi.gatewayApiKey;
+		this.durableLogEnabled = globalConfig.instanceAi.durableLog;
 	}
 
 	private requireInstanceAiEnabled(): void {
@@ -359,25 +368,17 @@ export class InstanceAiController {
 		// The client may have disconnected during the awaits above.
 		if (closed) return;
 
-		// 6. Replay missed events, emit run-sync frames, and flip to live delivery
-		//    in one synchronous block. The event bus store and emitter are
-		//    synchronous, so no event can slip between the replay and the live
-		//    handler taking over. Events that arrived during the awaits above are
-		//    already in the store (the early subscription in step 1 keeps relayed
-		//    events flowing in multi-main) and are included in the replay here.
-		const missed = this.eventBus.getEventsAfter(threadId, cursor);
-		for (const stored of missed) {
-			deliver(stored);
-		}
-
-		// 6b. Bootstrap sync: emit one run-sync control frame per live message
-		//     group. Each frame uses a named SSE event type (event: run-sync) with
-		//     NO id: field so the browser's lastEventId is unaffected and the
-		//     replay cursor stays consistent.
-		for (const [groupId, group] of liveGroups) {
-			const runEvents = this.eventBus.getEventsForRuns(threadId, group.runIds);
+		// 6b (used by both arms below). Emit one run-sync control frame for a live
+		//     message group. Each frame uses a named SSE event type
+		//     (event: run-sync) with NO id: field so the browser's lastEventId is
+		//     unaffected and the replay cursor stays consistent.
+		const writeRunSyncFrame = (
+			groupId: string,
+			group: { runIds: string[]; status: 'active' | 'suspended' | 'background' },
+			runEvents: InstanceAiEvent[],
+		) => {
 			const persistedSnapshot = persistedSnapshots.get(groupId);
-			if (runEvents.length === 0 && !persistedSnapshot) continue;
+			if (runEvents.length === 0 && !persistedSnapshot) return;
 
 			const eventTree = buildAgentTreeFromEvents(runEvents);
 			const agentTree = InstanceAiController.selectBootstrapTree(
@@ -394,6 +395,64 @@ export class InstanceAiController {
 					backgroundTasks: threadStatus.backgroundTasks,
 				})}\n\n`,
 			);
+		};
+
+		if (this.durableLogEnabled) {
+			// 6. Replay missed events from the DURABLE log — survives restarts and is
+			//    valid on any main (the table is in the shared DB). The reads are
+			//    async, so unlike the old synchronous memory-store replay, live
+			//    events can land mid-bootstrap: buffer them across every await (the
+			//    replay read AND the run-sync tree reads) and flush with seq dedupe
+			//    only when no await remains before live delivery takes over (the
+			//    drain persists before it emits, so a fact is never in neither
+			//    place). A flushed event may already be folded into a run-sync tree;
+			//    the shared reducer applies it idempotently, same as any live event
+			//    arriving after a frame.
+			const arrivedDuringReplay: StoredEvent[] = [];
+			const stopBuffering = this.eventBus.subscribe(threadId, (stored) => {
+				arrivedDuringReplay.push(stored);
+			});
+			try {
+				const missed = await this.eventLog.getEventsAfter(threadId, cursor);
+				// The client may have disconnected during the read: stop before
+				// writing to the dead response or arming the keep-alive below.
+				if (closed) return;
+				let lastReplayedSeq = cursor;
+				for (const stored of missed) {
+					deliver(stored);
+					if (stored.id !== undefined) lastReplayedSeq = stored.id;
+				}
+				// Build each live group's bootstrap tree from the durable log, so the
+				// group renders fully even when the bus cache was evicted, the process
+				// restarted, or this main never buffered the thread (sibling main).
+				for (const [groupId, group] of liveGroups) {
+					const runEvents = await this.eventLog.getEventsForRuns(threadId, group.runIds);
+					if (closed) return;
+					writeRunSyncFrame(groupId, group, runEvents);
+				}
+				for (const stored of arrivedDuringReplay) {
+					if (stored.id === undefined || stored.id > lastReplayedSeq) deliver(stored);
+				}
+				this.durableLogMetrics.recordReplay(missed.length, Math.max(0, lastReplayedSeq - cursor));
+			} finally {
+				// The buffering subscription must not outlive the bootstrap, even when
+				// a durable read throws.
+				stopBuffering();
+			}
+		} else {
+			// 6. Replay missed events, emit run-sync frames, and flip to live delivery
+			//    in one synchronous block. The event bus store and emitter are
+			//    synchronous, so no event can slip between the replay and the live
+			//    handler taking over. Events that arrived during the awaits above are
+			//    already in the store (the early subscription in step 1 keeps relayed
+			//    events flowing in multi-main) and are included in the replay here.
+			const missed = this.eventBus.getEventsAfter(threadId, cursor);
+			for (const stored of missed) {
+				deliver(stored);
+			}
+			for (const [groupId, group] of liveGroups) {
+				writeRunSyncFrame(groupId, group, this.eventBus.getEventsForRuns(threadId, group.runIds));
+			}
 		}
 		if (liveGroups.size > 0) res.flush?.();
 
@@ -694,8 +753,11 @@ export class InstanceAiController {
 
 		// Include the next SSE event ID so the frontend can skip past events
 		// already covered by these historical messages (prevents duplicates).
-		// Read from the shared sequence, so the cursor is valid against any main.
-		const nextEventId = await this.eventBus.getNextEventId(threadId);
+		// Flag on: durable authority, valid across restarts and mains. Flag off:
+		// the shared sequence, so the cursor is valid against any main.
+		const nextEventId = this.durableLogEnabled
+			? await this.eventLog.getNextEventId(threadId)
+			: await this.eventBus.getNextEventId(threadId);
 		return { ...result, nextEventId };
 	}
 
@@ -1138,8 +1200,12 @@ export class InstanceAiController {
 	}
 
 	private writeSseEvent(res: FlushableResponse, stored: StoredEvent): void {
-		// No `event:` field — events are discriminated by data.type per streaming-protocol.md
-		res.write(`id: ${stored.id}\ndata: ${JSON.stringify(stored.event)}\n\n`);
+		// No `event:` field — events are discriminated by data.type per streaming-protocol.md.
+		// Ephemeral events (deltas/status) carry no `id:` line, so the browser's
+		// Last-Event-ID only ever advances on durable facts — same precedent as
+		// the run-sync control frames above.
+		const idLine = stored.id !== undefined ? `id: ${stored.id}\n` : '';
+		res.write(`${idLine}data: ${JSON.stringify(stored.event)}\n\n`);
 		res.flush?.();
 	}
 }

packages/cli/src/modules/instance-ai/instance-ai.service.ts

@@ -112,6 +112,7 @@ import { Telemetry } from '@/telemetry';
 import { composeLocalMcpServers } from './browser/composite-local-mcp-server';
 import { InstanceAiBrowserSessionService } from './browser/instance-ai-browser-session.service';
 import { EvalThreadCredentialAllowlistService } from './eval/thread-credential-allowlist.service';
+import { DurableEventLog } from './event-bus/durable-event-log';
 import { InProcessEventBus } from './event-bus/in-process-event-bus';
 import { InstanceAiCreditService } from './instance-ai-credit.service';
 import { BROWSER_TOOL_CATEGORY, InstanceAiGatewayService } from './instance-ai-gateway.service';
@@ -489,6 +490,7 @@ export class InstanceAiService {
 		private readonly instanceSettings: InstanceSettings,
 		private readonly adapterService: InstanceAiAdapterService,
 		private readonly eventBus: InProcessEventBus,
+		private readonly eventLog: DurableEventLog,
 		private readonly settingsService: InstanceAiSettingsService,
 		private readonly gatewayService: InstanceAiGatewayService,
 		private readonly browserSessionService: InstanceAiBrowserSessionService,
@@ -577,7 +579,21 @@ export class InstanceAiService {
 			aiService: this.aiService,
 		});
 		this.terminalOutcome = new InstanceAiTerminalOutcomeService({
-			eventBus: this.eventBus,
+			durableLog: globalConfig.instanceAi.durableLog,
+			// Flag-resolved reads: the terminal guard and outcome-replay dedup must
+			// see the run's events after a restart too, which only the durable log
+			// can provide (the bus cache is empty in a fresh process).
+			eventBus: {
+				publish: (threadId, event) => this.eventBus.publish(threadId, event),
+				getEventsForRun: async (threadId, runId) =>
+					this.instanceAiConfig.durableLog
+						? await this.readDurableEventsForRuns(threadId, [runId])
+						: this.eventBus.getEventsForRun(threadId, runId),
+				getEventsForRuns: async (threadId, runIds) =>
+					this.instanceAiConfig.durableLog
+						? await this.readDurableEventsForRuns(threadId, runIds)
+						: this.eventBus.getEventsForRuns(threadId, runIds),
+			},
 			dbSnapshotStorage: this.dbSnapshotStorage,
 			agentMemory: this.agentMemory,
 			telemetry: this.telemetry,
@@ -1444,6 +1460,10 @@ export class InstanceAiService {
 		this.domainAccessTrackersByThread.clear();
 		this.tracing.clear();
 
+		// Durable-log flag: flush in-flight drains + open coalesce buffers so the
+		// tail of every streamed segment survives the restart. No-op when off.
+		await this.eventLog.flushAll();
+
 		this.eventBus.clear();
 		await this._mcpClientManager?.disconnect();
 		this.logger.debug('Instance AI service shut down');
@@ -3010,7 +3030,7 @@ export class InstanceAiService {
 			// Check if already cancelled before starting agent work
 			if (signal.aborted) {
 				await this.persistInterruptedUserMessage(threadId, user.id, message, turnStartedAt);
-				this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'cancelled', {
+				await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'cancelled', {
 					messageGroupId,
 					correlationId: messageId,
 				});
@@ -3370,7 +3390,7 @@ export class InstanceAiService {
 					});
 				}
 
-				const waitingDecision = this.terminalOutcome.evaluateWaitingResponse(
+				const waitingDecision = await this.terminalOutcome.evaluateWaitingResponse(
 					threadId,
 					runId,
 					result.confirmationEvent,
@@ -3471,7 +3491,7 @@ export class InstanceAiService {
 			const userFacingErrorCode =
 				result.status === 'errored' ? getUserFacingErrorCode(result.error) : undefined;
 			if (runControl.shouldEmitTerminalOutcome(result.stopReason)) {
-				this.terminalOutcome.evaluateTerminalResponse(threadId, runId, result.status, {
+				await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, result.status, {
 					messageGroupId,
 					correlationId: messageId,
 					workSummary: result.workSummary,
@@ -3559,7 +3579,7 @@ export class InstanceAiService {
 				if (cancellationReason === INSTANCE_AI_RUN_TIMEOUT_REASON) {
 					this.liveness.publishRunTimeoutNotice(threadId, runId);
 				}
-				this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'cancelled', {
+				await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'cancelled', {
 					messageGroupId,
 					correlationId: messageId,
 				});
@@ -3614,7 +3634,7 @@ export class InstanceAiService {
 				...buildInstanceAiObservabilityContext(errCtx),
 			});
 			this.instanceAiErrorReporter.report(error, { component: 'instance-ai-run', ...errCtx });
-			this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'errored', {
+			await this.terminalOutcome.evaluateTerminalResponse(threadId, runId, 'errored', {
 				messageGroupId,
 				correlationId: messageId,
 				errorMessage: userFacingErrorMessage,
@@ -4513,7 +4533,7 @@ export class InstanceAiService {
 				}
 
 				const messageGroupId = this.tracing.getMessageGroupId(opts.runId);
-				const waitingDecision = this.terminalOutcome.evaluateWaitingResponse(
+				const waitingDecision = await this.terminalOutcome.evaluateWaitingResponse(
 					opts.threadId,
 					opts.runId,
 					result.confirmationEvent,
@@ -4605,15 +4625,20 @@ export class InstanceAiService {
 			const userFacingErrorCode =
 				result.status === 'errored' ? getUserFacingErrorCode(result.error) : undefined;
 			if (runControl.shouldEmitTerminalOutcome(result.stopReason)) {
-				this.terminalOutcome.evaluateTerminalResponse(opts.threadId, opts.runId, result.status, {
-					messageGroupId,
-					workSummary: result.workSummary,
-					errorMessage: userFacingErrorMessage,
-					errorCode: userFacingErrorCode,
-					suppressCompletedFallback:
-						opts.checkpoint?.isCheckpointFollowUp === true ||
-						opts.plannedBuild?.isPlannedBuildFollowUp === true,
-				});
+				await this.terminalOutcome.evaluateTerminalResponse(
+					opts.threadId,
+					opts.runId,
+					result.status,
+					{
+						messageGroupId,
+						workSummary: result.workSummary,
+						errorMessage: userFacingErrorMessage,
+						errorCode: userFacingErrorCode,
+						suppressCompletedFallback:
+							opts.checkpoint?.isCheckpointFollowUp === true ||
+							opts.plannedBuild?.isPlannedBuildFollowUp === true,
+					},
+				);
 			}
 			const finalStatus = result.status === 'errored' ? 'error' : result.status;
 			await this.tracing.finalizeRunTracing(opts.runId, opts.tracing, {
@@ -4689,9 +4714,14 @@ export class InstanceAiService {
 				if (cancellationReason === INSTANCE_AI_RUN_TIMEOUT_REASON) {
 					this.liveness.publishRunTimeoutNotice(opts.threadId, opts.runId);
 				}
-				this.terminalOutcome.evaluateTerminalResponse(opts.threadId, opts.runId, 'cancelled', {
-					messageGroupId,
-				});
+				await this.terminalOutcome.evaluateTerminalResponse(
+					opts.threadId,
+					opts.runId,
+					'cancelled',
+					{
+						messageGroupId,
+					},
+				);
 				await this.tracing.finalizeRunTracing(opts.runId, opts.tracing, {
 					status: 'cancelled',
 					reason: cancellationReason,
@@ -4741,7 +4771,7 @@ export class InstanceAiService {
 				...buildInstanceAiObservabilityContext(errCtx),
 			});
 			this.instanceAiErrorReporter.report(error, { component: 'instance-ai-run', ...errCtx });
-			this.terminalOutcome.evaluateTerminalResponse(opts.threadId, opts.runId, 'errored', {
+			await this.terminalOutcome.evaluateTerminalResponse(opts.threadId, opts.runId, 'errored', {
 				messageGroupId,
 				errorMessage: userFacingErrorMessage,
 				errorCode: userFacingErrorCode,
@@ -5381,6 +5411,21 @@ export class InstanceAiService {
 		return '';
 	}
 
+	/**
+	 * Read-own-writes barrier for decision reads from the durable log: settle
+	 * the thread's drain (including open coalesce buffers) so everything
+	 * published before this call is visible to the read. Used at run
+	 * boundaries — terminal-guard inputs and snapshot builds — where closing
+	 * the open segment early is correct anyway.
+	 */
+	private async readDurableEventsForRuns(
+		threadId: string,
+		runIds: string[],
+	): Promise<InstanceAiEvent[]> {
+		await this.eventLog.flush(threadId);
+		return await this.eventLog.getEventsForRuns(threadId, runIds);
+	}
+
 	/**
 	 * 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).
@@ -5403,10 +5448,18 @@ export class InstanceAiService {
 					const snapshot = await snapshotStorage.getLatest(threadId, { messageGroupId, runId });
 					groupRunIds = snapshot?.runIds?.length ? snapshot.runIds : [runId];
 				}
-				events = this.eventBus.getEventsForRuns(threadId, groupRunIds);
+				events = this.instanceAiConfig.durableLog
+					? await this.readDurableEventsForRuns(threadId, groupRunIds)
+					: this.eventBus.getEventsForRuns(threadId, groupRunIds);
 			} else {
-				events = this.eventBus.getEventsForRun(threadId, runId);
+				events = this.instanceAiConfig.durableLog
+					? await this.readDurableEventsForRuns(threadId, [runId])
+					: this.eventBus.getEventsForRun(threadId, runId);
 			}
+			// Durable-log flag on: the tree input comes from the DB, so long runs can
+			// no longer out-evict their own snapshot input (the empty-agentTree bug
+			// class). The snapshot write itself stays during migration so pre-log
+			// threads keep rendering; INS-841 moves history to fold-on-read.
 			if (isUpdate && events.length === 0) {
 				this.logger.warn('Skipped updating empty Instance AI agent tree snapshot', {
 					threadId,

packages/cli/src/modules/instance-ai/message-parser.ts

@@ -333,6 +333,13 @@ function buildSnapshotMessage(snapshot: AgentTreeSnapshot): InstanceAiMessage {
 // Main parser
 // ---------------------------------------------------------------------------
 
+/**
+ * Durable-log instrumentation: counts assistant messages that rendered from
+ * the message-derived fallback ladder instead of a renderable snapshot tree.
+ * Forwarded to the metrics pipeline via DurableLogMetrics.notifyParserFallbacks.
+ */
+export const messageParserStats = { fallbackActivations: 0 };
+
 /**
  * Converts persisted native agent messages into rich InstanceAiMessage objects
  * with agent trees (from snapshots or reconstructed flat trees).
@@ -462,6 +469,7 @@ export function parseStoredMessages(
 			// empty one.
 			const snapshotIsRenderable = snapshot !== undefined && isRenderableTree(snapshot.tree);
 			const agentTree = snapshotIsRenderable ? snapshot.tree : messageFlatTree;
+			if (!snapshotIsRenderable && messageFlatTree) messageParserStats.fallbackActivations++;
 
 			const assistantMessage: InstanceAiMessage = {
 				id: msg.id,

packages/cli/src/modules/instance-ai/output-redaction-config.ts

@@ -8,8 +8,15 @@ import type { InstanceAiConfig } from '@n8n/config';
 /**
  * Resolve the Instance AI output-redaction policy from env config.
  * Returns `false` when disabled so the redactor passes events through untouched.
+ *
+ * Durable-log flag: raw-at-rest storage policy (team decision, 2026-07-06) —
+ * the stream-side redactor moves off the publish path so the log captures raw
+ * values, consistent with workflow execution data. Redaction applies at egress
+ * boundaries instead; the stricter LangSmith telemetry redactor
+ * (trace-payloads.ts) is a separate layer and is unchanged.
  */
 export function resolveOutputRedaction(config: InstanceAiConfig): RedactionOptions | false {
+	if (config.durableLog) return false;
 	if (!config.outputRedactionEnabled) return false;
 
 	const detect = config.outputRedactionPii

packages/cli/src/scaling/pubsub/pubsub.event-map.ts

@@ -164,9 +164,11 @@ export type PubSubCommandMap = {
 		/**
 		 * Producer-assigned stored event. The id comes from the shared per-thread
 		 * sequence, so every main stores and serves identical event ids and the
-		 * frontend's replay cursor is valid against any main.
+		 * frontend's replay cursor is valid against any main. With the durable
+		 * log enabled ids are DB-assigned seqs and ephemeral events (deltas,
+		 * status) carry no id at all: they are live-only.
 		 */
-		storedEvent: { id: number; event: InstanceAiEvent };
+		storedEvent: { id?: number; event: InstanceAiEvent };
 	};
 
 	/**