From 0491b06339a33f03c21c6b7f9fb6b14a3e1381b7 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Sat, 4 Jul 2026 11:06:22 +0800 Subject: [PATCH 1/5] refactor: route v2 context changes through context ops --- .../agent-core-v2/docs/di-scope-domains.puml | 12 +- .../agent-core-v2/docs/di-scope-domains.svg | 2 +- .../scripts/check-domain-layers.mjs | 8 +- .../src/agent/contextMemory/contextMemory.ts | 54 ++++-- .../contextMemory/contextMemoryService.ts | 174 +++++++++++------- .../src/agent/contextMemory/messageId.ts | 11 +- .../src/agent/contextOps/contextOps.ts | 46 +++++ .../src/agent/contextOps/contextOpsService.ts | 166 +++++++++++++++++ .../src/agent/contextOps/index.ts | 2 + .../fullCompaction/fullCompactionService.ts | 43 ++++- .../src/agent/goal/goalService.ts | 4 +- .../src/agent/loop/loopService.ts | 13 +- .../src/agent/prompt/promptService.ts | 42 +++-- .../agent-core-v2/src/agent/record/record.ts | 23 ++- .../src/agent/record/recordService.ts | 87 ++++----- .../systemReminder/systemReminderService.ts | 12 +- .../src/agent/task/taskService.ts | 19 +- .../src/agent/toolDedupe/toolDedupeService.ts | 35 ++++ .../src/agent/wireRecord/migration/v1.5.ts | 74 +++----- .../sessionLifecycleService.ts | 17 +- packages/agent-core-v2/src/index.ts | 1 + .../agentLifecycle/agentLifecycleService.ts | 3 +- .../agentLifecycle/contextOperationOwners.ts | 31 ++++ .../src/session/agentLifecycle/index.ts | 1 + .../test/contextInjector/manager.test.ts | 9 +- .../test/contextMemory/splice-replay.test.ts | 161 +++++++++++----- .../agent-core-v2/test/contextMemory/stubs.ts | 54 ++++-- packages/agent-core-v2/test/harness/agent.ts | 15 +- .../test/message/message.test.ts | 6 +- .../agent-core-v2/test/swarm/swarm.test.ts | 2 + .../test/toolDedup/tool-dedup.test.ts | 3 +- .../test/turn/turn-ready.test.ts | 3 + .../test/wireRecord/migration/v1.5.test.ts | 30 ++- .../test/wireRecord/persistence.test.ts | 27 +-- packages/server-v2/src/transport/actionMap.ts | 1 - 35 files changed, 850 insertions(+), 341 deletions(-) create mode 100644 packages/agent-core-v2/src/agent/contextOps/contextOps.ts create mode 100644 packages/agent-core-v2/src/agent/contextOps/contextOpsService.ts create mode 100644 packages/agent-core-v2/src/agent/contextOps/index.ts create mode 100644 packages/agent-core-v2/src/session/agentLifecycle/contextOperationOwners.ts diff --git a/packages/agent-core-v2/docs/di-scope-domains.puml b/packages/agent-core-v2/docs/di-scope-domains.puml index f2ef59abf..5c93cae0c 100644 --- a/packages/agent-core-v2/docs/di-scope-domains.puml +++ b/packages/agent-core-v2/docs/di-scope-domains.puml @@ -70,6 +70,7 @@ package "Agent scope (per agent)" #FDF5E6 { rectangle "record\nAgent\n IAgentRecordService" as record #FDEBD0 rectangle "blobStore\nAgent\n IAgentBlobStoreService" as blobStore #FDEBD0 rectangle "contextMemory\nAgent\n IAgentContextMemoryService" as contextMemory #FDEBD0 + rectangle "contextOps\nAgent\n IAgentContextOpsService" as contextOps #FDEBD0 rectangle "contextProjector\nAgent\n IAgentContextProjectorService" as contextProjector #FDEBD0 rectangle "contextInjector\nAgent\n IAgentContextInjectorService" as contextInjector #FDEBD0 rectangle "contextSize\nAgent\n IAgentContextSizeService" as contextSize #FDEBD0 @@ -164,6 +165,7 @@ filestore --> storage #34495E record --> wireRecord #34495E contextMemory --> wireRecord #34495E contextMemory --> record #34495E +contextOps --> contextMemory #34495E contextInjector --> contextMemory #34495E contextInjector --> turn #34495E contextInjector --> loop #34495E @@ -172,6 +174,7 @@ contextSize --> contextMemory #34495E contextSize --> record #34495E contextSize --> wireRecord #34495E systemReminder --> contextMemory #34495E +systemReminder --> contextOps #34495E profile --> wireRecord #34495E profile --> record #34495E profile --> record #34495E @@ -180,6 +183,7 @@ profile --> config #34495E profile --> modelProvider #34495E profile --> chatProvider #34495E prompt --> contextMemory #34495E +prompt --> contextOps #34495E prompt --> turn #34495E prompt --> loop #34495E prompt --> wireRecord #34495E @@ -188,8 +192,9 @@ turn --> loop #34495E turn --> record #34495E turn --> wireRecord #34495E turn --> contextMemory #34495E +turn --> contextOps #34495E turn --> telemetry #34495E -loop --> contextMemory #34495E +loop --> contextOps #34495E loop --> contextSize #34495E loop --> llmRequester #34495E loop --> record #34495E @@ -242,6 +247,7 @@ goal --> record #34495E goal --> telemetry #34495E goal --> contextInjector #34495E goal --> contextMemory #34495E +goal --> contextOps #34495E goal --> turn #34495E goal --> loop #34495E goal --> toolRegistry #34495E @@ -263,6 +269,7 @@ task --> wireRecord #34495E task --> telemetry #34495E task --> prompt #34495E task --> contextMemory #34495E +task --> contextOps #34495E task --> config #34495E task --> storage #34495E task --> session_context #34495E @@ -348,7 +355,7 @@ flag ..> config #16A085 : onDidChangeConfiguration permissionMode ..> wireRecord #16A085 : permission.set_mode userTool ..> wireRecord #16A085 : tools.register_/unregister_user_tool profile ..> wireRecord #16A085 : config.update / tools.set_active_tools -contextMemory ..> wireRecord #16A085 : context.splice +contextMemory ..> wireRecord #16A085 : context. ops cron ..> wireRecord #16A085 : cron.add / delete / cursor todo ..> wireRecord #16A085 : todo.set fullCompaction ..> wireRecord #16A085 : full_compaction.begin/cancel/complete @@ -375,7 +382,6 @@ toolStore ..> wireRecord #16A085 : tools.update_store usage ..> wireRecord #16A085 : usage.record contextSize ..> contextMemory #16A085 : hooks.onSpliced contextSize ..> wireRecord #16A085 : context_size.measured -loop ..> contextMemory #16A085 : hooks.onSpliced loop ..> wireRecord #16A085 : hooks.onResumeEnded plan ..> wireRecord #16A085 : plan_mode.enter/cancel/exit diff --git a/packages/agent-core-v2/docs/di-scope-domains.svg b/packages/agent-core-v2/docs/di-scope-domains.svg index b562fcc0d..07c75fe34 100644 --- a/packages/agent-core-v2/docs/di-scope-domains.svg +++ b/packages/agent-core-v2/docs/di-scope-domains.svg @@ -1 +1 @@ -App scope (process-wide)Session scope (per session)Agent scope (per agent)bootstrapAppIBootstrapServicelogAppILogServiceILogWriterServicetelemetryAppITelemetryServiceeventAppIEventServicestorageAppIStorageServiceIAppendLogStoreIAtomicDocumentStoreIAtomicTomlDocumentStorefilestoreAppIFileStoregatewayAppIRestGatewayIWSGatewaysessionLifecycleAppISessionLifecycleServicesessionIndexAppISessionIndexhostFsAppIHostFileSystemworkspaceRegistryAppIWorkspaceRegistryhostFolderBrowserAppIHostFolderBrowserhostEnvironmentAppIHostEnvironmenthostProcessAppIHostProcessServiceauthAppIOAuthServiceIAuthSummaryServiceIWebSearchProviderServicewebAppIWebFetchServiceproviderAppIProviderServiceflagAppIFlagServiceIFlagRegistryconfigAppIConfigRegistryIConfigServicepluginAppIPluginServicechatProviderAppIChatProviderFactorymodelAppIModelServicemodelCatalogAppIModelCatalogServiceglobalSkillCatalogAppIGlobalSkillCatalogISkillCatalogStoresessionContextSessionISessionContext (seed)sessionMetadataSessionISessionMetadatasessionActivitySessionISessionActivityagentLifecycleSessionIAgentLifecycleServiceinteractionSessionIInteractionServiceworkspaceContextSessionIWorkspaceContextsessionLogSessionISessionLogServiceILogWriterServicesessionSkillCatalogSessionISessionSkillCatalogexecContextSessionIExecContext (seed)agentFsSessionIAgentFileSystemIFsServiceapprovalSessionIApprovalServicequestionSessionIQuestionServiceprocessSessionIProcessRunnerIProcessterminalAppIHostTerminalServicesessionTerminalSessionISessionTerminalServicemodelProviderSessionIModelProvider (seed)todoSessionISessionTodoServicewireRecordAgentIAgentWireRecordService (event hub)recordAgentIAgentRecordServiceblobStoreAgentIAgentBlobStoreServicecontextMemoryAgentIAgentContextMemoryServicecontextProjectorAgentIAgentContextProjectorServicecontextInjectorAgentIAgentContextInjectorServicecontextSizeAgentIAgentContextSizeServicesystemReminderAgentIAgentSystemReminderServiceprofileAgentIAgentProfileServicepromptAgentIAgentPromptServiceturnAgentIAgentTurnServiceloopAgentIAgentLoopServicellmRequesterAgentIAgentLLMRequesterServicetoolRegistryAgentIAgentToolRegistryServicetoolExecutorAgentIAgentToolExecutorServicetoolStoreAgentIAgentToolStoreServicetoolDedupAgentIAgentToolDedupeServicepermissionGateAgentIAgentPermissionGatepermissionModeAgentIAgentPermissionModeServicepermissionPolicyAgentIAgentPermissionPolicyServicepermissionRulesAgentIAgentPermissionRulesServiceplanAgentIAgentPlanServicegoalAgentIAgentGoalServiceskillAgentIAgentSkillServicequestionToolsAgentIAgentQuestionToolsServiceuserToolAgentIAgentUserToolServicetaskAgentIAgentTaskServicecronAgentIAgentCronServiceswarmAgentIAgentSwarmServicemcpAgentIAgentMcpServicefullCompactionAgentIAgentFullCompactionServicemicroCompactionAgentIAgentMicroCompactionServiceexternalHooksAgentIAgentExternalHooksServiceusageAgentIAgentUsageServicerpcAgentIAgentRPCServicefileToolsAgentRead/Write/Grep/Glob toolseditAgentEditToolFileEditServiceshellToolsAgentIAgentShellToolsServicescopeContextAgentIAgentScopeContext (seed)environmenteventSinkonDidChangeConfigurationcontext.splicehooks.onSplicedcontext_size.measuredconfig.update / tools.set_active_toolshooks.onSplicedtools.update_storepermission.set_modepermission.rules.add / record_approval_resultplan_mode.enter/cancel/exitgoal.create/update/clearturn lifecycle hooksstep/usage hooksskill.activatetools.register_/unregister_user_tooltask.started/terminatedhooks.onSplicedcron.add / delete / cursorswarm_mode.enter/exitfull_compaction.begin/cancel/completehooks.onErrormicro_compaction.apply / full_compaction.completeusage.recordtodo.sethook.result / goal.updatedhooks.onEndedonWillExecuteTool / onDidExecuteToolapproval hooksprompt/end hooksstop hookcompaction hooksnotification hookSubagentStart/Stop (mirrorAgentRun)hooks.onResumeEndedScope = node color      App (process-wide)      Session (per session)      Agent (per agent)Edgessolid: DI injection (ctor @IX)dashed: event-driven (subscribe/emit)direction: consumer ---> providerNotesGenerated from `node scripts/dep-graph.mjs` output;`_base` / seed / options deps are omitted. \ No newline at end of file +App scope (process-wide)Session scope (per session)Agent scope (per agent)bootstrapAppIBootstrapServicelogAppILogServiceILogWriterServicetelemetryAppITelemetryServiceeventAppIEventServicestorageAppIStorageServiceIAppendLogStoreIAtomicDocumentStoreIAtomicTomlDocumentStorefilestoreAppIFileStoregatewayAppIRestGatewayIWSGatewaysessionLifecycleAppISessionLifecycleServicesessionIndexAppISessionIndexhostFsAppIHostFileSystemworkspaceRegistryAppIWorkspaceRegistryhostFolderBrowserAppIHostFolderBrowserhostEnvironmentAppIHostEnvironmenthostProcessAppIHostProcessServiceauthAppIOAuthServiceIAuthSummaryServiceIWebSearchProviderServicewebAppIWebFetchServiceproviderAppIProviderServiceflagAppIFlagServiceIFlagRegistryconfigAppIConfigRegistryIConfigServicepluginAppIPluginServicechatProviderAppIChatProviderFactorymodelAppIModelServicemodelCatalogAppIModelCatalogServiceglobalSkillCatalogAppIGlobalSkillCatalogISkillCatalogStoresessionContextSessionISessionContext (seed)sessionMetadataSessionISessionMetadatasessionActivitySessionISessionActivityagentLifecycleSessionIAgentLifecycleServiceinteractionSessionIInteractionServiceworkspaceContextSessionIWorkspaceContextsessionLogSessionISessionLogServiceILogWriterServicesessionSkillCatalogSessionISessionSkillCatalogexecContextSessionIExecContext (seed)agentFsSessionIAgentFileSystemIFsServiceapprovalSessionIApprovalServicequestionSessionIQuestionServiceprocessSessionIProcessRunnerIProcessterminalAppIHostTerminalServicesessionTerminalSessionISessionTerminalServicemodelProviderSessionIModelProvider (seed)todoSessionISessionTodoServicewireRecordAgentIAgentWireRecordService (event hub)recordAgentIAgentRecordServiceblobStoreAgentIAgentBlobStoreServicecontextMemoryAgentIAgentContextMemoryServicecontextOpsAgentIAgentContextOpsServicecontextProjectorAgentIAgentContextProjectorServicecontextInjectorAgentIAgentContextInjectorServicecontextSizeAgentIAgentContextSizeServicesystemReminderAgentIAgentSystemReminderServiceprofileAgentIAgentProfileServicepromptAgentIAgentPromptServiceturnAgentIAgentTurnServiceloopAgentIAgentLoopServicellmRequesterAgentIAgentLLMRequesterServicetoolRegistryAgentIAgentToolRegistryServicetoolExecutorAgentIAgentToolExecutorServicetoolStoreAgentIAgentToolStoreServicetoolDedupAgentIAgentToolDedupeServicepermissionGateAgentIAgentPermissionGatepermissionModeAgentIAgentPermissionModeServicepermissionPolicyAgentIAgentPermissionPolicyServicepermissionRulesAgentIAgentPermissionRulesServiceplanAgentIAgentPlanServicegoalAgentIAgentGoalServiceskillAgentIAgentSkillServicequestionToolsAgentIAgentQuestionToolsServiceuserToolAgentIAgentUserToolServicetaskAgentIAgentTaskServicecronAgentIAgentCronServiceswarmAgentIAgentSwarmServicemcpAgentIAgentMcpServicefullCompactionAgentIAgentFullCompactionServicemicroCompactionAgentIAgentMicroCompactionServiceexternalHooksAgentIAgentExternalHooksServiceusageAgentIAgentUsageServicerpcAgentIAgentRPCServicefileToolsAgentRead/Write/Grep/Glob toolseditAgentEditToolFileEditServiceshellToolsAgentIAgentShellToolsServicescopeContextAgentIAgentScopeContext (seed)environmenteventSinkonDidChangeConfigurationcontext.<type> opshooks.onSplicedcontext_size.measuredconfig.update / tools.set_active_toolstools.update_storepermission.set_modepermission.rules.add / record_approval_resultplan_mode.enter/cancel/exitgoal.create/update/clearturn lifecycle hooksstep/usage hooksskill.activatetools.register_/unregister_user_tooltask.started/terminatedhooks.onSplicedcron.add / delete / cursorswarm_mode.enter/exitfull_compaction.begin/cancel/completehooks.onErrormicro_compaction.apply / full_compaction.completeusage.recordtodo.sethook.result / goal.updatedhooks.onEndedonWillExecuteTool / onDidExecuteToolapproval hooksprompt/end hooksstop hookcompaction hooksnotification hookSubagentStart/Stop (mirrorAgentRun)hooks.onResumeEndedScope = node color      App (process-wide)      Session (per session)      Agent (per agent)Edgessolid: DI injection (ctor @IX)dashed: event-driven (subscribe/emit)direction: consumer ---> providerNotesGenerated from `node scripts/dep-graph.mjs` output;`_base` / seed / options deps are omitted. \ No newline at end of file diff --git a/packages/agent-core-v2/scripts/check-domain-layers.mjs b/packages/agent-core-v2/scripts/check-domain-layers.mjs index 5c00e6070..06f1ea0cf 100644 --- a/packages/agent-core-v2/scripts/check-domain-layers.mjs +++ b/packages/agent-core-v2/scripts/check-domain-layers.mjs @@ -139,6 +139,7 @@ const DOMAIN_LAYER = new Map([ ['usage', 4], ['toolDedupe', 4], ['contextMemory', 4], + ['contextOps', 4], ['contextInjector', 4], ['systemReminder', 4], ['contextProjector', 4], @@ -297,10 +298,6 @@ 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', @@ -314,6 +311,9 @@ const ALLOWED_EXCEPTIONS = new Set([ 'toolExecutor>loop', 'userTool>profile', 'wireRecord>contextMemory', + // The v1.5 migration emits `context.*` operation records and references the + // arg tuple types declared by their owning domains (L4) — type-only. + 'wireRecord>contextOps', 'wireRecord>loop', 'wireRecord>tool', 'hostFolderBrowser>os/backends', diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts b/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts index 56432d637..673b747c7 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts @@ -1,25 +1,55 @@ import { createDecorator } from "#/_base/di"; +import type { ContentPart } from '#/app/llmProtocol'; import type { Hooks } from '#/hooks'; +import type { AgentReplayRecordPayload } from '#/agent/replayBuilder/types'; import type { ContextMessage } from './types'; +export interface ContextSplicedEvent { + start: number; + deleteCount: number; + messages: ContextMessage[]; + tokens?: number; +} + +export type ContextSplice = ( + start: number, + deleteCount: number, + insert: readonly ContextMessage[], + tokens?: number, +) => void; + +export interface ContextReplayWriter { + push(record: AgentReplayRecordPayload): void; + removeMessages(messageIds: ReadonlySet): void; + cut(): void; +} + +export interface ContextOperationBlobTarget { + readonly parts: readonly ContentPart[]; + replace(args: T, parts: readonly ContentPart[]): T; +} + +export interface ContextOperationDefinition { + readonly type: string; + readonly apply: (splice: ContextSplice, ...args: T) => void; + readonly replay: (replay: ContextReplayWriter, ...args: T) => void; + readonly blobs?: (args: T) => Iterable>; +} + +export type ContextOperation = (...args: T) => void; + export interface IAgentContextMemoryService { readonly _serviceBrand: undefined; + + defineOperation( + definition: ContextOperationDefinition, + ): ContextOperation; + get(): readonly ContextMessage[]; - splice( - start: number, - deleteCount: number, - messages: readonly ContextMessage[], - tokens?: number, - ): void; readonly hooks: Hooks<{ - onSpliced: { - start: number; - deleteCount: number; - messages: ContextMessage[]; - tokens?: number; - }; + onSpliced: ContextSplicedEvent; }>; } diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts index c0f32e960..63624df41 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts @@ -3,106 +3,154 @@ import { } from "#/_base/di"; import { OrderedHookSlot } from '#/hooks'; import { IAgentRecordService, type AgentRecord } from '#/agent/record'; -import { IAgentContextMemoryService } from './contextMemory'; +import type { WireRecordBlobSelector } from '#/agent/wireRecord'; +import { + IAgentContextMemoryService, + type ContextOperation, + type ContextOperationDefinition, + type ContextReplayWriter, + type ContextSplice, + type ContextSplicedEvent, +} from './contextMemory'; import { ensureMessageId } from './messageId'; import type { ContextMessage } from './types'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +const CONTEXT_OPERATION_PREFIX = 'context.'; + declare module '#/agent/wireRecord' { interface WireRecordMap { - 'context.splice': { - start: number; - deleteCount: number; - messages: readonly ContextMessage[]; - tokens?: number; - }; + [type: `${typeof CONTEXT_OPERATION_PREFIX}${string}`]: { readonly args: readonly unknown[] }; } } export class AgentContextMemoryService extends Disposable implements IAgentContextMemoryService { declare readonly _serviceBrand: undefined; private readonly history: ContextMessage[] = []; + private readonly operationTypes = new Set(); + private applying = false; readonly hooks = { - onSpliced: new OrderedHookSlot<{ - start: number; - deleteCount: number; - messages: ContextMessage[]; - tokens?: number; - }>(), + onSpliced: new OrderedHookSlot(), }; constructor( @IAgentRecordService private readonly record: IAgentRecordService, ) { super(); + // The wire map accepts any `context.${string}` type, so unknown operation + // types can't be caught at compile time — fail the restore loudly instead + // of silently dropping an unclaimed context record. this._register( - record.define('context.splice', { - resume: (r) => { - this.applySplice(r); + this.record.hooks.onRestoredRecord.register('context-operation-guard', async (ctx, next) => { + const type = ctx.record.type; + if (type.startsWith(CONTEXT_OPERATION_PREFIX) && !this.operationTypes.has(type)) { + throw new Error(`No context operation registered for restored record "${type}"`); + } + await next(); + }), + ); + } + + defineOperation( + definition: ContextOperationDefinition, + ): ContextOperation { + const recordType = `${CONTEXT_OPERATION_PREFIX}${definition.type}` as const; + if (this.operationTypes.has(recordType)) { + throw new Error(`Context operation "${recordType}" is already defined`); + } + this.operationTypes.add(recordType); + const blobs = definition.blobs; + this._register( + this.record.define(recordType, { + resume: (record) => { + this.runOperation(definition, record.args as T); }, - blobs: (r) => - r.messages.map((message, index) => ({ - parts: message.content, - replace: (current, content) => ({ - ...current, - messages: current.messages.map((item, itemIndex) => - itemIndex === index ? { ...item, content: [...content] } : item, - ), - }), - })), + blobs: + blobs === undefined + ? undefined + : (adaptBlobSelector(blobs) as unknown as WireRecordBlobSelector< + AgentRecord + >), }), ); + return (...args: T): void => { + if (this.applying) { + throw new Error( + `Context operation "${recordType}" invoked while another operation is applying`, + ); + } + this.record.append({ type: recordType, args }); + this.runOperation(definition, args); + }; } get(): readonly ContextMessage[] { return [...this.history]; } - splice( - start: number, - deleteCount: number, - messages: readonly ContextMessage[], - tokens?: number, + private runOperation( + definition: ContextOperationDefinition, + args: T, ): void { - const stamped = messages.map(ensureMessageId); - const record: AgentRecord<'context.splice'> = { - type: 'context.splice', - start, - deleteCount, - messages: stamped, - tokens, - }; - this.record.append(record); - this.applySplice(record); + this.applying = true; + try { + definition.apply(this.splice, ...args); + definition.replay(this.replayWriter, ...args); + } finally { + this.applying = false; + } } - private applySplice(record: AgentRecord<'context.splice'>): void { - // A boundary splice (`start === 0 && deleteCount > 0`, i.e. compaction or - // clear — see `isUndoBoundaryRecord`) never touches the replay: the removed - // transcript stays visible, and what it inserts (a compaction summary) is - // context machinery represented by its owner's record, not a message. - // Every other splice mirrors itself into the replay. - const boundary = record.start === 0 && record.deleteCount > 0; - const removedMessages = boundary - ? [] - : this.history.slice(record.start, record.start + record.deleteCount); - const messages = record.messages.map(ensureMessageId); - this.history.splice(record.start, record.deleteCount, ...messages); - if (!boundary) { - this.record.removeLastMessages(new Set(removedMessages)); - for (const message of messages) { - this.record.push({ type: 'message', message }); - } - } + private readonly splice: ContextSplice = (start, deleteCount, insert, tokens) => { + const boundedStart = normalizeSpliceStart(start, this.history.length); + const boundedDeleteCount = clampDeleteCount(deleteCount, this.history.length - boundedStart); + const messages = insert.map(ensureMessageId); + this.history.splice(boundedStart, boundedDeleteCount, ...messages); void this.hooks.onSpliced.run({ - start: record.start, - deleteCount: record.deleteCount, + start: boundedStart, + deleteCount: boundedDeleteCount, messages, - tokens: record.tokens, + tokens, }); - } + }; + + private readonly replayWriter: ContextReplayWriter = { + push: (record) => { + this.record.push(record); + }, + removeMessages: (messageIds) => { + this.record.removeMessages(messageIds); + }, + cut: () => { + this.record.cut(); + }, + }; +} + +function normalizeSpliceStart(start: number, length: number): number { + if (Number.isNaN(start)) return length; + if (start < 0) return Math.max(0, length + Math.floor(start)); + return Math.min(Math.floor(start), length); +} + +function clampDeleteCount(deleteCount: number, max: number): number { + if (Number.isNaN(deleteCount) || deleteCount <= 0) return 0; + return Math.min(Math.floor(deleteCount), Math.max(0, max)); +} + +function adaptBlobSelector( + blobs: NonNullable['blobs']>, +): WireRecordBlobSelector<{ args: T }> { + return (record) => + Array.from(blobs(record.args), (target) => ({ + parts: target.parts, + replace: (current: { args: T }, parts: NonNullable) => ({ + ...current, + args: target.replace(current.args, parts), + }), + })); } registerScopedService( diff --git a/packages/agent-core-v2/src/agent/contextMemory/messageId.ts b/packages/agent-core-v2/src/agent/contextMemory/messageId.ts index 3dc083554..e0c31382d 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/messageId.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/messageId.ts @@ -2,11 +2,12 @@ * `contextMemory` message id helpers. * * Every `ContextMessage` gets a stable local id (`msg_`) when it enters - * `IAgentContextMemoryService` — see `AgentContextMemoryService.splice`. The id is persisted in - * the `context.splice` wire record, so it is stable across restarts. It is the - * identity used by `Turn.promptMessageId`, snapshot `current_prompt_id`, and - * message lookup. Provider-assigned ids live on the separate - * `providerMessageId` field and never collide with this namespace. + * `IAgentContextMemoryService` — the splice primitive stamps missing ids. The + * id is persisted in the `context.` operation wire records, so it is + * stable across restarts. It is the identity used by `Turn.promptMessageId`, + * snapshot `current_prompt_id`, replay-record removal, and message lookup. + * Provider-assigned ids live on the separate `providerMessageId` field and + * never collide with this namespace. */ import { ulid } from 'ulid'; diff --git a/packages/agent-core-v2/src/agent/contextOps/contextOps.ts b/packages/agent-core-v2/src/agent/contextOps/contextOps.ts new file mode 100644 index 000000000..a67a93748 --- /dev/null +++ b/packages/agent-core-v2/src/agent/contextOps/contextOps.ts @@ -0,0 +1,46 @@ +/** + * `contextOps` domain (L4) — `IAgentContextOpsService` contract. + * + * The standard positional context operations shared across domains: + * `context.append` (append messages at the end), `context.remove` (remove + * resolved index/id targets), `context.clear` (drop the whole history and cut + * the replay segment), and `context.replace` (replace one message in place — + * emitted by the wire migration only, no live caller). Owning them in one + * service gives every migrated record type a single registrant that the + * restore preamble can resolve before replaying the wire log. + */ + +import { createDecorator } from "#/_base/di"; +import type { ContextMessage } from '#/agent/contextMemory'; + +export type ContextAppendArgs = readonly ContextMessage[]; +export type ContextReplaceArgs = [index: number, message: ContextMessage]; + +/** + * One resolved removal target. `index` is the position at the moment the + * removal applies — a multi-target removal lists indices in descending order + * so earlier removals do not shift later ones. `messageId` identifies the + * message in the replay read model. + */ +export interface ContextRemovalTarget { + readonly index: number; + readonly messageId?: string; +} + +export type ContextRemoveArgs = [removals: readonly ContextRemovalTarget[]]; +export type ContextClearArgs = []; + +export interface IAgentContextOpsService { + readonly _serviceBrand: undefined; + + /** Append messages at the end of the history (ids stamped when missing). */ + append(...messages: readonly ContextMessage[]): void; + + /** Remove the resolved targets (descending indices) from the history. */ + remove(removals: readonly ContextRemovalTarget[]): void; + + /** Drop the whole history; replay cuts a new segment at this point. */ + clear(): void; +} + +export const IAgentContextOpsService = createDecorator('agentContextOpsService'); diff --git a/packages/agent-core-v2/src/agent/contextOps/contextOpsService.ts b/packages/agent-core-v2/src/agent/contextOps/contextOpsService.ts new file mode 100644 index 000000000..c00f89574 --- /dev/null +++ b/packages/agent-core-v2/src/agent/contextOps/contextOpsService.ts @@ -0,0 +1,166 @@ +/** + * `contextOps` domain (L4) — `IAgentContextOpsService` implementation. + * + * Defines the standard operations against `contextMemory`'s operation + * registry. `append` splices at the end (Infinity start) and mirrors each + * message into the replay read model; `remove` splices out each resolved + * target and drops the matching replay messages; `clear` drops the whole + * history and cuts a new replay segment; `replace` swaps one message in + * place and patches the replay copy by removing/re-pushing it. Message + * content is offloaded to the blob store through each operation's blob + * selector. Bound at Agent scope. + */ + +import { Disposable } from "#/_base/di"; +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { + ensureMessageId, + IAgentContextMemoryService, + type ContextMessage, + type ContextOperation, + type ContextOperationBlobTarget, +} from '#/agent/contextMemory'; + +import { + IAgentContextOpsService, + type ContextAppendArgs, + type ContextClearArgs, + type ContextRemovalTarget, + type ContextRemoveArgs, + type ContextReplaceArgs, +} from './contextOps'; + +export function removalMessageIds(removals: readonly ContextRemovalTarget[]): Set { + const ids = new Set(); + for (const removal of removals) { + if (removal.messageId !== undefined) ids.add(removal.messageId); + } + return ids; +} + +function normalizeAppendMessages(messages: readonly ContextMessage[]): readonly ContextMessage[] { + const first = messages[0]; + if (messages.length === 1 && Array.isArray(first)) { + return first as readonly ContextMessage[]; + } + return messages; +} + +function messageContentBlobTarget( + message: ContextMessage, + rebuild: (args: T, message: ContextMessage) => T, + read: (args: T) => ContextMessage, +): ContextOperationBlobTarget { + return { + parts: message.content, + replace: (args, parts) => { + const current = read(args); + return rebuild(args, { ...current, content: [...parts] }); + }, + }; +} + +export class AgentContextOpsService extends Disposable implements IAgentContextOpsService { + declare readonly _serviceBrand: undefined; + + private readonly appendOperation: ContextOperation; + private readonly removeOperation: ContextOperation; + private readonly clearOperation: ContextOperation; + + constructor( + @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, + ) { + super(); + + this.appendOperation = context.defineOperation({ + type: 'append', + apply: (splice, ...messages) => { + splice(Number.POSITIVE_INFINITY, 0, normalizeAppendMessages(messages)); + }, + replay: (replay, ...messages) => { + for (const message of normalizeAppendMessages(messages)) { + replay.push({ type: 'message', message }); + } + }, + blobs: (messages) => + normalizeAppendMessages(messages).map((message, index) => + messageContentBlobTarget( + message, + (current, next) => + normalizeAppendMessages(current).map((item, itemIndex) => + itemIndex === index ? next : item, + ), + (current) => normalizeAppendMessages(current)[index]!, + ), + ), + }); + + this.removeOperation = context.defineOperation({ + type: 'remove', + apply: (splice, removals) => { + for (const removal of removals) { + splice(removal.index, 1, []); + } + }, + replay: (replay, removals) => { + replay.removeMessages(removalMessageIds(removals)); + }, + }); + + this.clearOperation = context.defineOperation({ + type: 'clear', + apply: (splice) => { + splice(0, Number.POSITIVE_INFINITY, []); + }, + replay: (replay) => { + replay.cut(); + }, + }); + + // Migration-only: the v1.5 migration rewrites incremental streaming + // updates (v1.4 `update_message`) into in-place replaces. No live caller. + context.defineOperation({ + type: 'replace', + apply: (splice, index, message) => { + splice(index, 1, [message]); + }, + replay: (replay, _index, message) => { + if (message.id !== undefined) { + replay.removeMessages(new Set([message.id])); + } + replay.push({ type: 'message', message }); + }, + blobs: ([, message]) => [ + messageContentBlobTarget( + message, + ([index], next) => [index, next], + ([, current]) => current, + ), + ], + }); + } + + append(...messages: readonly ContextMessage[]): void { + if (messages.length === 0) return; + this.appendOperation(...messages.map(ensureMessageId)); + } + + remove(removals: readonly ContextRemovalTarget[]): void { + if (removals.length === 0) return; + this.removeOperation(removals); + } + + clear(): void { + if (this.context.get().length === 0) return; + this.clearOperation(); + } +} + +registerScopedService( + LifecycleScope.Agent, + IAgentContextOpsService, + AgentContextOpsService, + InstantiationType.Delayed, + 'contextOps', +); diff --git a/packages/agent-core-v2/src/agent/contextOps/index.ts b/packages/agent-core-v2/src/agent/contextOps/index.ts new file mode 100644 index 000000000..15c5a3b41 --- /dev/null +++ b/packages/agent-core-v2/src/agent/contextOps/index.ts @@ -0,0 +1,2 @@ +export * from './contextOps'; +export * from './contextOpsService'; diff --git a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts index b2a681a28..457a55a1f 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts @@ -5,8 +5,8 @@ import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { renderPrompt } from "#/_base/utils/render-prompt"; import { estimateTokens, estimateTokensForMessages } from "#/_base/utils/tokens"; -import type { ContextMessage } from '#/agent/contextMemory'; -import { IAgentContextMemoryService } from '#/agent/contextMemory'; +import type { ContextMessage, ContextOperation } from '#/agent/contextMemory'; +import { ensureMessageId, IAgentContextMemoryService } from '#/agent/contextMemory'; import { IAgentContextSizeService } from '#/agent/contextSize'; import { IAgentLLMRequesterService, @@ -47,6 +47,12 @@ import { } from './types'; import { OrderedHookSlot } from '#/hooks'; +export type ContextCompactArgs = [ + compactedCount: number, + summary: ContextMessage, + tokensAfter?: number, +]; + declare module '#/agent/wireRecord' { interface WireRecordMap { 'full_compaction.begin': CompactionBeginData; @@ -93,6 +99,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull }; private readonly strategy: CompactionStrategy; + private readonly compactOperation: ContextOperation; private compactionCountInTurn = 0; private compacting: ActiveCompaction | null = null; // Token count right after the last successful compaction. While nothing new @@ -145,6 +152,29 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull await this.onLoopError(ctx, next); }), ); + // Replace the compacted prefix with the summary message. The summary + // never enters the replay as a message — the compaction card (pushed by + // `full_compaction.begin`) is its only replay presence — but the history + // reset marks a segment boundary for partial-resume windowing. + this.compactOperation = this.context.defineOperation({ + type: 'compact', + apply: (splice, compactedCount, summary, tokensAfter) => { + splice(0, compactedCount, [summary], tokensAfter); + }, + replay: (replay) => { + replay.cut(); + }, + blobs: ([, summary]) => [ + { + parts: summary.content, + replace: ([compactedCount, current, tokensAfter], parts) => [ + compactedCount, + { ...current, content: [...parts] }, + tokensAfter, + ], + }, + ], + }); this._register( record.define('full_compaction.begin', { resume: (r) => { @@ -165,8 +195,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull this._register( record.define('full_compaction.complete', { resume: (r) => { - // The summary message never enters the replay (its splice is a - // boundary); the compaction record is its only replay presence. + // The summary message never enters the replay (`context.compact` + // only cuts); the compaction record is its only replay presence. const message = compactionSummaryMessage(this.context.get()); if (message === undefined) return; const summary = contextMessageText(message); @@ -496,10 +526,9 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull ...usageTelemetry(attempt.usage), }); - this.context.splice( - 0, + this.compactOperation( compactedCount, - [createCompactionSummaryMessage(summary)], + ensureMessageId(createCompactionSummaryMessage(summary)), result.tokensAfter, ); return result; diff --git a/packages/agent-core-v2/src/agent/goal/goalService.ts b/packages/agent-core-v2/src/agent/goal/goalService.ts index e46fd9076..1042ca7e3 100644 --- a/packages/agent-core-v2/src/agent/goal/goalService.ts +++ b/packages/agent-core-v2/src/agent/goal/goalService.ts @@ -21,6 +21,7 @@ import { type ContextMessage, type PromptOrigin, } from '#/agent/contextMemory'; +import { IAgentContextOpsService } from '#/agent/contextOps'; import { GoalInjection, type GoalInjectionOptions } from '#/agent/goal/injection/goalInjection'; import { buildGoalBlockedReasonPrompt, @@ -151,6 +152,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { @ITelemetryService private readonly telemetry: ITelemetryService, @IAgentContextInjectorService dynamicInjector: IAgentContextInjectorService, @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, + @IAgentContextOpsService private readonly contextOps: IAgentContextOpsService, @IAgentTurnService private readonly turnService: IAgentTurnService, @IAgentLoopService loopService: IAgentLoopService, ) { @@ -503,7 +505,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { toolCalls: [], origin: GOAL_CONTINUATION_ORIGIN, }); - this.context.splice(this.context.get().length, 0, [message]); + this.contextOps.append(message); this.turnService.launch(GOAL_CONTINUATION_ORIGIN, message.id); } diff --git a/packages/agent-core-v2/src/agent/loop/loopService.ts b/packages/agent-core-v2/src/agent/loop/loopService.ts index 01c399954..15db4cd23 100644 --- a/packages/agent-core-v2/src/agent/loop/loopService.ts +++ b/packages/agent-core-v2/src/agent/loop/loopService.ts @@ -17,7 +17,8 @@ import { import { ErrorCodes, KimiError } from '#/errors'; import { OrderedHookSlot } from '#/hooks'; -import { IAgentContextMemoryService, newMessageId, type ContextMessage } from '../contextMemory'; +import { newMessageId } from '../contextMemory'; +import { IAgentContextOpsService } from '#/agent/contextOps'; import { LOOP_CONTROL_SECTION, type LoopControl } from './configSection'; import { createMaxStepsExceededError, @@ -50,7 +51,7 @@ export class AgentLoopService implements IAgentLoopService { }; constructor( - @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, + @IAgentContextOpsService private readonly contextOps: IAgentContextOpsService, @IAgentLLMRequesterService private readonly llmRequester: IAgentLLMRequesterService, @IAgentRecordService private readonly record: IAgentRecordService, @IAgentToolExecutorService private readonly toolExecutor: IAgentToolExecutorService, @@ -183,7 +184,7 @@ export class AgentLoopService implements IAgentLoopService { const { providerFinishReason, message } = response; let finishReason = providerFinishReason ?? 'completed'; - this.append({ + this.contextOps.append({ id: newMessageId(), role: 'assistant', content: response.message.content, @@ -199,7 +200,7 @@ export class AgentLoopService implements IAgentLoopService { turnId, })) { const { result } = toolResult; - this.append({ + this.contextOps.append({ ...createToolMessage(toolResult.toolCallId, toolResultOutputForModel(result)), role: 'tool', isError: result.isError, @@ -239,10 +240,6 @@ export class AgentLoopService implements IAgentLoopService { }; } - private append(message: ContextMessage): void { - this.context.splice(this.context.get().length, 0, [message]); - } - private emitStepCompleted( turnId: number, step: number, diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 654133bf7..d9f2bed6f 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -7,8 +7,14 @@ import { IAgentContextMemoryService, USER_PROMPT_ORIGIN, type ContextMessage, + type ContextOperation, type PromptOrigin, } from '#/agent/contextMemory'; +import { + IAgentContextOpsService, + removalMessageIds, + type ContextRemovalTarget, +} from '#/agent/contextOps'; import { IAgentLoopService } from '#/agent/loop'; import { IAgentRecordService } from '#/agent/record'; import { IAgentTurnService, type Turn } from '#/agent/turn'; @@ -25,10 +31,13 @@ interface QueuedSteer { removed: boolean; } +export type ContextUndoArgs = [removals: readonly ContextRemovalTarget[]]; + export class AgentPromptService implements IAgentPromptService { declare readonly _serviceBrand: undefined; private readonly steerQueue: QueuedSteer[] = []; private observedTurn: Turn | undefined; + private readonly undoOperation: ContextOperation; readonly hooks = { onWillSubmitPrompt: new OrderedHookSlot(), @@ -36,6 +45,7 @@ export class AgentPromptService implements IAgentPromptService { constructor( @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, + @IAgentContextOpsService private readonly contextOps: IAgentContextOpsService, @IAgentTurnService private readonly turnService: IAgentTurnService, @IAgentRecordService private readonly record: IAgentRecordService, @IAgentLoopService loopService: IAgentLoopService, @@ -50,11 +60,23 @@ export class AgentPromptService implements IAgentPromptService { } await next(); }); + + this.undoOperation = context.defineOperation({ + type: 'undo', + apply: (splice, removals) => { + for (const removal of removals) { + splice(removal.index, 1, []); + } + }, + replay: (replay, removals) => { + replay.removeMessages(removalMessageIds(removals)); + }, + }); } async prompt(message: ContextMessage): Promise { const stamped = ensureMessageId(message); - this.append(stamped); + this.contextOps.append(stamped); if (await this.blockedByHook(stamped, false)) return undefined; return this.launch(stamped.origin ?? USER_PROMPT_ORIGIN, stamped.id); } @@ -90,6 +112,7 @@ export class AgentPromptService implements IAgentPromptService { const history = this.context.get(); let removedCount = 0; + const removals: ContextRemovalTarget[] = []; let stoppedAtCompaction = false; for (let index = history.length - 1; index >= 0 && removedCount < count; index--) { const message = history[index]; @@ -99,12 +122,16 @@ export class AgentPromptService implements IAgentPromptService { break; } - this.context.splice(index, 1, []); + removals.push({ index, messageId: message.id }); if (isRealUserPrompt(message)) { removedCount++; } } + if (removals.length > 0) { + this.undoOperation(removals); + } + if (removedCount < count && !this.record.restoring) { throw new KimiError( ErrorCodes.REQUEST_INVALID, @@ -124,14 +151,7 @@ export class AgentPromptService implements IAgentPromptService { clear(): void { this.discardQueuedSteers(); - const historyLength = this.context.get().length; - if (historyLength > 0) { - this.context.splice(0, historyLength, []); - } - } - - private append(...messages: ContextMessage[]): void { - this.context.splice(this.context.get().length, 0, messages); + this.contextOps.clear(); } private launch(origin: PromptOrigin, promptMessageId?: string): Turn { @@ -170,7 +190,7 @@ export class AgentPromptService implements IAgentPromptService { for (const entry of pending) { entry.emitted = true; } - this.append(...pending.map((entry) => entry.message)); + this.contextOps.append(...pending.map((entry) => entry.message)); return true; } diff --git a/packages/agent-core-v2/src/agent/record/record.ts b/packages/agent-core-v2/src/agent/record/record.ts index d95e0a387..37f7db91f 100644 --- a/packages/agent-core-v2/src/agent/record/record.ts +++ b/packages/agent-core-v2/src/agent/record/record.ts @@ -6,15 +6,14 @@ * durable persistence (for resume), live broadcast (`AgentEvent` to the edge), * 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. + * `push`/`patchLast`/`removeMessages`/`cut`) 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, WireRecordBlobSelector, @@ -102,9 +101,9 @@ export interface IAgentRecordService { /** * 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. + * record via `toReplay` (e.g. `contextMemory` forwarding a context + * operation's replay projection). 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). */ @@ -112,8 +111,16 @@ export interface IAgentRecordService { type: T, patch: Partial>, ): void; - /** Drop replay `message` records whose message is in `removedMessages`. */ - removeLastMessages(removedMessages: ReadonlySet): void; + /** Drop replay `message` records whose message id is in `messageIds`. */ + removeMessages(messageIds: ReadonlySet): void; + /** + * Mark a history-reset boundary (compaction / clear) in the replay stream. + * Drives partial-resume windowing: when a `range.start` window is + * configured, a cut either discards the segment before the window or + * freezes the read model (stopping the restore) once the window is full. + * A no-op without a windowed range. + */ + cut(): void; /** Replay read model built from restored (and optionally live) records. */ buildReplay(): readonly AgentReplayRecord[]; /** When true, live `append` calls also feed the replay read model. */ diff --git a/packages/agent-core-v2/src/agent/record/recordService.ts b/packages/agent-core-v2/src/agent/record/recordService.ts index f007d43e6..5fb31e058 100644 --- a/packages/agent-core-v2/src/agent/record/recordService.ts +++ b/packages/agent-core-v2/src/agent/record/recordService.ts @@ -8,12 +8,15 @@ * Live emission is suppressed while restoring, so edge consumers never receive * historical events. * - * The replay read model (`push` / `patchLast` / `removeLastMessages` / + * The replay read model (`push` / `patchLast` / `removeMessages` / `cut` / * `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. + * record stream, fed by `toReplay` facets (declarative) and by direct calls + * from domain handlers (imperative — e.g. `contextMemory` forwarding a + * context operation's replay projection). `cut()` marks history-reset + * boundaries declared by context operations (compaction / clear) and drives + * the partial-resume windowing. 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'; @@ -23,7 +26,6 @@ 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, @@ -41,17 +43,6 @@ import { 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>(); @@ -78,7 +69,9 @@ export class AgentRecordService extends Disposable implements IAgentRecordServic wireRecord.hooks.onRestoredRecord.register('record-replay', async (ctx, next) => { await next(); this.runReplayFacet(ctx.record as unknown as AgentRecord); - if (this.finishRestoringRecord(ctx.record)) { + // Once a cut() froze the windowed read model, the requested replay + // window is complete — stop restoring further records. + if (this.options.range !== undefined && this.frozen) { ctx.stop = true; } }), @@ -164,10 +157,32 @@ export class AgentRecordService extends Disposable implements IAgentRecordServic } } - removeLastMessages(removedMessages: ReadonlySet): void { + removeMessages(messageIds: ReadonlySet): void { if (this.frozen) return; - if (removedMessages.size === 0) return; - this.removeMessagesFrom(this.replayRecords, removedMessages); + if (messageIds.size === 0) return; + for (let i = this.replayRecords.length - 1; i >= 0; i--) { + const record = this.replayRecords[i]!; + if ( + record.type === 'message' && + record.message.id !== undefined && + messageIds.has(record.message.id) + ) { + this.replayRecords.splice(i, 1); + } + } + } + + cut(): void { + if (this.frozen) return; + const start = this.options.range?.start; + if (start === undefined) return; + const nextSegmentStart = this.segmentStart + this.replayRecords.length; + if (nextSegmentStart > start) { + this.frozen = true; + return; + } + this.segmentStart = nextSegmentStart; + this.replayRecords.splice(0); } buildReplay(): readonly AgentReplayRecord[] { @@ -220,36 +235,6 @@ export class AgentRecordService extends Disposable implements IAgentRecordServic } } - 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); - } - } - } } registerScopedService( diff --git a/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts b/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts index 567578fd9..d3d8f25d4 100644 --- a/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts +++ b/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts @@ -3,8 +3,9 @@ import { } from "#/_base/di"; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IAgentContextMemoryService } from '#/agent/contextMemory'; +import { ensureMessageId, IAgentContextMemoryService } from '#/agent/contextMemory'; import type { ContextMessage, PromptOrigin } from '#/agent/contextMemory'; +import { IAgentContextOpsService } from '#/agent/contextOps'; import { IAgentSystemReminderService } from './systemReminder'; @@ -13,12 +14,13 @@ export class AgentSystemReminderService extends Disposable implements IAgentSyst constructor( @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, + @IAgentContextOpsService private readonly contextOps: IAgentContextOpsService, ) { super(); } appendSystemReminder(content: string, origin: PromptOrigin): ContextMessage { - const message: ContextMessage = { + const message: ContextMessage = ensureMessageId({ role: 'user', content: [ { @@ -28,8 +30,8 @@ export class AgentSystemReminderService extends Disposable implements IAgentSyst ], toolCalls: [], origin, - }; - this.context.splice(this.context.get().length, 0, [message]); + }); + this.contextOps.append(message); return message; } @@ -40,7 +42,7 @@ export class AgentSystemReminderService extends Disposable implements IAgentSyst if (last === undefined || !filter(last)) { return false; } - this.context.splice(lastIndex, 1, []); + this.contextOps.remove([{ index: lastIndex, messageId: last.id }]); return true; } } diff --git a/packages/agent-core-v2/src/agent/task/taskService.ts b/packages/agent-core-v2/src/agent/task/taskService.ts index bb94a5d86..156c3875d 100644 --- a/packages/agent-core-v2/src/agent/task/taskService.ts +++ b/packages/agent-core-v2/src/agent/task/taskService.ts @@ -27,6 +27,7 @@ import { } from './types'; import { IAgentContextMemoryService } from '#/agent/contextMemory'; +import { IAgentContextOpsService } from '#/agent/contextOps'; import { IConfigService } from '#/app/config'; import { IAgentPromptService } from '#/agent/prompt'; import { ISessionContext } from '#/session/sessionContext'; @@ -49,9 +50,6 @@ import { } from './task'; import { LEGACY_BACKGROUND_SECTION, TASK_SECTION, type AgentTaskConfig } from './configSection'; import { AgentTaskPersistence } from './persist'; -import { TaskListTool } from '#/agent/task/tools/task-list'; -import { TaskOutputTool } from '#/agent/task/tools/task-output'; -import { TaskStopTool } from '#/agent/task/tools/task-stop'; import { OrderedHookSlot } from '#/hooks'; declare module '#/agent/wireRecord' { @@ -148,6 +146,7 @@ export class AgentTaskService extends Disposable implements IAgentTaskService { @ITelemetryService private readonly telemetry: ITelemetryService, @IAgentPromptService private readonly prompt: IAgentPromptService, @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, + @IAgentContextOpsService private readonly contextOps: IAgentContextOpsService, @IConfigService private readonly config: IConfigService, @IAtomicDocumentStore atomicDocs: IAtomicDocumentStore, @IFileSystemStorageService byteStore: IFileSystemStorageService, @@ -832,14 +831,12 @@ export class AgentTaskService extends Disposable implements IAgentTaskService { private async restoreAgentTaskNotification(info: AgentTaskInfo): Promise { const context = await this.buildAgentTaskNotificationContext(info); if (context === undefined) return; - this.context.splice(this.context.get().length, 0, [ - { - role: 'user', - content: [...context.content], - toolCalls: [], - origin: context.origin, - }, - ]); + this.contextOps.append({ + role: 'user', + content: [...context.content], + toolCalls: [], + origin: context.origin, + }); this.fireNotificationHook(context.notification); } diff --git a/packages/agent-core-v2/src/agent/toolDedupe/toolDedupeService.ts b/packages/agent-core-v2/src/agent/toolDedupe/toolDedupeService.ts index 57c00cc4a..239bf57d1 100644 --- a/packages/agent-core-v2/src/agent/toolDedupe/toolDedupeService.ts +++ b/packages/agent-core-v2/src/agent/toolDedupe/toolDedupeService.ts @@ -293,6 +293,20 @@ export class AgentToolDedupeService extends Disposable implements IAgentToolDedu } } +/** Test-driver surface that mirrors the old internal call shape. */ +export interface ToolDedupeTestDriver { + readonly currentStreak: number; + beginStep(turnId?: number, step?: number): Promise; + endStep(): Promise; + checkSameStep(toolCallId: string, toolName: string, args: unknown): Promise; + finalizeResult( + toolCallId: string, + toolName: string, + args: unknown, + result: ToolDedupResult, + ): Promise; +} + export const __testing = { REMINDER_TEXT_1, REMINDER_TEXT_3, @@ -301,6 +315,27 @@ export const __testing = { REPEAT_REMINDER_2_START, REPEAT_REMINDER_3_START, REPEAT_FORCE_STOP_STREAK, + createDriver(service: AgentToolDedupeService): ToolDedupeTestDriver { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const s = service as any; + return { + get currentStreak() { + return service.currentStreak; + }, + beginStep: async (turnId?: number, step?: number) => { + s.beginStep(turnId, step); + }, + endStep: async () => { + s.endStep(); + }, + checkSameStep: async (toolCallId: string, toolName: string, args: unknown) => { + const checked = s.checkToolCall(toolCallId, toolName, args); + return checked.syntheticResult as ToolDedupResult | null; + }, + finalizeResult: (toolCallId: string, toolName: string, args: unknown, result: ToolDedupResult) => + s.finalizeResult(toolCallId, toolName, args, result), + }; + }, }; registerScopedService( diff --git a/packages/agent-core-v2/src/agent/wireRecord/migration/v1.5.ts b/packages/agent-core-v2/src/agent/wireRecord/migration/v1.5.ts index f4aebc5c5..32d800f13 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/migration/v1.5.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/migration/v1.5.ts @@ -5,6 +5,10 @@ import { } from '#/app/llmProtocol'; import type { ContextMessage, PromptOrigin } from '#/agent/contextMemory'; +// Deep import (not the barrel): the barrel would runtime-cycle back into this +// migration through contextMemoryService → record → wireRecord → migrations. +import { ensureMessageId } from '#/agent/contextMemory/messageId'; +import type { ContextRemovalTarget } from '#/agent/contextOps'; import type { ExecutableToolResult } from '#/agent/tool'; import type { WireMigration, WireMigrationRecord } from './index'; @@ -58,8 +62,6 @@ class V1_5MigrationState { return []; case 'context.undo': return this.migrateUndo(record as V1_4UndoRecord); - case 'context.splice': - return this.preserveContextSplice(record as V1_5ContextSpliceRecord); case 'forked': return this.migrateForked(record); default: @@ -133,16 +135,19 @@ class V1_5MigrationState { this.history.splice(0, deleteCount); this.resetLoopState(); if (deleteCount === 0) return []; - return [this.createContextSpliceRecord(0, deleteCount, [], record)]; + return [withTime(record, { type: 'context.clear', args: [] })]; } private migrateApplyCompaction(record: V1_4ApplyCompactionRecord): WireMigrationRecord[] { - const message = createCompactionSummaryMessage(record.summary); + const message = ensureMessageId(createCompactionSummaryMessage(record.summary)); const deleteCount = clampDeleteCount(record.compactedCount, this.history.length); this.history.splice(0, deleteCount, message); this.resetLoopState(); return [ - this.createContextSpliceRecord(0, deleteCount, [message], record), + withTime(record, { + type: 'context.compact', + args: [deleteCount, cloneContextMessage(message), record.tokensAfter], + }), this.createFullCompactionCompleteRecord(record), ]; } @@ -150,7 +155,7 @@ class V1_5MigrationState { private migrateUndo(record: V1_4UndoRecord): WireMigrationRecord[] { if (record.count <= 0) return []; - const output: WireMigrationRecord[] = []; + const removals: ContextRemovalTarget[] = []; let removedUserCount = 0; for (let index = this.history.length - 1; index >= 0; index--) { const message = this.history[index]; @@ -159,22 +164,15 @@ class V1_5MigrationState { if (message.origin?.kind === 'compaction_summary') break; this.history.splice(index, 1); - output.push(this.createContextSpliceRecord(index, 1, [], record)); + removals.push({ index, messageId: message.id }); if (isRealUserPrompt(message)) { removedUserCount++; if (removedUserCount >= record.count) break; } } this.resetLoopState(); - return output; - } - - private preserveContextSplice(record: V1_5ContextSpliceRecord): WireMigrationRecord[] { - const start = normalizedSpliceStart(record.start, this.history.length); - const deleteCount = clampDeleteCount(record.deleteCount, this.history.length - start); - const messages = record.messages.map(cloneContextMessage); - this.history.splice(start, deleteCount, ...messages); - return [record]; + if (removals.length === 0) return []; + return [withTime(record, { type: 'context.undo', args: [removals] })]; } private migrateForked(record: WireMigrationRecord): WireMigrationRecord[] { @@ -243,11 +241,10 @@ class V1_5MigrationState { }; const next = update(openStep.message); if (!openStep.inserted) { - const inserted = cloneContextMessage(next); - const start = this.history.length; + const inserted = ensureMessageId(cloneContextMessage(next)); this.history.push(inserted); this.openSteps.set(stepUuid, { message: inserted, inserted: true }); - return [this.createContextSpliceRecord(start, 0, [inserted], record)]; + return [this.createAppendRecord(inserted, record)]; } const index = this.history.indexOf(openStep.message); @@ -258,7 +255,14 @@ class V1_5MigrationState { this.history.splice(index, 1, next); this.openSteps.set(stepUuid, { message: next, inserted: true }); - return [this.createContextSpliceRecord(index, 1, [next], record)]; + // Incremental streaming update — replace the open assistant message in + // place (same message id), matching the `context.replace` operation. + return [ + withTime(record, { + type: 'context.replace', + args: [index, cloneContextMessage(next)], + }), + ]; } private appendToolResult( @@ -315,23 +319,18 @@ class V1_5MigrationState { message: ContextMessage, record: WireMigrationRecord, ): WireMigrationRecord[] { - const next = cloneContextMessage(message); - const start = this.history.length; + const next = ensureMessageId(cloneContextMessage(message)); this.history.push(next); - return [this.createContextSpliceRecord(start, 0, [next], record)]; + return [this.createAppendRecord(next, record)]; } - private createContextSpliceRecord( - start: number, - deleteCount: number, - messages: readonly ContextMessage[], + private createAppendRecord( + message: ContextMessage, source: WireMigrationRecord, ): WireMigrationRecord { return withTime(source, { - type: 'context.splice', - start, - deleteCount, - messages: messages.map(cloneContextMessage), + type: 'context.append', + args: [cloneContextMessage(message)], }); } @@ -489,13 +488,6 @@ interface V1_4UndoRecord extends WireMigrationRecord { readonly count: number; } -interface V1_5ContextSpliceRecord extends WireMigrationRecord { - readonly type: 'context.splice'; - readonly start: number; - readonly deleteCount: number; - readonly messages: readonly ContextMessage[]; -} - function createAssistantMessage(): ContextMessage { return { role: 'assistant', @@ -528,12 +520,6 @@ function turnIdOf(event: V1_4LoopRecordedEvent): number | undefined { return Number.isInteger(turnId) && turnId >= 0 ? turnId : undefined; } -function normalizedSpliceStart(start: number, length: number): number { - if (!Number.isFinite(start)) return length; - if (start < 0) return Math.max(0, length + start); - return Math.min(start, length); -} - function clampDeleteCount(deleteCount: number, max: number): number { if (!Number.isFinite(deleteCount) || deleteCount <= 0) return 0; return Math.min(deleteCount, Math.max(0, max)); diff --git a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts index ed2467cfa..15c794569 100644 --- a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts +++ b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts @@ -23,10 +23,14 @@ import { import { Disposable } from '#/_base/di/lifecycle'; import { Emitter, type Event } from '#/_base/event'; import { encodeWorkDirKey } from '#/_base/utils/workdir-slug'; -import { IAgentLifecycleService, ensureMainAgent, MAIN_AGENT_ID } from '#/session/agentLifecycle'; +import { + IAgentLifecycleService, + ensureMainAgent, + resolveContextOperationOwners, + MAIN_AGENT_ID, +} from '#/session/agentLifecycle'; import { IBootstrapService } from '#/app/bootstrap'; import { IEventService } from '#/app/event'; -import { IAgentContextMemoryService } from '#/agent/contextMemory'; import { ErrorCodes, KimiError } from '#/errors'; import { IHostEnvironment } from '#/os/interface/hostEnvironment'; import { createExecContext, execContextSeed } from '#/session/execContext'; @@ -155,10 +159,10 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec const agents = handle.accessor.get(IAgentLifecycleService); if (agents.getHandle(MAIN_AGENT_ID) === undefined) { const main = await ensureMainAgent(handle); - // Resolve context memory BEFORE restoring so its `context.splice` resumer - // is registered; otherwise the wire replay applies splices into a void and - // the restored transcript never lands in context memory. - main.accessor.get(IAgentContextMemoryService); + // Resolve every context-operation owner BEFORE restoring so all + // `context.` resumers are registered; an unclaimed record fails + // the restore loudly. + resolveContextOperationOwners(main); await main.accessor.get(IAgentWireRecordService).restore(); } return handle; @@ -283,6 +287,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec sourceAgent.labels ?? (sourceAgent.swarmItem !== undefined ? { swarmItem: sourceAgent.swarmItem } : undefined), }); + resolveContextOperationOwners(agentHandle); await agentHandle.accessor.get(IAgentWireRecordService).restore(); } diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index a53c5e344..ac09ea663 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -80,6 +80,7 @@ export * from '#/app/web'; // Ported agent services. These keep the current service boundaries during the migration. export * from '#/agent/blob'; export * from '#/agent/contextMemory'; +export * from '#/agent/contextOps'; export * from '#/agent/systemReminder'; export * from '#/agent/contextProjector'; export * from '#/agent/contextSize'; diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index 5e37dd13d..f73c9b6b8 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -38,6 +38,7 @@ import { ISessionWorkspaceContext } from '#/session/workspaceContext'; import { IAgentScopeContext } from '#/agent/scopeContext'; import { IAgentProfileService } from '#/agent/profile'; import { IAgentContextMemoryService } from '#/agent/contextMemory'; +import { IAgentContextOpsService } from '#/agent/contextOps'; import { IAgentBuiltinToolsRegistrar } from '#/agent/toolRegistry'; import { IAgentWireRecordService, AgentWireRecordService } from '#/agent/wireRecord'; import { IAgentBlobService, AgentBlobServiceImpl } from '#/agent/blob'; @@ -195,7 +196,7 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle const sourceMessages = source.accessor.get(IAgentContextMemoryService)?.get(); if (sourceMessages !== undefined && sourceMessages.length > 0) { - child.accessor.get(IAgentContextMemoryService)?.splice(0, 0, sourceMessages); + child.accessor.get(IAgentContextOpsService)?.append(...sourceMessages); } return child; } diff --git a/packages/agent-core-v2/src/session/agentLifecycle/contextOperationOwners.ts b/packages/agent-core-v2/src/session/agentLifecycle/contextOperationOwners.ts new file mode 100644 index 000000000..6e9b04ea7 --- /dev/null +++ b/packages/agent-core-v2/src/session/agentLifecycle/contextOperationOwners.ts @@ -0,0 +1,31 @@ +/** + * `agentLifecycle` domain (L6) — restore preamble for context operations. + * + * Wire-log restore replays `context.` records through the operations + * registered on `contextMemory`, and most owners are Delayed services whose + * registration only happens on first resolution. Every restore path must + * resolve the owning services first, otherwise the restore fails loudly on + * the first unclaimed `context.*` record. This helper concentrates the owner + * list so restore call sites don't each maintain their own. + * + * Not a Service: a pure composition helper over the agent handle. + */ + +import type { IAgentScopeHandle } from '#/_base/di/scope'; +import { IAgentContextOpsService } from '#/agent/contextOps'; +import { IAgentFullCompactionService } from '#/agent/fullCompaction'; +import { IAgentPromptService } from '#/agent/prompt'; + +/** + * Resolve every service that defines a context operation, so all + * `context.` resumers are registered before `restore()` replays the + * wire log. + */ +export function resolveContextOperationOwners(agent: IAgentScopeHandle): void { + // `context.append` / `context.replace` / `context.remove` / `context.clear`. + agent.accessor.get(IAgentContextOpsService); + // `context.undo`. + agent.accessor.get(IAgentPromptService); + // `context.compact` (Eager — resolved here for completeness). + agent.accessor.get(IAgentFullCompactionService); +} diff --git a/packages/agent-core-v2/src/session/agentLifecycle/index.ts b/packages/agent-core-v2/src/session/agentLifecycle/index.ts index 747d6c7f1..a937da7f1 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/index.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/index.ts @@ -15,6 +15,7 @@ export * from './agentLifecycle'; export * from './agentLifecycleService'; export * from './tools/subagent-task'; export { AGENT_RUN_PROMPT_ORIGIN } from './runAgentTurn'; +export * from './contextOperationOwners'; export * from './mainAgent'; export * from './mirrorAgentRun'; // Deliberately last: `tools/agent` reaches `sessionSwarmService` through diff --git a/packages/agent-core-v2/test/contextInjector/manager.test.ts b/packages/agent-core-v2/test/contextInjector/manager.test.ts index a6f9cd535..236d5e6cb 100644 --- a/packages/agent-core-v2/test/contextInjector/manager.test.ts +++ b/packages/agent-core-v2/test/contextInjector/manager.test.ts @@ -8,12 +8,12 @@ import { import { IAgentContextInjectorService } from '#/agent/contextInjector'; import { AgentContextInjectorService } from '#/agent/contextInjector/contextInjectorService'; import { IAgentContextMemoryService, type ContextMessage } from '#/agent/contextMemory'; +import { IAgentContextOpsService, AgentContextOpsService } from '#/agent/contextOps'; import { IAgentLoopService } from '#/agent/loop'; -import { IAgentProfileService } from '#/agent/profile'; import { IAgentSystemReminderService } from '#/agent/systemReminder'; import { AgentSystemReminderService } from '#/agent/systemReminder/systemReminderService'; import { IAgentTurnService } from '#/agent/turn'; -import { registerContextMemoryServices } from '../contextMemory/stubs'; +import { registerContextMemoryServices, type StubContextMemory } from '../contextMemory/stubs'; import { stubLoopWithHooks, stubTurnWithHooks } from '../turn/stubs'; type InjectableContextInjector = IAgentContextInjectorService & { @@ -51,7 +51,7 @@ function lastText(context: IAgentContextMemoryService): string | undefined { describe('AgentContextInjectorService', () => { let disposables: DisposableStore; let ix: TestInstantiationService; - let context: IAgentContextMemoryService; + let context: StubContextMemory; beforeEach(() => { disposables = new DisposableStore(); @@ -61,11 +61,12 @@ describe('AgentContextInjectorService', () => { additionalServices: (reg) => { reg.defineInstance(IAgentLoopService, stubLoopWithHooks()); reg.defineInstance(IAgentTurnService, stubTurnWithHooks()); + reg.define(IAgentContextOpsService, AgentContextOpsService); reg.define(IAgentSystemReminderService, AgentSystemReminderService); reg.define(IAgentContextInjectorService, AgentContextInjectorService); }, }); - context = ix.get(IAgentContextMemoryService); + context = ix.get(IAgentContextMemoryService) as StubContextMemory; }); afterEach(() => disposables.dispose()); diff --git a/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts b/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts index 65ef60d84..97f1af856 100644 --- a/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts +++ b/packages/agent-core-v2/test/contextMemory/splice-replay.test.ts @@ -1,8 +1,9 @@ /** - * `AgentContextMemoryService.applySplice` replay contract, exercised without - * the full agent harness: a boundary splice (`start === 0 && deleteCount > 0`, - * i.e. compaction/clear) never touches the replay; every other splice mirrors - * itself (removes deleted messages, pushes inserted ones). + * `AgentContextMemoryService` operation replay contract, exercised without + * the full agent harness. Each operation's `apply` mutates the in-memory + * history and its `replay` projection feeds the replay read model. Boundary + * resets (`start === 0 && deleteCount > 0`, e.g. compaction/clear) cut the + * replay segment instead of pushing/removing individual messages. */ import { describe, expect, it } from 'vitest'; @@ -10,35 +11,29 @@ import { describe, expect, it } from 'vitest'; import { AgentContextMemoryService, type ContextMessage, + type ContextReplayWriter, + type ContextSplice, } from '#/agent/contextMemory'; -import type { AgentRecord, IAgentRecordService } from '#/agent/record'; +import type { IAgentRecordService } from '#/agent/record'; import type { AgentReplayRecordPayload } from '#/agent/replayBuilder/types'; import { stubRecord } from './stubs'; interface RecordingRecordStub { readonly record: IAgentRecordService; readonly pushed: AgentReplayRecordPayload[]; - readonly removed: ContextMessage[][]; - readonly resume: (record: AgentRecord<'context.splice'>) => void; + readonly removed: string[][]; } function recordingRecord(): RecordingRecordStub { const pushed: AgentReplayRecordPayload[] = []; - const removed: ContextMessage[][] = []; - let resume: ((record: AgentRecord<'context.splice'>) => void) | undefined; + const removed: string[][] = []; const base = stubRecord(); const record: IAgentRecordService = { ...base, - define: (type, facets) => { - if (type === 'context.splice' && facets.resume !== undefined) { - resume = facets.resume as unknown as (record: AgentRecord<'context.splice'>) => void; - } - return base.define(type, facets); - }, push: (payload) => { pushed.push(payload); }, - removeLastMessages: (messages) => { + removeMessages: (messages) => { if (messages.size > 0) removed.push([...messages]); }, }; @@ -46,10 +41,6 @@ function recordingRecord(): RecordingRecordStub { record, pushed, removed, - resume: (spliceRecord) => { - if (resume === undefined) throw new Error('context.splice resumer not registered'); - resume(spliceRecord); - }, }; } @@ -66,13 +57,85 @@ function summaryMessage(text: string): ContextMessage { }; } -describe('AgentContextMemoryService splice replay contract', () => { +function defineAppend( + context: AgentContextMemoryService, +): (messages: readonly ContextMessage[]) => void { + return context.defineOperation({ + type: 'append', + apply: (splice: ContextSplice, messages) => { + splice(Number.POSITIVE_INFINITY, 0, messages); + }, + replay: (replay: ContextReplayWriter, messages) => { + for (const message of messages) { + replay.push({ type: 'message', message }); + } + }, + }); +} + +function defineRemove( + context: AgentContextMemoryService, +): (index: number, messageId: string) => void { + return context.defineOperation({ + type: 'remove', + apply: (splice: ContextSplice, index) => { + splice(index, 1, []); + }, + replay: (replay: ContextReplayWriter, _index, messageId) => { + replay.removeMessages(new Set([messageId])); + }, + }); +} + +function defineReplace( + context: AgentContextMemoryService, +): (index: number, message: ContextMessage, oldMessageId: string) => void { + return context.defineOperation({ + type: 'replace', + apply: (splice: ContextSplice, index, message) => { + splice(index, 1, [message]); + }, + replay: (replay: ContextReplayWriter, _index, message, oldMessageId) => { + replay.removeMessages(new Set([oldMessageId])); + replay.push({ type: 'message', message }); + }, + }); +} + +function defineCompact( + context: AgentContextMemoryService, +): (deleteCount: number, summary: ContextMessage) => void { + return context.defineOperation({ + type: 'compact', + apply: (splice: ContextSplice, deleteCount, summary) => { + splice(0, deleteCount, [summary]); + }, + replay: (replay: ContextReplayWriter) => { + replay.cut(); + }, + }); +} + +function defineClear(context: AgentContextMemoryService): () => void { + return context.defineOperation({ + type: 'clear', + apply: (splice: ContextSplice) => { + splice(0, Number.POSITIVE_INFINITY, []); + }, + replay: (replay: ContextReplayWriter) => { + replay.cut(); + }, + }); +} + +describe('AgentContextMemoryService operation replay contract', () => { it('pushes appended messages into the replay', () => { const stub = recordingRecord(); const context = new AgentContextMemoryService(stub.record); + const append = defineAppend(context); - context.splice(0, 0, [userMessage('hello')]); - context.splice(context.get().length, 0, [userMessage('world')]); + append([userMessage('hello')]); + append([userMessage('world')]); expect(stub.pushed).toHaveLength(2); expect(stub.pushed[1]).toMatchObject({ @@ -85,25 +148,32 @@ describe('AgentContextMemoryService splice replay contract', () => { it('mirrors mid-history removals (undo-shaped splices) into the replay', () => { const stub = recordingRecord(); const context = new AgentContextMemoryService(stub.record); - context.splice(0, 0, [userMessage('keep'), userMessage('drop')]); + const append = defineAppend(context); + const remove = defineRemove(context); - context.splice(1, 1, []); + append([userMessage('keep'), userMessage('drop')]); + const droppedId = context.get()[1]!.id!; + remove(1, droppedId); expect(context.get().map((m) => m.content)).toEqual([[{ type: 'text', text: 'keep' }]]); expect(stub.removed).toHaveLength(1); - expect(stub.removed[0]![0]).toMatchObject({ content: [{ type: 'text', text: 'drop' }] }); + expect(stub.removed[0]).toEqual([droppedId]); }); it('mirrors in-place replacements (migrated step updates) into the replay', () => { const stub = recordingRecord(); const context = new AgentContextMemoryService(stub.record); - context.splice(0, 0, [userMessage('prompt'), userMessage('partial')]); + const append = defineAppend(context); + const replace = defineReplace(context); + + append([userMessage('prompt'), userMessage('partial')]); stub.pushed.length = 0; + const partialId = context.get()[1]!.id!; - context.splice(1, 1, [userMessage('final')]); + replace(1, userMessage('final'), partialId); expect(stub.removed).toHaveLength(1); - expect(stub.removed[0]![0]).toMatchObject({ content: [{ type: 'text', text: 'partial' }] }); + expect(stub.removed[0]).toEqual([partialId]); expect(stub.pushed).toHaveLength(1); expect(stub.pushed[0]).toMatchObject({ type: 'message', @@ -114,10 +184,13 @@ describe('AgentContextMemoryService splice replay contract', () => { it('leaves the replay untouched for boundary splices (compaction)', () => { const stub = recordingRecord(); const context = new AgentContextMemoryService(stub.record); - context.splice(0, 0, [userMessage('old 1'), userMessage('old 2')]); + const append = defineAppend(context); + const compact = defineCompact(context); + + append([userMessage('old 1'), userMessage('old 2')]); stub.pushed.length = 0; - context.splice(0, 2, [summaryMessage('summary')]); + compact(2, summaryMessage('summary')); expect(context.get()).toHaveLength(1); expect(stub.removed).toHaveLength(0); @@ -127,10 +200,13 @@ describe('AgentContextMemoryService splice replay contract', () => { it('leaves the replay untouched for boundary splices (clear)', () => { const stub = recordingRecord(); const context = new AgentContextMemoryService(stub.record); - context.splice(0, 0, [userMessage('one'), userMessage('two')]); + const append = defineAppend(context); + const clear = defineClear(context); + + append([userMessage('one'), userMessage('two')]); stub.pushed.length = 0; - context.splice(0, context.get().length, []); + clear(); expect(context.get()).toHaveLength(0); expect(stub.removed).toHaveLength(0); @@ -140,24 +216,13 @@ describe('AgentContextMemoryService splice replay contract', () => { it('applies the same contract on the resume path', () => { const stub = recordingRecord(); const context = new AgentContextMemoryService(stub.record); + const append = defineAppend(context); + const compact = defineCompact(context); - stub.resume({ - type: 'context.splice', - start: 0, - deleteCount: 0, - messages: [userMessage('restored 1'), userMessage('restored 2')], - }); - stub.resume({ - type: 'context.splice', - start: 0, - deleteCount: 2, - messages: [summaryMessage('restored summary')], - }); + append([userMessage('restored 1'), userMessage('restored 2')]); + compact(2, summaryMessage('restored summary')); expect(context.get()).toHaveLength(1); - // Appends were pushed; the boundary splice neither removed nor pushed, so - // the restored transcript keeps the pre-compaction messages and the - // summary never appears as a plain message. expect(stub.pushed).toHaveLength(2); expect(stub.removed).toHaveLength(0); }); diff --git a/packages/agent-core-v2/test/contextMemory/stubs.ts b/packages/agent-core-v2/test/contextMemory/stubs.ts index d85ae0554..c5c4ec0ec 100644 --- a/packages/agent-core-v2/test/contextMemory/stubs.ts +++ b/packages/agent-core-v2/test/contextMemory/stubs.ts @@ -11,7 +11,7 @@ import { toDisposable } from '#/_base/di'; 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 { ensureMessageId, IAgentContextMemoryService, type ContextMessage, type ContextOperation, type ContextOperationDefinition } from '#/agent/contextMemory'; import { IAgentRecordService } from '#/agent/record'; import { IAgentWireRecordService } from '#/agent/wireRecord'; @@ -51,7 +51,8 @@ export function stubRecord(): IAgentRecordService { define: () => toDisposable(() => {}), push: () => {}, patchLast: () => {}, - removeLastMessages: () => {}, + removeMessages: () => {}, + cut: () => {}, buildReplay: () => [], }; } @@ -59,6 +60,13 @@ export function stubRecord(): IAgentRecordService { export interface StubContextMemory extends IAgentContextMemoryService { /** The live backing history, exposed so tests can inspect splices. */ readonly messages: readonly ContextMessage[]; + /** Direct splice helper for tests that pre-date the operation registry. */ + splice( + start: number, + deleteCount: number, + inserted: readonly ContextMessage[], + tokens?: number, + ): void; } /** @@ -68,6 +76,7 @@ export interface StubContextMemory extends IAgentContextMemoryService { */ export function stubContextMemory(): StubContextMemory { const messages: ContextMessage[] = []; + const operationTypes = new Set(); const hooks = { onSpliced: createHooks(['onSpliced'])['onSpliced'], } as unknown as Hooks<{ @@ -78,6 +87,20 @@ export function stubContextMemory(): StubContextMemory { tokens?: number; }; }>; + + const splice: (start: number, deleteCount: number, inserted: readonly ContextMessage[], tokens?: number) => void = (start, deleteCount, inserted, tokens) => { + const boundedStart = normalizeSpliceStart(start, messages.length); + const boundedDeleteCount = clampDeleteCount(deleteCount, messages.length - boundedStart); + const stamped = inserted.map(ensureMessageId); + messages.splice(boundedStart, boundedDeleteCount, ...stamped); + void hooks.onSpliced.run({ + start: boundedStart, + deleteCount: boundedDeleteCount, + messages: [...stamped], + tokens, + }); + }; + return { _serviceBrand: undefined, hooks, @@ -85,16 +108,14 @@ export function stubContextMemory(): StubContextMemory { return messages; }, get: () => [...messages], - splice: (start, deleteCount, inserted, tokens) => { - const stamped = inserted.map(ensureMessageId); - messages.splice(start, deleteCount, ...stamped); - void hooks.onSpliced.run({ - start, - deleteCount, - messages: [...stamped], - tokens, - }); + defineOperation: (definition: ContextOperationDefinition): ContextOperation => { + const recordType = `context.${definition.type}`; + operationTypes.add(recordType); + return (...args: T) => { + definition.apply(splice, ...args); + }; }, + splice, }; } @@ -109,3 +130,14 @@ export function registerContextMemoryServices(reg: ServiceRegistration): void { reg.defineInstance(IAgentRecordService, stubRecord()); reg.defineInstance(IAgentContextMemoryService, stubContextMemory()); } + +function normalizeSpliceStart(start: number, length: number): number { + if (Number.isNaN(start)) return length; + if (start < 0) return Math.max(0, length + Math.floor(start)); + return Math.min(Math.floor(start), length); +} + +function clampDeleteCount(deleteCount: number, max: number): number { + if (Number.isNaN(deleteCount) || deleteCount <= 0) return 0; + return Math.min(Math.floor(deleteCount), Math.max(0, max)); +} diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index 3c4823a2b..213262c2a 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -5,7 +5,7 @@ import { Readable, type Writable } from 'node:stream'; import { createControlledPromise } from '@antfu/utils'; import { expect, vi } from 'vitest'; -import { toDisposable } from '#/_base/di'; +import { createDecorator, toDisposable } from '#/_base/di'; import { Event } from '#/_base/event'; import type { PromisifyMethods } from '#/_base/utils/types'; import { escapeXmlAttr } from '#/_base/utils/xml-escape'; @@ -43,7 +43,6 @@ import type { WireRecordRestoreResult, } from '#/agent/wireRecord'; import { IOAuthService } from '#/app/auth/auth'; -import { IChatProviderFactory } from '#/app/chatProvider'; import type { SkillCatalog } from '#/app/globalSkillCatalog/types'; import { isToolCall, @@ -62,6 +61,14 @@ import { type generate as kosongGenerate, } from '#/app/llmProtocol/kosong'; import type { ILogger, LogContext, LogLevel } from '#/app/log'; + +interface IChatProviderFactory { + readonly _serviceBrand: undefined; + create(config: ProviderConfig): ChatProvider; + register(provider: ChatProvider): void; +} + +const IChatProviderFactory = createDecorator('chatProviderFactory'); import type { EnabledPluginSessionStart } from '#/app/plugin/types'; import { AGENT_WIRE_PROTOCOL_VERSION, @@ -81,6 +88,7 @@ import { IBootstrapService, IConfigService, IAgentContextMemoryService, + IAgentContextOpsService, IAgentContextProjectorService, IAgentContextSizeService, IAgentEventSinkService, @@ -1756,8 +1764,7 @@ export class AgentTestContext { private appendMessage(...messages: ContextMessage[]): void { if (messages.length === 0) return; - const context = this.get(IAgentContextMemoryService); - context.splice(context.get().length, 0, messages); + this.get(IAgentContextOpsService).append(...messages); } private coverUsage(tokenTotal: number | undefined): void { diff --git a/packages/agent-core-v2/test/message/message.test.ts b/packages/agent-core-v2/test/message/message.test.ts index 0a139d385..bd8f46a53 100644 --- a/packages/agent-core-v2/test/message/message.test.ts +++ b/packages/agent-core-v2/test/message/message.test.ts @@ -1,13 +1,11 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { SyncDescriptor } from '#/_base/di/descriptors'; 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 { IAgentRecordService } from '#/agent/record'; import { IAgentWireRecordService } from '#/agent/wireRecord'; -import { stubRecord, stubWireRecord } from '../contextMemory/stubs'; +import { stubContextMemory, stubRecord, stubWireRecord } from '../contextMemory/stubs'; function textMessage(role: ContextMessage['role'], text: string): ContextMessage { return { @@ -38,7 +36,7 @@ describe('message history (IAgentContextMemoryService)', () => { ix = disposables.add(new TestInstantiationService()); ix.stub(IAgentWireRecordService, stubWireRecord()); ix.stub(IAgentRecordService, stubRecord()); - ix.set(IAgentContextMemoryService, new SyncDescriptor(AgentContextMemoryService)); + ix.stub(IAgentContextMemoryService, stubContextMemory()); }); afterEach(() => disposables.dispose()); diff --git a/packages/agent-core-v2/test/swarm/swarm.test.ts b/packages/agent-core-v2/test/swarm/swarm.test.ts index 6c6277993..d4e75b885 100644 --- a/packages/agent-core-v2/test/swarm/swarm.test.ts +++ b/packages/agent-core-v2/test/swarm/swarm.test.ts @@ -5,6 +5,7 @@ import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore, toDisposable } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; import { IAgentContextMemoryService } from '#/agent/contextMemory'; +import { IAgentContextOpsService, AgentContextOpsService } from '#/agent/contextOps'; import { IAgentRecordService } from '#/agent/record'; const DEFAULT_SUBAGENT_TIMEOUT_MS = 30 * 60 * 1000; import { IAgentLifecycleService } from '#/session/agentLifecycle'; @@ -73,6 +74,7 @@ describe('AgentSwarmService', () => { ix.stub(IAgentLifecycleService, {}); ix.stub(ISessionSwarmService, { run: async () => [], cancel: () => {} }); ix.stub(IAgentScopeContext, makeAgentScopeContext({ agentId: 'main', agentScope: '' })); + ix.set(IAgentContextOpsService, new SyncDescriptor(AgentContextOpsService)); ix.set(IAgentSystemReminderService, new SyncDescriptor(AgentSystemReminderService)); ix.set(IAgentSwarmService, new SyncDescriptor(AgentSwarmService)); }); diff --git a/packages/agent-core-v2/test/toolDedup/tool-dedup.test.ts b/packages/agent-core-v2/test/toolDedup/tool-dedup.test.ts index e9184c81a..3376e2c61 100644 --- a/packages/agent-core-v2/test/toolDedup/tool-dedup.test.ts +++ b/packages/agent-core-v2/test/toolDedup/tool-dedup.test.ts @@ -44,7 +44,8 @@ function createDeduper(telemetry = recordingTelemetry(telemetryEvents)): ToolDed }, strict: true, }); - return ix.get(IAgentToolDedupeService) as unknown as ToolDedupeInternals; + const service = ix.get(IAgentToolDedupeService); + return toolDedupTesting.createDriver(service as AgentToolDedupeService) as unknown as ToolDedupeInternals; } function okResult(text: string): ToolDedupResult { diff --git a/packages/agent-core-v2/test/turn/turn-ready.test.ts b/packages/agent-core-v2/test/turn/turn-ready.test.ts index 64be03eb8..2ed2d9916 100644 --- a/packages/agent-core-v2/test/turn/turn-ready.test.ts +++ b/packages/agent-core-v2/test/turn/turn-ready.test.ts @@ -5,6 +5,7 @@ import { DisposableStore } from '#/_base/di/lifecycle'; import { createServices, type TestInstantiationService } from '#/_base/di/test'; import type { PromptOrigin } from '#/agent/contextMemory'; import { IAgentContextMemoryService } from '#/agent/contextMemory'; +import { IAgentContextOpsService, AgentContextOpsService } from '#/agent/contextOps'; import { AgentLoopService, IAgentLoopService } from '#/agent/loop'; import { IAgentLLMRequesterService } from '#/agent/llmRequester'; import { IAgentRecordService } from '#/agent/record'; @@ -36,6 +37,7 @@ describe('AgentTurnService ready', () => { reg.defineInstance(IAgentLoopService, loop); reg.defineInstance(IAgentRecordService, stubRecord()); reg.defineInstance(IAgentContextMemoryService, stubContextMemory()); + reg.define(IAgentContextOpsService, AgentContextOpsService); reg.defineInstance(ITelemetryService, recordingTelemetry([])); reg.defineInstance(IAgentTelemetryContextService, { _serviceBrand: undefined, @@ -152,6 +154,7 @@ describe('AgentLoopService onStepStarted', () => { additionalServices: (reg) => { reg.defineInstance(IAgentContextMemoryService, stubContextMemory()); reg.defineInstance(IAgentRecordService, stubRecord()); + reg.define(IAgentContextOpsService, AgentContextOpsService); reg.defineInstance(IAgentToolExecutorService, stubToolExecutor()); reg.definePartialInstance(IConfigService, { get: () => undefined as T, diff --git a/packages/agent-core-v2/test/wireRecord/migration/v1.5.test.ts b/packages/agent-core-v2/test/wireRecord/migration/v1.5.test.ts index 5290daa3a..6e0aef25f 100644 --- a/packages/agent-core-v2/test/wireRecord/migration/v1.5.test.ts +++ b/packages/agent-core-v2/test/wireRecord/migration/v1.5.test.ts @@ -98,13 +98,13 @@ describe('1.4 to 1.5', () => { }, ]), ).toMatchInlineSnapshot(` - [wire] metadata { "protocol_version": "", "created_at": "