fix(agent-core-v2): recover session persistence after disk-full instead of losing buffered records

.changeset/rich-pandas-heal.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fix loss of session history when the disk runs out of space mid-session.

packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts

@@ -3,7 +3,11 @@ import { LifecycleScope } from '#/app/scopes';
 import { ScopeActivation, registerScopedService } from '#/_base/di/scope';
 import { Emitter, type Event } from '#/_base/event';
 
-import { IFileSystemStorageService } from '#/persistence/interface/storage';
+import {
+  IFileSystemStorageService,
+  StorageError,
+  StorageErrors,
+} from '#/persistence/interface/storage';
 import {
   AppendLogCorruptedError,
   IAppendLogStore,
@@ -14,6 +18,24 @@ import {
 
 const textEncoder = new TextEncoder();
 
+const RECOVERY_BASE_DELAY_MS = 1_000;
+const RECOVERY_MAX_DELAY_MS = 30_000;
+const NEWLINE = 0x0a;
+
+interface RecoveryState {
+  readonly failedBatch: readonly unknown[];
+  attempts: number;
+  timer: ReturnType<typeof setTimeout> | undefined;
+}
+
+function isRecoverableStorageError(error: unknown): boolean {
+  return (
+    error instanceof StorageError &&
+    (error.code === StorageErrors.codes.STORAGE_DISK_FULL ||
+      (StorageErrors.retryable as readonly string[]).includes(error.code))
+  );
+}
+
 const pendingRetirements = new Set<Promise<void>>();
 
 export async function drainAppendLogRetirements(): Promise<void> {
@@ -27,6 +49,7 @@ interface LogState {
   flushPromise: Promise<void> | undefined;
   flushScheduled: boolean;
   storageFailure: { readonly error: unknown } | undefined;
+  recovery: RecoveryState | undefined;
   cutoverEpoch: number;
   refCount: number;
   retired: boolean;
@@ -41,9 +64,18 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
   private readonly logs = new Map<string, LogState>();
   private readonly writeEmitter = this._register(new Emitter<AppendLogWrite>());
   readonly onDidWrite: Event<AppendLogWrite> = this.writeEmitter.event;
+  private readonly recoverEmitter = this._register(new Emitter<AppendLogWrite>());
+  readonly onDidRecover: Event<AppendLogWrite> = this.recoverEmitter.event;
 
   constructor(@IFileSystemStorageService private readonly storage: IFileSystemStorageService) {
     super();
+    this._register(
+      toDisposable(() => {
+        for (const state of this.logs.values()) {
+          if (state.recovery?.timer !== undefined) clearTimeout(state.recovery.timer);
+        }
+      }),
+    );
   }
 
   append<R>(scope: string, key: string, record: R, options?: AppendLogOptions): void {
@@ -114,6 +146,7 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
     const encoded = encodeBatch(records);
     const state = this.state(scope, key);
     state.cutoverEpoch++;
+    this.cancelRecovery(state);
     const prior = state.flushPromise ?? state.ready;
     const priorSettled = prior.then(
       () => undefined,
@@ -126,6 +159,10 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
         return true;
       } catch (error) {
         state.storageFailure = { error };
+        if (state.recovery === undefined && isRecoverableStorageError(error)) {
+          state.recovery = { failedBatch: state.pending.slice(), attempts: 0, timer: undefined };
+          this.scheduleRecovery(scope, key, state);
+        }
         throw error;
       }
     });
@@ -144,6 +181,15 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
   }
 
   async close(): Promise<void> {
+    const recoveries: Promise<void>[] = [];
+    for (const [id, state] of this.logs) {
+      if (state.recovery === undefined) continue;
+      if (state.recovery.timer !== undefined) clearTimeout(state.recovery.timer);
+      state.recovery.timer = undefined;
+      const { scope, key } = fromLogId(id);
+      recoveries.push(this.attemptRecovery(scope, key, state));
+    }
+    await Promise.all(recoveries);
     await this.flush();
   }
 
@@ -169,6 +215,7 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
         flushPromise: undefined,
         flushScheduled: false,
         storageFailure: undefined,
+        recovery: undefined,
         cutoverEpoch: 0,
         refCount: 0,
         retired: false,
@@ -182,6 +229,7 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
 
   private scheduleFlush(scope: string, key: string, state: LogState): void {
     if (state.flushScheduled || state.flushPromise !== undefined) return;
+    if (state.storageFailure !== undefined) return;
     state.flushScheduled = true;
     queueMicrotask(() => {
       state.flushScheduled = false;
@@ -213,13 +261,129 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
 
   private async settleRetiredState(scope: string, key: string, state: LogState): Promise<void> {
     try {
+      if (state.recovery !== undefined) {
+        if (state.recovery.timer !== undefined) clearTimeout(state.recovery.timer);
+        state.recovery.timer = undefined;
+        await this.attemptRecovery(scope, key, state);
+      }
       await this.flushState(scope, key, state);
     } finally {
       const id = logId(scope, key);
       if (this.logs.get(id) === state) this.logs.delete(id);
     }
   }
 
+  private cancelRecovery(state: LogState): void {
+    if (state.recovery?.timer !== undefined) clearTimeout(state.recovery.timer);
+    state.recovery = undefined;
+  }
+
+  private scheduleRecovery(scope: string, key: string, state: LogState): void {
+    const recovery = state.recovery;
+    if (recovery === undefined || state.retired || recovery.timer !== undefined) return;
+    const delay = Math.min(RECOVERY_BASE_DELAY_MS * 2 ** recovery.attempts, RECOVERY_MAX_DELAY_MS);
+    recovery.timer = setTimeout(() => {
+      recovery.timer = undefined;
+      void this.attemptRecovery(scope, key, state);
+    }, delay);
+    recovery.timer.unref();
+  }
+
+  private async attemptRecovery(scope: string, key: string, state: LogState): Promise<void> {
+    const recovery = state.recovery;
+    if (recovery === undefined) return;
+    state.recovery = undefined;
+    try {
+      if (state.flushPromise !== undefined) {
+        await state.flushPromise.catch(() => undefined);
+      }
+      if (state.storageFailure === undefined) return;
+      if (state.recovery !== undefined) return;
+      if (state.flushPromise !== undefined) {
+        state.recovery = recovery;
+        this.scheduleRecovery(scope, key, state);
+        return;
+      }
+      await this.ownFlush(
+        scope,
+        key,
+        state,
+        this.reconcileCommittedTail(scope, key, state, recovery.failedBatch),
+        { value: false },
+      );
+      state.storageFailure = undefined;
+      this.recoverEmitter.fire({ scope, key });
+    } catch (error) {
+      if (state.recovery !== undefined) {
+        state.recovery.attempts = recovery.attempts + 1;
+        if (state.recovery.timer !== undefined) clearTimeout(state.recovery.timer);
+        state.recovery.timer = undefined;
+        this.scheduleRecovery(scope, key, state);
+        return;
+      }
+      if (isRecoverableStorageError(error)) {
+        state.recovery = recovery;
+        recovery.attempts++;
+        this.scheduleRecovery(scope, key, state);
+      }
+    }
+  }
+
+  private async reconcileCommittedTail(
+    scope: string,
+    key: string,
+    state: LogState,
+    failedBatch: readonly unknown[],
+  ): Promise<boolean> {
+    const encoded = encodeBatch(failedBatch);
+    if (encoded.byteLength === 0) return false;
+    const epoch = state.cutoverEpoch;
+    const size = await this.storage.size(scope, key);
+    if (size === undefined || size === 0) return false;
+    const window = Math.min(size, encoded.byteLength);
+    const tail = new Uint8Array(window);
+    let offset = 0;
+    for await (const chunk of this.storage.readStream(scope, key, {
+      start: size - window,
+      end: size - 1,
+    })) {
+      tail.set(chunk, offset);
+      offset += chunk.byteLength;
+    }
+    if (state.cutoverEpoch !== epoch) {
+      throw new StorageError(
+        StorageErrors.codes.STORAGE_CORRUPTED,
+        'append-log recovery aborted: log rewritten concurrently',
+        { details: { scope, key } },
+      );
+    }
+    const overlap = committedTailOverlap(tail, encoded);
+    if (overlap === encoded.byteLength) {
+      state.pending.splice(0, failedBatch.length);
+      return false;
+    }
+    if (overlap > 0) {
+      const full = await this.storage.read(scope, key);
+      if (full === undefined || full.byteLength !== size) {
+        throw new StorageError(
+          StorageErrors.codes.STORAGE_CORRUPTED,
+          'append-log recovery aborted: log changed while reconciling',
+          { details: { scope, key } },
+        );
+      }
+      await this.storage.write(scope, key, full.subarray(0, size - overlap), { atomic: true });
+      return false;
+    }
+    if (tail[tail.byteLength - 1] !== NEWLINE) {
+      throw new StorageError(
+        StorageErrors.codes.STORAGE_CORRUPTED,
+        'append-log tail does not match the failed batch; automatic recovery aborted',
+        { details: { scope, key } },
+      );
+    }
+    return false;
+  }
+
   private ownFlush(
     scope: string,
     key: string,
@@ -279,8 +443,12 @@ export class AppendLogStore extends Disposable implements IAppendLogStore {
         wrote = true;
         if (wroteBox !== undefined) wroteBox.value = true;
       } catch (error) {
-        const failure = (state.storageFailure ??= { error });
-        throw failure.error;
+        state.storageFailure = { error };
+        if (state.recovery === undefined && isRecoverableStorageError(error)) {
+          state.recovery = { failedBatch: batch, attempts: 0, timer: undefined };
+          this.scheduleRecovery(scope, key, state);
+        }
+        throw error;
       }
       if (state.cutoverEpoch !== cutoverEpoch) return wrote;
       state.pending.splice(0, batch.length);
@@ -304,6 +472,18 @@ function encodeBatch(records: readonly unknown[]): Uint8Array {
   return textEncoder.encode(content);
 }
 
+function committedTailOverlap(existing: Uint8Array, batch: Uint8Array): number {
+  const max = Math.min(existing.byteLength, batch.byteLength);
+  for (let k = max; k > 0; k--) {
+    const before = existing.byteLength - k;
+    if (before !== 0 && existing[before - 1] !== NEWLINE) continue;
+    let i = 0;
+    while (i < k && existing[before + i] === batch[i]) i++;
+    if (i === k) return k;
+  }
+  return 0;
+}
+
 registerScopedService(
   LifecycleScope.App,
   IAppendLogStore,

packages/agent-core-v2/src/persistence/interface/appendLogStore.ts

@@ -41,6 +41,7 @@ export interface IAppendLogStore {
   readonly _serviceBrand: undefined;
 
   readonly onDidWrite: Event<AppendLogWrite>;
+  readonly onDidRecover: Event<AppendLogWrite>;
 
   append<R>(scope: string, key: string, record: R, options?: AppendLogOptions): void;
   read<R>(scope: string, key: string, options?: AppendLogReadOptions): AsyncIterable<R>;

packages/agent-core-v2/src/wire/wireService.ts

@@ -11,7 +11,7 @@ import {
   type AppendLogTruncation,
   IAppendLogStore,
 } from '#/persistence/interface/appendLogStore';
-import { IFileSystemStorageService, StorageError, StorageErrors } from '#/persistence/interface/storage';
+import { IFileSystemStorageService, isStorageError, StorageError, StorageErrors } from '#/persistence/interface/storage';
 
 import { IWireService } from './wire';
 import { WireError, WireErrors } from './errors';
@@ -59,6 +59,15 @@ export class WireService extends Service implements IWireService {
     this.wireScope = scopeContext.scope();
     this.agentId = scopeContext.agentId;
     this._register(this.log.acquire(this.wireScope, AGENT_WIRE_RECORD_KEY));
+    this._register(
+      this.log.onDidRecover((write) => {
+        if (write.scope === this.wireScope && write.key === AGENT_WIRE_RECORD_KEY) {
+          this.logger.info('session wire persistence recovered after storage failure', {
+            scope: this.wireScope,
+          });
+        }
+      }),
+    );
   }
 
   async seal(): Promise<void> {
@@ -343,7 +352,15 @@ export class WireService extends Service implements IWireService {
 
   private appendRecordLow(record: WireRecord): void {
     this.log.append(this.wireScope, AGENT_WIRE_RECORD_KEY, record, {
-      onError: onUnexpectedError,
+      onError: (error) => {
+        onUnexpectedError(error);
+        if (isStorageError(error, StorageErrors.codes.STORAGE_DISK_FULL)) {
+          this.logger.error(
+            'session wire persistence degraded: no space left on device; new records stay buffered in memory and the store retries in the background — free disk space to resume durable writes',
+            { scope: this.wireScope },
+          );
+        }
+      },
     });
     this.lines += 1;
     if (record.type === 'context.clear') this.lastClearLine = this.lines;

packages/agent-core-v2/test/harness/agent.ts

@@ -963,6 +963,7 @@ function reassertServiceOverrides(
 class PersistenceAppendLogStore implements IAppendLogStore {
   declare readonly _serviceBrand: undefined;
   readonly onDidWrite: IAppendLogStore['onDidWrite'] = Event.None as IAppendLogStore['onDidWrite'];
+  readonly onDidRecover: IAppendLogStore['onDidRecover'] = Event.None as IAppendLogStore['onDidRecover'];
   private readonly history: WireRecord[] = [];
   private historySeeded = false;