mirror of
https://github.com/MoonshotAI/kimi-code.git
synced 2026-08-21 06:35:50 +00:00
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.
This commit is contained in:
haozhe.yang
parent
643acf454a
commit
8ee741007b
24 changed files with 293 additions and 391 deletions
|
|
@ -70,7 +70,6 @@ package "Agent scope (per agent)" #FDF5E6 {
|
|||
rectangle "<b>contextInjector</b>\n<size:9><i>Agent</i></size>\n IAgentContextInjectorService" as contextInjector #FDEBD0
|
||||
rectangle "<b>contextSize</b>\n<size:9><i>Agent</i></size>\n IAgentContextSizeService" as contextSize #FDEBD0
|
||||
rectangle "<b>systemReminder</b>\n<size:9><i>Agent</i></size>\n IAgentSystemReminderService" as systemReminder #FDEBD0
|
||||
rectangle "<b>replayBuilder</b>\n<size:9><i>Agent</i></size>\n IAgentReplayBuilderService" as replayBuilder #FDEBD0
|
||||
rectangle "<b>profile</b>\n<size:9><i>Agent</i></size>\n IAgentProfileService" as profile #FDEBD0
|
||||
rectangle "<b>prompt</b>\n<size:9><i>Agent</i></size>\n IAgentPromptService" as prompt #FDEBD0
|
||||
rectangle "<b>turn</b>\n<size:9><i>Agent</i></size>\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
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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<K extends keyof AgentRecordMap> {
|
|||
readonly blobs?: WireRecordBlobSelector<AgentRecord<K>>;
|
||||
}
|
||||
|
||||
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<K extends keyof AgentRecordMap>(type: K, facets: RecordFacets<K>): IDisposable;
|
||||
|
||||
restore(
|
||||
records?: readonly PersistedWireRecord[],
|
||||
options?: WireRecordRestoreOptions,
|
||||
): Promise<WireRecordRestoreResult>;
|
||||
flush(): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
|
||||
/** 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<T extends AgentReplayRecord['type']>(
|
||||
type: T,
|
||||
patch: Partial<Extract<AgentReplayRecord, { type: T }>>,
|
||||
): void;
|
||||
/** Drop replay `message` records whose message is in `removedMessages`. */
|
||||
removeLastMessages(removedMessages: ReadonlySet<ContextMessage>): 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;
|
||||
|
|
|
|||
|
|
@ -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<AgentEvent>`), 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<AgentEvent>`), 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<keyof AgentRecordMap, RecordFacets<keyof AgentRecordMap>>();
|
||||
private readonly liveEmitter = this._register(new Emitter<AgentEvent>());
|
||||
|
||||
// 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<WireRecordRestoreResult> {
|
||||
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<void> {
|
||||
return this.wireRecord.flush();
|
||||
patchLast<T extends AgentReplayRecord['type']>(
|
||||
type: T,
|
||||
patch: Partial<Extract<AgentReplayRecord, { type: T }>>,
|
||||
): 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<void> {
|
||||
return this.wireRecord.close();
|
||||
removeLastMessages(removedMessages: ReadonlySet<ContextMessage>): 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<ContextMessage>,
|
||||
): 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
|
|||
|
|
@ -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<T extends AgentReplayRecord['type']>(
|
||||
type: T,
|
||||
patch: Partial<Extract<AgentReplayRecord, { type: T }>>,
|
||||
): void;
|
||||
removeLastMessages(removedMessages: ReadonlySet<ContextMessage>): void;
|
||||
finishRestoringRecord(record: WireRecord): boolean;
|
||||
buildResult(): readonly AgentReplayRecord[];
|
||||
}
|
||||
|
||||
export const IAgentReplayBuilderService = createDecorator<IAgentReplayBuilderService>(
|
||||
'agentReplayBuilderService',
|
||||
);
|
||||
|
|
@ -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<T extends AgentReplayRecord['type']>(
|
||||
type: T,
|
||||
patch: Partial<Extract<AgentReplayRecord, { type: T }>>,
|
||||
): 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<ContextMessage>): 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<ContextMessage>,
|
||||
): 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',
|
||||
);
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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)],
|
||||
],
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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: {
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
|
|
|
|||
|
|
@ -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<typeof createWireStub>;
|
||||
readonly eventSink: ReturnType<typeof createEventSinkStub>;
|
||||
readonly replay: ReturnType<typeof createReplayStub>;
|
||||
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);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<typeof vi.spyOn>;
|
||||
let track: ReturnType<typeof vi.spyOn>;
|
||||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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, {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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, [{}]));
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue