fix(agent-core-v2): persist active goal time (#1695)

* fix(agent-core-v2): persist active goal time

* fix(agent-core-v2): migrate active goal anchors

* fix(agent-core-v2): migrate envelope-less goal logs

* fix(agent-core-v2): advance migrated goal checkpoints

* fix(agent-core-v2): align crash recovery with wire domain

.changeset/fix-goal-crash-wallclock.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Preserve active goal elapsed time across crash recovery.

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<GoalBudgetLimits>().optional(),
     actor: z.custom<GoalActor>().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 };
     }

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<number, string>();
   private readonly countedGoalTurns = new Set<number>();
@@ -312,14 +312,15 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
   async createGoal(input: CreateGoalInput, actor: GoalActor = 'user'): Promise<GoalSnapshot> {
     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;
   }

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 {

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')))
+  );
+}

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,

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' }),

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": "<protocol-version>", "created_at": "<time>" }
+      [wire] metadata      { "protocol_version": "1.4", "created_at": "<time>" }
       [wire] goal.create   { "goalId": "goal-1", "objective": "ship the feature", "completionCriterion": "tests pass", "time": "<time>" }
       [wire] goal.update   { "tokensUsed": 5, "wallClockMs": 0, "time": "<time>" }
       [wire] goal.update   { "turnsUsed": 1, "time": "<time>" }

packages/agent-core-v2/test/wire/migration/v1.5.test.ts

@@ -0,0 +1,73 @@
+/**
+ * Scenario: migrate persisted goal lifecycle records from wire protocol 1.4 to 1.5.
+ * Responsibilities: recover missing active wall-clock anchors without replacing persisted ones.
+ * Wiring: pure migration exercised through the shared migration test surface.
+ * Run: `pnpm --filter @moonshot-ai/agent-core-v2 exec vitest run test/wire/migration/v1.5.test.ts`.
+ */
+import { describe, expect, it } from 'vitest';
+
+import { migrateV1_4ToV1_5 } from '#/wire/migration/migration';
+import { runMigration } from './utils';
+
+describe('1.4 to 1.5 active wall-clock anchor migration', () => {
+  it('backfills missing anchors from create and resume record timestamps', () => {
+    expect(
+      runMigration(migrateV1_4ToV1_5, [
+        {
+          type: 'metadata',
+          protocol_version: '1.4',
+          created_at: 1,
+        },
+        {
+          type: 'goal.create',
+          goalId: 'goal-1',
+          objective: 'ship the feature',
+          time: 10,
+        },
+        {
+          type: 'goal.update',
+          status: 'paused',
+          wallClockMs: 20,
+          time: 30,
+        },
+        {
+          type: 'goal.update',
+          status: 'active',
+          time: 40,
+        },
+      ]),
+    ).toMatchInlineSnapshot(`
+      [wire] metadata      { "protocol_version": "<protocol-version>", "created_at": "<time>" }
+      [wire] goal.create   { "goalId": "goal-1", "objective": "ship the feature", "time": "<time>", "wallClockResumedAt": 10 }
+      [wire] goal.update   { "status": "paused", "wallClockMs": 20, "time": "<time>" }
+      [wire] goal.update   { "status": "active", "time": "<time>", "wallClockResumedAt": 40 }
+    `);
+  });
+
+  it('preserves an existing active wall-clock anchor', () => {
+    expect(
+      runMigration(migrateV1_4ToV1_5, [
+        {
+          type: 'goal.update',
+          status: 'active',
+          wallClockResumedAt: 35,
+          time: 40,
+        },
+      ]),
+    ).toMatchInlineSnapshot(`[wire] goal.update   { "status": "active", "wallClockResumedAt": 35, "time": "<time>" }`);
+  });
+
+  it('advances a missing anchor from a wall-clock checkpoint timestamp', () => {
+    expect(
+      runMigration(migrateV1_4ToV1_5, [
+        {
+          type: 'goal.update',
+          wallClockMs: 3_000,
+          time: 4_000,
+        },
+      ]),
+    ).toMatchInlineSnapshot(
+      `[wire] goal.update   { "wallClockMs": 3000, "time": "<time>", "wallClockResumedAt": 4000 }`,
+    );
+  });
+});

packages/agent-core-v2/test/wire/resume.test.ts

