feat(core): Persist agent session timelines during execution (#35490)

Co-authored-by: Cursor <[email protected]>

docs/generated/postgres-schema/public.agent_execution.md

@@ -30,7 +30,7 @@
 | Name | Type | Definition |
 | ---- | ---- | ---------- |
 | CHK_agent_execution_hitlStatus | CHECK | CHECK ((("hitlStatus")::text = ANY ((ARRAY['suspended'::character varying, 'resumed'::character varying])::text[]))) |
-| CHK_agent_execution_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['success'::character varying, 'error'::character varying])::text[]))) |
+| CHK_agent_execution_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['running'::character varying, 'success'::character varying, 'error'::character varying, 'cancelled'::character varying, 'interrupted'::character varying])::text[]))) |
 | CHK_agent_execution_storedAt | CHECK | CHECK ((("storedAt")::text = ANY ((ARRAY['db'::character varying, 'fs'::character varying, 's3'::character varying, 'az'::character varying])::text[]))) |
 | FK_add2432fb6034cc18b6af299dce | FOREIGN KEY | FOREIGN KEY ("threadId") REFERENCES agent_execution_threads(id) ON DELETE CASCADE |
 | PK_ba438acc8532addc12d1ef17049 | PRIMARY KEY | PRIMARY KEY (id) |
@@ -47,6 +47,7 @@
 | Name | Definition |
 | ---- | ---------- |
 | IDX_63d3c3a68b9cebf05f967f0b1c | CREATE INDEX "IDX_63d3c3a68b9cebf05f967f0b1c" ON public.agent_execution USING btree ("threadId", "createdAt") |
+| IDX_agent_execution_status | CREATE INDEX "IDX_agent_execution_status" ON public.agent_execution USING btree (status) WHERE ((status)::text = 'running'::text) |
 | PK_ba438acc8532addc12d1ef17049 | CREATE UNIQUE INDEX "PK_ba438acc8532addc12d1ef17049" ON public.agent_execution USING btree (id) |
 
 ## Relations

docs/generated/sqlite-schema/agent_execution.md

@@ -6,7 +6,7 @@
 <summary><strong>Table Definition</strong></summary>
 
 ```sql
-CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "storedAt" varchar(2) NOT NULL DEFAULT ('db'), "attachments" text, CONSTRAINT "CHK_agent_execution_storedAt" CHECK (("storedAt" IN ('db', 'fs', 's3', 'az'))), CONSTRAINT "CHK_agent_execution_status" CHECK ((((("status" IN ('success', 'error')))))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK ((((("hitlStatus" IN ('suspended', 'resumed')))))), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)
+CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId" varchar(128) NOT NULL, "status" varchar(16) NOT NULL, "startedAt" datetime(3), "stoppedAt" datetime(3), "duration" integer NOT NULL DEFAULT (0), "userMessage" text, "model" varchar(255), "promptTokens" integer, "completionTokens" integer, "totalTokens" integer, "cost" real, "timeline" text, "error" text, "hitlStatus" varchar(16), "source" varchar(32), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "storedAt" varchar(2) NOT NULL DEFAULT ('db'), "attachments" text, CONSTRAINT "CHK_agent_execution_storedAt" CHECK (("storedAt" IN ('db', 'fs', 's3', 'az'))), CONSTRAINT "CHK_agent_execution_hitlStatus" CHECK ((((("hitlStatus" IN ('suspended', 'resumed')))))), CONSTRAINT "CHK_agent_execution_status" CHECK ("status" IN ('running', 'success', 'error', 'cancelled', 'interrupted')), CONSTRAINT "FK_add2432fb6034cc18b6af299dce" FOREIGN KEY ("threadId") REFERENCES "agent_execution_threads" ("id") ON DELETE CASCADE ON UPDATE NO ACTION)
 ```
 
 </details>
@@ -41,8 +41,8 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
 | Name | Type | Definition |
 | ---- | ---- | ---------- |
 | - | CHECK | CHECK (("storedAt" IN ('db', 'fs', 's3', 'az'))) |
-| - | CHECK | CHECK ((((("status" IN ('success', 'error')))))) |
 | - | CHECK | CHECK ((((("hitlStatus" IN ('suspended', 'resumed')))))) |
