From 667898b9dffca5b3109acf377b30bd6f2101ff54 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Thu, 9 Jul 2026 11:22:17 +0800 Subject: [PATCH] refactor(agent-core-v2): split blob service helpers, rewrite tests - extract rewriteMediaUrls and blobref parse/format helpers to dedupe URL rewriting - move the byte-bounded LRU cache into a module-private ByteLruCache with focused unit tests, dropping the protected maxCacheSize test seam - rewrite blob service tests against the contract on in-memory storage, removing cache-internals cases --- .../src/agent/blob/agentBlobServiceImpl.ts | 115 +++---- .../src/agent/blob/byteLruCache.ts | 61 ++++ .../test/blob/agentBlobService.test.ts | 219 +++++++++++++ .../test/blob/blobService.test.ts | 172 ---------- .../agent-core-v2/test/blob/blobref.test.ts | 308 ------------------ .../test/blob/byteLruCache.test.ts | 91 ++++++ 6 files changed, 417 insertions(+), 549 deletions(-) create mode 100644 packages/agent-core-v2/src/agent/blob/byteLruCache.ts create mode 100644 packages/agent-core-v2/test/blob/agentBlobService.test.ts delete mode 100644 packages/agent-core-v2/test/blob/blobService.test.ts delete mode 100644 packages/agent-core-v2/test/blob/blobref.test.ts create mode 100644 packages/agent-core-v2/test/blob/byteLruCache.test.ts diff --git a/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts b/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts index 61bd4ee51..befdf3688 100644 --- a/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts +++ b/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts @@ -18,6 +18,7 @@ import { IAgentBlobService, MISSING_MEDIA_PLACEHOLDER, } from './agentBlobService'; +import { ByteLruCache } from './byteLruCache'; const DEFAULT_THRESHOLD = 4096; const DEFAULT_MAX_CACHE_SIZE = 50 * 1024 * 1024; @@ -27,9 +28,7 @@ export class AgentBlobServiceImpl implements IAgentBlobService { declare readonly _serviceBrand: undefined; private readonly storageScope: string; - private readonly cache = new Map(); - private readonly cacheSizes = new Map(); - private currentCacheSize = 0; + private readonly cache = new ByteLruCache(DEFAULT_MAX_CACHE_SIZE); constructor( @IBlobStore private readonly blobs: IBlobStore, @@ -42,10 +41,6 @@ export class AgentBlobServiceImpl implements IAgentBlobService { return DEFAULT_THRESHOLD; } - protected get maxCacheSize(): number { - return DEFAULT_MAX_CACHE_SIZE; - } - isBlobRef(url: string): boolean { return url.startsWith(BLOBREF_PROTOCOL); } @@ -72,7 +67,21 @@ export class AgentBlobServiceImpl implements IAgentBlobService { return changed ? out : parts; } - private async offloadContentPart(part: ContentPart): Promise { + private offloadContentPart(part: ContentPart): Promise { + return this.rewriteMediaUrls(part, (url) => this.maybeOffloadString(url)); + } + + private loadContentPart(part: ContentPart): Promise { + return this.rewriteMediaUrls(part, async (url) => { + if (!this.isBlobRef(url)) return url; + return (await this.loadBlobRefUrl(url)) ?? MISSING_MEDIA_PLACEHOLDER; + }); + } + + private async rewriteMediaUrls( + part: ContentPart, + transformUrl: (url: string) => Promise, + ): Promise { let updated: Record | undefined; for (const [key, value] of Object.entries(part)) { const mediaObj = asMediaContainer(value); @@ -81,7 +90,7 @@ export class AgentBlobServiceImpl implements IAgentBlobService { const url = mediaObj.url; if (typeof url !== 'string') continue; - const newUrl = await this.maybeOffloadString(url); + const newUrl = await transformUrl(url); if (newUrl === url) continue; if (updated === undefined) updated = { ...part }; @@ -90,50 +99,26 @@ export class AgentBlobServiceImpl implements IAgentBlobService { return updated === undefined ? part : (updated as unknown as ContentPart); } - private async loadContentPart(part: ContentPart): Promise { - let updated: Record | undefined; - for (const [key, value] of Object.entries(part)) { - const mediaObj = asMediaContainer(value); - if (mediaObj === undefined) continue; - - const url = mediaObj.url; - if (typeof url !== 'string' || !this.isBlobRef(url)) continue; - - const newUrl = await this.loadBlobRefUrl(url); - if (updated === undefined) updated = { ...part }; - updated[key] = { ...(value as object), url: newUrl ?? MISSING_MEDIA_PLACEHOLDER }; - } - return updated === undefined ? part : (updated as unknown as ContentPart); - } - private async loadBlobRefUrl(url: string): Promise { - const rest = url.slice(BLOBREF_PROTOCOL.length); - const semiIdx = rest.indexOf(';'); - if (semiIdx === -1) return undefined; + const ref = parseBlobRef(url); + if (ref === undefined) return undefined; - const mimeType = rest.slice(0, semiIdx); - const hash = rest.slice(semiIdx + 1); - if (hash.length === 0) return undefined; - - const payload = await this.readBlob(hash); + const payload = await this.readBlob(ref.hash); if (payload === undefined) return undefined; - return `data:${mimeType};base64,${payload.toString('base64')}`; + return formatDataUri(ref.mimeType, payload); } private async readBlob(hash: string): Promise { const cached = this.cache.get(hash); - if (cached !== undefined) { - this.cache.delete(hash); - this.cache.set(hash, cached); - return cached; - } + if (cached !== undefined) return cached; const payload = await this.blobs.get(this.storageScope, hash).catch(() => undefined); - if (payload !== undefined) { - this.setCache(hash, Buffer.from(payload)); - } - return payload !== undefined ? Buffer.from(payload) : undefined; + if (payload === undefined) return undefined; + + const buffer = Buffer.from(payload); + this.cache.set(hash, buffer); + return buffer; } private async maybeOffloadString(value: string): Promise { @@ -153,35 +138,27 @@ export class AgentBlobServiceImpl implements IAgentBlobService { const hash = createHash('sha256').update(base64Payload, 'utf8').digest('hex'); const binary = Buffer.from(base64Payload, 'base64'); await this.blobs.put(this.storageScope, hash, binary); - this.setCache(hash, binary); - return `${BLOBREF_PROTOCOL}${mimeType};${hash}`; + this.cache.set(hash, binary); + return formatBlobRef(mimeType, hash); } +} - private setCache(hash: string, payload: Buffer): void { - const size = payload.byteLength; - if (this.cache.has(hash)) { - const oldSize = this.cacheSizes.get(hash) ?? 0; - this.currentCacheSize += size - oldSize; - this.cache.delete(hash); - } else { - if (size > this.maxCacheSize) return; - while (this.currentCacheSize + size > this.maxCacheSize && this.cache.size > 0) { - this.evictLRU(); - } - this.currentCacheSize += size; - } - this.cache.set(hash, payload); - this.cacheSizes.set(hash, size); - } +function formatBlobRef(mimeType: string, hash: string): string { + return `${BLOBREF_PROTOCOL}${mimeType};${hash}`; +} - private evictLRU(): void { - const lru = this.cache.keys().next().value; - if (lru === undefined) return; - const size = this.cacheSizes.get(lru) ?? 0; - this.currentCacheSize -= size; - this.cache.delete(lru); - this.cacheSizes.delete(lru); - } +function parseBlobRef(url: string): { mimeType: string; hash: string } | undefined { + if (!url.startsWith(BLOBREF_PROTOCOL)) return undefined; + const rest = url.slice(BLOBREF_PROTOCOL.length); + const semiIdx = rest.indexOf(';'); + if (semiIdx === -1) return undefined; + const hash = rest.slice(semiIdx + 1); + if (hash.length === 0) return undefined; + return { mimeType: rest.slice(0, semiIdx), hash }; +} + +function formatDataUri(mimeType: string, payload: Buffer): string { + return `data:${mimeType};base64,${payload.toString('base64')}`; } function asMediaContainer(value: unknown): { url: unknown } | undefined { diff --git a/packages/agent-core-v2/src/agent/blob/byteLruCache.ts b/packages/agent-core-v2/src/agent/blob/byteLruCache.ts new file mode 100644 index 000000000..02ada9c84 --- /dev/null +++ b/packages/agent-core-v2/src/agent/blob/byteLruCache.ts @@ -0,0 +1,61 @@ +/** + * `blob` domain — byte-bounded LRU cache. + * + * A small, dependency-free cache whose capacity is measured in **bytes** rather + * than entries. Hits refresh an entry to most-recently-used; inserts evict the + * least-recently-used entries until the payload fits. A single payload larger + * than `maxBytes` is never cached. + * + * Module-private helper for `AgentBlobServiceImpl`; not part of the package + * surface. Owned as a value (not a DI service) so each agent keeps its own + * cache. Promote to a shared util only when a second caller appears. + */ + +export class ByteLruCache { + private readonly map = new Map(); + private currentBytes = 0; + + constructor(private readonly maxBytes: number) {} + + get(key: string): Buffer | undefined { + const value = this.map.get(key); + if (value === undefined) return undefined; + // Refresh to most-recently-used. + this.map.delete(key); + this.map.set(key, value); + return value; + } + + set(key: string, value: Buffer): void { + const size = value.byteLength; + const existing = this.map.get(key); + + if (size > this.maxBytes) { + if (existing !== undefined) { + this.currentBytes -= existing.byteLength; + this.map.delete(key); + } + return; + } + + if (existing !== undefined) { + this.currentBytes -= existing.byteLength; + this.map.delete(key); + } else { + while (this.map.size > 0 && this.currentBytes + size > this.maxBytes) { + this.evictOldest(); + } + } + + this.currentBytes += size; + this.map.set(key, value); + } + + private evictOldest(): void { + const oldest = this.map.keys().next().value; + if (oldest === undefined) return; + const value = this.map.get(oldest)!; + this.currentBytes -= value.byteLength; + this.map.delete(oldest); + } +} diff --git a/packages/agent-core-v2/test/blob/agentBlobService.test.ts b/packages/agent-core-v2/test/blob/agentBlobService.test.ts new file mode 100644 index 000000000..315441ae9 --- /dev/null +++ b/packages/agent-core-v2/test/blob/agentBlobService.test.ts @@ -0,0 +1,219 @@ +/** + * Scenario: the agent blob service offloads large inline media (data URIs) into + * content-addressed blobs and loads them back on read. + * + * Responsibilities asserted: + * - sub-threshold data URIs pass through unchanged (and keep the same array ref) + * - large data URIs become `blobref:` URLs and are persisted under the agent scope + * - offload is non-mutating, idempotent, and handles every media container + * - load restores blobrefs, leaves other URLs alone, and substitutes a + * placeholder when the blob is missing + * - content-addressing deduplicates identical payloads and isolates per agent + * + * Wiring: real `BlobStoreService` over the in-memory storage backend, with the + * service resolved through the DI scope tree — no stubbed boundary, no real fs. + * + * Run: `pnpm test -- test/blob/agentBlobService.test.ts` + */ + +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import type { ContentPart } from '#/app/llmProtocol/message'; +import { SyncDescriptor } from '#/_base/di/descriptors'; +import { type ServiceIdentifier } from '#/_base/di/instantiation'; +import { LifecycleScope } from '#/_base/di/scope'; +import { createScopedTestHost, stubPair } from '#/_base/di/test'; +import { + BLOBREF_PROTOCOL, + IAgentBlobService, + MISSING_MEDIA_PLACEHOLDER, +} from '#/agent/blob/agentBlobService'; +import { AgentBlobServiceImpl } from '#/agent/blob/agentBlobServiceImpl'; +import { IAgentScopeContext, makeAgentScopeContext } from '#/agent/scopeContext/scopeContext'; +import { BlobStoreService } from '#/persistence/backends/node-fs/blobStoreService'; +import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; +import { IBlobStore } from '#/persistence/interface/blobStore'; +import { IFileSystemStorageService } from '#/persistence/interface/storage'; + +// The default offload threshold is 4096 base64 chars; LARGE straddles it. +const LARGE = 'A'.repeat(5000); +const SMALL = 'AQID'; + +function dataUri(mimeType: string, payload: string): string { + return `data:${mimeType};base64,${payload}`; +} + +function imagePart(url: string): ContentPart { + return { type: 'image_url', imageUrl: { url } }; +} + +function videoPart(url: string): ContentPart { + return { type: 'video_url', videoUrl: { url } }; +} + +function imageUrl(part: ContentPart): string { + return (part as { imageUrl: { url: string } }).imageUrl.url; +} + +function videoUrl(part: ContentPart): string { + return (part as { videoUrl: { url: string } }).videoUrl.url; +} + +describe('agent blob service (offload/load of inline media)', () => { + let host: ReturnType; + let blobs: IBlobStore; + + beforeEach(() => { + host = createScopedTestHost([ + stubPair(IFileSystemStorageService, new InMemoryStorageService()), + [IBlobStore as ServiceIdentifier, new SyncDescriptor(BlobStoreService, [])], + ]); + blobs = host.app.accessor.get(IBlobStore); + }); + + afterEach(() => { + host.dispose(); + }); + + function createService(agentId: string, agentScope: string): IAgentBlobService { + const agent = host.child(LifecycleScope.Agent, agentId, [ + stubPair(IAgentScopeContext, makeAgentScopeContext({ agentId, agentScope })), + [IAgentBlobService as ServiceIdentifier, new SyncDescriptor(AgentBlobServiceImpl)], + ]); + return agent.accessor.get(IAgentBlobService); + } + + function service(): IAgentBlobService { + return createService('agent', ''); + } + + it('offload leaves a sub-threshold data URI unchanged and returns the same array', async () => { + const svc = service(); + const uri = dataUri('image/png', SMALL); + const parts: ContentPart[] = [imagePart(uri)]; + + const out = await svc.offloadParts(parts); + + expect(out).toBe(parts); + expect(imageUrl(out[0]!)).toBe(uri); + }); + + it('offload rewrites a large data URI to a blobref persisted under the agent scope', async () => { + const svc = service(); + const uri = dataUri('image/png', LARGE); + + const out = await svc.offloadParts([imagePart(uri)]); + + const ref = imageUrl(out[0]!); + expect(svc.isBlobRef(ref)).toBe(true); + expect(ref.startsWith(`${BLOBREF_PROTOCOL}image/png;`)).toBe(true); + + const keys = await blobs.list('blobs'); + expect(keys).toHaveLength(1); + expect(Buffer.from((await blobs.get('blobs', keys[0]!))!).toString('base64')).toBe(LARGE); + }); + + it('offload then load restores the original data URI', async () => { + const svc = service(); + const uri = dataUri('image/jpeg', LARGE); + + const out = await svc.offloadParts([imagePart(uri)]); + const back = await svc.loadParts(out); + + expect(imageUrl(back[0]!)).toBe(uri); + }); + + it('offload does not mutate the input array or its media objects', async () => { + const svc = service(); + const uri = dataUri('image/png', LARGE); + const inner = { url: uri }; + const part = { type: 'image_url', imageUrl: inner } as ContentPart; + const parts = [part]; + + const out = await svc.offloadParts(parts); + + expect(out).not.toBe(parts); + expect(out[0]).not.toBe(part); + expect((out[0]! as { imageUrl: { url: string } }).imageUrl).not.toBe(inner); + expect(inner.url).toBe(uri); + }); + + it('offload rewrites every media container in the part list', async () => { + const svc = service(); + const uri = dataUri('image/png', LARGE); + + const out = await svc.offloadParts([imagePart(uri), videoPart(uri)]); + + expect(svc.isBlobRef(imageUrl(out[0]!))).toBe(true); + expect(svc.isBlobRef(videoUrl(out[1]!))).toBe(true); + + const back = await svc.loadParts(out); + expect(imageUrl(back[0]!)).toBe(uri); + expect(videoUrl(back[1]!)).toBe(uri); + }); + + it('offload returns the input unchanged when no part carries media', async () => { + const svc = service(); + const parts: ContentPart[] = [{ type: 'text', text: 'just text' }]; + + expect(await svc.offloadParts(parts)).toBe(parts); + }); + + it('offload leaves an existing blobref untouched', async () => { + const svc = service(); + const parts: ContentPart[] = [imagePart('blobref:image/png;deadbeef')]; + + expect(await svc.offloadParts(parts)).toBe(parts); + }); + + it('offload maps identical payloads to the same blobref and stores them once', async () => { + const svc = service(); + const uri = dataUri('image/png', LARGE); + + const first = await svc.offloadParts([imagePart(uri)]); + const second = await svc.offloadParts([imagePart(uri)]); + + expect(imageUrl(first[0]!)).toBe(imageUrl(second[0]!)); + expect(await blobs.list('blobs')).toHaveLength(1); + }); + + it('offload isolates blobs per agent scope so agents do not collide', async () => { + const a1 = createService('a1', 'sessions/s1/agents/a1'); + const a2 = createService('a2', 'sessions/s1/agents/a2'); + const uri = dataUri('image/png', LARGE); + + const out1 = await a1.offloadParts([imagePart(uri)]); + const out2 = await a2.offloadParts([imagePart(uri)]); + + expect(await blobs.list('sessions/s1/agents/a1/blobs')).toHaveLength(1); + expect(await blobs.list('sessions/s1/agents/a2/blobs')).toHaveLength(1); + expect(imageUrl((await a1.loadParts(out1))[0]!)).toBe(uri); + expect(imageUrl((await a2.loadParts(out2))[0]!)).toBe(uri); + }); + + it('load leaves non-blobref URLs unchanged and returns the same array', async () => { + const svc = service(); + const parts: ContentPart[] = [ + imagePart('https://example.com/a.png'), + imagePart(dataUri('image/png', SMALL)), + ]; + + expect(await svc.loadParts(parts)).toBe(parts); + }); + + it('load substitutes the missing-media placeholder when the blob is absent', async () => { + const svc = service(); + + const out = await svc.loadParts([imagePart('blobref:image/png;deadbeef')]); + + expect(imageUrl(out[0]!)).toBe(MISSING_MEDIA_PLACEHOLDER); + }); + + it('isBlobRef recognizes only the blobref protocol', () => { + const svc = service(); + + expect(svc.isBlobRef('blobref:image/png;abc')).toBe(true); + expect(svc.isBlobRef('data:image/png;base64,AQID')).toBe(false); + expect(svc.isBlobRef('https://example.com/a.png')).toBe(false); + }); +}); diff --git a/packages/agent-core-v2/test/blob/blobService.test.ts b/packages/agent-core-v2/test/blob/blobService.test.ts deleted file mode 100644 index a16f8ed2e..000000000 --- a/packages/agent-core-v2/test/blob/blobService.test.ts +++ /dev/null @@ -1,172 +0,0 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; - -import type { ContentPart } from '#/app/llmProtocol/message'; -import { SyncDescriptor } from '#/_base/di/descriptors'; -import { type ServiceIdentifier } from '#/_base/di/instantiation'; -import { LifecycleScope } from '#/_base/di/scope'; -import { createScopedTestHost, stubPair } from '#/_base/di/test'; -import { IAgentBlobService } from '#/agent/blob/agentBlobService'; -import { AgentBlobServiceImpl } from '#/agent/blob/agentBlobServiceImpl'; -import { IAgentScopeContext, makeAgentScopeContext } from '#/agent/scopeContext/scopeContext'; -import { IBootstrapService } from '#/app/bootstrap/bootstrap'; -import { IFileSystemStorageService } from '#/persistence/interface/storage'; -import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; -import { IBlobStore } from '#/persistence/interface/blobStore'; -import { BlobStoreService } from '#/persistence/backends/node-fs/blobStoreService'; - -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('AgentBlobServiceImpl', () => { - let host: ReturnType; - - beforeEach(() => { - host = createScopedTestHost([ - stubPair(IFileSystemStorageService, new InMemoryStorageService()), - stubPair(IBootstrapService, { homeDir: '/home' } as unknown as IBootstrapService), - [IBlobStore as ServiceIdentifier, new SyncDescriptor(BlobStoreService, [])], - ]); - }); - - afterEach(() => { - host.dispose(); - }); - - function getBlobStore(): IAgentBlobService { - const agent = host.child(LifecycleScope.Agent, 'test-agent', [ - [ - IAgentScopeContext as ServiceIdentifier, - makeAgentScopeContext({ agentId: 'test-agent', agentScope: '' }), - ], - [ - IAgentBlobService as ServiceIdentifier, - new SyncDescriptor(AgentBlobServiceImpl), - ], - ]); - return agent.accessor.get(IAgentBlobService); - } - - 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.app.accessor.get(IFileSystemStorageService); - 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.loadParts(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.loadParts(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.loadParts(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, - ); - }); - - it('persists blobs under the per-agent scope when homedir is seeded', async () => { - const agent = host.child(LifecycleScope.Agent, 'a1', [ - [ - IAgentScopeContext as ServiceIdentifier, - makeAgentScopeContext({ - agentId: 'a1', - agentScope: 'sessions/s1/agents/a1', - }), - ], - [ - IAgentBlobService as ServiceIdentifier, - new SyncDescriptor(AgentBlobServiceImpl), - ], - ]); - const store = agent.accessor.get(IAgentBlobService); - const { uri, payload } = makeLargeDataUri(); - const parts: ContentPart[] = [{ type: 'image_url', imageUrl: { url: uri } }]; - - const offloaded = await store.offloadParts(parts); - expect(store.isBlobRef((offloaded[0]! as { imageUrl: { url: string } }).imageUrl.url)).toBe(true); - - const backend = host.app.accessor.get(IFileSystemStorageService); - const perAgentScope = 'sessions/s1/agents/a1/blobs'; - const perAgentKeys = await backend.list(perAgentScope); - expect(perAgentKeys).toHaveLength(1); - expect( - Buffer.from((await backend.read(perAgentScope, perAgentKeys[0]!))!).toString('base64'), - ).toBe(payload); - expect(await backend.list('blobs')).toHaveLength(0); - - const rehydrated = await store.loadParts(offloaded); - expect((rehydrated[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe(uri); - }); -}); diff --git a/packages/agent-core-v2/test/blob/blobref.test.ts b/packages/agent-core-v2/test/blob/blobref.test.ts deleted file mode 100644 index 6265e5fc9..000000000 --- a/packages/agent-core-v2/test/blob/blobref.test.ts +++ /dev/null @@ -1,308 +0,0 @@ -import { randomBytes } from 'node:crypto'; -import { mkdir, readdir, readFile, rm } from 'node:fs/promises'; -import { tmpdir } from 'node:os'; -import { join } from 'pathe'; - -import type { ContentPart } from '#/app/llmProtocol/message'; -import { afterEach, describe, expect, it } from 'vitest'; - -import { SyncDescriptor } from '#/_base/di/descriptors'; -import { DisposableStore } from '#/_base/di/lifecycle'; -import { TestInstantiationService } from '#/_base/di/test'; -import { - BLOBREF_PROTOCOL, - IAgentBlobService, - MISSING_MEDIA_PLACEHOLDER, -} from '#/agent/blob/agentBlobService'; -import { AgentBlobServiceImpl } from '#/agent/blob/agentBlobServiceImpl'; -import { IBootstrapService } from '#/app/bootstrap/bootstrap'; -import { IAgentScopeContext, makeAgentScopeContext } from '#/agent/scopeContext/scopeContext'; -import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; -import { IFileSystemStorageService } from '#/persistence/interface/storage'; -import { IBlobStore } from '#/persistence/interface/blobStore'; -import { BlobStoreService } from '#/persistence/backends/node-fs/blobStoreService'; - -const cleanups: string[] = []; -const disposables: DisposableStore[] = []; - -afterEach(async () => { - for (const store of disposables.splice(0)) { - store.dispose(); - } - for (const dir of cleanups.splice(0)) { - await rm(dir, { recursive: true, force: true }).catch(() => {}); - } -}); - -function imagePart(url: string): ContentPart { - return { type: 'image_url', imageUrl: { url } } as ContentPart; -} - -function videoPart(url: string): ContentPart { - return { type: 'video_url', videoUrl: { url } } as ContentPart; -} - -function firstImageUrl(parts: readonly ContentPart[]): string { - return (parts[0] as unknown as { imageUrl: { url: string } }).imageUrl.url; -} - -function secondVideoUrl(parts: readonly ContentPart[]): string { - return (parts[1] as unknown as { videoUrl: { url: string } }).videoUrl.url; -} - -function imageUrlObject(part: ContentPart): { url: string } { - return (part as unknown as { imageUrl: { url: string } }).imageUrl; -} - -async function makeHomeDir(): Promise<{ homeDir: string; blobsDir: string }> { - const homeDir = join(tmpdir(), `blobref-test-${randomBytes(6).toString('hex')}`); - await mkdir(homeDir, { recursive: true }); - cleanups.push(homeDir); - return { homeDir, blobsDir: join(homeDir, 'blobs') }; -} - -function createStore( - homeDir: string, - ctor: typeof AgentBlobServiceImpl = AgentBlobServiceImpl, -): IAgentBlobService { - const disposable = new DisposableStore(); - disposables.push(disposable); - - const ix = disposable.add(new TestInstantiationService()); - ix.set(IFileSystemStorageService, new FileStorageService(homeDir)); - ix.set(IBlobStore, new SyncDescriptor(BlobStoreService, [])); - ix.set(IBootstrapService, { homeDir } as unknown as IBootstrapService); - ix.set( - IAgentScopeContext, - makeAgentScopeContext({ agentId: 'test', agentScope: '' }), - ); - ix.set(IAgentBlobService, new SyncDescriptor(ctor)); - return ix.get(IAgentBlobService); -} - -async function makeStore(): Promise<{ store: IAgentBlobService; blobsDir: string }> { - const { homeDir, blobsDir } = await makeHomeDir(); - return { - store: createStore(homeDir), - blobsDir, - }; -} - -class TwoBlobCacheStoreService extends AgentBlobServiceImpl { - protected override get maxCacheSize(): number { - return 8_000; - } -} - -class OneBlobCacheStoreService extends AgentBlobServiceImpl { - protected override get maxCacheSize(): number { - return 4_000; - } -} - -describe('blobref', () => { - it('offloads large data URIs and replaces with blobref', async () => { - const { store, blobsDir } = await makeStore(); - const payload = 'A'.repeat(5000); - const dataUri = `data:image/png;base64,${payload}`; - - const offloaded = await store.offloadParts([imagePart(dataUri)]); - const url = firstImageUrl(offloaded); - - expect(store.isBlobRef(url)).toBe(true); - expect(url.startsWith(BLOBREF_PROTOCOL)).toBe(true); - expect(url.startsWith('blobref:image/png;')).toBe(true); - - const files = await readdir(blobsDir); - expect(files).toHaveLength(1); - expect((await readFile(join(blobsDir, files[0]!))).toString('base64')).toBe(payload); - }); - - it('does not mutate the input array or content parts', async () => { - const { store } = await makeStore(); - const payload = 'M'.repeat(5000); - const dataUri = `data:image/png;base64,${payload}`; - const innerImageUrl = { url: dataUri }; - const part = { type: 'image_url', imageUrl: innerImageUrl } as unknown as ContentPart; - const parts = [part]; - - const offloaded = await store.offloadParts(parts); - - expect(parts[0]).toBe(part); - expect(imageUrlObject(part)).toBe(innerImageUrl); - expect(innerImageUrl.url).toBe(dataUri); - - expect(offloaded).not.toBe(parts); - expect(offloaded[0]).not.toBe(part); - expect(imageUrlObject(offloaded[0]!)).not.toBe(innerImageUrl); - expect(firstImageUrl(offloaded).startsWith('blobref:image/png;')).toBe(true); - }); - - it('offloads every media container in content parts', async () => { - const { store, blobsDir } = await makeStore(); - const payload = 'X'.repeat(5000); - const dataUri = `data:image/png;base64,${payload}`; - const parts = [imagePart(dataUri), videoPart(dataUri)]; - - const offloaded = await store.offloadParts(parts); - - expect(firstImageUrl(offloaded).startsWith('blobref:image/png;')).toBe(true); - expect(secondVideoUrl(offloaded).startsWith('blobref:image/png;')).toBe(true); - expect(firstImageUrl(offloaded)).toBe(secondVideoUrl(offloaded)); - - const files = await readdir(blobsDir); - expect(files).toHaveLength(1); - }); - - it('returns the same array reference when nothing needs offloading', async () => { - const { store } = await makeStore(); - const parts: readonly ContentPart[] = [{ type: 'text', text: 'just text' }]; - - const offloaded = await store.offloadParts(parts); - expect(offloaded).toBe(parts); - }); - - it('skips small data URIs below threshold', async () => { - const { store, blobsDir } = await makeStore(); - const payload = 'short'; - const dataUri = `data:image/png;base64,${payload}`; - const parts = [imagePart(dataUri)]; - - const offloaded = await store.offloadParts(parts); - - expect(offloaded).toBe(parts); - const files = await readdir(blobsDir).catch(() => []); - expect(files).toHaveLength(0); - }); - - it('skips existing blobrefs during offload', async () => { - const { store } = await makeStore(); - const parts = [imagePart('blobref:image/png;abc')]; - - const offloaded = await store.offloadParts(parts); - - expect(offloaded).toBe(parts); - }); - - it('rehydrates blobrefs back to data URIs', async () => { - const { store } = await makeStore(); - const payload = 'B'.repeat(5000); - const dataUri = `data:image/jpeg;base64,${payload}`; - - const offloaded = await store.offloadParts([imagePart(dataUri)]); - const rehydrated = await store.loadParts(offloaded); - - expect(firstImageUrl(rehydrated)).toBe(dataUri); - expect(firstImageUrl(offloaded)).toMatch(/^blobref:image\/jpeg;/); - }); - - it('replaces missing blobs with placeholder text', async () => { - const { store } = await makeStore(); - const rehydrated = await store.loadParts([imagePart('blobref:image/png;deadbeef')]); - - expect(firstImageUrl(rehydrated)).toBe(MISSING_MEDIA_PLACEHOLDER); - }); - - it('deduplicates identical payloads by hash', async () => { - const { store, blobsDir } = await makeStore(); - const payload = 'C'.repeat(5000); - const dataUri = `data:image/png;base64,${payload}`; - - await store.offloadParts([imagePart(dataUri)]); - await store.offloadParts([imagePart(dataUri)]); - - const files = await readdir(blobsDir); - expect(files).toHaveLength(1); - }); - - it('rehydrates from write-through cache after blob file is deleted', async () => { - const { store, blobsDir } = await makeStore(); - const payload = 'E'.repeat(5000); - const dataUri = `data:image/png;base64,${payload}`; - - const offloaded = await store.offloadParts([imagePart(dataUri)]); - const files = await readdir(blobsDir); - expect(files).toHaveLength(1); - await rm(join(blobsDir, files[0]!)); - - const rehydrated = await store.loadParts(offloaded); - expect(firstImageUrl(rehydrated)).toBe(dataUri); - }); - - it('rehydrates from read cache after first disk read', async () => { - const { homeDir, blobsDir } = await makeHomeDir(); - const writer = createStore(homeDir); - const reader = createStore(homeDir); - const payload = 'F'.repeat(5000); - const dataUri = `data:image/png;base64,${payload}`; - - const offloaded = await writer.offloadParts([imagePart(dataUri)]); - const blobref = firstImageUrl(offloaded); - - const firstRead = await reader.loadParts([imagePart(blobref)]); - expect(firstImageUrl(firstRead)).toBe(dataUri); - - const files = await readdir(blobsDir); - expect(files).toHaveLength(1); - await rm(join(blobsDir, files[0]!)); - - const secondRead = await reader.loadParts([imagePart(blobref)]); - expect(firstImageUrl(secondRead)).toBe(dataUri); - }); - - it('evicts least-recently-used entries when cache size limit is exceeded', async () => { - const { homeDir, blobsDir } = await makeHomeDir(); - const store = createStore(homeDir, TwoBlobCacheStoreService); - const payloadA = 'A'.repeat(5000); - const payloadB = 'B'.repeat(5000); - const payloadC = 'C'.repeat(5000); - - const offloadedA = await store.offloadParts([imagePart(`data:image/png;base64,${payloadA}`)]); - const offloadedB = await store.offloadParts([imagePart(`data:image/png;base64,${payloadB}`)]); - const blobrefA = firstImageUrl(offloadedA); - const blobrefB = firstImageUrl(offloadedB); - - await store.loadParts([imagePart(blobrefA)]); - const offloadedC = await store.offloadParts([imagePart(`data:image/png;base64,${payloadC}`)]); - const blobrefC = firstImageUrl(offloadedC); - - const files = await readdir(blobsDir); - for (const file of files) { - await rm(join(blobsDir, file)); - } - - expect(firstImageUrl(await store.loadParts([imagePart(blobrefA)]))).toBe( - `data:image/png;base64,${payloadA}`, - ); - expect(firstImageUrl(await store.loadParts([imagePart(blobrefB)]))).toBe( - MISSING_MEDIA_PLACEHOLDER, - ); - expect(firstImageUrl(await store.loadParts([imagePart(blobrefC)]))).toBe( - `data:image/png;base64,${payloadC}`, - ); - }); - - it('skips caching a blob larger than the entire cache cap', async () => { - const { homeDir, blobsDir } = await makeHomeDir(); - const store = createStore(homeDir, OneBlobCacheStoreService); - const small = 'S'.repeat(5000); - const large = 'L'.repeat(10000); - - const offloadedSmall = await store.offloadParts([imagePart(`data:image/png;base64,${small}`)]); - const offloadedLarge = await store.offloadParts([imagePart(`data:image/png;base64,${large}`)]); - const smallBlobref = firstImageUrl(offloadedSmall); - const largeBlobref = firstImageUrl(offloadedLarge); - - const files = await readdir(blobsDir); - for (const file of files) { - await rm(join(blobsDir, file)); - } - - expect(firstImageUrl(await store.loadParts([imagePart(smallBlobref)]))).toBe( - `data:image/png;base64,${small}`, - ); - expect(firstImageUrl(await store.loadParts([imagePart(largeBlobref)]))).toBe( - MISSING_MEDIA_PLACEHOLDER, - ); - }); -}); diff --git a/packages/agent-core-v2/test/blob/byteLruCache.test.ts b/packages/agent-core-v2/test/blob/byteLruCache.test.ts new file mode 100644 index 000000000..a4c380805 --- /dev/null +++ b/packages/agent-core-v2/test/blob/byteLruCache.test.ts @@ -0,0 +1,91 @@ +/** + * Scenario: the byte-bounded LRU cache used by the agent blob service. + * + * Responsibilities asserted: hit returns the stored value, miss is undefined, + * least-recently-used eviction on overflow, recency refresh on get, oversize + * payloads are never cached, replacement re-accounts size, and multiple entries + * evict to make room. Pure data-structure tests — no DI, no IO. + * + * Run: `pnpm test -- test/blob/byteLruCache.test.ts` + */ + +import { describe, expect, it } from 'vitest'; + +import { ByteLruCache } from '#/agent/blob/byteLruCache'; + +const buf = (n: number): Buffer => Buffer.alloc(n); + +describe('ByteLruCache', () => { + it('returns the stored buffer on a hit', () => { + const cache = new ByteLruCache(16); + cache.set('a', Buffer.from('hello')); + + expect(cache.get('a')?.equals(Buffer.from('hello'))).toBe(true); + }); + + it('returns undefined for a missing key', () => { + const cache = new ByteLruCache(16); + + expect(cache.get('nope')).toBeUndefined(); + }); + + it('evicts the least-recently-used entry when capacity is exceeded', () => { + const cache = new ByteLruCache(10); + cache.set('a', buf(5)); + cache.set('b', buf(5)); + + cache.set('c', buf(5)); + + expect(cache.get('a')).toBeUndefined(); + expect(cache.get('b')).toBeDefined(); + expect(cache.get('c')).toBeDefined(); + }); + + it('refreshes recency on get so a read entry survives eviction', () => { + const cache = new ByteLruCache(10); + cache.set('a', buf(5)); + cache.set('b', buf(5)); + + cache.get('a'); + cache.set('c', buf(5)); + + expect(cache.get('b')).toBeUndefined(); + expect(cache.get('a')).toBeDefined(); + expect(cache.get('c')).toBeDefined(); + }); + + it('does not cache a payload larger than maxBytes and keeps existing entries', () => { + const cache = new ByteLruCache(10); + cache.set('a', buf(5)); + + cache.set('big', buf(11)); + + expect(cache.get('big')).toBeUndefined(); + expect(cache.get('a')).toBeDefined(); + }); + + it('re-accounts size when an existing key is replaced', () => { + const cache = new ByteLruCache(10); + cache.set('a', buf(4)); + cache.set('a', buf(9)); + + cache.set('b', buf(2)); + + expect(cache.get('a')).toBeUndefined(); + expect(cache.get('b')).toBeDefined(); + }); + + it('evicts multiple entries to make room for a larger payload', () => { + const cache = new ByteLruCache(10); + cache.set('a', buf(3)); + cache.set('b', buf(3)); + cache.set('c', buf(3)); + + cache.set('d', buf(5)); + + expect(cache.get('a')).toBeUndefined(); + expect(cache.get('b')).toBeUndefined(); + expect(cache.get('c')).toBeDefined(); + expect(cache.get('d')).toBeDefined(); + }); +});