feat(core): Add owned-session queries to the Agents thread repositories (no-changelog) (#40615)

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

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

@@ -2,9 +2,9 @@ import type { AgentSessionOrigin, AgentSessionQueryFilters } from '@n8n/api-type
 import type { SerializableAgentState } from '@n8n/agents';
 import {
 	BaseRepository,
-	TransactionRunner,
 	escapeLike,
 	LIKE_ESCAPE_CLAUSE,
+	TransactionRunner,
 	type OperationContext,
 } from '@n8n/db';
 import { Service } from '@n8n/di';
@@ -41,6 +41,27 @@ export interface AgentExecutionThreadPage {
 	nextCursor: string | null;
 }
 
+/** Position of the last thread on a keyset page: `updatedAt`, then `id` breaks ties. */
+export interface AgentExecutionThreadKeyset {
+	updatedAt: Date;
+	id: string;
+}
+
+export interface AgentExecutionThreadKeysetPage {
+	threads: AgentExecutionThread[];
+	hasMore: boolean;
+}
+
+function ownedSessionWhere(agentId: string, ownerId: string, threadId: string) {
+	return {
+		id: threadId,
+		agentId,
+		ownerId,
+		accessScope: 'user' as const,
+		parentThreadId: IsNull(),
+	};
+}
+
 interface AgentSessionDeletionRefs {
 	attachmentBinaryDataIds: string[];
 	executionLogs: Array<Pick<AgentExecution, 'id' | 'storedAt'>>;
@@ -404,6 +425,125 @@ export class AgentExecutionThreadRepository extends BaseRepository<AgentExecutio
 		}
 	}
 
+	/** Top-level private sessions of one user with one agent, newest first. */
+	async findOwnedByAgent(
+		agentId: string,
+		ownerId: string,
+		options: { limit?: number } = {},
+	): Promise<AgentExecutionThread[]> {
+		return await this.find({
+			where: { agentId, ownerId, accessScope: 'user', parentThreadId: IsNull() },
+			order: { updatedAt: 'DESC', id: 'DESC' },
+			...(options.limit ? { take: options.limit } : {}),
+		});
+	}
+
+	/** One top-level private session of a user with an agent. */
+	async findOwnedById(
+		agentId: string,
+		ownerId: string,
+		threadId: string,
+	): Promise<AgentExecutionThread | null> {
+		return await this.findOneBy(ownedSessionWhere(agentId, ownerId, threadId));
+	}
+
+	/**
+	 * Rename a top-level private session of a user with an agent. Returns false
+	 * when no such session exists.
+	 */
+	async renameOwned(
+		agentId: string,
+		ownerId: string,
+		threadId: string,
+		title: string,
+		ctx: OperationContext,
+	): Promise<boolean> {
+		const result = await this.managerFor(ctx).update(
+			AgentExecutionThread,
+			ownedSessionWhere(agentId, ownerId, threadId),
+			{ title },
+		);
+		return result.affected === 1;
+	}
+
+	/**
+	 * Move a top-level private session of a user with an agent to another
+	 * project. Returns false when no such session exists. The caller must check
+	 * that the user can use the target project.
+	 */
+	async moveOwned(
+		agentId: string,
+		ownerId: string,
+		threadId: string,
+		projectId: string,
+		ctx: OperationContext,
+	): Promise<boolean> {
+		const result = await this.managerFor(ctx).update(
+			AgentExecutionThread,
+			ownedSessionWhere(agentId, ownerId, threadId),
+			{ projectId },
+		);
+		return result.affected === 1;
+	}
+
+	/**
+	 * One page of the top-level private sessions of a user with an agent,
+	 * newest first. `before` is the last session of the previous page. The
+	 * `id` breaks ties between sessions with the same `updatedAt`, so no
+	 * session is skipped or repeated across pages.
+	 */
+	async findOwnedHistoryPage(
+		agentId: string,
+		ownerId: string,
+		limit: number,
+		search?: string,
+		before?: AgentExecutionThreadKeyset,
+	): Promise<AgentExecutionThreadKeysetPage> {
+		const updatedAt = this.updatedAtExpression();
+		const query = this.createQueryBuilder('thread')
+			.where('thread.agentId = :agentId', { agentId })
+			.andWhere('thread.ownerId = :ownerId', { ownerId })
+			.andWhere("thread.accessScope = 'user'")
+			.andWhere('thread.parentThreadId IS NULL');
+		const term = search?.trim().toLowerCase();
+		if (term) {
+			query.andWhere(`LOWER(thread.title) LIKE :search ${LIKE_ESCAPE_CLAUSE}`, {
+				search: `%${escapeLike(term)}%`,
+			});
+		}
+		if (before) {
+			query.andWhere(
+				`(${updatedAt} < :beforeAt OR (${updatedAt} = :beforeAt AND thread.id < :beforeId))`,
+				{ beforeAt: before.updatedAt, beforeId: before.id },
+			);
+		}
+		// Order by the same expression as the keyset, so ties stored in another format stay together.
+		const threads = await query
+			.orderBy(updatedAt, 'DESC')
+			.addOrderBy('thread.id', 'DESC')
+			.limit(limit + 1)
+			.getMany();
+		const hasMore = threads.length > limit;
+		if (hasMore) threads.pop();
+		return { threads, hasMore };
+	}
+
+	/** Sessions of an agent last updated before `cutoff`, oldest first, for pruning. */
+	async findByAgentUpdatedBefore(
+		agentId: string,
+		cutoff: Date,
+		limit: number,
+	): Promise<AgentExecutionThread[]> {
+		const updatedAt = this.updatedAtExpression();
+		return await this.createQueryBuilder('thread')
+			.where('thread.agentId = :agentId', { agentId })
+			.andWhere(`${updatedAt} < :cutoff`, { cutoff })
+			.orderBy(updatedAt, 'ASC')
+			.addOrderBy('thread.id', 'ASC')
+			.limit(limit)
+			.getMany();
+	}
+
 	async findByParentThreadId(
 		parentThreadId: string,
 		projectId: string,

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

@@ -42,6 +42,15 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
 		});
 	}
 
