diff --git a/.changeset/activity-view-last-turn-seed.md b/.changeset/activity-view-last-turn-seed.md new file mode 100644 index 000000000..fb0d14bdd --- /dev/null +++ b/.changeset/activity-view-last-turn-seed.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Restore how the last turn ended (completed, cancelled, or failed) when a session is resumed after a server restart, so clients can still surface a previously failed turn instead of the session looking silently stopped. diff --git a/packages/agent-core-v2/src/agent/activityView/activityViewService.ts b/packages/agent-core-v2/src/agent/activityView/activityViewService.ts index 57a722d71..bc73d64d4 100644 --- a/packages/agent-core-v2/src/agent/activityView/activityViewService.ts +++ b/packages/agent-core-v2/src/agent/activityView/activityViewService.ts @@ -6,14 +6,16 @@ * events drive the live phase/stream/retry detail, permission approval events * drive the pending-approval list, while task and full-compaction events drive * the background-work slice. The view seeds once from `IAgentLoopService`, - * `IAgentTaskService`, and `IAgentFullCompactionService` (reads, never writes) - * and otherwise holds only derived state, so it can be discarded and rebuilt - * at any time. The mutable view state (`lifecycle`, `turn`, `lastTurn`, - * `background`, `current`) is registered into `agentState` - * (`IAgentStateService`) and read/written through it; the event-bus - * subscription handles stay mechanism held by the `Disposable` base, and - * `MutableTurn`'s in-place-mutated Maps stay instance fields of that - * per-turn class. Bound at Agent scope. + * `IAgentTaskService`, and `IAgentFullCompactionService`, and recovers the + * last turn's outcome from the wire `TurnModel` through `IWireService`, so + * a cold-resumed agent still reports how its previous turn ended (reads, + * never writes). Otherwise the view holds only derived state, so it can be + * discarded and rebuilt at any time. The mutable view state (`lifecycle`, + * `turn`, `lastTurn`, `background`, `current`) is registered into + * `agentState` (`IAgentStateService`) and read/written through it; the + * event-bus subscription handles stay mechanism held by the `Disposable` + * base, and `MutableTurn`'s in-place-mutated Maps stay instance fields of + * that per-turn class. Bound at Agent scope. */ import { Disposable } from '#/_base/di/lifecycle'; @@ -21,12 +23,14 @@ import { LifecycleScope, ScopeActivation, registerScopedService } from '#/_base/ import { defineState } from '#/_base/state/stateRegistry'; import { IEventBus } from '#/app/event/eventBus'; import { IAgentLoopService } from '#/agent/loop/loop'; +import { TurnModel } from '#/agent/loop/turnOps'; import { IAgentStateService } from '#/agent/state/agentState'; import { IAgentTaskService } from '#/agent/task/task'; import { IAgentFullCompactionService } from '#/agent/fullCompaction/fullCompaction'; import { USER_PROMPT_ORIGIN } from '#/agent/contextMemory/types'; import type { PromptOrigin } from '#/agent/contextMemory/types'; import type { TurnEndReason } from '#/agent/loop/turnEvents'; +import { IWireService } from '#/wire/wire'; import type { ActivityLastTurnState, @@ -74,6 +78,7 @@ export class AgentActivityView extends Disposable implements IAgentActivityView @IAgentTaskService private readonly tasks: IAgentTaskService, @IAgentFullCompactionService private readonly fullCompaction: IAgentFullCompactionService, @IAgentStateService private readonly states: IAgentStateService, + @IWireService private readonly wire: IWireService, ) { super(); this.states.register(activityViewLifecycleKey); @@ -84,6 +89,12 @@ export class AgentActivityView extends Disposable implements IAgentActivityView this.seedFromLoop(); this.seedFromTasks(); this.seedFromFullCompaction(); + this._register( + this.wire.hooks.onDidRestore.register('activityView', async (_ctx, next) => { + this.seedLastTurnFromWire(); + await next(); + }), + ); this._register(this.eventBus.subscribe('turn.started', (e) => this.onTurnStarted(e.turnId, e.origin))); this._register(this.eventBus.subscribe('turn.step.started', (e) => this.onStepStarted(e.step))); @@ -220,8 +231,24 @@ export class AgentActivityView extends Disposable implements IAgentActivityView private seedFromLoop(): void { const status = this.loop.status(); - if (status.state !== 'running' || status.activeTurnId === undefined) return; - this.turn = new MutableTurn(status.activeTurnId, USER_PROMPT_ORIGIN); + if (status.state === 'running' && status.activeTurnId !== undefined) { + this.turn = new MutableTurn(status.activeTurnId, USER_PROMPT_ORIGIN); + this.publish(); + return; + } + this.seedLastTurnFromWire(); + } + + private seedLastTurnFromWire(): void { + if (this.turn !== undefined || this.lastTurn !== undefined) return; + const lastEnded = this.wire.getModel(TurnModel).lastEnded; + if (lastEnded === undefined) return; + this.lastTurn = { + turnId: lastEnded.turnId, + reason: lastEnded.reason, + durationMs: lastEnded.durationMs, + at: Date.now(), + }; this.publish(); } diff --git a/packages/agent-core-v2/src/agent/loop/turnOps.ts b/packages/agent-core-v2/src/agent/loop/turnOps.ts index ad96e5b63..9901b077e 100644 --- a/packages/agent-core-v2/src/agent/loop/turnOps.ts +++ b/packages/agent-core-v2/src/agent/loop/turnOps.ts @@ -4,9 +4,9 @@ * * Owns the next available turn id, including cancelled queued reservations and * legacy loop-event observations. Also persists the terminal `turn.ended` - * record (reason / error / durationMs) so downstream history rebuilds can - * recover how a turn ended; the record carries no engine-restorable state, so - * its `apply` is a no-op. Consumed by the Agent-scope `loopService`; the + * record (reason / error / durationMs) so downstream history rebuilds and + * cold-resumed read models (e.g. the activity view) can recover how the last + * turn ended. Consumed by the Agent-scope `loopService`; the * `interruptionReminder` domain projects `turn.cancel` into its own model. */ @@ -20,6 +20,11 @@ import type { PromptOrigin } from '#/agent/contextMemory/types'; export interface TurnModelState { readonly nextTurnId: number; readonly cancelledTurnIds: readonly number[]; + readonly lastEnded?: { + readonly turnId: number; + readonly reason: 'completed' | 'cancelled' | 'failed' | 'blocked'; + readonly durationMs?: number; + }; } export const TurnModel = defineModel( @@ -33,9 +38,13 @@ export const TurnModel = defineModel( } const turnId = Number.parseInt(event.turnId, 10); - return Number.isInteger(turnId) && turnId >= state.nextTurnId - ? advanceTurnClock(state, turnId + 1) - : state; + if (!Number.isInteger(turnId)) return state; + let next = state; + if (turnId >= state.nextTurnId) next = advanceTurnClock(state, turnId + 1); + if (next.lastEnded !== undefined && turnId > next.lastEnded.turnId) { + next = { ...next, lastEnded: undefined }; + } + return next; }, }, }, @@ -85,7 +94,10 @@ export const endTurn = TurnModel.defineOp('turn.ended', { error: z.custom().optional(), durationMs: z.number().optional(), }), - apply: (s) => s, + apply: (s, { turnId, reason, durationMs }) => ({ + ...s, + lastEnded: { turnId, reason, durationMs }, + }), }); function advanceTurnClock( @@ -98,6 +110,7 @@ function advanceTurnClock( ); while (pendingCancellations.delete(nextTurnId)) nextTurnId += 1; return { + ...state, nextTurnId, cancelledTurnIds: [...pendingCancellations].toSorted((a, b) => a - b), }; diff --git a/packages/agent-core-v2/test/agent/activityView/activityView.test.ts b/packages/agent-core-v2/test/agent/activityView/activityView.test.ts index 0b0c7f4cb..fe97eff8a 100644 --- a/packages/agent-core-v2/test/agent/activityView/activityView.test.ts +++ b/packages/agent-core-v2/test/agent/activityView/activityView.test.ts @@ -17,10 +17,10 @@ import { IAgentTaskService } from '#/agent/task/task'; import type { AgentTaskInfo } from '#/agent/task/types'; import { AgentActivityView } from '#/agent/activityView/activityViewService'; import { IAgentActivityView, type AgentActivityState } from '#/agent/activityView/activityView'; -import { - IAgentFullCompactionService, - type FullCompactionTask, -} from '#/agent/fullCompaction/fullCompaction'; +import { IAgentFullCompactionService } from '#/agent/fullCompaction/fullCompaction'; +import type { FullCompactionTask } from '#/agent/fullCompaction/fullCompaction'; +import { TurnModel, type TurnModelState } from '#/agent/loop/turnOps'; +import { IWireService } from '#/wire/wire'; class FakeBus { private readonly byType = new Map void>>(); @@ -64,16 +64,38 @@ let disposables: DisposableStore; function harness( seedTasks: readonly AgentTaskInfo[] = [], compacting: FullCompactionTask | null = null, + lastEnded?: TurnModelState['lastEnded'], ) { const bus = new FakeBus(); const loop = { status: () => ({ state: 'idle', pendingTurnIds: [], hasPendingRequests: false }), } as unknown as IAgentLoopService; const tasks = { list: () => seedTasks } as unknown as IAgentTaskService; + const wireState: { lastEnded?: TurnModelState['lastEnded'] } = { lastEnded }; + const restoreHooks: Array<() => Promise> = []; + const wire = { + getModel: (model: unknown) => + model === TurnModel + ? { nextTurnId: 1, cancelledTurnIds: [], lastEnded: wireState.lastEnded } + : undefined, + hooks: { + onDidRestore: { + register: (_id: string, fn: (ctx: undefined, next: () => Promise) => Promise) => { + restoreHooks.push(async () => fn(undefined, async () => {})); + return { dispose: () => {} }; + }, + }, + }, + } as unknown as IWireService; + const restore = async (ended: TurnModelState['lastEnded']): Promise => { + wireState.lastEnded = ended; + for (const hook of restoreHooks) await hook(); + }; const ix = disposables.add(new TestInstantiationService()); ix.stub(IEventBus, bus as unknown as IEventBus); ix.stub(IAgentLoopService, loop); ix.stub(IAgentTaskService, tasks); + ix.stub(IWireService, wire); ix.set(IAgentStateService, new AgentStateService()); ix.stub(IAgentFullCompactionService, { _serviceBrand: undefined, @@ -85,7 +107,7 @@ function harness( bus.published .filter((e) => e.type === 'agent.activity.updated') .map((e) => e as unknown as AgentActivityState); - return { bus, view, updates }; + return { bus, view, updates, restore }; } describe('AgentActivityView', () => { @@ -119,6 +141,30 @@ describe('AgentActivityView', () => { expect(view.state().background).toEqual([{ kind: 'process', id: 'bash-9', since: 100 }]); }); + it('seeds lastTurn from the wire TurnModel when the view is built after restore', () => { + const { view } = harness([], null, { turnId: 7, reason: 'failed', durationMs: 1234 }); + expect(view.state().lastTurn).toMatchObject({ turnId: 7, reason: 'failed', durationMs: 1234 }); + }); + + it('seeds lastTurn when the wire restore lands after construction (cold resume ordering)', async () => { + const { view, restore } = harness(); + expect(view.state().lastTurn).toBeUndefined(); + await restore({ turnId: 7, reason: 'failed', durationMs: 1234 }); + expect(view.state().lastTurn).toMatchObject({ turnId: 7, reason: 'failed', durationMs: 1234 }); + }); + + it('does not overwrite a live lastTurn when the restore hook runs', async () => { + const { bus, view, restore } = harness([], null, { turnId: 7, reason: 'failed' }); + bus.publish({ type: 'turn.ended', turnId: 9, reason: 'completed' }); + await restore({ turnId: 7, reason: 'failed' }); + expect(view.state().lastTurn).toMatchObject({ turnId: 9, reason: 'completed' }); + }); + + it('leaves lastTurn empty when the wire has no ended turn', () => { + const { view } = harness(); + expect(view.state().lastTurn).toBeUndefined(); + }); + it('folds full compaction into the background slice', () => { const { bus, view } = harness(); diff --git a/packages/agent-core-v2/test/agent/loop/turnOps.test.ts b/packages/agent-core-v2/test/agent/loop/turnOps.test.ts new file mode 100644 index 000000000..65b2dabfa --- /dev/null +++ b/packages/agent-core-v2/test/agent/loop/turnOps.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, it } from 'vitest'; + +import { MODEL_CROSS_REDUCERS } from '#/wire/model'; +import { TurnModel, cancelTurn, endTurn, promptTurn } from '#/agent/loop/turnOps'; + +function foldLoopEvent(s: import('#/agent/loop/turnOps').TurnModelState, turnId: string) { + const entries = MODEL_CROSS_REDUCERS.get('context.append_loop_event') ?? []; + const entry = entries.find((e) => e.model === TurnModel); + if (entry === undefined) throw new Error('turn model cross-reducer not registered'); + return entry.reducer(s, { event: { type: 'step.begin', turnId } }) as typeof s; +} + +describe('TurnModel lastEnded', () => { + it('keeps the stored outcome across prompts and queued cancels', () => { + let s = TurnModel.initial(); + s = promptTurn.apply(s, { input: [], origin: { kind: 'user' } }); + s = endTurn.apply(s, { turnId: 0, reason: 'failed', durationMs: 10 }); + expect(s.lastEnded).toMatchObject({ turnId: 0, reason: 'failed' }); + s = promptTurn.apply(s, { input: [], origin: { kind: 'user' } }); + expect(s.lastEnded?.reason).toBe('failed'); + s = cancelTurn.apply(s, { turnId: 1, target: 'queued' }); + expect(s.lastEnded?.reason).toBe('failed'); + s = endTurn.apply(s, { turnId: 1, reason: 'completed' }); + expect(s.lastEnded).toMatchObject({ turnId: 1, reason: 'completed' }); + }); + + it('clears the stored outcome once a newer turn starts producing', () => { + let s = TurnModel.initial(); + s = promptTurn.apply(s, { input: [], origin: { kind: 'user' } }); + s = endTurn.apply(s, { turnId: 0, reason: 'failed' }); + s = promptTurn.apply(s, { input: [], origin: { kind: 'user' } }); + s = foldLoopEvent(s, '1'); + expect(s.lastEnded).toBeUndefined(); + }); + + it('keeps the stored outcome on the same turn’s own events', () => { + let s = TurnModel.initial(); + s = promptTurn.apply(s, { input: [], origin: { kind: 'user' } }); + s = foldLoopEvent(s, '0'); + s = endTurn.apply(s, { turnId: 0, reason: 'completed' }); + s = foldLoopEvent(s, '0'); + expect(s.lastEnded?.reason).toBe('completed'); + }); + + it('starts without a stored outcome', () => { + expect(TurnModel.initial().lastEnded).toBeUndefined(); + }); +});