@@ -6,6 +6,7 @@ import { describe, expect, it, vi } from 'vitest';
 
 import {
   WIRE_PROTOCOL_VERSION,
+  IAgentGoalService,
   type WireRecord,
   type PromptOrigin,
 } from '#/index';
@@ -565,6 +566,119 @@ describe('Agent resume', () => {
     expect(ctx.context.get()).toHaveLength(0);
   });
 
+  it('restores an envelope-less active interval into a budget-reached paused goal', async () => {
+    const now = vi.spyOn(Date, 'now').mockReturnValue(6_000);
+    const persistence = new RecordingAgentPersistence(
+      [
+        {
+          type: 'goal.create',
+          goalId: 'goal-1',
+          objective: 'ship work',
+          time: 100,
+        },
+        {
+          type: 'goal.update',
+          status: 'paused',
+          wallClockMs: 2_000,
+          actor: 'user',
+          time: 500,
+        },
+        {
+          type: 'goal.update',
+          status: 'active',
+          budgetLimits: { wallClockBudgetMs: 6_000 },
+          actor: 'user',
+          time: 1_000,
+        },
+      ] as unknown as WireRecord[],
+      false,
+    );
+    const ctx = testAgent({ persistence, autoConfigure: false });
+
+    try {
+      await ctx.restorePersisted();
+
+      const goal = ctx.get(IAgentGoalService).getGoal().goal;
+      expect(goal).toMatchObject({
+        status: 'paused',
+        wallClockMs: 7_000,
+        budget: {
+          wallClockBudgetReached: true,
+          remainingWallClockMs: 0,
+          overBudget: true,
+        },
+      });
+      expect(persistence.appended).toEqual([
+        expect.objectContaining({
+          type: 'goal.update',
+          status: 'paused',
+          reason: 'Paused after agent resume',
+          wallClockMs: 7_000,
+        }),
+      ]);
+      expect(persistence.rewritten).toContainEqual(
+        expect.objectContaining({
+          type: 'goal.update',
+          status: 'active',
+          wallClockResumedAt: 1_000,
+        }),
+      );
+    } finally {
+      now.mockRestore();
+      await ctx.dispose();
+    }
+  });
+
+  it('restores only post-checkpoint active time from a 1.3 wall-clock checkpoint', async () => {
+    const now = vi.spyOn(Date, 'now').mockReturnValue(6_000);
+    const persistence = new RecordingAgentPersistence([
+      {
+        type: 'metadata',
+        protocol_version: '1.3',
+        created_at: 1,
+      },
+      {
+        type: 'goal.create',
+        goalId: 'goal-1',
+        objective: 'ship work',
+        time: 1_000,
+      },
+      {
+        type: 'goal.account_usage',
+        goalId: 'goal-1',
+        wallClockMs: 3_000,
+        time: 4_000,
+      },
+    ] as unknown as WireRecord[]);
+    const ctx = testAgent({ persistence, autoConfigure: false });
+
+    try {
+      await ctx.restorePersisted();
+
+      expect(ctx.get(IAgentGoalService).getGoal().goal).toMatchObject({
+        status: 'paused',
+        wallClockMs: 5_000,
+      });
+      expect(persistence.appended).toEqual([
+        expect.objectContaining({
+          type: 'goal.update',
+          status: 'paused',
+          wallClockMs: 5_000,
+        }),
+      ]);
+      expect(persistence.rewritten).toContainEqual(
+        expect.objectContaining({
+          type: 'goal.update',
+          wallClockMs: 3_000,
+          wallClockResumedAt: 4_000,
+        }),
+      );
+    } finally {
+      now.mockRestore();
+      await ctx.dispose();
+    }
+  });
+
   it('restores context after undo and removes undone messages from replay', async () => {
     const persistence = new RecordingAgentPersistence([
       {
@@ -664,8 +778,8 @@ class RecordingAgentPersistence extends InMemoryWireRecordPersistence {
   readonly appended: WireRecord[] = [];
   rewritten: readonly WireRecord[] | undefined;
 
-  constructor(events: readonly WireRecord[]) {
-    super(withMetadata(events));
+  constructor(events: readonly WireRecord[], addMetadata = true) {
+    super(addMetadata ? withMetadata(events) : events);
   }
 
   override append(input: WireRecord): void {