From 8ee741007b7c576db92cfe97ef77830d2e30f0e1 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Thu, 2 Jul 2026 19:19:50 +0800 Subject: [PATCH] refactor(agent-core-v2): fold replayBuilder into record The record stream had three services: wireRecord (durable log), record (live/facet facade), and replayBuilder (replay buffer). replayBuilder was just one more projection of the same record stream, so fold it into record and model the split as Event Store + Projection: - record now owns the replay read model: push/patchLast/removeLastMessages/ buildReplay/captureLiveRecords, plus the range/segment logic for partial resume (seeded via RecordServiceOptions). - The two onRestoredRecord handlers collapse into one; record no longer injects IAgentReplayBuilderService. - Drop the unused restore/flush/close forwarders from IAgentRecordService; orchestration keeps using wireRecord directly. - replayBuilder keeps only its shared cross-domain types (AgentReplayRecord, ResumeSessionResult) as a types-only module; the service is removed. - Update consumers (contextMemory, fullCompaction, goal, messageLegacy, agentLifecycle), the test harness, and stubs to IAgentRecordService. - Add a record>contextMemory domain-layer exception: replay message records carry ContextMessage. --- .../agent-core-v2/docs/di-scope-domains.puml | 16 +- .../scripts/check-domain-layers.mjs | 4 + .../contextMemory/contextMemoryService.ts | 6 +- .../fullCompaction/fullCompactionService.ts | 10 +- .../src/agent/goal/goalService.ts | 6 +- .../agent-core-v2/src/agent/record/record.ts | 51 ++++-- .../src/agent/record/recordService.ts | 153 ++++++++++++++---- .../src/agent/replayBuilder/index.ts | 10 +- .../src/agent/replayBuilder/replayBuilder.ts | 33 ---- .../replayBuilder/replayBuilderService.ts | 149 ----------------- .../src/app/messageLegacy/messageLegacy.ts | 2 +- .../app/messageLegacy/messageLegacyService.ts | 6 +- .../agentLifecycle/agentLifecycleService.ts | 11 +- .../agent-core-v2/test/contextMemory/stubs.ts | 22 ++- packages/agent-core-v2/test/goal/goal.test.ts | 10 +- packages/agent-core-v2/test/harness/agent.ts | 14 +- .../test/message/message.test.ts | 6 +- .../test/record/recordService.test.ts | 115 +++++++------ .../test/skill/skill-tool-manager.test.ts | 8 +- .../agent-core-v2/test/skill/skill.test.ts | 13 -- .../agent-core-v2/test/usage/usage.test.ts | 7 - .../test/wireRecord/index.test.ts | 14 +- .../test/wireRecord/persistence.test.ts | 6 +- .../test/wireRecord/resume.test.ts | 12 +- 24 files changed, 293 insertions(+), 391 deletions(-) delete mode 100644 packages/agent-core-v2/src/agent/replayBuilder/replayBuilder.ts delete mode 100644 packages/agent-core-v2/src/agent/replayBuilder/replayBuilderService.ts diff --git a/packages/agent-core-v2/docs/di-scope-domains.puml b/packages/agent-core-v2/docs/di-scope-domains.puml index 610cfba80..93fbcf2eb 100644 --- a/packages/agent-core-v2/docs/di-scope-domains.puml +++ b/packages/agent-core-v2/docs/di-scope-domains.puml @@ -70,7 +70,6 @@ package "Agent scope (per agent)" #FDF5E6 { rectangle "contextInjector\nAgent\n IAgentContextInjectorService" as contextInjector #FDEBD0 rectangle "contextSize\nAgent\n IAgentContextSizeService" as contextSize #FDEBD0 rectangle "systemReminder\nAgent\n IAgentSystemReminderService" as systemReminder #FDEBD0 - rectangle "replayBuilder\nAgent\n IAgentReplayBuilderService" as replayBuilder #FDEBD0 rectangle "profile\nAgent\n IAgentProfileService" as profile #FDEBD0 rectangle "prompt\nAgent\n IAgentPromptService" as prompt #FDEBD0 rectangle "turn\nAgent\n IAgentTurnService" as turn #FDEBD0 @@ -167,7 +166,7 @@ blobStore --> environment #34495E filestore --> storage #34495E record --> wireRecord #34495E contextMemory --> wireRecord #34495E -contextMemory --> replayBuilder #34495E +contextMemory --> record #34495E contextInjector --> contextMemory #34495E contextInjector --> turn #34495E contextInjector --> loop #34495E @@ -176,10 +175,9 @@ contextSize --> contextMemory #34495E contextSize --> record #34495E contextSize --> wireRecord #34495E systemReminder --> contextMemory #34495E -replayBuilder --> wireRecord #34495E profile --> wireRecord #34495E profile --> record #34495E -profile --> replayBuilder #34495E +profile --> record #34495E profile --> telemetry #34495E profile --> config #34495E profile --> modelProvider #34495E @@ -224,24 +222,24 @@ permissionGate --> telemetry #34495E permissionGate --> toolExecutor #34495E permissionMode --> wireRecord #34495E permissionMode --> record #34495E -permissionMode --> replayBuilder #34495E +permissionMode --> record #34495E permissionMode --> contextInjector #34495E permissionRules --> wireRecord #34495E -permissionRules --> replayBuilder #34495E +permissionRules --> record #34495E permissionRules --> config #34495E plan --> contextMemory #34495E plan --> wireRecord #34495E plan --> record #34495E plan --> agentFs #34495E plan --> profile #34495E -plan --> replayBuilder #34495E +plan --> record #34495E plan --> toolRegistry #34495E plan --> contextInjector #34495E plan --> telemetry #34495E goal --> wireRecord #34495E goal --> record #34495E goal --> systemReminder #34495E -goal --> replayBuilder #34495E +goal --> record #34495E goal --> telemetry #34495E goal --> contextInjector #34495E goal --> contextMemory #34495E @@ -294,7 +292,7 @@ fullCompaction --> toolStore #34495E fullCompaction --> telemetry #34495E fullCompaction --> wireRecord #34495E fullCompaction --> record #34495E -fullCompaction --> replayBuilder #34495E +fullCompaction --> record #34495E fullCompaction --> externalHooks #34495E fullCompaction --> turn #34495E fullCompaction --> loop #34495E diff --git a/packages/agent-core-v2/scripts/check-domain-layers.mjs b/packages/agent-core-v2/scripts/check-domain-layers.mjs index 8712b13a9..42dd8b3f1 100644 --- a/packages/agent-core-v2/scripts/check-domain-layers.mjs +++ b/packages/agent-core-v2/scripts/check-domain-layers.mjs @@ -257,6 +257,10 @@ const ALLOWED_EXCEPTIONS = new Set([ 'permissionPolicy>profile', 'permissionRules>replayBuilder', 'record>replayBuilder', + // `record` owns the replay read model, whose `message` records carry + // `ContextMessage` (L4). `removeLastMessages` takes a set of them, so the + // projection side references the context message type by structure only. + 'record>contextMemory', 'plugin>externalHooks', 'plugin>mcp', 'profile>session', diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts index 043b0452d..6e83881c7 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts @@ -2,7 +2,6 @@ import { Disposable, } from "#/_base/di"; import { OrderedHookSlot } from '#/hooks'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; import { IAgentRecordService, type AgentRecord } from '#/agent/record'; import { IAgentContextMemoryService } from './contextMemory'; import { ensureMessageId } from './messageId'; @@ -36,7 +35,6 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte constructor( @IAgentRecordService private readonly record: IAgentRecordService, - @IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService, ) { super(); this._register( @@ -87,9 +85,9 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte : []; const messages = record.messages.map(ensureMessageId); this.history.splice(record.start, record.deleteCount, ...messages); - this.replayBuilder.removeLastMessages(new Set(removedMessages)); + this.record.removeLastMessages(new Set(removedMessages)); for (const message of messages) { - this.replayBuilder.push({ type: 'message', message }); + this.record.push({ type: 'message', message }); } void this.hooks.onSpliced.run({ start: record.start, diff --git a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts index c1f88e1ff..5513d3c0f 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts @@ -20,7 +20,6 @@ import { IAgentLoopService, type TurnContextOverflowContext } from '#/agent/loop import { isAbortError } from '#/agent/loop/errors'; import { IAgentProfileService } from '#/agent/profile'; import { IAgentRecordService } from '#/agent/record'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; import { TODO_STORE_KEY, renderTodoList, @@ -106,7 +105,6 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull @IAgentToolStoreService private readonly toolStore: IAgentToolStoreService, @ITelemetryService private readonly telemetry: ITelemetryService, @IAgentRecordService private readonly record: IAgentRecordService, - @IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService, @IAgentExternalHooksService private readonly externalHooks: IAgentExternalHooksService, @IAgentTurnService turnService: IAgentTurnService, @IAgentLoopService loopService: IAgentLoopService, @@ -141,7 +139,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull this._register( record.define('full_compaction.begin', { resume: (r) => { - this.replayBuilder.push({ + this.record.push({ type: 'compaction', instruction: r.instruction, }); @@ -151,7 +149,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull this._register( record.define('full_compaction.cancel', { resume: () => { - this.replayBuilder.patchLast('compaction', { result: 'cancelled' }); + this.record.patchLast('compaction', { result: 'cancelled' }); }, }), ); @@ -161,8 +159,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull const message = compactionSummaryMessage(this.context.get()); if (message === undefined) return; const summary = contextMessageText(message); - this.replayBuilder.removeLastMessages(new Set([message])); - this.replayBuilder.patchLast('compaction', { + this.record.removeLastMessages(new Set([message])); + this.record.patchLast('compaction', { result: { summary, compactedCount: r.compactedCount, diff --git a/packages/agent-core-v2/src/agent/goal/goalService.ts b/packages/agent-core-v2/src/agent/goal/goalService.ts index 8dafbf8db..77ccf1f93 100644 --- a/packages/agent-core-v2/src/agent/goal/goalService.ts +++ b/packages/agent-core-v2/src/agent/goal/goalService.ts @@ -34,7 +34,6 @@ import { type TurnStepUsageContext, } from '#/agent/loop'; import { IAgentRecordService, type AgentRecord } from '#/agent/record'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; import { IAgentSystemReminderService } from '#/agent/systemReminder'; import { IAgentTurnService, @@ -164,7 +163,6 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { private readonly options: GoalServiceOptions = {}, @IAgentRecordService private readonly record: IAgentRecordService, @IAgentSystemReminderService private readonly reminders: IAgentSystemReminderService, - @IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService, @ITelemetryService private readonly telemetry: ITelemetryService, @IAgentContextInjectorService dynamicInjector: IAgentContextInjectorService, @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, @@ -582,7 +580,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { budgetLimits: {}, }; this.state = state; - this.replayBuilder.push({ + this.record.push({ type: 'goal_updated', snapshot: this.toSnapshot(state), change: { kind: 'created' }, @@ -608,7 +606,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { if (record.budgetLimits !== undefined) state.budgetLimits = record.budgetLimits; if (status === undefined) return; - this.replayBuilder.push({ + this.record.push({ type: 'goal_updated', snapshot: this.toSnapshot(state), change: diff --git a/packages/agent-core-v2/src/agent/record/record.ts b/packages/agent-core-v2/src/agent/record/record.ts index f39a839f3..d95e0a387 100644 --- a/packages/agent-core-v2/src/agent/record/record.ts +++ b/packages/agent-core-v2/src/agent/record/record.ts @@ -4,22 +4,21 @@ * Single entry point for recording facts that happen inside an agent. One * `append(record)` call fans out to every facet declared for the record type: * durable persistence (for resume), live broadcast (`AgentEvent` to the edge), - * and replay capture. `signal(event)` emits a live-only event that is never - * recorded (deltas / progress). Bound at Agent scope. + * and the replay read model. `signal(event)` emits a live-only event that is + * never recorded (deltas / progress). The replay read model (`buildReplay`, + * `push`/`patchLast`/`removeLastMessages`) is owned here too — it is just one + * more projection of the same record stream. Bound at Agent scope. */ import type { AgentEvent } from '@moonshot-ai/protocol'; import type { IDisposable } from '#/_base/di'; import { createDecorator } from '#/_base/di'; +import type { ContextMessage } from '#/agent/contextMemory'; import type { IAgentWireRecordService, - WireRecord, WireRecordBlobSelector, WireRecordMap, - WireRecordRestoreOptions, - WireRecordRestoreResult, - PersistedWireRecord, WireRecordRestoringContext, } from '#/agent/wireRecord'; import type { AgentReplayRecord, AgentReplayRecordPayload } from '#/agent/replayBuilder/types'; @@ -67,6 +66,20 @@ export interface RecordFacets { readonly blobs?: WireRecordBlobSelector>; } +export interface ReplayRangeOptions { + readonly start?: number; + readonly count?: number; +} + +/** + * Static construction options for `AgentRecordService`, supplied through a + * `SyncDescriptor` when the service is seeded into a scope. `range` limits the + * replay read model to a slice of the restored stream (used by partial-resume). + */ +export interface RecordServiceOptions { + readonly range?: ReplayRangeOptions; +} + export interface IAgentRecordService { readonly _serviceBrand: undefined; @@ -86,16 +99,24 @@ export interface IAgentRecordService { */ define(type: K, facets: RecordFacets): IDisposable; - restore( - records?: readonly PersistedWireRecord[], - options?: WireRecordRestoreOptions, - ): Promise; - flush(): Promise; - close(): Promise; - - /** Replay result built from restored (and optionally live) records. */ + /** + * Append a record to the replay read model directly. Used when the projected + * data is computed inside a domain handler rather than derived from a single + * record via `toReplay` (e.g. `contextMemory` projecting spliced messages). + * Gated by phase: captured while restoring/post-restoring, or always when + * `captureLiveRecords` is set. + */ + push(record: AgentReplayRecordPayload): void; + /** Patch the most recent replay record of `type` (restore-time only). */ + patchLast( + type: T, + patch: Partial>, + ): void; + /** Drop replay `message` records whose message is in `removedMessages`. */ + removeLastMessages(removedMessages: ReadonlySet): void; + /** Replay read model built from restored (and optionally live) records. */ buildReplay(): readonly AgentReplayRecord[]; - /** When true, live `append` calls also feed the replay buffer. */ + /** When true, live `append` calls also feed the replay read model. */ captureLiveRecords: boolean; readonly restoring: WireRecordRestoringContext | null; diff --git a/packages/agent-core-v2/src/agent/record/recordService.ts b/packages/agent-core-v2/src/agent/record/recordService.ts index f059ed6d5..f007d43e6 100644 --- a/packages/agent-core-v2/src/agent/record/recordService.ts +++ b/packages/agent-core-v2/src/agent/record/recordService.ts @@ -3,12 +3,17 @@ * * Owns the unified `append` / `signal` / `define` API: one `append(record)` * fans out to durable persistence (delegated to `wireRecord`), live broadcast - * (an owned `Emitter`), and replay capture (delegated to - * `replayBuilder`). `signal(event)` emits a live-only event that is never - * recorded. Live emission is suppressed while restoring, so edge consumers - * never receive historical events. The former `eventSink` service is folded - * into this class; `wireRecord` / `replayBuilder` remain registered backends - * that this service coordinates. + * (an owned `Emitter`), and the replay read model (an owned + * buffer). `signal(event)` emits a live-only event that is never recorded. + * Live emission is suppressed while restoring, so edge consumers never receive + * historical events. + * + * The replay read model (`push` / `patchLast` / `removeLastMessages` / + * `buildReplay`) is owned here too — it is one more projection of the same + * record stream, fed by `toReplay` facets (declarative) and by direct `push` + * calls from domain handlers (imperative). The former `eventSink` and + * `replayBuilder` services are folded into this class; `wireRecord` remains the + * registered persistence backend that this service coordinates. */ import { Disposable, toDisposable } from '#/_base/di'; @@ -18,43 +23,64 @@ import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import type { AgentEvent } from '@moonshot-ai/protocol'; +import type { ContextMessage } from '#/agent/contextMemory'; import { IAgentWireRecordService, type WireRecord, type WireRecordBlobSelector, type WireRecordMap, - type WireRecordRestoreOptions, - type WireRecordRestoreResult, - type PersistedWireRecord, type WireRecordRestoringContext, } from '#/agent/wireRecord'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; -import type { AgentReplayRecord } from '#/agent/replayBuilder/types'; +import type { AgentReplayRecord, AgentReplayRecordPayload } from '#/agent/replayBuilder/types'; import { IAgentRecordService, type AgentRecord, type AgentRecordMap, type RecordFacets, + type RecordServiceOptions, } from './record'; +// An undo boundary is a `context.splice` that removes messages from the start of +// the history. It is the canonical (post v1.5 migration) equivalent of the legacy +// `context.clear` and `context.apply_compaction` records, both of which the v1.5 +// migration rewrites into a `context.splice` with `start === 0` and +// `deleteCount > 0` (see wireRecord/migration/v1.5.ts). A splice that only +// appends (`deleteCount === 0`) or removes messages from the middle/end of the +// history (`start > 0`, e.g. a migrated `context.undo`) is not a boundary. +function isUndoBoundaryRecord(record: WireRecord): boolean { + return record.type === 'context.splice' && record.start === 0 && record.deleteCount > 0; +} + export class AgentRecordService extends Disposable implements IAgentRecordService { declare readonly _serviceBrand: undefined; private readonly facets = new Map>(); private readonly liveEmitter = this._register(new Emitter()); + // Replay read model state. + captureLiveRecords = false; + private readonly replayRecords: AgentReplayRecord[] = []; + private _postRestoring = false; + private frozen = false; + private segmentStart = 0; + constructor( + private readonly options: RecordServiceOptions = {}, @IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService, - @IAgentReplayBuilderService private readonly replayBuilder: IAgentReplayBuilderService, ) { super(); // Restore-time replay capture: every restored record runs its `toReplay` - // facet. `replayBuilder.push` gates by phase, so this only records while - // restoring / post-restoring. + // facet (declarative projection), then the range/segment logic applies. + // Domain resumers (which run before this hook) perform the imperative + // `push` calls, so by the time we run here their contributions are already + // in the buffer. this._register( wireRecord.hooks.onRestoredRecord.register('record-replay', async (ctx, next) => { await next(); this.runReplayFacet(ctx.record as unknown as AgentRecord); + if (this.finishRestoringRecord(ctx.record)) { + ctx.stop = true; + } }), ); } @@ -109,31 +135,55 @@ export class AgentRecordService extends Disposable implements IAgentRecordServic }); } - restore( - records?: readonly PersistedWireRecord[], - options?: WireRecordRestoreOptions, - ): Promise { - return this.wireRecord.restore(records, options); + push(record: AgentReplayRecordPayload): void { + if ( + !this.captureLiveRecords && + this.wireRecord.restoring === null && + !this.postRestoring + ) { + return; + } + if (this.frozen) return; + + this.replayRecords.push({ + ...record, + time: this.wireRecord.restoring?.time ?? Date.now(), + }); } - flush(): Promise { - return this.wireRecord.flush(); + patchLast( + type: T, + patch: Partial>, + ): void { + if (this.frozen) return; + if (this.wireRecord.restoring === null) return; + + const last = this.replayRecords.at(-1); + if (last?.type === type) { + Object.assign(last, patch); + } } - close(): Promise { - return this.wireRecord.close(); + removeLastMessages(removedMessages: ReadonlySet): void { + if (this.frozen) return; + if (removedMessages.size === 0) return; + this.removeMessagesFrom(this.replayRecords, removedMessages); } buildReplay(): readonly AgentReplayRecord[] { - return this.replayBuilder.buildResult(); - } - - get captureLiveRecords(): boolean { - return this.replayBuilder.captureLiveRecords; - } - - set captureLiveRecords(value: boolean) { - this.replayBuilder.captureLiveRecords = value; + const range = this.options.range; + if (range !== undefined) { + if (range.start === undefined && range.count !== undefined) { + const offset = Math.max(0, this.replayRecords.length - range.count); + return this.replayRecords.slice(offset); + } + const start = range.start ?? 0; + const offset = Math.max(0, start - this.segmentStart); + const count = range.count; + const end = count === undefined ? undefined : offset + count; + return this.replayRecords.slice(offset, end); + } + return this.replayRecords; } get restoring(): WireRecordRestoringContext | null { @@ -141,7 +191,11 @@ export class AgentRecordService extends Disposable implements IAgentRecordServic } get postRestoring(): boolean { - return this.replayBuilder.postRestoring; + return this._postRestoring || this.wireRecord.postRestoring; + } + + set postRestoring(value: boolean) { + this._postRestoring = value; } get hooks(): IAgentWireRecordService['hooks'] { @@ -162,7 +216,38 @@ export class AgentRecordService extends Disposable implements IAgentRecordServic if (out === undefined) return; const list = Array.isArray(out) ? out : [out]; for (const replayRecord of list) { - this.replayBuilder.push(replayRecord); + this.push(replayRecord); + } + } + + private finishRestoringRecord(record: WireRecord): boolean { + const range = this.options.range; + if (range === undefined) return false; + if (this.frozen) return true; + if (!isUndoBoundaryRecord(record)) return false; + if (range.start === undefined) return false; + + const start = range.start; + const nextSegmentStart = this.segmentStart + this.replayRecords.length; + if (nextSegmentStart > start) { + this.frozen = true; + return true; + } + + this.segmentStart = nextSegmentStart; + this.replayRecords.splice(0); + return false; + } + + private removeMessagesFrom( + records: AgentReplayRecord[], + removedMessages: ReadonlySet, + ): void { + for (let i = records.length - 1; i >= 0; i--) { + const record = records[i]!; + if (record.type === 'message' && removedMessages.has(record.message)) { + records.splice(i, 1); + } } } } diff --git a/packages/agent-core-v2/src/agent/replayBuilder/index.ts b/packages/agent-core-v2/src/agent/replayBuilder/index.ts index 15560ee7d..f85c1eb85 100644 --- a/packages/agent-core-v2/src/agent/replayBuilder/index.ts +++ b/packages/agent-core-v2/src/agent/replayBuilder/index.ts @@ -1,6 +1,10 @@ /** - * `replayBuilder` domain barrel - re-exports the replayBuilder service contract and implementation. + * `replayBuilder` barrel — shared replay read-model types. + * + * The replay buffer service was folded into `IAgentRecordService` (the + * projection side of the record stream). Only the cross-domain types remain + * here: the replay record shapes produced by `record` and the resume-result + * shapes consumed by the edge (`rpc` / `core-api`). */ -export * from './replayBuilder'; -export * from './replayBuilderService'; +export * from './types'; diff --git a/packages/agent-core-v2/src/agent/replayBuilder/replayBuilder.ts b/packages/agent-core-v2/src/agent/replayBuilder/replayBuilder.ts deleted file mode 100644 index ccb102483..000000000 --- a/packages/agent-core-v2/src/agent/replayBuilder/replayBuilder.ts +++ /dev/null @@ -1,33 +0,0 @@ -import { createDecorator } from "#/_base/di"; -import type { ContextMessage } from '#/agent/contextMemory'; -import type { WireRecord } from "#/agent/wireRecord"; -import type { AgentReplayRecord, AgentReplayRecordPayload } from './types'; - -export interface ReplayRangeOptions { - readonly start?: number; - readonly count?: number; -} - -export interface ReplayBuilderServiceOptions { - readonly range?: ReplayRangeOptions; -} - -export interface IAgentReplayBuilderService { - readonly _serviceBrand: undefined; - - postRestoring: boolean; - captureLiveRecords: boolean; - - push(record: AgentReplayRecordPayload): void; - patchLast( - type: T, - patch: Partial>, - ): void; - removeLastMessages(removedMessages: ReadonlySet): void; - finishRestoringRecord(record: WireRecord): boolean; - buildResult(): readonly AgentReplayRecord[]; -} - -export const IAgentReplayBuilderService = createDecorator( - 'agentReplayBuilderService', -); diff --git a/packages/agent-core-v2/src/agent/replayBuilder/replayBuilderService.ts b/packages/agent-core-v2/src/agent/replayBuilder/replayBuilderService.ts deleted file mode 100644 index 71430977b..000000000 --- a/packages/agent-core-v2/src/agent/replayBuilder/replayBuilderService.ts +++ /dev/null @@ -1,149 +0,0 @@ -import { - Disposable, -} from "#/_base/di"; -import { InstantiationType } from '#/_base/di/extensions'; -import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import type { AgentReplayRecord, AgentReplayRecordPayload } from './types'; - -import type { ContextMessage } from "#/agent/contextMemory"; -import type { WireRecord } from "#/agent/wireRecord"; -import { IAgentWireRecordService } from '#/agent/wireRecord'; -import { - IAgentReplayBuilderService, - type ReplayBuilderServiceOptions, -} from './replayBuilder'; - -// An undo boundary is a `context.splice` that removes messages from the start of -// the history. It is the canonical (post v1.5 migration) equivalent of the legacy -// `context.clear` and `context.apply_compaction` records, both of which the v1.5 -// migration rewrites into a `context.splice` with `start === 0` and -// `deleteCount > 0` (see wireRecord/migration/v1.5.ts). A splice that only -// appends (`deleteCount === 0`) or removes messages from the middle/end of the -// history (`start > 0`, e.g. a migrated `context.undo`) is not a boundary. -function isUndoBoundaryRecord(record: WireRecord): boolean { - return record.type === 'context.splice' && record.start === 0 && record.deleteCount > 0; -} - -export class AgentReplayBuilderService extends Disposable implements IAgentReplayBuilderService { - declare readonly _serviceBrand: undefined; - - captureLiveRecords = false; - - private readonly records: AgentReplayRecord[] = []; - private _postRestoring = false; - private frozen = false; - private segmentStart = 0; - - constructor( - private readonly options: ReplayBuilderServiceOptions = {}, - @IAgentWireRecordService private readonly wireRecord: IAgentWireRecordService, - ) { - super(); - this._register( - wireRecord.hooks.onRestoredRecord.register('replay-builder', async (context, next) => { - await next(); - if (this.finishRestoringRecord(context.record)) { - context.stop = true; - } - }), - ); - } - - get postRestoring(): boolean { - return this._postRestoring || this.wireRecord.postRestoring; - } - - set postRestoring(value: boolean) { - this._postRestoring = value; - } - - push(record: AgentReplayRecordPayload): void { - if ( - !this.captureLiveRecords && - this.wireRecord.restoring === null && - !this.postRestoring - ) { - return; - } - if (this.frozen) return; - - this.records.push({ - ...record, - time: this.wireRecord.restoring?.time ?? Date.now(), - }); - } - - patchLast( - type: T, - patch: Partial>, - ): void { - if (this.frozen) return; - if (this.wireRecord.restoring === null) return; - - const last = this.records.at(-1); - if (last?.type === type) { - Object.assign(last, patch); - } - } - - removeLastMessages(removedMessages: ReadonlySet): void { - if (this.frozen) return; - if (removedMessages.size === 0) return; - this.removeMessagesFrom(this.records, removedMessages); - } - - finishRestoringRecord(record: WireRecord): boolean { - const range = this.options.range; - if (range === undefined) return false; - if (this.frozen) return true; - if (!isUndoBoundaryRecord(record)) return false; - if (range.start === undefined) return false; - - const start = range.start; - const nextSegmentStart = this.segmentStart + this.records.length; - if (nextSegmentStart > start) { - this.frozen = true; - return true; - } - - this.segmentStart = nextSegmentStart; - this.records.splice(0); - return false; - } - - buildResult(): readonly AgentReplayRecord[] { - const range = this.options.range; - if (range !== undefined) { - if (range.start === undefined && range.count !== undefined) { - const offset = Math.max(0, this.records.length - range.count); - return this.records.slice(offset); - } - const start = range.start ?? 0; - const offset = Math.max(0, start - this.segmentStart); - const count = range.count; - const end = count === undefined ? undefined : offset + count; - return this.records.slice(offset, end); - } - return this.records; - } - - private removeMessagesFrom( - records: AgentReplayRecord[], - removedMessages: ReadonlySet, - ): void { - for (let i = records.length - 1; i >= 0; i--) { - const record = records[i]!; - if (record.type === 'message' && removedMessages.has(record.message)) { - records.splice(i, 1); - } - } - } -} - -registerScopedService( - LifecycleScope.Agent, - IAgentReplayBuilderService, - AgentReplayBuilderService, - InstantiationType.Delayed, - 'replayBuilder', -); diff --git a/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts b/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts index b2121c141..a07aac7cb 100644 --- a/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts +++ b/packages/agent-core-v2/src/app/messageLegacy/messageLegacy.ts @@ -8,7 +8,7 @@ * holds the model's CURRENT, folded context and is left untouched. For a live * session this adapter reads that folded history (its transcript is in memory * by definition); for a cold session it loads the session, restores the main - * agent's wire log, and reads the FULL transcript from `IAgentReplayBuilderService` + * agent's wire log, and reads the FULL transcript from `IAgentRecordService` * (pre-compaction messages preserved, matching v1's `wire.jsonl` rebuild). The * `ContextMessage → Message` projection is shared with the `snapshot` and * `:undo` edges via `contextMemory/messageProjection`. Bound at App scope — a diff --git a/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts b/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts index b74828e35..e7888bea6 100644 --- a/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts +++ b/packages/agent-core-v2/src/app/messageLegacy/messageLegacyService.ts @@ -6,7 +6,7 @@ * shape. Live sessions are read from the main agent's `IAgentContextMemoryService` (the * folded history already in memory); cold sessions are loaded, their main agent * is restored from the persisted wire log, and the FULL transcript is read from - * `IAgentReplayBuilderService` — v2's own replay reducer, so no reduction logic is + * `IAgentRecordService` — v2's own replay read model, so no reduction logic is * duplicated here. Pagination, id derivation, and the role filter mirror v1's * `MessageService` (`packages/agent-core/src/services/message/messageService.ts`). */ @@ -22,7 +22,7 @@ import { type ContextMessage, } from '#/agent/contextMemory'; import { ErrorCodes, KimiError } from '#/errors'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; +import { IAgentRecordService } from '#/agent/record'; import { ISessionIndex } from '#/app/sessionIndex'; import { ISessionLifecycleService } from '#/app/sessionLifecycle'; @@ -109,7 +109,7 @@ export class MessageLegacyService implements IMessageLegacyService { // agent was restored and the replay holds the full pre-compaction history. // Otherwise the agent is fresh (never restored) and the folded context // memory IS the transcript. - const replay = agent.accessor.get(IAgentReplayBuilderService).buildResult(); + const replay = agent.accessor.get(IAgentRecordService).buildReplay(); const restored: ContextMessage[] = []; for (const record of replay) { if (record.type === 'message') restored.push(record.message); diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index f685eb2f7..306511864 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -34,10 +34,6 @@ import { IAgentContextMemoryService } from '#/agent/contextMemory'; import { IAgentBuiltinToolsRegistrar } from '#/agent/toolRegistry'; import { IAgentWireRecordService, AgentWireRecordService } from '#/agent/wireRecord'; import { IAgentBlobStoreService, AgentBlobStoreService } from '#/agent/blobStore'; -import { - IAgentReplayBuilderService, - AgentReplayBuilderService, -} from '#/agent/replayBuilder'; import { IAgentExternalHooksService, AgentExternalHooksService, @@ -106,12 +102,11 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle [IAgentWireRecordService, new SyncDescriptor(AgentWireRecordService, [{ homedir: agentHomedir }])], [IAgentBlobStoreService, new SyncDescriptor(AgentBlobStoreService, [{}])], [IAgentMcpService, new SyncDescriptor(AgentMcpService, [{ manager: this.getMcpManager() }])], - // These two carry a leading static `options` param; the scoped + // External hooks carries a leading static `options` param; the scoped // registry supplies none, so seed an empty one to satisfy the DI // contract (static args must fill the slots before the first `@IX`). - // Kept delayed so they only instantiate (with their full dependency - // set) when a turn actually resolves them. - [IAgentReplayBuilderService, new SyncDescriptor(AgentReplayBuilderService, [{}], true)], + // Kept delayed so it only instantiates (with its full dependency set) + // when a turn actually resolves it. [IAgentExternalHooksService, new SyncDescriptor(AgentExternalHooksService, [{}], true)], ], }, diff --git a/packages/agent-core-v2/test/contextMemory/stubs.ts b/packages/agent-core-v2/test/contextMemory/stubs.ts index 78ecbdb73..d85ae0554 100644 --- a/packages/agent-core-v2/test/contextMemory/stubs.ts +++ b/packages/agent-core-v2/test/contextMemory/stubs.ts @@ -1,6 +1,6 @@ /** * `contextMemory` test stubs — shared doubles for `IAgentContextMemoryService` and its - * collaborators (`IAgentWireRecordService`, `IAgentReplayBuilderService`). + * collaborators (`IAgentWireRecordService`, `IAgentRecordService`). * * Lives under `test/` (not `src/`) so test-support code stays out of the * production tree. Import from a relative path (`./stubs` or @@ -12,7 +12,7 @@ import type { ServiceRegistration } from '#/_base/di/test'; import { createHooks } from '#/hooks'; import type { Hooks } from '#/hooks'; import { ensureMessageId, IAgentContextMemoryService, type ContextMessage } from '#/agent/contextMemory'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; +import { IAgentRecordService } from '#/agent/record'; import { IAgentWireRecordService } from '#/agent/wireRecord'; /** @@ -36,17 +36,23 @@ export function stubWireRecord(): IAgentWireRecordService { }; } -/** A no-op `IAgentReplayBuilderService` — every mutator is a no-op. */ -export function stubReplayBuilder(): IAgentReplayBuilderService { +/** A no-op `IAgentRecordService` — every mutator is a no-op and `buildReplay` is empty. */ +export function stubRecord(): IAgentRecordService { + const hooks = createHooks(['onRestoredRecord', 'onResumeEnded']) as IAgentRecordService['hooks']; return { _serviceBrand: undefined, + restoring: null, postRestoring: false, captureLiveRecords: false, + hooks, + append: () => {}, + on: () => toDisposable(() => {}), + signal: () => {}, + define: () => toDisposable(() => {}), push: () => {}, patchLast: () => {}, removeLastMessages: () => {}, - finishRestoringRecord: () => false, - buildResult: () => [], + buildReplay: () => [], }; } @@ -94,12 +100,12 @@ export function stubContextMemory(): StubContextMemory { /** * Register the default collaborators consumed by `AgentContextMemoryService` - * (`IAgentWireRecordService`, `IAgentReplayBuilderService`) and an in-memory `IAgentContextMemoryService`. + * (`IAgentWireRecordService`, `IAgentRecordService`) and an in-memory `IAgentContextMemoryService`. * Tests that exercise the real `AgentContextMemoryService` should override * `IAgentContextMemoryService` via `additionalServices`. */ export function registerContextMemoryServices(reg: ServiceRegistration): void { reg.defineInstance(IAgentWireRecordService, stubWireRecord()); - reg.defineInstance(IAgentReplayBuilderService, stubReplayBuilder()); + reg.defineInstance(IAgentRecordService, stubRecord()); reg.defineInstance(IAgentContextMemoryService, stubContextMemory()); } diff --git a/packages/agent-core-v2/test/goal/goal.test.ts b/packages/agent-core-v2/test/goal/goal.test.ts index 9f928f7b0..3afd495e2 100644 --- a/packages/agent-core-v2/test/goal/goal.test.ts +++ b/packages/agent-core-v2/test/goal/goal.test.ts @@ -6,7 +6,7 @@ import { IAgentContextMemoryService } from '#/agent/contextMemory'; import { IAgentEventSinkService } from '#/agent/eventSink'; import { IAgentGoalService, type AgentGoalService } from '#/agent/goal'; import { IAgentLoopService } from '#/agent/loop'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; +import { IAgentRecordService } from '#/agent/record'; import { IAgentTurnService, type Turn, type TurnResult } from '#/agent/turn'; import type { PersistedWireRecord, WireRecord } from '#/agent/wireRecord'; import { recordingTelemetry, type TelemetryRecord } from '../telemetry/stubs'; @@ -95,7 +95,7 @@ describe('AgentGoalService', () => { let context: IAgentContextMemoryService; let goals: GoalServiceTestManager; let records: PersistedWireRecord[]; - let replayBuilder: IAgentReplayBuilderService; + let replayBuilder: IAgentRecordService; let events: Array<{ readonly type: string; readonly snapshot?: GoalSnapshot | null; readonly change?: GoalChange }>; let telemetry: TelemetryRecord[]; @@ -110,7 +110,7 @@ describe('AgentGoalService', () => { context = ctx.get(IAgentContextMemoryService); goals = ctx.get(IAgentGoalService) as GoalServiceTestManager; records = persistence.records; - replayBuilder = ctx.get(IAgentReplayBuilderService); + replayBuilder = ctx.get(IAgentRecordService); const eventSink = ctx.get(IAgentEventSinkService); eventSink.on((event) => { if (event.type === 'goal.updated') events.push(event); @@ -428,7 +428,7 @@ describe('AgentGoalService', () => { }, ]); - expect(replayBuilder.buildResult()).toEqual([ + expect(replayBuilder.buildReplay()).toEqual([ expect.objectContaining({ type: 'goal_updated', snapshot: expect.objectContaining({ objective: 'work', status: 'active' }), @@ -478,7 +478,7 @@ describe('AgentGoalService', () => { }, ]); - expect(replayBuilder.buildResult().at(-1)).toMatchObject({ + expect(replayBuilder.buildReplay().at(-1)).toMatchObject({ type: 'goal_updated', snapshot: { status: 'paused', terminalReason: 'Paused after agent resume' }, change: { diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index 53df5db7b..a4f794628 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -129,10 +129,10 @@ import { type InteractionResolution, } from '#/session/interaction'; import { - IAgentReplayBuilderService, - AgentReplayBuilderService, - type ReplayBuilderServiceOptions, -} from '#/agent/replayBuilder'; + AgentRecordService, + IAgentRecordService, + type RecordServiceOptions, +} from '#/agent/record'; import type { AgentAPI } from '#/agent/rpc/core-api'; import { IAgentSkillService } from '#/agent/skill/skill'; import { ISessionSkillCatalog } from '#/session/sessionSkillCatalog/skillCatalog'; @@ -145,7 +145,6 @@ import type { WireRecordRestoreOptions, WireRecordRestoreResult, } from '#/agent/wireRecord'; -import { IAgentRecordService } from '#/agent/record'; import type { PathAccessOperation } from '#/session/workspaceContext'; import { createFakeAgentFs, createFakeProcessRunner } from '../tools/fixtures/fake-exec'; @@ -623,8 +622,8 @@ export function goalServices(options: GoalServiceOptions): TestAgentServiceOverr return agentService(IAgentGoalService, new SyncDescriptor(AgentGoalService, [options])); } -export function replayServices(options: ReplayBuilderServiceOptions = {}): TestAgentServiceOverride { - return agentService(IAgentReplayBuilderService, new SyncDescriptor(AgentReplayBuilderService, [options])); +export function replayServices(options: RecordServiceOptions = {}): TestAgentServiceOverride { + return agentService(IAgentRecordService, new SyncDescriptor(AgentRecordService, [options])); } /** @@ -1024,7 +1023,6 @@ export class AgentTestContext { reg.defineDescriptor(IAgentCronService, new SyncDescriptor(AgentCronService, [{}])); reg.defineDescriptor(IAgentBackgroundService, new SyncDescriptor(AgentBackgroundService)); reg.defineDescriptor(IAgentMcpService, new SyncDescriptor(AgentMcpService, [{}])); - reg.defineDescriptor(IAgentReplayBuilderService, new SyncDescriptor(AgentReplayBuilderService, [{}])); reg.defineDescriptor(IAgentGoalService, new SyncDescriptor(AgentGoalService, [{}])); reg.defineDescriptor(IAgentSkillService, new SyncDescriptor(AgentSkillService)); reg.defineDescriptor(IAgentUserToolService, new SyncDescriptor(AgentUserToolService)); diff --git a/packages/agent-core-v2/test/message/message.test.ts b/packages/agent-core-v2/test/message/message.test.ts index 037d29c59..0a139d385 100644 --- a/packages/agent-core-v2/test/message/message.test.ts +++ b/packages/agent-core-v2/test/message/message.test.ts @@ -5,9 +5,9 @@ import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; import { IAgentContextMemoryService, type ContextMessage } from '#/agent/contextMemory'; import { AgentContextMemoryService } from '#/agent/contextMemory/contextMemoryService'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; +import { IAgentRecordService } from '#/agent/record'; import { IAgentWireRecordService } from '#/agent/wireRecord'; -import { stubReplayBuilder, stubWireRecord } from '../contextMemory/stubs'; +import { stubRecord, stubWireRecord } from '../contextMemory/stubs'; function textMessage(role: ContextMessage['role'], text: string): ContextMessage { return { @@ -37,7 +37,7 @@ describe('message history (IAgentContextMemoryService)', () => { disposables = new DisposableStore(); ix = disposables.add(new TestInstantiationService()); ix.stub(IAgentWireRecordService, stubWireRecord()); - ix.stub(IAgentReplayBuilderService, stubReplayBuilder()); + ix.stub(IAgentRecordService, stubRecord()); ix.set(IAgentContextMemoryService, new SyncDescriptor(AgentContextMemoryService)); }); afterEach(() => disposables.dispose()); diff --git a/packages/agent-core-v2/test/record/recordService.test.ts b/packages/agent-core-v2/test/record/recordService.test.ts index c53eda864..1dc523073 100644 --- a/packages/agent-core-v2/test/record/recordService.test.ts +++ b/packages/agent-core-v2/test/record/recordService.test.ts @@ -1,10 +1,9 @@ import { describe, expect, it, vi } from 'vitest'; import { DisposableStore, toDisposable } from '#/_base/di'; +import { SyncDescriptor } from '#/_base/di/descriptors'; import { createServices } from '#/_base/di/test'; import { OrderedHookSlot } from '#/hooks'; -import { IAgentEventSinkService } from '#/agent/eventSink'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; import { IAgentWireRecordService } from '#/agent/wireRecord'; import type { WireRecord, WireRecordRestoredContext } from '#/agent/wireRecord'; import { @@ -13,7 +12,7 @@ import { type AgentRecord, } from '#/agent/record'; import type { AgentEvent } from '@moonshot-ai/protocol'; -import type { AgentReplayRecord, AgentReplayRecordPayload } from '#/agent/replayBuilder/types'; +import type { AgentReplayRecordPayload } from '#/agent/replayBuilder/types'; declare module '#/agent/record' { interface AgentRecordMap { @@ -24,8 +23,6 @@ declare module '#/agent/record' { interface StubHost { readonly record: IAgentRecordService; readonly wire: ReturnType; - readonly eventSink: ReturnType; - readonly replay: ReturnType; readonly dispose: () => void; } @@ -49,6 +46,7 @@ function createWireStub() { restore: vi.fn(async () => ({}) as { warning?: string }), flush: vi.fn(async () => {}), close: vi.fn(async () => {}), + getRecords: vi.fn(() => []), get restoring() { return restoring; }, @@ -59,53 +57,31 @@ function createWireStub() { }; } -function createEventSinkStub() { - const emitted: AgentEvent[] = []; - return { - emitted, - emit: vi.fn((event: AgentEvent) => emitted.push(event)), - on: vi.fn(() => toDisposable(() => {})), - }; -} - -function createReplayStub() { - const records: AgentReplayRecord[] = []; - return { - records, - push: vi.fn((record: AgentReplayRecordPayload) => - records.push(record as unknown as AgentReplayRecord), - ), - buildResult: vi.fn(() => records), - captureLiveRecords: false, - postRestoring: false, - }; -} - -function createHost(): StubHost { +function createHost(captureLiveRecords = false): StubHost { const wire = createWireStub(); - const eventSink = createEventSinkStub(); - const replay = createReplayStub(); const disposables = new DisposableStore(); - const services = createServices(disposables, { + const ix = createServices(disposables, { additionalServices: (reg) => { reg.definePartialInstance(IAgentWireRecordService, wire); - reg.definePartialInstance(IAgentEventSinkService, eventSink); - reg.definePartialInstance(IAgentReplayBuilderService, replay); - reg.define(IAgentRecordService, AgentRecordService); }, }); + // Seed the leading static `options` argument (range) so createInstance does + // not warn about a static/service-dependency conflict. + ix.set(IAgentRecordService, new SyncDescriptor(AgentRecordService, [{}])); + const record = ix.get(IAgentRecordService); + record.captureLiveRecords = captureLiveRecords; return { - record: services.get(IAgentRecordService), + record, wire, - eventSink, - replay, dispose: () => disposables.dispose(), }; } describe('AgentRecordService facade', () => { it('append fans out to durable + live + replay facets', () => { - const host = createHost(); + const host = createHost(true); + const live: AgentEvent[] = []; + host.record.on((event) => live.push(event)); host.record.define('test.fact', { toLive: (r) => ({ type: 'test.live', value: r.value }) as unknown as AgentEvent, toReplay: (r) => @@ -117,36 +93,51 @@ describe('AgentRecordService facade', () => { expect(host.wire.append).toHaveBeenCalledWith( expect.objectContaining({ type: 'test.fact', value: 42 }), ); - expect(host.eventSink.emitted).toContainEqual( - expect.objectContaining({ type: 'test.live', value: 42 }), - ); - expect(host.replay.records).toContainEqual( + expect(live).toContainEqual(expect.objectContaining({ type: 'test.live', value: 42 })); + expect(host.record.buildReplay()).toContainEqual( expect.objectContaining({ type: 'message', value: 42 }), ); host.dispose(); }); it('append omits facets that are not declared', () => { - const host = createHost(); + const host = createHost(true); + const live: AgentEvent[] = []; + host.record.on((event) => live.push(event)); host.record.define('test.fact', {}); host.record.append({ type: 'test.fact', value: 1 }); expect(host.wire.append).toHaveBeenCalledTimes(1); - expect(host.eventSink.emit).not.toHaveBeenCalled(); - expect(host.replay.push).not.toHaveBeenCalled(); + expect(live).toHaveLength(0); + expect(host.record.buildReplay()).toHaveLength(0); + host.dispose(); + }); + + it('live append does not feed the replay buffer unless captureLiveRecords is set', () => { + const host = createHost(false); + host.record.define('test.fact', { + toReplay: (r) => + ({ type: 'message', value: r.value }) as unknown as AgentReplayRecordPayload, + }); + + host.record.append({ type: 'test.fact', value: 7 }); + + expect(host.wire.append).toHaveBeenCalledTimes(1); + expect(host.record.buildReplay()).toHaveLength(0); host.dispose(); }); it('signal emits live only and never persists or captures replay', () => { - const host = createHost(); + const host = createHost(true); + const live: AgentEvent[] = []; + host.record.on((event) => live.push(event)); + host.record.signal({ type: 'test.delta', delta: 'x' } as unknown as AgentEvent); - expect(host.eventSink.emitted).toContainEqual( - expect.objectContaining({ type: 'test.delta', delta: 'x' }), - ); + expect(live).toContainEqual(expect.objectContaining({ type: 'test.delta', delta: 'x' })); expect(host.wire.append).not.toHaveBeenCalled(); - expect(host.replay.push).not.toHaveBeenCalled(); + expect(host.record.buildReplay()).toHaveLength(0); host.dispose(); }); @@ -155,7 +146,7 @@ describe('AgentRecordService facade', () => { const resume = vi.fn(); host.record.define('test.fact', { resume }); - expect(host.wire.register).toHaveBeenCalledWith('test.fact', expect.any(Function)); + expect(host.wire.register).toHaveBeenCalledWith('test.fact', expect.any(Function), undefined); const registered = host.wire.resumers.get('test.fact'); expect(registered).toBeDefined(); await registered?.({ type: 'test.fact', value: 7 }); @@ -177,22 +168,30 @@ describe('AgentRecordService facade', () => { stop: false, }); - expect(host.replay.records).toContainEqual( - expect.objectContaining({ type: 'message', value: 5 }), + expect(host.record.buildReplay()).toContainEqual( + expect.objectContaining({ type: 'message', value: 5, time: 123 }), ); host.dispose(); }); - it('on() delegates to the live event sink', () => { + it('suppresses live emission while restoring', () => { const host = createHost(); - const handler = vi.fn(); - host.record.on(handler); - expect(host.eventSink.on).toHaveBeenCalledWith(handler); + const live: AgentEvent[] = []; + host.record.on((event) => live.push(event)); + host.record.define('test.fact', { + toLive: (r) => ({ type: 'test.live', value: r.value }) as unknown as AgentEvent, + }); + host.wire.setRestoring({ time: 1 }); + + host.record.append({ type: 'test.fact', value: 3 }); + + expect(host.wire.append).toHaveBeenCalledTimes(1); + expect(live).toHaveLength(0); host.dispose(); }); it('dispose returned by define unregisters the resumer and facets', () => { - const host = createHost(); + const host = createHost(true); const subscription = host.record.define('test.fact', { resume: vi.fn() }); expect(host.wire.resumers.has('test.fact')).toBe(true); diff --git a/packages/agent-core-v2/test/skill/skill-tool-manager.test.ts b/packages/agent-core-v2/test/skill/skill-tool-manager.test.ts index 687b4d87c..6b9144914 100644 --- a/packages/agent-core-v2/test/skill/skill-tool-manager.test.ts +++ b/packages/agent-core-v2/test/skill/skill-tool-manager.test.ts @@ -8,7 +8,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { IAgentContextMemoryService } from '#/agent/contextMemory'; import { IAgentEventSinkService } from '#/agent/eventSink'; import { IAgentProfileService } from '#/agent/profile'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; +import { IAgentRecordService } from '#/agent/record'; import { InMemorySkillCatalog, type SkillCatalog, type SkillDefinition } from '#/app/globalSkillCatalog'; import { IAgentToolRegistryService } from '#/agent/toolRegistry'; import { @@ -282,7 +282,7 @@ describe('ToolManager SkillTool wire behavior', () => { describe('ToolManager SkillTool restore behavior', () => { let ctx: TestAgentContext; let context: IAgentContextMemoryService; - let replay: IAgentReplayBuilderService; + let replay: IAgentRecordService; let skills: InMemorySkillCatalog; let emit: ReturnType; let track: ReturnType; @@ -298,7 +298,7 @@ describe('ToolManager SkillTool restore behavior', () => { ); context = ctx.get(IAgentContextMemoryService); const events = ctx.get(IAgentEventSinkService); - replay = ctx.get(IAgentReplayBuilderService); + replay = ctx.get(IAgentRecordService); emit = vi.spyOn(events, 'emit'); }); @@ -351,7 +351,7 @@ describe('ToolManager SkillTool restore behavior', () => { ); expect(track).not.toHaveBeenCalledWith('skill_invoked', expect.anything()); expect(context.get()).toMatchObject([message]); - expect(replay.buildResult()).toContainEqual( + expect(replay.buildReplay()).toContainEqual( expect.objectContaining({ type: 'message', message: expect.objectContaining({ diff --git a/packages/agent-core-v2/test/skill/skill.test.ts b/packages/agent-core-v2/test/skill/skill.test.ts index b2b7cc5fe..1d2db0cc1 100644 --- a/packages/agent-core-v2/test/skill/skill.test.ts +++ b/packages/agent-core-v2/test/skill/skill.test.ts @@ -18,7 +18,6 @@ import { ITelemetryService } from '#/app/telemetry'; import { IAgentToolRegistryService } from '#/agent/toolRegistry'; import type { Turn } from '#/agent/turn'; import { IAgentWireRecordService } from '#/agent/wireRecord'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; import { AgentRecordService, IAgentRecordService } from '#/agent/record'; import { stubWireRecord } from '../contextMemory/stubs'; import { executeTool } from '../tools/fixtures/execute-tool'; @@ -67,12 +66,6 @@ describe('AgentSkillService', () => { clear: () => {}, }); reg.defineInstance(IAgentWireRecordService, stubWireRecord()); - reg.definePartialInstance(IAgentReplayBuilderService, { - push: () => {}, - buildResult: () => [], - captureLiveRecords: false, - postRestoring: false, - }); reg.define(IAgentRecordService, AgentRecordService); reg.definePartialInstance(ITelemetryService, { track: () => {} }); reg.definePartialInstance(IAgentToolRegistryService, { @@ -170,12 +163,6 @@ describe('SkillTool', () => { clear: () => {}, }); reg.defineInstance(IAgentWireRecordService, stubWireRecord()); - reg.definePartialInstance(IAgentReplayBuilderService, { - push: () => {}, - buildResult: () => [], - captureLiveRecords: false, - postRestoring: false, - }); reg.define(IAgentRecordService, AgentRecordService); reg.definePartialInstance(ITelemetryService, { track: () => {} }); reg.definePartialInstance(IAgentToolRegistryService, { diff --git a/packages/agent-core-v2/test/usage/usage.test.ts b/packages/agent-core-v2/test/usage/usage.test.ts index bd61f5078..15709820a 100644 --- a/packages/agent-core-v2/test/usage/usage.test.ts +++ b/packages/agent-core-v2/test/usage/usage.test.ts @@ -5,7 +5,6 @@ import { DisposableStore } from '#/_base/di/lifecycle'; import { createServices, type TestInstantiationService } from '#/_base/di/test'; import { OrderedHookSlot } from '#/hooks'; import { IAgentEventSinkService } from '#/agent/eventSink'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; import { AgentRecordService, IAgentRecordService } from '#/agent/record'; import { IAgentUsageService, type UsageStatus } from '#/agent/usage'; import { AgentUsageService } from '#/agent/usage/usageService'; @@ -243,12 +242,6 @@ function createUsageHarness(): { }, on: () => toDisposable(() => {}), }); - reg.definePartialInstance(IAgentReplayBuilderService, { - push: () => {}, - buildResult: () => [], - captureLiveRecords: false, - postRestoring: false, - }); reg.define(IAgentRecordService, AgentRecordService); reg.define(IAgentUsageService, AgentUsageService); }, diff --git a/packages/agent-core-v2/test/wireRecord/index.test.ts b/packages/agent-core-v2/test/wireRecord/index.test.ts index 660acb08e..fcfcc01d2 100644 --- a/packages/agent-core-v2/test/wireRecord/index.test.ts +++ b/packages/agent-core-v2/test/wireRecord/index.test.ts @@ -5,12 +5,12 @@ import { IAgentContextMemoryService, IAgentContextSizeService, IAgentFullCompactionService, - IAgentReplayBuilderService, + IAgentRecordService, IAgentWireRecordService, type ContextMessage, type PersistedWireRecord, } from '#/index'; -import type { ReplayRangeOptions } from '#/agent/replayBuilder'; +import type { ReplayRangeOptions } from '#/agent/record'; import { InMemoryWireRecordPersistence, createTestAgent, @@ -26,7 +26,7 @@ describe('AgentRecords persistence metadata', () => { let expectResumeMatches: boolean; let persistence: RecordingInMemoryWireRecordPersistence; let records: IAgentWireRecordService; - let replay: IAgentReplayBuilderService; + let replay: IAgentRecordService; beforeEach(() => { expectResumeMatches = true; @@ -35,7 +35,7 @@ describe('AgentRecords persistence metadata', () => { context = ctx.get(IAgentContextMemoryService); contextSize = ctx.get(IAgentContextSizeService); records = ctx.get(IAgentWireRecordService); - replay = ctx.get(IAgentReplayBuilderService); + replay = ctx.get(IAgentRecordService); }); afterEach(async () => { @@ -237,7 +237,7 @@ describe('AgentRecords persistence metadata', () => { await expect(ctx.restorePersisted()).resolves.toEqual({}); expect(context.get()).toHaveLength(0); - expect(replay.buildResult()).toEqual([ + expect(replay.buildReplay()).toEqual([ expect.objectContaining({ type: 'goal_updated', snapshot: expect.objectContaining({ goalId: 'g1', status: 'active' }), @@ -652,12 +652,12 @@ async function buildReplayFromPersistence( replayServices(range === undefined ? {} : { range }), ); const fullCompaction = ctx.get(IAgentFullCompactionService); - const replay = ctx.get(IAgentReplayBuilderService); + const replay = ctx.get(IAgentRecordService); try { const isCompacting = fullCompaction.isCompacting; if (isCompacting) throw new Error('Unexpected active compaction before restore'); await ctx.restorePersisted({ rewriteMigratedRecords: false }); - return replay.buildResult(); + return replay.buildReplay(); } finally { try { await ctx.expectResumeMatches(); diff --git a/packages/agent-core-v2/test/wireRecord/persistence.test.ts b/packages/agent-core-v2/test/wireRecord/persistence.test.ts index c592b5bdc..1ee817660 100644 --- a/packages/agent-core-v2/test/wireRecord/persistence.test.ts +++ b/packages/agent-core-v2/test/wireRecord/persistence.test.ts @@ -14,7 +14,7 @@ import { IBootstrapService } from '#/app/bootstrap'; import { IHostFileSystem, HostFileSystem } from '#/app/hostFs'; import { AgentContextMemoryService } from '#/agent/contextMemory/contextMemoryService'; import { IAgentContextMemoryService, type ContextMessage } from '#/agent/contextMemory'; -import { IAgentReplayBuilderService } from '#/agent/replayBuilder'; +import { IAgentRecordService } from '#/agent/record'; import { AppendLogStore, AGENT_WIRE_PROTOCOL_VERSION, @@ -29,7 +29,7 @@ import { FileStorageService } from '#/app/storage/fileStorageService'; import { InMemoryStorageService } from '#/app/storage/inMemoryStorageService'; import type { IStorageService } from '#/app/storage'; import { stubBootstrap } from '../bootstrap/stubs'; -import { stubReplayBuilder } from '../contextMemory/stubs'; +import { stubRecord } from '../contextMemory/stubs'; const cleanups: string[] = []; const disposables: DisposableStore[] = []; @@ -101,7 +101,7 @@ async function createWireHarness(): Promise<{ ix.stub(IBlobStorage, storage); ix.stub(IBootstrapService, stubBootstrap(dir)); ix.stub(IHostFileSystem, new HostFileSystem()); - ix.stub(IAgentReplayBuilderService, stubReplayBuilder()); + ix.stub(IAgentRecordService, stubRecord()); ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); ix.set(IAgentBlobStoreService, new SyncDescriptor(AgentBlobStoreService)); ix.set(IAgentWireRecordService, new SyncDescriptor(AgentWireRecordService, [{}])); diff --git a/packages/agent-core-v2/test/wireRecord/resume.test.ts b/packages/agent-core-v2/test/wireRecord/resume.test.ts index fcf9f8682..f10dec7f3 100644 --- a/packages/agent-core-v2/test/wireRecord/resume.test.ts +++ b/packages/agent-core-v2/test/wireRecord/resume.test.ts @@ -6,7 +6,7 @@ import { describe, expect, it, vi } from 'vitest'; import { AGENT_WIRE_PROTOCOL_VERSION, - IAgentReplayBuilderService, + IAgentRecordService, type PersistedWireRecord, type PromptOrigin, } from '#/index'; @@ -483,7 +483,7 @@ describe('Agent resume', () => { origin: { kind: 'compaction_summary' }, }), ]); - expect(ctx.get(IAgentReplayBuilderService).buildResult()).toEqual([ + expect(ctx.get(IAgentRecordService).buildReplay()).toEqual([ expect.objectContaining({ type: 'message', message: expect.objectContaining({ @@ -519,7 +519,7 @@ describe('Agent resume', () => { await ctx.restorePersisted(); - expect(ctx.get(IAgentReplayBuilderService).buildResult()).toEqual([ + expect(ctx.get(IAgentRecordService).buildReplay()).toEqual([ expect.objectContaining({ type: 'compaction', result: 'cancelled', @@ -632,7 +632,7 @@ describe('Agent resume', () => { await ctx.restorePersisted(); - expect(ctx.get(IAgentReplayBuilderService).buildResult()).toContainEqual( + expect(ctx.get(IAgentRecordService).buildReplay()).toContainEqual( expect.objectContaining({ type: 'message', message: expect.objectContaining({ @@ -733,7 +733,7 @@ describe('Agent resume', () => { await ctx.restorePersisted(); expect(ctx.context.get()).toHaveLength(0); - expect(ctx.get(IAgentReplayBuilderService).buildResult()).toContainEqual( + expect(ctx.get(IAgentRecordService).buildReplay()).toContainEqual( expect.objectContaining({ type: 'goal_updated', snapshot: expect.objectContaining({ @@ -847,7 +847,7 @@ describe('Agent resume', () => { expect(ctx.context.get()[0]?.role).toBe('user'); expect(ctx.context.get()[1]?.role).toBe('assistant'); - const replay = ctx.get(IAgentReplayBuilderService).buildResult(); + const replay = ctx.get(IAgentRecordService).buildReplay(); expect(replay).toHaveLength(2); expect(replay[0]).toMatchObject({ type: 'message',