+	/**
+	 * Whether the thread has a queued or claimed message, hidden machine turns
+	 * included. A hidden follow-up turn waits here between two executions, so
+	 * the thread still has work when no execution runs.
+	 */
+	async hasItems(threadId: string, ctx: OperationContext = {}): Promise<boolean> {
+		return await this.managerFor(ctx).existsBy(AgentMessageQueue, { threadId });
+	}
+
 	async findItem(threadId: string, id: string, ctx: OperationContext) {
 		return await this.managerFor(ctx).findOne(AgentMessageQueue, {
 			where: { threadId, id },

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

@@ -14,6 +14,10 @@ export class AgentThreadGrantRepository extends BaseRepository<AgentThreadGrant>
 		await this.createQueryBuilder().insert().values({ threadId, grantKey }).orIgnore().execute();
 	}
 
+	async revoke(threadId: string, grantKey: string): Promise<void> {
+		await this.delete({ threadId, grantKey });
+	}
+
 	async findKeys(threadId: string): Promise<Set<string>> {
 		const grants = await this.find({ where: { threadId }, select: ['grantKey'] });
 		return new Set(grants.map(({ grantKey }) => grantKey));

packages/cli/test/integration/agents/agent-thread-repository-queries.test.ts

@@ -0,0 +1,381 @@
+import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils';
+import { Container } from '@n8n/di';
+import { v4 as uuid } from 'uuid';
+
+import type { Agent } from '@/modules/agents/entities/agent.entity';
+import type { AgentExecutionThread } from '@/modules/agents/entities/agent-execution-thread.entity';
+import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository';
+import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository';
+import { AgentMessageQueueRepository } from '@/modules/agents/repositories/agent-message-queue.repository';
+import { AgentMessageRepository } from '@/modules/agents/repositories/agent-message.repository';
+import { AgentRepository } from '@/modules/agents/repositories/agent.repository';
+import { AgentThreadGrantRepository } from '@/modules/agents/repositories/agent-thread-grant.repository';
+
+import { createMember } from '../shared/db/users';
+
+describe('Agent thread repository queries for system agents', () => {
+	let threadRepo: AgentExecutionThreadRepository;
+	let agentRepo: AgentRepository;
+	let grantRepo: AgentThreadGrantRepository;
+	let queueRepo: AgentMessageQueueRepository;
+	let messageRepo: AgentMessageRepository;
+	let executionRepo: AgentExecutionRepository;
+	let projectId: string;
+	let agentId: string;
+	let otherAgentId: string;
+	let ownerId: string;
+	let otherUserId: string;
+
+	beforeAll(async () => {
+		await testModules.loadModules(['agents']);
+		await testDb.init();
+		threadRepo = Container.get(AgentExecutionThreadRepository);
+		agentRepo = Container.get(AgentRepository);
+		grantRepo = Container.get(AgentThreadGrantRepository);
+		queueRepo = Container.get(AgentMessageQueueRepository);
+		messageRepo = Container.get(AgentMessageRepository);
+		executionRepo = Container.get(AgentExecutionRepository);
+		ownerId = (await createMember()).id;
+		otherUserId = (await createMember()).id;
+	});
+
+	beforeEach(async () => {
+		projectId = (await createTeamProject()).id;
+		agentId = (await createAgent('Agent')).id;
+		otherAgentId = (await createAgent('Other agent')).id;
+	});
+
+	afterEach(async () => {
+		await queueRepo.delete({});
+		await executionRepo.delete({});
+		await threadRepo.delete({});
+		await agentRepo.delete({});
+	});
+
+	afterAll(async () => {
+		await testDb.terminate();
+	});
+
+	async function createAgent(name: string): Promise<Agent> {
+		return await agentRepo.save(
+			agentRepo.create({ id: uuid(), name, projectId, integrations: [], tools: {}, skills: {} }),
+		);
+	}
+
+	async function createThread(
+		overrides: Partial<AgentExecutionThread> = {},
+		updatedAt?: string,
+	): Promise<AgentExecutionThread> {
+		const thread = await threadRepo.save(
+			threadRepo.create({
+				id: uuid(),
+				agentId,
+				agentName: 'Agent',
+				projectId,
+				accessScope: 'user',
+				ownerId,
+				...overrides,
+			}),
+		);
+		if (updatedAt) {
+			await threadRepo.update({ id: thread.id }, { updatedAt: new Date(updatedAt) });
+			thread.updatedAt = new Date(updatedAt);
+		}
+		return thread;
+	}
+
+	const ids = (threads: AgentExecutionThread[]) => threads.map(({ id }) => id);
+
+	describe('findOwnedByAgent', () => {
+		it("returns the owner's top-level private sessions with the agent, newest first", async () => {
+			const older = await createThread({}, '2024-01-01T00:00:00.000Z');
+			const newer = await createThread({}, '2024-02-01T00:00:00.000Z');
+			await createThread({ parentThreadId: newer.id });
+			await createThread({ ownerId: otherUserId });
+			await createThread({ accessScope: 'project', ownerId: null });
+			await createThread({ agentId: otherAgentId });
+
+			expect(ids(await threadRepo.findOwnedByAgent(agentId, ownerId))).toEqual([
+				newer.id,
+				older.id,
+			]);
+		});
+
+		it('applies the limit and orders sessions with the same updatedAt by id', async () => {
+			const created = [];
+			for (let index = 0; index < 3; index++) {
+				created.push(await createThread({}, '2024-01-01T00:00:00.000Z'));
+			}
+			const expected = ids(created).sort().reverse();
+
+			expect(ids(await threadRepo.findOwnedByAgent(agentId, ownerId, { limit: 2 }))).toEqual(
+				expected.slice(0, 2),
+			);
+		});
+	});
+
+	describe('findOwnedById', () => {
+		it('returns a top-level private session only for its owner and agent', async () => {
+			const thread = await createThread();
+			const shared = await createThread({ accessScope: 'project', ownerId: null });
+			const child = await createThread({ parentThreadId: thread.id });
+
+			expect((await threadRepo.findOwnedById(agentId, ownerId, thread.id))?.id).toBe(thread.id);
+			expect(await threadRepo.findOwnedById(agentId, otherUserId, thread.id)).toBeNull();
+			expect(await threadRepo.findOwnedById(otherAgentId, ownerId, thread.id)).toBeNull();
+			expect(await threadRepo.findOwnedById(agentId, ownerId, shared.id)).toBeNull();
+			expect(await threadRepo.findOwnedById(agentId, ownerId, child.id)).toBeNull();
+			expect(await threadRepo.findOwnedById(agentId, ownerId, 'missing')).toBeNull();
+		});
+	});
+
+	describe('renameOwned', () => {
+		it('renames a session and keeps its project', async () => {
+			const thread = await createThread({ title: 'Old title' });
+
+			expect(await threadRepo.renameOwned(agentId, ownerId, thread.id, 'New title', {})).toBe(true);
+			expect(await threadRepo.findOneByOrFail({ id: thread.id })).toMatchObject({
+				title: 'New title',
+				projectId,
+			});
+		});
+
+		it('does not rename a session of another owner, another agent or a parent session', async () => {
+			const parent = await createThread({ title: 'Parent' });
+			const child = await createThread({ parentThreadId: parent.id, title: 'Child' });
+			const shared = await createThread({ accessScope: 'project', ownerId: null, title: 'Shared' });
+			const rename = async (agent: string, owner: string, threadId: string) =>
+				await threadRepo.renameOwned(agent, owner, threadId, 'Changed', {});
+
+			expect(await rename(agentId, otherUserId, parent.id)).toBe(false);
+			expect(await rename(otherAgentId, ownerId, parent.id)).toBe(false);
+			expect(await rename(agentId, ownerId, child.id)).toBe(false);
+			expect(await rename(agentId, ownerId, shared.id)).toBe(false);
+			expect(await rename(agentId, ownerId, 'missing')).toBe(false);
+			const stored = await threadRepo.findBy({});
+			expect(stored.map(({ title }) => title).sort()).toEqual(['Child', 'Parent', 'Shared']);
+		});
+	});
+
+	describe('moveOwned', () => {
+		it('moves a session to another project and keeps its title', async () => {
+			const thread = await createThread({ title: 'Title' });
+			const otherProjectId = (await createTeamProject()).id;
+
+			expect(await threadRepo.moveOwned(agentId, ownerId, thread.id, otherProjectId, {})).toBe(
+				true,
+			);
+			expect(await threadRepo.findOneByOrFail({ id: thread.id })).toMatchObject({
+				title: 'Title',
+				projectId: otherProjectId,
+			});
+		});
+
+		it('does not move a session of another owner, another agent or a parent session', async () => {
+			const parent = await createThread();
+			const child = await createThread({ parentThreadId: parent.id });
+			const shared = await createThread({ accessScope: 'project', ownerId: null });
+			const otherProjectId = (await createTeamProject()).id;
+			const move = async (agent: string, owner: string, threadId: string) =>
+				await threadRepo.moveOwned(agent, owner, threadId, otherProjectId, {});
+
+			expect(await move(agentId, otherUserId, parent.id)).toBe(false);
+			expect(await move(otherAgentId, ownerId, parent.id)).toBe(false);
+			expect(await move(agentId, ownerId, child.id)).toBe(false);
+			expect(await move(agentId, ownerId, shared.id)).toBe(false);
+			expect(await move(agentId, ownerId, 'missing')).toBe(false);
+			const stored = await threadRepo.findBy({});
+			expect(stored.every((thread) => thread.projectId === projectId)).toBe(true);
+		});
+	});
+
+	describe('findOwnedHistoryPage', () => {
+		async function readAllPages(limit: number, search?: string) {
+			const pages: string[][] = [];
+			let before: { updatedAt: Date; id: string } | undefined;
+			// Stop a broken keyset that repeats rows, so the test fails instead of hanging.
+			while (pages.length < 20) {
+				const page = await threadRepo.findOwnedHistoryPage(agentId, ownerId, limit, search, before);
+				pages.push(ids(page.threads));
+				const last = page.threads.at(-1);
+				if (!page.hasMore || !last) break;
+				before = { updatedAt: last.updatedAt, id: last.id };
+			}
+			return pages;
+		}
+
+		it("pages through the owner's top-level private sessions, newest first", async () => {
+			const first = await createThread({}, '2024-03-01T00:00:00.000Z');
+			const second = await createThread({}, '2024-02-01T00:00:00.000Z');
+			const third = await createThread({}, '2024-01-01T00:00:00.000Z');
+			await createThread({ parentThreadId: first.id }, '2024-04-01T00:00:00.000Z');
+			await createThread({ ownerId: otherUserId }, '2024-04-01T00:00:00.000Z');
+			await createThread({ accessScope: 'project', ownerId: null }, '2024-04-01T00:00:00.000Z');
+			await createThread({ agentId: otherAgentId }, '2024-04-01T00:00:00.000Z');
+
+			expect(await readAllPages(2)).toEqual([[first.id, second.id], [third.id]]);
+		});
+
+		it('reports no next page when the last page is exactly full', async () => {
+			const first = await createThread({}, '2024-02-01T00:00:00.000Z');
+			const second = await createThread({}, '2024-01-01T00:00:00.000Z');
+
+			expect(await threadRepo.findOwnedHistoryPage(agentId, ownerId, 2)).toEqual({
+				threads: [
+					expect.objectContaining({ id: first.id }),
+					expect.objectContaining({ id: second.id }),
+				],
+				hasMore: false,
+			});
+		});
+
+		it('neither skips nor repeats sessions with the same updatedAt across pages', async () => {
+			const newest = await createThread({}, '2024-02-01T00:00:00.000Z');
+			const tied = [];
+			for (let index = 0; index < 5; index++) {
+				tied.push(await createThread({}, '2024-01-01T00:00:00.000Z'));
+			}
+			const oldest = await createThread({}, '2023-12-01T00:00:00.000Z');
+
+			const pages = await readAllPages(2);
+
+			expect(pages.flat()).toEqual([newest.id, ...ids(tied).sort().reverse(), oldest.id]);
+			expect(pages.map((page) => page.length)).toEqual([2, 2, 2, 1]);
+		});
+
+		it('pages across a tie stored without milliseconds', async () => {
+			const later = await createThread({}, '2024-01-01T00:00:01.500Z');
+			const tied = [await createThread(), await createThread()];
+			// SQLite can store timestamps without milliseconds. The keyset must still match them.
+			for (const thread of tied) {
+				await threadRepo
+					.createQueryBuilder()
+					.update()
+					.set({ updatedAt: () => "'2024-01-01 00:00:00'" })
+					.where('id = :id', { id: thread.id })
+					.execute();
+			}
+
+			const pages = await readAllPages(1);
+
+			expect(pages).toEqual([
+				[later.id],
+				...ids(tied)
+					.sort()
+					.reverse()
+					.map((id) => [id]),
+			]);
+		});
+
+		it('pages across a tie stored in two formats', async () => {
+			const tied = [];
+			for (let index = 0; index < 4; index++) tied.push(await createThread());
+			// SQLite can store the same time with and without milliseconds.
+			const formats = ["'2024-01-01 00:00:00'", "'2024-01-01 00:00:00.000'"];
+			for (const [index, thread] of tied.entries()) {
+				await threadRepo
+					.createQueryBuilder()
+					.update()
+					.set({ updatedAt: () => formats[index % 2] })
+					.where('id = :id', { id: thread.id })
+					.execute();
+			}
+
+			expect((await readAllPages(1)).flat()).toEqual(ids(tied).sort().reverse());
+		});
+
+		it('filters by title without case sensitivity and matches wildcards literally', async () => {
+			const match = await createThread(
+				{ title: 'Weekly Sales report' },
+				'2024-02-01T00:00:00.000Z',
+			);
+			const literal = await createThread({ title: '100% done_now' }, '2024-01-01T00:00:00.000Z');
+			await createThread({ title: 'Inventory' });
+			await createThread({ title: '100 percent donexnow' });
+			await createThread({ title: null });
+
+			expect((await readAllPages(10, '  sales REPORT ')).flat()).toEqual([match.id]);
+			expect((await readAllPages(10, '0% done_')).flat()).toEqual([literal.id]);
+			expect((await readAllPages(10, '   ')).flat()).toHaveLength(5);
+		});
+
+		it('applies the search on every page', async () => {
+			const first = await createThread({ title: 'Report A' }, '2024-03-01T00:00:00.000Z');
+			await createThread({ title: 'Other' }, '2024-02-15T00:00:00.000Z');
+			const second = await createThread({ title: 'Report B' }, '2024-02-01T00:00:00.000Z');
+
+			expect(await readAllPages(1, 'report')).toEqual([[first.id], [second.id]]);
+		});
+	});
+
+	describe('findByAgentUpdatedBefore', () => {
+		it('returns sessions of the agent updated before the cutoff, oldest first', async () => {
+			const oldest = await createThread({}, '2024-01-01T00:00:00.000Z');
+			const tied = [
+				await createThread({ ownerId: otherUserId }, '2024-01-02T00:00:00.000Z'),
+				await createThread({ accessScope: 'project', ownerId: null }, '2024-01-02T00:00:00.000Z'),
+			];
+			await createThread({}, '2024-01-03T00:00:00.000Z');
+			await createThread({ agentId: otherAgentId }, '2024-01-01T00:00:00.000Z');
+			const cutoff = new Date('2024-01-03T00:00:00.000Z');
+
+			expect(ids(await threadRepo.findByAgentUpdatedBefore(agentId, cutoff, 10))).toEqual([
+				oldest.id,
+				...ids(tied).sort(),
+			]);
+			expect(ids(await threadRepo.findByAgentUpdatedBefore(agentId, cutoff, 1))).toEqual([
+				oldest.id,
+			]);
+		});
+	});
+
+	describe('AgentThreadGrantRepository.revoke', () => {
+		it('removes one grant of one thread', async () => {
+			const thread = await createThread();
+			const other = await createThread();
+			await grantRepo.grant(thread.id, 'tool-a');
+			await grantRepo.grant(thread.id, 'tool-b');
+			await grantRepo.grant(other.id, 'tool-a');
+
+			await grantRepo.revoke(thread.id, 'tool-a');
+			await grantRepo.revoke(thread.id, 'missing');
+
+			expect(await grantRepo.findKeys(thread.id)).toEqual(new Set(['tool-b']));
+			expect(await grantRepo.findKeys(other.id)).toEqual(new Set(['tool-a']));
+		});
+	});
+
+	describe('AgentMessageQueueRepository.hasItems', () => {
+		async function enqueue(threadId: string, hidden = false) {
+			const message = await messageRepo.createInput(
+				{
+					threadId,
+					resourceId: ownerId,
+					content: { role: 'user', content: [{ type: 'text', text: 'Input' }] },
+					origin: hidden ? { source: 'system', hidden: true } : { source: 'system' },
+				},
+				{},
+			);
+			return await queueRepo.enqueue(threadId, message.id, { kind: 'system' }, {});
+		}
+
+		it('is true while a queued or claimed item exists, hidden turns included', async () => {
+			const thread = await createThread();
+			const other = await createThread();
+			expect(await queueRepo.hasItems(thread.id)).toBe(false);
+
+			const item = await enqueue(thread.id, true);
+			expect(await queueRepo.hasItems(thread.id)).toBe(true);
+			expect(await queueRepo.hasItems(other.id)).toBe(false);
+
+			const execution = await executionRepo.save(
+				executionRepo.create({ id: uuid(), threadId: thread.id, status: 'running' }),
+			);
+			expect(await queueRepo.linkExecution(item.id, null, execution.id, {})).toBe(true);
+			expect(await queueRepo.hasItems(thread.id)).toBe(true);
+
+			await queueRepo.removeActive(thread.id, execution.id, {});
+			expect(await queueRepo.hasItems(thread.id)).toBe(false);
+		});
+	});
+});