+| - | CHECK | CHECK ("status" IN ('running', 'success', 'error', 'cancelled', 'interrupted')) |
 | - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (threadId) REFERENCES agent_execution_threads (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE |
 | id | PRIMARY KEY | PRIMARY KEY (id) |
 | sqlite_autoindex_agent_execution_1 | PRIMARY KEY | PRIMARY KEY (id) |
@@ -52,6 +52,7 @@ CREATE TABLE "agent_execution" ("id" varchar(36) PRIMARY KEY NOT NULL, "threadId
 | Name | Definition |
 | ---- | ---------- |
 | IDX_63d3c3a68b9cebf05f967f0b1c | CREATE INDEX "IDX_63d3c3a68b9cebf05f967f0b1c" ON "agent_execution" ("threadId", "createdAt")  |
+| IDX_agent_execution_status | CREATE INDEX "IDX_agent_execution_status" ON "agent_execution" ("status") WHERE "status" = 'running' |
 | sqlite_autoindex_agent_execution_1 | PRIMARY KEY (id) |
 
 ## Relations

packages/@n8n/api-types/src/agents/types.ts

@@ -241,7 +241,7 @@ export interface AgentPersistedMessageDto {
 	/** Agent-execution turn id when this message was produced from an execution transcript. */
 	executionId?: string;
 	/** Outcome of the execution that produced this message. */
-	executionStatus?: 'success' | 'error';
+	executionStatus?: 'running' | 'success' | 'error' | 'cancelled' | 'interrupted';
 }
 
 export const AGENT_BUILDER_DEFAULT_MODEL = 'claude-sonnet-4-6' as const;

packages/@n8n/api-types/src/push/agents.ts

@@ -0,0 +1,11 @@
+export type AgentExecutionUpdated = {
+	type: 'agentExecutionUpdated';
+	data: {
+		projectId: string;
+		agentId: string;
+		threadId: string;
+		executionId: string;
+	};
+};
+
+export type AgentsPushMessage = AgentExecutionUpdated;

packages/@n8n/api-types/src/push/index.ts

@@ -1,3 +1,4 @@
+import type { AgentsPushMessage } from './agents';
 import type { BuilderCreditsPushMessage } from './builder-credits';
 import type { ChatHubPushMessage } from './chat-hub';
 import type { CollaborationPushMessage } from './collaboration';
@@ -11,6 +12,7 @@ import type { WorkflowPushMessage } from './workflow';
 import type { WorkflowReviewPushMessage } from './workflow-review';
 
 export type PushMessage =
+	| AgentsPushMessage
 	| ExecutionPushMessage
 	| WorkflowPushMessage
 	| HotReloadPushMessage

packages/@n8n/db/src/migrations/common/1785828155091-AddAgentExecutionRuntimeState.ts

@@ -0,0 +1,38 @@
+import type { MigrationContext, ReversibleMigration } from '../migration-types';
+
+const executionTable = 'agent_execution';
+const statusesBefore = ['success', 'error'];
+const statusesAfter = ['running', 'success', 'error', 'cancelled', 'interrupted'];
+
+export class AddAgentExecutionRuntimeState1785828155091 implements ReversibleMigration {
+	async up(ctx: MigrationContext) {
+		const { schemaBuilder } = ctx;
+
+		await schemaBuilder.dropEnumCheck(executionTable, 'status', { recreatesOnSqlite: true });
+		await schemaBuilder.addEnumCheck(executionTable, 'status', statusesAfter, {
+			recreatesOnSqlite: true,
+		});
+		await schemaBuilder.createIndex(
+			executionTable,
+			['status'],
+			false,
+			undefined,
+			`${ctx.escape.columnName('status')} = 'running'`,
+		);
+	}
+
+	async down(ctx: MigrationContext) {
+		const { schemaBuilder } = ctx;
+
+		await schemaBuilder.dropIndex(executionTable, ['status']);
+		await ctx.runQuery(
+			`UPDATE ${ctx.escape.tableName(executionTable)} ` +
+				`SET ${ctx.escape.columnName('status')} = 'error' ` +
+				`WHERE ${ctx.escape.columnName('status')} IN ('running', 'cancelled', 'interrupted')`,
+		);
+		await schemaBuilder.dropEnumCheck(executionTable, 'status', { recreatesOnSqlite: true });
+		await schemaBuilder.addEnumCheck(executionTable, 'status', statusesBefore, {
+			recreatesOnSqlite: true,
+		});
+	}
+}

packages/@n8n/db/src/migrations/postgresdb/index.ts

@@ -237,6 +237,7 @@ import { CrashStaleEnqueuedExecutions1785247194306 } from '../common/17852471943
 import { AddMisfirePolicyToScheduler1785247194307 } from '../common/1785247194307-AddMisfirePolicyToScheduler';
 import { CreateAgentChatAttachmentsTable1785255306000 } from '../common/1785255306000-CreateAgentChatAttachmentsTable';
 import { AddSetupCompletedAtToAgents1785500832626 } from '../common/1785500832626-AddSetupCompletedAtToAgents';
+import { AddAgentExecutionRuntimeState1785828155091 } from '../common/1785828155091-AddAgentExecutionRuntimeState';
 import type { Migration } from '../migration-types';
 
 export const postgresMigrations: Migration[] = [
@@ -479,4 +480,5 @@ export const postgresMigrations: Migration[] = [
 	AddMisfirePolicyToScheduler1785247194307,
 	CreateAgentChatAttachmentsTable1785255306000,
 	AddSetupCompletedAtToAgents1785500832626,
+	AddAgentExecutionRuntimeState1785828155091,
 ];

packages/@n8n/db/src/migrations/sqlite/index.ts

@@ -229,6 +229,7 @@ import { CreateAgentEvalTables1784815940112 } from '../common/1784815940112-Crea
 import { AddAgentFileStorageColumns1785186578138 } from '../common/1785186578138-AddAgentFileStorageColumns';
 import { CrashStaleEnqueuedExecutions1785247194306 } from '../common/1785247194306-CrashStaleEnqueuedExecutions';
 import { CreateAgentChatAttachmentsTable1785255306000 } from '../common/1785255306000-CreateAgentChatAttachmentsTable';
+import { AddAgentExecutionRuntimeState1785828155091 } from '../common/1785828155091-AddAgentExecutionRuntimeState';
 
 const sqliteMigrations: Migration[] = [
 	InitialMigration1588102412422,
@@ -461,6 +462,7 @@ const sqliteMigrations: Migration[] = [
 	AddMisfirePolicyToScheduler1785247194307,
 	CreateAgentChatAttachmentsTable1785255306000,
 	AddSetupCompletedAtToAgents1785500832626,
+	AddAgentExecutionRuntimeState1785828155091,
 ];
 
 export { sqliteMigrations };

packages/@n8n/decorators/src/pubsub/pubsub-metadata.ts

@@ -25,6 +25,7 @@ export type PubSubEventName =
 	| 'response-to-get-worker-status'
 	| 'restart-event-bus'
 	| 'relay-execution-lifecycle-event'
+	| 'relay-agent-execution-update'
 	| 'relay-chat-stream-event'
 	| 'relay-instance-ai-event'
 	| 'relay-instance-ai-task-control'

packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts

@@ -79,8 +79,12 @@ function makeRuntime(chunks: StreamChunk[] = [{ type: 'finish', finishReason: 's
 		agent: {
 			name: 'Runtime Agent',
 			snapshot: { model: { provider: 'anthropic', name: 'claude-sonnet-4-5' } },
-			stream: vi.fn().mockResolvedValue({ stream: makeReadableStream(chunks) }),
-			resume: vi.fn().mockResolvedValue({ stream: makeReadableStream(chunks) }),
+			stream: vi
+				.fn()
+				.mockResolvedValue({ runId: 'runtime-run-1', stream: makeReadableStream(chunks) }),
+			resume: vi
+				.fn()
+				.mockResolvedValue({ runId: 'runtime-run-1', stream: makeReadableStream(chunks) }),
 			structuredOutput: vi.fn(),
 			close: vi.fn(),
 		} as unknown as RuntimeAgent & {
@@ -104,7 +108,8 @@ function makeService() {
 	const agentRunTracingService = mock<AgentRunTracingService>();
 	const externalHooks = mock<ExternalHooks>();
 
-	executionService.recordMessage.mockResolvedValue('execution-1');
+	executionService.startExecutionRecording.mockResolvedValue('execution-1');
+	executionService.finalizeExecution.mockResolvedValue('execution-1');
 	agentRunTracingService.build.mockResolvedValue(undefined);
 
 	const service = new AgentExecutionOrchestratorService(
@@ -172,6 +177,54 @@ describe('AgentExecutionOrchestratorService', () => {
 		vi.clearAllMocks();
 	});
 
+	it('starts durable recording before consuming timeline events and finalizes the same row', async () => {
+		const { service, executionService } = makeService();
+		executionService.startExecutionRecording.mockResolvedValue('execution-running');
+		executionService.finalizeExecution.mockResolvedValue('execution-running');
+		const runtime = makeRuntime([
+			{ type: 'text-delta', id: 'text-1', delta: 'Working' },
+			{ type: 'finish', finishReason: 'stop' },
+		]);
+
+		await collect(
+			service.streamChatResponse({
+				agentInstance: runtime.agent,
+				toolRegistry: runtime.toolRegistry,
+				agentId,
+				userId,
+				message: 'hello',
+				memory: { threadId: 'thread-1', resourceId: 'resource-1' },
+				projectId,
+				telemetry: telemetryContext,
+			}),
+		);
+
+		expect(executionService.startExecutionRecording).toHaveBeenCalledWith(
+			expect.objectContaining({ threadId: 'thread-1', userMessage: 'hello' }),
+			expect.any(Date),
+		);
+		expect(executionService.startExecutionRecording.mock.invocationCallOrder[0]).toBeLessThan(
+			executionService.recordTimelineSnapshot.mock.invocationCallOrder[0],
+		);
+		expect(executionService.recordTimelineSnapshot).toHaveBeenCalledWith(
+			expect.objectContaining({
+				projectId,
+				agentId,
+				threadId: 'thread-1',
+				executionId: 'execution-running',
+			}),
+		);
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-running',
+			expect.objectContaining({
+				record: expect.objectContaining({ assistantResponse: 'Working' }),
+			}),
+		);
+		const startedAt = executionService.startExecutionRecording.mock.calls[0][1];
+		const finalizedRecord = executionService.finalizeExecution.mock.calls[0][1].record;
+		expect(startedAt.getTime()).toBe(finalizedRecord.startTime);
+	});
+
 	it('streams chat responses and records suspended executions', async () => {
 		const { service, executionService } = makeService();
 		const abortController = new AbortController();
@@ -209,7 +262,8 @@ describe('AgentExecutionOrchestratorService', () => {
 				abortSignal: abortController.signal,
 			}),
 		);
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				threadId: 'thread-1',
 				userMessage: 'hello',
@@ -219,7 +273,7 @@ describe('AgentExecutionOrchestratorService', () => {
 		);
 	});
 
-	it('awaits recordMessage and notifies onExecutionRecorded with the returned id', async () => {
+	it('awaits finalization and notifies onExecutionRecorded with the returned id', async () => {
 		const { service, executionService } = makeService();
 		const runtime = makeRuntime([{ type: 'finish', finishReason: 'stop' }]);
 		const onExecutionRecorded = vi.fn();
@@ -238,7 +292,7 @@ describe('AgentExecutionOrchestratorService', () => {
 			}),
 		);
 
-		expect(executionService.recordMessage).toHaveBeenCalled();
+		expect(executionService.finalizeExecution).toHaveBeenCalled();
 		expect(onExecutionRecorded).toHaveBeenCalledWith('execution-1');
 	});
 
@@ -259,7 +313,7 @@ describe('AgentExecutionOrchestratorService', () => {
 			}),
 		);
 
-		expect(executionService.recordMessage).toHaveBeenCalled();
+		expect(executionService.finalizeExecution).toHaveBeenCalled();
 	});
 
 	it('executes in-app chat against the draft runtime', async () => {
@@ -311,7 +365,8 @@ describe('AgentExecutionOrchestratorService', () => {
 			}),
 		);
 		expect(externalHooks.run).not.toHaveBeenCalled();
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				source: undefined,
 				taskId: undefined,
@@ -364,7 +419,8 @@ describe('AgentExecutionOrchestratorService', () => {
 		expect(externalHooks.run.mock.invocationCallOrder[0] ?? 0).toBeLessThan(
 			runtimeCacheService.getRuntime.mock.invocationCallOrder[0] ?? 0,
 		);
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				source: 'slack',
 				telemetry: {
@@ -411,7 +467,8 @@ describe('AgentExecutionOrchestratorService', () => {
 		expect(externalHooks.run.mock.invocationCallOrder[0] ?? 0).toBeLessThan(
 			runtimeCacheService.getRuntime.mock.invocationCallOrder[0] ?? 0,
 		);
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				source: 'task',
 				taskId: 'task-1',
@@ -491,7 +548,8 @@ describe('AgentExecutionOrchestratorService', () => {
 
 		expect(generatedTextIndex).toBeGreaterThan(-1);
 		expect(generatedTextIndex).toBeLessThan(finishIndex);
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				record: expect.objectContaining({
 					assistantResponse: expect.stringContaining('maximum number of iterations'),
@@ -520,7 +578,8 @@ describe('AgentExecutionOrchestratorService', () => {
 			),
 		).rejects.toThrow('reader failed while consuming stream');
 
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				threadId: 'thread-1',
 				agentId,
@@ -560,7 +619,8 @@ describe('AgentExecutionOrchestratorService', () => {
 		abortController.abort();
 		await collect(stream);
 
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				userMessage: 'hello',
 				record: expect.objectContaining({
@@ -654,7 +714,8 @@ describe('AgentExecutionOrchestratorService', () => {
 		);
 		expect(externalHooks.run).not.toHaveBeenCalled();
 		expect(JSON.stringify(runtime.agent.resume.mock.calls[0])).not.toContain('platform-user-1');
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				threadId: 'thread-1',
 				userMessage: null,
@@ -696,7 +757,8 @@ describe('AgentExecutionOrchestratorService', () => {
 		abortController.abort();
 		await collect(stream);
 
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				userMessage: null,
 				hitlStatus: 'resumed',
@@ -1029,7 +1091,8 @@ describe('AgentExecutionOrchestratorService', () => {
 			}),
 		);
 
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({ threadId: 'thread-1', userMessage: null, hitlStatus: 'suspended' }),
 		);
 	});

packages/cli/src/modules/agents/__tests__/agent-execution-update-broadcaster.test.ts

@@ -0,0 +1,87 @@
+import type { Logger } from '@n8n/backend-common';
+import type { ProjectRelationRepository } from '@n8n/db';
+import type { InstanceSettings } from 'n8n-core';
+import { mock } from 'vitest-mock-extended';
+
+import type { Push } from '@/push';
+import type { Publisher } from '@/scaling/pubsub/publisher.service';
+
+import { AgentExecutionUpdateBroadcaster } from '../agent-execution-update-broadcaster';
+
+const update = {
+	projectId: 'project-1',
+	agentId: 'agent-1',
+	threadId: 'thread-1',
+	executionId: 'execution-1',
+};
+
+describe('AgentExecutionUpdateBroadcaster', () => {
+	const logger = mock<Logger>();
+	const projectRelationRepository = mock<ProjectRelationRepository>();
+	const push = mock<Push>();
+	const publisher = mock<Publisher>();
+	const instanceSettings = mock<InstanceSettings>();
+	let broadcaster: AgentExecutionUpdateBroadcaster;
+
+	beforeEach(() => {
+		vi.clearAllMocks();
+		logger.scoped.mockReturnValue(logger);
+		projectRelationRepository.findUserIdsByProjectId.mockResolvedValue(['user-1', 'user-2']);
+		Object.defineProperties(instanceSettings, {
+			isWorker: { value: false, configurable: true },
+			isMultiMain: { value: false, configurable: true },
+		});
+		broadcaster = new AgentExecutionUpdateBroadcaster(
+			logger,
+			projectRelationRepository,
+			push,
+			publisher,
+			instanceSettings,
+		);
+	});
+
+	it('sends project-scoped invalidations locally and relays them from workers', async () => {
+		Object.defineProperty(instanceSettings, 'isWorker', { value: true, configurable: true });
+
+		broadcaster.notify(update);
+
+		await vi.waitFor(() =>
+			expect(publisher.publishCommand).toHaveBeenCalledWith({
+				command: 'relay-agent-execution-update',
+				payload: { data: update, userIds: ['user-1', 'user-2'] },
+			}),
+		);
+		expect(push.sendToUsers).toHaveBeenCalledWith({ type: 'agentExecutionUpdated', data: update }, [
+			'user-1',
+			'user-2',
+		]);
+	});
+
+	it('delivers relayed invalidations locally without publishing them again', () => {
+		broadcaster.handleRelay({ data: update, userIds: ['user-2'] });
+
+		expect(push.sendToUsers).toHaveBeenCalledWith({ type: 'agentExecutionUpdated', data: update }, [
+			'user-2',
+		]);
+		expect(publisher.publishCommand).not.toHaveBeenCalled();
+	});
+
+	it('contains project lookup and relay failures', async () => {
+		projectRelationRepository.findUserIdsByProjectId.mockRejectedValueOnce(
+			new Error('lookup failed'),
+		);
+		broadcaster.notify(update);
+		await vi.waitFor(() => expect(logger.warn).toHaveBeenCalledTimes(1));
+		expect(push.sendToUsers).not.toHaveBeenCalled();
+
+		Object.defineProperty(instanceSettings, 'isMultiMain', { value: true, configurable: true });
+		publisher.publishCommand.mockRejectedValueOnce(new Error('relay failed'));
+		broadcaster.notify(update);
+		await vi.waitFor(() => expect(logger.warn).toHaveBeenCalledTimes(2));
+
+		expect(push.sendToUsers).toHaveBeenCalledWith({ type: 'agentExecutionUpdated', data: update }, [
+			'user-1',
+			'user-2',
+		]);
+	});
+});

packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts

@@ -6,10 +6,11 @@ import type { ErrorReporter, StorageConfig } from 'n8n-core';
 import type { Telemetry } from '@/telemetry';
 
 import type { AgentChatAttachmentService } from '../agent-chat-attachment.service';
-import { AgentExecutionService } from '../agent-execution.service';
+import { AgentExecutionService, type RecordMessageParams } from '../agent-execution.service';
+import type { AgentExecutionUpdateBroadcaster } from '../agent-execution-update-broadcaster';
 import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity';
 import type { AgentExecution } from '../entities/agent-execution.entity';
-import type { MessageRecord } from '../execution-recorder';
+import type { MessageRecord, TimelineEvent } from '../execution-recorder';
 import type { AgentExecutionLogStore } from '../execution-log/agent-execution-log-store';
 import type { N8nMemory } from '../integrations/n8n-memory';
 import type { AgentExecutionThreadRepository } from '../repositories/agent-execution-thread.repository';
@@ -64,11 +65,14 @@ describe('AgentExecutionService', () => {
 	let storageConfig: Mocked<StorageConfig>;
 	let errorReporter: Mocked<ErrorReporter>;
 	let agentChatAttachmentService: Mocked<AgentChatAttachmentService>;
+	let executionUpdateBroadcaster: Mocked<AgentExecutionUpdateBroadcaster>;
 
 	beforeEach(() => {
 		vi.clearAllMocks();
 
 		agentExecutionRepository = mock<AgentExecutionRepository>();
+		agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
+		agentExecutionRepository.updateTimelineIfRunning.mockResolvedValue(true);
 		agentExecutionThreadRepository = mock<AgentExecutionThreadRepository>();
 		n8nMemory = mock<N8nMemory>();
 		memoryBackend = mock<N8nMemoryImplementation>();
@@ -78,6 +82,7 @@ describe('AgentExecutionService', () => {
 		storageConfig = mock<StorageConfig>({ modeTag: 'db' });
 		errorReporter = mock<ErrorReporter>();
 		agentChatAttachmentService = mock<AgentChatAttachmentService>();
+		executionUpdateBroadcaster = mock<AgentExecutionUpdateBroadcaster>();
 
 		service = new AgentExecutionService(
 			mockLogger(),
@@ -89,10 +94,180 @@ describe('AgentExecutionService', () => {
 			agentExecutionLogStore,
 			storageConfig,
 			errorReporter,
+			executionUpdateBroadcaster,
 		);
 	});
 
-	describe('recordMessage', () => {
+	async function recordExecution(params: RecordMessageParams): Promise<string> {
+		const { record, ...startParams } = params;
+		const executionId = await service.startExecutionRecording(
+			startParams,
+			new Date(record.startTime),
+		);
+		return await service.finalizeExecution(executionId, params);
+	}
+
+	describe('startExecutionRecording', () => {
+		it('keeps a running execution alive until it is finalized', async () => {
+			vi.useFakeTimers();
+			try {
+				agentExecutionThreadRepository.findOrCreate.mockResolvedValue({
+					thread: makeThread(),
+					created: true,
+				});
+				agentExecutionRepository.create.mockImplementation((data) => data as AgentExecution);
+				agentExecutionRepository.save.mockResolvedValue({
+					id: 'execution-1',
+				} as AgentExecution);
+				agentExecutionRepository.touchRunning.mockResolvedValue();
+				agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
+
+				const executionId = await service.startExecutionRecording(
+					{
+						threadId: 'thread-1',
+						agentId: 'agent-1',
+						agentName: 'Agent',
+						projectId: 'project-1',
+						userMessage: 'Run',
+					},
+					new Date(),
+				);
+				await vi.advanceTimersByTimeAsync(30_000);
+
+				expect(executionUpdateBroadcaster.notify).toHaveBeenCalledWith({
+					projectId: 'project-1',
+					agentId: 'agent-1',
+					threadId: 'thread-1',
+					executionId,
+				});
+				expect(agentExecutionRepository.create).toHaveBeenCalledWith(
+					expect.objectContaining({ status: 'running' }),
+				);
+				expect(agentExecutionRepository.touchRunning).toHaveBeenCalledWith(executionId);
+
+				await service.finalizeExecution(executionId, {
+					threadId: 'thread-1',
+					agentId: 'agent-1',
+					agentName: 'Agent',
+					projectId: 'project-1',
+					userMessage: 'Run',
+					record: makeMessageRecord(),
+				});
+				await vi.advanceTimersByTimeAsync(30_000);
+
+				expect(agentExecutionRepository.touchRunning).toHaveBeenCalledOnce();
+			} finally {
+				vi.useRealTimers();
+			}
+		});
+	});
+
+	it('serializes timeline snapshot updates', async () => {
+		let releaseFirstWrite!: () => void;
+		agentExecutionRepository.updateTimelineIfRunning
+			.mockImplementationOnce(
+				async () =>
+					await new Promise<boolean>((resolve) => {
+						releaseFirstWrite = () => resolve(true);
+					}),
+			)
+			.mockResolvedValue(true);
+		const first: TimelineEvent[] = [{ type: 'text', content: 'First', timestamp: 1 }];
+		const second: TimelineEvent[] = [{ type: 'text', content: 'Second', timestamp: 1 }];
+
+		service.recordTimelineSnapshot({
+			executionId: 'execution-1',
+			projectId: 'project-1',
+			agentId: 'agent-1',
+			threadId: 'thread-1',
+			timeline: first,
+		});
+		service.recordTimelineSnapshot({
+			executionId: 'execution-1',
+			projectId: 'project-1',
+			agentId: 'agent-1',
+			threadId: 'thread-1',
+			timeline: second,
+		});
+		await vi.waitFor(() =>
+			expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledTimes(1),
+		);
+		releaseFirstWrite();
+		await vi.waitFor(() =>
+			expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledTimes(2),
+		);
+
+		expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenLastCalledWith(
+			'execution-1',
+			second,
+		);
+	});
+
+	it('notifies only after a timeline snapshot retry persists', async () => {
+		vi.useFakeTimers();
+		try {
+			agentExecutionRepository.updateTimelineIfRunning
+				.mockRejectedValueOnce(new Error('temporarily unavailable'))
+				.mockResolvedValueOnce(true);
+
+			service.recordTimelineSnapshot({
+				executionId: 'execution-1',
+				projectId: 'project-1',
+				agentId: 'agent-1',
+				threadId: 'thread-1',
+				timeline: [{ type: 'text', content: 'Working', timestamp: 1 }],
+			});
+			await vi.advanceTimersByTimeAsync(0);
+
+			expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalledTimes(1);
+			expect(executionUpdateBroadcaster.notify).not.toHaveBeenCalled();
+
+			await vi.advanceTimersByTimeAsync(1_000);
+
+			expect(executionUpdateBroadcaster.notify).toHaveBeenCalledWith({
+				projectId: 'project-1',
+				agentId: 'agent-1',
+				threadId: 'thread-1',
+				executionId: 'execution-1',
+			});
+		} finally {
+			vi.useRealTimers();
+		}
+	});
+
+	it('does not notify when a late timeline snapshot is rejected', async () => {
+		agentExecutionRepository.updateTimelineIfRunning.mockResolvedValue(false);
+
+		service.recordTimelineSnapshot({
+			executionId: 'execution-1',
+			projectId: 'project-1',
+			agentId: 'agent-1',
+			threadId: 'thread-1',
+			timeline: [{ type: 'text', content: 'Too late', timestamp: 1 }],
+		});
+		await vi.waitFor(() =>
+			expect(agentExecutionRepository.updateTimelineIfRunning).toHaveBeenCalled(),
+		);
+
+		expect(executionUpdateBroadcaster.notify).not.toHaveBeenCalled();
+	});
+
+	it('does not notify when execution finalization loses the running-state race', async () => {
+		agentExecutionRepository.updateIfRunning.mockResolvedValue(false);
+
+		await service.finalizeExecution('execution-1', {
+			threadId: 'thread-1',
+			agentId: 'agent-1',
+			agentName: 'Agent',
+			projectId: 'project-1',
+			userMessage: 'Run',
+			record: makeMessageRecord(),
+		});
+
+		expect(executionUpdateBroadcaster.notify).not.toHaveBeenCalled();
+	});
+
+	describe('execution lifecycle', () => {
 		it('writes the timeline to blob storage in non-db mode', async () => {
 			storageConfig = mock<StorageConfig>({ modeTag: 'fs' });
 			service = new AgentExecutionService(
@@ -105,6 +280,7 @@ describe('AgentExecutionService', () => {
 				agentExecutionLogStore,
 				storageConfig,
 				errorReporter,
+				executionUpdateBroadcaster,
 			);
 
 			const record = makeMessageRecord({
@@ -129,7 +305,7 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.create.mockImplementation((data) => data as AgentExecution);
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -138,7 +314,8 @@ describe('AgentExecutionService', () => {
 				record,
 			});
 
-			expect(agentExecutionRepository.create).toHaveBeenCalledWith(
+			expect(agentExecutionRepository.updateIfRunning).toHaveBeenCalledWith(
+				'execution-1',
 				expect.objectContaining({ timeline: null, storedAt: 'fs' }),
 			);
 			expect(agentExecutionLogStore.write).toHaveBeenCalledWith(
@@ -146,9 +323,16 @@ describe('AgentExecutionService', () => {
 				{ timeline: record.timeline },
 				'fs',
 			);
+			expect(executionUpdateBroadcaster.notify).toHaveBeenLastCalledWith({
+				projectId: 'project-1',
+				agentId: 'agent-1',
+				threadId: 'thread-1',
+				executionId: 'execution-1',
+			});
+			expect(executionUpdateBroadcaster.notify).toHaveBeenCalledTimes(2);
 		});
 
-		it('keeps the execution row when the blob write fails', async () => {
+		it('stores the finalized execution in the database when the blob write fails', async () => {
 			storageConfig = mock<StorageConfig>({ modeTag: 'fs' });
 			service = new AgentExecutionService(
 				mockLogger(),
@@ -160,6 +344,7 @@ describe('AgentExecutionService', () => {
 				agentExecutionLogStore,
 				storageConfig,
 				errorReporter,
+				executionUpdateBroadcaster,
 			);
 
 			const record = makeMessageRecord({
@@ -185,18 +370,23 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 			agentExecutionLogStore.write.mockRejectedValue(new Error('disk full'));
 
-			await expect(
-				service.recordMessage({
-					threadId: 'thread-1',
-					agentId: 'agent-1',
-					agentName: 'Agent',
-					projectId: 'project-1',
-					userMessage: 'Run',
-					record,
-				}),
-			).resolves.toBe('execution-1');
+			await recordExecution({
+				threadId: 'thread-1',
+				agentId: 'agent-1',
+				agentName: 'Agent',
+				projectId: 'project-1',
+				userMessage: 'Run',
+				record,
+			});
 
-			expect(errorReporter.error).toHaveBeenCalledOnce();
+			expect(agentExecutionRepository.updateIfRunning).toHaveBeenCalledWith(
+				'execution-1',
+				expect.objectContaining({
+					status: 'success',
+					timeline: record.timeline,
+					storedAt: 'db',
+				}),
+			);
 		});
 
 		it('passes thread metadata when creating a subagent execution session', async () => {
@@ -216,7 +406,7 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.create.mockImplementation((entity) => entity as AgentExecution);
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -252,7 +442,7 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.create.mockImplementation((data) => data as AgentExecution);
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -290,7 +480,7 @@ describe('AgentExecutionService', () => {
 				updatedAt: new Date(),
 			});
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -312,7 +502,7 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.create.mockImplementation((data) => data as AgentExecution);
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -333,7 +523,7 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.create.mockImplementation((data) => data as AgentExecution);
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -402,7 +592,7 @@ describe('AgentExecutionService', () => {
 			});
 
 			await expect(
-				service.recordMessage({
+				recordExecution({
 					threadId: 'thread-1',
 					agentId: 'agent-1',
 					agentName: 'Agent',
@@ -444,7 +634,7 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.create.mockImplementation((data) => data as AgentExecution);
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -485,7 +675,7 @@ describe('AgentExecutionService', () => {
 			agentExecutionRepository.create.mockImplementation((data) => data as AgentExecution);
 			agentExecutionRepository.save.mockResolvedValue({ id: 'execution-1' } as AgentExecution);
 
-			await service.recordMessage({
+			await recordExecution({
 				threadId: 'thread-1',
 				agentId: 'agent-1',
 				agentName: 'Agent',
@@ -514,6 +704,94 @@ describe('AgentExecutionService', () => {
 		});
 	});
 
+	describe('finalizeExecution', () => {
+		it('makes the terminal timeline authoritative', async () => {
+			const record = makeMessageRecord({
+				finishReason: 'cancelled',
+				timeline: [{ type: 'text', content: 'Done', timestamp: 1, endTime: 2 }],
+			});
+			agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
+
+			await service.finalizeExecution('execution-1', {
+				threadId: 'thread-1',
+				agentId: 'agent-1',
+				agentName: 'Agent',
+				projectId: 'project-1',
+				userMessage: 'Run',
+				record,
+				telemetry: {
+					runType: 'test',
+					configuration: {
+						model: null,
+						channels: [],
+						tool_types: [],
+						tool_count: 0,
+						num_skills: 0,
+						memory_type: 'none',
+					},
+				},
+			});
+
+			expect(agentExecutionRepository.updateIfRunning).toHaveBeenCalledWith(
+				'execution-1',
+				expect.objectContaining({
+					status: 'cancelled',
+					timeline: record.timeline,
+					storedAt: 'db',
+				}),
+			);
+			expect(telemetry.trackAgentTurnFinished).toHaveBeenCalledWith(
+				expect.objectContaining({ turn_status: 'failed' }),
+			);
+		});
+
+		it('preserves an interrupted execution inline without overwriting blob storage', async () => {
+			storageConfig = mock<StorageConfig>({ modeTag: 'fs' });
+			service = new AgentExecutionService(
+				mockLogger(),
+				agentExecutionRepository,
+				agentExecutionThreadRepository,
+				n8nMemory,
+				telemetry,
+				agentChatAttachmentService,
+				agentExecutionLogStore,
+				storageConfig,
+				errorReporter,
+				executionUpdateBroadcaster,
+			);
+			const partial = [{ type: 'text', content: 'Partial', timestamp: 1, endTime: 2 }] as const;
+			agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
+			agentExecutionThreadRepository.findOneBy.mockResolvedValue(makeThread());
+
+			await service.finalizeInterruptedExecution({
+				id: 'execution-1',
+				threadId: 'thread-1',
+				startedAt: new Date(Date.now() - 100),
+				timeline: [...partial],
+				thread: makeThread(),
+			} as AgentExecution);
+
+			await vi.waitFor(() =>
+				expect(executionUpdateBroadcaster.notify).toHaveBeenCalledWith({
+					projectId: 'project-1',
+					agentId: 'agent-1',
+					threadId: 'thread-1',
+					executionId: 'execution-1',
+				}),
+			);
+			expect(agentExecutionRepository.updateIfRunning).toHaveBeenCalledWith(
+				'execution-1',
+				expect.objectContaining({
+					status: 'interrupted',
+					timeline: partial,
+					storedAt: 'db',
+					error: expect.stringContaining('interrupted'),
+				}),
+			);
+			expect(agentExecutionLogStore.write).not.toHaveBeenCalled();
+		});
+	});
+
 	describe('getThreadDetail', () => {
 		it('returns thread executions after ownership validation', async () => {
 			const thread = makeThread();
@@ -526,6 +804,23 @@ describe('AgentExecutionService', () => {
 			expect(result).toEqual({ thread, executions });
 		});
 
+		it('returns inline progress for running executions', async () => {
+			const thread = makeThread();
+			const partial = [{ type: 'text', content: 'Working', timestamp: 1, endTime: 2 }] as const;
+			const execution = {
+				id: 'execution-1',
+				status: 'running',
+				storedAt: 'db',
+				timeline: [...partial],
+			} as AgentExecution;
+			agentExecutionThreadRepository.findOneBy.mockResolvedValue(thread);
+			agentExecutionRepository.findByThreadIdOrdered.mockResolvedValue([execution]);
+
+			const result = await service.getThreadDetail('thread-1', 'project-1', 'agent-1');
+
+			expect(result?.executions[0]?.timeline).toEqual(partial);
+		});
+
 		it('hydrates blob-stored timelines from the log store', async () => {
 			const dbEvent = {
 				type: 'tool-call' as const,

packages/cli/src/modules/agents/__tests__/agent-interrupted-execution-sweeper.test.ts

@@ -0,0 +1,55 @@
+import { mockLogger } from '@n8n/backend-test-utils';
+import { mock } from 'vitest-mock-extended';
+
+import { AgentInterruptedExecutionSweeper } from '../agent-interrupted-execution-sweeper';
+import type { AgentExecutionService } from '../agent-execution.service';
+import type { AgentExecution } from '../entities/agent-execution.entity';
+import type { AgentExecutionRepository } from '../repositories/agent-execution.repository';
+
+describe('AgentInterruptedExecutionSweeper', () => {
+	it('terminalizes an abandoned running execution', async () => {
+		const repository = mock<AgentExecutionRepository>();
+		const executionService = mock<AgentExecutionService>();
+		const execution = {
+			id: 'execution-1',
+			threadId: 'thread-1',
+			status: 'running',
+			startedAt: new Date(0),
+			updatedAt: new Date(0),
+		} as AgentExecution;
+		repository.findRunning.mockResolvedValue([execution]);
+		executionService.finalizeInterruptedExecution.mockResolvedValue(true);
+		const sweeper = new AgentInterruptedExecutionSweeper(
+			mockLogger(),
+			repository,
+			executionService,
+		);
+
+		await sweeper.sweep();
+
+		expect(executionService.finalizeInterruptedExecution).toHaveBeenCalledWith(execution);
+	});
+
+	it('leaves a recently active execution running in another process', async () => {
+		const repository = mock<AgentExecutionRepository>();
+		const executionService = mock<AgentExecutionService>();
+		repository.findRunning.mockResolvedValue([
+			{
+				id: 'execution-1',
+				threadId: 'thread-1',
+				status: 'running',
+				startedAt: new Date(Date.now() - AgentInterruptedExecutionSweeper.LIVENESS_GRACE_MS * 2),
+				updatedAt: new Date(),
+			} as AgentExecution,
+		]);
+		const sweeper = new AgentInterruptedExecutionSweeper(
+			mockLogger(),
+			repository,
+			executionService,
+		);
+
+		await sweeper.sweep();
+
+		expect(executionService.finalizeInterruptedExecution).not.toHaveBeenCalled();
+	});
+});

packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts

@@ -219,7 +219,6 @@ describe('AgentRuntimeReconstructionService integration tools', () => {
 		n8nMemory.getImplementation.mockReturnValue(memoryBackend);
 		n8nCheckpointStorage = mock<N8NCheckpointStorage>();
 		agentExecutionService = mock<AgentExecutionService>();
-		agentExecutionService.recordMessage.mockResolvedValue('exec-id');
 		chatIntegrationService = mock<ChatIntegrationService>();
 		agentKnowledgeService = mock<AgentKnowledgeService>();
 		publisher = mock<Publisher>();

packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts

@@ -76,8 +76,12 @@ function makeRuntime(chunks: StreamChunk[] = [{ type: 'finish', finishReason: 's
 	return {
 		agent: {
 			name: 'Runtime Agent',
-			stream: vi.fn().mockResolvedValue({ stream: makeReadableStream(chunks) }),
-			resume: vi.fn().mockResolvedValue({ stream: makeReadableStream(chunks) }),
+			stream: vi
+				.fn()
+				.mockResolvedValue({ runId: 'runtime-run-1', stream: makeReadableStream(chunks) }),
+			resume: vi
+				.fn()
+				.mockResolvedValue({ runId: 'runtime-run-1', stream: makeReadableStream(chunks) }),
 			structuredOutput: vi.fn(),
 			close: vi.fn(),
 		} as unknown as RuntimeAgent & {
@@ -108,7 +112,8 @@ function makeService() {
 	const agentRunTracingService = mock<AgentRunTracingService>();
 	const executionLevelTracer = mock<ExecutionLevelTracer>();
 
-	executionService.recordMessage.mockResolvedValue('execution-1');
+	executionService.startExecutionRecording.mockResolvedValue('execution-1');
+	executionService.finalizeExecution.mockResolvedValue('execution-1');
 	agentRunTracingService.build.mockResolvedValue(undefined);
 	executionLevelTracer.getActiveContext.mockReturnValue(undefined);
 
@@ -156,6 +161,8 @@ describe('AgentWorkflowExecutionService', () => {
 
 		agentRepository.findByIdAndProjectId.mockResolvedValue(makeAgent());
 		reconstructionService.reconstructFromAgentEntity.mockResolvedValue(runtime);
+		executionService.startExecutionRecording.mockResolvedValue('agent-execution-1');
+		executionService.finalizeExecution.mockResolvedValue('agent-execution-1');
 
 		const result = await service.executeForWorkflow(
 			agentId,
@@ -193,7 +200,16 @@ describe('AgentWorkflowExecutionService', () => {
 			run_type: 'production',
 			message_count: 1,
 		});
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.recordTimelineSnapshot).toHaveBeenCalledWith(
+			expect.objectContaining({
+				projectId,
+				agentId,
+				threadId: 'thread-1',
+				executionId: 'agent-execution-1',
+			}),
+		);
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'agent-execution-1',
 			expect.objectContaining({
 				telemetry: expect.objectContaining({
 					runType: 'production',
@@ -203,6 +219,9 @@ describe('AgentWorkflowExecutionService', () => {
 				}),
 			}),
 		);
+		const startedAt = executionService.startExecutionRecording.mock.calls[0][1];
+		const finalizedRecord = executionService.finalizeExecution.mock.calls[0][1].record;
+		expect(startedAt.getTime()).toBe(finalizedRecord.startTime);
 		expect(agentRunTracingService.build).toHaveBeenCalledWith(
 			expect.objectContaining({
 				agentId,
@@ -215,6 +234,26 @@ describe('AgentWorkflowExecutionService', () => {
 		);
 	});
 
+	it('records workflow stream setup failures', async () => {
+		const { service, agentRepository, reconstructionService, executionService } = makeService();
+		const runtime = makeRuntime();
+		runtime.agent.stream.mockRejectedValue(new Error('stream setup failed'));
+		agentRepository.findByIdAndProjectId.mockResolvedValue(makeAgent());
+		reconstructionService.reconstructFromAgentEntity.mockResolvedValue(runtime);
+		executionService.startExecutionRecording.mockResolvedValue('fallback-execution-1');
+
+		await expect(
+			service.executeForWorkflow(agentId, 'hello', 'execution-1', 'thread-1', projectId),
+		).rejects.toThrow('stream setup failed');
+
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'fallback-execution-1',
+			expect.objectContaining({
+				record: expect.objectContaining({ error: 'stream setup failed', finishReason: 'error' }),
+			}),
+		);
+	});
+
 	it('records a null input for a tool-result with no matching tool-call', async () => {
 		const { service, agentRepository, reconstructionService } = makeService();
 		const runtime = makeRuntime([
@@ -596,7 +635,7 @@ describe('AgentWorkflowExecutionService', () => {
 			expect(result.response).toBe('answer');
 			// Inline runs have no persisted session.
 			expect(result.session).toBeNull();
-			expect(executionService.recordMessage).not.toHaveBeenCalled();
+			expect(executionService.startExecutionRecording).not.toHaveBeenCalled();
 
 			// Thread-scoped persistence: stable across executions, so a reused
 			// session id continues the same conversation.
@@ -970,7 +1009,8 @@ describe('AgentWorkflowExecutionService', () => {
 		await expect(execution).rejects.toThrow(OperationalError);
 		await expect(execution).rejects.toThrow("Couldn't get structured output matching the schema");
 		await expect(execution).rejects.not.toThrow('Check the stream');
-		expect(executionService.recordMessage).toHaveBeenCalledWith(
+		expect(executionService.finalizeExecution).toHaveBeenCalledWith(
+			'execution-1',
 			expect.objectContaining({
 				record: expect.objectContaining({
 					error: expect.stringContaining("Couldn't get structured output matching the schema"),

packages/cli/src/modules/agents/__tests__/execution-recorder.test.ts

@@ -1,6 +1,6 @@
 import type { BuiltTool, StreamChunk } from '@n8n/agents';
 
-import { ExecutionRecorder } from '../execution-recorder';
+import { ExecutionRecorder, type TimelineEvent } from '../execution-recorder';
 import { buildToolRegistry } from '../tool-registry';
 
 function makeToolCallChunk(toolName: string, input: unknown, toolCallId = 'tc1'): StreamChunk {
@@ -976,3 +976,87 @@ describe('ExecutionRecorder — subagent-chunk', () => {
 		expect(recorder.getMessageRecord().timeline).toEqual([]);
 	});
 });
+
+describe('ExecutionRecorder — durable timeline events', () => {
+	it('emits replacing live snapshots without splitting the final response', () => {
+		vi.useFakeTimers();
+		try {
+			const snapshots: TimelineEvent[][] = [];
+			const recorder = new ExecutionRecorder(undefined, (timeline) => snapshots.push(timeline));
+			recorder.record({ type: 'text-delta', id: 'text-1', delta: 'First' });
+			vi.advanceTimersByTime(999);
+			expect(snapshots).toEqual([]);
+			vi.advanceTimersByTime(1);
+			expect(snapshots).toEqual([[expect.objectContaining({ type: 'text', content: 'First' })]]);
+
+			recorder.record({ type: 'text-delta', id: 'text-1', delta: ' second' });
+			vi.advanceTimersByTime(1_000);
+
+			expect(snapshots).toEqual([
+				[expect.objectContaining({ type: 'text', content: 'First' })],
+				[expect.objectContaining({ type: 'text', content: 'First second' })],
+			]);
+			const record = recorder.getMessageRecord();
+			expect(record.assistantResponse).toBe('First second');
+			expect(record.timeline).toEqual([
+				expect.objectContaining({ type: 'text', content: 'First second' }),
+			]);
+		} finally {
+			vi.useRealTimers();
+		}
+	});
+
+	it('emits a tool snapshot when execution starts and updates it on completion', () => {
+		const snapshots: TimelineEvent[][] = [];
+		const recorder = new ExecutionRecorder(undefined, (timeline) => snapshots.push(timeline));
+
+		recorder.record({ type: 'text-delta', id: 'text-1', delta: 'Checking' });
+		expect(snapshots).toEqual([]);
+
+		recorder.record({
+			type: 'tool-call',
+			toolCallId: 'tool-1',
+			toolName: 'lookup',
+			input: { id: 1 },
+		} as StreamChunk);
+		expect(snapshots).toEqual([[expect.objectContaining({ type: 'text', content: 'Checking' })]]);
+
+		recorder.record({
+			type: 'tool-execution-start',
+			toolCallId: 'tool-1',
+			toolName: 'lookup',
+			startTime: 10,
+		});
+		expect(snapshots[1]?.[1]).toMatchObject({
+			type: 'tool-call',
+			startTime: 10,
+			endTime: 0,
+		});
+
+		recorder.record({
+			type: 'tool-execution-end',
+			toolCallId: 'tool-1',
+			toolName: 'lookup',
+			isError: false,
+			endTime: 20,
+		});
+		recorder.record({
+			type: 'tool-result',
+			toolCallId: 'tool-1',
+			toolName: 'lookup',
+			output: { name: 'Ada' },
+		});
+
+		expect(snapshots).toHaveLength(4);
+		expect(snapshots[2]?.[1]).toMatchObject({
+			type: 'tool-call',
+			endTime: 20,
+			output: undefined,
+		});
+		expect(snapshots[3]?.[1]).toMatchObject({
+			type: 'tool-call',
+			endTime: 20,
+			output: { name: 'Ada' },
+		});
+	});
+});

packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts

@@ -17,7 +17,11 @@ import { ExternalHooks } from '@/external-hooks';
 import type { AgentRunTelemetryType, IAgentConfigurationTelemetryProperties } from '@/interfaces';
 import { Telemetry } from '@/telemetry';
 
-import { AgentExecutionService, type RecordMessageParams } from './agent-execution.service';
+import {
+	AgentExecutionService,
+	type RecordMessageParams,
+	type StartExecutionParams,
+} from './agent-execution.service';
 import { AgentRunTracingService, modelIdFromSnapshot } from './agent-run-tracing.service';
 import { AgentRuntimeCacheService } from './agent-runtime-cache.service';
 import { ExecutionRecorder, type MessageRecord } from './execution-recorder';
@@ -344,7 +348,13 @@ export class AgentExecutionOrchestratorService {
 		});
 
 		const { agent: agentInstance, toolRegistry } = runtime;
-		const recorder = new ExecutionRecorder(toolRegistry);
+		let executionId: string | undefined;
+		const recorder = this.createRecorder(toolRegistry, () => executionId, {
+			projectId,
+			agentId,
+			threadId,
+		});
+		const startedAt = recorder.startedAt;
 		const runType: AgentRunTelemetryType = usePublishedVersion ? 'production' : 'test';
 
 		try {
@@ -376,7 +386,22 @@ export class AgentExecutionOrchestratorService {
 				...(tracing ? { telemetry: tracing } : {}),
 				...(abortSignal ? { abortSignal } : {}),
 			});
-
+			const startParams: StartExecutionParams = {
+				threadId,
+				agentId,
+				agentName: agentInstance.name,
+				projectId,
+				userMessage: null,
+				telemetry: {
+					runType,
+					configuration: runtime.telemetryConfiguration,
+				},
+			};
+			executionId = await this.tryStartExecution(
+				startParams,
+				startedAt,
+				'Failed to start resumed agent execution recording',
+			);
 			for await (const value of streamAgentChunks(resultStream.stream)) {
 				recorder.record(value);
 				yield value;
@@ -391,6 +416,7 @@ export class AgentExecutionOrchestratorService {
 			// pre-suspension execution already has it.
 			const messageRecord = normalizeAbortedMessageRecord(recorder.getMessageRecord(), abortSignal);
 			await this.persistRecordedExecution({
+				executionId,
 				onExecutionRecorded,
 				failureMessage: 'Failed to record resumed agent execution',
 				params: {
@@ -589,7 +615,13 @@ export class AgentExecutionOrchestratorService {
 		} = config;
 		const { threadId, resourceId } = memory;
 
-		const recorder = new ExecutionRecorder(toolRegistry);
+		let executionId: string | undefined;
+		const recorder = this.createRecorder(toolRegistry, () => executionId, {
+			projectId,
+			agentId,
+			threadId,
+		});
+		const startedAt = recorder.startedAt;
 
 		try {
 			const tracing = await this.agentRunTracingService.build({
@@ -612,7 +644,23 @@ export class AgentExecutionOrchestratorService {
 				...(tracing ? { telemetry: tracing } : {}),
 				...(abortSignal ? { abortSignal } : {}),
 			});
-
+			const startParams: StartExecutionParams = {
+				threadId,
+				agentId,
+				agentName: agentInstance.name,
+				projectId,
+				userMessage: message,
+				attachments,
+				source,
+				taskId,
+				taskVersionId,
+				telemetry,
+			};
+			executionId = await this.tryStartExecution(
+				startParams,
+				startedAt,
+				'Failed to start agent execution recording',
+			);
 			for await (const value of streamAgentChunks(resultStream.stream)) {
 				recorder.record(value);
 				if (value.type === 'tool-call-suspended') {
@@ -639,6 +687,7 @@ export class AgentExecutionOrchestratorService {
 			// response text and tool calls are valuable.
 			const messageRecord = normalizeAbortedMessageRecord(recorder.getMessageRecord(), abortSignal);
 			await this.persistRecordedExecution({
+				executionId,
 				onExecutionRecorded,
 				failureMessage: 'Failed to record agent execution',
 				params: {
@@ -659,35 +708,57 @@ export class AgentExecutionOrchestratorService {
 		}
 	}
 
-	/**
-	 * Persist a streamed turn. Awaits when `onExecutionRecorded` is set so SSE
-	 * `done` can carry executionId; otherwise fire-and-forget.
-	 */
+	private createRecorder(
+		toolRegistry: ToolRegistry,
+		getExecutionId: () => string | undefined,
+		context: Pick<StartExecutionParams, 'projectId' | 'agentId' | 'threadId'>,
+	): ExecutionRecorder {
+		return new ExecutionRecorder(toolRegistry, (timeline) => {
+			const executionId = getExecutionId();
+			if (executionId) {
+				this.agentExecutionService.recordTimelineSnapshot({
+					...context,
+					executionId,
+					timeline,
+				});
+			}
+		});
+	}
+
+	private async tryStartExecution(
+		params: StartExecutionParams,
+		startedAt: Date,
+		failureMessage: string,
+	): Promise<string | undefined> {
+		try {
+			return await this.agentExecutionService.startExecutionRecording(params, startedAt);
+		} catch (error) {
+			this.logger.warn(failureMessage, {
+				agentId: params.agentId,
+				threadId: params.threadId,
+				error: error instanceof Error ? error.message : String(error),
+			});
+			return undefined;
+		}
+	}
+
 	private async persistRecordedExecution(args: {
+		executionId?: string;
 		onExecutionRecorded?: (executionId: string) => void;
 		params: RecordMessageParams;
 		failureMessage: string;
 	}): Promise<void> {
-		const { onExecutionRecorded, params, failureMessage } = args;
-		const persist = async () => {
-			const executionId = await this.agentExecutionService.recordMessage(params);
-			onExecutionRecorded?.(executionId);
-		};
-		const logFailure = (error: unknown) => {
+		const { executionId, onExecutionRecorded, params, failureMessage } = args;
+		if (!executionId) return;
+		try {
+			const recordedId = await this.agentExecutionService.finalizeExecution(executionId, params);
+			onExecutionRecorded?.(recordedId);
+		} catch (error) {
 			this.logger.warn(failureMessage, {
 				agentId: params.agentId,
 				threadId: params.threadId,
 				error: error instanceof Error ? error.message : String(error),
 			});
-		};
-		if (onExecutionRecorded) {
-			try {
-				await persist();
-			} catch (error) {
-				logFailure(error);
-			}
-		} else {
-			void persist().catch(logFailure);
 		}
 	}
 }

packages/cli/src/modules/agents/agent-execution-update-broadcaster.ts

@@ -0,0 +1,54 @@
+import type { PushPayload } from '@n8n/api-types';
+import { Logger } from '@n8n/backend-common';
+import { ProjectRelationRepository } from '@n8n/db';
+import { OnPubSubEvent } from '@n8n/decorators';
+import { Service } from '@n8n/di';
+import { InstanceSettings } from 'n8n-core';
+
+import { Push } from '@/push';
+import type { PubSubCommandMap } from '@/scaling/pubsub/pubsub.event-map';
+import { Publisher } from '@/scaling/pubsub/publisher.service';
+
+type AgentExecutionUpdate = PushPayload<'agentExecutionUpdated'>;
+
+@Service()
+export class AgentExecutionUpdateBroadcaster {
+	constructor(
+		private readonly logger: Logger,
+		private readonly projectRelationRepository: ProjectRelationRepository,
+		private readonly push: Push,
+		private readonly publisher: Publisher,
+		private readonly instanceSettings: InstanceSettings,
+	) {
+		this.logger = this.logger.scoped('agents');
+	}
+
+	notify(data: AgentExecutionUpdate): void {
+		void this.broadcast(data).catch((error: unknown) => {
+			this.logger.warn('Failed to broadcast agent execution update', {
+				executionId: data.executionId,
+				threadId: data.threadId,
+				error: error instanceof Error ? error.message : String(error),
+			});
+		});
+	}
+
+	private async broadcast(data: AgentExecutionUpdate): Promise<void> {
+		const userIds = await this.projectRelationRepository.findUserIdsByProjectId(data.projectId);
+		if (userIds.length === 0) return;
+
+		this.push.sendToUsers({ type: 'agentExecutionUpdated', data }, userIds);
+
+		if (this.instanceSettings.isWorker || this.instanceSettings.isMultiMain) {
+			await this.publisher.publishCommand({
+				command: 'relay-agent-execution-update',
+				payload: { data, userIds },
+			});
+		}
+	}
+
+	@OnPubSubEvent('relay-agent-execution-update', { instanceType: 'main' })
+	handleRelay({ data, userIds }: PubSubCommandMap['relay-agent-execution-update']): void {
+		this.push.sendToUsers({ type: 'agentExecutionUpdated', data }, userIds);
+	}
+}

packages/cli/src/modules/agents/agent-execution.service.ts

@@ -12,14 +12,18 @@ import {
 	AgentChatAttachmentService,
 	type StoredAttachmentRef,
 } from './agent-chat-attachment.service';
+import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadcaster';
 import { AgentExecutionThread } from './entities/agent-execution-thread.entity';
 import { AgentExecution } from './entities/agent-execution.entity';
-import type { MessageRecord } from './execution-recorder';
+import type { MessageRecord, TimelineEvent } from './execution-recorder';
 import { AgentExecutionLogStore } from './execution-log/agent-execution-log-store';
 import { N8nMemory } from './integrations/n8n-memory';
 import { AgentExecutionThreadRepository } from './repositories/agent-execution-thread.repository';
 import type { AgentExecutionThreadMetadata } from './repositories/agent-execution-thread.repository';
-import { AgentExecutionRepository } from './repositories/agent-execution.repository';
+import {
+	AgentExecutionRepository,
+	type RunningAgentExecution,
+} from './repositories/agent-execution.repository';
 
 export interface RecordMessageParams {
 	threadId: string;
@@ -47,6 +51,16 @@ export interface RecordMessageParams {
 	};
 }
 
+export type StartExecutionParams = Omit<RecordMessageParams, 'record' | 'hitlStatus'>;
+
+interface TimelineSnapshotParams {
+	executionId: string;
+	projectId: string;
+	agentId: string;
+	threadId: string;
+	timeline: TimelineEvent[];
+}
+
 export interface ThreadDetail {
 	thread: AgentExecutionThread;
 	executions: AgentExecution[];
@@ -58,11 +72,26 @@ export interface ThreadListItem extends Omit<AgentExecutionThread, 'generateId'
 	source: string | null;
 }
 
+const TIMELINE_SNAPSHOT_RETRY_DELAY_MS = 1_000;
+
 @Service()
 export class AgentExecutionService {
 	/** Batch size mirrors ExecutionPersistence.bulkDeletionBatchSize. */
 	private static readonly logDeletionBatchSize = 500;
 
+	private static readonly heartbeatIntervalMs = 30_000;
+
+	private readonly heartbeatTimers = new Map<string, NodeJS.Timeout>();
+
+	private readonly pendingTimelineSnapshots = new Map<
+		string,
+		Omit<TimelineSnapshotParams, 'executionId'>
+	>();
+
+	private readonly timelineSnapshotWrites = new Map<string, Promise<void>>();
+
+	private readonly executionsNeedingTitleSync = new Set<string>();
+
 	constructor(
 		private readonly logger: Logger,
 		private readonly agentExecutionRepository: AgentExecutionRepository,
@@ -73,70 +102,72 @@ export class AgentExecutionService {
 		private readonly agentExecutionLogStore: AgentExecutionLogStore,
 		private readonly storageConfig: StorageConfig,
 		private readonly errorReporter: ErrorReporter,
+		private readonly executionUpdateBroadcaster: AgentExecutionUpdateBroadcaster,
 	) {}
 
-	/**
-	 * Record a single agent run within a thread.
-	 * Creates or updates the thread, then inserts one row into agent_execution.
-	 */
-	async recordMessage(params: RecordMessageParams): Promise<string> {
-		const {
-			threadId,
-			agentId,
-			agentName,
-			projectId,
-			record,
-			source,
-			hitlStatus,
-			threadMetadata,
-			taskId,
-			taskVersionId,
-		} = params;
-
-		// Ensure the thread exists and bump its updatedAt
-		const { thread, created } = await this.agentExecutionThreadRepository.findOrCreate(
-			threadId,
-			agentId,
-			agentName,
-			projectId,
-			threadMetadata,
-			taskId,
-			taskVersionId,
-		);
-		if (!created) {
-			await this.agentExecutionThreadRepository.bumpUpdatedAt(threadId);
-			// Sync title from the SDK memory thread if we don't have one yet
-			if (!thread.title) {
-				await this.syncTitleFromMemory(threadId, agentId);
-			}
-		}
-
-		// Replace platform mentions (e.g. Slack's <@U0ANB4K6611> or plain @U0ANB4K6611)
-		const userMessage =
-			params.userMessage === null
-				? null
-				: (() => {
-						const cleanedMessage = params.userMessage
-							.replace(/<@[A-Z0-9]+>/gi, `@${agentName}`)
-							.replace(/@[A-Z0-9]{8,}/gi, `@${agentName}`)
-							.trim();
-						return cleanedMessage.length > 0 ? cleanedMessage : null;
-					})();
-
-		const status: AgentExecution['status'] = record.error ? 'error' : 'success';
-		const startedAt = new Date(record.startTime);
-		const stoppedAt = new Date(record.startTime + record.duration);
-		const storedAt: AgentExecution['storedAt'] =
-			record.timeline.length > 0 ? this.storageConfig.modeTag : 'db';
-
+	async startExecutionRecording(params: StartExecutionParams, startedAt: Date): Promise<string> {
+		const { userMessage, created } = await this.prepareThread(params);
 		const inserted = await this.agentExecutionRepository.save(
 			this.agentExecutionRepository.create({
-				threadId,
-				status,
+				threadId: params.threadId,
+				status: 'running',
 				startedAt,
-				stoppedAt,
-				duration: record.duration,
+				stoppedAt: null,
+				duration: 0,
 				userMessage,
+				model: null,
+				promptTokens: null,
+				completionTokens: null,
+				totalTokens: null,
+				cost: null,
+				timeline: null,
+				storedAt: 'db',
+				error: null,
+				hitlStatus: null,
+				source: params.source ?? null,
+				attachments: params.attachments?.length ? params.attachments : null,
+			}),
+		);
+		if (created) this.executionsNeedingTitleSync.add(inserted.id);
+		this.startHeartbeat(inserted.id);
+		this.executionUpdateBroadcaster.notify({
+			projectId: params.projectId,
+			agentId: params.agentId,
+			threadId: params.threadId,
+			executionId: inserted.id,
+		});
+		return inserted.id;
+	}
+
+	recordTimelineSnapshot({ executionId, ...snapshot }: TimelineSnapshotParams): void {
+		this.pendingTimelineSnapshots.set(executionId, snapshot);
+		this.ensureTimelineSnapshotWrite(executionId);
+	}
+
+	async finalizeExecution(executionId: string, params: RecordMessageParams): Promise<string> {
+		const { record, hitlStatus } = params;
+		const status = executionStatus(record);
+		let storedAt: AgentExecution['storedAt'] =
+			record.timeline.length > 0 ? this.storageConfig.modeTag : 'db';
+
+		try {
+			if (storedAt !== 'db') {
+				try {
+					await this.agentExecutionLogStore.write(
+						{ agentId: params.agentId, threadId: params.threadId, executionId },
+						{ timeline: record.timeline },
+						storedAt,
+					);
+				} catch (error) {
+					this.errorReporter.error(error);
+					storedAt = 'db';
+				}
+			}
+
+			const finalized = await this.agentExecutionRepository.updateIfRunning(executionId, {
+				status,
+				stoppedAt: new Date(record.startTime + record.duration),
+				duration: record.duration,
 				model: record.model,
 				promptTokens: record.usage?.promptTokens ?? null,
 				completionTokens: record.usage?.completionTokens ?? null,
@@ -146,31 +177,163 @@ export class AgentExecutionService {
 				storedAt,
 				error: record.error,
 				hitlStatus: hitlStatus ?? null,
-				source: source ?? null,
-				attachments: params.attachments?.length ? params.attachments : null,
-			}),
-		);
+			});
+			if (!finalized) return executionId;
 
-		if (storedAt !== 'db') {
+			this.executionUpdateBroadcaster.notify({
+				projectId: params.projectId,
+				agentId: params.agentId,
+				threadId: params.threadId,
+				executionId,
+			});
+			await this.completeRecordedExecution(params, executionId, status);
+			return executionId;
+		} catch (error) {
+			this.errorReporter.error(error);
+			throw error;
+		} finally {
+			this.stopHeartbeat(executionId);
+			this.executionsNeedingTitleSync.delete(executionId);
+		}
+	}
+
+	async finalizeInterruptedExecution(execution: RunningAgentExecution): Promise<boolean> {
+		const timeline = execution.timeline ?? [];
+		const stoppedAt = new Date();
+		const duration = execution.startedAt
+			? Math.max(0, stoppedAt.getTime() - execution.startedAt.getTime())
+			: 0;
+		const finalized = await this.agentExecutionRepository.updateIfRunning(execution.id, {
+			status: 'interrupted',
+			stoppedAt,
+			duration,
+			timeline: timeline.length > 0 ? timeline : null,
+			storedAt: 'db',
+			error: 'Agent execution was interrupted by a process restart.',
+		});
+		if (finalized) void this.notifyInterruptedExecution(execution);
+		return finalized;
+	}
+
+	private async notifyInterruptedExecution(execution: RunningAgentExecution): Promise<void> {
+		try {
+			const thread = await this.agentExecutionThreadRepository.findOneBy({
+				id: execution.threadId,
+			});
+			if (!thread) return;
+			this.executionUpdateBroadcaster.notify({
+				projectId: thread.projectId,
+				agentId: thread.agentId,
+				threadId: execution.threadId,
+				executionId: execution.id,
+			});
+		} catch (error) {
+			this.logger.warn('Failed to resolve an interrupted agent execution update', {
+				executionId: execution.id,
+				threadId: execution.threadId,
+				error: error instanceof Error ? error.message : String(error),
+			});
+		}
+	}
+
+	private startHeartbeat(executionId: string): void {
+		const timer = setInterval(() => {
+			void this.agentExecutionRepository.touchRunning(executionId).catch((error: unknown) => {
+				this.logger.warn('Failed to heartbeat a running agent execution', {
+					executionId,
+					error: error instanceof Error ? error.message : String(error),
+				});
+			});
+		}, AgentExecutionService.heartbeatIntervalMs);
+		timer.unref();
+		this.heartbeatTimers.set(executionId, timer);
+	}
+
+	private stopHeartbeat(executionId: string): void {
+		const timer = this.heartbeatTimers.get(executionId);
+		if (timer) clearInterval(timer);
+		this.heartbeatTimers.delete(executionId);
+	}
+
+	private ensureTimelineSnapshotWrite(executionId: string): void {
+		if (
+			this.timelineSnapshotWrites.has(executionId) ||
+			!this.pendingTimelineSnapshots.has(executionId)
+		) {
+			return;
+		}
+		const write = this.drainTimelineSnapshots(executionId).finally(() => {
+			this.timelineSnapshotWrites.delete(executionId);
+			this.ensureTimelineSnapshotWrite(executionId);
+		});
+		this.timelineSnapshotWrites.set(executionId, write);
+	}
+
+	private async drainTimelineSnapshots(executionId: string): Promise<void> {
+		while (true) {
+			const snapshot = this.pendingTimelineSnapshots.get(executionId);
+			if (!snapshot) return;
+			this.pendingTimelineSnapshots.delete(executionId);
 			try {
-				await this.agentExecutionLogStore.write(
-					{ agentId, threadId, executionId: inserted.id },
-					{ timeline: record.timeline },
-					storedAt,
-				);
+				if (
+					!(await this.agentExecutionRepository.updateTimelineIfRunning(
+						executionId,
+						snapshot.timeline,
+					))
+				) {
+					this.pendingTimelineSnapshots.delete(executionId);
+					return;
+				}
+				this.executionUpdateBroadcaster.notify({
+					projectId: snapshot.projectId,
+					agentId: snapshot.agentId,
+					threadId: snapshot.threadId,
+					executionId,
+				});
 			} catch (error) {
-				// Keep the row even without its blob; the read path treats a missing log as no timeline.
-				this.errorReporter.error(error);
+				if (!this.pendingTimelineSnapshots.has(executionId)) {
+					this.pendingTimelineSnapshots.set(executionId, snapshot);
+				}
+				this.logger.warn('Failed to persist an agent execution timeline snapshot; retrying', {
+					executionId,
+					error: error instanceof Error ? error.message : String(error),
+				});
+				await new Promise<void>((resolve) => {
+					const timer = setTimeout(resolve, TIMELINE_SNAPSHOT_RETRY_DELAY_MS);
+					timer.unref();
+				});
 			}
 		}
+	}
 
-		// When a resumed execution completes with usage data, backfill any
-		// preceding suspended executions in the same thread that are missing it.
+	private async prepareThread(
+		params: StartExecutionParams,
+	): Promise<{ userMessage: string | null; created: boolean }> {
+		const { thread, created } = await this.agentExecutionThreadRepository.findOrCreate(
+			params.threadId,
+			params.agentId,
+			params.agentName,
+			params.projectId,
+			params.threadMetadata,
+			params.taskId,
+			params.taskVersionId,
+		);
+		if (!created) {
+			await this.agentExecutionThreadRepository.bumpUpdatedAt(params.threadId);
+			if (!thread.title) await this.syncTitleFromMemory(params.threadId, params.agentId);
+		}
+		return { userMessage: cleanUserMessage(params.userMessage, params.agentName), created };
+	}
+
+	private async completeRecordedExecution(
+		params: RecordMessageParams,
+		executionId: string,
+		status: AgentExecution['status'],
+	): Promise<void> {
+		const { threadId, agentId, record, hitlStatus } = params;
 		if (hitlStatus === 'resumed' && record.model) {
 			await this.backfillSuspendedExecutions(threadId, record.model);
 		}
-
-		// Atomically increment token/cost/duration counters on the thread
 		if (record.usage) {
 			await this.agentExecutionThreadRepository.incrementUsage(
 				threadId,
@@ -180,20 +343,18 @@ export class AgentExecutionService {
 				record.duration,
 			);
 		}
-
 		if (params.telemetry) {
 			try {
 				this.telemetry.trackAgentTurnFinished({
 					agent_id: agentId,
 					thread_id: threadId,
 					run_type: params.telemetry.runType,
-					turn_status:
-						record.error !== null || record.finishReason === 'error' ? 'failed' : 'succeeded',
+					turn_status: status === 'success' ? 'succeeded' : 'failed',
 					configuration: params.telemetry.configuration,
 					latency_ms: record.duration,
 					cost: record.totalCost ?? 0,
 					token_count: record.usage?.totalTokens ?? 0,
-					tool_call_count: record.timeline.filter((t) => t.type === 'tool-call').length,
+					tool_call_count: record.timeline.filter((event) => event.type === 'tool-call').length,
 				});
 			} catch (error) {
 				this.logger.warn('Failed to track agent execution telemetry', {
@@ -203,23 +364,16 @@ export class AgentExecutionService {
 				});
 			}
 		}
-
 		this.logger.debug('Recorded agent execution', {
-			executionId: inserted.id,
+			executionId,
 			threadId,
 			agentId,
 			status,
 			duration: record.duration,
 		});
-
-		// Title generation now runs synchronously (sync: true) before the stream
-		// ends, so by this point the memory thread should have the title.
-		// Sync it to our execution thread on the first message.
-		if (created) {
+		if (this.executionsNeedingTitleSync.has(executionId)) {
 			await this.syncTitleFromMemory(threadId, agentId);
 		}
-
-		return inserted.id;
 	}
 
 	/**
@@ -430,6 +584,21 @@ export class AgentExecutionService {
 	}
 }
 
+function cleanUserMessage(message: string | null, agentName: string): string | null {
+	if (message === null) return null;
+	const cleaned = message
+		.replace(/<@[A-Z0-9]+>/gi, `@${agentName}`)
+		.replace(/@[A-Z0-9]{8,}/gi, `@${agentName}`)
+		.trim();
+	return cleaned.length > 0 ? cleaned : null;
+}
+
+function executionStatus(record: MessageRecord): AgentExecution['status'] {
+	if (record.error !== null || record.finishReason === 'error') return 'error';
+	if (record.finishReason === 'cancelled') return 'cancelled';
+	return 'success';
+}
+
 /**
  * True if `thread` belongs to the given project and agent.
  * Returns false for threads from a different project/agent so the caller can

packages/cli/src/modules/agents/agent-interrupted-execution-sweeper.ts

@@ -0,0 +1,51 @@
+import { Logger } from '@n8n/backend-common';
+import { Service } from '@n8n/di';
+
+import { AgentExecutionService } from './agent-execution.service';
+import { AgentExecutionRepository } from './repositories/agent-execution.repository';
+
+@Service()
+export class AgentInterruptedExecutionSweeper {
+	static readonly LIVENESS_GRACE_MS = 2 * 60 * 1000;
+
+	constructor(
+		private readonly logger: Logger,
+		private readonly executionRepository: AgentExecutionRepository,
+		private readonly executionService: AgentExecutionService,
+	) {
+		this.logger = this.logger.scoped('agents');
+	}
+
+	async sweep(): Promise<void> {
+		let running;
+		try {
+			running = await this.executionRepository.findRunning();
+		} catch (error) {
+			this.logger.error('Failed to query running agent executions', { error });
+			return;
+		}
+
+		for (const execution of running) {
+			try {
+				if (
+					execution.updatedAt.getTime() >
+					Date.now() - AgentInterruptedExecutionSweeper.LIVENESS_GRACE_MS
+				) {
+					continue;
+				}
+				if (await this.executionService.finalizeInterruptedExecution(execution)) {
+					this.logger.info('Marked abandoned agent execution as interrupted', {
+						executionId: execution.id,
+						threadId: execution.threadId,
+					});
+				}
+			} catch (error) {
+				this.logger.error('Failed to finalize interrupted agent execution', {
+					executionId: execution.id,
+					threadId: execution.threadId,
+					error,
+				});
+			}
+		}
+	}
+}

packages/cli/src/modules/agents/agent-workflow-execution.service.ts

@@ -24,7 +24,7 @@ import type { AgentRunTelemetryType } from '@/interfaces';
 import { ExecutionLevelTracer } from '@/modules/otel/execution-level-tracer';
 import { Telemetry } from '@/telemetry';
 
-import { AgentExecutionService } from './agent-execution.service';
+import { AgentExecutionService, type StartExecutionParams } from './agent-execution.service';
 import { AgentRunTracingService } from './agent-run-tracing.service';
 import { AgentRuntimeReconstructionService } from './agent-runtime-reconstruction.service';
 import {
@@ -215,6 +215,7 @@ export class AgentWorkflowExecutionService {
 			nodeId?: string;
 			nodeName?: string;
 		};
+		recordingParams?: StartExecutionParams;
 	}): Promise<WorkflowAgentRunOutcome> {
 		const {
 			agentInstance,
@@ -225,9 +226,22 @@ export class AgentWorkflowExecutionService {
 			runType,
 			outputSchema,
 			tracing,
+			recordingParams,
 		} = params;
 
-		const recorder = new ExecutionRecorder();
+		let agentExecutionId: string | undefined;
+		const recorder = new ExecutionRecorder(undefined, (timeline) => {
+			if (agentExecutionId && recordingParams) {
+				this.agentExecutionService.recordTimelineSnapshot({
+					projectId: recordingParams.projectId,
+					agentId: recordingParams.agentId,
+					threadId: recordingParams.threadId,
+					executionId: agentExecutionId,
+					timeline,
+				});
+			}
+		});
+		const startedAt = recorder.startedAt;
 
 		let structuredOutput: unknown = null;
 		const toolCalls: ExecuteAgentData['toolCalls'] = [];
@@ -277,6 +291,21 @@ export class AgentWorkflowExecutionService {
 					...(telemetry ? { telemetry } : {}),
 				});
 
+				if (recordingParams) {
+					try {
+						agentExecutionId = await this.agentExecutionService.startExecutionRecording(
+							recordingParams,
+							startedAt,
+						);
+					} catch (error) {
+						this.logger.warn('Failed to start agent execution recording from workflow', {
+							agentId: recordingParams.agentId,
+							threadId,
+							error: error instanceof Error ? error.message : String(error),
+						});
+					}
+				}
+
 				for await (const value of streamAgentChunks(resultStream.stream)) {
 					recorder.record(value);
 
@@ -308,12 +337,28 @@ export class AgentWorkflowExecutionService {
 			await run();
 		}
 
+		if (streamError && recordingParams && !agentExecutionId) {
+			try {
+				agentExecutionId = await this.agentExecutionService.startExecutionRecording(
+					recordingParams,
+					startedAt,
+				);
+			} catch (error) {
+				this.logger.warn('Failed to start agent execution recording from workflow', {
+					agentId: recordingParams.agentId,
+					threadId,
+					error: error instanceof Error ? error.message : String(error),
+				});
+			}
+		}
+
 		return {
 			recorder,
 			messageRecord: recorder.getMessageRecord(),
 			structuredOutput,
 			toolCalls,
 			streamError,
+			agentExecutionId,
 		};
 	}
 
@@ -429,29 +474,43 @@ export class AgentWorkflowExecutionService {
 				nodeId: workflowContext?.callingNodeId,
 				nodeName: workflowContext?.callingNodeName,
 			},
-		});
-
-		void this.agentExecutionService
-			.recordMessage({
+			recordingParams: {
 				threadId,
 				agentId,
 				agentName: agentInstance.name,
 				projectId,
 				userMessage: message,
-				record: run.messageRecord,
 				source: AGENT_WORKFLOW_TRIGGER_TYPE,
 				telemetry: {
 					runType,
 					configuration: telemetryConfiguration,
 				},
-			})
-			.catch((error) => {
+			},
+		});
+
+		if (run.agentExecutionId) {
+			try {
+				await this.agentExecutionService.finalizeExecution(run.agentExecutionId, {
+					threadId,
+					agentId,
+					agentName: agentInstance.name,
+					projectId,
+					userMessage: message,
+					record: run.messageRecord,
+					source: AGENT_WORKFLOW_TRIGGER_TYPE,
+					telemetry: {
+						runType,
+						configuration: telemetryConfiguration,
+					},
+				});
+			} catch (error) {
 				this.logger.warn('Failed to record agent execution from workflow', {
 					agentId,
 					threadId,
 					error: error instanceof Error ? error.message : String(error),
 				});
-			});
+			}
+		}
 
 		return this.buildWorkflowResult({
 			run,
@@ -620,4 +679,5 @@ interface WorkflowAgentRunOutcome {
 	structuredOutput: unknown;
 	toolCalls: ExecuteAgentData['toolCalls'];
 	streamError?: Error;
+	agentExecutionId?: string;
 }

packages/cli/src/modules/agents/agents.module.ts

@@ -1,12 +1,14 @@
 import { Logger } from '@n8n/backend-common';
 import { AgentsConfig } from '@n8n/config';
 import type { ModuleInterface } from '@n8n/decorators';
-import { BackendModule } from '@n8n/decorators';
+import { BackendModule, OnShutdown } from '@n8n/decorators';
 import { Container } from '@n8n/di';
 import { InstanceSettings } from 'n8n-core';
 
 @BackendModule({ name: 'agents' })
 export class AgentsModule implements ModuleInterface {
+	private interruptedExecutionSweepTimer?: NodeJS.Timeout;
+
 	async init() {
 		await import('./agents-catalog.controller.js');
 		await import('./agent-threads.controller.js');
@@ -96,6 +98,24 @@ export class AgentsModule implements ModuleInterface {
 		const taskService = Container.get(AgentTaskService);
 		const logger = Container.get(Logger);
 		const instanceSettings = Container.get(InstanceSettings);
+		if (instanceSettings.instanceType === 'main') {
+			const { AgentInterruptedExecutionSweeper } = await import(
+				'./agent-interrupted-execution-sweeper.js'
+			);
+			const sweep = () => {
+				void Container.get(AgentInterruptedExecutionSweeper)
+					.sweep()
+					.catch((error: unknown) => {
+						logger.error('[Agents] Interrupted execution sweep failed', { error });
+					});
+			};
+			sweep();
+			this.interruptedExecutionSweepTimer = setInterval(
+				sweep,
+				AgentInterruptedExecutionSweeper.LIVENESS_GRACE_MS,
+			);
+			this.interruptedExecutionSweepTimer.unref();
+		}
 		void chatService.reconnectAll().catch((error) => {
 			logger.error('[Agents] Failed to reconnect integrations on startup', {
 				error: error instanceof Error ? error.message : String(error),
@@ -112,6 +132,13 @@ export class AgentsModule implements ModuleInterface {
 		}
 	}
 
+	@OnShutdown()
+	async shutdown() {
+		if (this.interruptedExecutionSweepTimer) {
+			clearInterval(this.interruptedExecutionSweepTimer);
+		}
+	}
+
 	async settings() {
 		const config = Container.get(AgentsConfig);
 		const { isAgentKnowledgeBaseEnabled } = await import('./agent-knowledge-gate.js');

packages/cli/src/modules/agents/entities/agent-execution.entity.ts

@@ -9,7 +9,7 @@ import { Column, Entity, Index, JoinColumn, ManyToOne } from '@n8n/typeorm';
 import { AgentExecutionThread } from './agent-execution-thread.entity';
 import type { TimelineEvent } from '../execution-recorder';
 
-export type AgentExecutionStatus = 'success' | 'error';
+export type AgentExecutionStatus = 'running' | 'success' | 'error' | 'cancelled' | 'interrupted';
 export type AgentExecutionHitlStatus = 'suspended' | 'resumed';
 
 /**
@@ -25,6 +25,7 @@ export type AgentExecutionHitlStatus = 'suspended' | 'resumed';
  */
 @Entity({ name: 'agent_execution' })
 @Index(['threadId', 'createdAt'])
+@Index(['status'], { where: '"status" = \'running\'' })
 export class AgentExecution extends WithTimestampsAndStringId {
 	@ManyToOne(() => AgentExecutionThread, { onDelete: 'CASCADE' })
 	@JoinColumn({ name: 'threadId' })

packages/cli/src/modules/agents/execution-recorder.ts

@@ -14,6 +14,7 @@ import type { ToolRegistry } from './tool-registry';
 /** Cap on child trace characters persisted per delegation. Tighter than the
  *  live forwarding budget because this is written into every parent execution row. */
 const CHILD_TRACE_PERSIST_CHAR_BUDGET = 4_000;
+const TIMELINE_BLOCK_MAX_DURATION_MS = 1_000;
 
 /**
  * Walk a nodeParameters tree and substitute templated values with what the
@@ -258,7 +259,10 @@ export interface MessageRecord {
 export class ExecutionRecorder {
 	private readonly registry: ToolRegistry;
 
-	constructor(registry?: ToolRegistry) {
+	constructor(
+		registry?: ToolRegistry,
+		private readonly onTimelineSnapshot?: (timeline: TimelineEvent[]) => void,
+	) {
 		this.registry = registry ?? new Map();
 	}
 
@@ -286,6 +290,8 @@ export class ExecutionRecorder {
 	/** Wall-clock when the current reasoning segment started. */
 	private reasoningStartTime: number | null = null;
 
+	private timelineSnapshotTimer?: NodeJS.Timeout;
+
 	private _suspended = false;
 
 	private error: string | null = null;
@@ -299,9 +305,12 @@ export class ExecutionRecorder {
 		switch (chunk.type) {
 			case 'text-delta':
 				this.flushReasoningBuffer();
-				if (this.textStartTime === null) this.textStartTime = Date.now();
+				if (this.textStartTime === null) {
+					this.textStartTime = Date.now();
+				}
 				this.textParts.push(chunk.delta);
 				this.textBuffer.push(chunk.delta);
+				this.scheduleTimelineSnapshot();
 				break;
 			case 'reasoning-start':
 				this.flushTextBuffer();
@@ -310,8 +319,11 @@ export class ExecutionRecorder {
 				break;
 			case 'reasoning-delta':
 				this.flushTextBuffer();
-				if (this.reasoningStartTime === null) this.reasoningStartTime = Date.now();
+				if (this.reasoningStartTime === null) {
+					this.reasoningStartTime = Date.now();
+				}
 				this.reasoningBuffer.push(chunk.delta);
+				this.scheduleTimelineSnapshot();
 				break;
 			case 'reasoning-end':
 				this.flushReasoningBuffer();
@@ -370,7 +382,7 @@ export class ExecutionRecorder {
 				this.flushReasoningBuffer();
 				this.flushTextBuffer();
 				this._suspended = true;
-				this.timeline.push({
+				this.appendCompletedEvent({
 					type: 'suspension',
 					toolName: chunk.toolName ?? '',
 					toolCallId: chunk.toolCallId ?? '',
@@ -391,6 +403,10 @@ export class ExecutionRecorder {
 		return this._suspended;
 	}
 
+	get startedAt(): Date {
+		return new Date(this.startTime);
+	}
+
 	/** Build the final message record after the stream has ended. */
 	getMessageRecord(): MessageRecord {
 		this.flushReasoningBuffer();
@@ -410,11 +426,15 @@ export class ExecutionRecorder {
 
 	/** Flush accumulated text into a timeline event. */
 	private flushTextBuffer(): void {
+		if (this.textStartTime !== null && this.timelineSnapshotTimer) {
+			clearTimeout(this.timelineSnapshotTimer);
+			this.timelineSnapshotTimer = undefined;
+		}
 		if (this.textBuffer.length === 0) return;
 		const content = this.textBuffer.join('');
 		if (content.trim()) {
 			const now = Date.now();
-			this.timeline.push({
+			this.appendCompletedEvent({
 				type: 'text',
 				content,
 				// Generation start (first text-delta) → end (now). Falls back to `now`
@@ -429,6 +449,10 @@ export class ExecutionRecorder {
 
 	/** Flush accumulated reasoning without including it in `assistantResponse`. */
 	private flushReasoningBuffer(): void {
+		if (this.reasoningStartTime !== null && this.timelineSnapshotTimer) {
+			clearTimeout(this.timelineSnapshotTimer);
+			this.timelineSnapshotTimer = undefined;
+		}
 		if (this.reasoningBuffer.length === 0) {
 			this.reasoningStartTime = null;
 			return;
@@ -436,7 +460,7 @@ export class ExecutionRecorder {
 		const content = this.reasoningBuffer.join('');
 		if (content.trim()) {
 			const now = Date.now();
-			this.timeline.push({
+			this.appendCompletedEvent({
 				type: 'reasoning',
 				content,
 				timestamp: this.reasoningStartTime ?? now,
@@ -447,6 +471,30 @@ export class ExecutionRecorder {
 		this.reasoningStartTime = null;
 	}
 
+	private scheduleTimelineSnapshot(): void {
+		if (this.timelineSnapshotTimer) return;
+		this.timelineSnapshotTimer = setTimeout(() => {
+			this.timelineSnapshotTimer = undefined;
+			const now = Date.now();
+			if (this.textStartTime !== null && this.textBuffer.length > 0) {
+				this.emitTimelineSnapshot({
+					type: 'text',
+					content: this.textBuffer.join(''),
+					timestamp: this.textStartTime,
+					endTime: now,
+				});
+			} else if (this.reasoningStartTime !== null && this.reasoningBuffer.length > 0) {
+				this.emitTimelineSnapshot({
+					type: 'reasoning',
+					content: this.reasoningBuffer.join(''),
+					timestamp: this.reasoningStartTime,
+					endTime: now,
+				});
+			}
+		}, TIMELINE_BLOCK_MAX_DURATION_MS);
+		this.timelineSnapshotTimer.unref();
+	}
+
 	/**
 	 * Record a discrete `tool-call` chunk from the stream. The matching
 	 * `tool-result` chunk closes the timeline entry.
@@ -498,7 +546,10 @@ export class ExecutionRecorder {
 	private recordToolExecutionStart(toolCallId: string, startTime: number): void {
 		if (!toolCallId) return;
 		const entry = this.findOpenTimelineToolCall(toolCallId);
-		if (entry) entry.startTime = startTime;
+		if (entry) {
+			entry.startTime = startTime;
+			this.emitTimelineSnapshot();
+		}
 	}
 
 	/**
@@ -514,6 +565,7 @@ export class ExecutionRecorder {
 			entry.endTime = endTime;
 			entry.success = !isError;
 			if (entry.childTrace) settleChildTrace(entry.childTrace);
+			this.emitTimelineSnapshot();
 		}
 	}
 
@@ -572,6 +624,7 @@ export class ExecutionRecorder {
 					pendingTimeline.workflowExecutionId = execId;
 				}
 			}
+			this.emitTimelineSnapshot();
 			return;
 		}
 
@@ -606,6 +659,17 @@ export class ExecutionRecorder {
 				synthesized.workflowExecutionId = execId;
 			}
 		}
-		this.timeline.push(synthesized);
+		this.appendCompletedEvent(synthesized);
+	}
+
+	private appendCompletedEvent(event: TimelineEvent): void {
+		this.timeline.push(event);
+		this.emitTimelineSnapshot();
+	}
+
+	private emitTimelineSnapshot(activeEvent?: TimelineEvent): void {
+		if (!this.onTimelineSnapshot) return;
+		const timeline = activeEvent ? [...this.timeline, activeEvent] : this.timeline;
+		this.onTimelineSnapshot(structuredClone(timeline));
 	}
 }

packages/cli/src/modules/agents/repositories/agent-execution.repository.ts

@@ -1,8 +1,25 @@
 import { Service } from '@n8n/di';
 import { DataSource, IsNull, Not, Repository } from '@n8n/typeorm';
+import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity';
 
 import { AgentExecution } from '../entities/agent-execution.entity';
 
+export type RunningAgentExecution = Pick<
+	AgentExecution,
+	'id' | 'threadId' | 'startedAt' | 'updatedAt' | 'timeline'
+>;
+
+type AgentExecutionFinalizationValues = Pick<
+	AgentExecution,
+	'status' | 'stoppedAt' | 'duration' | 'timeline' | 'storedAt' | 'error'
+> &
+	Partial<
+		Pick<
+			AgentExecution,
+			'model' | 'promptTokens' | 'completionTokens' | 'totalTokens' | 'cost' | 'hitlStatus'
+		>
+	>;
+
 @Service()
 export class AgentExecutionRepository extends Repository<AgentExecution> {
 	constructor(dataSource: DataSource) {
@@ -14,6 +31,39 @@ export class AgentExecutionRepository extends Repository<AgentExecution> {
 		return await this.find({ where: { threadId }, order: { createdAt: 'ASC' } });
 	}
 
+	async findRunning(): Promise<RunningAgentExecution[]> {
+		return await this.find({
+			select: ['id', 'threadId', 'startedAt', 'updatedAt', 'timeline'],
+			where: { status: 'running' },
+		});
+	}
+
+	async touchRunning(executionId: string): Promise<void> {
+		await this.update({ id: executionId, status: 'running' }, { updatedAt: new Date() });
+	}
+
+	async updateTimelineIfRunning(
+		executionId: string,
+		timeline: AgentExecution['timeline'],
+	): Promise<boolean> {
+		const result = await this.update({ id: executionId, status: 'running' }, {
+			timeline,
+			updatedAt: new Date(),
+		} as QueryDeepPartialEntity<AgentExecution>);
+		return result.affected === 1;
+	}
+
+	async updateIfRunning(
+		executionId: string,
+		values: AgentExecutionFinalizationValues,
+	): Promise<boolean> {
+		const result = await this.update(
+			{ id: executionId, status: 'running' },
+			values as QueryDeepPartialEntity<AgentExecution>,
+		);
+		return result.affected === 1;
+	}
+
 	/**
 	 * The first user-message text in each of the given threads. Used by the
 	 * sessions list to render a preview before the LLM-generated title is

packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-foreground-runner.test.ts

@@ -122,6 +122,8 @@ describe('SubAgentForegroundRunner', () => {
 		reconstructionService = mock<AgentRuntimeReconstructionService>();
 		Container.set(AgentRuntimeReconstructionService, reconstructionService);
 		agentExecutionService = mock<AgentExecutionService>();
+		agentExecutionService.startExecutionRecording.mockResolvedValue('agent-execution-1');
+		agentExecutionService.finalizeExecution.mockResolvedValue('agent-execution-1');
 		checkpointStorage = mock<N8NCheckpointStorage>();
 		logger = mock<Logger>();
 		runner = new SubAgentForegroundRunner(
@@ -153,6 +155,8 @@ describe('SubAgentForegroundRunner', () => {
 	});
 
 	it('rebuilds the child through the shared reconstruction service and runs it with a fresh prompt', async () => {
+		agentExecutionService.startExecutionRecording.mockResolvedValue('agent-execution-1');
+		agentExecutionService.finalizeExecution.mockResolvedValue('agent-execution-1');
 		const result = await runner.runForeground(spawnRequest, {
 			projectId,
 			credentialProvider,
@@ -196,7 +200,20 @@ describe('SubAgentForegroundRunner', () => {
 		const childPrompt = childAgent.stream.mock.calls[0]?.[0] as string;
 		expect(childPrompt).toContain('CONTEXT:\nFocus on auth endpoints.');
 		expect(childPrompt).toContain('EXPECTED OUTPUT:\nA concise summary.');
-		expect(agentExecutionService.recordMessage).toHaveBeenCalledWith(
+		expect(agentExecutionService.startExecutionRecording).toHaveBeenCalledWith(
+			expect.objectContaining({ threadId: result.threadId, source: 'subagent' }),
+			expect.any(Date),
+		);
+		expect(agentExecutionService.recordTimelineSnapshot).toHaveBeenCalledWith(
+			expect.objectContaining({
+				projectId,
+				agentId: 'agent-1',
+				threadId: result.threadId,
+				executionId: 'agent-execution-1',
+			}),
+		);
+		expect(agentExecutionService.finalizeExecution).toHaveBeenCalledWith(
+			'agent-execution-1',
 			expect.objectContaining({
 				threadId: result.threadId,
 				agentId: 'agent-1',
@@ -209,6 +226,9 @@ describe('SubAgentForegroundRunner', () => {
 				},
 			}),
 		);
+		const startedAt = agentExecutionService.startExecutionRecording.mock.calls[0][1];
+		const finalizedRecord = agentExecutionService.finalizeExecution.mock.calls[0][1].record;
+		expect(startedAt.getTime()).toBe(finalizedRecord.startTime);
 	});
 
 	it('records the child turn with the parent run type, not its own published state', async () => {
@@ -218,7 +238,8 @@ describe('SubAgentForegroundRunner', () => {
 			runType: 'test',
 		});
 
-		expect(agentExecutionService.recordMessage).toHaveBeenCalledWith(
+		expect(agentExecutionService.finalizeExecution).toHaveBeenCalledWith(
+			'agent-execution-1',
 			expect.objectContaining({
 				threadId: result.threadId,
 				agentId: 'agent-1',
@@ -308,7 +329,8 @@ describe('SubAgentForegroundRunner', () => {
 				},
 			}),
 		);
-		expect(agentExecutionService.recordMessage).toHaveBeenCalledWith(
+		expect(agentExecutionService.finalizeExecution).toHaveBeenCalledWith(
+			'agent-execution-1',
 			expect.objectContaining({
 				threadId: result.threadId,
 				agentId: 'agent-2',
@@ -374,7 +396,8 @@ describe('SubAgentForegroundRunner', () => {
 				],
 			},
 		});
-		expect(agentExecutionService.recordMessage).toHaveBeenCalledWith(
+		expect(agentExecutionService.finalizeExecution).toHaveBeenCalledWith(
+			'agent-execution-1',
 			expect.objectContaining({ hitlStatus: 'suspended' }),
 		);
 		expect(childAgent.close).toHaveBeenCalledTimes(1);
@@ -416,7 +439,8 @@ describe('SubAgentForegroundRunner', () => {
 			status: 'completed',
 			result: { runId: 'child-run-1', finishReason: 'stop' },
 		});
-		expect(agentExecutionService.recordMessage).toHaveBeenCalledWith(
+		expect(agentExecutionService.finalizeExecution).toHaveBeenCalledWith(
+			'agent-execution-1',
 			expect.objectContaining({
 				threadId: 'child-thread-1',
 				agentId: 'agent-1',

packages/cli/src/modules/agents/sub-agents/sub-agent-foreground-runner.ts

@@ -187,6 +187,20 @@ export class SubAgentForegroundRunner {
 		const telemetry = deriveSubAgentTelemetry(context.telemetry);
 		const userMessage =
 			operation.type === 'run' ? renderDelegateSubAgentPrompt(operation.request) : null;
+		let executionId: string | undefined;
+		const recorder = new ExecutionRecorder(undefined, (timeline) => {
+			if (executionId) {
+				this.agentExecutionService.recordTimelineSnapshot({
+					projectId: context.projectId,
+					agentId: runtimeSource.source.sourceId,
+					threadId,
+					executionId,
+					timeline,
+				});
+			}
+		});
+		const startedAt = recorder.startedAt;
+		let recorded = false;
 		try {
 			const executionOptions = {
 				...(context.abortSignal !== undefined ? { abortSignal: context.abortSignal } : {}),
@@ -208,7 +222,41 @@ export class SubAgentForegroundRunner {
 							runId: operation.request.childRunId,
 							toolCallId: operation.request.childToolCallId,
 						});
-			const { messageRecord, result } = await consumeAgentStream(resultStream, context.onChunk);
+			try {
+				const currentExecutionId = await this.agentExecutionService.startExecutionRecording(
+					{
+						threadId,
+						agentId: runtimeSource.source.sourceId,
+						agentName: runtimeSource.source.config.name,
+						projectId: context.projectId,
+						userMessage,
+						source: 'subagent',
+						threadMetadata: {
+							parentThreadId: operation.request.parentThreadId,
+							parentAgentId: context.parentAgentId,
+						},
+						telemetry: {
+							runType: context.runType,
+							configuration: buildAgentConfigurationTelemetryFromConfig(
+								runtimeSource.source.config,
+							),
+						},
+					},
+					startedAt,
+				);
+				executionId = currentExecutionId;
+			} catch (error) {
+				this.logger.warn('Failed to start subagent execution recording', {
+					agentId: runtimeSource.source.sourceId,
+					taskPath: operation.taskPath,
+					error: error instanceof Error ? error.message : String(error),
+				});
+			}
+			const { messageRecord, result } = await consumeAgentStream(
+				resultStream,
+				recorder,
+				context.onChunk,
+			);
 			const suspended = result.pendingSuspend !== undefined && result.pendingSuspend.length > 0;
 			const hitlStatus = suspended
 				? 'suspended'
@@ -225,8 +273,10 @@ export class SubAgentForegroundRunner {
 				taskPath: operation.taskPath,
 				userMessage,
 				record: messageRecord,
+				executionId,
 				...(hitlStatus !== undefined ? { hitlStatus } : {}),
 			});
+			recorded = true;
 
 			return {
 				taskPath: operation.taskPath,
@@ -239,6 +289,25 @@ export class SubAgentForegroundRunner {
 				result,
 				...(suspended ? { resumeContext: createResumeContext(runtimeSource.source) } : {}),
 			};
+		} catch (error) {
+			if (!recorded) {
+				recorder.record({ type: 'error', error });
+				recorder.record({ type: 'finish', finishReason: 'error' });
+				await this.recordSubAgentExecution({
+					runtimeSource: runtimeSource.source,
+					projectId: context.projectId,
+					threadId,
+					parentThreadId: operation.request.parentThreadId,
+					parentAgentId: context.parentAgentId,
+					runType: context.runType,
+					taskPath: operation.taskPath,
+					userMessage,
+					record: recorder.getMessageRecord(),
+					executionId,
+					...(operation.type === 'resume' ? { hitlStatus: 'resumed' as const } : {}),
+				});
+			}
+			throw error;
 		} finally {
 			await agent.close().catch((error) => {
 				this.logger.warn(`Failed to close subagent after ${operation.type}`, {
@@ -260,6 +329,7 @@ export class SubAgentForegroundRunner {
 		taskPath: SubAgentTaskPath;
 		userMessage: string | null;
 		record: MessageRecord;
+		executionId?: string;
 		hitlStatus?: 'suspended' | 'resumed';
 	}): Promise<void> {
 		const {
@@ -272,11 +342,13 @@ export class SubAgentForegroundRunner {
 			taskPath,
 			userMessage,
 			record,
+			executionId,
 			hitlStatus,
 		} = params;
 
+		if (!executionId) return;
 		try {
-			await this.agentExecutionService.recordMessage({
+			await this.agentExecutionService.finalizeExecution(executionId, {
 				threadId,
 				agentId: runtimeSource.sourceId,
 				agentName: runtimeSource.config.name,
@@ -319,9 +391,9 @@ async function getReconstructionService() {
 
 async function consumeAgentStream(
 	resultStream: StreamResult,
+	recorder: ExecutionRecorder,
 	onChunk?: (chunk: StreamChunk) => void,
 ): Promise<{ messageRecord: MessageRecord; result: GenerateResult }> {
-	const recorder = new ExecutionRecorder();
 	const pendingSuspend: NonNullable<GenerateResult['pendingSuspend']> = [];
 	let structuredOutput: unknown;
 

packages/cli/src/scaling/constants.ts

@@ -38,6 +38,7 @@ export const IMMEDIATE_COMMANDS = new Set<PubSub.Command['command']>([
 	'remove-triggers-and-pollers',
 	'relay-execution-lifecycle-event',
 	'relay-chat-stream-event',
+	'relay-agent-execution-update',
 	'relay-instance-ai-event',
 	'relay-instance-ai-task-control',
 	'agent-chat-subscription-changed',

packages/cli/src/scaling/pubsub/__tests__/publisher.service.test.ts

@@ -143,6 +143,7 @@ describe('Publisher', () => {
 		});
 
 		it.each([
+			'relay-agent-execution-update',
 			'display-workflow-activation',
 			'display-workflow-deactivation',
 			'display-workflow-activation-error',

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

@@ -3,6 +3,7 @@ import type {
 	ChatHubMessageStatus,
 	InstanceAiEvent,
 	PushMessage,
+	PushPayload,
 	WorkerStatus,
 	WorkflowPublicationStatusMessage,
 } from '@n8n/api-types';
@@ -124,6 +125,11 @@ export type PubSubCommandMap = {
 		asBinary: boolean;
 	};
 
+	'relay-agent-execution-update': {
+		data: PushPayload<'agentExecutionUpdated'>;
+		userIds: string[];
+	};
+
 	'clear-test-webhooks': {
 		webhookKey: string;
 		workflowEntity: IWorkflowBase;

packages/cli/src/scaling/pubsub/pubsub.types.ts

@@ -59,6 +59,7 @@ export namespace PubSub {
 		export type DisplayWorkflowPublicationStatus = ToCommand<'display-workflow-publication-status'>;
 		export type WorkflowPublishWakeUp = ToCommand<'workflow-publish-wake-up'>;
 		export type RelayExecutionLifecycleEvent = ToCommand<'relay-execution-lifecycle-event'>;
+		export type RelayAgentExecutionUpdate = ToCommand<'relay-agent-execution-update'>;
 		export type RelayChatStreamEvent = ToCommand<'relay-chat-stream-event'>;
 		export type RelayInstanceAiEvent = ToCommand<'relay-instance-ai-event'>;
 		export type RelayInstanceAiTaskControl = ToCommand<'relay-instance-ai-task-control'>;
@@ -99,6 +100,7 @@ export namespace PubSub {
 		| Commands.DisplayWorkflowPublicationStatus
 		| Commands.WorkflowPublishWakeUp
 		| Commands.RelayExecutionLifecycleEvent
+		| Commands.RelayAgentExecutionUpdate
 		| Commands.RelayChatStreamEvent
 		| Commands.RelayInstanceAiEvent
 		| Commands.RelayInstanceAiTaskControl

packages/cli/test/migration/1785828155091-add-agent-execution-runtime-state.test.ts

@@ -0,0 +1,118 @@
+import {
+	createTestMigrationContext,
+	initDbUpToMigration,
+	runSingleMigration,
+	type TestMigrationContext,
+} from '@n8n/backend-test-utils';
+import { DbConnection } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { DataSource } from '@n8n/typeorm';
+import { randomUUID } from 'node:crypto';
+
+const MIGRATION_NAME = 'AddAgentExecutionRuntimeState1785828155091';
+
+describe('AddAgentExecutionRuntimeState migration', () => {
+	let dataSource: DataSource;
+	let context: TestMigrationContext;
+	let executionId: string;
+
+	beforeAll(async () => {
+		await Container.get(DbConnection).init();
+		dataSource = Container.get(DataSource);
+	});
+
+	beforeEach(async () => {
+		context = createTestMigrationContext(dataSource);
+		await context.queryRunner.clearDatabase();
+		await context.queryRunner.release();
+		await initDbUpToMigration(MIGRATION_NAME);
+
+		context = createTestMigrationContext(dataSource);
+		const projectId = randomUUID();
+		const agentId = randomUUID();
+		const threadId = randomUUID();
+		executionId = randomUUID();
+		const now = new Date();
+		await context.runQuery(
+			`INSERT INTO ${context.escape.tableName('project')} ("id", "name", "type", "createdAt", "updatedAt")
+			 VALUES (:id, :name, 'team', :createdAt, :updatedAt)`,
+			{ id: projectId, name: 'Project', createdAt: now, updatedAt: now },
+		);
+		await context.runQuery(
+			`INSERT INTO ${context.escape.tableName('agents')} ("id", "name", "projectId", "integrations", "tools", "skills", "createdAt", "updatedAt")
+			 VALUES (:id, :name, :projectId, :integrations, :tools, :skills, :createdAt, :updatedAt)`,
+			{
+				id: agentId,
+				name: 'Agent',
+				projectId,
+				integrations: '[]',
+				tools: '{}',
+				skills: '{}',
+				createdAt: now,
+				updatedAt: now,
+			},
+		);
+		await context.runQuery(
+			`INSERT INTO ${context.escape.tableName('agent_execution_threads')}
+			 ("id", "agentId", "agentName", "projectId", "createdAt", "updatedAt")
+			 VALUES (:id, :agentId, :agentName, :projectId, :createdAt, :updatedAt)`,
+			{ id: threadId, agentId, agentName: 'Agent', projectId, createdAt: now, updatedAt: now },
+		);
+		await context.runQuery(
+			`INSERT INTO ${context.escape.tableName('agent_execution')}
+			 ("id", "threadId", "status", "duration", "storedAt", "createdAt", "updatedAt")
+			 VALUES (:id, :threadId, 'success', 0, 'db', :createdAt, :updatedAt)`,
+			{ id: executionId, threadId, createdAt: now, updatedAt: now },
+		);
+		await context.queryRunner.release();
+
+		await runSingleMigration(MIGRATION_NAME);
+	});
+
+	afterAll(async () => {
+		await Container.get(DbConnection).close();
+	});
+
+	it('supports runtime statuses and indexes running execution scans', async () => {
+		const ctx = createTestMigrationContext(dataSource);
+		const executionTable = ctx.escape.tableName('agent_execution');
+
+		for (const status of ['running', 'cancelled', 'interrupted']) {
+			await expect(
+				ctx.runQuery(`UPDATE ${executionTable} SET "status" = :status WHERE "id" = :id`, {
+					id: executionId,
+					status,
+				}),
+			).resolves.not.toThrow();
+		}
+		expect(
+			(await ctx.queryRunner.getTable(`${ctx.tablePrefix}agent_execution`))?.indices.some(
+				(index) => index.columnNames[0] === 'status' && index.where?.includes("'running'"),
+			),
+		).toBe(true);
+		await ctx.queryRunner.release();
+	});
+
+	it('maps runtime-only statuses to error on rollback', async () => {
+		const runtimeCtx = createTestMigrationContext(dataSource);
+		await runtimeCtx.runQuery(
+			`UPDATE ${runtimeCtx.escape.tableName('agent_execution')} SET "status" = 'running' WHERE "id" = :id`,
+			{ id: executionId },
+		);
+		await runtimeCtx.queryRunner.release();
+
+		await dataSource.undoLastMigration({ transaction: 'each' });
+		const ctx = createTestMigrationContext(dataSource);
+		const row = await ctx.runQuery<Array<{ status: string }>>(
+			`SELECT "status" FROM ${ctx.escape.tableName('agent_execution')} WHERE "id" = :id`,
+			{ id: executionId },
+		);
+		expect(row).toEqual([{ status: 'error' }]);
+		expect(
+			(await ctx.queryRunner.getTable(`${ctx.tablePrefix}agent_execution`))?.indices.some(
+				(index) => index.columnNames[0] === 'status',
+			),
+		).toBe(false);
+		await ctx.queryRunner.release();
+	});
+});

packages/frontend/editor-ui/src/features/agents/__tests__/AgentSessionTimelinePanel.spec.ts

@@ -1,11 +1,23 @@
 /* eslint-disable import-x/no-extraneous-dependencies -- test-only patterns */
-import { describe, it, expect, vi, beforeEach } from 'vitest';
-import { mount, flushPromises } from '@vue/test-utils';
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
+import type { PushMessage } from '@n8n/api-types';
+import { enableAutoUnmount, mount, flushPromises } from '@vue/test-utils';
+import { reactive } from 'vue';
 import AgentSessionTimelinePanel from '../components/AgentSessionTimelinePanel.vue';
 import type { ThreadDetail } from '../composables/useAgentThreadsApi';
 
 const getThreadDetail = vi.fn();
 const showError = vi.fn();
+const pushHandlers = new Set<(event: PushMessage) => void>();
+const pushStore = reactive({
+	isConnected: false,
+	pushConnect: vi.fn(),
+	pushDisconnect: vi.fn(),
+	addEventListener: vi.fn((handler: (event: PushMessage) => void) => {
+		pushHandlers.add(handler);
+		return () => pushHandlers.delete(handler);
+	}),
+});
 
 vi.mock('@n8n/composables/useToast', () => ({
 	useToast: () => ({ showError }),
@@ -15,6 +27,10 @@ vi.mock('@/features/agents/agentSessions.store', () => ({
 	useAgentSessionsStore: () => ({ getThreadDetail }),
 }));
 
+vi.mock('@/app/stores/pushConnection.store', () => ({
+	usePushConnectionStore: () => pushStore,
+}));
+
 // Avoid the project-agents list dependency — the panel only reads `.value`.
 vi.mock('@/features/agents/composables/useSubAgentNames', () => ({
 	useSubAgentNames: () => ({ subAgentNameById: { value: new Map() } }),
@@ -22,7 +38,12 @@ vi.mock('@/features/agents/composables/useSubAgentNames', () => ({
 
 const stubs = {
 	SessionTimelineChart: { template: '<div data-test-id="chart-stub" />' },
-	SessionTimelineTable: { template: '<div data-test-id="table-stub" />' },
+	SessionTimelineTable: {
+		name: 'SessionTimelineTable',
+		props: ['selectedIndex', 'items'],
+		template:
+			'<div data-test-id="table-stub" :data-selected-index="selectedIndex ?? undefined" :data-item-count="items.length" />',
+	},
 	SessionEventFilter: { template: '<div data-test-id="filter-stub" />' },
 	SessionDetailPanel: { template: '<div data-test-id="detail-stub" />' },
 	N8nInput: { template: '<input data-test-id="search-stub" />' },
@@ -34,6 +55,18 @@ const detail: ThreadDetail = {
 	executions: [],
 };
 
+const update: PushMessage = {
+	type: 'agentExecutionUpdated',
+	data: {
+		projectId: 'p1',
+		agentId: 'a1',
+		threadId: 't1',
+		executionId: 'exec-1',
+	},
+};
+
+enableAutoUnmount(afterEach);
+
 function mountPanel(props: Partial<{ projectId: string; agentId: string; threadId: string }> = {}) {
 	return mount(AgentSessionTimelinePanel, {
 		props: { projectId: 'p1', agentId: 'a1', threadId: 't1', ...props },
@@ -41,9 +74,15 @@ function mountPanel(props: Partial<{ projectId: string; agentId: string; threadI
 	});
 }
 
+function emitPush(message: PushMessage = update) {
+	pushHandlers.forEach((handler) => handler(message));
+}
+
 describe('AgentSessionTimelinePanel', () => {
 	beforeEach(() => {
 		vi.clearAllMocks();
+		pushHandlers.clear();
+		pushStore.isConnected = false;
 		getThreadDetail.mockResolvedValue(detail);
 	});
 
@@ -91,4 +130,152 @@ describe('AgentSessionTimelinePanel', () => {
 
 		expect(showError).toHaveBeenCalled();
 	});
+
+	it('refreshes the rendered thread only for matching invalidations', async () => {
+		const refreshed = {
+			...detail,
+			executions: [
+				{
+					id: 'exec-1',
+					status: 'running',
+					timeline: [{ type: 'text', content: 'Working', timestamp: 1 }],
+				},
+			],
+		} as unknown as ThreadDetail;
+		const wrapper = mountPanel();
+		await flushPromises();
+
+		for (const data of [
+			{ ...update.data, projectId: 'other-project' },
+			{ ...update.data, agentId: 'other-agent' },
+			{ ...update.data, threadId: 'other-thread' },
+		]) {
+			emitPush({ type: 'agentExecutionUpdated', data });
+		}
+		await flushPromises();
+		expect(getThreadDetail).toHaveBeenCalledTimes(1);
+
+		getThreadDetail.mockResolvedValueOnce(refreshed);
+		emitPush();
+		await flushPromises();
+
+		expect(getThreadDetail).toHaveBeenCalledTimes(2);
+		expect(wrapper.emitted('loaded')?.at(-1)).toEqual([refreshed]);
+		expect(wrapper.find('[data-test-id="table-stub"]').attributes('data-item-count')).toBe('1');
+	});
+
+	it('reconciles state after the push connection reconnects', async () => {
+		const refreshed = { ...detail, thread: { ...detail.thread, title: 'Updated' } };
+		getThreadDetail.mockResolvedValueOnce(detail).mockResolvedValueOnce(refreshed);
+		const wrapper = mountPanel();
+		await flushPromises();
+
+		pushStore.isConnected = true;
+		await flushPromises();
+
+		expect(getThreadDetail).toHaveBeenCalledTimes(2);
+		expect(wrapper.emitted('loaded')?.at(-1)).toEqual([refreshed]);
+	});
+
+	it('coalesces invalidations while a refresh is in flight', async () => {
+		let resolveRefresh!: (value: ThreadDetail) => void;
+		const refreshed = { ...detail, thread: { ...detail.thread, title: 'Refreshed' } };
+		getThreadDetail
+			.mockResolvedValueOnce(detail)
+			.mockReturnValueOnce(new Promise<ThreadDetail>((resolve) => (resolveRefresh = resolve)))
+			.mockResolvedValueOnce(refreshed);
+		const wrapper = mountPanel();
+		await flushPromises();
+
+		emitPush();
+		emitPush();
+		emitPush();
+		await flushPromises();
+		expect(getThreadDetail).toHaveBeenCalledTimes(2);
+
+		resolveRefresh(detail);
+		await flushPromises();
+
+		expect(getThreadDetail).toHaveBeenCalledTimes(3);
+		expect(wrapper.emitted('loaded')?.at(-1)).toEqual([refreshed]);
+	});
+
+	it('ignores stale responses and invalidations after the selected thread changes', async () => {
+		let resolveStaleRefresh!: (value: ThreadDetail) => void;
+		const nextDetail = {
+			...detail,
+			thread: { ...detail.thread, id: 't2', title: 'Current thread' },
+		} as ThreadDetail;
+		const staleDetail = {
+			...detail,
+			thread: { ...detail.thread, title: 'Stale thread' },
+		} as ThreadDetail;
+		getThreadDetail
+			.mockResolvedValueOnce(detail)
+			.mockReturnValueOnce(new Promise<ThreadDetail>((resolve) => (resolveStaleRefresh = resolve)))
+			.mockResolvedValueOnce(nextDetail);
+		const wrapper = mountPanel();
+		await flushPromises();
+
+		emitPush();
+		await flushPromises();
+		await wrapper.setProps({ threadId: 't2' });
+		await flushPromises();
+		emitPush();
+		resolveStaleRefresh(staleDetail);
+		await flushPromises();
+
+		expect(getThreadDetail).toHaveBeenCalledTimes(3);
+		expect(getThreadDetail).toHaveBeenLastCalledWith('p1', 'a1', 't2');
+		expect(wrapper.emitted('loaded')?.at(-1)).toEqual([nextDetail]);
+	});
+
+	it('disconnects and ignores invalidations after unmount', async () => {
+		const wrapper = mountPanel();
+		await flushPromises();
+
+		wrapper.unmount();
+		emitPush();
+		await flushPromises();
+
+		expect(pushStore.pushConnect).toHaveBeenCalledOnce();
+		expect(pushStore.pushDisconnect).toHaveBeenCalledOnce();
+		expect(pushHandlers.size).toBe(0);
+		expect(getThreadDetail).toHaveBeenCalledOnce();
+	});
+
+	it('keeps the selected timeline item through reconciliation', async () => {
+		const execution = {
+			id: 'exec-1',
+			status: 'running',
+			createdAt: '2026-08-04T07:00:00.000Z',
+			startedAt: '2026-08-04T07:00:00.000Z',
+			stoppedAt: null,
+			duration: 0,
+			userMessage: 'Question',
+			timeline: [{ type: 'text', content: 'Answer', timestamp: 1 }],
+		};
+		const runningDetail = { ...detail, executions: [execution] } as unknown as ThreadDetail;
+		const refreshedDetail = {
+			...detail,
+			executions: [
+				{
+					...execution,
+					timeline: [
+						{ type: 'text', content: 'Earlier update', timestamp: 0 },
+						{ type: 'text', content: 'Answer', timestamp: 1 },
+					],
+				},
+			],
+		} as unknown as ThreadDetail;
+		getThreadDetail.mockResolvedValueOnce(runningDetail).mockResolvedValueOnce(refreshedDetail);
+		const wrapper = mountPanel();
+		await flushPromises();
+		const table = wrapper.findComponent({ name: 'SessionTimelineTable' });
+		table.vm.$emit('select', 1);
+		emitPush();
+		await flushPromises();
+
+		expect(wrapper.find('[data-test-id="table-stub"]').attributes('data-selected-index')).toBe('2');
+	});
 });

packages/frontend/editor-ui/src/features/agents/__tests__/session-timeline.utils.spec.ts

@@ -319,7 +319,7 @@ describe('flattenExecutionsToTimelineItems', () => {
 		});
 	});
 
-	it('maps a regular tool-call timeline event to kind:tool', () => {
+	it('maps a running tool-call without marking it failed', () => {
 		const items = flattenExecutionsToTimelineItems([
 			withTimeline([
 				{
@@ -328,10 +328,10 @@ describe('flattenExecutionsToTimelineItems', () => {
 					name: 'http',
 					toolCallId: 'tc-1',
 					input: { method: 'GET' },
-					output: { ok: true },
+					output: undefined,
 					startTime: 1000,
-					endTime: 1200,
-					success: true,
+					endTime: 0,
+					success: false,
 				},
 			]),
 		]);
@@ -339,9 +339,10 @@ describe('flattenExecutionsToTimelineItems', () => {
 		expect(tool).toMatchObject({
 			toolName: 'http',
 			timestamp: 1000,
-			endTimestamp: 1200,
+			endTimestamp: 1000,
 			toolInput: { method: 'GET' },
-			toolOutput: { ok: true },
+			toolOutput: undefined,
+			toolSuccess: undefined,
 		});
 		expect(tool?.workflowId).toBeUndefined();
 	});

packages/frontend/editor-ui/src/features/agents/components/AgentSessionTimelinePanel.vue

@@ -1,5 +1,7 @@
 <script lang="ts" setup>
+import type { PushMessage } from '@n8n/api-types';
 import { useToast } from '@n8n/composables/useToast';
+import { usePushConnectionStore } from '@/app/stores/pushConnection.store';
 import { useAgentSessionsStore } from '@/features/agents/agentSessions.store';
 import type {
 	AgentExecution,
@@ -24,8 +26,8 @@ import { shouldIgnoreCanvasShortcut } from '@/features/workflows/canvas/canvas.u
 import type { FilterOption, TimelineItem } from '@/features/agents/session-timeline.types';
 import { useI18n } from '@n8n/i18n';
 import { N8nIcon, N8nInput } from '@n8n/design-system';
-import { computed, ref, watch } from 'vue';
-import { useActiveElement, useEventListener } from '@vueuse/core';
+import { computed, onBeforeUnmount, onMounted, ref, watch } from 'vue';
+import { useActiveElement, useDocumentVisibility, useEventListener } from '@vueuse/core';
 
 const props = defineProps<{
 	projectId: string;
@@ -43,7 +45,9 @@ const emit = defineEmits<{
 const i18n = useI18n();
 const toast = useToast();
 const sessionsStore = useAgentSessionsStore();
+const pushStore = usePushConnectionStore();
 const activeElement = useActiveElement();
+const documentVisibility = useDocumentVisibility();
 
 const projectId = computed(() => props.projectId);
 
@@ -53,7 +57,10 @@ const selectedIndex = ref<number | null>(null);
 const highlightedIndex = ref<number | null>(null);
 const selectedFilters = ref<Set<string>>(new Set());
 const searchQuery = ref('');
-let loadThreadDetailRequestId = 0;
+let threadDetailRequestId = 0;
+let refreshPending = false;
+let removePushListener: (() => void) | undefined;
+let activeRequest: { identity: string; promise: Promise<void> } | undefined;
 
 const baseItems = computed<TimelineItem[]>(() =>
 	flattenExecutionsToTimelineItems(executions.value),
@@ -149,6 +156,10 @@ function selectTimelineItem(index: number | null) {
 	highlightedIndex.value = index;
 }
 
+function timelineItemKey(item: TimelineItem): string {
+	return `${item.executionId}:${item.kind}:${item.toolCallId ?? item.timestamp}`;
+}
+
 function onKeyDown(event: KeyboardEvent) {
 	if (activeElement.value && shouldIgnoreCanvasShortcut(activeElement.value)) return;
 
@@ -189,18 +200,27 @@ function onKeyUp(event: KeyboardEvent) {
 
 useEventListener(document, 'keyup', onKeyUp);
 
-async function loadThreadDetail() {
-	const currentProjectId = props.projectId;
-	const currentAgentId = props.agentId;
-	const currentThreadId = props.threadId;
-	const requestId = ++loadThreadDetailRequestId;
-
+function loadThreadDetail() {
 	executions.value = [];
 	selectedFilters.value = new Set();
 	searchQuery.value = '';
 	selectTimelineItem(null);
 	loading.value = true;
 	emit('loaded', null);
+	refreshPending = false;
+	startThreadDetailRequest(true);
+}
+
+function threadIdentity(): string {
+	return `${props.projectId}:${props.agentId}:${props.threadId}`;
+}
+
+async function fetchThreadDetail(initial: boolean) {
+	const currentProjectId = props.projectId;
+	const currentAgentId = props.agentId;
+	const currentThreadId = props.threadId;
+	const identity = threadIdentity();
+	const requestId = ++threadDetailRequestId;
 
 	try {
 		const result = await sessionsStore.getThreadDetail(
@@ -208,19 +228,80 @@ async function loadThreadDetail() {
 			currentAgentId,
 			currentThreadId,
 		);
-		if (requestId !== loadThreadDetailRequestId) return;
+		if (requestId !== threadDetailRequestId || identity !== threadIdentity()) {
+			return;
+		}
+		const selectedKey = !initial && selectedItem.value ? timelineItemKey(selectedItem.value) : null;
 		executions.value = result.executions;
+		if (selectedKey) {
+			const nextIndex = items.value.findIndex((item) => timelineItemKey(item) === selectedKey);
+			selectTimelineItem(nextIndex >= 0 ? nextIndex : null);
+		}
 		emit('loaded', result);
 	} catch (error) {
-		if (requestId !== loadThreadDetailRequestId) return;
-		toast.showError(error, i18n.baseText('agentSessions.showError.load'));
+		if (requestId !== threadDetailRequestId) return;
+		if (initial) toast.showError(error, i18n.baseText('agentSessions.showError.load'));
 	} finally {
-		if (requestId === loadThreadDetailRequestId) {
-			loading.value = false;
-		}
+		if (initial && requestId === threadDetailRequestId) loading.value = false;
 	}
 }
 
+function startThreadDetailRequest(initial: boolean) {
+	const identity = threadIdentity();
+	const request = { identity, promise: fetchThreadDetail(initial) };
+	activeRequest = request;
+	void request.promise.finally(() => {
+		if (activeRequest !== request) return;
+		activeRequest = undefined;
+		if (refreshPending && identity === threadIdentity()) {
+			refreshPending = false;
+			refreshThreadDetail();
+		}
+	});
+}
+
+function refreshThreadDetail() {
+	if (activeRequest?.identity === threadIdentity()) {
+		refreshPending = true;
+		return;
+	}
+	startThreadDetailRequest(false);
+}
+
+function onPushMessage(event: PushMessage) {
+	if (
+		event.type === 'agentExecutionUpdated' &&
+		event.data.projectId === props.projectId &&
+		event.data.agentId === props.agentId &&
+		event.data.threadId === props.threadId
+	) {
+		refreshThreadDetail();
+	}
+}
+
+watch(documentVisibility, (visibility) => {
+	if (visibility === 'visible') refreshThreadDetail();
+});
+
+watch(
+	() => pushStore.isConnected,
+	(isConnected, wasConnected) => {
+		if (isConnected && !wasConnected) refreshThreadDetail();
+	},
+);
+
+onMounted(() => {
+	pushStore.pushConnect();
+	removePushListener = pushStore.addEventListener(onPushMessage);
+});
+
+onBeforeUnmount(() => {
+	threadDetailRequestId++;
+	refreshPending = false;
+	removePushListener?.();
+	pushStore.pushDisconnect();
+});
+
 watch([() => props.projectId, () => props.agentId, () => props.threadId], loadThreadDetail, {
 	immediate: true,
 });

packages/frontend/editor-ui/src/features/agents/composables/useAgentThreadsApi.ts

@@ -24,7 +24,7 @@ export interface AgentExecutionThread {
 	source?: string | null;
 }
 
-export type AgentExecutionStatus = 'success' | 'error';
+export type AgentExecutionStatus = 'running' | 'success' | 'error' | 'cancelled' | 'interrupted';
 export type AgentExecutionHitlStatus = 'suspended' | 'resumed';
 
 /**

packages/frontend/editor-ui/src/features/agents/session-timeline.utils.ts

@@ -267,7 +267,7 @@ export function flattenExecutionsToTimelineItems(executions: AgentExecution[]):
 					toolCallId: event.toolCallId,
 					toolInput: event.input,
 					toolOutput: event.output,
-					toolSuccess: event.success,
+					toolSuccess: event.endTime === 0 ? undefined : event.success,
 					timestamp: event.startTime,
 					endTimestamp: event.endTime || event.startTime,
 					workflowId: isWorkflow ? event.workflowId : undefined,