diff --git a/.changeset/fix-goal-crash-wallclock.md b/.changeset/fix-goal-crash-wallclock.md new file mode 100644 index 0000000000..ba48b45370 --- /dev/null +++ b/.changeset/fix-goal-crash-wallclock.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Preserve active goal elapsed time across crash recovery. diff --git a/packages/agent-core-v2/src/agent/goal/goalOps.ts b/packages/agent-core-v2/src/agent/goal/goalOps.ts index 03bf62bc4e..70eaba0d29 100644 --- a/packages/agent-core-v2/src/agent/goal/goalOps.ts +++ b/packages/agent-core-v2/src/agent/goal/goalOps.ts @@ -5,14 +5,21 @@ * * Declares the current goal as `GoalState | null` (initial `null`); `GoalState` * holds the persistent, replayable fields — identity, objective, status, - * `turnsUsed` / `tokensUsed`, the accumulated `wallClockMs`, `budgetLimits`, - * and `terminalReason`. The non-deterministic bits stay OUT of `apply`: - * `goalId` is minted at the call site and carried in the `goal.create` payload; - * the `wallClockMs` `Date.now()` accumulation is computed by the live service - * 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` fact is + * `turnsUsed` / `tokensUsed`, the accumulated `wallClockMs`, the current + * active interval's epoch-ms `wallClockResumedAt`, `budgetLimits`, and + * `terminalReason`. The persistence contract charges an active interval from + * its persisted create/resume anchor through the first recovery clock read, + * then folds that interval into `wallClockMs` while recovery pauses the goal. + * This intentionally includes unobservable crash downtime: a monotonic clock + * cannot span processes, while learning the crash instant would require + * periodic durable writes. System-clock rollback is clamped to zero. The + * 1.4 -> 1.5 compatibility transform (also applied before sealing + * envelope-less logs) derives missing create/resume/checkpoint anchors from + * those records' existing epoch-ms `time` stamps. The + * non-deterministic values stay OUT of `apply`: `goalId` and the wall-clock + * anchor/totals are computed by the live service and carried in Op payloads. + * Each `apply` returns the same reference when nothing changes so the 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.restore` rebuilds the Model silently and the * service's `wire.hooks.onDidRestore` @@ -40,6 +47,7 @@ export interface GoalState { readonly turnsUsed: number; readonly tokensUsed: number; readonly wallClockMs: number; + readonly wallClockResumedAt?: number; readonly budgetLimits: GoalBudgetLimits; readonly terminalReason?: string; } @@ -71,6 +79,7 @@ export const createGoal = GoalModel.defineOp('goal.create', { goalId: z.string(), objective: z.string(), completionCriterion: z.string().optional(), + wallClockResumedAt: z.number().optional(), }), apply: (_s, p) => ({ goalId: p.goalId, @@ -80,6 +89,7 @@ export const createGoal = GoalModel.defineOp('goal.create', { turnsUsed: 0, tokensUsed: 0, wallClockMs: 0, + wallClockResumedAt: p.wallClockResumedAt, budgetLimits: {}, }), }); @@ -91,6 +101,7 @@ export const updateGoal = GoalModel.defineOp('goal.update', { turnsUsed: z.number().optional(), tokensUsed: z.number().optional(), wallClockMs: z.number().optional(), + wallClockResumedAt: z.number().optional(), budgetLimits: z.custom().optional(), actor: z.custom().optional(), }), @@ -102,6 +113,8 @@ export const updateGoal = GoalModel.defineOp('goal.update', { ...(next ?? s), status: p.status, terminalReason: p.status === 'active' ? undefined : p.reason, + wallClockResumedAt: + p.status === 'active' ? p.wallClockResumedAt : undefined, }; } if (p.turnsUsed !== undefined && p.turnsUsed !== s.turnsUsed) { @@ -113,6 +126,13 @@ export const updateGoal = GoalModel.defineOp('goal.update', { if (p.wallClockMs !== undefined && p.wallClockMs !== s.wallClockMs) { next = { ...(next ?? s), wallClockMs: p.wallClockMs }; } + if ( + p.wallClockResumedAt !== undefined && + (p.status ?? s.status) === 'active' && + p.wallClockResumedAt !== s.wallClockResumedAt + ) { + next = { ...(next ?? s), wallClockResumedAt: p.wallClockResumedAt }; + } if (p.budgetLimits !== undefined && p.budgetLimits !== s.budgetLimits) { next = { ...(next ?? s), budgetLimits: p.budgetLimits }; } diff --git a/packages/agent-core-v2/src/agent/goal/goalService.ts b/packages/agent-core-v2/src/agent/goal/goalService.ts index 8828c2cad5..ab4794a8aa 100644 --- a/packages/agent-core-v2/src/agent/goal/goalService.ts +++ b/packages/agent-core-v2/src/agent/goal/goalService.ts @@ -7,9 +7,10 @@ * publishes `goal.updated` live to `IEventBus`, and forces a replayed `active` * goal back to `paused` via `wire.hooks.onDidRestore`. 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 replay and (re)started on the live path. A `forked` wire Op - * clears the Model + * `Date.now()` inside `apply`); the active interval's epoch-ms + * `wallClockResumedAt` anchor is + * persisted at create/resume boundaries so recovery can settle crash-spanning + * elapsed time without periodic writes. A `forked` wire Op clears the Model * at a fork boundary; the `goal.*` payload shapes are registered in * `PersistedOpMap` (`#/wire/types`) inside `goalOps` because they still ride * the Agent wire journal restored into the Model. @@ -194,7 +195,6 @@ function isGoalContinuationOrigin(origin: TurnStartedEvent['origin']): boolean { export class AgentGoalService extends Disposable implements IAgentGoalService { declare readonly _serviceBrand: undefined; - private wallClockResumedAt?: number; private liveTurnId?: number; private readonly goalDrivenTurns = new Map(); private readonly countedGoalTurns = new Set(); @@ -312,14 +312,15 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { async createGoal(input: CreateGoalInput, actor: GoalActor = 'user'): Promise { const objective = this.validateObjective(input.objective); this.prepareForGoalCreation(input.replace === true); + const wallClockResumedAt = Date.now(); this.wire.dispatch( createGoal({ goalId: randomUUID(), objective, completionCriterion: normalizeCompletionCriterion(input.completionCriterion), + wallClockResumedAt, }), ); - this.wallClockResumedAt = Date.now(); this.adoptStarterTurn(actor); const state = this.requireState(); this.emitGoalUpdated(this.toSnapshot(state)); @@ -457,7 +458,6 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { private dispatchCompletion(state: GoalState, reason: string | undefined, actor: GoalActor): void { const wallClockMs = this.settleWallClock(state); - this.wallClockResumedAt = undefined; this.wire.dispatch(updateGoal({ status: 'complete', reason, wallClockMs, actor })); } @@ -763,7 +763,6 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { this.appendForkClearedReminder(); const state = this.goalState; if (state === null) return; - this.wallClockResumedAt = undefined; if (state.status === 'complete') { this.clearInternal('runtime', { emit: false, track: false }); return; @@ -796,7 +795,6 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { ): void { if (this.goalState === null) return; this.cancelPendingContinuation(opts.preserveLiveContinuation === true); - this.wallClockResumedAt = undefined; this.wire.dispatch(clearGoal({})); if (opts.emit !== false) this.emitGoalUpdated(null); if (opts.track !== false) this.telemetry.track2('goal_cleared', { actor }); @@ -810,13 +808,13 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { opts: { readonly preserveLiveContinuation?: boolean } = {}, ): GoalSnapshot { const wallClockMs = this.settleWallClock(state); - if (status === 'active') { - this.wallClockResumedAt = Date.now(); - } else if (state.status === 'active') { + const wallClockResumedAt = status === 'active' ? Date.now() : undefined; + if (status !== 'active' && state.status === 'active') { this.cancelPendingContinuation(opts.preserveLiveContinuation === true); - this.wallClockResumedAt = undefined; } - this.wire.dispatch(updateGoal({ status, reason, wallClockMs, actor })); + this.wire.dispatch( + updateGoal({ status, reason, wallClockMs, wallClockResumedAt, actor }), + ); const next = this.requireState(); if (status === 'active') this.adoptStarterTurn(actor); this.emitGoalUpdated(this.toSnapshot(next), { kind: 'lifecycle', status, reason, actor }); @@ -848,15 +846,15 @@ export class AgentGoalService extends Disposable implements IAgentGoalService { } private settleWallClock(state: GoalState): number { - if (state.status === 'active' && this.wallClockResumedAt !== undefined) { - return state.wallClockMs + Math.max(0, Date.now() - this.wallClockResumedAt); + if (state.status === 'active' && state.wallClockResumedAt !== undefined) { + return state.wallClockMs + Math.max(0, Date.now() - state.wallClockResumedAt); } return state.wallClockMs; } private liveWallClockMs(state: GoalState): number { - if (state.status === 'active' && this.wallClockResumedAt !== undefined) { - return state.wallClockMs + Math.max(0, Date.now() - this.wallClockResumedAt); + if (state.status === 'active' && state.wallClockResumedAt !== undefined) { + return state.wallClockMs + Math.max(0, Date.now() - state.wallClockResumedAt); } return state.wallClockMs; } diff --git a/packages/agent-core-v2/src/wire/migration/migration.ts b/packages/agent-core-v2/src/wire/migration/migration.ts index 73508ed104..f572b886c8 100644 --- a/packages/agent-core-v2/src/wire/migration/migration.ts +++ b/packages/agent-core-v2/src/wire/migration/migration.ts @@ -4,15 +4,17 @@ import { migrateV1_0ToV1_1 } from './v1.1'; import { migrateV1_1ToV1_2 } from './v1.2'; import { migrateV1_2ToV1_3 } from './v1.3'; import { migrateV1_3ToV1_4 } from './v1.4'; +import { migrateV1_4ToV1_5 } from './v1.5'; export { migrateV1_0ToV1_1, migrateV1_1ToV1_2, migrateV1_2ToV1_3, migrateV1_3ToV1_4, + migrateV1_4ToV1_5, }; -export const WIRE_PROTOCOL_VERSION = '1.4'; +export const WIRE_PROTOCOL_VERSION = '1.5'; export type WireMigrationRecord = WireRecord; @@ -27,6 +29,7 @@ const MIGRATIONS: readonly WireMigration[] = [ migrateV1_1ToV1_2, migrateV1_2ToV1_3, migrateV1_3ToV1_4, + migrateV1_4ToV1_5, ]; export function isNewerWireVersion(readVersion: string): boolean { diff --git a/packages/agent-core-v2/src/wire/migration/v1.5.ts b/packages/agent-core-v2/src/wire/migration/v1.5.ts new file mode 100644 index 0000000000..8b18e7b07e --- /dev/null +++ b/packages/agent-core-v2/src/wire/migration/v1.5.ts @@ -0,0 +1,28 @@ +/** + * Wire protocol 1.5 persists an epoch-ms anchor at every goal create/resume + * boundary and wall-clock checkpoint. Version 1.4 records already carry an + * epoch-ms `time`, so the migration can recover that boundary without + * inventing a crash timestamp or adding periodic checkpoint writes. Existing + * anchors are authoritative. + */ +import type { WireMigration, WireMigrationRecord } from './migration'; + +export const migrateV1_4ToV1_5: WireMigration = { + sourceVersion: '1.4', + targetVersion: '1.5', + migrateRecord(record: WireMigrationRecord): WireMigrationRecord { + if (!advancesActiveInterval(record)) return record; + if (record['wallClockResumedAt'] !== undefined) return record; + if (typeof record['time'] !== 'number') return record; + return { ...record, wallClockResumedAt: record['time'] }; + }, +}; + +function advancesActiveInterval(record: WireMigrationRecord): boolean { + return ( + record.type === 'goal.create' || + (record.type === 'goal.update' && + (record['status'] === 'active' || + (record['status'] === undefined && typeof record['wallClockMs'] === 'number'))) + ); +} diff --git a/packages/agent-core-v2/src/wire/wireService.ts b/packages/agent-core-v2/src/wire/wireService.ts index d4a12a093e..18cc5ec9ce 100644 --- a/packages/agent-core-v2/src/wire/wireService.ts +++ b/packages/agent-core-v2/src/wire/wireService.ts @@ -29,6 +29,7 @@ import { WireError, WireErrors } from './errors'; import { WIRE_PROTOCOL_VERSION, isNewerWireVersion, + migrateV1_4ToV1_5, migrateWireRecord, resolveWireMigrations, type WireMigration, @@ -153,6 +154,7 @@ export class WireService extends Disposable implements IWireService { hasRecords = true; if (sourceRecord.type !== 'metadata') { rewrittenRecords = [createWireMetadataRecord()]; + migrations = [migrateV1_4ToV1_5]; } else if (!isWireMetadataRecord(sourceRecord)) { throw new StorageError( StorageErrors.codes.STORAGE_CORRUPTED, 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 17bcdb260b..3e9aa33e0a 100644 --- a/packages/agent-core-v2/test/agent/goal/goal.test.ts +++ b/packages/agent-core-v2/test/agent/goal/goal.test.ts @@ -498,6 +498,7 @@ describe('AgentGoalService', () => { goalId: expect.any(String), objective: 'work', completionCriterion: 'tests pass', + wallClockResumedAt: expect.any(Number), }), expect.objectContaining({ type: 'goal.update', tokensUsed: 5 }), expect.objectContaining({ type: 'goal.update', turnsUsed: 1 }), @@ -514,6 +515,7 @@ describe('AgentGoalService', () => { expect.objectContaining({ type: 'goal.update', status: 'active', + wallClockResumedAt: expect.any(Number), actor: 'user', }), expect.objectContaining({ type: 'goal.clear' }), diff --git a/packages/agent-core-v2/test/wire/migration/v1.4.test.ts b/packages/agent-core-v2/test/wire/migration/v1.4.test.ts index 0e632cfcb0..4dc3ee6358 100644 --- a/packages/agent-core-v2/test/wire/migration/v1.4.test.ts +++ b/packages/agent-core-v2/test/wire/migration/v1.4.test.ts @@ -64,7 +64,7 @@ describe('1.3 to 1.4', () => { }, ]), ).toMatchInlineSnapshot(` - [wire] metadata { "protocol_version": "", "created_at": "