diff --git a/packages/agent-core-v2/src/agent/blob/agentBlobService.ts b/packages/agent-core-v2/src/agent/blob/agentBlobService.ts index 591dce5fb..585bbbbd5 100644 --- a/packages/agent-core-v2/src/agent/blob/agentBlobService.ts +++ b/packages/agent-core-v2/src/agent/blob/agentBlobService.ts @@ -2,7 +2,7 @@ * `blob` domain — `IAgentBlobService` contract. * * Offloads large inline media payloads to content-addressed blob storage and - * rehydrates them on read. Bound at Agent scope. + * loads them back on read. Bound at Agent scope. */ import type { ContentPart } from '#/app/llmProtocol'; @@ -16,7 +16,7 @@ export interface IAgentBlobService { readonly _serviceBrand: undefined; offloadParts(parts: readonly ContentPart[]): Promise; - rehydrateParts(parts: readonly ContentPart[]): Promise; + loadParts(parts: readonly ContentPart[]): Promise; isBlobRef(url: string): boolean; } diff --git a/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts b/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts index f35e3749f..9ef97cef2 100644 --- a/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts +++ b/packages/agent-core-v2/src/agent/blob/agentBlobServiceImpl.ts @@ -2,7 +2,7 @@ * `blob` domain — `IAgentBlobService` implementation. * * Offloads large inline media payloads into content-addressed blobs and - * rehydrates them on read; persists bytes through `IBlobStore` under the + * loads them back on read; persists bytes through `IBlobStore` under the * agent's `scope('blobs')` root, matching the v1 `/blobs/` * layout. Bound at Agent scope. */ @@ -61,11 +61,11 @@ export class AgentBlobServiceImpl implements IAgentBlobService { return changed ? out : parts; } - async rehydrateParts(parts: readonly ContentPart[]): Promise { + async loadParts(parts: readonly ContentPart[]): Promise { let changed = false; const out: ContentPart[] = []; for (const part of parts) { - const next = await this.rehydrateContentPart(part); + const next = await this.loadContentPart(part); if (next !== part) changed = true; out.push(next); } @@ -90,7 +90,7 @@ export class AgentBlobServiceImpl implements IAgentBlobService { return updated === undefined ? part : (updated as unknown as ContentPart); } - private async rehydrateContentPart(part: ContentPart): Promise { + private async loadContentPart(part: ContentPart): Promise { let updated: Record | undefined; for (const [key, value] of Object.entries(part)) { const mediaObj = asMediaContainer(value); @@ -99,14 +99,14 @@ export class AgentBlobServiceImpl implements IAgentBlobService { const url = mediaObj.url; if (typeof url !== 'string' || !this.isBlobRef(url)) continue; - const newUrl = await this.rehydrateBlobRefUrl(url); + 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 rehydrateBlobRefUrl(url: string): Promise { + private async loadBlobRefUrl(url: string): Promise { const rest = url.slice(BLOBREF_PROTOCOL.length); const semiIdx = rest.indexOf(';'); if (semiIdx === -1) return undefined; diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts index d01ea715b..7cda8fcd8 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts @@ -9,8 +9,8 @@ * the Model silently and never invokes these methods), so existing subscribers * (micro-compaction, context-injector, task-notification) observe the same * splice-shaped change events regardless of which 1.4 Op was persisted. Message - * ids are stamped at the dispatch call site so `apply` stays pure. Blob offload - * lives in the `WireService` hook seeded with `contextBlobSelector`. Bound at + * ids are stamped at the dispatch call site so `apply` stays pure. Blob + * dehydrate/rehydrate is declared on `ContextModel.blobs`. Bound at * Agent scope. */ diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts index 83f53cc20..e49fb8af2 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts @@ -3,8 +3,7 @@ * 1.4 Ops `context.append_message` (`contextAppendMessage`) / `context.clear` * (`contextClear`) / `context.apply_compaction` (`contextApplyCompaction`) / * `context.undo` (`contextUndo`) for the per-agent conversation history, plus the - * legacy `context.splice` (`contextSplice`) Op and the `contextBlobSelector` that - * drives blob offload for persisted message parts. + * legacy `context.splice` (`contextSplice`) Op. * * Declares the history as `ContextMessage[]` (initial `[]`); every Op's `apply` * is a pure array transform that returns a NEW reference on change and the SAME @@ -23,18 +22,18 @@ * still replay (newer-version passthrough, no migration) and for the few internal * single-delete mutations that have no 1.4 spelling. * - * Blob handling uses two complementary mechanisms: - * - `contextBlobSelector` (record-level): offloads oversized content parts to - * blob storage on append, replacing data URIs with `blobref:` references. - * - `ContextModel.rehydrate` (model-level): after replay, traverses the - * surviving final state and rehydrates `blobref:` URLs back to inline data - * URIs — skipping I/O for data that was compacted away during the session. - * - * The selector is seeded into the Agent wire by `agentLifecycle`. + * Blob handling is declared as a `ModelBlobCodec` on `ContextModel.blobs`: + * - `dehydrate(record, transform)`: at dispatch time, traverses message content + * in `context.splice` and `context.append_message` records, passing each + * `ContentPart[]` through `transform` to offload oversized data URIs. + * - `rehydrate(state, transform)`: after replay, traverses the surviving final + * state and loads `blobref:` URLs back to inline data — skipping I/O for + * data that was compacted away during the session. */ import type { ContentPart } from '#/app/llmProtocol'; -import { defineModel, defineOp, type WireBlobSelector } from '#/wire'; +import { defineModel, defineOp, type PartsTransformer } from '#/wire'; +import type { PersistedRecord } from '#/wire'; import { foldAppendMessage, @@ -44,20 +43,51 @@ import { } from './loopEventFold'; import type { ContextMessage } from './types'; -export const ContextModel = defineModel('contextMemory', () => [], { - rehydrate: async (state, rehydrateParts) => { - let changed = false; - const result: ContextMessage[] = []; - for (const msg of state) { - const parts = await rehydrateParts(msg.content); - if (parts !== msg.content) { - changed = true; - result.push({ ...msg, content: [...parts] as ContentPart[] }); - } else { - result.push(msg); - } +async function dehydrateMessages( + messages: readonly ContextMessage[], + transform: PartsTransformer, +): Promise<{ changed: boolean; result: ContextMessage[] }> { + let changed = false; + const result: ContextMessage[] = []; + for (const msg of messages) { + const parts = await transform(msg.content); + if (parts !== msg.content) { + changed = true; + result.push({ ...msg, content: [...parts] as ContentPart[] }); + } else { + result.push(msg); } - return changed ? result : state; + } + return { changed, result }; +} + +async function dehydrateRecord( + record: PersistedRecord, + transform: PartsTransformer, +): Promise { + if (record.type === 'context.splice') { + const messages = record['messages']; + if (!Array.isArray(messages)) return record; + const { changed, result } = await dehydrateMessages(messages as ContextMessage[], transform); + return changed ? { ...record, messages: result } : record; + } + if (record.type === 'context.append_message') { + const message = record['message'] as ContextMessage | undefined; + if (message === undefined) return record; + const parts = await transform(message.content); + if (parts === message.content) return record; + return { ...record, message: { ...message, content: [...parts] } }; + } + return record; +} + +export const ContextModel = defineModel('contextMemory', () => [], { + blobs: { + dehydrate: dehydrateRecord, + rehydrate: async (state, transform) => { + const { changed, result } = await dehydrateMessages(state, transform); + return changed ? result : state; + }, }, }); @@ -174,33 +204,3 @@ function isRealUserPrompt(message: ContextMessage): boolean { origin.trigger === 'user-slash' ); } - -export const contextBlobSelector: WireBlobSelector = (record) => { - if (record.type === 'context.splice') { - const messages = record['messages']; - if (!Array.isArray(messages)) return []; - return (messages as readonly ContextMessage[]).map((message, index) => ({ - parts: message.content, - replace: (current, parts) => ({ - ...current, - messages: (current['messages'] as readonly ContextMessage[]).map((item, itemIndex) => - itemIndex === index ? { ...item, content: [...parts] } : item, - ), - }), - })); - } - if (record.type === 'context.append_message') { - const message = record['message'] as ContextMessage | undefined; - if (message === undefined) return []; - return [ - { - parts: message.content, - replace: (current, parts) => ({ - ...current, - message: { ...(current['message'] as ContextMessage), content: [...parts] }, - }), - }, - ]; - } - return []; -}; diff --git a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts index 5d53cbc52..2c45760bf 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts @@ -257,7 +257,7 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco let current = record; for (const selector of [...selectors] as BlobSelector[]) { for (const target of selector(current)) { - const parts = await blobStore.rehydrateParts(target.parts); + const parts = await blobStore.loadParts(target.parts); if (parts !== target.parts) { current = target.replace(current, parts); } diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index d94fc6efc..8ffd32331 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -42,7 +42,7 @@ import { ISessionMetadata } from '#/session/sessionMetadata'; import { ISessionWorkspaceContext } from '#/session/workspaceContext'; import { IAgentScopeContext } from '#/agent/scopeContext'; import { IAgentProfileService } from '#/agent/profile'; -import { contextBlobSelector, IAgentContextMemoryService } from '#/agent/contextMemory'; +import { IAgentContextMemoryService } from '#/agent/contextMemory'; import { IAgentBuiltinToolsRegistrar } from '#/agent/toolRegistry'; import { AGENT_WIRE_PROTOCOL_VERSION, @@ -141,7 +141,7 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle } satisfies IAgentScopeContext, ], [IAgentWireRecordService, new SyncDescriptor(AgentWireRecordService, [{ homedir: agentHomedir }])], - [IAgentWireService, new SyncDescriptor(WireService, [{ logScope: agentScope, logKey: WIRE_RECORD_FILENAME, blobSelector: contextBlobSelector }])], + [IAgentWireService, new SyncDescriptor(WireService, [{ logScope: agentScope, logKey: WIRE_RECORD_FILENAME }])], [IAgentBlobService, new SyncDescriptor(AgentBlobServiceImpl)], [ IAgentMcpService, diff --git a/packages/agent-core-v2/src/wire/model.ts b/packages/agent-core-v2/src/wire/model.ts index 2a0c3fba7..f0699e5ed 100644 --- a/packages/agent-core-v2/src/wire/model.ts +++ b/packages/agent-core-v2/src/wire/model.ts @@ -1,23 +1,30 @@ /** * `wire` domain (L2) — Model definition primitive (`ModelDef` / `defineModel`), * `DeepReadonly` (the compile-time half of immutability), and the - * `ModelRehydrateFn` / `PartsRehydrator` types that let a model declare how to - * rehydrate blob references in its state after replay. + * `ModelBlobCodec` / `PartsTransformer` types that let a model declare how to + * dehydrate large inline media before persistence and rehydrate blob references + * in its state after replay. * * A `ModelDef` is a stateless descriptor: it names a model and manufactures its * initial state via `initial`. It never holds state itself — per-scope state * instances are owned by `IWireService`, and domain services read them through - * `wire.getModel(model)`. The optional `rehydrate` function declares how to - * traverse the model's state and replace blob references with inline data after - * replay — only models whose state contains `ContentPart[]` (e.g. ContextModel) - * need it; all others leave it undefined (no-op). `WireService.replay` applies - * all records first (blobref URLs enter the model state as-is, zero I/O), then - * calls `rehydrate` on each model that declares it — so only the *surviving* - * state is rehydrated, skipping data that was later removed by compaction. + * `wire.getModel(model)`. The optional `blobs` codec declares both directions + * of the blob offload pipeline: + * - `dehydrate(record, transform)`: called per-record at dispatch time; the + * model traverses its record structure, passes each `ContentPart[]` through + * `transform` (which offloads oversized data URIs to blob storage and returns + * parts with `blobref:` URLs), and returns the transformed record. + * - `rehydrate(state, transform)`: called once after replay; the model + * traverses the surviving final state, passes each `ContentPart[]` through + * `transform` (which loads blob references back to inline data URIs), and + * returns the transformed state. Only the *surviving* state is rehydrated, + * skipping data that was later removed by compaction. * - * `PartsRehydrator` uses `readonly unknown[]` rather than `ContentPart[]` so + * Both directions receive a `PartsTransformer` — the same function shape — so + * the model owns the traversal logic and `WireService` owns the storage I/O. + * `PartsTransformer` uses `readonly unknown[]` rather than `ContentPart[]` so * this file stays free of `app/llmProtocol` imports (L2 → L3 boundary); the - * cast happens once inside `WireService.rehydrateModels`. + * cast happens once inside `WireService`. * * `DeepReadonly` recursively maps a state type to its deeply-readonly view * for the references returned by `getModel` / `subscribe`: functions pass @@ -27,17 +34,19 @@ * applied by `WireService` after every `apply`. Scope-agnostic. */ -export type PartsRehydrator = (parts: readonly unknown[]) => Promise; +import type { PersistedRecord } from './wireService'; -export type ModelRehydrateFn = ( - state: S, - rehydrateParts: PartsRehydrator, -) => S | Promise; +export type PartsTransformer = (parts: readonly unknown[]) => Promise; + +export interface ModelBlobCodec { + dehydrate(record: PersistedRecord, transform: PartsTransformer): PersistedRecord | Promise; + rehydrate(state: S, transform: PartsTransformer): S | Promise; +} export interface ModelDef { readonly name: string; readonly initial: () => S; - readonly rehydrate?: ModelRehydrateFn; + readonly blobs?: ModelBlobCodec; } export interface DerivedModelDef { @@ -45,24 +54,24 @@ export interface DerivedModelDef { readonly initial: () => S; // eslint-disable-next-line @typescript-eslint/no-explicit-any readonly reducers: Readonly S>>; - readonly rehydrate?: ModelRehydrateFn; + readonly blobs?: ModelBlobCodec; } export function defineModel( name: string, initial: () => S, - opts?: { rehydrate?: ModelRehydrateFn }, + opts?: { blobs?: ModelBlobCodec }, ): ModelDef { - return { name, initial, rehydrate: opts?.rehydrate }; + return { name, initial, blobs: opts?.blobs }; } export function defineDerivedModel( name: string, initial: () => S, reducers: Record S>, - opts?: { rehydrate?: ModelRehydrateFn }, + opts?: { blobs?: ModelBlobCodec }, ): DerivedModelDef { - return { name, initial, reducers, rehydrate: opts?.rehydrate }; + return { name, initial, reducers, blobs: opts?.blobs }; } diff --git a/packages/agent-core-v2/src/wire/wireService.ts b/packages/agent-core-v2/src/wire/wireService.ts index fbeb01198..15e64f237 100644 --- a/packages/agent-core-v2/src/wire/wireService.ts +++ b/packages/agent-core-v2/src/wire/wireService.ts @@ -4,11 +4,11 @@ * * The scope-agnostic state-machine engine: `dispatch` persists + applies + * notifies (OpGroup `{ silent: false }`), `replay` (async — rehydrates blob - * references first) applies only (`{ silent: true }`); `flush` drains the - * serialized persist queue. Reads go through `getModel` / `subscribe`; the live - * append-log record stream streams via `onEmission`, restore completion via - * `onRestored`, and Op-derived facts flow out through `IEventBus` (see `op.ts` - * `toEvent`). A single implementation serves every + * references via `ModelDef.blobs` first) applies only (`{ silent: true }`); + * `flush` drains the serialized persist queue. Reads go through `getModel` / + * `subscribe`; the live append-log record stream streams via `onEmission`, + * restore completion via `onRestored`, and Op-derived facts flow out through + * `IEventBus` (see `op.ts` `toEvent`). A single implementation serves every * scope — instances are isolated per scope through the distinct DI tokens in * `tokens`, each seeded with its own persistence key. `PersistedRecord` is the * on-the-wire append-log shape (`wire.jsonl`): intentionally flat diff --git a/packages/agent-core-v2/src/wire/wireServiceImpl.ts b/packages/agent-core-v2/src/wire/wireServiceImpl.ts index a190254ac..d0f46b7ca 100644 --- a/packages/agent-core-v2/src/wire/wireServiceImpl.ts +++ b/packages/agent-core-v2/src/wire/wireServiceImpl.ts @@ -1,18 +1,16 @@ /** * `wire` domain (L2) — `WireService`, the single scope-agnostic implementation - * of `IWireService`, plus its construction options (`WireServiceOptions`), the - * optional blob offload/rehydrate seam (`WireBlobSelector` / `WireBlobTarget`), + * of `IWireService`, plus its construction options (`WireServiceOptions`) * and the coded `CycleError`. * * One class serves every scope: per-scope isolation comes from the distinct DI * tokens in `tokens`, each seeded with its own `WireServiceOptions` - * (`logScope` / `logKey`, and optionally a `blobSelector`) as the leading - * (non-service) constructor argument through a `SyncDescriptor`, mirroring - * `WireRecordServiceOptions`. `dispatch` and `replay` both lower to one - * primitive, `execute(OpGroup)` — apply-all THEN onChange-all, so a subscriber - * never observes a partially-applied group — with `dispatch` adding persistence - * + emission + Op-derived `IEventBus` events (`silent: false`) and `replay` - * staying silent (apply only, skipping + * (`logScope` / `logKey`) as the leading (non-service) constructor argument + * through a `SyncDescriptor`, mirroring `WireRecordServiceOptions`. `dispatch` + * and `replay` both lower to one primitive, `execute(OpGroup)` — apply-all THEN + * onChange-all, so a subscriber never observes a partially-applied group — with + * `dispatch` adding persistence + emission + Op-derived `IEventBus` events + * (`silent: false`) and `replay` staying silent (apply only, skipping * unknown record types, then `onRestored`). A reentrancy guard (`dispatching` + * `queue` + `drain`, capped by `MAX_DRAIN = 100`) lets onChange handlers enqueue * further ops without reentering `execute`; a cascade past the cap throws @@ -28,19 +26,21 @@ * flat `{ type, ...payload }` record — scalar / array payloads nested so a * JSONL line stays an object, with `type` / `time` stripped back out on replay. * - * Blob handling has two asymmetric paths: + * Blob handling is driven by each `ModelDef`'s optional `blobs` codec + * (`ModelBlobCodec`), which declares two symmetric directions: * - * - **Offload (dispatch → persist)**: record-level `WireBlobSelector` rewrites - * oversized inline parts to `blobref:` references before the record reaches - * the append log. `apply` and the live emission still see the original inline - * payload. Records with no offloadable targets short-circuit synchronously. + * - **Dehydrate (dispatch → persist)**: `model.blobs.dehydrate(record, transform)` + * lets the model traverse its own record structure, pass each `ContentPart[]` + * through `transform` (which offloads oversized inline data to blob storage), + * and return the transformed record. `apply` and the live emission still see + * the original inline payload. Records whose model has no `blobs` codec + * short-circuit synchronously (no queue, no microtask). * - * - **Rehydrate (replay → model)**: `replay` applies all records first with - * blobref URLs entering the model state as-is (zero I/O). After all records - * are applied, `rehydrateModels` calls `ModelDef.rehydrate` on each model - * that declares it, replacing blobref URLs with inline data *only* in the - * surviving final state. This skips I/O for data later removed by compaction - * — a 20×+ speedup for long sessions with many images. + * - **Rehydrate (replay → model)**: after all records are applied, + * `rehydrateModels` calls `model.blobs.rehydrate(state, transform)` on each + * model that declares a `blobs` codec, replacing blobref URLs with inline data + * *only* in the surviving final state — skipping I/O for data later removed by + * compaction (a 20×+ speedup for long sessions with many images). * * Scope-agnostic. */ @@ -53,7 +53,7 @@ import { type DomainEvent, IEventBus } from '#/app/event'; import type { ContentPart } from '#/app/llmProtocol'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; -import type { DeepReadonly, DerivedModelDef, ModelDef, PartsRehydrator } from './model'; +import type { DeepReadonly, DerivedModelDef, ModelDef, PartsTransformer } from './model'; import type { Op } from './op'; import { OP_REGISTRY } from './op'; import type { @@ -75,17 +75,9 @@ export class CycleError extends Error { } } -export interface WireBlobTarget { - readonly parts: readonly ContentPart[]; - replace(record: PersistedRecord, parts: readonly ContentPart[]): PersistedRecord; -} - -export type WireBlobSelector = (record: PersistedRecord) => Iterable; - export interface WireServiceOptions { readonly logScope: string; readonly logKey: string; - readonly blobSelector?: WireBlobSelector; } interface ModelInstance { @@ -238,7 +230,7 @@ export class WireService extends Disposable implements IWireService { inst.state = Object.freeze(op.descriptor.apply(prev, op.payload)); if (!group.silent) { const record = this.toRecord(op); - this.appendToWireLog(record); + this.appendToWireLog(record, op.descriptor.model); this.emissionEmitter.fire({ type: 'record', record }); const event = op.descriptor.toEvent?.(op.payload, inst.state); if (event !== undefined && this.eventBus !== undefined) { @@ -289,20 +281,27 @@ export class WireService extends Disposable implements IWireService { return { type: op.type, payload }; } - private appendToWireLog(record: PersistedRecord): void { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + private appendToWireLog(record: PersistedRecord, model: ModelDef): void { if (this.log === undefined) return; - // When the blob hook is active, every append rides the serialized queue so a - // record with no offloadable targets cannot leapfrog a pending offload and - // reorder the log; otherwise append directly (no microtask, no queue). - if (this.blobService === undefined || this.options.blobSelector === undefined) { + // When the model declares a blob codec and the blob service is available, + // every append rides the serialized queue so a record with no offloadable + // targets cannot leapfrog a pending dehydrate and reorder the log; otherwise + // append directly (no microtask, no queue). + if (this.blobService === undefined || model.blobs?.dehydrate === undefined) { this.log.append(this.options.logScope, this.options.logKey, record, { onError: onUnexpectedError, }); return; } + const dehydrate = model.blobs.dehydrate; + const transform: PartsTransformer = (parts) => + this.blobService!.offloadParts( + parts as readonly ContentPart[], + ) as Promise; this.persistQueue = this.persistQueue .then(async () => { - const prepared = this.prepareRecord(record); + const prepared = dehydrate(record, transform); const offloaded = isPromise(prepared) ? await prepared : prepared; this.log?.append(this.options.logScope, this.options.logKey, offloaded, { onError: onUnexpectedError, @@ -311,44 +310,20 @@ export class WireService extends Disposable implements IWireService { .catch((error: unknown) => onUnexpectedError(error)); } - private prepareRecord(record: PersistedRecord): PersistedRecord | Promise { - const blobService = this.blobService; - const selector = this.options.blobSelector; - if (blobService === undefined || selector === undefined) return record; - const targets = [...selector(record)]; - if (targets.length === 0) return record; - return this.offloadTargets(record, targets, blobService); - } - - private async offloadTargets( - record: PersistedRecord, - targets: readonly WireBlobTarget[], - blobService: IAgentBlobService, - ): Promise { - let current = record; - for (const target of targets) { - const parts = await blobService.offloadParts(target.parts); - if (parts !== target.parts) { - current = target.replace(current, parts); - } - } - return current; - } - private async rehydrateModels(): Promise { if (this.blobService === undefined) return; - const rehydrateParts: PartsRehydrator = (parts) => - this.blobService!.rehydrateParts( + const transform: PartsTransformer = (parts) => + this.blobService!.loadParts( parts as readonly ContentPart[], ) as Promise; for (const [def, inst] of this.models) { - if (def.rehydrate === undefined) continue; - const result = def.rehydrate(inst.state, rehydrateParts); + if (def.blobs?.rehydrate === undefined) continue; + const result = def.blobs.rehydrate(inst.state, transform); inst.state = Object.freeze(isPromise(result) ? await result : result); } for (const [def, inst] of this.derivedModels) { - if (def.rehydrate === undefined) continue; - const result = def.rehydrate(inst.state, rehydrateParts); + if (def.blobs?.rehydrate === undefined) continue; + const result = def.blobs.rehydrate(inst.state, transform); inst.state = Object.freeze(isPromise(result) ? await result : result); } } diff --git a/packages/agent-core-v2/test/agentLifecycle/agentLifecycle.test.ts b/packages/agent-core-v2/test/agentLifecycle/agentLifecycle.test.ts index 238fcfcdf..b7388f863 100644 --- a/packages/agent-core-v2/test/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/agentLifecycle/agentLifecycle.test.ts @@ -102,7 +102,7 @@ function stubBlobPassThrough(ix: TestInstantiationService): void { ix.stub(IAgentBlobService, { _serviceBrand: undefined, offloadParts: async (parts) => parts, - rehydrateParts: async (parts) => parts, + loadParts: async (parts) => parts, isBlobRef: () => false, } satisfies IAgentBlobService); } diff --git a/packages/agent-core-v2/test/blob/blobService.test.ts b/packages/agent-core-v2/test/blob/blobService.test.ts index b2a0e452b..711f6c446 100644 --- a/packages/agent-core-v2/test/blob/blobService.test.ts +++ b/packages/agent-core-v2/test/blob/blobService.test.ts @@ -88,7 +88,7 @@ describe('AgentBlobServiceImpl', () => { const parts: ContentPart[] = [{ type: 'image_url', imageUrl: { url: uri } }]; const offloaded = await store.offloadParts(parts); - const rehydrated = await store.rehydrateParts(offloaded); + const rehydrated = await store.loadParts(offloaded); expect((rehydrated[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe(uri); }); @@ -106,7 +106,7 @@ describe('AgentBlobServiceImpl', () => { 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); + 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); }); @@ -117,7 +117,7 @@ describe('AgentBlobServiceImpl', () => { { type: 'image_url', imageUrl: { url: 'blobref:image/png;deadbeef' } }, ]; - const result = await store.rehydrateParts(parts); + const result = await store.loadParts(parts); expect((result[0]! as { imageUrl: { url: string } }).imageUrl.url).toBe('[media missing]'); }); @@ -165,7 +165,7 @@ describe('AgentBlobServiceImpl', () => { ).toBe(payload); expect(await backend.list('blobs')).toHaveLength(0); - const rehydrated = await store.rehydrateParts(offloaded); + 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 index 18d82ffa4..9c98f5d9e 100644 --- a/packages/agent-core-v2/test/blob/blobref.test.ts +++ b/packages/agent-core-v2/test/blob/blobref.test.ts @@ -190,7 +190,7 @@ describe('blobref', () => { const dataUri = `data:image/jpeg;base64,${payload}`; const offloaded = await store.offloadParts([imagePart(dataUri)]); - const rehydrated = await store.rehydrateParts(offloaded); + const rehydrated = await store.loadParts(offloaded); expect(firstImageUrl(rehydrated)).toBe(dataUri); expect(firstImageUrl(offloaded)).toMatch(/^blobref:image\/jpeg;/); @@ -198,7 +198,7 @@ describe('blobref', () => { it('replaces missing blobs with placeholder text', async () => { const { store } = await makeStore(); - const rehydrated = await store.rehydrateParts([imagePart('blobref:image/png;deadbeef')]); + const rehydrated = await store.loadParts([imagePart('blobref:image/png;deadbeef')]); expect(firstImageUrl(rehydrated)).toBe(MISSING_MEDIA_PLACEHOLDER); }); @@ -225,7 +225,7 @@ describe('blobref', () => { expect(files).toHaveLength(1); await rm(join(blobsDir, files[0]!)); - const rehydrated = await store.rehydrateParts(offloaded); + const rehydrated = await store.loadParts(offloaded); expect(firstImageUrl(rehydrated)).toBe(dataUri); }); @@ -239,14 +239,14 @@ describe('blobref', () => { const offloaded = await writer.offloadParts([imagePart(dataUri)]); const blobref = firstImageUrl(offloaded); - const firstRead = await reader.rehydrateParts([imagePart(blobref)]); + 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.rehydrateParts([imagePart(blobref)]); + const secondRead = await reader.loadParts([imagePart(blobref)]); expect(firstImageUrl(secondRead)).toBe(dataUri); }); @@ -262,7 +262,7 @@ describe('blobref', () => { const blobrefA = firstImageUrl(offloadedA); const blobrefB = firstImageUrl(offloadedB); - await store.rehydrateParts([imagePart(blobrefA)]); + await store.loadParts([imagePart(blobrefA)]); const offloadedC = await store.offloadParts([imagePart(`data:image/png;base64,${payloadC}`)]); const blobrefC = firstImageUrl(offloadedC); @@ -271,13 +271,13 @@ describe('blobref', () => { await rm(join(blobsDir, file)); } - expect(firstImageUrl(await store.rehydrateParts([imagePart(blobrefA)]))).toBe( + expect(firstImageUrl(await store.loadParts([imagePart(blobrefA)]))).toBe( `data:image/png;base64,${payloadA}`, ); - expect(firstImageUrl(await store.rehydrateParts([imagePart(blobrefB)]))).toBe( + expect(firstImageUrl(await store.loadParts([imagePart(blobrefB)]))).toBe( MISSING_MEDIA_PLACEHOLDER, ); - expect(firstImageUrl(await store.rehydrateParts([imagePart(blobrefC)]))).toBe( + expect(firstImageUrl(await store.loadParts([imagePart(blobrefC)]))).toBe( `data:image/png;base64,${payloadC}`, ); }); @@ -298,10 +298,10 @@ describe('blobref', () => { await rm(join(blobsDir, file)); } - expect(firstImageUrl(await store.rehydrateParts([imagePart(smallBlobref)]))).toBe( + expect(firstImageUrl(await store.loadParts([imagePart(smallBlobref)]))).toBe( `data:image/png;base64,${small}`, ); - expect(firstImageUrl(await store.rehydrateParts([imagePart(largeBlobref)]))).toBe( + expect(firstImageUrl(await store.loadParts([imagePart(largeBlobref)]))).toBe( MISSING_MEDIA_PLACEHOLDER, ); }); diff --git a/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts b/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts index 1fe9104e5..62ce85061 100644 --- a/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts +++ b/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts @@ -1,11 +1,11 @@ /** * `AgentContextMemoryService` wire contract, exercised without the full agent * harness (mirror of `test/goal/goal-wire.test.ts`): a `TestInstantiationService` - * + `InMemoryStorageService` + `AppendLogStore` + `WireService` seeded with the - * `contextBlobSelector` and a stub `IAgentBlobService`. Covers the splice Ops' - * NEW-reference + flat-record shape, the live-only `onSpliced` hook (silent on - * replay), and — load-bearing — the blob offload-on-dispatch ↔ - * rehydrate-on-replay round-trip. + * + `InMemoryStorageService` + `AppendLogStore` + `WireService` + stub + * `IAgentBlobService`. Covers the splice Ops' NEW-reference + flat-record shape, + * the live-only `onSpliced` hook (silent on replay), and — load-bearing — the + * blob dehydrate-on-dispatch ↔ rehydrate-on-replay round-trip via + * `ContextModel.blobs`. */ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; @@ -16,7 +16,6 @@ import { TestInstantiationService } from '#/_base/di/test'; import { IAgentBlobService } from '#/agent/blob'; import { AgentContextMemoryService, - contextBlobSelector, ContextModel, contextAppendMessage, contextApplyCompaction, @@ -54,7 +53,7 @@ class StubBlobService implements IAgentBlobService { declare readonly _serviceBrand: undefined; readonly store = new Map(); offloadCalls = 0; - rehydrateCalls = 0; + loadCalls = 0; private seq = 0; isBlobRef(url: string): boolean { @@ -71,7 +70,7 @@ class StubBlobService implements IAgentBlobService { return changed ? out : parts; } - async rehydrateParts(parts: readonly ContentPart[]): Promise { + async loadParts(parts: readonly ContentPart[]): Promise { let changed = false; const out = parts.map((part) => { const next = this.rehydratePart(part); @@ -109,7 +108,7 @@ class StubBlobService implements IAgentBlobService { const sha = rest.slice(semi + 1); const payload = this.store.get(sha); if (payload === undefined) continue; - this.rehydrateCalls++; + this.loadCalls++; return { ...obj, [key]: { ...media, url: `data:${mime};base64,${payload}` } } as unknown as ContentPart; } return part; @@ -158,7 +157,7 @@ function buildHost(key: string): Host { ix.set( IAgentWireService, new SyncDescriptor(WireService, [ - { logScope: SCOPE, logKey: key, blobSelector: contextBlobSelector }, + { logScope: SCOPE, logKey: key }, ]), ); ix.stub(IAgentBlobService, blob); @@ -301,7 +300,7 @@ describe('AgentContextMemoryService (wire-backed)', () => { const replay = buildHost(REPLAY_KEY); await replay.wire.replay(...records); - expect(blob.rehydrateCalls).toBeGreaterThanOrEqual(1); + expect(blob.loadCalls).toBeGreaterThanOrEqual(1); const rebuilt = replay.wire.getModel(ContextModel) as readonly ContextMessage[]; expect(rebuilt).toEqual(live);