diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 2eec9e7a69f..5e42b825c37 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -441,12 +441,12 @@ erDiagram } "public.agent_files" { varchar_36_ agentId FK - text binaryDataId timestamp_3__with_time_zone createdAt varchar_255_ fileName integer fileSizeBytes varchar_16_ id varchar_255_ mimeType + varchar_2_ storedAt timestamp_3__with_time_zone updatedAt } "public.agent_history" { diff --git a/docs/generated/postgres-schema/public.agent_files.md b/docs/generated/postgres-schema/public.agent_files.md index 12041cef160..7e4e96d49a8 100644 --- a/docs/generated/postgres-schema/public.agent_files.md +++ b/docs/generated/postgres-schema/public.agent_files.md @@ -5,27 +5,28 @@ | Name | Type | Default | Nullable | Children | Parents | Comment | | ---- | ---- | ------- | -------- | -------- | ------- | ------- | | agentId | varchar(36) | | false | | [public.agents](public.agents.md) | Agent that owns this uploaded file | -| binaryDataId | text | | false | | | Opaque BinaryDataService reference (mode-prefixed, e.g. "filesystem-v2:\"); not an FK to binary_data, which only has rows in DB storage mode | | createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | | fileName | varchar(255) | | false | | | | | fileSizeBytes | integer | | false | | | Uploaded file size in bytes | | id | varchar(16) | | false | | | Application-generated n8n nano ID | | mimeType | varchar(255) | | false | | | | +| storedAt | varchar(2) | 'fs'::character varying | false | | | | | updatedAt | timestamp(3) with time zone | CURRENT_TIMESTAMP(3) | false | | | | ## Constraints | Name | Type | Definition | | ---- | ---- | ---------- | +| CHK_agent_files_storedAt | CHECK | CHECK ((("storedAt")::text = ANY ((ARRAY['fs'::character varying, 's3'::character varying, 'az'::character varying])::text[]))) | | FK_aca4514cb500494b64356c2e164 | FOREIGN KEY | FOREIGN KEY ("agentId") REFERENCES agents(id) ON DELETE CASCADE | | PK_692920e59217af7d124cd95106f | PRIMARY KEY | PRIMARY KEY (id) | | agent_files_agentId_not_null | n | NOT NULL "agentId" | -| agent_files_binaryDataId_not_null | n | NOT NULL "binaryDataId" | | agent_files_createdAt_not_null | n | NOT NULL "createdAt" | | agent_files_fileName_not_null | n | NOT NULL "fileName" | | agent_files_fileSizeBytes_not_null | n | NOT NULL "fileSizeBytes" | | agent_files_id_not_null | n | NOT NULL id | | agent_files_mimeType_not_null | n | NOT NULL "mimeType" | +| agent_files_storedAt_not_null | n | NOT NULL "storedAt" | | agent_files_updatedAt_not_null | n | NOT NULL "updatedAt" | ## Indexes @@ -33,7 +34,6 @@ | Name | Definition | | ---- | ---------- | | IDX_45dafc48fe2ce95eac30fc8ffd | CREATE INDEX "IDX_45dafc48fe2ce95eac30fc8ffd" ON public.agent_files USING btree ("agentId", "createdAt") | -| IDX_agent_files_agentId_binaryDataId | CREATE UNIQUE INDEX "IDX_agent_files_agentId_binaryDataId" ON public.agent_files USING btree ("agentId", "binaryDataId") | | IDX_agent_files_agentId_fileName | CREATE UNIQUE INDEX "IDX_agent_files_agentId_fileName" ON public.agent_files USING btree ("agentId", "fileName") | | PK_692920e59217af7d124cd95106f | CREATE UNIQUE INDEX "PK_692920e59217af7d124cd95106f" ON public.agent_files USING btree (id) | @@ -46,12 +46,12 @@ erDiagram "public.agent_files" { varchar_36_ agentId FK - text binaryDataId timestamp_3__with_time_zone createdAt varchar_255_ fileName integer fileSizeBytes varchar_16_ id varchar_255_ mimeType + varchar_2_ storedAt timestamp_3__with_time_zone updatedAt } "public.agents" { diff --git a/docs/generated/postgres-schema/public.agents.md b/docs/generated/postgres-schema/public.agents.md index 358b5a4b071..30441a4c89a 100644 --- a/docs/generated/postgres-schema/public.agents.md +++ b/docs/generated/postgres-schema/public.agents.md @@ -136,12 +136,12 @@ erDiagram } "public.agent_files" { varchar_36_ agentId FK - text binaryDataId timestamp_3__with_time_zone createdAt varchar_255_ fileName integer fileSizeBytes varchar_16_ id varchar_255_ mimeType + varchar_2_ storedAt timestamp_3__with_time_zone updatedAt } "public.agent_task_definition" { diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 0f5cae0de59..dd7965ed784 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -428,12 +428,12 @@ erDiagram } "agent_files" { varchar_36_ agentId FK - TEXT binaryDataId datetime_3_ createdAt varchar_255_ fileName INTEGER fileSizeBytes varchar_16_ id PK varchar_255_ mimeType + VARCHAR_2_ storedAt datetime_3_ updatedAt } "agent_history" { diff --git a/docs/generated/sqlite-schema/agent_files.md b/docs/generated/sqlite-schema/agent_files.md index 43dd427be19..96948bfc4d9 100644 --- a/docs/generated/sqlite-schema/agent_files.md +++ b/docs/generated/sqlite-schema/agent_files.md @@ -6,7 +6,7 @@ Table Definition ```sql -CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" varchar(36) NOT NULL, "binaryDataId" text NOT NULL, "fileName" varchar(255) NOT NULL, "mimeType" varchar(255) NOT NULL, "fileSizeBytes" integer NOT NULL, "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')), CONSTRAINT "FK_aca4514cb500494b64356c2e164" FOREIGN KEY ("agentId") REFERENCES "agents" ("id") ON DELETE CASCADE) +CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" varchar(36) NOT NULL, "fileName" varchar(255) NOT NULL, "mimeType" varchar(255) NOT NULL, "fileSizeBytes" integer NOT NULL, "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 'fs' CONSTRAINT "CHK_agent_files_storedAt" CHECK("storedAt" IN ('fs', 's3', 'az')), CONSTRAINT "FK_aca4514cb500494b64356c2e164" FOREIGN KEY ("agentId") REFERENCES "agents" ("id") ON DELETE CASCADE ON UPDATE NO ACTION) ``` @@ -16,18 +16,19 @@ CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" var | Name | Type | Default | Nullable | Children | Parents | Comment | | ---- | ---- | ------- | -------- | -------- | ------- | ------- | | agentId | varchar(36) | | false | | [agents](agents.md) | | -| binaryDataId | TEXT | | false | | | | | createdAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | | fileName | varchar(255) | | false | | | | | fileSizeBytes | INTEGER | | false | | | | | id | varchar(16) | | false | | | | | mimeType | varchar(255) | | false | | | | +| storedAt | VARCHAR(2) | 'fs' | false | | | | | updatedAt | datetime(3) | STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW') | false | | | | ## Constraints | Name | Type | Definition | | ---- | ---- | ---------- | +| - | CHECK | CHECK("storedAt" IN ('fs', 's3', 'az')) | | - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (agentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE | | id | PRIMARY KEY | PRIMARY KEY (id) | | sqlite_autoindex_agent_files_1 | PRIMARY KEY | PRIMARY KEY (id) | @@ -37,7 +38,6 @@ CREATE TABLE "agent_files" ("id" varchar(16) PRIMARY KEY NOT NULL, "agentId" var | Name | Definition | | ---- | ---------- | | IDX_45dafc48fe2ce95eac30fc8ffd | CREATE INDEX "IDX_45dafc48fe2ce95eac30fc8ffd" ON "agent_files" ("agentId", "createdAt") | -| IDX_agent_files_agentId_binaryDataId | CREATE UNIQUE INDEX "IDX_agent_files_agentId_binaryDataId" ON "agent_files" ("agentId", "binaryDataId") | | IDX_agent_files_agentId_fileName | CREATE UNIQUE INDEX "IDX_agent_files_agentId_fileName" ON "agent_files" ("agentId", "fileName") | | sqlite_autoindex_agent_files_1 | PRIMARY KEY (id) | @@ -50,12 +50,12 @@ erDiagram "agent_files" { varchar_36_ agentId FK - TEXT binaryDataId datetime_3_ createdAt varchar_255_ fileName INTEGER fileSizeBytes varchar_16_ id PK varchar_255_ mimeType + VARCHAR_2_ storedAt datetime_3_ updatedAt } "agents" { diff --git a/docs/generated/sqlite-schema/agents.md b/docs/generated/sqlite-schema/agents.md index d925895eee7..5ad4662ac89 100644 --- a/docs/generated/sqlite-schema/agents.md +++ b/docs/generated/sqlite-schema/agents.md @@ -140,12 +140,12 @@ erDiagram } "agent_files" { varchar_36_ agentId FK - TEXT binaryDataId datetime_3_ createdAt varchar_255_ fileName INTEGER fileSizeBytes varchar_16_ id PK varchar_255_ mimeType + VARCHAR_2_ storedAt datetime_3_ updatedAt } "agent_task_definition" { diff --git a/packages/@n8n/db/src/migrations/common/1784900752603-ReplaceAgentFileBinaryDataIdWithStoredAt.ts b/packages/@n8n/db/src/migrations/common/1784900752603-ReplaceAgentFileBinaryDataIdWithStoredAt.ts new file mode 100644 index 00000000000..69a066230ac --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1784900752603-ReplaceAgentFileBinaryDataIdWithStoredAt.ts @@ -0,0 +1,55 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +/** + * Replaces BinaryDataService's opaque `binaryDataId` with a blob-storage + * `storedAt` location. Existing knowledge-file rows are disposable (feature at + * low rollout) and are deleted rather than migrated — the bytes they point at + * are no longer readable through the new store. + */ +export class ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603 implements ReversibleMigration { + async up({ + escape, + runQuery, + tablePrefix, + schemaBuilder: { dropIndex, dropColumns }, + }: MigrationContext) { + const agentFiles = escape.tableName('agent_files'); + const binaryData = escape.tableName('binary_data'); + const sourceType = escape.columnName('sourceType'); + const storedAt = escape.columnName('storedAt'); + + await runQuery(`DELETE FROM ${agentFiles}`); + await runQuery(`DELETE FROM ${binaryData} WHERE ${sourceType} = 'agent_file'`); + + await dropIndex('agent_files', ['agentId', 'binaryDataId'], { skipIfMissing: true }); + await dropColumns('agent_files', ['binaryDataId'], { recreatesOnSqlite: true }); + + await runQuery( + `ALTER TABLE ${agentFiles} ADD COLUMN ${storedAt} VARCHAR(2) NOT NULL DEFAULT 'fs' ` + + `CONSTRAINT "CHK_${tablePrefix}agent_files_storedAt" CHECK(${storedAt} IN ('fs', 's3', 'az'))`, + ); + } + + async down({ + escape, + runQuery, + schemaBuilder: { addColumns, createIndex, column }, + }: MigrationContext) { + const agentFiles = escape.tableName('agent_files'); + const storedAt = escape.columnName('storedAt'); + + await runQuery(`DELETE FROM ${agentFiles}`); + await runQuery(`ALTER TABLE ${agentFiles} DROP COLUMN ${storedAt}`); + + await addColumns( + 'agent_files', + [ + column('binaryDataId').text.notNull.comment( + 'Opaque BinaryDataService reference (mode-prefixed); not an FK to binary_data', + ), + ], + { recreatesOnSqlite: true }, + ); + await createIndex('agent_files', ['agentId', 'binaryDataId'], true); + } +} diff --git a/packages/@n8n/db/src/migrations/postgresdb/index.ts b/packages/@n8n/db/src/migrations/postgresdb/index.ts index fcdd15e889a..0ed6c6dbe53 100644 --- a/packages/@n8n/db/src/migrations/postgresdb/index.ts +++ b/packages/@n8n/db/src/migrations/postgresdb/index.ts @@ -230,6 +230,7 @@ import { AddStoredAtToAgentExecution1784815940110 } from '../common/178481594011 import { AddInstanceCredentials1784815940111 } from '../common/1784815940111-AddInstanceCredentials'; import { CreateAgentEvalTables1784815940112 } from '../common/1784815940112-CreateAgentEvalTables'; import type { Migration } from '../migration-types'; +import { ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603 } from '../common/1784900752603-ReplaceAgentFileBinaryDataIdWithStoredAt'; export const postgresMigrations: Migration[] = [ InitialMigration1587669153312, @@ -463,4 +464,5 @@ export const postgresMigrations: Migration[] = [ AddStoredAtToAgentExecution1784815940110, AddInstanceCredentials1784815940111, CreateAgentEvalTables1784815940112, + ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603, ]; diff --git a/packages/@n8n/db/src/migrations/sqlite/index.ts b/packages/@n8n/db/src/migrations/sqlite/index.ts index 3679349e891..cda15ae924f 100644 --- a/packages/@n8n/db/src/migrations/sqlite/index.ts +++ b/packages/@n8n/db/src/migrations/sqlite/index.ts @@ -221,6 +221,7 @@ import { CreateWorkflowReviewRequestTables1784000000052 } from '../common/178400 import { AddStoredAtToAgentExecution1784815940110 } from '../common/1784815940110-AddStoredAtToAgentExecution'; import { AddInstanceCredentials1784815940111 } from '../common/1784815940111-AddInstanceCredentials'; import { CreateAgentEvalTables1784815940112 } from '../common/1784815940112-CreateAgentEvalTables'; +import { ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603 } from '../common/1784900752603-ReplaceAgentFileBinaryDataIdWithStoredAt'; const sqliteMigrations: Migration[] = [ InitialMigration1588102412422, @@ -445,6 +446,7 @@ const sqliteMigrations: Migration[] = [ AddStoredAtToAgentExecution1784815940110, AddInstanceCredentials1784815940111, CreateAgentEvalTables1784815940112, + ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603, ]; export { sqliteMigrations }; diff --git a/packages/cli/src/modules/agents/__tests__/agent-knowledge-commands.test.ts b/packages/cli/src/modules/agents/__tests__/agent-knowledge-commands.test.ts index 2b66c8a341e..161d82b77bf 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-knowledge-commands.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-knowledge-commands.test.ts @@ -18,7 +18,8 @@ import { const mobyDickFile: AgentKnowledgeFileReference = { file: 'moby-dick.txt', fileId: 'file-1', - binaryDataId: 'filesystem-v2:agents/agent-1/knowledge-files/file-1/binary_data/uuid', + agentId: 'agent-1', + storedAt: 'fs', displayName: 'moby-dick.txt', mimeType: 'text/plain', fileSizeBytes: 123, diff --git a/packages/cli/src/modules/agents/__tests__/agent-knowledge-file-store.test.ts b/packages/cli/src/modules/agents/__tests__/agent-knowledge-file-store.test.ts new file mode 100644 index 00000000000..716316300bf --- /dev/null +++ b/packages/cli/src/modules/agents/__tests__/agent-knowledge-file-store.test.ts @@ -0,0 +1,104 @@ +import type { Logger } from '@n8n/backend-common'; +import { FsByteStore } from '@n8n/blob-storage'; +import { UnexpectedError } from 'n8n-workflow'; +import type { StorageConfig } from 'n8n-core'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { mock } from 'vitest-mock-extended'; + +import { AgentKnowledgeFileStore } from '../agent-knowledge-file-store'; + +vi.unmock('node:fs'); +vi.unmock('node:fs/promises'); + +describe('AgentKnowledgeFileStore', () => { + let storagePath: string; + let storageConfig: StorageConfig; + let logger: ReturnType>; + let store: AgentKnowledgeFileStore; + + beforeAll(async () => { + storagePath = await mkdtemp(join(tmpdir(), 'n8n-agent-knowledge-file-store-')); + }); + + beforeEach(async () => { + for (const entry of await (await import('node:fs/promises')).readdir(storagePath)) { + await rm(join(storagePath, entry), { recursive: true, force: true }); + } + storageConfig = { storagePath, modeTag: 'db' } as StorageConfig; + logger = mock(); + const fsByteStore = new FsByteStore({ + storagePath, + reportError: () => {}, + }); + store = new AgentKnowledgeFileStore(fsByteStore as never, storageConfig, logger); + }); + + afterAll(async () => { + await rm(storagePath, { recursive: true, force: true }); + }); + + it('falls back to fs when execution data storage mode is database and round-trips bytes', async () => { + const body = Buffer.from('knowledge-bytes', 'utf-8'); + const ref = { agentId: 'agent-1', fileId: 'file-1' }; + + const storedAt = await store.write(ref, body, { + fileName: 'notes.txt', + mimeType: 'text/plain', + }); + + expect(storedAt).toBe('fs'); + expect(logger.warn).toHaveBeenCalledWith( + "Execution data storage mode is 'database'; agent knowledge files will be stored on the local filesystem", + ); + await expect(store.readAsBuffer({ ...ref, storedAt: 'fs' })).resolves.toEqual(body); + }); + + it('writes to a registered s3 location when modeTag is s3', async () => { + storageConfig = { storagePath, modeTag: 's3' } as StorageConfig; + const fsByteStore = new FsByteStore({ storagePath, reportError: () => {} }); + store = new AgentKnowledgeFileStore(fsByteStore as never, storageConfig, logger); + + const s3Path = join(storagePath, 's3-root'); + const s3Store = new FsByteStore({ storagePath: s3Path, reportError: () => {} }); + store.registerByteStore('s3', s3Store); + + const body = Buffer.from('s3-bytes', 'utf-8'); + const ref = { agentId: 'agent-1', fileId: 'file-2' }; + + const storedAt = await store.write(ref, body, { mimeType: 'text/plain' }); + + expect(storedAt).toBe('s3'); + await expect(store.readAsBuffer({ ...ref, storedAt: 's3' })).resolves.toEqual(body); + }); + + it('returns null for a missing blob and throws for an unregistered storedAt', async () => { + await expect( + store.readAsBuffer({ agentId: 'agent-1', fileId: 'missing', storedAt: 'fs' }), + ).resolves.toBeNull(); + + await expect( + store.readAsBuffer({ agentId: 'agent-1', fileId: 'file-1', storedAt: 's3' }), + ).rejects.toThrow(UnexpectedError); + }); + + it('deletes registered blobs and skips unregistered locations without throwing', async () => { + const kept = { agentId: 'agent-1', fileId: 'keep' }; + const body = Buffer.from('delete-me', 'utf-8'); + await store.write(kept, body, { mimeType: 'text/plain' }); + + await expect( + store.delete([ + { ...kept, storedAt: 'fs' }, + { agentId: 'agent-1', fileId: 'orphan', storedAt: 's3' }, + ]), + ).resolves.toBeUndefined(); + + await expect(store.readAsBuffer({ ...kept, storedAt: 'fs' })).resolves.toBeNull(); + expect(logger.warn).toHaveBeenCalledWith( + 'Skipped deleting agent knowledge files for unconfigured storage', + expect.objectContaining({ storedAt: 's3', count: 1 }), + ); + }); +}); diff --git a/packages/cli/src/modules/agents/__tests__/agent-knowledge-sandbox.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-knowledge-sandbox.service.test.ts index 3bfd91343b4..f04181aa25d 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-knowledge-sandbox.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-knowledge-sandbox.service.test.ts @@ -3,10 +3,11 @@ import type { Logger } from '@n8n/backend-common'; import type { AgentsConfig } from '@n8n/config'; import type { AiAssistantClient } from '@n8n_io/ai-assistant-sdk'; import { mock } from 'vitest-mock-extended'; -import type { BinaryDataService, InstanceSettings } from 'n8n-core'; +import type { InstanceSettings } from 'n8n-core'; import type { AiService } from '../../../services/ai.service'; +import type { AgentKnowledgeFileStore } from '../agent-knowledge-file-store'; import type { Agent } from '../entities/agent.entity'; import type { AgentFile } from '../entities/agent-file.entity'; import { KNOWLEDGE_MIRROR_FILES_DIR } from '../agent-knowledge-storage'; @@ -102,10 +103,10 @@ function makePublishedAgentRepository(): ReturnType return repository; } -function makeBinaryDataService(): Mocked { - const service = mock(); - service.getAsBuffer.mockResolvedValue(Buffer.from('mock file content')); - return service; +function makeKnowledgeFileStore(): Mocked { + const store = mock(); + store.readAsBuffer.mockResolvedValue(Buffer.from('mock file content')); + return store; } function makeService( @@ -115,7 +116,7 @@ function makeService( instanceSettings: InstanceSettings = mock({ instanceId }), agentFileRepository: AgentFileRepository = mock(), agentRepository: AgentRepository = makePublishedAgentRepository(), - binaryDataService: BinaryDataService = makeBinaryDataService(), + agentKnowledgeFileStore: AgentKnowledgeFileStore = makeKnowledgeFileStore(), ): AgentKnowledgeSandboxService { return new AgentKnowledgeSandboxService( { @@ -134,7 +135,7 @@ function makeService( instanceSettings, agentFileRepository, agentRepository, - binaryDataService, + agentKnowledgeFileStore, ); } @@ -144,7 +145,7 @@ function makeAgentFile(overrides: Partial = {}): AgentFile { return { id, agentId, - binaryDataId: `filesystem-v2:agents/${agentId}/knowledge-files/${id}/binary_data/uuid`, + storedAt: 'fs', fileName, mimeType: 'text/plain', fileSizeBytes: 100, @@ -547,7 +548,7 @@ describe('AgentKnowledgeSandboxService', () => { ]); const agentRepository = makePublishedAgentRepository(); agentRepository.existsBy.mockResolvedValue(true); - const binaryDataService = makeBinaryDataService(); + const agentKnowledgeFileStore = makeKnowledgeFileStore(); const service = makeService( {}, mock(), @@ -555,30 +556,30 @@ describe('AgentKnowledgeSandboxService', () => { mock({ instanceId }), agentFileRepository, agentRepository, - binaryDataService, + agentKnowledgeFileStore, ); await service.searchKnowledge(projectId, agentId, { pattern: 'foo' }); let commands = sandbox.process.executeCommand.mock.calls.map(([command]) => command); expect(commands.filter(isManifestReadCommand)).toHaveLength(1); expect(commands.filter(isMirrorSyncCommand)).toHaveLength(1); - expect(binaryDataService.getAsBuffer).toHaveBeenCalledTimes(2); + expect(agentKnowledgeFileStore.readAsBuffer).toHaveBeenCalledTimes(2); expect(sandbox.fs.uploadFiles).toHaveBeenCalledTimes(1); manifestState = 'doc1.txt\ndoc2.txt\n'; sandbox.process.executeCommand.mockClear(); - binaryDataService.getAsBuffer.mockClear(); + agentKnowledgeFileStore.readAsBuffer.mockClear(); sandbox.fs.uploadFiles.mockClear(); await service.searchKnowledge(projectId, agentId, { pattern: 'bar' }); commands = sandbox.process.executeCommand.mock.calls.map(([command]) => command); expect(commands.filter(isManifestReadCommand)).toHaveLength(0); expect(commands.filter(isMirrorSyncCommand)).toHaveLength(0); expect(commands).toHaveLength(1); - expect(binaryDataService.getAsBuffer).not.toHaveBeenCalled(); + expect(agentKnowledgeFileStore.readAsBuffer).not.toHaveBeenCalled(); expect(sandbox.fs.uploadFiles).not.toHaveBeenCalled(); sandbox.process.executeCommand.mockClear(); - binaryDataService.getAsBuffer.mockClear(); + agentKnowledgeFileStore.readAsBuffer.mockClear(); sandbox.fs.uploadFiles.mockClear(); agentFileRepository.findByAgentId.mockResolvedValue([ makeMirrorFile('file-1', 'doc1.txt'), @@ -593,7 +594,7 @@ describe('AgentKnowledgeSandboxService', () => { // Only the newly-added name should be fetched and staged for move — // the manifest rewrite (which always lists every expected name) is // a separate, later part of the finalize command. - expect(binaryDataService.getAsBuffer).toHaveBeenCalledTimes(1); + expect(agentKnowledgeFileStore.readAsBuffer).toHaveBeenCalledTimes(1); expect(syncCommands[0]).toContain('.tmp-doc3.txt'); expect(syncCommands[0]).not.toContain('.tmp-doc1.txt'); expect(syncCommands[0]).not.toContain('.tmp-doc2.txt'); @@ -614,8 +615,10 @@ describe('AgentKnowledgeSandboxService', () => { ]); const agentRepository = makePublishedAgentRepository(); agentRepository.existsBy.mockResolvedValue(true); - const binaryDataService = makeBinaryDataService(); - binaryDataService.getAsBuffer.mockResolvedValue(Buffer.alloc(MIRROR_UPLOAD_BATCH_BYTES)); + const agentKnowledgeFileStore = makeKnowledgeFileStore(); + agentKnowledgeFileStore.readAsBuffer.mockResolvedValue( + Buffer.alloc(MIRROR_UPLOAD_BATCH_BYTES), + ); const service = makeService( {}, mock(), @@ -623,7 +626,7 @@ describe('AgentKnowledgeSandboxService', () => { mock({ instanceId }), agentFileRepository, agentRepository, - binaryDataService, + agentKnowledgeFileStore, ); await service.searchKnowledge(projectId, agentId, { pattern: 'foo' }); @@ -691,7 +694,7 @@ describe('AgentKnowledgeSandboxService', () => { ); }); - it('skips a file that fails to load from binary data storage and retries it next sync', async () => { + it('skips a file that fails to load from the knowledge file store and retries it next sync', async () => { const sandbox = makeSandbox('started'); sandbox.process.executeCommand.mockResolvedValue({ exitCode: 0, @@ -702,8 +705,8 @@ describe('AgentKnowledgeSandboxService', () => { agentFileRepository.findByAgentId.mockResolvedValue([makeMirrorFile('file-1', 'doc1.txt')]); const agentRepository = makePublishedAgentRepository(); agentRepository.existsBy.mockResolvedValue(true); - const binaryDataService = makeBinaryDataService(); - binaryDataService.getAsBuffer.mockRejectedValueOnce(new Error('missing on disk')); + const agentKnowledgeFileStore = makeKnowledgeFileStore(); + agentKnowledgeFileStore.readAsBuffer.mockRejectedValueOnce(new Error('missing on disk')); const logger = mock(); const service = makeService( {}, @@ -712,7 +715,7 @@ describe('AgentKnowledgeSandboxService', () => { mock({ instanceId }), agentFileRepository, agentRepository, - binaryDataService, + agentKnowledgeFileStore, ); await expect( @@ -727,12 +730,12 @@ describe('AgentKnowledgeSandboxService', () => { // The failed file was left out of the cached manifest, so the next // sync's expected hash mismatches and it retries — this time - // `getAsBuffer` succeeds (the mock's default resolves). + // `readAsBuffer` succeeds (the mock's default resolves). await expect( service.searchKnowledge(projectId, agentId, { pattern: 'bar' }), ).resolves.toBeDefined(); - expect(binaryDataService.getAsBuffer).toHaveBeenCalledTimes(2); + expect(agentKnowledgeFileStore.readAsBuffer).toHaveBeenCalledTimes(2); expect(sandbox.fs.uploadFiles).toHaveBeenCalledTimes(1); }); }); diff --git a/packages/cli/src/modules/agents/__tests__/agent-knowledge.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-knowledge.service.test.ts index 874c20225d9..c27e78ced15 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-knowledge.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-knowledge.service.test.ts @@ -7,8 +7,8 @@ import { Readable } from 'node:stream'; import { MAX_AGENT_KNOWLEDGE_BASE_SIZE_BYTES } from '@n8n/api-types'; import type { Logger } from '@n8n/backend-common'; -import type { BinaryDataConfig, BinaryDataService } from 'n8n-core'; +import type { AgentKnowledgeFileStore } from '../agent-knowledge-file-store'; import { AgentKnowledgeService } from '../agent-knowledge.service'; import type { AgentKnowledgeSandboxService } from '../agent-knowledge-sandbox.service'; import type { AgentFile } from '../entities/agent-file.entity'; @@ -28,10 +28,9 @@ vi.mock('@n8n/ai-utilities', () => ({ })); /** - * Real `BinaryDataService` managers fully consume the stream before - * resolving/rejecting; mock implementations must too, or an unconsumed - * `createReadStream` lazily opens after the test's temp-file cleanup runs - * and throws an unhandled ENOENT. + * The knowledge file store writes streams by consuming them; mock + * implementations must too, or an unconsumed `createReadStream` lazily opens + * after the test's temp-file cleanup runs and throws an unhandled ENOENT. */ async function drainIfStream(content: unknown): Promise { if (content instanceof Readable) { @@ -65,7 +64,7 @@ function makeAgentFile(overrides: Partial = {}): AgentFile { return { id: 'file-1', agentId, - binaryDataId: 'filesystem-v2:agents/agent-1/knowledge-files/file-1/binary_data/uuid', + storedAt: 'fs', fileName: 'first.txt', mimeType: 'text/plain', fileSizeBytes: 4, @@ -84,7 +83,7 @@ class InMemoryAgentFileRepository { return { id: input.id ?? 'generated-id', agentId: input.agentId ?? agentId, - binaryDataId: input.binaryDataId ?? '', + storedAt: input.storedAt ?? 'fs', fileName: input.fileName ?? '', mimeType: input.mimeType ?? '', fileSizeBytes: input.fileSizeBytes ?? 0, @@ -134,39 +133,33 @@ describe('AgentKnowledgeService', () => { let agentRepository: Mocked; let agentFileRepository: InMemoryAgentFileRepository; let agentKnowledgeSandboxService: Mocked; - let binaryDataService: Mocked; - let binaryDataConfig: BinaryDataConfig; + let agentKnowledgeFileStore: Mocked; let logger: Mocked; let service: AgentKnowledgeService; - let storeSequence: number; beforeEach(() => { vi.clearAllMocks(); - storeSequence = 0; agentRepository = mock(); agentFileRepository = new InMemoryAgentFileRepository(); agentKnowledgeSandboxService = mock(); - binaryDataService = mock(); - binaryDataService.store.mockImplementation(async (_location, content, binaryData) => { - storeSequence += 1; + agentKnowledgeFileStore = mock(); + agentKnowledgeFileStore.write.mockImplementation(async (_ref, content) => { await drainIfStream(content); - return { ...binaryData, id: `filesystem-v2:mock-${storeSequence}` }; + return 'fs'; }); - binaryDataService.deleteManyByBinaryDataId.mockResolvedValue(undefined); - binaryDataConfig = { mode: 'filesystem' } as BinaryDataConfig; + agentKnowledgeFileStore.delete.mockResolvedValue(undefined); logger = mock(); service = new AgentKnowledgeService( agentRepository, agentFileRepository as unknown as AgentFileRepository, agentKnowledgeSandboxService, - binaryDataService, - binaryDataConfig, + agentKnowledgeFileStore, logger, ); loadMock.mockResolvedValue([{ pageContent: 'extracted pdf text' }]); }); - it('stores text and PDF files via BinaryDataService, creates DB rows, and cleans temp files', async () => { + it('stores text and PDF files via the knowledge file store, creates DB rows, and cleans temp files', async () => { agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never); const tempDirectory = await mkdtemp(path.join(tmpdir(), 'agent-knowledge-upload-')); const textFilePath = path.join(tempDirectory, 'notes.txt'); @@ -206,16 +199,9 @@ describe('AgentKnowledgeService', () => { }), ]); - expect(binaryDataService.store).toHaveBeenCalledTimes(2); - const [textCall, pdfCall] = binaryDataService.store.mock.calls; - expect(textCall[0]).toEqual( - expect.objectContaining({ - type: 'custom', - pathSegments: ['agents', agentId, 'knowledge-files', expect.any(String)], - sourceType: 'agent_file', - sourceId: expect.any(String), - }), - ); + expect(agentKnowledgeFileStore.write).toHaveBeenCalledTimes(2); + const [textCall, pdfCall] = agentKnowledgeFileStore.write.mock.calls; + expect(textCall[0]).toEqual({ agentId, fileId: expect.any(String) }); expect(textCall[1]).toBeInstanceOf(Readable); expect(textCall[2]).toEqual( expect.objectContaining({ mimeType: 'text/plain', fileName: 'notes.txt' }), @@ -226,8 +212,8 @@ describe('AgentKnowledgeService', () => { ); const [storedTextFile, storedPdfFile] = agentFileRepository.all(); - expect(storedTextFile.binaryDataId).toBe('filesystem-v2:mock-1'); - expect(storedPdfFile.binaryDataId).toBe('filesystem-v2:mock-2'); + expect(storedTextFile.storedAt).toBe('fs'); + expect(storedPdfFile.storedAt).toBe('fs'); await expect(access(textFilePath)).rejects.toThrow(); await expect(access(pdfFilePath)).rejects.toThrow(); @@ -241,7 +227,7 @@ describe('AgentKnowledgeService', () => { ).rejects.toThrow('Invalid knowledge file name'); expect(agentFileRepository.all()).toEqual([]); - expect(binaryDataService.store).not.toHaveBeenCalled(); + expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled(); }); it('cleans up already-stored files when a later upload in the batch fails', async () => { @@ -251,13 +237,13 @@ describe('AgentKnowledgeService', () => { const secondPath = path.join(tempDirectory, 'second.txt'); await writeFile(firstPath, 'hello'); await writeFile(secondPath, 'world'); - binaryDataService.store + agentKnowledgeFileStore.write .mockReset() - .mockImplementationOnce(async (_location, content, binaryData) => { + .mockImplementationOnce(async (_ref, content) => { await drainIfStream(content); - return { ...binaryData, id: 'filesystem-v2:mock-1' }; + return 'fs'; }) - .mockImplementationOnce(async (_location, content) => { + .mockImplementationOnce(async (_ref, content) => { await drainIfStream(content); throw new Error('store failed'); }); @@ -275,48 +261,11 @@ describe('AgentKnowledgeService', () => { ).rejects.toThrow('store failed'); expect(agentFileRepository.all()).toEqual([]); - expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledWith([ - 'filesystem-v2:mock-1', + expect(agentKnowledgeFileStore.delete).toHaveBeenCalledWith([ + expect.objectContaining({ agentId, storedAt: 'fs' }), ]); }); - it('rejects uploads when the binary data service has no persisted storage mode', async () => { - agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never); - const tempDirectory = await mkdtemp(path.join(tmpdir(), 'agent-knowledge-upload-')); - const filePath = path.join(tempDirectory, 'notes.txt'); - await writeFile(filePath, 'hello'); - binaryDataService.store.mockImplementationOnce(async (_location, content, binaryData) => { - await drainIfStream(content); - return { ...binaryData, id: undefined }; - }); - - await expect( - service.uploadFiles(agentId, projectId, [ - makeMulterFile({ originalname: 'notes.txt', path: filePath, size: 5, buffer: undefined }), - ]), - ).rejects.toThrow('Agent knowledge base requires a persisted binary data storage mode'); - expect(agentFileRepository.all()).toEqual([]); - }); - - it('rejects uploads up front when binary data mode is in-memory', async () => { - agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never); - binaryDataConfig = { mode: 'default' } as BinaryDataConfig; - service = new AgentKnowledgeService( - agentRepository, - agentFileRepository as unknown as AgentFileRepository, - agentKnowledgeSandboxService, - binaryDataService, - binaryDataConfig, - logger, - ); - - await expect( - service.uploadFiles(agentId, projectId, [makeMulterFile({ originalname: 'notes.txt' })]), - ).rejects.toThrow('Agent knowledge base requires a persisted binary data storage mode'); - expect(binaryDataService.store).not.toHaveBeenCalled(); - expect(agentFileRepository.all()).toEqual([]); - }); - it('allows uploads that bring the knowledge base exactly to the size limit', async () => { agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never); await agentFileRepository.save( @@ -347,7 +296,7 @@ describe('AgentKnowledgeService', () => { }), ]); expect(agentFileRepository.all()).toHaveLength(2); - expect(binaryDataService.store).toHaveBeenCalledTimes(1); + expect(agentKnowledgeFileStore.write).toHaveBeenCalledTimes(1); }); it('rejects uploads that would exceed the knowledge base size limit', async () => { @@ -369,7 +318,7 @@ describe('AgentKnowledgeService', () => { ]), ).rejects.toThrow('Knowledge base limit reached'); expect(agentFileRepository.all()).toHaveLength(1); - expect(binaryDataService.store).not.toHaveBeenCalled(); + expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled(); }); it('rejects uploads when existing knowledge files already exceed the size limit', async () => { @@ -391,48 +340,42 @@ describe('AgentKnowledgeService', () => { ]), ).rejects.toThrow('Knowledge base limit reached'); expect(agentFileRepository.all()).toHaveLength(1); - expect(binaryDataService.store).not.toHaveBeenCalled(); + expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled(); }); - it('deletes the DB row and its binary data', async () => { + it('deletes the DB row and its blob', async () => { agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never); - await agentFileRepository.save( - makeAgentFile({ id: 'file-1', binaryDataId: 'filesystem-v2:file-1.txt' }), - ); + await agentFileRepository.save(makeAgentFile({ id: 'file-1', storedAt: 'fs' })); await expect(service.deleteFile(agentId, projectId, 'file-1')).resolves.toBeUndefined(); expect(agentFileRepository.all()).toEqual([]); - expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledWith([ - 'filesystem-v2:file-1.txt', + expect(agentKnowledgeFileStore.delete).toHaveBeenCalledWith([ + { agentId, fileId: 'file-1', storedAt: 'fs' }, ]); }); - it('logs binary data deletion failures without restoring the DB row', async () => { + it('logs blob deletion failures without restoring the DB row', async () => { agentRepository.findByIdAndProjectId.mockResolvedValue({ id: agentId, projectId } as never); - await agentFileRepository.save( - makeAgentFile({ id: 'file-1', binaryDataId: 'filesystem-v2:file-1.txt' }), - ); - binaryDataService.deleteManyByBinaryDataId.mockRejectedValueOnce(new Error('delete failed')); + await agentFileRepository.save(makeAgentFile({ id: 'file-1', storedAt: 'fs' })); + agentKnowledgeFileStore.delete.mockRejectedValueOnce(new Error('delete failed')); await expect(service.deleteFile(agentId, projectId, 'file-1')).resolves.toBeUndefined(); expect(agentFileRepository.all()).toEqual([]); - expect(logger.warn).toHaveBeenCalledWith('Failed to delete knowledge file binary data', { + expect(logger.warn).toHaveBeenCalledWith('Failed to delete knowledge file blob', { agentId, fileId: 'file-1', error: 'delete failed', }); }); - it('deletes all agent files and their binary data', async () => { - await agentFileRepository.save( - makeAgentFile({ id: 'file-1', binaryDataId: 'filesystem-v2:file-1.txt' }), - ); + it('deletes all agent files and their blobs', async () => { + await agentFileRepository.save(makeAgentFile({ id: 'file-1', storedAt: 'fs' })); await agentFileRepository.save( makeAgentFile({ id: 'file-2', - binaryDataId: 'filesystem-v2:file-2.md', + storedAt: 's3', fileName: 'guide.md', mimeType: 'text/markdown', }), @@ -441,11 +384,10 @@ describe('AgentKnowledgeService', () => { await expect(service.deleteAllFilesForAgent(projectId, agentId)).resolves.toBeUndefined(); expect(agentFileRepository.all()).toEqual([]); - expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledWith([ - 'filesystem-v2:file-1.txt', - 'filesystem-v2:file-2.md', + expect(agentKnowledgeFileStore.delete).toHaveBeenCalledWith([ + { agentId, fileId: 'file-1', storedAt: 'fs' }, + { agentId, fileId: 'file-2', storedAt: 's3' }, ]); - expect(binaryDataService.deleteMany).not.toHaveBeenCalled(); }); describe('published-only gating', () => { @@ -460,7 +402,7 @@ describe('AgentKnowledgeService', () => { 'Knowledge base is only available for published agents. Publish the agent first.', ); expect(agentFileRepository.all()).toEqual([]); - expect(binaryDataService.store).not.toHaveBeenCalled(); + expect(agentKnowledgeFileStore.write).not.toHaveBeenCalled(); }); it('rejects warmup for an unpublished agent', async () => { diff --git a/packages/cli/src/modules/agents/agent-knowledge-file-store.ts b/packages/cli/src/modules/agents/agent-knowledge-file-store.ts new file mode 100644 index 00000000000..62033e69174 --- /dev/null +++ b/packages/cli/src/modules/agents/agent-knowledge-file-store.ts @@ -0,0 +1,121 @@ +import { Logger } from '@n8n/backend-common'; +import { + FsByteStore, + type ByteStore, + type PreWriteBlobMetadata, + type StorageLocation, +} from '@n8n/blob-storage'; +import { Service } from '@n8n/di'; +import { ErrorReporter, StorageConfig } from 'n8n-core'; +import { UnexpectedError } from 'n8n-workflow'; +import type { Readable } from 'node:stream'; + +export type AgentKnowledgeFileRef = { agentId: string; fileId: string }; +export type StoredAgentKnowledgeFileRef = AgentKnowledgeFileRef & { storedAt: StorageLocation }; + +@Service() +export class AgentKnowledgeFileFsByteStore extends FsByteStore { + constructor(storageConfig: StorageConfig, errorReporter: ErrorReporter) { + super({ + storagePath: storageConfig.storagePath, + reportError: (error) => errorReporter.error(error), + }); + } +} + +/** + * Stores agent knowledge file bytes via pluggable backends. The `fs` backend is + * always available; `s3` and `az` are registered at module init when configured. + * When execution data storage mode is `database`, writes fall back to `fs`. + */ +@Service() +export class AgentKnowledgeFileStore { + private readonly byteStores = new Map(); + private warnedAboutDbFallback = false; + + constructor( + fsByteStore: AgentKnowledgeFileFsByteStore, + private readonly storageConfig: StorageConfig, + private readonly logger: Logger, + ) { + this.byteStores.set('fs', fsByteStore); + } + + registerByteStore(loc: StorageLocation, store: ByteStore) { + this.byteStores.set(loc, store); + } + + hasLocation(loc: StorageLocation) { + return this.byteStores.has(loc); + } + + get writeLocation(): StorageLocation { + if (this.storageConfig.modeTag === 'db') { + if (!this.warnedAboutDbFallback) { + this.warnedAboutDbFallback = true; + this.logger.warn( + "Execution data storage mode is 'database'; agent knowledge files will be stored on the local filesystem", + ); + } + return 'fs'; + } + return this.storageConfig.modeTag; + } + + async write( + ref: AgentKnowledgeFileRef, + body: Buffer | Readable, + metadata: PreWriteBlobMetadata, + ): Promise { + const loc = this.writeLocation; + await this.getByteStore(loc).write(this.key(ref), body, metadata); + return loc; + } + + async readAsBuffer(ref: StoredAgentKnowledgeFileRef): Promise { + return await this.getByteStore(ref.storedAt).read(this.key(ref)); + } + + async delete(refs: StoredAgentKnowledgeFileRef[]): Promise { + if (refs.length === 0) return; + + const groups = new Map(); + for (const ref of refs) { + const group = groups.get(ref.storedAt) ?? []; + group.push(ref); + groups.set(ref.storedAt, group); + } + + await Promise.all( + [...groups].map(async ([loc, group]) => { + const store = this.byteStores.get(loc); + if (!store) { + this.logger.warn('Skipped deleting agent knowledge files for unconfigured storage', { + storedAt: loc, + count: group.length, + }); + return; + } + await store.delete(group.map((ref) => this.key(ref))); + }), + ); + } + + private key(ref: AgentKnowledgeFileRef) { + return [ + 'agents', + encodeURIComponent(ref.agentId), + 'knowledge-files', + encodeURIComponent(ref.fileId), + 'content', + ].join('/'); + } + + private getByteStore(loc: StorageLocation): ByteStore { + const store = this.byteStores.get(loc); + if (!store) { + throw new UnexpectedError(`Knowledge file store for location "${loc}" is not configured.`); + } + return store; + } +} diff --git a/packages/cli/src/modules/agents/agent-knowledge-retrieval.ts b/packages/cli/src/modules/agents/agent-knowledge-retrieval.ts index f9288f59f34..b03deff8a30 100644 --- a/packages/cli/src/modules/agents/agent-knowledge-retrieval.ts +++ b/packages/cli/src/modules/agents/agent-knowledge-retrieval.ts @@ -1,6 +1,8 @@ -import { BadRequestError } from '@/errors/response-errors/bad-request.error'; +import type { StorageLocation } from '@n8n/blob-storage'; import { z } from 'zod'; +import { BadRequestError } from '@/errors/response-errors/bad-request.error'; + import { hasControlCharacter } from './agent-knowledge-storage'; const MAX_SEARCH_PATTERN_LENGTH = 500; @@ -186,7 +188,8 @@ export type ReadKnowledgeRequest = z.infer; export interface AgentKnowledgeFileReference { file: string; fileId: string; - binaryDataId: string; + agentId: string; + storedAt: StorageLocation; displayName: string; mimeType: string; fileSizeBytes: number; diff --git a/packages/cli/src/modules/agents/agent-knowledge-sandbox.service.ts b/packages/cli/src/modules/agents/agent-knowledge-sandbox.service.ts index 2de628424dc..ab2785d331d 100644 --- a/packages/cli/src/modules/agents/agent-knowledge-sandbox.service.ts +++ b/packages/cli/src/modules/agents/agent-knowledge-sandbox.service.ts @@ -5,7 +5,7 @@ import { Logger } from '@n8n/backend-common'; import { AgentsConfig } from '@n8n/config'; import { Time } from '@n8n/constants'; import { Service } from '@n8n/di'; -import { BinaryDataService, InstanceSettings } from 'n8n-core'; +import { InstanceSettings } from 'n8n-core'; import { OperationalError } from 'n8n-workflow'; import { nanoid } from 'nanoid'; import { createHash } from 'node:crypto'; @@ -30,6 +30,7 @@ import { parseRipgrepFilesOutput, parseRipgrepOutput, } from './agent-knowledge-commands'; +import { AgentKnowledgeFileStore } from './agent-knowledge-file-store'; import { isAgentKnowledgeBaseEnabled } from './agent-knowledge-gate'; import { assertValidKnowledgeFilePath, @@ -216,7 +217,7 @@ export class AgentKnowledgeSandboxService { private readonly instanceSettings: InstanceSettings, private readonly agentFileRepository: AgentFileRepository, private readonly agentRepository: AgentRepository, - private readonly binaryDataService: BinaryDataService, + private readonly agentKnowledgeFileStore: AgentKnowledgeFileStore, ) {} async warmSandbox(projectId: string, agentId: string): Promise { @@ -497,8 +498,8 @@ export class AgentKnowledgeSandboxService { const filesByName = new Map(files.map((file) => [file.file, file])); const copiedNames = await this.uploadMirrorFiles(sandbox, toCopy, filesByName, sandboxName); - // Files that failed to load from BinaryDataService are left out of the - // manifest, so the next sync attempt retries them as `toCopy` again. + // Files that failed to load from the knowledge file store are left out of + // the manifest, so the next sync attempt retries them as `toCopy` again. const finalManifestNames = expectedNames.filter( (name) => copiedNames.has(name) || presentSet.has(name), ); @@ -521,11 +522,11 @@ export class AgentKnowledgeSandboxService { } /** - * Fetches each `names` entry from BinaryDataService and uploads it to the - * sandbox mirror under a `.tmp-` prefix; `buildMirrorFinalizeCommand` moves - * it into place so a concurrent search never sees a partially-written file. - * Uploads flush in `MIRROR_UPLOAD_BATCH_BYTES`-sized batches so the whole - * knowledge base (up to 1.5 GB) is never held in memory at once. + * Fetches each `names` entry from the knowledge file store and uploads it to + * the sandbox mirror under a `.tmp-` prefix; `buildMirrorFinalizeCommand` + * moves it into place so a concurrent search never sees a partially-written + * file. Uploads flush in `MIRROR_UPLOAD_BATCH_BYTES`-sized batches so the + * whole knowledge base (up to 1.5 GB) is never held in memory at once. * Returns the subset of `names` that were fetched and uploaded successfully. */ private async uploadMirrorFiles( @@ -557,11 +558,19 @@ export class AgentKnowledgeSandboxService { if (!file) continue; try { - const buffer = await this.binaryDataService.getAsBuffer({ - id: file.binaryDataId, - data: '', - mimeType: file.mimeType, + const buffer = await this.agentKnowledgeFileStore.readAsBuffer({ + agentId: file.agentId, + fileId: file.fileId, + storedAt: file.storedAt, }); + if (!buffer) { + this.logger.warn('Failed to load agent knowledge file for mirror sync', { + sandboxName, + file: name, + error: 'not found', + }); + continue; + } batch.push({ source: buffer, destination: `${KNOWLEDGE_MIRROR_FILES_DIR}/.tmp-${name}` }); batchNames.push(name); batchBytes += buffer.length; @@ -639,7 +648,8 @@ export class AgentKnowledgeSandboxService { return files.map((file) => ({ file: storageFileNameForOriginalFileName(file.fileName), fileId: file.id, - binaryDataId: file.binaryDataId, + agentId: file.agentId, + storedAt: file.storedAt, displayName: file.fileName, mimeType: file.mimeType, fileSizeBytes: file.fileSizeBytes, diff --git a/packages/cli/src/modules/agents/agent-knowledge-storage.ts b/packages/cli/src/modules/agents/agent-knowledge-storage.ts index d0a162aaa9c..ada00e12a19 100644 --- a/packages/cli/src/modules/agents/agent-knowledge-storage.ts +++ b/packages/cli/src/modules/agents/agent-knowledge-storage.ts @@ -1,6 +1,4 @@ import type { AgentFileDto } from '@n8n/api-types'; -import type { SourceType } from '@n8n/db'; -import { FileLocation } from 'n8n-core'; import path from 'node:path'; import { BadRequestError } from '@/errors/response-errors/bad-request.error'; @@ -8,15 +6,11 @@ import { BadRequestError } from '@/errors/response-errors/bad-request.error'; import type { AgentFile } from './entities/agent-file.entity'; // Local sandbox disk mirroring the DB-backed knowledge files, so repeated -// reads/searches avoid re-fetching from BinaryDataService each time. +// reads/searches avoid re-fetching from the knowledge file store each time. export const KNOWLEDGE_MIRROR_DIR = '/home/daytona/knowledge-mirror'; export const KNOWLEDGE_MIRROR_FILES_DIR = `${KNOWLEDGE_MIRROR_DIR}/files`; export const KNOWLEDGE_MIRROR_MANIFEST = `${KNOWLEDGE_MIRROR_DIR}/manifest`; -// Typed against `SourceType` so a drift from the `binary_data` schema enum -// (see `packages/@n8n/db/src/entities/binary-data-file.ts`) is a compile error. -const AGENT_FILE_SOURCE_TYPE: SourceType = 'agent_file'; - export function hasControlCharacter(value: string): boolean { for (const character of value) { if (character.charCodeAt(0) < 32) { @@ -33,15 +27,6 @@ function sanitizePathCharacter(character: string): string { return character; } -/** One directory per file, so deleting a single file never touches others. */ -export function buildKnowledgeFileLocation(agentId: string, fileId: string) { - return FileLocation.ofCustom({ - pathSegments: ['agents', agentId, 'knowledge-files', fileId], - sourceType: AGENT_FILE_SOURCE_TYPE, - sourceId: fileId, - }); -} - export function assertKnowledgePathSegment(segment: string, label: string): void { if ( !segment || diff --git a/packages/cli/src/modules/agents/agent-knowledge.service.ts b/packages/cli/src/modules/agents/agent-knowledge.service.ts index 14d1207e52a..27febed3123 100644 --- a/packages/cli/src/modules/agents/agent-knowledge.service.ts +++ b/packages/cli/src/modules/agents/agent-knowledge.service.ts @@ -5,11 +5,10 @@ import { } from '@n8n/api-types'; import { N8nPdfLoader } from '@n8n/ai-utilities'; import { Logger } from '@n8n/backend-common'; +import type { StorageLocation } from '@n8n/blob-storage'; import { Service } from '@n8n/di'; import { QueryFailedError } from '@n8n/typeorm'; import { generateNanoId } from '@n8n/utils/generate-nano-id'; -import { BinaryDataConfig, BinaryDataService } from 'n8n-core'; -import { OperationalError, type IBinaryData } from 'n8n-workflow'; import { createReadStream } from 'node:fs'; import { unlink } from 'node:fs/promises'; import path from 'node:path'; @@ -17,11 +16,8 @@ import type { Readable } from 'node:stream'; import { BadRequestError } from '@/errors/response-errors/bad-request.error'; import { NotFoundError } from '@/errors/response-errors/not-found.error'; -import { - buildKnowledgeFileLocation, - storageFileNameForOriginalFileName, - toAgentFileDto, -} from './agent-knowledge-storage'; +import { AgentKnowledgeFileStore } from './agent-knowledge-file-store'; +import { storageFileNameForOriginalFileName, toAgentFileDto } from './agent-knowledge-storage'; import { AgentKnowledgeSandboxService } from './agent-knowledge-sandbox.service'; import type { AgentFile } from './entities/agent-file.entity'; import type { Agent } from './entities/agent.entity'; @@ -52,8 +48,7 @@ export class AgentKnowledgeService { private readonly agentRepository: AgentRepository, private readonly agentFileRepository: AgentFileRepository, private readonly agentKnowledgeSandboxService: AgentKnowledgeSandboxService, - private readonly binaryDataService: BinaryDataService, - private readonly binaryDataConfig: BinaryDataConfig, + private readonly agentKnowledgeFileStore: AgentKnowledgeFileStore, private readonly logger: Logger, ) {} @@ -65,11 +60,6 @@ export class AgentKnowledgeService { try { const agent = await this.ensureAgentBelongsToProject(agentId, projectId); this.assertAgentPublished(agent); - if (this.binaryDataConfig.mode === 'default') { - throw new OperationalError( - 'Agent knowledge base requires a persisted binary data storage mode', - ); - } this.validateUploadMetadata(files); await this.validateUploadBatch(agentId, files); @@ -113,13 +103,15 @@ export class AgentKnowledgeService { } await this.agentFileRepository.delete({ id: fileId, agentId }); - await this.binaryDataService.deleteManyByBinaryDataId([file.binaryDataId]).catch((error) => { - this.logger.warn('Failed to delete knowledge file binary data', { - agentId, - fileId: file.id, - error: error instanceof Error ? error.message : error, + await this.agentKnowledgeFileStore + .delete([{ agentId, fileId: file.id, storedAt: file.storedAt }]) + .catch((error) => { + this.logger.warn('Failed to delete knowledge file blob', { + agentId, + fileId: file.id, + error: error instanceof Error ? error.message : error, + }); }); - }); this.agentKnowledgeSandboxService.invalidateMirror(projectId, agentId); this.agentKnowledgeSandboxService.prewarmMirrorInBackground(projectId, agentId); } @@ -128,10 +120,16 @@ export class AgentKnowledgeService { const files = await this.agentFileRepository.findByAgentId(agentId); await this.agentFileRepository.delete({ agentId }); if (files.length > 0) { - await this.binaryDataService - .deleteManyByBinaryDataId(files.map((file) => file.binaryDataId)) + await this.agentKnowledgeFileStore + .delete( + files.map((file) => ({ + agentId, + fileId: file.id, + storedAt: file.storedAt, + })), + ) .catch((error) => { - this.logger.warn('Failed to delete knowledge files binary data', { + this.logger.warn('Failed to delete knowledge file blobs', { agentId, error: error instanceof Error ? error.message : error, }); @@ -145,32 +143,21 @@ export class AgentKnowledgeService { await this.agentKnowledgeSandboxService.destroySandbox(projectId, agentId); } - /** Stores the file's bytes via BinaryDataService, then reserves its DB row. */ + /** Stores the file's bytes via AgentKnowledgeFileStore, then reserves its DB row. */ private async storeAgentFile(agentId: string, file: Express.Multer.File): Promise { const fileId = generateNanoId(); const storageFileName = storageFileNameForOriginalFileName(file.originalname); const content = await this.prepareUploadContent(file); - const binaryData: IBinaryData = { - data: '', - mimeType: file.mimetype, + const storedAt = await this.agentKnowledgeFileStore.write({ agentId, fileId }, content, { fileName: storageFileName, - }; - const stored = await this.binaryDataService.store( - buildKnowledgeFileLocation(agentId, fileId), - content, - binaryData, - ); - if (!stored.id) { - throw new OperationalError( - 'Agent knowledge base requires a persisted binary data storage mode', - ); - } + mimeType: file.mimetype, + }); try { - return await this.saveAgentFile(agentId, fileId, file, stored.id); + return await this.saveAgentFile(agentId, fileId, file, storedAt); } catch (error) { - await this.binaryDataService.deleteManyByBinaryDataId([stored.id]).catch(() => {}); + await this.agentKnowledgeFileStore.delete([{ agentId, fileId, storedAt }]).catch(() => {}); if (isUniqueConstraintError(error)) { throw this.duplicateFileNameError(file.originalname); } @@ -182,12 +169,12 @@ export class AgentKnowledgeService { agentId: string, fileId: string, file: Express.Multer.File, - binaryDataId: string, + storedAt: StorageLocation, ): Promise { const agentFile = this.agentFileRepository.create({ id: fileId, agentId, - binaryDataId, + storedAt, fileName: file.originalname, mimeType: file.mimetype, fileSizeBytes: file.size, @@ -212,7 +199,9 @@ export class AgentKnowledgeService { private async cleanupUploadedFiles(files: AgentFile[]): Promise { for (const file of files) { - await this.binaryDataService.deleteManyByBinaryDataId([file.binaryDataId]).catch(() => {}); + await this.agentKnowledgeFileStore + .delete([{ agentId: file.agentId, fileId: file.id, storedAt: file.storedAt }]) + .catch(() => {}); await this.agentFileRepository.delete({ id: file.id, agentId: file.agentId }).catch(() => {}); } } diff --git a/packages/cli/src/modules/agents/agents.module.ts b/packages/cli/src/modules/agents/agents.module.ts index 10a87e37550..f3bb0a711c7 100644 --- a/packages/cli/src/modules/agents/agents.module.ts +++ b/packages/cli/src/modules/agents/agents.module.ts @@ -34,31 +34,32 @@ export class AgentsModule implements ModuleInterface { const { AgentExecutionService } = await import('./agent-execution.service.js'); Container.get(AgentExecutionService); - // Register blob backends for agent execution logs. The fs backend is always - // available; s3/az reuse the clients base-command already initialized - // (initBinaryDataService runs before module init in all commands) and are - // only registered when that init succeeded — mirroring ExecutionDataJsonStore, - // so a configured-but-unreachable backend degrades instead of throwing. + // Register blob backends for agent execution logs and knowledge files. + // The fs backend is always available; s3/az reuse the clients base-command + // already initialized (initBinaryDataService runs before module init in + // all commands) and are only registered when that init succeeded — + // mirroring ExecutionDataJsonStore, so a configured-but-unreachable + // backend degrades instead of throwing. const { AzureByteStore, S3ByteStore } = await import('@n8n/blob-storage'); const { AgentExecutionLogStore } = await import('./execution-log/agent-execution-log-store.js'); + const { AgentKnowledgeFileStore } = await import('./agent-knowledge-file-store.js'); const { ExecutionDataJsonStore } = await import( '@/executions/execution-data/execution-data-json-store.js' ); const agentExecutionLogStore = Container.get(AgentExecutionLogStore); + const agentKnowledgeFileStore = Container.get(AgentKnowledgeFileStore); const executionDataJsonStore = Container.get(ExecutionDataJsonStore); if (executionDataJsonStore.hasLocation('s3')) { const { ObjectStoreService } = await import('@n8n/blob-storage/object-store'); - agentExecutionLogStore.registerByteStore( - 's3', - new S3ByteStore(Container.get(ObjectStoreService)), - ); + const s3Store = new S3ByteStore(Container.get(ObjectStoreService)); + agentExecutionLogStore.registerByteStore('s3', s3Store); + agentKnowledgeFileStore.registerByteStore('s3', s3Store); } if (executionDataJsonStore.hasLocation('az')) { const { AzureBlobService } = await import('@n8n/blob-storage/azure-blob'); - agentExecutionLogStore.registerByteStore( - 'az', - new AzureByteStore(Container.get(AzureBlobService)), - ); + const azStore = new AzureByteStore(Container.get(AzureBlobService)); + agentExecutionLogStore.registerByteStore('az', azStore); + agentKnowledgeFileStore.registerByteStore('az', azStore); } const { AgentRuntimeCacheService } = await import('./agent-runtime-cache.service.js'); diff --git a/packages/cli/src/modules/agents/entities/agent-file.entity.ts b/packages/cli/src/modules/agents/entities/agent-file.entity.ts index 1c1548f0a38..f0bb836a610 100644 --- a/packages/cli/src/modules/agents/entities/agent-file.entity.ts +++ b/packages/cli/src/modules/agents/entities/agent-file.entity.ts @@ -1,3 +1,4 @@ +import type { StorageLocation } from '@n8n/blob-storage'; import { WithTimestampsAndStringId } from '@n8n/db'; import { Column, Entity, Index, JoinColumn, ManyToOne, type Relation } from '@n8n/typeorm'; @@ -6,7 +7,6 @@ import { Agent } from './agent.entity'; @Entity({ name: 'agent_files' }) @Index(['agentId', 'createdAt']) @Index(['agentId', 'fileName'], { unique: true }) -@Index(['agentId', 'binaryDataId'], { unique: true }) export class AgentFile extends WithTimestampsAndStringId { @ManyToOne(() => Agent, { onDelete: 'CASCADE' }) @JoinColumn({ name: 'agentId' }) @@ -16,17 +16,16 @@ export class AgentFile extends WithTimestampsAndStringId { agentId: string; /** - * BinaryDataService id (e.g. `filesystem-v2:agents//knowledge-files//binary_data/`). - * Not a DB FK — see `BinaryDataService` in `n8n-core` for how the bytes are resolved from this id. + * Blob-storage location for the file bytes: 'fs', 's3', or 'az'. Never 'db'. + * Bytes are resolved via AgentKnowledgeFileStore. */ - @Column({ type: 'text' }) - binaryDataId: string; + @Column({ type: 'varchar', length: 2, nullable: false, default: 'fs' }) + storedAt: StorageLocation; // fileName/mimeType/fileSizeBytes are intentionally denormalized rather than - // joined from binary_data: (1) binaryDataId is an opaque storage reference, - // not an FK; (2) we keep the original user-facing values, which differ from - // the stored binary for converted uploads (a PDF is stored as extracted - // text with a different byte size). + // joined from storage metadata: we keep the original user-facing values, + // which differ from the stored bytes for converted uploads (a PDF is stored + // as extracted text with a different byte size). @Column({ type: 'varchar', length: 255 }) fileName: string; diff --git a/packages/cli/test/migration/1784900752603-replace-agent-file-binary-data-id-with-stored-at.test.ts b/packages/cli/test/migration/1784900752603-replace-agent-file-binary-data-id-with-stored-at.test.ts new file mode 100644 index 00000000000..e3d11b94ab6 --- /dev/null +++ b/packages/cli/test/migration/1784900752603-replace-agent-file-binary-data-id-with-stored-at.test.ts @@ -0,0 +1,144 @@ +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'; + +const MIGRATION_NAME = 'ReplaceAgentFileBinaryDataIdWithStoredAt1784900752603'; + +async function columnNames(context: TestMigrationContext, table: string): Promise { + if (context.isSqlite) { + const rows = await context.runQuery>( + `PRAGMA table_info(${context.escape.tableName(table)})`, + ); + return rows.map((row) => row.name); + } + const fullName = `${context.tablePrefix}${table}`; + const rows = await context.runQuery>( + `SELECT column_name FROM information_schema.columns WHERE table_name = :name`, + { name: fullName }, + ); + return rows.map((row) => row.column_name); +} + +describe('ReplaceAgentFileBinaryDataIdWithStoredAt Migration', () => { + let dataSource: DataSource; + + beforeAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.init(); + + dataSource = Container.get(DataSource); + + const context = createTestMigrationContext(dataSource); + await context.queryRunner.clearDatabase(); + await context.queryRunner.release(); + + await initDbUpToMigration(MIGRATION_NAME); + }); + + afterAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.close(); + }); + + async function insertProject(context: TestMigrationContext, id: string): Promise { + const table = context.escape.tableName('project'); + const now = new Date(); + await context.runQuery( + `INSERT INTO ${table} ("id", "name", "type", "customTelemetryTags", "createdAt", "updatedAt") VALUES (:id, :name, :type, :tags, :createdAt, :updatedAt)`, + { id, name: 'Test Project', type: 'team', tags: '[]', createdAt: now, updatedAt: now }, + ); + } + + async function insertAgent( + context: TestMigrationContext, + data: { id: string; projectId: string }, + ): Promise { + const table = context.escape.tableName('agents'); + const now = new Date(); + await context.runQuery( + `INSERT INTO ${table} ("id", "name", "projectId", "integrations", "tools", "skills", "createdAt", "updatedAt") VALUES (:id, :name, :projectId, :integrations, :tools, :skills, :createdAt, :updatedAt)`, + { + id: data.id, + name: 'Test Agent', + projectId: data.projectId, + integrations: '[]', + tools: '{}', + skills: '{}', + createdAt: now, + updatedAt: now, + }, + ); + } + + it('deletes existing knowledge files and replaces binaryDataId with storedAt', async () => { + const seedContext = createTestMigrationContext(dataSource); + try { + await insertProject(seedContext, 'project-1'); + await insertAgent(seedContext, { id: 'agent-1', projectId: 'project-1' }); + + const agentFiles = seedContext.escape.tableName('agent_files'); + const now = new Date(); + await seedContext.runQuery( + `INSERT INTO ${agentFiles} ("id", "agentId", "binaryDataId", "fileName", "mimeType", "fileSizeBytes", "createdAt", "updatedAt") VALUES (:id, :agentId, :binaryDataId, :fileName, :mimeType, :fileSizeBytes, :createdAt, :updatedAt)`, + { + id: 'file-1', + agentId: 'agent-1', + binaryDataId: 'filesystem-v2:agents/agent-1/knowledge-files/file-1/binary_data/uuid', + fileName: 'notes.txt', + mimeType: 'text/plain', + fileSizeBytes: 11, + createdAt: now, + updatedAt: now, + }, + ); + + const binaryData = seedContext.escape.tableName('binary_data'); + await seedContext.runQuery( + `INSERT INTO ${binaryData} ("fileId", "sourceType", "sourceId", "data", "mimeType", "fileName", "fileSize", "createdAt", "updatedAt") VALUES (:fileId, :sourceType, :sourceId, :data, :mimeType, :fileName, :fileSize, :createdAt, :updatedAt)`, + { + fileId: '11111111-1111-1111-1111-111111111111', + sourceType: 'agent_file', + sourceId: 'file-1', + data: Buffer.from('notes'), + mimeType: 'text/plain', + fileName: 'notes.txt', + fileSize: 5, + createdAt: now, + updatedAt: now, + }, + ); + } finally { + await seedContext.queryRunner.release(); + } + + await runSingleMigration(MIGRATION_NAME); + + const assertContext = createTestMigrationContext(dataSource); + try { + const agentFiles = assertContext.escape.tableName('agent_files'); + const binaryData = assertContext.escape.tableName('binary_data'); + + const remainingFiles = await assertContext.runQuery>( + `SELECT "id" FROM ${agentFiles}`, + ); + expect(remainingFiles).toEqual([]); + + const remainingBinary = await assertContext.runQuery>( + `SELECT "fileId" FROM ${binaryData} WHERE "sourceType" = 'agent_file'`, + ); + expect(remainingBinary).toEqual([]); + + const columns = await columnNames(assertContext, 'agent_files'); + expect(columns).toContain('storedAt'); + expect(columns).not.toContain('binaryDataId'); + } finally { + await assertContext.queryRunner.release(); + } + }); +});