feat(core): Add atomic thread metadata patches to Agents memory (no-changelog) (#40610)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

packages/cli/src/modules/agents/__tests__/agent-thread.repository.integration.test.ts

@@ -0,0 +1,96 @@
+import { testDb, testModules } from '@n8n/backend-test-utils';
+import { GlobalConfig } from '@n8n/config';
+import { Container } from '@n8n/di';
+import { jsonParse } from 'n8n-workflow';
+
+import { AgentThreadRepository } from '../repositories/agent-thread.repository';
+
+describe('AgentThreadRepository', () => {
+	let repository: AgentThreadRepository;
+
+	beforeAll(async () => {
+		// The concurrent patch test needs real parallel transactions on Postgres.
+		Container.get(GlobalConfig).database.postgresdb.poolSize = 4;
+		await testModules.loadModules(['agents']);
+		await testDb.init();
+		repository = Container.get(AgentThreadRepository);
+	});
+
+	beforeEach(async () => {
+		await repository.delete({});
+	});
+
+	afterAll(async () => {
+		await testDb.terminate();
+	});
+
+	async function seedThread(metadata: Record<string, unknown>) {
+		return await repository.save(
+			repository.create({
+				id: 'thread-1',
+				resourceId: 'user-1',
+				title: 'Chat',
+				metadata: JSON.stringify(metadata),
+			}),
+		);
+	}
+
+	async function readMetadata(): Promise<unknown> {
+		const row = await repository.findOneByOrFail({ id: 'thread-1' });
+		return jsonParse(row.metadata ?? 'null');
+	}
+
+	describe('patchThread', () => {
+		it('replaces the metadata, so a patch can remove keys', async () => {
+			await seedThread({ keep: 1, drop: 2 });
+
+			const saved = await repository.patchThread(
+				'thread-1',
+				() => ({ title: 'Renamed', metadata: JSON.stringify({ keep: 1 }) }),
+				{},
+			);
+
+			expect(saved?.title).toBe('Renamed');
+			expect(await readMetadata()).toEqual({ keep: 1 });
+		});
+
+		it('writes nothing when the update returns null', async () => {
+			const seeded = await seedThread({ count: 1 });
+
+			const result = await repository.patchThread('thread-1', () => null, {});
+
+			expect(result).toMatchObject({ id: 'thread-1', title: 'Chat' });
+			const row = await repository.findOneByOrFail({ id: 'thread-1' });
+			expect(row.updatedAt.getTime()).toBe(seeded.updatedAt.getTime());
+			expect(await readMetadata()).toEqual({ count: 1 });
+		});
+
+		it('returns null for a missing thread without calling the update', async () => {
+			const update = vi.fn();
+
+			await expect(repository.patchThread('missing', update, {})).resolves.toBeNull();
+			expect(update).not.toHaveBeenCalled();
+		});
+
+		it('keeps every update when patches run at the same time', async () => {
+			await seedThread({ count: 0 });
+
+			// Two mains can patch one thread at the same time. On Postgres the row
+			// lock makes each patch wait for the previous commit. SQLite
+			// serializes the two transactions through the write connection.
+			const increment = async () =>
+				await repository.patchThread(
+					'thread-1',
+					(row) => {
+						const metadata = jsonParse<{ count: number }>(row.metadata ?? '{}');
+						return { metadata: JSON.stringify({ count: metadata.count + 1 }) };
+					},
+					{},
+				);
+
+			await Promise.all(Array.from({ length: 10 }, increment));
+
+			expect(await readMetadata()).toEqual({ count: 10 });
+		});
+	});
+});

packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.integration.test.ts

@@ -0,0 +1,79 @@
+import { testDb, testModules } from '@n8n/backend-test-utils';
+import { GlobalConfig } from '@n8n/config';
+import { Container } from '@n8n/di';
+import { jsonParse } from 'n8n-workflow';
+
+import { AgentThreadRepository } from '../../repositories/agent-thread.repository';
+import { N8nMemory, type N8nMemoryImpl } from '../n8n-memory';
+
+describe('N8nMemoryImpl', () => {
+	let repository: AgentThreadRepository;
+	let memory: N8nMemoryImpl;
+
+	beforeAll(async () => {
+		// The concurrent tests need real parallel transactions on Postgres.
+		Container.get(GlobalConfig).database.postgresdb.poolSize = 4;
+		await testModules.loadModules(['agents']);
+		await testDb.init();
+		repository = Container.get(AgentThreadRepository);
+		memory = Container.get(N8nMemory).getImplementation('agent-1');
+	});
+
+	beforeEach(async () => {
+		await repository.delete({});
+	});
+
+	afterAll(async () => {
+		await testDb.terminate();
+	});
+
+	async function readMetadata(): Promise<unknown> {
+		const row = await repository.findOneByOrFail({ id: 'thread-1' });
+		return jsonParse(row.metadata ?? 'null');
+	}
+
+	describe('saveThread', () => {
+		it('keeps every write when saves and patches run at the same time', async () => {
+			await memory.saveThread({ id: 'thread-1', resourceId: 'user-1', metadata: { count: 0 } });
+
+			const patch = async () =>
+				await memory.patchThread({
+					threadId: 'thread-1',
+					update: ({ metadata }) => ({
+						metadata: { ...metadata, count: (metadata?.count as number) + 1 },
+					}),
+				});
+			const touch = async () => await memory.saveThread({ id: 'thread-1', resourceId: 'user-1' });
+			const save = async (i: number) =>
+				await memory.saveThread({
+					id: 'thread-1',
+					resourceId: 'user-1',
+					metadata: { [`key${i}`]: true },
+				});
+
+			await Promise.all(
+				Array.from({ length: 10 }, async (_, i) => await Promise.all([patch(), touch(), save(i)])),
+			);
+
+			expect(await readMetadata()).toEqual({
+				count: 10,
+				...Object.fromEntries(Array.from({ length: 10 }, (_, i) => [`key${i}`, true])),
+			});
+		});
+
+		it('writes nothing when the save carries no title or metadata', async () => {
+			const created = await memory.saveThread({
+				id: 'thread-1',
+				resourceId: 'user-1',
+				title: 'Chat',
+				metadata: { keep: 1 },
+			});
+
+			const saved = await memory.saveThread({ id: 'thread-1', resourceId: 'user-2' });
+
+			expect(saved).toMatchObject({ resourceId: 'user-1', title: 'Chat', metadata: { keep: 1 } });
+			const row = await repository.findOneByOrFail({ id: 'thread-1' });
+			expect(row.updatedAt.getTime()).toBe(created.updatedAt.getTime());
+		});
+	});
+});

packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.test.ts

@@ -1,5 +1,7 @@
 import { hashEpisodicMemoryEvidence, type NewObservationLogEntry } from '@n8n/agents';
+import type { OperationContext, Transaction } from '@n8n/db';
 import { In, IsNull, Like } from '@n8n/typeorm';
+import { jsonParse } from 'n8n-workflow';
 import type { Mock, Mocked } from 'vitest';
 import { mock } from 'vitest-mock-extended';
 
@@ -282,42 +284,42 @@ describe('N8nMemory', () => {
 	});
 
 	describe('saveThread — existing row', () => {
+		/** Applies the update to `existing` the way the locked repository write does. */
+		function lockedRow(existing: Partial<AgentThreadEntity>) {
+			const row = { id: 'thread-1', resourceId: 'original-user', ...existing } as AgentThreadEntity;
+			threadRepository.patchThread.mockImplementation(async (threadId, update) => {
+				if (threadId !== row.id) return null;
+				const patch = update(row);
+				if (!patch) return row;
+				if (patch.title !== undefined) row.title = patch.title;
+				if (patch.metadata !== undefined) row.metadata = patch.metadata;
+				return row;
+			});
+			return row;
+		}
+
 		it('preserves the original resourceId instead of overwriting with the caller’s', async () => {
 			// Shared threads (e.g. the test-chat thread keyed by agentId) are
 			// written to by multiple users. The first writer owns the row;
 			// subsequent saves must not re-stamp it with the current caller.
-			const existing = {
-				id: 'thread-1',
-				resourceId: 'original-user',
-				title: null,
-				metadata: null,
-			} as unknown as AgentThreadEntity;
-			threadRepository.findOneBy.mockResolvedValue(existing);
-			threadRepository.save.mockImplementation(async (e) => e as AgentThreadEntity);
+			const row = lockedRow({ title: null, metadata: null });
 
-			await memory.saveThread({
+			const saved = await memory.saveThread({
 				id: 'thread-1',
 				resourceId: 'different-user',
-				title: undefined,
+				title: 'Renamed',
 				metadata: undefined,
 			});
 
-			expect(threadRepository.save).toHaveBeenCalledWith(
-				expect.objectContaining({ id: 'thread-1', resourceId: 'original-user' }),
-			);
+			expect(row.resourceId).toBe('original-user');
+			expect(saved.resourceId).toBe('original-user');
+			expect(threadRepository.save).not.toHaveBeenCalled();
 		});
 
 		it('still ensures the caller’s resource row exists (for message-level writes)', async () => {
 			// resourceId on the thread is preserved, but messages saved afterwards
 			// carry the current user's resourceId — that row must exist.
-			const existing = {
-				id: 'thread-1',
-				resourceId: 'original-user',
-				title: null,
-				metadata: null,
-			} as unknown as AgentThreadEntity;
-			threadRepository.findOneBy.mockResolvedValue(existing);
-			threadRepository.save.mockImplementation(async (e) => e as AgentThreadEntity);
+			lockedRow({ title: null, metadata: null });
 
 			await memory.saveThread({
 				id: 'thread-1',
@@ -333,21 +335,14 @@ describe('N8nMemory', () => {
 			expect(resourceInsertQueryBuilder.orIgnore).toHaveBeenCalled();
 		});
 
-		it('merges metadata updates instead of replacing existing thread metadata', async () => {
+		it('merges metadata updates into the locked row instead of replacing it', async () => {
 			const currentMessageContext = {
 				integrationConnectionId: 'slack:cred-1',
 				platform: 'slack',
 				target: { type: 'thread', threadId: 'thread-1' },
 				updatedAt: '2026-05-18T10:00:00.000Z',
 			};
-			const existing = {
-				id: 'thread-1',
-				resourceId: 'original-user',
-				title: null,
-				metadata: JSON.stringify({ currentMessageContext }),
-			} as unknown as AgentThreadEntity;
-			threadRepository.findOneBy.mockResolvedValue(existing);
-			threadRepository.save.mockImplementation(async (e) => e as AgentThreadEntity);
+			const row = lockedRow({ title: null, metadata: JSON.stringify({ currentMessageContext }) });
 
 			await memory.saveThread({
 				id: 'thread-1',
@@ -356,15 +351,144 @@ describe('N8nMemory', () => {
 				metadata: { summary: 'Support thread' },
 			});
 
-			expect(threadRepository.save).toHaveBeenCalledWith(
-				expect.objectContaining({ metadata: expect.any(String) }),
-			);
-			const savedThread = threadRepository.save.mock.calls[0][0] as AgentThreadEntity;
-			expect(JSON.parse(savedThread.metadata ?? '{}')).toEqual({
+			expect(JSON.parse(row.metadata ?? '{}')).toEqual({
 				currentMessageContext,
 				summary: 'Support thread',
 			});
 		});
+
+		it('writes nothing when the save carries no title or metadata', async () => {
+			lockedRow({ title: 'Chat', metadata: null });
+
+			await memory.saveThread({ id: 'thread-1', resourceId: 'original-user' });
+
+			const update = threadRepository.patchThread.mock.calls[0][1];
+			expect(update({ title: 'Chat', metadata: '{"a":1}' } as AgentThreadEntity)).toBeNull();
+		});
+	});
+
+	describe('saveThread — new row', () => {
+		it('creates the thread when no row exists', async () => {
+			threadRepository.patchThread.mockResolvedValue(null);
+			threadRepository.create.mockImplementation((e) => e as AgentThreadEntity);
+			threadRepository.save.mockImplementation(async (e) => e as AgentThreadEntity);
+
+			await memory.saveThread({
+				id: 'thread-1',
+				resourceId: 'user-1',
+				title: 'Chat',
+				metadata: { keep: 1 },
+			});
+
+			expect(threadRepository.save).toHaveBeenCalledWith(
+				expect.objectContaining({
+					id: 'thread-1',
+					resourceId: 'user-1',
+					title: 'Chat',
+					metadata: JSON.stringify({ keep: 1 }),
+				}),
+			);
+		});
+	});
+
+	describe('patchThread', () => {
+		let writes: number;
+
+		/** A row store behind the repository double that counts writes. */
+		function seedThread(metadata: Record<string, unknown> | null) {
+			writes = 0;
+			const row = {
+				id: 'thread-1',
+				resourceId: 'user-1',
+				title: 'Chat',
+				metadata: metadata ? JSON.stringify(metadata) : null,
+			} as AgentThreadEntity;
+			threadRepository.patchThread.mockImplementation(async (threadId, update) => {
+				if (threadId !== row.id) return null;
+				const patch = update(row);
+				if (!patch) return { ...row };
+				if (patch.title !== undefined) row.title = patch.title;
+				if (patch.metadata !== undefined) row.metadata = patch.metadata;
+				writes++;
+				return { ...row } as AgentThreadEntity;
+			});
+			return row;
+		}
+
+		it('replaces the metadata, so a patch can remove keys', async () => {
+			const row = seedThread({ keep: 1, drop: 2 });
+
+			const updated = await memory.patchThread({
+				threadId: 'thread-1',
+				update: ({ metadata }) => {
+					const { drop: _drop, ...rest } = metadata ?? {};
+					return { title: 'Renamed', metadata: { ...rest, added: true } };
+				},
+			});
+
+			expect(jsonParse(row.metadata ?? '{}')).toEqual({ keep: 1, added: true });
+			expect(row.title).toBe('Renamed');
+			expect(updated).toMatchObject({
+				id: 'thread-1',
+				title: 'Renamed',
+				metadata: { keep: 1, added: true },
+			});
+		});
+
+		it('keeps the title when the patch only sets metadata', async () => {
+			const row = seedThread({ count: 1 });
+
+			await memory.patchThread({
+				threadId: 'thread-1',
+				update: () => ({ metadata: {} }),
+			});
+
+			expect(row.title).toBe('Chat');
+			expect(row.metadata).toBe('{}');
+		});
+
+		it('returns the current thread without a write when the update returns null', async () => {
+			seedThread({ count: 1 });
+
+			const result = await memory.patchThread({ threadId: 'thread-1', update: () => null });
+
+			expect(result).toMatchObject({ id: 'thread-1', title: 'Chat', metadata: { count: 1 } });
+			expect(writes).toBe(0);
+		});
+
+		it('returns null for a missing thread without calling the update', async () => {
+			seedThread({});
+			const update = vi.fn();
+
+			expect(await memory.patchThread({ threadId: 'missing', update })).toBeNull();
+			expect(update).not.toHaveBeenCalled();
+			expect(writes).toBe(0);
+		});
+
+		it('passes a root operation context to the repository by default', async () => {
+			seedThread({});
+
+			await memory.patchThread({ threadId: 'thread-1', update: () => null });
+
+			expect(threadRepository.patchThread).toHaveBeenCalledWith(
+				'thread-1',
+				expect.any(Function),
+				{},
+			);
+		});
+
+		it("passes the caller's operation context to the repository", async () => {
+			seedThread({});
+			const ctx: OperationContext = { trx: mock<Transaction>() };
+
+			await memory.patchThread({ threadId: 'thread-1', update: () => null }, ctx);
+
+			expect(threadRepository.patchThread).toHaveBeenCalledWith(
+				'thread-1',
+				expect.any(Function),
+				ctx,
+			);
+		});
 	});
 
 	describe('deleteThread', () => {

packages/cli/src/modules/agents/integrations/n8n-memory.ts

@@ -165,22 +165,29 @@ export class N8nMemoryImpl
 	async saveThread(thread: Omit<Thread, 'createdAt' | 'updatedAt'>): Promise<Thread> {
 		await this.resourceRepository.ensureExists(thread.resourceId);
 
-		const existing = await this.threadRepository.findOneBy({ id: thread.id });
-
-		if (existing) {
-			// `resourceId` is treated as immutable on existing threads. Some thread
-			// IDs can receive messages from more than one resource; overwriting the
-			// column on each save would make ownership depend on the last writer.
-			// Per-user scoping is enforced at the message level via resourceId.
-			if (thread.title !== undefined) existing.title = thread.title;
-			if (thread.metadata !== undefined) {
-				// Thread metadata carries runtime-owned keys such as integration
-				// message context, so caller updates merge into the existing blob.
-				existing.metadata = this.mergeThreadMetadata(existing.metadata, thread.metadata);
-			}
-			const saved = await this.threadRepository.save(existing);
-			return this.toThread(saved);
-		}
+		// `resourceId` is treated as immutable on existing threads. Some thread
+		// IDs can receive messages from more than one resource; overwriting the
+		// column on each save would make ownership depend on the last writer.
+		// Per-user scoping is enforced at the message level via resourceId.
+		// The merge runs on the locked row, so a save cannot revert a concurrent
+		// `patchThread`.
+		const updated = await this.threadRepository.patchThread(
+			thread.id,
+			(row) => {
+				if (thread.title === undefined && thread.metadata === undefined) return null;
+				return {
+					title: thread.title,
+					// Thread metadata carries runtime-owned keys such as integration
+					// message context, so caller updates merge into the existing blob.
+					metadata:
+						thread.metadata === undefined
+							? undefined
+							: this.mergeThreadMetadata(row.metadata, thread.metadata),
+				};
+			},
+			{},
+		);
+		if (updated) return this.toThread(updated);
 
 		const entity = this.threadRepository.create({
 			id: thread.id,
@@ -192,6 +199,38 @@ export class N8nMemoryImpl
 		return this.toThread(saved);
 	}
 
+	/**
+	 * Read, update and write one thread in one transaction. Unlike `saveThread`,
+	 * the update replaces the metadata, so a caller can remove keys. Concurrent
+	 * patches do not lose updates: see `AgentThreadRepository.patchThread`.
+	 *
+	 * When `update` returns `null` or `undefined`, nothing is written and the
+	 * current thread is returned. Returns `null` when the thread does not exist.
+	 */
+	async patchThread(
+		args: {
+			threadId: string;
+			update: (
+				current: Thread,
+			) => { title?: string; metadata?: Record<string, unknown> } | null | undefined;
+		},
+		ctx: OperationContext = {},
+	): Promise<Thread | null> {
+		const row = await this.threadRepository.patchThread(
+			args.threadId,
+			(locked) => {
+				const patch = args.update(this.toThread(locked));
+				if (!patch) return null;
+				return {
+					title: patch.title,
+					metadata: patch.metadata === undefined ? undefined : JSON.stringify(patch.metadata),
+				};
+			},
+			ctx,
+		);
+		return row ? this.toThread(row) : null;
+	}
+
 	async deleteThread(threadId: string, ctx: OperationContext = {}): Promise<void> {
 		await this.threadRepository.runInTransaction(ctx, async (trx) => {
 			await this.dropEpisodicEntriesWithoutSources(trx, threadId);

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

@@ -1,12 +1,54 @@
 import { BaseRepository, TransactionRunner } from '@n8n/db';
+import type { OperationContext } from '@n8n/db';
 import { Service } from '@n8n/di';
 import { DataSource } from '@n8n/typeorm';
 
 import { AgentThreadEntity } from '../entities/agent-thread.entity';
 
+/** Column values that a thread patch can write. `undefined` keeps the stored value. */
+export interface AgentThreadRowPatch {
+	title?: string | null;
+	metadata?: string | null;
+}
+
 @Service()
 export class AgentThreadRepository extends BaseRepository<AgentThreadEntity> {
 	constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
 		super(AgentThreadEntity, dataSource.manager, transactionRunner);
 	}
+
+	/**
+	 * Read, update and write the title and metadata of one thread in one
+	 * transaction. On Postgres the row stays locked until the write, so
+	 * concurrent patches from different mains cannot lose an update. On SQLite
+	 * (single main only) each transaction holds the single write connection
+	 * until it commits, so patches already run one after another.
+	 *
+	 * `update` gets the locked row and returns the columns to write. When it
+	 * returns `null` or `undefined`, nothing is written and the row is returned
+	 * as it is. Returns `null` when the thread does not exist.
+	 */
+	async patchThread(
+		threadId: string,
+		update: (row: Readonly<AgentThreadEntity>) => AgentThreadRowPatch | null | undefined,
+		ctx: OperationContext,
+	): Promise<AgentThreadEntity | null> {
+		return await this.runInTransaction(ctx, async (manager) => {
+			const row = await manager.findOne(AgentThreadEntity, {
+				where: { id: threadId },
+				lock:
+					manager.connection.options.type === 'postgres'
+						? { mode: 'pessimistic_write' }
+						: undefined,
+			});
+			if (!row) return null;
+
+			const patch = update(row);
+			if (!patch) return row;
+
+			if (patch.title !== undefined) row.title = patch.title;
+			if (patch.metadata !== undefined) row.metadata = patch.metadata;
+			return await manager.save(AgentThreadEntity, row);
+		});
+	}
 }