vrr/kimi-code
2
0
Fork 0
mirror of https://github.com/MoonshotAI/kimi-code.git synced 2026-07-10 01:39:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor(agent-core-v2): split blob service helpers, rewrite tests

Browse source 
- 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
This commit is contained in:
haozhe.yang 2026-07-09 11:22:17 +08:00
parent 5355756652
commit 667898b9df
6 changed files with 417 additions and 549 deletions
Download patch file Download diff file Expand all files Collapse all files

115
packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts
View file

@ -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<string, Buffer>();
private readonly cacheSizes = new Map<string, number>();
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<ContentPart> {
private offloadContentPart(part: ContentPart): Promise<ContentPart> {
return this.rewriteMediaUrls(part, (url) => this.maybeOffloadString(url));
}
private loadContentPart(part: ContentPart): Promise<ContentPart> {
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<string>,
): Promise<ContentPart> {
let updated: Record<string, unknown> | 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<ContentPart> {
let updated: Record<string, unknown> | 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<string | undefined> {
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<Buffer | undefined> {
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<string> {
@ -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 {

61
packages/agent-core-v2/src/agent/blob/byteLruCache.ts Normal file
View file

@ -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<string, Buffer>();
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);
}
}

219
packages/agent-core-v2/test/blob/agentBlobService.test.ts Normal file
View file

@ -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<typeof createScopedTestHost>;
let blobs: IBlobStore;
beforeEach(() => {
host = createScopedTestHost([
stubPair(IFileSystemStorageService, new InMemoryStorageService()),
[IBlobStore as ServiceIdentifier<unknown>, 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<unknown>, 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);
});
});

172
packages/agent-core-v2/test/blob/blobService.test.ts
View file

@ -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<typeof createScopedTestHost>;
beforeEach(() => {
host = createScopedTestHost([
stubPair(IFileSystemStorageService, new InMemoryStorageService()),
stubPair(IBootstrapService, { homeDir: '/home' } as unknown as IBootstrapService),
[IBlobStore as ServiceIdentifier<unknown>, new SyncDescriptor(BlobStoreService, [])],
]);
});
afterEach(() => {
host.dispose();
});
function getBlobStore(): IAgentBlobService {
const agent = host.child(LifecycleScope.Agent, 'test-agent', [
[
IAgentScopeContext as ServiceIdentifier<unknown>,
makeAgentScopeContext({ agentId: 'test-agent', agentScope: '' }),
],
[
IAgentBlobService as ServiceIdentifier<unknown>,
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<unknown>,
makeAgentScopeContext({
agentId: 'a1',
agentScope: 'sessions/s1/agents/a1',
}),
],
[
IAgentBlobService as ServiceIdentifier<unknown>,
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);
});
});

308
packages/agent-core-v2/test/blob/blobref.test.ts
View file

@ -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,
);
});
});

91
packages/agent-core-v2/test/blob/byteLruCache.test.ts Normal file
View file

@ -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();
});
});