Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/fix-goal-crash-wallclock.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@moonshot-ai/kimi-code": patch
---

Preserve active goal elapsed time across crash recovery.
36 changes: 28 additions & 8 deletions packages/agent-core-v2/src/agent/goal/goalOps.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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`
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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,
Expand All @@ -80,6 +89,7 @@ export const createGoal = GoalModel.defineOp('goal.create', {
turnsUsed: 0,
tokensUsed: 0,
wallClockMs: 0,
wallClockResumedAt: p.wallClockResumedAt,
budgetLimits: {},
}),
});
Expand All @@ -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(),
Comment thread
chengluyu marked this conversation as resolved.
budgetLimits: z.custom<GoalBudgetLimits>().optional(),
actor: z.custom<GoalActor>().optional(),
}),
Expand All @@ -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) {
Expand All @@ -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 };
}
Expand Down
32 changes: 15 additions & 17 deletions packages/agent-core-v2/src/agent/goal/goalService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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>();
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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 }));
}

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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 });
Expand All @@ -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 });
Expand Down Expand Up @@ -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;
}
Expand Down
5 changes: 4 additions & 1 deletion packages/agent-core-v2/src/wire/migration/migration.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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 {
Expand Down
28 changes: 28 additions & 0 deletions packages/agent-core-v2/src/wire/migration/v1.5.ts
Original file line number Diff line number Diff line change
@@ -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')))
);
}
2 changes: 2 additions & 0 deletions packages/agent-core-v2/src/wire/wireService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import { WireError, WireErrors } from './errors';
import {
WIRE_PROTOCOL_VERSION,
isNewerWireVersion,
migrateV1_4ToV1_5,
migrateWireRecord,
resolveWireMigrations,
type WireMigration,
Expand Down Expand Up @@ -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,
Expand Down
2 changes: 2 additions & 0 deletions packages/agent-core-v2/test/agent/goal/goal.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 }),
Expand All @@ -514,6 +515,7 @@ describe('AgentGoalService', () => {
expect.objectContaining({
type: 'goal.update',
status: 'active',
wallClockResumedAt: expect.any(Number),
actor: 'user',
}),
expect.objectContaining({ type: 'goal.clear' }),
Expand Down
2 changes: 1 addition & 1 deletion packages/agent-core-v2/test/wire/migration/v1.4.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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>" }
Expand Down
73 changes: 73 additions & 0 deletions packages/agent-core-v2/test/wire/migration/v1.5.test.ts
Original file line number Diff line number Diff line change
@@ -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 }`,
);
});
});
Loading
Loading