From 2ce7de5ccae7a162e5b2bb200a3e95822e2ddec1 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Mon, 13 Jul 2026 13:26:49 +0800 Subject: [PATCH 1/4] refactor(v2): move op payload declarations to PersistedOpMap Colocate persisted/transient op payload shapes with their op definitions via interface-merge into #/wire/types (PersistedOpMap / TransientOpMap), replacing the per-service WireRecordMap augmentations. defineOp apply handlers now infer payloads from the map, and taskService restores delivered-notification marks by iterating getRecords() on restore instead of a wireRecord restore hook. --- .../src/agent/contextMemory/contextOps.ts | 23 +++++++----- .../src/agent/fullCompaction/compactionOps.ts | 24 +++++++----- .../fullCompaction/fullCompactionService.ts | 14 ------- .../agent-core-v2/src/agent/goal/goalOps.ts | 37 ++++++++++++------- .../src/agent/goal/goalService.ts | 28 ++------------ .../src/agent/llmRequester/llmRequestOps.ts | 34 ++++++++--------- .../agent-core-v2/src/agent/loop/turnOps.ts | 28 +++++++------- .../src/agent/mcp/mcpDiscoveryOps.ts | 14 +++---- .../src/agent/profile/profileOps.ts | 8 +++- .../src/agent/profile/profileService.ts | 8 ---- .../agent-core-v2/src/agent/swarm/swarmOps.ts | 6 +++ .../src/agent/task/taskService.ts | 27 +++++++------- .../agent-core-v2/src/session/cron/cronOps.ts | 32 ++++++++++------ .../session/cron/sessionCronServiceImpl.ts | 15 -------- .../src/session/todo/sessionTodoService.ts | 9 ----- .../agent-core-v2/src/session/todo/todoOps.ts | 9 ++++- .../agent-core-v2/src/wire/wireServiceImpl.ts | 4 +- 17 files changed, 150 insertions(+), 170 deletions(-) diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts index d2a5d5ff9c..69ed2c55d7 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts @@ -124,23 +124,28 @@ export interface ContextMessagePayload { readonly message: ContextMessage; } -export const contextAppendMessage = defineOp(ContextModel, 'context.append_message', { - apply: (state, p: ContextMessagePayload): ContextMessage[] => - foldAppendMessage(state, p.message) as ContextMessage[], -}); - export interface ContextLoopEventPayload { readonly event: LoopRecordedEvent; } +declare module '#/wire/types' { + interface PersistedOpMap { + 'context.append_message': ContextMessagePayload; + 'context.append_loop_event': ContextLoopEventPayload; + } +} + +export const contextAppendMessage = defineOp(ContextModel, 'context.append_message', { + apply: (state, p): ContextMessage[] => + foldAppendMessage(state, p.message) as ContextMessage[], +}); + export const contextAppendLoopEvent = defineOp(ContextModel, 'context.append_loop_event', { - apply: (state, p: ContextLoopEventPayload): ContextMessage[] => - foldLoopEvent(state, p.event) as ContextMessage[], + apply: (state, p) => foldLoopEvent(state, p.event) as ContextMessage[], }); export const contextClear = defineOp(ContextModel, 'context.clear', { - apply: (state): ContextMessage[] => - state.length === 0 ? state : (resetFold([]) as ContextMessage[]), + apply: (state) => state.length === 0 ? state : (resetFold([]) as ContextMessage[]), }); interface ContextCompactionBasePayload { diff --git a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts index a4c7ad1afa..9e9c31b434 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts @@ -30,8 +30,8 @@ * `begin` Op's `toEvent`; the rest directly) and also emit live through * `wire.signal` (legacy channel, until Phase 3); they are declared here via * interface-merge (`error` is already declared by `mcp`, so it is not - * re-declared). The `full_compaction.*` record shapes stay declared in - * `WireRecordMap` (see `fullCompactionService.ts`) because the records still + * re-declared). The `full_compaction.*` record shapes are registered in + * `PersistedOpMap` (`#/wire/types`, below) because the records still * ride the per-agent `wire.jsonl` log read by `wireRecord.restore()` / * `getRecords()`. Consumed by the Agent-scope `fullCompactionService`. */ @@ -68,9 +68,18 @@ declare module '#/app/event/eventBus' { export type FullCompactionBeginPayload = CompactionBeginData; +export type FullCompactionCompletePayload = Record; + +declare module '#/wire/types' { + interface PersistedOpMap { + 'full_compaction.begin': FullCompactionBeginPayload; + 'full_compaction.cancel': {}; + 'full_compaction.complete': FullCompactionCompletePayload; + } +} + export const fullCompactionBegin = defineOp(CompactionModel, 'full_compaction.begin', { - apply: (s, _p: FullCompactionBeginPayload): CompactionState => - s.phase === 'running' ? s : { phase: 'running' }, + apply: (s, _p) => (s.phase === 'running' ? s : { phase: 'running' }), toEvent: (p) => ({ type: 'compaction.started' as const, trigger: p.source, @@ -79,12 +88,9 @@ export const fullCompactionBegin = defineOp(CompactionModel, 'full_compaction.be }); export const fullCompactionCancel = defineOp(CompactionModel, 'full_compaction.cancel', { - apply: (s): CompactionState => (s.phase === 'idle' ? s : { phase: 'idle' }), + apply: (s) => (s.phase === 'idle' ? s : { phase: 'idle' }), }); -export type FullCompactionCompletePayload = Record; - export const fullCompactionComplete = defineOp(CompactionModel, 'full_compaction.complete', { - apply: (s, _p: FullCompactionCompletePayload): CompactionState => - s.phase === 'idle' ? s : { phase: 'idle' }, + apply: (s, _p) => (s.phase === 'idle' ? s : { phase: 'idle' }), }); diff --git a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts index 0ae77f7c17..d1b55d83af 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts @@ -56,7 +56,6 @@ import { fullCompactionBegin, fullCompactionCancel, fullCompactionComplete, - type FullCompactionCompletePayload, } from './compactionOps'; import { type CompactionBeginData, @@ -65,19 +64,6 @@ import { import { Emitter, type Event } from '#/_base/event'; import { OrderedHookSlot } from '#/hooks'; -// The `full_compaction.*` record shapes stay declared in `WireRecordMap` -// because the records still ride the per-agent `wire.jsonl` log read by -// `wireRecord.restore()` / `getRecords()`. fullCompaction itself no longer -// registers resumers here — its state rebuilds from the same log via -// `wire.replay` into `CompactionModel`. -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { - 'full_compaction.begin': CompactionBeginData; - 'full_compaction.cancel': {}; - 'full_compaction.complete': FullCompactionCompletePayload; - } -} - export const MAX_COMPACTION_RETRY_ATTEMPTS = 5; const DEFAULT_COMPACTION_MAX_COMPLETION_TOKENS = 128 * 1024; const OVERFLOW_CONTEXT_SAFETY_RATIO = 0.85; diff --git a/packages/agent-core-v2/src/agent/goal/goalOps.ts b/packages/agent-core-v2/src/agent/goal/goalOps.ts index 65ccec79a5..0dd013cec0 100644 --- a/packages/agent-core-v2/src/agent/goal/goalOps.ts +++ b/packages/agent-core-v2/src/agent/goal/goalOps.ts @@ -61,8 +61,27 @@ export interface GoalCreatePayload { readonly completionCriterion?: string; } +export interface GoalUpdatePayload { + readonly status?: GoalStatus; + readonly reason?: string; + readonly turnsUsed?: number; + readonly tokensUsed?: number; + readonly wallClockMs?: number; + readonly budgetLimits?: GoalBudgetLimits; + readonly actor?: GoalActor; +} + +declare module '#/wire/types' { + interface PersistedOpMap { + 'goal.create': GoalCreatePayload; + 'goal.update': GoalUpdatePayload; + 'goal.clear': {}; + forked: {}; + } +} + export const createGoal = defineOp(GoalModel, 'goal.create', { - apply: (_s, p: GoalCreatePayload): GoalModelState => ({ + apply: (_s, p) => ({ goalId: p.goalId, objective: p.objective, completionCriterion: p.completionCriterion, @@ -74,18 +93,8 @@ export const createGoal = defineOp(GoalModel, 'goal.create', { }), }); -export interface GoalUpdatePayload { - readonly status?: GoalStatus; - readonly reason?: string; - readonly turnsUsed?: number; - readonly tokensUsed?: number; - readonly wallClockMs?: number; - readonly budgetLimits?: GoalBudgetLimits; - readonly actor?: GoalActor; -} - export const updateGoal = defineOp(GoalModel, 'goal.update', { - apply: (s, p: GoalUpdatePayload): GoalModelState => { + apply: (s, p) => { if (s === null) return null; let next: GoalState | undefined; if (p.status !== undefined && p.status !== s.status) { @@ -112,9 +121,9 @@ export const updateGoal = defineOp(GoalModel, 'goal.update', { }); export const clearGoal = defineOp(GoalModel, 'goal.clear', { - apply: (): GoalModelState => null, + apply: () => null, }); export const forkGoal = defineOp(GoalModel, 'forked', { - apply: (): GoalModelState => null, + apply: () => null, }); diff --git a/packages/agent-core-v2/src/agent/goal/goalService.ts b/packages/agent-core-v2/src/agent/goal/goalService.ts index b01aeeb551..21522a40b0 100644 --- a/packages/agent-core-v2/src/agent/goal/goalService.ts +++ b/packages/agent-core-v2/src/agent/goal/goalService.ts @@ -9,9 +9,10 @@ * lives in the Model (set from each Op payload, never by `Date.now()` inside * `apply`); the `wallClockResumedAt` cursor is a live-only field, reset on * replay and (re)started on the live path. A `forked` wire Op clears the Model - * at a fork boundary; the `goal.*` record shapes stay declared in - * `WireRecordMap` because they still ride the shared wire log read by - * `getRecords()` and replayed into the Model. Injects reminders through + * at a fork boundary; the `goal.*` payload shapes are registered in + * `PersistedOpMap` (`#/wire/types`) inside `goalOps` because they still ride + * the shared wire log read by `getRecords()` and replayed into the Model. + * Injects reminders through * `contextInjector`, drives continuation turns by enqueueing `newTurn` * `StepRequest`s onto `loop` (the continuation message materializes when the * loop pops it), accounts live @@ -69,27 +70,6 @@ import type { GoalToolResult, } from './types'; -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { - forked: {}; - 'goal.create': { - goalId: string; - objective: string; - completionCriterion?: string; - }; - 'goal.update': { - status?: GoalStatus; - reason?: string; - turnsUsed?: number; - tokensUsed?: number; - wallClockMs?: number; - budgetLimits?: GoalBudgetLimits; - actor?: GoalActor; - }; - 'goal.clear': {}; - } -} - const MAX_GOAL_OBJECTIVE_LENGTH = 4000; // The criterion is repeated in every goal reminder, so it is truncated instead diff --git a/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts b/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts index ac7be0bf73..7fe126860f 100644 --- a/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts +++ b/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts @@ -30,17 +30,6 @@ export interface LlmToolsSnapshotPayload { readonly tools: readonly LlmRequestToolSchema[]; } -export const llmToolsSnapshot = defineOp( - LlmRequestTraceModel, - 'llm.tools_snapshot', - { - apply: (s, p: LlmToolsSnapshotPayload): LlmRequestTraceState => { - if (s.seenToolsHashes.includes(p.hash)) return s; - return { seenToolsHashes: [...s.seenToolsHashes, p.hash] }; - }, - }, -); - export interface LlmRequestPayload { readonly kind: 'loop' | 'compaction'; readonly provider: string; @@ -64,13 +53,24 @@ export interface LlmRequestPayload { readonly droppedCount?: number; } -export const llmRequest = defineOp(LlmRequestTraceModel, 'llm.request', { - apply: (s, _p: LlmRequestPayload): LlmRequestTraceState => s, -}); - -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { +declare module '#/wire/types' { + interface PersistedOpMap { 'llm.tools_snapshot': LlmToolsSnapshotPayload; 'llm.request': LlmRequestPayload; } } + +export const llmToolsSnapshot = defineOp( + LlmRequestTraceModel, + 'llm.tools_snapshot', + { + apply: (s, p) => { + if (s.seenToolsHashes.includes(p.hash)) return s; + return { seenToolsHashes: [...s.seenToolsHashes, p.hash] }; + }, + }, +); + +export const llmRequest = defineOp(LlmRequestTraceModel, 'llm.request', { + apply: (s, _p) => s, +}); diff --git a/packages/agent-core-v2/src/agent/loop/turnOps.ts b/packages/agent-core-v2/src/agent/loop/turnOps.ts index 5516f461a9..f9d853d5b9 100644 --- a/packages/agent-core-v2/src/agent/loop/turnOps.ts +++ b/packages/agent-core-v2/src/agent/loop/turnOps.ts @@ -47,31 +47,31 @@ export interface PromptTurnPayload { readonly origin: PromptOrigin; } -export const promptTurn = defineOp(TurnModel, 'turn.prompt', { - apply: (s, _p: PromptTurnPayload): TurnModelState => ({ nextTurnId: s.nextTurnId + 1 }), -}); - export interface SteerTurnPayload { readonly input: readonly ContentPart[]; readonly origin: PromptOrigin; } -export const steerTurn = defineOp(TurnModel, 'turn.steer', { - apply: (s, _p: SteerTurnPayload): TurnModelState => s, -}); - export interface CancelTurnPayload { readonly turnId?: number; } -export const cancelTurn = defineOp(TurnModel, 'turn.cancel', { - apply: (s, _p: CancelTurnPayload): TurnModelState => s, -}); - -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { +declare module '#/wire/types' { + interface PersistedOpMap { 'turn.prompt': PromptTurnPayload; 'turn.steer': SteerTurnPayload; 'turn.cancel': CancelTurnPayload; } } + +export const promptTurn = defineOp(TurnModel, 'turn.prompt', { + apply: (s, _p) => ({ nextTurnId: s.nextTurnId + 1 }), +}); + +export const steerTurn = defineOp(TurnModel, 'turn.steer', { + apply: (s, _p) => s, +}); + +export const cancelTurn = defineOp(TurnModel, 'turn.cancel', { + apply: (s, _p) => s, +}); diff --git a/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts b/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts index c0cd0ce0ab..abe0969c86 100644 --- a/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts +++ b/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts @@ -33,16 +33,16 @@ export interface McpToolsDiscoveredPayload { readonly collisions?: readonly McpToolCollision[]; } +declare module '#/wire/types' { + interface PersistedOpMap { + 'mcp.tools_discovered': McpToolsDiscoveredPayload; + } +} + export const mcpToolsDiscovered = defineOp(McpDiscoveryModel, 'mcp.tools_discovered', { - apply: (s, p: McpToolsDiscoveredPayload): McpDiscoveryState => { + apply: (s, p) => { const key = `${p.serverName}\n${p.hash}`; if (s.seen.includes(key)) return s; return { seen: [...s.seen, key] }; }, }); - -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { - 'mcp.tools_discovered': McpToolsDiscoveredPayload; - } -} diff --git a/packages/agent-core-v2/src/agent/profile/profileOps.ts b/packages/agent-core-v2/src/agent/profile/profileOps.ts index de95ea6be5..b5000ae0bf 100644 --- a/packages/agent-core-v2/src/agent/profile/profileOps.ts +++ b/packages/agent-core-v2/src/agent/profile/profileOps.ts @@ -116,6 +116,12 @@ export interface SetActiveToolsPayload { readonly names: readonly string[]; } +declare module '#/wire/types' { + interface PersistedOpMap { + 'tools.set_active_tools': SetActiveToolsPayload; + } +} + export const setActiveTools = defineOp(ActiveToolsModel, 'tools.set_active_tools', { - apply: (s, p: SetActiveToolsPayload): ActiveToolsState => (p.names === s ? s : p.names), + apply: (s, p) => (p.names === s ? s : p.names), }); diff --git a/packages/agent-core-v2/src/agent/profile/profileService.ts b/packages/agent-core-v2/src/agent/profile/profileService.ts index 353b35b2a8..9440c1fabe 100644 --- a/packages/agent-core-v2/src/agent/profile/profileService.ts +++ b/packages/agent-core-v2/src/agent/profile/profileService.ts @@ -79,14 +79,6 @@ import { type ProfileModelState, } from './profileOps'; -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { - 'tools.set_active_tools': { - names: readonly string[]; - }; - } -} - declare module '#/app/event/eventBus' { interface DomainEventMap { // `warning` is owned by `profile` (the agents-md-oversized notice). diff --git a/packages/agent-core-v2/src/agent/swarm/swarmOps.ts b/packages/agent-core-v2/src/agent/swarm/swarmOps.ts index a53a41bb04..840ab1fba2 100644 --- a/packages/agent-core-v2/src/agent/swarm/swarmOps.ts +++ b/packages/agent-core-v2/src/agent/swarm/swarmOps.ts @@ -18,6 +18,12 @@ import type { SwarmModeTrigger } from './swarm'; export const SwarmModel = defineModel('swarm', () => null); +declare module '#/wire/types' { + interface PersistedOpMap { + 'swarm_mode.exit': Record; + } +} + export const swarmEnter = defineOp(SwarmModel, 'swarm_mode.enter', { apply: (_s, p: { trigger: SwarmModeTrigger }) => p.trigger, toEvent: () => ({ type: 'agent.status.updated' as const, swarmMode: true }), diff --git a/packages/agent-core-v2/src/agent/task/taskService.ts b/packages/agent-core-v2/src/agent/task/taskService.ts index ac5b307b7e..e738a4cabb 100644 --- a/packages/agent-core-v2/src/agent/task/taskService.ts +++ b/packages/agent-core-v2/src/agent/task/taskService.ts @@ -45,7 +45,10 @@ import { ISessionContext } from '#/session/sessionContext/sessionContext'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { ITelemetryService } from '#/app/telemetry/telemetry'; -import { IAgentWireRecordService, type WireRecord } from '#/agent/wireRecord/wireRecord'; +import { + IAgentWireRecordService, + type PersistedWireRecord, +} from '#/agent/wireRecord/wireRecord'; import { IAgentWireService } from '#/wire/tokens'; import type { IWireService } from '#/wire/wireService'; import { @@ -233,7 +236,14 @@ export class AgentTaskService extends Disposable implements IAgentTaskService { atomicDocs, byteStore, ); - this._register(this.wire.onRestored(() => this.restoreAfterReplay())); + this._register( + this.wire.onRestored(async () => { + for (const record of wireRecord.getRecords()) { + this.markDeliveredNotificationsFromRecord(record); + } + await this.restoreAfterReplay(); + }), + ); this._register( this.eventBus.subscribe('context.spliced', (e) => { if (isCompactionSplice(e)) { @@ -251,15 +261,6 @@ export class AgentTaskService extends Disposable implements IAgentTaskService { this.activeBackgroundTaskReminder(), ), ); - this._register( - wireRecord.hooks.onDidRestoreRecord.register( - 'task-delivered-notifications', - async (ctx, next) => { - this.markDeliveredNotificationsFromRecord(ctx.record); - await next(); - }, - ), - ); } private async restoreAfterReplay(): Promise { @@ -290,7 +291,7 @@ export class AgentTaskService extends Disposable implements IAgentTaskService { } } - private markDeliveredNotificationsFromRecord(record: WireRecord): void { + private markDeliveredNotificationsFromRecord(record: PersistedWireRecord): void { for (const origin of taskOriginsFromRecord(record)) { this.markDeliveredNotification(origin); } @@ -1213,7 +1214,7 @@ function notificationKey(origin: TaskNotificationOrigin): string { return `${origin.taskId}\0${origin.status}\0${origin.notificationId}`; } -function taskOriginsFromRecord(record: WireRecord): readonly TaskNotificationOrigin[] { +function taskOriginsFromRecord(record: PersistedWireRecord): readonly TaskNotificationOrigin[] { const raw = record as { readonly type: string; readonly message?: unknown; diff --git a/packages/agent-core-v2/src/session/cron/cronOps.ts b/packages/agent-core-v2/src/session/cron/cronOps.ts index 959430cfdb..11f0e64a86 100644 --- a/packages/agent-core-v2/src/session/cron/cronOps.ts +++ b/packages/agent-core-v2/src/session/cron/cronOps.ts @@ -38,22 +38,35 @@ export interface CronAddPayload { readonly task: CronTask; } +export interface CronDeletePayload { + readonly ids: readonly string[]; +} + +export interface CronCursorPayload { + readonly id: string; + readonly lastFiredAt: number; +} + +declare module '#/wire/types' { + interface TransientOpMap { + 'cron.add': CronAddPayload; + 'cron.delete': CronDeletePayload; + 'cron.cursor': CronCursorPayload; + } +} + export const cronAdd = defineOp(CronModel, 'cron.add', { persist: false, - apply: (s, p: CronAddPayload): CronModelState => { + apply: (s, p) => { const next = new Map(s); next.set(p.task.id, p.task); return next; }, }); -export interface CronDeletePayload { - readonly ids: readonly string[]; -} - export const cronDelete = defineOp(CronModel, 'cron.delete', { persist: false, - apply: (s, p: CronDeletePayload): CronModelState => { + apply: (s, p) => { let next: Map | undefined; for (const id of p.ids) { if (s.has(id)) { @@ -65,14 +78,9 @@ export const cronDelete = defineOp(CronModel, 'cron.delete', { }, }); -export interface CronCursorPayload { - readonly id: string; - readonly lastFiredAt: number; -} - export const cronCursor = defineOp(CronModel, 'cron.cursor', { persist: false, - apply: (s, p: CronCursorPayload): CronModelState => { + apply: (s, p) => { const task = s.get(p.id); if (task === undefined) return s; const next = new Map(s); diff --git a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts index 06d279e28c..fb97ed5399 100644 --- a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts +++ b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts @@ -56,21 +56,6 @@ export const CRON_FIRED = 'cron_fired' as const; export const CRON_MISSED = 'cron_missed' as const; export const CRON_DELETED = 'cron_deleted' as const; -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { - 'cron.add': { - task: CronTask; - }; - 'cron.delete': { - ids: readonly string[]; - }; - 'cron.cursor': { - id: string; - lastFiredAt: number; - }; - } -} - const STALE_THRESHOLD_MS = 7 * 24 * 60 * 60 * 1000; const DEFAULT_POLL_INTERVAL_MS = 1_000; const MAX_COALESCE_ITERATIONS = 10_000; diff --git a/packages/agent-core-v2/src/session/todo/sessionTodoService.ts b/packages/agent-core-v2/src/session/todo/sessionTodoService.ts index d64f43223c..db81f8b696 100644 --- a/packages/agent-core-v2/src/session/todo/sessionTodoService.ts +++ b/packages/agent-core-v2/src/session/todo/sessionTodoService.ts @@ -37,15 +37,6 @@ import { TodoModel, todoSet } from './todoOps'; import { TODO_LIST_TOOL_NAME, type TodoItem } from './todoItem'; import { TODO_LIST_REMINDER_VARIANT, todoListStaleReminder } from './todoListReminder'; -declare module '#/agent/wireRecord/wireRecord' { - interface WireRecordMap { - 'tools.update_store': { - key: string; - value: unknown; - }; - } -} - const MAIN_AGENT_ID = 'main'; export class SessionTodoService extends Disposable implements ISessionTodoService { diff --git a/packages/agent-core-v2/src/session/todo/todoOps.ts b/packages/agent-core-v2/src/session/todo/todoOps.ts index 12fb78e667..7f690da2e0 100644 --- a/packages/agent-core-v2/src/session/todo/todoOps.ts +++ b/packages/agent-core-v2/src/session/todo/todoOps.ts @@ -31,7 +31,12 @@ export interface ToolStoreUpdatePayload { readonly value: unknown; } +declare module '#/wire/types' { + interface PersistedOpMap { + 'tools.update_store': ToolStoreUpdatePayload; + } +} + export const todoSet = defineOp(TodoModel, 'tools.update_store', { - apply: (s, p: ToolStoreUpdatePayload): TodoModelState => - p.key === 'todo' ? readTodoItems(p.value) : s, + apply: (s, p) => (p.key === 'todo' ? readTodoItems(p.value) : s), }); diff --git a/packages/agent-core-v2/src/wire/wireServiceImpl.ts b/packages/agent-core-v2/src/wire/wireServiceImpl.ts index 721a86e25d..9e987e35dc 100644 --- a/packages/agent-core-v2/src/wire/wireServiceImpl.ts +++ b/packages/agent-core-v2/src/wire/wireServiceImpl.ts @@ -174,13 +174,13 @@ export class WireService extends Disposable implements IWireService { this._register(inst.emitter); this.derivedModels.set(model, inst); - for (const opType of Object.keys(model.reducers)) { + for (const [opType, reducer] of Object.entries(model.reducers)) { let list = this.reducerIndex.get(opType); if (list === undefined) { list = []; this.reducerIndex.set(opType, list); } - list.push({ inst, reducer: model.reducers[opType]! }); + list.push({ inst, reducer }); } return { From 29a298a09f2b539dd0e5c35d8cbbf6350df14107 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Mon, 13 Jul 2026 15:17:05 +0800 Subject: [PATCH 2/4] refactor(v2): bind op definitions to models and require zod payload schemas - ModelDef.defineOp(type, opts) replaces defineOp(model, type, opts); the standalone function stays as the internal primitive - schema (zod, before apply) is required and is the payload type's single source of truth; apply/toEvent payload and return types are inferred, so per-op payload interfaces are deleted and consumers use PayloadOf - Op registries (PersistedOpMap/TransientOpMap) now map op types to typeof the op; classification stays key-level to avoid type cycles - Every op is registered: 12 transient ops carry persist: false, 32 persisted ops ride the wire log --- .changeset/v2-wire-op-schemas.md | 5 + .../agent-core-v2/src/activity/activityOps.ts | 22 ++- .../src/agent/contextMemory/contextOps.ts | 115 ++++++------ .../src/agent/contextSize/contextSizeOps.ts | 15 +- .../src/agent/fullCompaction/compactionOps.ts | 26 +-- .../agent-core-v2/src/agent/goal/goalOps.ts | 51 +++--- .../src/agent/llmRequester/llmRequestOps.ts | 83 ++++----- .../agent/llmRequester/llmRequesterService.ts | 6 +- .../agent-core-v2/src/agent/loop/turnOps.ts | 56 +++--- .../src/agent/mcp/mcpDiscoveryOps.ts | 29 +-- .../agent/permissionMode/permissionModeOps.ts | 14 +- .../permissionRules/permissionRulesOps.ts | 24 ++- .../agent-core-v2/src/agent/plan/planOps.ts | 31 ++-- .../src/agent/profile/profileOps.ts | 39 +++-- .../src/agent/profile/profileService.ts | 8 +- .../replayBuilder/replayTimelineModel.ts | 73 ++++---- .../src/agent/runtime/runtimeOps.ts | 22 ++- .../agent-core-v2/src/agent/skill/skillOps.ts | 14 +- .../agent-core-v2/src/agent/swarm/swarmOps.ts | 14 +- .../agent-core-v2/src/agent/task/taskOps.ts | 22 ++- .../agent-core-v2/src/agent/usage/usageOps.ts | 20 ++- .../src/agent/userTool/userToolOps.ts | 52 +++--- .../src/agent/wireRecord/metadataOps.ts | 15 +- .../src/agent/wireRecord/wireRecord.ts | 64 +------ .../src/agent/wireRecord/wireRecordService.ts | 165 +----------------- .../agentLifecycle/agentLifecycleService.ts | 5 +- .../agent-core-v2/src/session/cron/cronOps.ts | 30 ++-- .../agent-core-v2/src/session/todo/todoOps.ts | 13 +- packages/agent-core-v2/src/wire/model.ts | 32 ++-- packages/agent-core-v2/src/wire/op.ts | 126 ++++++++++--- packages/agent-core-v2/src/wire/types.ts | 47 +++++ .../agent-core-v2/src/wire/wireService.ts | 2 +- .../agent-core-v2/src/wire/wireServiceImpl.ts | 1 + .../test/agent/contextMemory/stubs.ts | 15 +- .../test/agent/goal/goal.test.ts | 6 +- .../app/sessionExport/sessionExport.test.ts | 7 - .../lint/fixtures/duplicate-ops.fixture.ts | 7 +- .../test/lint/op-uniqueness.test.ts | 131 +++++++++++++- .../test/wire/store-event.test.ts | 22 ++- .../test/wire/wire-compat.test.ts | 12 +- .../test/wire/wireServiceImpl.test.ts | 18 +- 41 files changed, 787 insertions(+), 672 deletions(-) create mode 100644 .changeset/v2-wire-op-schemas.md create mode 100644 packages/agent-core-v2/src/wire/types.ts diff --git a/.changeset/v2-wire-op-schemas.md b/.changeset/v2-wire-op-schemas.md new file mode 100644 index 0000000000..32e79ab973 --- /dev/null +++ b/.changeset/v2-wire-op-schemas.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Declare v2 engine wire op payloads with required zod schemas and derive their types from the schemas, with ops declared on their models and every op type registered for replay classification. diff --git a/packages/agent-core-v2/src/activity/activityOps.ts b/packages/agent-core-v2/src/activity/activityOps.ts index 3eb95a4325..f7c7dc43cd 100644 --- a/packages/agent-core-v2/src/activity/activityOps.ts +++ b/packages/agent-core-v2/src/activity/activityOps.ts @@ -14,8 +14,9 @@ * Consumed by the Agent-scope `agentActivityService` and (PR5) the projector. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { PromptOrigin } from '#/agent/contextMemory/types'; import type { AgentLane, BackgroundActivityRef, SessionLane } from './activity'; @@ -46,10 +47,17 @@ export const LaneModel = defineModel('activityLane', () => ({ background: [], })); -export const setLane = defineOp(LaneModel, 'activity.set_lane', { +declare module '#/wire/types' { + interface TransientOpMap { + 'activity.set_lane': typeof setLane; + 'activity.set_session_lane': typeof setSessionLane; + } +} + +export const setLane = LaneModel.defineOp('activity.set_lane', { + schema: z.object({ next: z.custom() }), persist: false, - apply: (s, p: { next: LaneModelState }): LaneModelState => - laneEqual(s, p.next) ? s : p.next, + apply: (s, p) => (laneEqual(s, p.next) ? s : p.next), }); export function laneEqual(a: LaneModelState, b: LaneModelState): boolean { @@ -84,8 +92,8 @@ export const SessionLaneModel = defineModel('sessionActiv activeLeases: 0, })); -export const setSessionLane = defineOp(SessionLaneModel, 'activity.set_session_lane', { +export const setSessionLane = SessionLaneModel.defineOp('activity.set_session_lane', { + schema: z.object({ next: z.custom() }), persist: false, - apply: (s, p: { next: SessionLaneModelState }): SessionLaneModelState => - s.lane === p.next.lane && s.activeLeases === p.next.activeLeases ? s : p.next, + apply: (s, p) => (s.lane === p.next.lane && s.activeLeases === p.next.activeLeases ? s : p.next), }); diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts index 69ed2c55d7..e8ef66ff80 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts @@ -32,9 +32,10 @@ * data that was compacted away during the session. */ +import { z } from 'zod'; + import type { ContentPart } from '#/app/llmProtocol/message'; import { defineModel, type PartsTransformer } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { PersistedRecord } from '#/wire/wireService'; import { @@ -120,68 +121,71 @@ function popSwarmModeReminder(state: ContextMessage[], _payload: unknown): Conte return resetFold(state.slice(0, -1)) as ContextMessage[]; } -export interface ContextMessagePayload { - readonly message: ContextMessage; -} - -export interface ContextLoopEventPayload { - readonly event: LoopRecordedEvent; -} - declare module '#/wire/types' { interface PersistedOpMap { - 'context.append_message': ContextMessagePayload; - 'context.append_loop_event': ContextLoopEventPayload; + 'context.append_message': typeof contextAppendMessage; + 'context.append_loop_event': typeof contextAppendLoopEvent; + 'context.clear': typeof contextClear; + 'context.apply_compaction': typeof contextApplyCompaction; + 'context.undo': typeof contextUndo; } } -export const contextAppendMessage = defineOp(ContextModel, 'context.append_message', { - apply: (state, p): ContextMessage[] => - foldAppendMessage(state, p.message) as ContextMessage[], +// `ContextMessage` / `LoopRecordedEvent` are large domain unions owned by +// sibling modules; `z.custom` keeps their exact types without restating them. +const contextMessageSchema = z.custom(); +const loopRecordedEventSchema = z.custom(); + +export const contextAppendMessage = ContextModel.defineOp('context.append_message', { + schema: z.object({ message: contextMessageSchema }), + apply: (state, p) => foldAppendMessage(state, p.message) as ContextMessage[], }); -export const contextAppendLoopEvent = defineOp(ContextModel, 'context.append_loop_event', { +export const contextAppendLoopEvent = ContextModel.defineOp('context.append_loop_event', { + schema: z.object({ event: loopRecordedEventSchema }), apply: (state, p) => foldLoopEvent(state, p.event) as ContextMessage[], }); -export const contextClear = defineOp(ContextModel, 'context.clear', { - apply: (state) => state.length === 0 ? state : (resetFold([]) as ContextMessage[]), +export const contextClear = ContextModel.defineOp('context.clear', { + schema: z.object({}), + apply: (state) => (state.length === 0 ? state : (resetFold([]) as ContextMessage[])), }); -interface ContextCompactionBasePayload { - readonly tokensBefore?: number; - readonly tokensAfter?: number; - readonly keptUserMessageCount?: number; - readonly keptHeadUserMessageCount?: number; - readonly droppedCount?: number; - readonly legacyTail?: boolean; -} - -export interface TextSummaryCompactionPayload extends ContextCompactionBasePayload { - readonly summary: string; - readonly compactedCount: number; - readonly contextSummary?: string; -} - -interface ContextSummaryCompactionPayload extends ContextCompactionBasePayload { - readonly contextSummary: string; - readonly compactedCount: number; - readonly summary?: string; -} - -interface LegacyMessageSummaryCompactionPayload extends ContextCompactionBasePayload { - readonly summary: ContextMessage; - readonly count: number; - readonly compactedCount?: number; -} - -export type ContextCompactionPayload = - | TextSummaryCompactionPayload - | ContextSummaryCompactionPayload - | LegacyMessageSummaryCompactionPayload; - -export const contextApplyCompaction = defineOp(ContextModel, 'context.apply_compaction', { - apply: (state, p: ContextCompactionPayload): ContextMessage[] => { +const contextCompactionBaseShape = { + tokensBefore: z.number().optional(), + tokensAfter: z.number().optional(), + keptUserMessageCount: z.number().optional(), + keptHeadUserMessageCount: z.number().optional(), + droppedCount: z.number().optional(), + legacyTail: z.boolean().optional(), +}; + +const contextApplyCompactionSchema = z.union([ + z.object({ + ...contextCompactionBaseShape, + summary: z.string(), + compactedCount: z.number(), + contextSummary: z.string().optional(), + }), + z.object({ + ...contextCompactionBaseShape, + contextSummary: z.string(), + compactedCount: z.number(), + summary: z.string().optional(), + }), + z.object({ + ...contextCompactionBaseShape, + summary: contextMessageSchema, + count: z.number(), + compactedCount: z.number().optional(), + }), +]); + +type ContextCompactionPayload = z.infer; + +export const contextApplyCompaction = ContextModel.defineOp('context.apply_compaction', { + schema: contextApplyCompactionSchema, + apply: (state, p) => { const result = buildContextCompactionShape(state, readContextCompactionShapeInput(p)); return resetFold([...result.messages]) as ContextMessage[]; }, @@ -284,10 +288,6 @@ function isContextMessage(value: unknown): value is ContextMessage { return typeof message.role === 'string' && Array.isArray(message.content); } -export interface ContextUndoPayload { - readonly count: number; -} - export interface UndoCut { readonly cutIndex: number; readonly removedCount: number; @@ -375,8 +375,9 @@ export function formatUndoUnavailableMessage( } } -export const contextUndo = defineOp(ContextModel, 'context.undo', { - apply: (state, p: ContextUndoPayload): ContextMessage[] => { +export const contextUndo = ContextModel.defineOp('context.undo', { + schema: z.object({ count: z.number() }), + apply: (state, p) => { if (p.count <= 0 || state.length === 0) return state; const cut = computeUndoCut(state, p.count); if (!isFullyUndoable(cut, p.count)) return state; diff --git a/packages/agent-core-v2/src/agent/contextSize/contextSizeOps.ts b/packages/agent-core-v2/src/agent/contextSize/contextSizeOps.ts index 629604e67f..71a0eb0b04 100644 --- a/packages/agent-core-v2/src/agent/contextSize/contextSizeOps.ts +++ b/packages/agent-core-v2/src/agent/contextSize/contextSizeOps.ts @@ -22,8 +22,9 @@ * `contextSizeService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; export interface ContextSizeState { readonly length: number; @@ -35,14 +36,16 @@ export const ContextSizeModel = defineModel('contextSize', () tokens: 0, })); -export interface ContextSizeMeasuredPayload { - readonly length: number; - readonly tokens: number; +declare module '#/wire/types' { + interface TransientOpMap { + 'context_size.measured': typeof contextSizeMeasured; + } } -export const contextSizeMeasured = defineOp(ContextSizeModel, 'context_size.measured', { +export const contextSizeMeasured = ContextSizeModel.defineOp('context_size.measured', { + schema: z.object({ length: z.number(), tokens: z.number() }), persist: false, - apply: (s, p: ContextSizeMeasuredPayload): ContextSizeState => { + apply: (s, p) => { const length = normalizeMeasuredLength(p.length); const tokens = Math.max(0, p.tokens); if (s.length === length && s.tokens === tokens) return s; diff --git a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts index 9e9c31b434..9ca2959ff4 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts @@ -36,8 +36,9 @@ * `getRecords()`. Consumed by the Agent-scope `fullCompactionService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { CompactionBlockedEvent, CompactionCancelledEvent, @@ -66,20 +67,17 @@ declare module '#/app/event/eventBus' { } } -export type FullCompactionBeginPayload = CompactionBeginData; - -export type FullCompactionCompletePayload = Record; - declare module '#/wire/types' { interface PersistedOpMap { - 'full_compaction.begin': FullCompactionBeginPayload; - 'full_compaction.cancel': {}; - 'full_compaction.complete': FullCompactionCompletePayload; + 'full_compaction.begin': typeof fullCompactionBegin; + 'full_compaction.cancel': typeof fullCompactionCancel; + 'full_compaction.complete': typeof fullCompactionComplete; } } -export const fullCompactionBegin = defineOp(CompactionModel, 'full_compaction.begin', { - apply: (s, _p) => (s.phase === 'running' ? s : { phase: 'running' }), +export const fullCompactionBegin = CompactionModel.defineOp('full_compaction.begin', { + schema: z.custom(), + apply: (s) => (s.phase === 'running' ? s : { phase: 'running' }), toEvent: (p) => ({ type: 'compaction.started' as const, trigger: p.source, @@ -87,10 +85,12 @@ export const fullCompactionBegin = defineOp(CompactionModel, 'full_compaction.be }), }); -export const fullCompactionCancel = defineOp(CompactionModel, 'full_compaction.cancel', { +export const fullCompactionCancel = CompactionModel.defineOp('full_compaction.cancel', { + schema: z.object({}), apply: (s) => (s.phase === 'idle' ? s : { phase: 'idle' }), }); -export const fullCompactionComplete = defineOp(CompactionModel, 'full_compaction.complete', { - apply: (s, _p) => (s.phase === 'idle' ? s : { phase: 'idle' }), +export const fullCompactionComplete = CompactionModel.defineOp('full_compaction.complete', { + schema: z.object({}), + apply: (s) => (s.phase === 'idle' ? s : { phase: 'idle' }), }); diff --git a/packages/agent-core-v2/src/agent/goal/goalOps.ts b/packages/agent-core-v2/src/agent/goal/goalOps.ts index 0dd013cec0..8a5482cb48 100644 --- a/packages/agent-core-v2/src/agent/goal/goalOps.ts +++ b/packages/agent-core-v2/src/agent/goal/goalOps.ts @@ -19,8 +19,9 @@ * `goalService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { GoalActor, @@ -55,32 +56,21 @@ declare module '#/app/event/eventBus' { } } -export interface GoalCreatePayload { - readonly goalId: string; - readonly objective: string; - readonly completionCriterion?: string; -} - -export interface GoalUpdatePayload { - readonly status?: GoalStatus; - readonly reason?: string; - readonly turnsUsed?: number; - readonly tokensUsed?: number; - readonly wallClockMs?: number; - readonly budgetLimits?: GoalBudgetLimits; - readonly actor?: GoalActor; -} - declare module '#/wire/types' { interface PersistedOpMap { - 'goal.create': GoalCreatePayload; - 'goal.update': GoalUpdatePayload; - 'goal.clear': {}; - forked: {}; + 'goal.create': typeof createGoal; + 'goal.update': typeof updateGoal; + 'goal.clear': typeof clearGoal; + forked: typeof forkGoal; } } -export const createGoal = defineOp(GoalModel, 'goal.create', { +export const createGoal = GoalModel.defineOp('goal.create', { + schema: z.object({ + goalId: z.string(), + objective: z.string(), + completionCriterion: z.string().optional(), + }), apply: (_s, p) => ({ goalId: p.goalId, objective: p.objective, @@ -93,7 +83,16 @@ export const createGoal = defineOp(GoalModel, 'goal.create', { }), }); -export const updateGoal = defineOp(GoalModel, 'goal.update', { +export const updateGoal = GoalModel.defineOp('goal.update', { + schema: z.object({ + status: z.custom().optional(), + reason: z.string().optional(), + turnsUsed: z.number().optional(), + tokensUsed: z.number().optional(), + wallClockMs: z.number().optional(), + budgetLimits: z.custom().optional(), + actor: z.custom().optional(), + }), apply: (s, p) => { if (s === null) return null; let next: GoalState | undefined; @@ -120,10 +119,12 @@ export const updateGoal = defineOp(GoalModel, 'goal.update', { }, }); -export const clearGoal = defineOp(GoalModel, 'goal.clear', { +export const clearGoal = GoalModel.defineOp('goal.clear', { + schema: z.object({}), apply: () => null, }); -export const forkGoal = defineOp(GoalModel, 'forked', { +export const forkGoal = GoalModel.defineOp('forked', { + schema: z.object({}), apply: () => null, }); diff --git a/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts b/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts index 7fe126860f..9634594bc2 100644 --- a/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts +++ b/packages/agent-core-v2/src/agent/llmRequester/llmRequestOps.ts @@ -6,9 +6,10 @@ * the Agent-scope `llmRequester` implementation. */ +import { z } from 'zod'; + import type { ThinkingEffort } from '#/app/llmProtocol/thinkingEffort'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; export interface LlmRequestToolSchema { readonly name: string; @@ -25,52 +26,52 @@ export const LlmRequestTraceModel = defineModel( () => ({ seenToolsHashes: [] }), ); -export interface LlmToolsSnapshotPayload { - readonly hash: string; - readonly tools: readonly LlmRequestToolSchema[]; -} - -export interface LlmRequestPayload { - readonly kind: 'loop' | 'compaction'; - readonly provider: string; - readonly model: string; - readonly modelAlias?: string; - readonly thinkingEffort?: ThinkingEffort; - readonly thinkingKeep?: string; - readonly temperature?: number; - readonly topP?: number; - readonly maxTokens?: number; - readonly betaApi?: boolean; - /** Progressive tool disclosure in effect (env flag × model capability). */ - readonly toolSelect: boolean; - readonly systemPromptHash: string; - readonly systemPrompt?: string; - readonly toolsHash: string; - readonly messageCount: number; - readonly turnStep?: string; - readonly attempt?: string; - readonly projection?: 'strict'; - readonly droppedCount?: number; -} +const llmToolEntrySchema = z.object({ + name: z.string(), + description: z.string(), + parameters: z.record(z.string(), z.unknown()), +}); declare module '#/wire/types' { interface PersistedOpMap { - 'llm.tools_snapshot': LlmToolsSnapshotPayload; - 'llm.request': LlmRequestPayload; + 'llm.tools_snapshot': typeof llmToolsSnapshot; + 'llm.request': typeof llmRequest; } } -export const llmToolsSnapshot = defineOp( - LlmRequestTraceModel, - 'llm.tools_snapshot', - { - apply: (s, p) => { - if (s.seenToolsHashes.includes(p.hash)) return s; - return { seenToolsHashes: [...s.seenToolsHashes, p.hash] }; - }, +export const llmToolsSnapshot = LlmRequestTraceModel.defineOp('llm.tools_snapshot', { + schema: z.object({ + hash: z.string(), + tools: z.array(llmToolEntrySchema).readonly(), + }), + apply: (s, p) => { + if (s.seenToolsHashes.includes(p.hash)) return s; + return { seenToolsHashes: [...s.seenToolsHashes, p.hash] }; }, -); +}); -export const llmRequest = defineOp(LlmRequestTraceModel, 'llm.request', { - apply: (s, _p) => s, +export const llmRequest = LlmRequestTraceModel.defineOp('llm.request', { + schema: z.object({ + kind: z.enum(['loop', 'compaction']), + provider: z.string(), + model: z.string(), + modelAlias: z.string().optional(), + thinkingEffort: z.custom().optional(), + thinkingKeep: z.string().optional(), + temperature: z.number().optional(), + topP: z.number().optional(), + maxTokens: z.number().optional(), + betaApi: z.boolean().optional(), + /** Progressive tool disclosure in effect (env flag × model capability). */ + toolSelect: z.boolean(), + systemPromptHash: z.string(), + systemPrompt: z.string().optional(), + toolsHash: z.string(), + messageCount: z.number(), + turnStep: z.string().optional(), + attempt: z.string().optional(), + projection: z.literal('strict').optional(), + droppedCount: z.number().optional(), + }), + apply: (s) => s, }); diff --git a/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts b/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts index e6f529f7e2..7724c2e5da 100644 --- a/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts +++ b/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts @@ -51,6 +51,7 @@ import type { Protocol } from '#/app/protocol/protocol'; import type { ApiErrorEvent } from '#/app/telemetry/events'; import { ITelemetryService } from '#/app/telemetry/telemetry'; import { IAgentWireService } from '#/wire/tokens'; +import type { PayloadOf } from '#/wire/types'; import type { IWireService } from '#/wire/wireService'; import { THINKING_SECTION, type ThinkingConfig } from '#/agent/profile/configSection'; import { resolveThinkingKeep } from '#/agent/profile/thinking'; @@ -68,7 +69,6 @@ import { LlmRequestTraceModel, llmRequest, llmToolsSnapshot, - type LlmRequestPayload, type LlmRequestToolSchema, } from './llmRequestOps'; import { isAbortError } from '#/_base/utils/abort'; @@ -394,7 +394,7 @@ export class AgentLLMRequesterService implements IAgentLLMRequesterService { const models = this.config.get(MODELS_SECTION); const modelConfig = input.modelAlias === undefined ? undefined : models?.[input.modelAlias]; - const payload: LlmRequestPayload = { + const payload: PayloadOf = { kind: requestKindForRecord(fields), provider: input.protocol, model: input.modelName, @@ -488,7 +488,7 @@ function toolSignature(tools: readonly Tool[]): readonly LlmRequestToolSchema[] return tools.map(({ name, description, parameters }) => ({ name, description, parameters })); } -function requestKindForRecord(fields: LLMRequestLogFields): LlmRequestPayload['kind'] { +function requestKindForRecord(fields: LLMRequestLogFields): PayloadOf['kind'] { if (fields['kind'] === 'compaction') return 'compaction'; if (fields['requestKind'] === 'full_compaction') return 'compaction'; return 'loop'; diff --git a/packages/agent-core-v2/src/agent/loop/turnOps.ts b/packages/agent-core-v2/src/agent/loop/turnOps.ts index f9d853d5b9..8cad5371a3 100644 --- a/packages/agent-core-v2/src/agent/loop/turnOps.ts +++ b/packages/agent-core-v2/src/agent/loop/turnOps.ts @@ -19,8 +19,9 @@ * `activity` kernel (which reads the next turn id on admission). */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { ContentPart } from '#/app/llmProtocol/message'; import type { PromptOrigin } from '#/agent/contextMemory/types'; @@ -30,48 +31,43 @@ export interface TurnModelState { export const TurnModel = defineModel('turn', () => ({ nextTurnId: 0 }), { reducers: { - 'context.append_loop_event': (s, p: { event?: { turnId?: unknown } }): TurnModelState => { - const raw = p?.event?.turnId; - if (typeof raw !== 'string' && typeof raw !== 'number') return s; - const turnId = Number.parseInt(String(raw), 10); - if (Number.isInteger(turnId) && turnId >= s.nextTurnId) { - return { nextTurnId: turnId + 1 }; + 'context.append_loop_event': (state, { event }) => { + if (event.type === 'tool.result' || event.turnId === undefined) { + return state; } - return s; + + const turnId = Number.parseInt(event.turnId, 10); + return Number.isInteger(turnId) && turnId >= state.nextTurnId + ? { nextTurnId: turnId + 1 } + : state; }, }, }); -export interface PromptTurnPayload { - readonly input: readonly ContentPart[]; - readonly origin: PromptOrigin; -} - -export interface SteerTurnPayload { - readonly input: readonly ContentPart[]; - readonly origin: PromptOrigin; -} - -export interface CancelTurnPayload { - readonly turnId?: number; -} +const turnInputShape = { + input: z.custom(), + origin: z.custom(), +}; declare module '#/wire/types' { interface PersistedOpMap { - 'turn.prompt': PromptTurnPayload; - 'turn.steer': SteerTurnPayload; - 'turn.cancel': CancelTurnPayload; + 'turn.prompt': typeof promptTurn; + 'turn.steer': typeof steerTurn; + 'turn.cancel': typeof cancelTurn; } } -export const promptTurn = defineOp(TurnModel, 'turn.prompt', { - apply: (s, _p) => ({ nextTurnId: s.nextTurnId + 1 }), +export const promptTurn = TurnModel.defineOp('turn.prompt', { + schema: z.object(turnInputShape), + apply: (s) => ({ nextTurnId: s.nextTurnId + 1 }), }); -export const steerTurn = defineOp(TurnModel, 'turn.steer', { - apply: (s, _p) => s, +export const steerTurn = TurnModel.defineOp('turn.steer', { + schema: z.object(turnInputShape), + apply: (s) => s, }); -export const cancelTurn = defineOp(TurnModel, 'turn.cancel', { - apply: (s, _p) => s, +export const cancelTurn = TurnModel.defineOp('turn.cancel', { + schema: z.object({ turnId: z.number().optional() }), + apply: (s) => s, }); diff --git a/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts b/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts index abe0969c86..4ea63a17c0 100644 --- a/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts +++ b/packages/agent-core-v2/src/agent/mcp/mcpDiscoveryOps.ts @@ -5,8 +5,9 @@ * keyed by `${serverName}\n${hash}` entries already present in this log. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { MCPToolDefinition } from './types'; export interface McpToolCollision { @@ -25,21 +26,29 @@ export const McpDiscoveryModel = defineModel('mcp.discovery', seen: [], })); -export interface McpToolsDiscoveredPayload { - readonly serverName: string; - readonly hash: string; - readonly tools: readonly MCPToolDefinition[]; - readonly enabledNames: readonly string[]; - readonly collisions?: readonly McpToolCollision[]; -} +const mcpToolCollisionSchema = z.object({ + qualified: z.string(), + toolName: z.string(), + collidesWith: z.union([ + z.object({ kind: z.literal('same_server'), toolName: z.string() }), + z.object({ kind: z.literal('other_server'), serverName: z.string() }), + ]), +}); declare module '#/wire/types' { interface PersistedOpMap { - 'mcp.tools_discovered': McpToolsDiscoveredPayload; + 'mcp.tools_discovered': typeof mcpToolsDiscovered; } } -export const mcpToolsDiscovered = defineOp(McpDiscoveryModel, 'mcp.tools_discovered', { +export const mcpToolsDiscovered = McpDiscoveryModel.defineOp('mcp.tools_discovered', { + schema: z.object({ + serverName: z.string(), + hash: z.string(), + tools: z.custom(), + enabledNames: z.array(z.string()).readonly(), + collisions: z.array(mcpToolCollisionSchema).readonly().optional(), + }), apply: (s, p) => { const key = `${p.serverName}\n${p.hash}`; if (s.seen.includes(key)) return s; diff --git a/packages/agent-core-v2/src/agent/permissionMode/permissionModeOps.ts b/packages/agent-core-v2/src/agent/permissionMode/permissionModeOps.ts index 3b38144452..d17fb9670e 100644 --- a/packages/agent-core-v2/src/agent/permissionMode/permissionModeOps.ts +++ b/packages/agent-core-v2/src/agent/permissionMode/permissionModeOps.ts @@ -9,12 +9,20 @@ * type). Consumed by the Agent-scope `permissionModeService`. */ +import { z } from 'zod'; + import type { PermissionMode } from '#/agent/permissionPolicy/types'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; export const PermissionModeModel = defineModel('permissionMode', () => 'manual'); -export const setMode = defineOp(PermissionModeModel, 'permission.set_mode', { - apply: (_s, p: { mode: PermissionMode }) => p.mode, +declare module '#/wire/types' { + interface PersistedOpMap { + 'permission.set_mode': typeof setMode; + } +} + +export const setMode = PermissionModeModel.defineOp('permission.set_mode', { + schema: z.object({ mode: z.custom() }), + apply: (_s, p) => p.mode, }); diff --git a/packages/agent-core-v2/src/agent/permissionRules/permissionRulesOps.ts b/packages/agent-core-v2/src/agent/permissionRules/permissionRulesOps.ts index fcf33d8eca..2d5f38252c 100644 --- a/packages/agent-core-v2/src/agent/permissionRules/permissionRulesOps.ts +++ b/packages/agent-core-v2/src/agent/permissionRules/permissionRulesOps.ts @@ -19,8 +19,9 @@ * `permissionRulesService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { PermissionApprovalResultRecord, PermissionRule } from './permissionRules'; @@ -34,19 +35,30 @@ export const PermissionRulesModel = defineModel('perm sessionApprovalRulePatterns: [], })); -export const addPermissionRules = defineOp(PermissionRulesModel, 'permission.rules.add', { +declare module '#/wire/types' { + interface PersistedOpMap { + 'permission.record_approval_result': typeof recordApprovalResult; + } + + interface TransientOpMap { + 'permission.rules.add': typeof addPermissionRules; + } +} + +export const addPermissionRules = PermissionRulesModel.defineOp('permission.rules.add', { + schema: z.object({ rules: z.custom() }), persist: false, - apply: (s, p: { rules: readonly PermissionRule[] }): PermissionRulesModelState => { + apply: (s, p) => { if (p.rules.length === 0) return s; return { ...s, rules: [...s.rules, ...p.rules] }; }, }); -export const recordApprovalResult = defineOp( - PermissionRulesModel, +export const recordApprovalResult = PermissionRulesModel.defineOp( 'permission.record_approval_result', { - apply: (s, p: PermissionApprovalResultRecord): PermissionRulesModelState => { + schema: z.custom(), + apply: (s, p) => { const pattern = p.sessionApprovalRule; if ( p.result.decision !== 'approved' || diff --git a/packages/agent-core-v2/src/agent/plan/planOps.ts b/packages/agent-core-v2/src/agent/plan/planOps.ts index 5aeaa507e7..60e3dbd4f2 100644 --- a/packages/agent-core-v2/src/agent/plan/planOps.ts +++ b/packages/agent-core-v2/src/agent/plan/planOps.ts @@ -19,8 +19,9 @@ * Consumed by the Agent-scope `planService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; export interface PlanState { readonly active: boolean; @@ -29,26 +30,28 @@ export interface PlanState { export const PlanModel = defineModel('plan', () => ({ active: false })); -export interface PlanModeEnterPayload { - readonly id: string; -} - -export const planModeEnter = defineOp(PlanModel, 'plan_mode.enter', { - apply: (s, p: PlanModeEnterPayload): PlanState => - s.active && s.id === p.id ? s : { active: true, id: p.id }, +export const planModeEnter = PlanModel.defineOp('plan_mode.enter', { + schema: z.object({ id: z.string() }), + apply: (s, p) => (s.active && s.id === p.id ? s : { active: true, id: p.id }), toEvent: () => ({ type: 'agent.status.updated' as const, planMode: true }), }); -export interface PlanModeIdPayload { - readonly id?: string; +declare module '#/wire/types' { + interface PersistedOpMap { + 'plan_mode.enter': typeof planModeEnter; + 'plan_mode.cancel': typeof planModeCancel; + 'plan_mode.exit': typeof planModeExit; + } } -export const planModeCancel = defineOp(PlanModel, 'plan_mode.cancel', { - apply: (s, _p: PlanModeIdPayload): PlanState => (s.active === false ? s : { active: false }), +export const planModeCancel = PlanModel.defineOp('plan_mode.cancel', { + schema: z.object({ id: z.string().optional() }), + apply: (s) => (s.active === false ? s : { active: false }), toEvent: () => ({ type: 'agent.status.updated' as const, planMode: false }), }); -export const planModeExit = defineOp(PlanModel, 'plan_mode.exit', { - apply: (s, _p: PlanModeIdPayload): PlanState => (s.active === false ? s : { active: false }), +export const planModeExit = PlanModel.defineOp('plan_mode.exit', { + schema: z.object({ id: z.string().optional() }), + apply: (s) => (s.active === false ? s : { active: false }), toEvent: () => ({ type: 'agent.status.updated' as const, planMode: false }), }); diff --git a/packages/agent-core-v2/src/agent/profile/profileOps.ts b/packages/agent-core-v2/src/agent/profile/profileOps.ts index b5000ae0bf..b43ee84613 100644 --- a/packages/agent-core-v2/src/agent/profile/profileOps.ts +++ b/packages/agent-core-v2/src/agent/profile/profileOps.ts @@ -28,9 +28,11 @@ * Consumed by the Agent-scope `profileService`. */ +import { z } from 'zod'; + import type { ThinkingEffort } from '#/app/llmProtocol/thinkingEffort'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; +import type { PayloadOf } from '#/wire/types'; import { ProfileError, ProfileErrors } from './profile'; @@ -47,17 +49,16 @@ export const ProfileModel = defineModel('profile', () => ({ systemPrompt: '', })); -export interface ConfigUpdatePayload { - readonly cwd?: string; - readonly modelAlias?: string; - readonly profileName?: string; - readonly thinkingEffort?: ThinkingEffort; - readonly thinkingLevel?: ThinkingEffort; - readonly systemPrompt?: string; -} - -export const configUpdate = defineOp(ProfileModel, 'config.update', { - apply: (s, p: ConfigUpdatePayload): ProfileModelState => { +export const configUpdate = ProfileModel.defineOp('config.update', { + schema: z.object({ + cwd: z.string().optional(), + modelAlias: z.string().optional(), + profileName: z.string().optional(), + thinkingEffort: z.custom().optional(), + thinkingLevel: z.custom().optional(), + systemPrompt: z.string().optional(), + }), + apply: (s, p) => { let next: ProfileModelState | undefined; if (p.cwd !== undefined && p.cwd !== s.cwd) { next = { ...(next ?? s), cwd: p.cwd }; @@ -79,7 +80,9 @@ export const configUpdate = defineOp(ProfileModel, 'config.update', { }, }); -function configUpdateThinkingLevel(p: ConfigUpdatePayload): ThinkingEffort | undefined { +function configUpdateThinkingLevel( + p: PayloadOf, +): ThinkingEffort | undefined { if (p.thinkingEffort !== undefined && p.thinkingLevel !== undefined) { if (p.thinkingEffort !== p.thinkingLevel) { throw new ProfileError( @@ -112,16 +115,14 @@ export const ActiveToolsModel = defineModel( () => undefined, ); -export interface SetActiveToolsPayload { - readonly names: readonly string[]; -} - declare module '#/wire/types' { interface PersistedOpMap { - 'tools.set_active_tools': SetActiveToolsPayload; + 'config.update': typeof configUpdate; + 'tools.set_active_tools': typeof setActiveTools; } } -export const setActiveTools = defineOp(ActiveToolsModel, 'tools.set_active_tools', { +export const setActiveTools = ActiveToolsModel.defineOp('tools.set_active_tools', { + schema: z.object({ names: z.array(z.string()).readonly() }), apply: (s, p) => (p.names === s ? s : p.names), }); diff --git a/packages/agent-core-v2/src/agent/profile/profileService.ts b/packages/agent-core-v2/src/agent/profile/profileService.ts index 9440c1fabe..f40f65cf96 100644 --- a/packages/agent-core-v2/src/agent/profile/profileService.ts +++ b/packages/agent-core-v2/src/agent/profile/profileService.ts @@ -52,6 +52,7 @@ import { ITelemetryService } from '#/app/telemetry/telemetry'; import { IAgentTelemetryContextService } from '#/app/telemetry/agentTelemetryContext'; import type { ToolSource } from '#/agent/tool/toolContract'; import { IAgentWireService } from '#/wire/tokens'; +import type { PayloadOf } from '#/wire/types'; import type { IWireService } from '#/wire/wireService'; import { IEventBus } from '#/app/event/eventBus'; import { prepareSystemPromptContext } from './context'; @@ -75,7 +76,6 @@ import { ProfileModel, setActiveTools, type ActiveToolsState, - type ConfigUpdatePayload, type ProfileModelState, } from './profileOps'; @@ -388,8 +388,10 @@ export class AgentProfileService implements IAgentProfileService { private resolveConfigPayload( changed: Omit, - ): ConfigUpdatePayload { - const payload: { -readonly [K in keyof ConfigUpdatePayload]: ConfigUpdatePayload[K] } = {}; + ): PayloadOf { + const payload: { + -readonly [K in keyof PayloadOf]: PayloadOf[K]; + } = {}; if (changed.cwd !== undefined) payload.cwd = changed.cwd; if (changed.modelAlias !== undefined) payload.modelAlias = changed.modelAlias; if (changed.profileName !== undefined) payload.profileName = changed.profileName; diff --git a/packages/agent-core-v2/src/agent/replayBuilder/replayTimelineModel.ts b/packages/agent-core-v2/src/agent/replayBuilder/replayTimelineModel.ts index 775f03591e..f90eace3ef 100644 --- a/packages/agent-core-v2/src/agent/replayBuilder/replayTimelineModel.ts +++ b/packages/agent-core-v2/src/agent/replayBuilder/replayTimelineModel.ts @@ -16,85 +16,82 @@ import { contextAppendMessage, contextApplyCompaction, - type ContextCompactionPayload, - type ContextMessagePayload, } from '#/agent/contextMemory/contextOps'; import { fullCompactionBegin, fullCompactionCancel, fullCompactionComplete, - type FullCompactionBeginPayload, - type FullCompactionCompletePayload, } from '#/agent/fullCompaction/compactionOps'; -import { - clearGoal, - createGoal, - updateGoal, - type GoalCreatePayload, - type GoalUpdatePayload, -} from '#/agent/goal/goalOps'; -import { - planModeCancel, - planModeEnter, - planModeExit, - type PlanModeEnterPayload, - type PlanModeIdPayload, -} from '#/agent/plan/planOps'; -import { configUpdate, type ConfigUpdatePayload } from '#/agent/profile/profileOps'; +import { clearGoal, createGoal, updateGoal } from '#/agent/goal/goalOps'; +import { planModeCancel, planModeEnter, planModeExit } from '#/agent/plan/planOps'; +import { configUpdate } from '#/agent/profile/profileOps'; import type { PermissionMode } from '#/agent/permissionPolicy/types'; import { setMode } from '#/agent/permissionMode/permissionModeOps'; import type { PermissionApprovalResultRecord } from '#/agent/permissionRules/permissionRules'; import { recordApprovalResult } from '#/agent/permissionRules/permissionRulesOps'; import { type DerivedModelDef, defineDerivedModel } from '#/wire/model'; +import type { ModelReducers, OpPayload, OpType, PayloadOf } from '#/wire/types'; + +type TimelineMapperMap = { + [K in OpType]?: (payload: OpPayload) => unknown; +}; + +type TimelineEntry = { + [K in keyof M]: M[K] extends (...args: never[]) => infer E ? E : never; +}[keyof M]; + +type ErasedTimelineMapper = (payload: unknown) => E; -// eslint-disable-next-line @typescript-eslint/no-explicit-any -function defineDerivedTimeline any>>( +function defineDerivedTimeline( name: string, - mappers: M, -): DerivedModelDef[]> { - type E = ReturnType; - const reducers: Record readonly E[]> = {}; - for (const opType of Object.keys(mappers)) { - reducers[opType] = (s, p) => [...s, mappers[opType]!(p)]; - } + mappers: M & Record, never>, +): DerivedModelDef[]> { + type E = TimelineEntry; + const entries = Object.entries(mappers) as [OpType, ErasedTimelineMapper][]; + const reducers = Object.fromEntries( + entries.map( + ([opType, mapper]) => + [opType, (state: readonly E[], payload: unknown) => [...state, mapper(payload)]] as const, + ), + ) as ModelReducers; return defineDerivedModel(name, () => [], reducers); } export const ReplayTimelineModel = defineDerivedTimeline('agent.replayTimeline', { - [contextAppendMessage.type]: (p: ContextMessagePayload) => + [contextAppendMessage.type]: (p: PayloadOf) => ({ type: contextAppendMessage.type, payload: p }) as const, - [contextApplyCompaction.type]: (p: ContextCompactionPayload) => + [contextApplyCompaction.type]: (p: PayloadOf) => ({ type: contextApplyCompaction.type, payload: p }) as const, - [fullCompactionBegin.type]: (p: FullCompactionBeginPayload) => + [fullCompactionBegin.type]: (p: PayloadOf) => ({ type: fullCompactionBegin.type, payload: p }) as const, [fullCompactionCancel.type]: () => ({ type: fullCompactionCancel.type }) as const, - [fullCompactionComplete.type]: (p: FullCompactionCompletePayload) => + [fullCompactionComplete.type]: (p: PayloadOf) => ({ type: fullCompactionComplete.type, payload: p }) as const, - [createGoal.type]: (p: GoalCreatePayload) => + [createGoal.type]: (p: PayloadOf) => ({ type: createGoal.type, payload: p }) as const, - [updateGoal.type]: (p: GoalUpdatePayload) => + [updateGoal.type]: (p: PayloadOf) => ({ type: updateGoal.type, payload: p }) as const, [clearGoal.type]: () => ({ type: clearGoal.type }) as const, - [planModeEnter.type]: (p: PlanModeEnterPayload) => + [planModeEnter.type]: (p: PayloadOf) => ({ type: planModeEnter.type, payload: p }) as const, - [planModeCancel.type]: (p: PlanModeIdPayload) => + [planModeCancel.type]: (p: PayloadOf) => ({ type: planModeCancel.type, payload: p }) as const, - [planModeExit.type]: (p: PlanModeIdPayload) => + [planModeExit.type]: (p: PayloadOf) => ({ type: planModeExit.type, payload: p }) as const, - [configUpdate.type]: (p: ConfigUpdatePayload) => + [configUpdate.type]: (p: PayloadOf) => ({ type: configUpdate.type, payload: p }) as const, [setMode.type]: (p: { mode: PermissionMode }) => diff --git a/packages/agent-core-v2/src/agent/runtime/runtimeOps.ts b/packages/agent-core-v2/src/agent/runtime/runtimeOps.ts index 5186278c20..3711c0db49 100644 --- a/packages/agent-core-v2/src/agent/runtime/runtimeOps.ts +++ b/packages/agent-core-v2/src/agent/runtime/runtimeOps.ts @@ -14,8 +14,9 @@ * by the Agent-scope `runtimeService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { AgentPhase } from './runtime'; @@ -27,10 +28,17 @@ export const RuntimeModel = defineModel('runtime', () => ({ phase: { kind: 'idle' }, })); -export const setRuntimePhase = defineOp(RuntimeModel, 'runtime.set_phase', { +declare module '#/wire/types' { + interface TransientOpMap { + 'runtime.set_phase': typeof setRuntimePhase; + 'activity.set_snapshot': typeof setActivitySnapshot; + } +} + +export const setRuntimePhase = RuntimeModel.defineOp('runtime.set_phase', { + schema: z.object({ phase: z.custom() }), persist: false, - apply: (s, p: { phase: AgentPhase }): RuntimeModelState => - phaseEqual(s.phase, p.phase) ? s : { phase: p.phase }, + apply: (s, p) => (phaseEqual(s.phase, p.phase) ? s : { phase: p.phase }), toEvent: (p) => ({ type: 'agent.status.updated' as const, phase: p.phase }), }); @@ -105,10 +113,10 @@ export const ActivityModel = defineModel('activity', () = background: [], })); -export const setActivitySnapshot = defineOp(ActivityModel, 'activity.set_snapshot', { +export const setActivitySnapshot = ActivityModel.defineOp('activity.set_snapshot', { + schema: z.object({ next: z.custom() }), persist: false, - apply: (s, p: { next: AgentActivitySnapshot }): AgentActivitySnapshot => - snapshotEqual(s, p.next) ? s : p.next, + apply: (s, p) => (snapshotEqual(s, p.next) ? s : p.next), toEvent: (p) => ({ type: 'agent.activity.updated' as const, ...p.next }), }); diff --git a/packages/agent-core-v2/src/agent/skill/skillOps.ts b/packages/agent-core-v2/src/agent/skill/skillOps.ts index 48982bf5cc..69f019df76 100644 --- a/packages/agent-core-v2/src/agent/skill/skillOps.ts +++ b/packages/agent-core-v2/src/agent/skill/skillOps.ts @@ -12,8 +12,9 @@ * Consumed by the Agent-scope `skillService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { SkillActivationOrigin, SkillSource } from '#/agent/contextMemory/types'; @@ -32,9 +33,16 @@ declare module '#/app/event/eventBus' { export const SkillModel = defineModel('skill', () => null); -export const skillActivate = defineOp(SkillModel, 'skill.activate', { +declare module '#/wire/types' { + interface TransientOpMap { + 'skill.activate': typeof skillActivate; + } +} + +export const skillActivate = SkillModel.defineOp('skill.activate', { + schema: z.object({ origin: z.custom() }), persist: false, - apply: (s, _p: { origin: SkillActivationOrigin }) => s, + apply: (s) => s, toEvent: (p) => ({ type: 'skill.activated' as const, activationId: p.origin.activationId, diff --git a/packages/agent-core-v2/src/agent/swarm/swarmOps.ts b/packages/agent-core-v2/src/agent/swarm/swarmOps.ts index 840ab1fba2..3330615a4d 100644 --- a/packages/agent-core-v2/src/agent/swarm/swarmOps.ts +++ b/packages/agent-core-v2/src/agent/swarm/swarmOps.ts @@ -11,8 +11,9 @@ * Agent-scope `swarmService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { SwarmModeTrigger } from './swarm'; @@ -20,16 +21,19 @@ export const SwarmModel = defineModel('swarm', () => nu declare module '#/wire/types' { interface PersistedOpMap { - 'swarm_mode.exit': Record; + 'swarm_mode.enter': typeof swarmEnter; + 'swarm_mode.exit': typeof swarmExit; } } -export const swarmEnter = defineOp(SwarmModel, 'swarm_mode.enter', { - apply: (_s, p: { trigger: SwarmModeTrigger }) => p.trigger, +export const swarmEnter = SwarmModel.defineOp('swarm_mode.enter', { + schema: z.object({ trigger: z.custom() }), + apply: (_s, p) => p.trigger, toEvent: () => ({ type: 'agent.status.updated' as const, swarmMode: true }), }); -export const swarmExit = defineOp(SwarmModel, 'swarm_mode.exit', { +export const swarmExit = SwarmModel.defineOp('swarm_mode.exit', { + schema: z.object({}), apply: () => null, toEvent: () => ({ type: 'agent.status.updated' as const, swarmMode: false }), }); diff --git a/packages/agent-core-v2/src/agent/task/taskOps.ts b/packages/agent-core-v2/src/agent/task/taskOps.ts index 21e9bff4ec..d7055a1257 100644 --- a/packages/agent-core-v2/src/agent/task/taskOps.ts +++ b/packages/agent-core-v2/src/agent/task/taskOps.ts @@ -19,8 +19,9 @@ * Agent-scope `taskService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { AgentTaskInfo } from './types'; @@ -35,13 +36,19 @@ declare module '#/app/event/eventBus' { } } -export interface TaskInfoPayload { - readonly info: AgentTaskInfo; +const taskInfoSchema = z.object({ info: z.custom() }); + +declare module '#/wire/types' { + interface TransientOpMap { + 'task.started': typeof taskStarted; + 'task.terminated': typeof taskTerminated; + } } -export const taskStarted = defineOp(TaskModel, 'task.started', { +export const taskStarted = TaskModel.defineOp('task.started', { + schema: taskInfoSchema, persist: false, - apply: (s, p: TaskInfoPayload): TaskModelState => { + apply: (s, p) => { const next = new Map(s); next.set(p.info.taskId, p.info); return next; @@ -49,9 +56,10 @@ export const taskStarted = defineOp(TaskModel, 'task.started', { toEvent: (p) => ({ type: 'task.started' as const, info: p.info }), }); -export const taskTerminated = defineOp(TaskModel, 'task.terminated', { +export const taskTerminated = TaskModel.defineOp('task.terminated', { + schema: taskInfoSchema, persist: false, - apply: (s, p: TaskInfoPayload): TaskModelState => { + apply: (s, p) => { const next = new Map(s); next.set(p.info.taskId, p.info); return next; diff --git a/packages/agent-core-v2/src/agent/usage/usageOps.ts b/packages/agent-core-v2/src/agent/usage/usageOps.ts index a36b10daed..548b50baaf 100644 --- a/packages/agent-core-v2/src/agent/usage/usageOps.ts +++ b/packages/agent-core-v2/src/agent/usage/usageOps.ts @@ -14,10 +14,11 @@ * each dispatch (never on replay). Consumed by the Agent-scope `usageService`. */ +import { z } from 'zod'; + import { addUsage, type TokenUsage } from '#/app/llmProtocol/usage'; import type { AgentPhase } from '#/agent/runtime/runtime'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { UsageStatus } from './usage'; @@ -45,14 +46,19 @@ export interface UsageModelState { export const UsageModel = defineModel('usage', () => ({ byModel: {} })); -export interface UsageRecordPayload { - readonly model: string; - readonly usage: TokenUsage; - readonly usageScope?: UsageRecordScope; +declare module '#/wire/types' { + interface PersistedOpMap { + 'usage.record': typeof recordUsage; + } } -export const recordUsage = defineOp(UsageModel, 'usage.record', { - apply: (s, p: UsageRecordPayload): UsageModelState => { +export const recordUsage = UsageModel.defineOp('usage.record', { + schema: z.object({ + model: z.string(), + usage: z.custom(), + usageScope: z.custom().optional(), + }), + apply: (s, p) => { const current = s.byModel[p.model]; return { byModel: { diff --git a/packages/agent-core-v2/src/agent/userTool/userToolOps.ts b/packages/agent-core-v2/src/agent/userTool/userToolOps.ts index 876fe8b5fb..9b54e89634 100644 --- a/packages/agent-core-v2/src/agent/userTool/userToolOps.ts +++ b/packages/agent-core-v2/src/agent/userTool/userToolOps.ts @@ -18,8 +18,9 @@ * Consumed by the Agent-scope `userToolService`. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { UserToolRegistration } from './userTool'; @@ -27,6 +28,13 @@ export type UserToolModelState = Map; export const UserToolModel = defineModel('userTool', () => new Map()); +declare module '#/wire/types' { + interface PersistedOpMap { + 'tools.register_user_tool': typeof registerUserTool; + 'tools.unregister_user_tool': typeof unregisterUserTool; + } +} + function equalRegistration(a: UserToolRegistration, b: UserToolRegistration): boolean { return ( a.name === b.name && @@ -35,29 +43,23 @@ function equalRegistration(a: UserToolRegistration, b: UserToolRegistration): bo ); } -export const registerUserTool = defineOp( - UserToolModel, - 'tools.register_user_tool', - { - apply: (s, p: UserToolRegistration): UserToolModelState => { - const existing = s.get(p.name); - if (existing !== undefined && equalRegistration(existing, p)) return s; - const next = new Map(s); - next.set(p.name, p); - return next; - }, +export const registerUserTool = UserToolModel.defineOp('tools.register_user_tool', { + schema: z.custom(), + apply: (s, p) => { + const existing = s.get(p.name); + if (existing !== undefined && equalRegistration(existing, p)) return s; + const next = new Map(s); + next.set(p.name, p); + return next; }, -); - -export const unregisterUserTool = defineOp( - UserToolModel, - 'tools.unregister_user_tool', - { - apply: (s, p: { readonly name: string }): UserToolModelState => { - if (!s.has(p.name)) return s; - const next = new Map(s); - next.delete(p.name); - return next; - }, +}); + +export const unregisterUserTool = UserToolModel.defineOp('tools.unregister_user_tool', { + schema: z.object({ name: z.string() }), + apply: (s, p) => { + if (!s.has(p.name)) return s; + const next = new Map(s); + next.delete(p.name); + return next; }, -); +}); diff --git a/packages/agent-core-v2/src/agent/wireRecord/metadataOps.ts b/packages/agent-core-v2/src/agent/wireRecord/metadataOps.ts index 86872c115f..9e77139a53 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/metadataOps.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/metadataOps.ts @@ -9,17 +9,20 @@ * the same append path as every other Op. Scope-agnostic. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; const MetadataModel = defineModel('wire.metadata', () => null); -export interface WireMetadataPayload { - readonly protocol_version: string; - readonly created_at: number; +declare module '#/wire/types' { + interface PersistedOpMap { + metadata: typeof wireMetadata; + } } -export const wireMetadata = defineOp(MetadataModel, 'metadata', { +export const wireMetadata = MetadataModel.defineOp('metadata', { + schema: z.object({ protocol_version: z.string(), created_at: z.number() }), stamp: false, - apply: (s, _p: WireMetadataPayload): null => s, + apply: (s) => s, }); diff --git a/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts b/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts index 713cd4ff63..cb2f8b58c9 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts @@ -1,20 +1,9 @@ -import type { ContentPart } from '#/app/llmProtocol/message'; +import { createDecorator } from '#/_base/di/instantiation'; -import { createDecorator } from "#/_base/di/instantiation"; -import type { IDisposable } from "#/_base/di/lifecycle"; -import type { Event } from '#/_base/event'; - -import type { Hooks } from '#/hooks'; import type { WireMigrationRecord } from '#/agent/wireRecord/migration/migration'; export * from '#/agent/wireRecord/migration/migration'; -export interface WireRecordMap {} - -export type WireRecord = { - [T in K]: { readonly type: T; readonly time?: number } & Readonly; -}[K]; - export interface WireRecordMetadata { readonly type: 'metadata'; readonly protocol_version: string; @@ -22,16 +11,7 @@ export interface WireRecordMetadata { readonly time?: number; } -export type PersistedWireRecord = WireRecord | WireRecordMetadata | WireMigrationRecord; - -export interface WireRecordRestoringContext { - readonly time?: number; -} - -export interface WireRecordRestoredContext { - readonly record: WireRecord; - stop: boolean; -} +export type PersistedWireRecord = WireRecordMetadata | WireMigrationRecord; export interface WireRecordRestoreOptions { readonly rewriteMigratedRecords?: boolean; @@ -41,37 +21,9 @@ export interface WireRecordRestoreResult { readonly warning?: string; } -export interface WireRecordBlobTarget { - readonly parts: readonly ContentPart[]; - replace(record: TRecord, parts: readonly ContentPart[]): TRecord; -} - -export type WireRecordBlobSelector = ( - record: TRecord, -) => Iterable>; - -export interface WireRecordRegisterOptions { - readonly blobs?: WireRecordBlobSelector>; -} - -/** - * Static construction options for `AgentWireRecordService`, supplied through a - * `SyncDescriptor` when the service is seeded into a scope. Kept separate from - * injected services so each agent scope can pin its own persistence key. - */ -export interface WireRecordServiceOptions { - /** - * Per-agent home directory used to derive the wire-log persistence key. - * Falls back to `IBootstrapService.homeDir` (the global home) when omitted. - */ - readonly homedir?: string; -} - export interface IAgentWireRecordService { readonly _serviceBrand: undefined; - readonly restoring: WireRecordRestoringContext | null; - readonly postRestoring: boolean; /** * Snapshot of every record held in memory, in order, excluding the leading * `metadata` envelope: the records seeded by {@link restore} plus every record @@ -81,24 +33,12 @@ export interface IAgentWireRecordService { * transcript). */ getRecords(): readonly PersistedWireRecord[]; - register( - type: T, - resumer: (data: WireRecord) => void | Promise, - options?: WireRecordRegisterOptions, - ): IDisposable; restore( records?: readonly PersistedWireRecord[], options?: WireRecordRestoreOptions, ): Promise; flush(): Promise; close(): Promise; - - readonly hooks: Hooks<{ - onDidRestoreRecord: WireRecordRestoredContext; - }>; - - /** Fires once after a resume's replay pass has finished (live or restored). */ - readonly onDidFinishResume: Event; } export const IAgentWireRecordService = createDecorator('agentWireRecordService'); diff --git a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts index ff0c771b83..8fa579d9d7 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts @@ -1,16 +1,12 @@ import { relative } from 'pathe'; import { InstantiationType } from '#/_base/di/extensions'; +import { Disposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { Disposable, toDisposable } from "#/_base/di/lifecycle"; -import { Emitter, type Event } from '#/_base/event'; -import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { IBootstrapService } from '#/app/bootstrap/bootstrap'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; -import { OrderedHookSlot } from '#/hooks'; import { IAgentWireService } from '#/wire/tokens'; import type { IWireService } from '#/wire/wireService'; -import type { WireRecord, WireRecordMap } from './wireRecord'; import { AGENT_WIRE_PROTOCOL_VERSION, applyWireMigrations, @@ -22,39 +18,19 @@ import { import { IAgentWireRecordService, type PersistedWireRecord, - type WireRecordBlobSelector, type WireRecordMetadata, - type WireRecordRegisterOptions, - type WireRecordRestoredContext, type WireRecordRestoreOptions, type WireRecordRestoreResult, - type WireRecordServiceOptions, } from './wireRecord'; -type Resumer = (data: WireRecord) => void | Promise; -type BlobSelector = WireRecordBlobSelector>; - export class AgentWireRecordService extends Disposable implements IAgentWireRecordService { declare readonly _serviceBrand: undefined; - private readonly records: WireRecord[] = []; - private readonly resumers = new Map>>(); - private readonly blobSelectors = new Map< - keyof WireRecordMap, - BlobSelector[] - >(); + private readonly records: PersistedWireRecord[] = []; private readonly wireScope: string; - private _restoring: { time?: number } | null = null; - private _postRestoring = false; - readonly hooks = { - onDidRestoreRecord: new OrderedHookSlot(), - }; - private readonly _onDidFinishResume = this._register(new Emitter()); - readonly onDidFinishResume: Event = this._onDidFinishResume.event; constructor( - private readonly options: WireRecordServiceOptions = {}, + options: { readonly homedir?: string } = {}, @IBootstrapService bootstrap: IBootstrapService, - @IAgentBlobService private readonly blobStore?: IAgentBlobService, @IAppendLogStore private readonly log?: IAppendLogStore, @IAgentWireService private readonly wire?: IWireService, ) { @@ -79,48 +55,17 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco this._register( wire.onEmission((emission) => { if (emission.type === 'record' && emission.record.type !== 'metadata') { - this.records.push(emission.record as WireRecord); + this.records.push(emission.record as PersistedWireRecord); } }), ); } } - get restoring() { - return this._restoring; - } - - get postRestoring() { - return this._postRestoring; - } - getRecords(): readonly PersistedWireRecord[] { return [...this.records]; } - register( - type: T, - resumer: (data: WireRecord) => void | Promise, - options?: WireRecordRegisterOptions, - ) { - const typed = resumer as unknown as Resumer; - let set = this.resumers.get(type); - if (set === undefined) { - set = new Set(); - this.resumers.set(type, set); - } - set.add(typed); - const blobSelector = options?.blobs as BlobSelector | undefined; - const blobSet = this.registerBlobSelector(type, blobSelector); - return toDisposable(() => { - set?.delete(typed); - if (blobSelector !== undefined) { - const index = blobSet?.indexOf(blobSelector) ?? -1; - if (index !== -1) blobSet?.splice(index, 1); - } - }); - } - async restore( records?: readonly PersistedWireRecord[], options: WireRecordRestoreOptions = {}, @@ -132,7 +77,6 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco ? this.log.read(this.wireScope, WIRE_RECORD_FILENAME) : undefined); if (source === undefined) { - this.fireResumeEnded(); return {}; } @@ -144,11 +88,10 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco fromPersistence && this.log !== undefined; let migrations: readonly WireMigration[] = []; let shouldRewrite = false; - let completed = true; let warning: string | undefined; const sourceRecords: PersistedWireRecord[] = []; - for await (const record of toAsyncIterable(source)) { + for await (const record of source) { sourceRecords.push(record); } @@ -184,31 +127,19 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco } restoredRecords?.push(migratedRecord); if (migratedRecord.type === 'metadata') continue; - - if (await this.restoreRecord(await this.rehydrateRecord(migratedRecord as WireRecord))) { - completed = false; - break; - } + this.records.push(migratedRecord); } - if ( - completed && - shouldRewrite && - restoredRecords !== undefined && - this.log !== undefined - ) { + if (shouldRewrite && restoredRecords !== undefined && this.log !== undefined) { void this.log.rewrite(this.wireScope, WIRE_RECORD_FILENAME, restoredRecords); await this.log.flush(); } - if (completed) { - this.fireResumeEnded(); - } return warning === undefined ? {} : { warning }; } async flush(): Promise { - // Drain the wire service's async persist pipeline first: with a blob - // service registered, appends are queued on a microtask chain and only + // Drain the wire service's async persist pipeline first: with a model blob + // codec, appends are queued on a microtask chain and only // reach the log store once that queue settles. Flushing the log alone // would miss records still in flight. await this.wire?.flush(); @@ -218,84 +149,6 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco async close(): Promise { await this.log?.close(); } - - private async restoreRecord(record: WireRecord): Promise { - this.records.push(record); - this._restoring = { time: record.time ?? Date.now() }; - try { - const resumers = this.resumers.get(record.type); - if (resumers !== undefined) { - const currentResumers = Array.from(resumers); - for (const resumer of currentResumers) { - await resumer(record); - } - } - const context: WireRecordRestoredContext = { record, stop: false }; - await this.hooks.onDidRestoreRecord.run(context); - return context.stop; - } finally { - this._restoring = null; - } - } - - private fireResumeEnded(): void { - this._postRestoring = true; - try { - this._onDidFinishResume.fire(); - } finally { - this._postRestoring = false; - } - } - - private registerBlobSelector( - type: T, - selector: BlobSelector | undefined, - ): BlobSelector[] | undefined { - if (selector === undefined) return undefined; - - let selectors = this.blobSelectors.get(type); - if (selectors === undefined) { - selectors = []; - this.blobSelectors.set(type, selectors); - } - selectors.push(selector); - return selectors; - } - - private async rehydrateRecord( - record: WireRecord, - ): Promise> { - return this.applyBlobSelectors(record); - } - - private async applyBlobSelectors( - record: WireRecord, - ): Promise> { - const blobStore = this.blobStore; - if (blobStore === undefined) return record; - - const selectors = this.blobSelectors.get(record.type); - if (selectors === undefined) return record; - - let current = record; - for (const selector of [...selectors] as BlobSelector[]) { - for (const target of selector(current)) { - const parts = await blobStore.loadParts(target.parts); - if (parts !== target.parts) { - current = target.replace(current, parts); - } - } - } - return current; - } -} - -async function* toAsyncIterable( - source: Iterable | AsyncIterable, -): AsyncIterable { - for await (const item of source) { - yield item; - } } registerScopedService( diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index 67b5113b0a..c45956bd3b 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -66,8 +66,9 @@ import { AgentWireRecordService, WIRE_RECORD_FILENAME, } from '#/agent/wireRecord/wireRecordService'; -import { type WireMetadataPayload, wireMetadata } from '#/agent/wireRecord/metadataOps'; +import { wireMetadata } from '#/agent/wireRecord/metadataOps'; import { IAgentWireService } from '#/wire/tokens'; +import type { PayloadOf } from '#/wire/types'; import { WireService } from '#/wire/wireServiceImpl'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { AgentBlobServiceImpl } from '#/agent/blob/agentBlobServiceImpl'; @@ -487,7 +488,7 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle } } -function freshMetadataPayload(): WireMetadataPayload { +function freshMetadataPayload(): PayloadOf { return { protocol_version: AGENT_WIRE_PROTOCOL_VERSION, created_at: Date.now(), diff --git a/packages/agent-core-v2/src/session/cron/cronOps.ts b/packages/agent-core-v2/src/session/cron/cronOps.ts index 11f0e64a86..9936798ff3 100644 --- a/packages/agent-core-v2/src/session/cron/cronOps.ts +++ b/packages/agent-core-v2/src/session/cron/cronOps.ts @@ -18,9 +18,9 @@ */ import type { CronJobOrigin } from '@moonshot-ai/protocol'; +import { z } from 'zod'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import type { CronTask } from '#/app/cron/cronTask'; @@ -34,28 +34,16 @@ declare module '#/app/event/eventBus' { } } -export interface CronAddPayload { - readonly task: CronTask; -} - -export interface CronDeletePayload { - readonly ids: readonly string[]; -} - -export interface CronCursorPayload { - readonly id: string; - readonly lastFiredAt: number; -} - declare module '#/wire/types' { interface TransientOpMap { - 'cron.add': CronAddPayload; - 'cron.delete': CronDeletePayload; - 'cron.cursor': CronCursorPayload; + 'cron.add': typeof cronAdd; + 'cron.delete': typeof cronDelete; + 'cron.cursor': typeof cronCursor; } } -export const cronAdd = defineOp(CronModel, 'cron.add', { +export const cronAdd = CronModel.defineOp('cron.add', { + schema: z.object({ task: z.custom() }), persist: false, apply: (s, p) => { const next = new Map(s); @@ -64,7 +52,8 @@ export const cronAdd = defineOp(CronModel, 'cron.add', { }, }); -export const cronDelete = defineOp(CronModel, 'cron.delete', { +export const cronDelete = CronModel.defineOp('cron.delete', { + schema: z.object({ ids: z.array(z.string()).readonly() }), persist: false, apply: (s, p) => { let next: Map | undefined; @@ -78,7 +67,8 @@ export const cronDelete = defineOp(CronModel, 'cron.delete', { }, }); -export const cronCursor = defineOp(CronModel, 'cron.cursor', { +export const cronCursor = CronModel.defineOp('cron.cursor', { + schema: z.object({ id: z.string(), lastFiredAt: z.number() }), persist: false, apply: (s, p) => { const task = s.get(p.id); diff --git a/packages/agent-core-v2/src/session/todo/todoOps.ts b/packages/agent-core-v2/src/session/todo/todoOps.ts index 7f690da2e0..148d8346bf 100644 --- a/packages/agent-core-v2/src/session/todo/todoOps.ts +++ b/packages/agent-core-v2/src/session/todo/todoOps.ts @@ -17,8 +17,9 @@ * replays. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import { readTodoItems, type TodoItem } from './todoItem'; @@ -26,17 +27,13 @@ export type TodoModelState = readonly TodoItem[]; export const TodoModel = defineModel('todo', () => []); -export interface ToolStoreUpdatePayload { - readonly key: string; - readonly value: unknown; -} - declare module '#/wire/types' { interface PersistedOpMap { - 'tools.update_store': ToolStoreUpdatePayload; + 'tools.update_store': typeof todoSet; } } -export const todoSet = defineOp(TodoModel, 'tools.update_store', { +export const todoSet = TodoModel.defineOp('tools.update_store', { + schema: z.object({ key: z.string(), value: z.unknown() }), apply: (s, p) => (p.key === 'todo' ? readTodoItems(p.value) : s), }); diff --git a/packages/agent-core-v2/src/wire/model.ts b/packages/agent-core-v2/src/wire/model.ts index 87eb72af94..d387b273da 100644 --- a/packages/agent-core-v2/src/wire/model.ts +++ b/packages/agent-core-v2/src/wire/model.ts @@ -5,8 +5,10 @@ * dehydrate large inline media before persistence and rehydrate blob references * in its state after replay. * - * A `ModelDef` is a stateless descriptor: it names a model and manufactures its - * initial state via `initial`. It never holds state itself — per-scope state + * A `ModelDef` is a stateless descriptor: it names a model, manufactures its + * initial state via `initial`, and declares the model's Ops through + * `defineOp` (the model-bound form of the primitive in `op.ts`). It never + * holds state itself — per-scope state * instances are owned by `IWireService`, and domain services read them through * `wire.getModel(model)`. The optional `blobs` codec declares both directions * of the blob offload pipeline: @@ -38,7 +40,9 @@ * applied by `WireService` after every `apply`. Scope-agnostic. */ -import type { PersistedRecord } from './wireService'; +import { bindDefineOp, type DefineOpFn } from '#/wire/op'; +import type { ModelReducers } from '#/wire/types'; +import type { PersistedRecord } from '#/wire/wireService'; export type PartsTransformer = (parts: readonly unknown[]) => Promise; @@ -51,6 +55,11 @@ export interface ModelDef { readonly name: string; readonly initial: () => S; readonly blobs?: ModelBlobCodec; + /** + * Declare an Op on this model — `defineOp(model, ...)` with the model + * bound. Preferred call style: `MyModel.defineOp('my.op', { apply })`. + */ + readonly defineOp: DefineOpFn; } export interface ModelCrossReducerEntry { @@ -67,13 +76,18 @@ export function defineModel( initial: () => S, opts?: { blobs?: ModelBlobCodec; - // eslint-disable-next-line @typescript-eslint/no-explicit-any - reducers?: Record S>; + reducers?: ModelReducers; }, ): ModelDef { - const def: ModelDef = { name, initial, blobs: opts?.blobs }; + const def: ModelDef = { + name, + initial, + blobs: opts?.blobs, + defineOp: bindDefineOp(() => def), + }; if (opts?.reducers !== undefined) { for (const [opType, reducer] of Object.entries(opts.reducers)) { + if (reducer === undefined) continue; let list = MODEL_CROSS_REDUCERS.get(opType); if (list === undefined) { list = []; @@ -88,21 +102,19 @@ export function defineModel( export interface DerivedModelDef { readonly name: string; readonly initial: () => S; - // eslint-disable-next-line @typescript-eslint/no-explicit-any - readonly reducers: Readonly S>>; + readonly reducers: Readonly>; readonly blobs?: ModelBlobCodec; } export function defineDerivedModel( name: string, initial: () => S, - reducers: Record S>, + reducers: ModelReducers, opts?: { blobs?: ModelBlobCodec }, ): DerivedModelDef { return { name, initial, reducers, blobs: opts?.blobs }; } - export type DeepReadonly = T extends (...args: infer A) => infer R ? (...args: A) => R : T extends ReadonlyMap diff --git a/packages/agent-core-v2/src/wire/op.ts b/packages/agent-core-v2/src/wire/op.ts index 2a871245e3..20dbcf002a 100644 --- a/packages/agent-core-v2/src/wire/op.ts +++ b/packages/agent-core-v2/src/wire/op.ts @@ -9,20 +9,30 @@ * `goalCreate.type`). Every Op carries a mandatory pure `apply` and may carry * an optional `toEvent` that derives an `IEventBus` fact from the payload and * the post-apply state (published by `WireService` on `dispatch`, never on - * `replay`). The descriptor's payload is erased to `any` on `Op.descriptor` (mirroring - * `OP_REGISTRY`) so `Op` stays covariant in `P` — a heterogeneous batch of Ops, - * each with a different payload type, stays assignable to the single - * `dispatch(...ops: Op[])` rest parameter, while the precise payload type - * survives on `Op.payload` for the Op's own caller. Registering a duplicate - * `type` throws `DuplicateOpError` so the global Op-type namespace stays unique. - * Descriptors may opt out of persistence (`persist: false`) for live-only - * state, or opt out of timestamp stamping (`stamp: false`) for the metadata - * envelope. Both default to the v1-compatible persisted, stamped path. - * Scope-agnostic. + * `replay`). A mandatory `schema` (zod, declared before `apply`) is the + * payload's single source of truth: `P` is inferred from it, so Op authors + * never restate payload interfaces, and it is stored on the descriptor for + * payload validation at wire boundaries; the runtime paths (`dispatch` / + * `replay`) never consult it. The descriptor's payload is erased + * to `any` on `Op.descriptor` (mirroring `OP_REGISTRY`) so `Op` stays + * covariant in `P` — a heterogeneous batch of Ops, each with a different + * payload type, stays assignable to the single `dispatch(...ops: Op[])` rest + * parameter, while the precise payload type survives on `Op.payload` for the + * Op's own caller. Registering a duplicate `type` throws `DuplicateOpError` so + * the global Op-type namespace stays unique. Payloads flow from each Op + * definition into the `types.ts` registries (which map op types to `typeof` + * the Op); registration constrains only the persistence policy — a registered + * type must honor its map, an unregistered type keeps its free `persist` + * option. Descriptors may opt out of timestamp stamping (`stamp: false`) for + * the metadata envelope. Scope-agnostic. */ -import type { ModelDef } from './model'; +import type { z } from 'zod'; + +import type { ConflictingOpType, OpPersistenceOptions, OpType } from '#/wire/types'; + import { WireError, WireErrors } from './errors'; +import type { ModelDef } from './model'; export class DuplicateOpError extends WireError { constructor(readonly type: string) { @@ -36,6 +46,13 @@ export class DuplicateOpError extends WireError { export interface OpDescriptor { readonly type: K; readonly model: ModelDef; + /** + * Zod schema for the payload — the payload type's single source of truth + * (`P` is inferred from it). Stored on the descriptor so wire boundaries + * (replay of `wire.jsonl`, record export) can validate payloads against the + * Op's declared shape. Not consulted by `dispatch` / `replay` themselves. + */ + readonly schema: z.ZodType

