From 55def8cc2beef094a32100d18beddc4770650603 Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 9 Oct 2026 13:41:05 +0000 Subject: [PATCH] feat(core): Add atomic thread metadata patches to Agents memory (no-changelog) (#40610) Co-authored-by: Claude Opus 5.5 --- ...gent-thread.repository.integration.test.ts | 96 +++++++++ .../__tests__/n8n-memory.integration.test.ts | 79 +++++++ .../integrations/__tests__/n8n-memory.test.ts | 194 ++++++++++++++---- .../modules/agents/integrations/n8n-memory.ts | 71 +++++-- .../repositories/agent-thread.repository.ts | 42 ++++ 5 files changed, 431 insertions(+), 51 deletions(-) create mode 100644 packages/cli/src/modules/agents/__tests__/agent-thread.repository.integration.test.ts create mode 100644 packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.integration.test.ts diff --git a/packages/cli/src/modules/agents/__tests__/agent-thread.repository.integration.test.ts b/packages/cli/src/modules/agents/__tests__/agent-thread.repository.integration.test.ts new file mode 100644 index 00000000000..4f24687e9e9 --- /dev/null +++ b/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) { + return await repository.save( + repository.create({ + id: 'thread-1', + resourceId: 'user-1', + title: 'Chat', + metadata: JSON.stringify(metadata), + }), + ); + } + + async function readMetadata(): Promise { + 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 }); + }); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.integration.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.integration.test.ts new file mode 100644 index 00000000000..02ffeb90fda --- /dev/null +++ b/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 { + 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()); + }); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.test.ts index 31592cdee51..151d36f8bf3 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/n8n-memory.test.ts +++ b/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) { + 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 | 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() }; + + await memory.patchThread({ threadId: 'thread-1', update: () => null }, ctx); + + expect(threadRepository.patchThread).toHaveBeenCalledWith( + 'thread-1', + expect.any(Function), + ctx, + ); + }); }); describe('deleteThread', () => { diff --git a/packages/cli/src/modules/agents/integrations/n8n-memory.ts b/packages/cli/src/modules/agents/integrations/n8n-memory.ts index 71bb2292185..a4896311518 100644 --- a/packages/cli/src/modules/agents/integrations/n8n-memory.ts +++ b/packages/cli/src/modules/agents/integrations/n8n-memory.ts @@ -165,22 +165,29 @@ export class N8nMemoryImpl async saveThread(thread: Omit): Promise { 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 } | null | undefined; + }, + ctx: OperationContext = {}, + ): Promise { + 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 { await this.threadRepository.runInTransaction(ctx, async (trx) => { await this.dropEpisodicEntriesWithoutSources(trx, threadId); diff --git a/packages/cli/src/modules/agents/repositories/agent-thread.repository.ts b/packages/cli/src/modules/agents/repositories/agent-thread.repository.ts index 61f746bfde4..48d8f866e27 100644 --- a/packages/cli/src/modules/agents/repositories/agent-thread.repository.ts +++ b/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 { 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) => AgentThreadRowPatch | null | undefined, + ctx: OperationContext, + ): Promise { + 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); + }); + } }