diff --git a/.changeset/reroute-blob-storage.md b/.changeset/reroute-blob-storage.md new file mode 100644 index 000000000..550677441 --- /dev/null +++ b/.changeset/reroute-blob-storage.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/agent-core-v2": patch +"@moonshot-ai/server-v2": patch +--- + +Reroute the blob store backend from the host filesystem to the pluggable storage layer, so server-only deployments no longer require a local filesystem implementation. diff --git a/packages/agent-core-v2/src/blobStore/blobStore.ts b/packages/agent-core-v2/src/blobStore/blobStore.ts index ff999f5d9..ea0dd51ee 100644 --- a/packages/agent-core-v2/src/blobStore/blobStore.ts +++ b/packages/agent-core-v2/src/blobStore/blobStore.ts @@ -6,7 +6,11 @@ export const BLOBREF_PROTOCOL = 'blobref:'; export const MISSING_MEDIA_PLACEHOLDER = '[media missing]'; export interface BlobStoreServiceOptions { - readonly blobsDir?: string; + /** + * Storage scope used to namespace blob keys in the `IBlobStorage` backend. + * Defaults to `'blobs'`. + */ + readonly storageScope?: string; readonly threshold?: number; readonly maxCacheSize?: number; } diff --git a/packages/agent-core-v2/src/blobStore/blobStoreService.ts b/packages/agent-core-v2/src/blobStore/blobStoreService.ts index dcfc08fab..03100d565 100644 --- a/packages/agent-core-v2/src/blobStore/blobStoreService.ts +++ b/packages/agent-core-v2/src/blobStore/blobStoreService.ts @@ -1,10 +1,8 @@ -import { - createHash } from 'node:crypto'; -import { join } from 'pathe'; +import { createHash } from 'node:crypto'; import type { ContentPart } from '@moonshot-ai/kosong'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IHostFileSystem } from '#/hostFs'; +import { IBlobStorage, type IStorageService } from '#/storage'; import { BLOBREF_PROTOCOL, @@ -15,10 +13,11 @@ import { const DEFAULT_THRESHOLD = 4096; const DEFAULT_MAX_CACHE_SIZE = 50 * 1024 * 1024; +const DEFAULT_STORAGE_SCOPE = 'blobs'; const DATA_URI_HEADER_RE = /^data:([^;]+);base64,/; export class BlobStoreService implements IBlobStoreService { - private readonly blobsDir: string | undefined; + private readonly storageScope: string; private readonly threshold: number; private readonly maxCacheSize: number; private readonly cache = new Map(); @@ -26,10 +25,10 @@ export class BlobStoreService implements IBlobStoreService { private currentCacheSize = 0; constructor( + @IBlobStorage private readonly storage: IStorageService, options: BlobStoreServiceOptions = {}, - @IHostFileSystem private readonly hostFs: IHostFileSystem, ) { - this.blobsDir = options.blobsDir; + this.storageScope = options.storageScope ?? DEFAULT_STORAGE_SCOPE; this.threshold = options.threshold ?? DEFAULT_THRESHOLD; this.maxCacheSize = options.maxCacheSize ?? DEFAULT_MAX_CACHE_SIZE; } @@ -39,8 +38,6 @@ export class BlobStoreService implements IBlobStoreService { } async offloadParts(parts: readonly ContentPart[]): Promise { - if (this.blobsDir === undefined) return parts; - let changed = false; const out: ContentPart[] = []; for (const part of parts) { @@ -52,8 +49,6 @@ export class BlobStoreService implements IBlobStoreService { } async rehydrateParts(parts: readonly ContentPart[]): Promise { - if (this.blobsDir === undefined) return parts; - let changed = false; const out: ContentPart[] = []; for (const part of parts) { @@ -120,9 +115,8 @@ export class BlobStoreService implements IBlobStoreService { this.cache.set(hash, cached); return cached; } - if (this.blobsDir === undefined) return undefined; - const payload = await this.hostFs.readBytes(join(this.blobsDir, hash)).catch(() => undefined); + const payload = await this.storage.read(this.storageScope, hash).catch(() => undefined); if (payload !== undefined) { this.setCache(hash, Buffer.from(payload)); } @@ -143,14 +137,9 @@ export class BlobStoreService implements IBlobStoreService { } private async writeBlob(mimeType: string, base64Payload: string): Promise { - const blobsDir = this.blobsDir; - if (blobsDir === undefined) return `data:${mimeType};base64,${base64Payload}`; - - await this.hostFs.mkdir(blobsDir, { recursive: true }); const hash = createHash('sha256').update(base64Payload, 'utf8').digest('hex'); - const blobPath = join(blobsDir, hash); const binary = Buffer.from(base64Payload, 'base64'); - await this.hostFs.createExclusive(blobPath, binary); + await this.storage.write(this.storageScope, hash, binary, { atomic: true }); this.setCache(hash, binary); return `${BLOBREF_PROTOCOL}${mimeType};${hash}`; } diff --git a/packages/agent-core-v2/src/bootstrap/bootstrap.ts b/packages/agent-core-v2/src/bootstrap/bootstrap.ts index a768090ce..dff5c8194 100644 --- a/packages/agent-core-v2/src/bootstrap/bootstrap.ts +++ b/packages/agent-core-v2/src/bootstrap/bootstrap.ts @@ -21,7 +21,7 @@ import type { Environment } from '@moonshot-ai/kaos'; import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; import { createCoreScope, type Scope, type ScopeSeed } from '#/_base/di/scope'; -import { FileStorageService, IStorageService } from '#/storage'; +import { FileStorageService, IBlobStorage, IStorageService } from '#/storage'; export interface IBootstrapOptions { readonly homeDir: string; @@ -102,7 +102,10 @@ export function bootstrap(input: BootstrapInput = {}, extraSeeds: ScopeSeed = [] } function storageSeed(options: IBootstrapOptions): ScopeSeed { - return [[IStorageService as ServiceIdentifier, new FileStorageService(options.homeDir)]]; + return [ + [IStorageService as ServiceIdentifier, new FileStorageService(options.homeDir)], + [IBlobStorage as ServiceIdentifier, new FileStorageService(options.homeDir)], + ]; } export function resolveKimiHome( diff --git a/packages/agent-core-v2/src/storage/inMemoryStorageService.ts b/packages/agent-core-v2/src/storage/inMemoryStorageService.ts index 65b211711..7bc00b624 100644 --- a/packages/agent-core-v2/src/storage/inMemoryStorageService.ts +++ b/packages/agent-core-v2/src/storage/inMemoryStorageService.ts @@ -25,6 +25,7 @@ import { Emitter, type Event } from '#/_base/event'; import { IAppendLogStorage, IAtomicDocumentStorage, + IBlobStorage, IStorageService, type StorageAppendOptions, type StorageWriteOptions, @@ -167,3 +168,11 @@ registerScopedService( InstantiationType.Delayed, 'storage', ); + +registerScopedService( + LifecycleScope.Core, + IBlobStorage, + InMemoryStorageService, + InstantiationType.Delayed, + 'storage', +); diff --git a/packages/agent-core-v2/src/storage/storageService.ts b/packages/agent-core-v2/src/storage/storageService.ts index 8d23fe0d0..524b4dcbf 100644 --- a/packages/agent-core-v2/src/storage/storageService.ts +++ b/packages/agent-core-v2/src/storage/storageService.ts @@ -66,3 +66,15 @@ export const IAppendLogStorage: ServiceIdentifier = export const IAtomicDocumentStorage: ServiceIdentifier = createDecorator('atomicDocumentStorage'); + +/** + * `IBlobStorage` — role token for the blob-store backend. + * + * Like `IAppendLogStorage` and `IAtomicDocumentStorage`, this is the same + * `IStorageService` interface under a distinct identity so the composition root + * can route large, content-addressed blob objects to a dedicated backend + * (e.g., S3 in a server-only deployment) while other storage roles use a + * different backend. + */ +export const IBlobStorage: ServiceIdentifier = + createDecorator('blobStorage'); diff --git a/packages/agent-core-v2/src/wireRecord/wireRecordService.ts b/packages/agent-core-v2/src/wireRecord/wireRecordService.ts index 112a7facd..b0324329a 100644 --- a/packages/agent-core-v2/src/wireRecord/wireRecordService.ts +++ b/packages/agent-core-v2/src/wireRecord/wireRecordService.ts @@ -1,22 +1,13 @@ -import { InstantiationType } from '#/_base/di/extensions'; -import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { - join -} from 'pathe'; - import { createHash } from 'node:crypto'; +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { Disposable, toDisposable, } from "#/_base/di"; -import { - IBlobStoreService, - type BlobStoreServiceOptions, -} from '#/blobStore'; -import { IHostFileSystem } from '#/hostFs'; +import { IBlobStoreService } from '#/blobStore'; import { IAppendLogStore } from '#/storage'; -import { BlobStoreService } from '../blobStore/blobStoreService'; import { OrderedHookSlot } from '../hooks'; import type { WireRecord, WireRecordMap } from '../wireRecord'; import { @@ -63,10 +54,9 @@ export class WireRecordService extends Disposable implements IWireRecord { private readonly options: WireRecordServiceOptions = {}, @IBlobStoreService injectedBlobStore?: IBlobStoreService, @IAppendLogStore private readonly log?: IAppendLogStore, - @IHostFileSystem private readonly hostFs?: IHostFileSystem, ) { super(); - this.blobStore = this.resolveBlobStore(options, injectedBlobStore); + this.blobStore = options.blobStore ?? injectedBlobStore; this.persistKey = options.homedir === undefined ? undefined : hashKey(options.homedir); if (this.log !== undefined && this.persistKey !== undefined) { this._register(this.log.acquire('wire', this.persistKey)); @@ -324,19 +314,6 @@ export class WireRecordService extends Disposable implements IWireRecord { } return current; } - - private resolveBlobStore( - options: WireRecordServiceOptions, - injectedBlobStore: IBlobStoreService | undefined, - ): IBlobStoreService | undefined { - if (options.blobStore !== undefined) return options.blobStore; - if (options.homedir === undefined || this.hostFs === undefined) return injectedBlobStore; - - const blobOptions: BlobStoreServiceOptions = { - blobsDir: join(options.homedir, 'blobs'), - }; - return new BlobStoreService(blobOptions, this.hostFs); - } } async function* toAsyncIterable( diff --git a/packages/agent-core-v2/test/blobStore/blobStoreService.test.ts b/packages/agent-core-v2/test/blobStore/blobStoreService.test.ts new file mode 100644 index 000000000..389a66911 --- /dev/null +++ b/packages/agent-core-v2/test/blobStore/blobStoreService.test.ts @@ -0,0 +1,117 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import type { ContentPart } from '@moonshot-ai/kosong'; +import { LifecycleScope } from '#/_base/di/scope'; +import { createScopedTestHost, stubPair } from '#/_base/di/test'; +import { IBlobStoreService } from '#/blobStore'; +import { IBlobStorage, InMemoryStorageService } from '#/storage'; + +function makeLargeDataUri(mimeType = 'image/png'): { uri: string; payload: string } { + // The default threshold is 4096 base64 characters. + const payload = 'A'.repeat(5000); + return { uri: `data:${mimeType};base64,${payload}`, payload }; +} + +function makeSmallDataUri(mimeType = 'image/png'): { uri: string; payload: string } { + const payload = 'AQID'; + return { uri: `data:${mimeType};base64,${payload}`, payload }; +} + +describe('BlobStoreService', () => { + let host: ReturnType; + + beforeEach(() => { + host = createScopedTestHost([stubPair(IBlobStorage, new InMemoryStorageService())]); + }); + + afterEach(() => { + host.dispose(); + }); + + function getBlobStore(): IBlobStoreService { + const agent = host.core.createChild(LifecycleScope.Agent, 'test-agent'); + return agent.accessor.get(IBlobStoreService); + } + + it('leaves small data URIs unchanged', async () => { + const store = getBlobStore(); + const { uri } = makeSmallDataUri(); + const parts: ContentPart[] = [{ type: 'image_url', imageUrl: { url: uri } }]; + + const result = await store.offloadParts(parts); + + expect(result).toBe(parts); + expect((result[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe(uri); + }); + + it('offloads large data URIs to blobref URIs', async () => { + const store = getBlobStore(); + const { uri, payload } = makeLargeDataUri(); + const parts: ContentPart[] = [{ type: 'image_url', imageUrl: { url: uri } }]; + + const result = await store.offloadParts(parts); + + expect(result).not.toBe(parts); + const newUrl = (result[0]! as { imageUrl: { url: string } }).imageUrl.url; + expect(store.isBlobRef(newUrl)).toBe(true); + expect(newUrl.startsWith('blobref:image/png;')).toBe(true); + + const backend = host.core.accessor.get(IBlobStorage); + const keys = await backend.list('blobs'); + expect(keys).toHaveLength(1); + expect(Buffer.from((await backend.read('blobs', keys[0]!))!).toString('base64')).toBe(payload); + }); + + it('rehydrates blobref URIs back to data URIs', async () => { + const store = getBlobStore(); + const { uri } = makeLargeDataUri(); + const parts: ContentPart[] = [{ type: 'image_url', imageUrl: { url: uri } }]; + + const offloaded = await store.offloadParts(parts); + const rehydrated = await store.rehydrateParts(offloaded); + + expect((rehydrated[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe(uri); + }); + + it('rehydrates only blobref URLs and leaves other URLs alone', async () => { + const store = getBlobStore(); + const { uri: largeUri } = makeLargeDataUri(); + const { uri: smallUri } = makeSmallDataUri(); + const parts: ContentPart[] = [ + { type: 'image_url', imageUrl: { url: largeUri } }, + { type: 'audio_url', audioUrl: { url: smallUri } }, + ]; + + const offloaded = await store.offloadParts(parts); + expect(store.isBlobRef((offloaded[0]! as { imageUrl: { url: string } }).imageUrl.url)).toBe(true); + expect((offloaded[1]! as { audioUrl: { url: string } }).audioUrl.url).toBe(smallUri); + + const rehydrated = await store.rehydrateParts(offloaded); + expect((rehydrated[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe(largeUri); + expect((rehydrated[1]! as { audioUrl: { url: string } }).audioUrl.url).toBe(smallUri); + }); + + it('replaces missing blobs with a placeholder', async () => { + const store = getBlobStore(); + const parts: ContentPart[] = [ + { type: 'image_url', imageUrl: { url: 'blobref:image/png;deadbeef' } }, + ]; + + const result = await store.rehydrateParts(parts); + + expect((result[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe('[media missing]'); + }); + + it('is idempotent across offload/rehydrate cycles', async () => { + const store = getBlobStore(); + const { uri } = makeLargeDataUri(); + const parts: ContentPart[] = [{ type: 'image_url', imageUrl: { url: uri } }]; + + const first = await store.offloadParts(parts); + const second = await store.offloadParts(first); + + expect((second[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe( + (first[0]! as { imageUrl: { url: string } }).imageUrl.url, + ); + }); +}); diff --git a/packages/server-v2/src/start.ts b/packages/server-v2/src/start.ts index 4f09bcb0f..76295d27d 100644 --- a/packages/server-v2/src/start.ts +++ b/packages/server-v2/src/start.ts @@ -12,6 +12,7 @@ import { FileStorageService, IAppendLogStorage, IAtomicDocumentStorage, + IBlobStorage, logSeed, resolveConfigPath, resolveKimiHome, @@ -61,6 +62,7 @@ function durableStorageSeeds(homeDir: string): ScopeSeed { return [ [IAtomicDocumentStorage as ServiceIdentifier, new FileStorageService(homeDir)], [IAppendLogStorage as ServiceIdentifier, new FileStorageService(homeDir)], + [IBlobStorage as ServiceIdentifier, new FileStorageService(homeDir)], ]; } @@ -71,9 +73,9 @@ export async function startServer(opts: ServerStartOptions = {}): Promise