; readonly apply: (state: S, payload: P) => S; /** * Optional fact derivation: when present, `WireService` publishes the @@ -57,32 +74,91 @@ export interface Op { readonly type: K; readonly payload: P; // eslint-disable-next-line @typescript-eslint/no-explicit-any - readonly descriptor: OpDescriptor; + readonly descriptor: OpDescriptor; } // eslint-disable-next-line @typescript-eslint/no-explicit-any export const OP_REGISTRY = new Map>(); -export function defineOp( +interface OpBehaviorOptions { + readonly schema: z.ZodType

; + readonly apply: (state: S, payload: P) => S; + readonly toEvent?: (payload: P, state: S) => unknown; + readonly stamp?: boolean; +} + +/** + * Registry-derived constraint on a defined Op's options. A type registered in + * both maps is rejected outright; a registered type must honor its map's + * persistence policy (persisted Ops may not opt out, transient Ops must pass + * `persist: false`). Key-level only — never resolves the registry's member + * types, so Op definitions stay free of registry cycles. + */ +type RegisteredOpConstraint = K extends ConflictingOpType + ? never + : K extends OpType + ? OpPersistenceOptions + : unknown; + +type DefineOpOptions = OpBehaviorOptions & { + readonly persist?: boolean; +} & RegisteredOpConstraint; + +type DefinedOp = OpDescriptor & + ((payload: P) => Op); + +/** + * Call signature of `ModelDef.defineOp` — `defineOp` with the model bound. + * Lives here so `model.ts` can type the method without duplicating the + * registry-aware generics. + */ +export interface DefineOpFn { + ( + type: K & SingleStringLiteral, + opts: DefineOpOptions, S, P>, + ): DefinedOp; +} + +type SingleStringLiteral = {} extends Record + ? never + : K extends unknown + ? [Whole] extends [K] + ? K + : never + : never; + +/** + * Build `ModelDef.defineOp` for a model under construction. The getter defers + * the model read so `defineModel` can bind while the literal is initializing. + * The casts bypass TS's inability to re-prove the literal guard + * (`SingleStringLiteral`) on an already-validated abstract `K`; callers still + * get the full guard through `DefineOpFn`'s signature. + */ +export function bindDefineOp(getModel: () => ModelDef): DefineOpFn { + const bound = (type: string, opts: unknown): unknown => + defineOp(getModel(), type as never, opts as never); + return bound as DefineOpFn; +} + +export function defineOp( model: ModelDef, - type: K, - opts: { - apply: (state: S, payload: P) => S; - toEvent?: (payload: P, state: S) => unknown; - persist?: boolean; - stamp?: boolean; - }, -): OpDescriptor & ((payload: P) => Op) { + type: K & SingleStringLiteral, + opts: DefineOpOptions, S, P>, +): DefinedOp { if (OP_REGISTRY.has(type)) { throw new DuplicateOpError(type); } + const behavior: OpBehaviorOptions & { + readonly persist?: boolean; + } = opts; const descriptor: OpDescriptor = { type, model, - apply: opts.apply, - toEvent: opts.toEvent, - persist: opts.persist, - stamp: opts.stamp, + schema: behavior.schema, + apply: behavior.apply, + toEvent: behavior.toEvent, + persist: behavior.persist, + stamp: behavior.stamp, }; OP_REGISTRY.set(type, descriptor); const factory = (payload: P): Op => ({ type, payload, descriptor }); diff --git a/packages/agent-core-v2/src/wire/types.ts b/packages/agent-core-v2/src/wire/types.ts new file mode 100644 index 0000000000..8704b4b687 --- /dev/null +++ b/packages/agent-core-v2/src/wire/types.ts @@ -0,0 +1,47 @@ +/** + * `wire` domain (L2) — augmentable Op registries and their derived + * compile-time vocabulary. + * + * Domains contribute their defined Ops to `PersistedOpMap` or `TransientOpMap` + * via module augmentation (`'my.op': typeof myOp`). The selected map + * classifies whether a live dispatch writes the Op, while `OpPayload` recovers + * each Op's payload from the Op's own type: the payload flows from the Op + * definition into the registry, never the reverse, so Op authoring stays free + * of registry cycles. Persisted input remains an open wire boundary so replay + * can continue to tolerate historical and newer record types. Scope-agnostic. + */ + +export interface PersistedOpMap {} + +export interface TransientOpMap {} + +type StringKey = Extract; + +type PersistedOpKey = StringKey; +type TransientOpKey = StringKey; + +// Everything here is key-level: the maps' member types (`typeof` an Op) are +// resolved only by `OpPayload`, never by the classification aliases — an +// intersection of the maps would normalize members and re-enter Op +// definitions, forming a type cycle. +export type ConflictingOpType = Extract; +export type PersistedOpType = Exclude; +export type TransientOpType = Exclude; +export type OpType = PersistedOpType | TransientOpType; + +/** Payload carried by a defined Op (the result of `Model.defineOp(...)`). */ +export type PayloadOf = T extends (payload: infer P) => unknown ? P : never; + +export type OpPayload = K extends PersistedOpType + ? PayloadOf + : K extends TransientOpType + ? PayloadOf + : never; + +export type ModelReducers = { + [K in OpType]?: (state: S, payload: OpPayload) => S; +}; + +export type OpPersistenceOptions = K extends PersistedOpType + ? { readonly persist?: true } + : { readonly persist: false }; diff --git a/packages/agent-core-v2/src/wire/wireService.ts b/packages/agent-core-v2/src/wire/wireService.ts index 8601d6d526..71b3cb1bdd 100644 --- a/packages/agent-core-v2/src/wire/wireService.ts +++ b/packages/agent-core-v2/src/wire/wireService.ts @@ -13,7 +13,7 @@ * `tokens`, each seeded with its own persistence key. `PersistedRecord` is the * on-the-wire append-log shape (`wire.jsonl`): intentionally flat * (`{ type, ...payload }`, optional `time`) so it stays byte-compatible with the - * existing `WireRecord` journal (`{ type, time?, ...fields }`) — payload fields + * existing wire journal (`{ type, time?, ...fields }`) — payload fields * sit at the top level next to `type`, never nested under a `payload` key; the * index signature keeps it scope-agnostic and domains narrow via their Op * payload types. Scope-agnostic. diff --git a/packages/agent-core-v2/src/wire/wireServiceImpl.ts b/packages/agent-core-v2/src/wire/wireServiceImpl.ts index 9e987e35dc..9f908de139 100644 --- a/packages/agent-core-v2/src/wire/wireServiceImpl.ts +++ b/packages/agent-core-v2/src/wire/wireServiceImpl.ts @@ -175,6 +175,7 @@ export class WireService extends Disposable implements IWireService { this.derivedModels.set(model, inst); for (const [opType, reducer] of Object.entries(model.reducers)) { + if (reducer === undefined) continue; let list = this.reducerIndex.get(opType); if (list === undefined) { list = []; diff --git a/packages/agent-core-v2/test/agent/contextMemory/stubs.ts b/packages/agent-core-v2/test/agent/contextMemory/stubs.ts index 2426303c35..63c0516659 100644 --- a/packages/agent-core-v2/test/agent/contextMemory/stubs.ts +++ b/packages/agent-core-v2/test/agent/contextMemory/stubs.ts @@ -7,10 +7,7 @@ * `../contextMemory/stubs`). */ -import { toDisposable } from '#/_base/di/lifecycle'; import type { ServiceRegistration } from '#/_base/di/test'; -import { Event } from '#/_base/event'; -import { createHooks } from '#/hooks'; import { buildContextCompactionShape } from '#/agent/contextMemory/compactionHandoff'; import { IAgentContextMemoryService, @@ -24,20 +21,10 @@ import { IEventBus } from '#/app/event/eventBus'; import { EventBusService } from '#/app/event/eventBusService'; import { IAgentWireRecordService } from '#/agent/wireRecord/wireRecord'; -/** - * A no-op `IAgentWireRecordService`. `register` returns a disposable so services that - * `_register(wireRecord.register(...))` in their constructor can be disposed - * cleanly. - */ +/** A no-op `IAgentWireRecordService`. */ export function stubWireRecord(): IAgentWireRecordService { - const hooks = createHooks(['onDidRestoreRecord']) as IAgentWireRecordService['hooks']; return { _serviceBrand: undefined, - restoring: null, - postRestoring: false, - hooks, - onDidFinishResume: Event.None as Event, - register: () => toDisposable(() => {}), restore: () => Promise.resolve({}), flush: () => Promise.resolve(), close: () => Promise.resolve(), diff --git a/packages/agent-core-v2/test/agent/goal/goal.test.ts b/packages/agent-core-v2/test/agent/goal/goal.test.ts index 8ac12d5b74..a1047afcec 100644 --- a/packages/agent-core-v2/test/agent/goal/goal.test.ts +++ b/packages/agent-core-v2/test/agent/goal/goal.test.ts @@ -10,7 +10,7 @@ import { UpdateGoalTool, UpdateGoalToolInputSchema } from '#/agent/goal/tools/up import { IAgentLoopService, type AfterStepContext, type Turn } from '#/agent/loop/loop'; import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; import { IAgentUsageService } from '#/agent/usage/usage'; -import type { PersistedWireRecord, WireRecord } from '#/agent/wireRecord/wireRecord'; +import type { PersistedWireRecord } from '#/agent/wireRecord/wireRecord'; import { type DomainEvent, IEventBus } from '#/app/event/eventBus'; import { APIConnectionError, APIStatusError } from '#/app/llmProtocol/errors'; import type { ToolCall } from '#/app/llmProtocol/message'; @@ -31,7 +31,7 @@ import { recordingTelemetry, type TelemetryRecord } from '../../app/telemetry/st import { stubLoopWithHooks, type StubLoop } from '../loop/stubs'; type GoalServiceTestManager = IAgentGoalService & AgentGoalService; -type GoalRecord = Extract; +type GoalRecord = PersistedWireRecord & { type: `goal.${string}` }; type AgentEvent = DomainEvent; type GoalUpdatedEvent = Extract; type TurnEndedInput = { @@ -53,7 +53,7 @@ function goalRecords(records: readonly PersistedWireRecord[]): readonly GoalReco async function restoreGoalRecords( ctx: TestAgentContext, goals: IAgentGoalService, - records: readonly WireRecord[], + records: readonly PersistedWireRecord[], ): Promise { goals.getGoal(); await ctx.restore(records as readonly PersistedWireRecord[]); diff --git a/packages/agent-core-v2/test/app/sessionExport/sessionExport.test.ts b/packages/agent-core-v2/test/app/sessionExport/sessionExport.test.ts index 3ef81c699a..5d14bb8fd1 100644 --- a/packages/agent-core-v2/test/app/sessionExport/sessionExport.test.ts +++ b/packages/agent-core-v2/test/app/sessionExport/sessionExport.test.ts @@ -436,16 +436,9 @@ function stubAgentLifecycle(agents: readonly IAgentScopeHandle[]): IAgentLifecyc function stubAgentWire(flush: () => Promise = async () => {}): IAgentWireRecordService { return { _serviceBrand: undefined, - restoring: null, - postRestoring: false, getRecords: () => [], - register: () => noopDisposable, restore: async () => ({}), flush, close: async () => {}, - hooks: { - onDidRestoreRecord: { run: async () => {} }, - } as unknown as IAgentWireRecordService['hooks'], - onDidFinishResume: noopEvent, }; } diff --git a/packages/agent-core-v2/test/lint/fixtures/duplicate-ops.fixture.ts b/packages/agent-core-v2/test/lint/fixtures/duplicate-ops.fixture.ts index 7fcb93a2e5..65822696f5 100644 --- a/packages/agent-core-v2/test/lint/fixtures/duplicate-ops.fixture.ts +++ b/packages/agent-core-v2/test/lint/fixtures/duplicate-ops.fixture.ts @@ -5,10 +5,11 @@ * here — it exists purely to prove the scanner flags a planted duplicate. */ +import { z } from 'zod'; + import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; const FixtureModel = defineModel('fixture', () => ({})); -defineOp(FixtureModel, 'fixture.planted', { apply: (s) => s }); -defineOp(FixtureModel, 'fixture.planted', { apply: (s) => s }); +FixtureModel.defineOp('fixture.planted', { schema: z.object({}), apply: (s) => s }); +FixtureModel.defineOp('fixture.planted', { schema: z.object({}), apply: (s) => s }); diff --git a/packages/agent-core-v2/test/lint/op-uniqueness.test.ts b/packages/agent-core-v2/test/lint/op-uniqueness.test.ts index fc833f4bc0..f25dd09669 100644 --- a/packages/agent-core-v2/test/lint/op-uniqueness.test.ts +++ b/packages/agent-core-v2/test/lint/op-uniqueness.test.ts @@ -2,16 +2,132 @@ import { readdirSync, readFileSync, statSync } from 'node:fs'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { describe, expect, it } from 'vitest'; +import { z } from 'zod'; import { defineModel } from '#/wire/model'; -import { DuplicateOpError, defineOp } from '#/wire/op'; +import { DuplicateOpError, OP_REGISTRY } from '#/wire/op'; +import type { OpPayload } from '#/wire/types'; + +declare module '#/wire/types' { + interface PersistedOpMap { + 'test.op.persisted': typeof persistedOp; + 'test.op.conflicting': typeof persistedOp; + } + + interface TransientOpMap { + 'test.op.transient': typeof transientOp; + 'test.op.conflicting': typeof persistedOp; + } +} const __dirname = dirname(fileURLToPath(import.meta.url)); const PKG_ROOT = join(__dirname, '..', '..'); const SRC_ROOT = join(PKG_ROOT, 'src'); const FIXTURE_ROOT = join(__dirname, 'fixtures'); -const DEFINE_OP_RE = /defineOp\s*\(\s*\w+\s*,\s*['"]([^'"]+)['"]/g; +const DEFINE_OP_RE = /\.defineOp\s*\(\s*['"]([^'"]+)['"]/g; + +const testModel = defineModel('typecheck.model', () => ({ value: 0 })); + +const persistedOp = testModel.defineOp('test.op.persisted', { + schema: z.object({ value: z.number() }), + apply: (state, payload: { value: number }) => ({ value: state.value + payload.value }), +}); + +const transientOp = testModel.defineOp('test.op.transient', { + schema: z.object({ value: z.number() }), + persist: false, + apply: (_state, payload) => ({ value: payload.value }), +}); + +function typecheckRegisteredOps(): void { + // The registry recovers each Op's payload from the Op's own type. + type RegisteredPayload = OpPayload<'test.op.persisted'>; + const registeredPayload: RegisteredPayload = { value: 1 }; + persistedOp(registeredPayload); + // @ts-expect-error Op factories carry the payload from their own definition + persistedOp({ value: '1' }); + // @ts-expect-error transient Op factories carry the payload from their own definition + transientOp({ value: '1' }); + + const unregistered = testModel.defineOp('test.op.unregistered', { + schema: z.object({ by: z.number() }), + persist: false, + apply: (state, payload) => ({ + value: state.value + payload.by, + }), + toEvent: (payload) => ({ by: payload.by }), + }); + unregistered({ by: 1 }); + // @ts-expect-error unregistered Op factories retain their inferred payload + unregistered({ by: '1' }); + + const wrongPayloadSchema = z.object({ value: z.string() }); + testModel.defineOp('test.op.unregistered', { + // @ts-expect-error schemas must produce the Op's payload type + schema: wrongPayloadSchema, + apply: (state, payload: { value: number }) => ({ value: state.value + payload.value }), + }); + + const incorrectlyTransient = { + schema: z.object({ value: z.number() }), + persist: false as const, + apply: (state: { value: number }, payload: { value: number }) => ({ + value: state.value + payload.value, + }), + }; + // @ts-expect-error persisted Op types cannot opt out of persistence + testModel.defineOp('test.op.persisted', incorrectlyTransient); + + const missingTransientMarker = { + schema: z.object({ value: z.number() }), + apply: (state: { value: number }, payload: { value: number }) => ({ + value: state.value + payload.value, + }), + }; + // @ts-expect-error transient Op types require persist: false + testModel.defineOp('test.op.transient', missingTransientMarker); + + const incorrectlyPersistedTransient = { + schema: z.object({}), + persist: true as const, + apply: (state: { value: number }) => state, + }; + // @ts-expect-error transient Op types cannot opt into persistence + testModel.defineOp('test.op.transient', incorrectlyPersistedTransient); + + // @ts-expect-error the same Op type cannot belong to both registries + testModel.defineOp('test.op.conflicting', { + schema: z.object({ value: z.number() }), + apply: (_state: { value: number }, payload: { value: number }) => ({ + value: payload.value, + }), + }); + + const dynamicType: string = 'test.op.persisted'; + // @ts-expect-error Op definitions require a literal type so registry constraints cannot be bypassed + testModel.defineOp(dynamicType, { schema: z.object({}), apply: (state) => state }); + + const unionType = 'test.op.persisted' as 'test.op.persisted' | 'test.op.unregistered'; + // @ts-expect-error Op definitions reject unions that could mix registered and legacy types + testModel.defineOp(unionType, { schema: z.object({}), apply: (state) => state }); + + const templateType = 'test.op.persisted' as `test.op.${string}`; + // @ts-expect-error Op definitions reject non-literal template string types + testModel.defineOp(templateType, { + schema: z.object({}), + apply: (state: { value: number }) => state, + }); + + const brandedType = 'test.op.persisted' as string & { readonly opType: unique symbol }; + // @ts-expect-error Op definitions reject branded strings that could hide a registered type + testModel.defineOp(brandedType, { + schema: z.object({}), + apply: (state: { value: number }) => state, + }); +} + +void typecheckRegisteredOps; function walk(dir: string): string[] { const out: string[] = []; @@ -54,9 +170,14 @@ function duplicates(seen: Map): Map { describe('op-uniqueness', () => { it('defineOp throws DuplicateOpError when a type is registered twice', () => { const model = defineModel('lint.model', () => ({})); - const type = `lint.dup.${Date.now()}`; - defineOp(model, type, { apply: (s) => s }); - expect(() => defineOp(model, type, { apply: (s) => s })).toThrow(DuplicateOpError); + try { + model.defineOp('lint.duplicate.runtime', { schema: z.object({}), apply: (s) => s }); + expect(() => + model.defineOp('lint.duplicate.runtime', { schema: z.object({}), apply: (s) => s }), + ).toThrow(DuplicateOpError); + } finally { + OP_REGISTRY.delete('lint.duplicate.runtime'); + } }); it('finds no duplicate defineOp types across src/', () => { diff --git a/packages/agent-core-v2/test/wire/store-event.test.ts b/packages/agent-core-v2/test/wire/store-event.test.ts index e44d56ae5a..358774c2da 100644 --- a/packages/agent-core-v2/test/wire/store-event.test.ts +++ b/packages/agent-core-v2/test/wire/store-event.test.ts @@ -1,4 +1,5 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { z } from 'zod'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; @@ -10,7 +11,6 @@ import { InMemoryStorageService } from '#/persistence/backends/memory/inMemorySt import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import { IAgentWireService } from '#/wire/tokens'; import type { IWireService, PersistedRecord } from '#/wire/wireService'; import { WireService } from '#/wire/wireServiceImpl'; @@ -28,19 +28,23 @@ const KEY = 'store-event-test'; const CounterModel = defineModel('store-event.counter', () => ({ value: 0 })); const OtherModel = defineModel('store-event.other', () => ({ value: 0 })); -const addWithEvent = defineOp(CounterModel, 'store-event.counter.add', { - apply: (s, p: { by: number }) => ({ value: s.value + p.by }), +const addWithEvent = CounterModel.defineOp('store-event.counter.add', { + schema: z.object({ by: z.number() }), + apply: (s, p) => ({ value: s.value + p.by }), toEvent: (_p, state) => ({ type: 'store-event.added' as const, value: state.value }), }); -const addNoEvent = defineOp(CounterModel, 'store-event.counter.addNoEvent', { - apply: (s, p: { by: number }) => ({ value: s.value + p.by }), +const addNoEvent = CounterModel.defineOp('store-event.counter.addNoEvent', { + schema: z.object({ by: z.number() }), + apply: (s, p) => ({ value: s.value + p.by }), }); -const addUndefinedEvent = defineOp(CounterModel, 'store-event.counter.addUndef', { - apply: (s, p: { by: number }) => ({ value: s.value + p.by }), +const addUndefinedEvent = CounterModel.defineOp('store-event.counter.addUndef', { + schema: z.object({ by: z.number() }), + apply: (s, p) => ({ value: s.value + p.by }), toEvent: () => undefined, }); -const otherSet = defineOp(OtherModel, 'store-event.other.set', { - apply: (_s, p: { value: number }) => ({ value: p.value }), +const otherSet = OtherModel.defineOp('store-event.other.set', { + schema: z.object({ value: z.number() }), + apply: (_s, p) => ({ value: p.value }), toEvent: (p) => ({ type: 'store-event.otherSet' as const, value: p.value }), }); diff --git a/packages/agent-core-v2/test/wire/wire-compat.test.ts b/packages/agent-core-v2/test/wire/wire-compat.test.ts index 06b1f3a9c8..0498d5fbc3 100644 --- a/packages/agent-core-v2/test/wire/wire-compat.test.ts +++ b/packages/agent-core-v2/test/wire/wire-compat.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, describe, expect, it } from 'vitest'; +import { z } from 'zod'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; @@ -14,7 +15,6 @@ import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageSe import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import { IAgentWireService } from '#/wire/tokens'; import type { PersistedRecord } from '#/wire/wireService'; import { WireService } from '#/wire/wireServiceImpl'; @@ -25,11 +25,13 @@ const KEY = 'round-trip'; const CounterModel = defineModel('compat.counter', () => ({ value: 0 })); const TagsModel = defineModel('compat.tags', () => ({ tags: [] as string[] })); -const counterSet = defineOp(CounterModel, 'compat.counter.set', { - apply: (_s, p: { value: number }) => ({ value: p.value }), +const counterSet = CounterModel.defineOp('compat.counter.set', { + schema: z.object({ value: z.number() }), + apply: (_s, p) => ({ value: p.value }), }); -const tagsAdd = defineOp(TagsModel, 'compat.tags.add', { - apply: (s, p: { tag: string }) => ({ tags: [...s.tags, p.tag] }), +const tagsAdd = TagsModel.defineOp('compat.tags.add', { + schema: z.object({ tag: z.string() }), + apply: (s, p) => ({ tags: [...s.tags, p.tag] }), }); const cleanups: string[] = []; diff --git a/packages/agent-core-v2/test/wire/wireServiceImpl.test.ts b/packages/agent-core-v2/test/wire/wireServiceImpl.test.ts index 7ad75cdb31..722d389a0e 100644 --- a/packages/agent-core-v2/test/wire/wireServiceImpl.test.ts +++ b/packages/agent-core-v2/test/wire/wireServiceImpl.test.ts @@ -1,4 +1,5 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { z } from 'zod'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; @@ -9,7 +10,6 @@ import { InMemoryStorageService } from '#/persistence/backends/memory/inMemorySt import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { defineModel } from '#/wire/model'; -import { defineOp } from '#/wire/op'; import { IAgentWireService } from '#/wire/tokens'; import type { IWireService, PersistedRecord } from '#/wire/wireService'; import { CycleError, WireService } from '#/wire/wireServiceImpl'; @@ -24,23 +24,27 @@ const trace: string[] = []; const CounterModel = defineModel('store.counter', () => ({ value: 0 })); const OtherModel = defineModel('store.other', () => ({ value: 0 })); -const counterAdd = defineOp(CounterModel, 'store.counter.add', { - apply: (s, p: { by: number }) => { +const counterAdd = CounterModel.defineOp('store.counter.add', { + schema: z.object({ by: z.number() }), + apply: (s, p) => { trace.push('apply.counter'); return { value: s.value + p.by }; }, }); -const otherSet = defineOp(OtherModel, 'store.other.set', { - apply: (_s, p: { value: number }) => { +const otherSet = OtherModel.defineOp('store.other.set', { + schema: z.object({ value: z.number() }), + apply: (_s, p) => { trace.push('apply.other'); return { value: p.value }; }, }); -const otherInc = defineOp(OtherModel, 'store.other.inc', { +const otherInc = OtherModel.defineOp('store.other.inc', { + schema: z.object({}), apply: (s) => ({ value: s.value + 1 }), }); // Test-only op that violates the new-reference convention by mutating its input. -const mutateCounter = defineOp(CounterModel, 'store.counter.mutate', { +const mutateCounter = CounterModel.defineOp('store.counter.mutate', { + schema: z.object({}), apply: (s) => { (s as { value: number }).value = 123; return s; From 2b0e60fd753e9d7d5dab6b243ff4f390f9b10609 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Mon, 13 Jul 2026 15:20:04 +0800 Subject: [PATCH 3/4] docs(v2): drop retired wire.signal references from domain headers --- .../src/agent/contextSize/contextSizeService.ts | 6 +++--- .../src/agent/fullCompaction/compactionOps.ts | 7 +++---- packages/agent-core-v2/src/agent/goal/goalOps.ts | 7 ++++--- packages/agent-core-v2/src/agent/goal/goalService.ts | 4 ++-- 4 files changed, 12 insertions(+), 12 deletions(-) diff --git a/packages/agent-core-v2/src/agent/contextSize/contextSizeService.ts b/packages/agent-core-v2/src/agent/contextSize/contextSizeService.ts index b7ef86f0dd..ad87806d13 100644 --- a/packages/agent-core-v2/src/agent/contextSize/contextSizeService.ts +++ b/packages/agent-core-v2/src/agent/contextSize/contextSizeService.ts @@ -4,9 +4,9 @@ * Owns the last measured context token count in the wire `ContextSizeModel` * (`{ length, tokens }`): reads it through `wire.getModel`, writes it through * `wire.dispatch(contextSizeMeasured(...))` (called by `llmRequester` after each - * measured exchange), and emits the `contextTokens` slice of - * `agent.status.updated` live through `wire.signal` when the measured value - * changes. `get(start?, end?)` returns `{ size, measured, estimated }` for the + * measured exchange), and derives the `contextTokens` slice of + * `agent.status.updated` from the Op's `toEvent` (published to `IEventBus` on + * dispatch) when the measured value changes. `get(start?, end?)` returns `{ size, measured, estimated }` for the * context-message range `[start, end)`, resolved like `Array.prototype.slice` * (defaulting to the whole context; negative indices count back from the end; * an inverted range is empty): `measured` diff --git a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts index 9ca2959ff4..9de43c5971 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts @@ -27,10 +27,9 @@ * `wire.onRestored` handler (mirroring `goal`'s post-replay normalization). * * The `compaction.*` events publish to `IEventBus` (`compaction.started` via the - * `begin` Op's `toEvent`; the rest directly) and also emit live through - * `wire.signal` (legacy channel, until Phase 3); they are declared here via - * interface-merge (`error` is already declared by `mcp`, so it is not - * re-declared). The `full_compaction.*` record shapes are registered in + * `begin` Op's `toEvent`; the rest directly from the service); they are + * declared here via interface-merge (`error` is already declared by `mcp`, so + * it is not re-declared). The `full_compaction.*` record shapes are registered in * `PersistedOpMap` (`#/wire/types`, below) because the records still * ride the per-agent `wire.jsonl` log read by `wireRecord.restore()` / * `getRecords()`. Consumed by the Agent-scope `fullCompactionService`. diff --git a/packages/agent-core-v2/src/agent/goal/goalOps.ts b/packages/agent-core-v2/src/agent/goal/goalOps.ts index 8a5482cb48..3fbf3f286b 100644 --- a/packages/agent-core-v2/src/agent/goal/goalOps.ts +++ b/packages/agent-core-v2/src/agent/goal/goalOps.ts @@ -12,9 +12,10 @@ * when leaving `active` and carried in the `goal.update` payload; and * `wallClockResumedAt` is a live-only service field (never persisted, reset on * replay). Each `apply` returns the same reference when nothing changes so the - * wire's reference-equality gate stays quiet. The `goal.updated` signal is - * emitted live through `wire.signal` (declared here via interface-merge); - * `wire.replay` rebuilds the Model silently and the service's `wire.onRestored` + * wire's reference-equality gate stays quiet. The `goal.updated` fact is + * published live to `IEventBus` by the service (declared here via + * interface-merge); `wire.replay` rebuilds the Model silently and the + * service's `wire.onRestored` * forces a replayed `active` goal back to `paused`. Consumed by the Agent-scope * `goalService`. */ diff --git a/packages/agent-core-v2/src/agent/goal/goalService.ts b/packages/agent-core-v2/src/agent/goal/goalService.ts index 21522a40b0..d62040dc7d 100644 --- a/packages/agent-core-v2/src/agent/goal/goalService.ts +++ b/packages/agent-core-v2/src/agent/goal/goalService.ts @@ -3,8 +3,8 @@ * * Owns the per-agent goal lifecycle; persists the goal in the `wire` * `GoalModel` (`GoalState | null`) through the `goal.create` / `goal.update` / - * `goal.clear` Ops (`wire.dispatch`), reads it through `wire.getModel`, emits - * `goal.updated` live through `wire.signal`, and forces a replayed `active` + * `goal.clear` Ops (`wire.dispatch`), reads it through `wire.getModel`, + * publishes `goal.updated` live to `IEventBus`, and forces a replayed `active` * goal back to `paused` via `wire.onRestored`. The accumulated `wallClockMs` * lives in the Model (set from each Op payload, never by `Date.now()` inside * `apply`); the `wallClockResumedAt` cursor is a live-only field, reset on From 4821912d08ee248e00d9b17fac9f830ccf685218 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Mon, 13 Jul 2026 15:21:09 +0800 Subject: [PATCH 4/4] docs(v2): drop one more retired wire signal channel reference --- .../agent-core-v2/src/session/cron/sessionCronServiceImpl.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts index fb97ed5399..85d1270bc5 100644 --- a/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts +++ b/packages/agent-core-v2/src/session/cron/sessionCronServiceImpl.ts @@ -6,8 +6,8 @@ * (tick / coalesce / jitter / cursor), persists mutations through the * App-scoped `ICronTaskPersistence`, mirrors mutations as `cron.add` / * `cron.delete` / `cron.cursor` Ops on the main agent's `wire` (cross-scope - * borrow) so `wire.replay` can rebuild the `CronModel`, fires `cron.fired` - * through the main agent's `wire` signal channel, steers the main agent + * borrow) so `wire.replay` can rebuild the `CronModel`, publishes `cron.fired` + * to the main agent's `IEventBus`, steers the main agent * through `IAgentPromptService` when a task fires, and registers the cron * tools (`CronCreate` / `CronList` / `CronDelete`) into the main agent's * `IAgentToolRegistryService` once `IAgentLifecycleService` signals