feat(agent-core-v2): persist the last turn outcome into session metadata for cold listings

.changeset/cold-session-last-turn-outcome.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Session listings keep how the last turn ended (completed, cancelled, or failed) across server restarts, so clients can mark previously failed sessions before they are opened.

packages/agent-core-v2/docs/state-manifest.d.ts

@@ -465,6 +465,7 @@ export interface SessionStateSnapshot {
       readonly swarmItem?: string;
     }>>;
     readonly custom?: Record<string, unknown>;
+    readonly lastTurnReason?: 'completed' | 'cancelled' | 'failed';
   } | undefined;
   // src/session/sessionSkillCatalog/skillCatalogService.ts
   'sessionSkillCatalog.contributions': Map<string, {

packages/agent-core-v2/src/app/sessionIndex/sessionIndex.ts

@@ -50,6 +50,7 @@ export interface SessionSummary {
   readonly updatedAt: number;
   readonly archived: boolean;
   readonly custom?: Record<string, unknown>;
+  readonly lastTurnReason?: 'completed' | 'cancelled' | 'failed';
 }
 
 export interface SessionListQuery {

packages/agent-core-v2/src/app/sessionIndex/sessionIndexSource.ts

@@ -37,6 +37,10 @@ export function parseTime(value: unknown): number {
   return 0;
 }
 
+export function parseTurnOutcome(value: unknown): 'completed' | 'cancelled' | 'failed' | undefined {
+  return value === 'completed' || value === 'cancelled' || value === 'failed' ? value : undefined;
+}
+
 export function recoverCwd(meta: Record<string, unknown>): string | undefined {
   if (typeof meta['cwd'] === 'string' && meta['cwd'].length > 0) return meta['cwd'];
   if (typeof meta['workDir'] === 'string' && meta['workDir'].length > 0) {
@@ -63,6 +67,7 @@ export function buildSessionSummary(fields: {
   updatedAt: number;
   archived: boolean;
   custom?: Record<string, unknown>;
+  lastTurnReason?: 'completed' | 'cancelled' | 'failed';
 }): SessionSummary {
   return {
     id: fields.id,
@@ -74,6 +79,7 @@ export function buildSessionSummary(fields: {
     updatedAt: fields.updatedAt,
     archived: fields.archived,
     custom: fields.custom,
+    lastTurnReason: fields.lastTurnReason,
   };
 }
 
@@ -102,6 +108,7 @@ export function summaryEquals(a: SessionSummary, b: SessionSummary): boolean {
     a.createdAt === b.createdAt &&
     a.updatedAt === b.updatedAt &&
     a.archived === b.archived &&
+    a.lastTurnReason === b.lastTurnReason &&
     JSON.stringify(a.custom) === JSON.stringify(b.custom)
   );
 }
@@ -153,6 +160,7 @@ export async function readSessionSummary(
     updatedAt: parseTime(meta['updatedAt']),
     archived: meta['archived'] === true,
     custom,
+    lastTurnReason: parseTurnOutcome(meta['lastTurnReason']),
   });
 }
 

packages/agent-core-v2/src/index.ts

@@ -98,6 +98,8 @@ export * from '#/session/sessionMetadata/sessionMetadata';
 export * from '#/session/sessionMetadata/sessionMetadataService';
 export * from '#/session/sessionActivity/sessionActivity';
 export * from '#/session/sessionActivity/sessionActivityService';
+export * from '#/session/sessionActivity/sessionOutcomeMirror';
+export * from '#/session/sessionActivity/sessionOutcomeMirrorService';
 export * from '#/session/sessionToolPolicy/sessionToolPolicy';
 export * from '#/session/sessionToolPolicy/sessionToolPolicyService';
 export * from '#/app/config/config';

packages/agent-core-v2/src/session/sessionActivity/sessionOutcomeMirror.ts

@@ -0,0 +1,19 @@
+/**
+ * `sessionActivity` domain — `ISessionOutcomeMirror` contract: persist the
+ * latest main-turn outcome into durable session metadata.
+ *
+ * The activity aggregate's `lastTurnReason` is live fold state, rebuilt per
+ * process; this mirror is the write side that lands terminal outcomes in
+ * the session's metadata document, so the session index (and therefore cold
+ * listings after a restart) keep reporting them. Session-scoped — one
+ * instance per session.
+ */
+
+import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation';
+
+export interface ISessionOutcomeMirror {
+  readonly _serviceBrand: undefined;
+}
+
+export const ISessionOutcomeMirror: ServiceIdentifier<ISessionOutcomeMirror> =
+  createDecorator<ISessionOutcomeMirror>('sessionOutcomeMirror');

packages/agent-core-v2/src/session/sessionActivity/sessionOutcomeMirrorService.ts

@@ -0,0 +1,128 @@
+/**
+ * `sessionActivity` domain — `ISessionOutcomeMirror` implementation.
+ *
+ * Persists the main agent's terminal turn outcomes through `ISessionMetadata`
+ * (observed via `agentLifecycle` and the main agent's `eventBus`), so the
+ * session index keeps reporting them across restarts. Persisted on turn end
+ * (completed/failed, or a user's stop), cleared when a new turn starts, and
+ * backfilled from a cold resume's restored outcome — backfills never bump
+ * `updatedAt`, and programmatic aborts (including scope-teardown cancels)
+ * are deliberately never persisted, and backfills only apply to a pure
+ * resume (no turn started in this process — a live turn end owns its write,
+ * recency bump included). Writes are deduped against the last value this
+ * process persisted. Bound at Session scope.
+ */
+
+import { Disposable, DisposableStore } from '#/_base/di/lifecycle';
+import { LifecycleScope, ScopeActivation, registerScopedService } from '#/_base/di/scope';
+import { IEventBus } from '#/app/event/eventBus';
+import {
+  IAgentLifecycleService,
+  MAIN_AGENT_ID,
+} from '#/session/agentLifecycle/agentLifecycle';
+import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata';
+
+import type { SessionTurnOutcome } from './sessionActivity';
+import { ISessionOutcomeMirror } from './sessionOutcomeMirror';
+
+export class SessionOutcomeMirror extends Disposable implements ISessionOutcomeMirror {
+  declare readonly _serviceBrand: undefined;
+
+  private lastPersisted: SessionTurnOutcome | undefined;
+  private adopted = false;
+  private turnStartedHere = false;
+  private mainSubscription: DisposableStore | undefined;
+
+  constructor(
+    @IAgentLifecycleService private readonly agents: IAgentLifecycleService,
+    @ISessionMetadata private readonly metadata: ISessionMetadata,
+  ) {
+    super();
+    void this.metadata
+      .read()
+      .then((meta) => {
+        if (!this.adopted) this.lastPersisted = meta.lastTurnReason;
+      })
+      .catch(() => {});
+    this.attachMain();
+    this._register(this.agents.onDidCreate((handle) => {
+      if (handle.id === MAIN_AGENT_ID) this.attachMain();
+    }));
+    this._register(this.agents.onDidDispose((agentId) => {
+      if (agentId !== MAIN_AGENT_ID) return;
+      this.mainSubscription?.dispose();
+      this.mainSubscription = undefined;
+    }));
+    this._register({
+      dispose: () => {
+        this.mainSubscription?.dispose();
+        this.mainSubscription = undefined;
+      },
+    });
+  }
+
+  private attachMain(): void {
+    if (this.mainSubscription !== undefined) return;
+    const bus = this.agents.get(MAIN_AGENT_ID)?.accessor.get(IEventBus) as IEventBus | undefined;
+    if (bus === undefined) return;
+    const subscription = new DisposableStore();
+    this.mainSubscription = subscription;
+    subscription.add(
+      bus.subscribe('turn.ended', (event) => {
+        if (event.type !== 'turn.ended') return;
+        const reason = (event as { reason?: unknown }).reason;
+        const interruptReason = (event as { interruptReason?: unknown }).interruptReason;
+        if (reason === 'completed') {
+          this.write('completed');
+          return;
+        }
+        if (reason === 'failed' || reason === 'blocked') {
+          this.write('failed');
+          return;
+        }
+        if (reason === 'cancelled' && interruptReason === 'user_cancelled') {
+          this.write('cancelled');
+        }
+      }),
+    );
+    subscription.add(
+      bus.subscribe('turn.started', () => {
+        this.turnStartedHere = true;
+        this.write(undefined);
+      }),
+    );
+    subscription.add(
+      bus.subscribe('agent.activity.updated', (event) => {
+        if (this.turnStartedHere) return;
+        if (this.lastPersisted !== undefined) return;
+        const lastTurn = (event as { lastTurn?: { reason?: unknown } }).lastTurn;
+        const reason = lastTurn?.reason;
+        if (reason === 'completed') this.write('completed', { touchUpdatedAt: false });
+        else if (reason === 'failed' || reason === 'blocked') {
+          this.write('failed', { touchUpdatedAt: false });
+        }
+      }),
+    );
+  }
+
+  private write(
+    outcome: SessionTurnOutcome | undefined,
+    opts?: { readonly touchUpdatedAt?: boolean },
+  ): void {
+    if (outcome === this.lastPersisted) return;
+    this.adopted = true;
+    const previous = this.lastPersisted;
+    this.lastPersisted = outcome;
+    void this.metadata.update({ lastTurnReason: outcome }, opts).catch(() => {
+      if (this.lastPersisted === outcome) this.lastPersisted = previous;
+    });
+  }
+}
+
+registerScopedService(
+  LifecycleScope.Session,
+  ISessionOutcomeMirror,
+  SessionOutcomeMirror,
+  ScopeActivation.OnScopeCreated,
+  'sessionActivity',
+);

packages/agent-core-v2/src/session/sessionMetadata/sessionMetadata.ts

@@ -3,7 +3,8 @@
  *
  * Defines the `SessionMeta` model and the `ISessionMetadata` used by upper
  * layers to read and update the session's durable metadata (title, timestamps,
- * archived flag, fork provenance). Owns the in-memory copy, persists it as a
+ * archived flag, fork provenance, the latest main turn's terminal outcome).
+ * Owns the in-memory copy, persists it as a
  * single atomic document through `storage`, and notifies changes via
  * `onDidChangeMetadata`. Session-scoped — one instance per session. The initial
  * document is materialized when the session is created.
@@ -36,6 +37,7 @@ export interface SessionMeta {
   readonly forkedFrom?: string;
   readonly agents?: Readonly<Record<string, AgentMeta>>;
   readonly custom?: Record<string, unknown>;
+  readonly lastTurnReason?: 'completed' | 'cancelled' | 'failed';
 }
 
 export type SessionMetaPatch = Partial<Omit<SessionMeta, 'id' | 'createdAt'>>;
@@ -50,7 +52,7 @@ export interface ISessionMetadata {
   readonly ready: Promise<void>;
   readonly onDidChangeMetadata: Event<SessionMetadataChangedEvent>;
   read(): Promise<SessionMeta>;
-  update(patch: SessionMetaPatch): Promise<void>;
+  update(patch: SessionMetaPatch, opts?: { readonly touchUpdatedAt?: boolean }): Promise<void>;
   setTitle(title: string): Promise<void>;
   setArchived(archived: boolean): Promise<void>;
   registerAgent(agentId: string, meta: AgentMeta): Promise<void>;

packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts

@@ -23,7 +23,10 @@
  * derived store (no mirror flush, no query-store lock). First-time creation in
  * `load()` records too — a new session must appear in listings immediately
  * (the mirror's pending queue feeds the index's read-your-writes merge);
- * loading an *existing* document (session resume) stays silent.
+ * loading an *existing* document (session resume) stays silent. Queued writes
+ * are tracked in a module-level pending set, drained through
+ * `drainSessionMetadataWrites()` by hosts before the sessions root may be
+ * torn down (the query-store/mirror drain pattern).
  */
 
 import { Disposable } from '#/_base/di/lifecycle';
@@ -48,13 +51,21 @@ import {
 
 const META_KEY = 'state.json';
 
+const pendingWrites = new Set<Promise<void>>();
+
+export async function drainSessionMetadataWrites(): Promise<void> {
+  await Promise.all(pendingWrites);
+}
+
 export const sessionMetadataDataKey = defineState<SessionMeta | undefined>(
   'sessionMetadata.data',
   () => undefined,
 );
 
 export class SessionMetadata extends Disposable implements ISessionMetadata {
   declare readonly _serviceBrand: undefined;
+
+  private disposed = false;
   readonly ready: Promise<void>;
   readonly onDidChangeMetadata: Event<SessionMetadataChangedEvent>;
 
@@ -72,6 +83,11 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
     @ISessionIndexMirror private readonly mirror: ISessionIndexMirror,
   ) {
     super();
+    this._register({
+      dispose: () => {
+        this.disposed = true;
+      },
+    });
     this.states.register(sessionMetadataDataKey);
     this.scope = ctx.metaScope;
     this.onDidChangeMetadata = this._onDidChangeMetadata.event;
@@ -91,14 +107,26 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
     return this.data;
   }
 
-  async update(patch: SessionMetaPatch): Promise<void> {
-    return this.enqueueUpdate(() => this.applyUpdate(patch));
+  async update(
+    patch: SessionMetaPatch,
+    opts?: { readonly touchUpdatedAt?: boolean },
+  ): Promise<void> {
+    return this.enqueueUpdate(() => this.applyUpdate(patch, opts));
   }
 
-  private async applyUpdate(patch: SessionMetaPatch): Promise<void> {
+  private async applyUpdate(
+    patch: SessionMetaPatch,
+    opts?: { readonly touchUpdatedAt?: boolean },
+  ): Promise<void> {
     await this.ready;
-    this.data = { ...this.data, ...patch, updatedAt: Date.now() };
+    // The update queue outlives its callers: a patch queued before scope
+    // disposal must not write (or recreate files) once the session is being
+    // torn down.
+    if (this.disposed) return;
+    const updatedAt = opts?.touchUpdatedAt === false ? this.data.updatedAt : Date.now();
+    this.data = { ...this.data, ...patch, updatedAt };
     await this.store.set(this.scope, META_KEY, this.data);
+    if (this.disposed) return;
     this.mirrorToReadModel();
     this._onDidChangeMetadata.fire({
       changed: Object.keys(patch) as (keyof SessionMeta)[],
@@ -125,7 +153,10 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
 
   private enqueueUpdate(work: () => Promise<void>): Promise<void> {
     const run = this.updateQueue.then(work, work);
-    this.updateQueue = run.catch(() => {});
+    const tracked = run.catch(() => {});
+    this.updateQueue = tracked;
+    pendingWrites.add(tracked);
+    void tracked.finally(() => pendingWrites.delete(tracked));
     return run;
   }
 
@@ -141,6 +172,7 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
         updatedAt: this.data.updatedAt,
         archived: this.data.archived === true,
         custom: this.data.custom,
+        lastTurnReason: this.data.lastTurnReason,
       }),
     );
   }