feat(agent-core-v2): self-heal corrupted wire journals during restore

Author: elkaix

.changeset/wire-journal-corruption-recovery.md

@@ -0,0 +1,5 @@
+---
+"@pymodel/pythinker-code": patch
+---
+
+Fix sessions that fail to resume when their session journal is truncated or corrupted, for example after the disk fills up.

packages/agent-core-v2/src/app/telemetry/events.ts

@@ -479,6 +479,13 @@ export interface SessionLoadFailedEvent {
   reason: string;
 }
 
+export interface WireRepairEvent {
+  kind: 'corrupted' | 'truncated';
+  outcome: 'repaired' | 'failed';
+  dropped_count: number;
+  backup_created: boolean;
+}
+
 export interface FirstLaunchEvent {}
 
 export interface ExitEvent {
@@ -1058,6 +1065,16 @@ export const telemetryEventDefinitions = {
     comment: 'A session resume fails.',
     properties: { reason: 'Error code, error name, or unknown' },
   }),
+  wire_repair: defineTelemetryEvent<WireRepairEvent>({
+    owner: 'pythinker-code',
+    comment: 'A corrupted wire journal is truncated to its valid prefix and healed on disk.',
+    properties: {
+      kind: 'Corruption kind: unparseable middle line or torn final line',
+      outcome: 'Whether the on-disk repair succeeded',
+      dropped_count: 'Journal lines dropped from the corrupted tail',
+      backup_created: 'Whether a first-time .bak backup of the corrupted file was created',
+    },
+  }),
   first_launch: defineTelemetryEvent<FirstLaunchEvent>({
     owner: 'pythinker-code',
     comment: 'The CLI runs for the first time on this device.',

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

@@ -7,6 +7,7 @@ import {
   AppendLogCorruptedError,
   IAppendLogStore,
   type AppendLogOptions,
+  type AppendLogReadOptions,
 } from '#/persistence/interface/appendLogStore';
 
 const textEncoder = new TextEncoder();
@@ -48,8 +49,9 @@ export class AppendLogStore implements IAppendLogStore {
     this.scheduleFlush(scope, key, state);
   }
 
-  async *read<R>(scope: string, key: string): AsyncIterable<R> {
+  async *read<R>(scope: string, key: string, options?: AppendLogReadOptions): AsyncIterable<R> {
     await this.flushLog(scope, key);
+    const onTruncate = options?.onTruncate;
     const textDecoder = new TextDecoder();
     let pending = '';
     let lineNumber = 0;
@@ -60,7 +62,14 @@ export class AppendLogStore implements IAppendLogStore {
         const raw = pending.slice(0, newlineIndex);
         pending = pending.slice(newlineIndex + 1);
         lineNumber++;
-        const record = this.parseLine<R>(raw, scope, key, lineNumber, false);
+        let record: R | undefined;
+        try {
+          record = this.parseLine<R>(raw, scope, key, lineNumber, false);
+        } catch (error) {
+          if (onTruncate === undefined) throw error;
+          onTruncate({ lineNumber, reason: 'corrupted', cause: error });
+          return;
+        }
         if (record !== undefined) yield record;
         newlineIndex = pending.indexOf('\n');
       }
@@ -69,7 +78,12 @@ export class AppendLogStore implements IAppendLogStore {
     if (pending.length > 0) {
       lineNumber++;
       const record = this.parseLine<R>(pending, scope, key, lineNumber, true);
-      if (record !== undefined) yield record;
+      if (record !== undefined) {
+        yield record;
+      } else if (onTruncate !== undefined) {
+        const line = pending.endsWith('\r') ? pending.slice(0, -1) : pending;
+        if (line.length > 0) onTruncate({ lineNumber, reason: 'truncated' });
+      }
     }
   }
 

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

@@ -21,11 +21,21 @@ export interface AppendLogOptions {
   readonly onError?: (error: unknown) => void;
 }
 
+export interface AppendLogTruncation {
+  readonly lineNumber: number;
+  readonly reason: 'corrupted' | 'truncated';
+  readonly cause?: unknown;
+}
+
+export interface AppendLogReadOptions {
+  readonly onTruncate?: (truncation: AppendLogTruncation) => void;
+}
+
 export interface IAppendLogStore {
   readonly _serviceBrand: undefined;
 
   append<R>(scope: string, key: string, record: R, options?: AppendLogOptions): void;
-  read<R>(scope: string, key: string): AsyncIterable<R>;
+  read<R>(scope: string, key: string, options?: AppendLogReadOptions): AsyncIterable<R>;
   rewrite<R>(scope: string, key: string, records: readonly R[]): Promise<void>;
   flush(): Promise<void>;
   close(): Promise<void>;

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

@@ -0,0 +1,77 @@
+import type { ILogService } from '#/_base/log/log';
+import type { ITelemetryService } from '#/app/telemetry/telemetry';
+import type {
+  AppendLogTruncation,
+  IAppendLogStore,
+} from '#/persistence/interface/appendLogStore';
+import type { IFileSystemStorageService } from '#/persistence/interface/storage';
+
+export interface WireJournalRepairServices {
+  readonly appendLog: IAppendLogStore;
+  readonly storage: IFileSystemStorageService;
+  readonly log: ILogService;
+  readonly telemetry: ITelemetryService;
+}
+
+export function wireJournalBackupKey(key: string): string {
+  return `${key}.bak`;
+}
+
+export async function repairWireJournal(
+  services: WireJournalRepairServices,
+  scope: string,
+  key: string,
+  records: readonly unknown[],
+  truncation: AppendLogTruncation,
+): Promise<void> {
+  const { appendLog, storage, log, telemetry } = services;
+  let backupCreated = false;
+  let outcome: 'repaired' | 'failed' = 'repaired';
+  let droppedCount = 0;
+  let repairError: unknown;
+  try {
+    const original = await storage.read(scope, key);
+    if (original !== undefined) {
+      droppedCount = Math.max(0, countJournalLines(original) - records.length);
+      const backupKey = wireJournalBackupKey(key);
+      if ((await storage.size(scope, backupKey)) === undefined) {
+        await storage.write(scope, backupKey, original, { atomic: true });
+        backupCreated = true;
+      }
+    }
+    await appendLog.rewrite(scope, key, records);
+  } catch (error) {
+    outcome = 'failed';
+    repairError = error;
+  }
+  log.warn('corrupted wire journal truncated to its valid prefix', {
+    scope,
+    key,
+    lineNumber: truncation.lineNumber,
+    reason: truncation.reason,
+    outcome,
+    droppedCount,
+    backupCreated,
+    error: repairError instanceof Error ? repairError.message : undefined,
+  });
+  telemetry.track2('wire_repair', {
+    kind: truncation.reason,
+    outcome,
+    dropped_count: droppedCount,
+    backup_created: backupCreated,
+  });
+}
+
+function countJournalLines(data: Uint8Array): number {
+  let lines = 0;
+  let hasContent = false;
+  for (const byte of data) {
+    if (byte === 0x0a) {
+      lines++;
+      hasContent = false;
+    } else {
+      hasContent = true;
+    }
+  }
+  return hasContent ? lines + 1 : lines;
+}

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

@@ -1,15 +1,21 @@
 import { onUnexpectedError } from '#/_base/errors/unexpectedError';
 import { Service } from '#/_base/di/service';
+import { ILogService } from '#/_base/log/log';
 import { LifecycleScope } from '#/app/scopes';
 import { ScopeActivation, registerScopedService } from '#/_base/di/scope';
 import { IAgentBlobService } from '#/agent/blob/agentBlobService';
 import { IAgentScopeContext } from '#/agent/scopeContext/scopeContext';
+import { ITelemetryService } from '#/app/telemetry/telemetry';
 import type { ContentPart } from '#/kosong/contract/message';
-import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
-import { StorageError, StorageErrors } from '#/persistence/interface/storage';
+import {
+  type AppendLogTruncation,
+  IAppendLogStore,
+} from '#/persistence/interface/appendLogStore';
+import { IFileSystemStorageService, StorageError, StorageErrors } from '#/persistence/interface/storage';
 
 import { IWireService } from './wire';
 import { WireError, WireErrors } from './errors';
+import { repairWireJournal } from './repair';
 import {
   WIRE_PROTOCOL_VERSION,
   isNewerWireVersion,
@@ -38,14 +44,18 @@ export class WireService extends Service implements IWireService {
     @IAgentScopeContext scopeContext: IAgentScopeContext,
     @IAppendLogStore private readonly log: IAppendLogStore,
     @IAgentBlobService private readonly blobService: IAgentBlobService,
+    @IFileSystemStorageService private readonly storage: IFileSystemStorageService,
+    @ILogService private readonly logger: ILogService,
+    @ITelemetryService private readonly telemetry: ITelemetryService,
   ) {
     super();
     this.wireScope = scopeContext.scope();
     this._register(this.log.acquire(this.wireScope, AGENT_WIRE_RECORD_KEY));
   }
 
   async seal(): Promise<void> {
-    for await (const record of this.log.read(this.wireScope, AGENT_WIRE_RECORD_KEY)) {
+    const tolerate = { onTruncate: () => {} };
+    for await (const record of this.log.read(this.wireScope, AGENT_WIRE_RECORD_KEY, tolerate)) {
       void record;
       return;
     }
@@ -78,7 +88,12 @@ export class WireService extends Service implements IWireService {
   }
 
   async *readJournal(): AsyncIterable<WireRecord> {
-    const source = this.log.read<WireRecord>(this.wireScope, AGENT_WIRE_RECORD_KEY);
+    let truncation: AppendLogTruncation | undefined;
+    const source = this.log.read<WireRecord>(this.wireScope, AGENT_WIRE_RECORD_KEY, {
+      onTruncate: (info) => {
+        truncation = info;
+      },
+    });
     let migrations: readonly WireMigration[] = [];
     let rewrittenRecords: WireRecord[] | undefined;
     let newerWireVersion = false;
@@ -128,11 +143,42 @@ export class WireService extends Service implements IWireService {
     if (!hasRecords) {
       rewrittenRecords = [createWireMetadataRecord()];
     }
-    if (rewrittenRecords !== undefined) {
+    if (truncation !== undefined) {
+      await this.repairJournal(truncation, rewrittenRecords);
+    } else if (rewrittenRecords !== undefined) {
       await this.log.rewrite(this.wireScope, AGENT_WIRE_RECORD_KEY, rewrittenRecords);
     }
   }
 
+  private async repairJournal(
+    truncation: AppendLogTruncation,
+    rewrittenRecords: WireRecord[] | undefined,
+  ): Promise<void> {
+    let records: WireRecord[] = rewrittenRecords ?? [];
+    if (rewrittenRecords === undefined) {
+      const tolerate = { onTruncate: () => {} };
+      for await (const record of this.log.read<WireRecord>(
+        this.wireScope,
+        AGENT_WIRE_RECORD_KEY,
+        tolerate,
+      )) {
+        records.push(record);
+      }
+    }
+    await repairWireJournal(
+      {
+        appendLog: this.log,
+        storage: this.storage,
+        log: this.logger,
+        telemetry: this.telemetry,
+      },
+      this.wireScope,
+      AGENT_WIRE_RECORD_KEY,
+      records,
+      truncation,
+    );
+  }
+
   async flush(): Promise<void> {
     await this.persistQueue;
     await this.log.flush();

packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycleService.ts

@@ -10,6 +10,7 @@ import {
 } from '#/_base/di/scope';
 import { unwrapErrorCause } from '#/_base/errors/errors';
 import { AsyncEmitter, Emitter, type Event, type IWaitUntil } from '#/_base/event';
+import { ILogService } from '#/_base/log/log';
 import { drainLogCloses } from '#/_base/log/logService';
 import { DEFAULT_PLAN_MODE_SECTION } from '#/features/plan/configSection';
 import { IAgentPlanService } from '#/features/plan/plan';
@@ -27,8 +28,12 @@ import {
 import { ITelemetryService } from '#/app/telemetry/telemetry';
 import { ErrorCodes, Error2, isError2 } from '#/errors';
 import { IHostFileSystem, type HostDirEntry } from '#/os/interface/hostFileSystem';
-import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
+import {
+  type AppendLogTruncation,
+  IAppendLogStore,
+} from '#/persistence/interface/appendLogStore';
 import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
+import { IFileSystemStorageService } from '#/persistence/interface/storage';
 import {
   IAgentLifecycleService,
   MAIN_AGENT_ID,
@@ -51,6 +56,7 @@ import {
   createWireMetadataRecord,
   type WireRecord,
 } from '#/wire/record';
+import { repairWireJournal } from '#/wire/repair';
 import { IModelCatalog } from '#/kosong/model/catalog';
 import { IModelService } from '#/kosong/model/model';
 import { IProviderService } from '#/kosong/provider/provider';
@@ -145,6 +151,8 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
     @ISessionIndexMirror private readonly indexMirror: ISessionIndexMirror,
     @IAppendLogStore private readonly appendLogStore: IAppendLogStore,
     @IAtomicDocumentStore private readonly docs: IAtomicDocumentStore,
+    @IFileSystemStorageService private readonly storage: IFileSystemStorageService,
+    @ILogService private readonly log: ILogService,
     @IHostFileSystem private readonly hostFs: IHostFileSystem,
     @IEventService private readonly event: IEventService,
     @ITelemetryService private readonly telemetry: ITelemetryService,
@@ -663,12 +671,30 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
         await agentHandle.accessor.get(IEventDispatcher).flush();
       }
     }
-    return collect(
-      this.appendLogStore.read<WireRecord>(
-        agentScopeOf(sessionScopeOf(this.handlerScope, sourceSessionId), agentId),
-        AGENT_WIRE_RECORD_KEY,
-      ),
+    const scope = agentScopeOf(sessionScopeOf(this.handlerScope, sourceSessionId), agentId);
+    let truncation: AppendLogTruncation | undefined;
+    const records = await collect(
+      this.appendLogStore.read<WireRecord>(scope, AGENT_WIRE_RECORD_KEY, {
+        onTruncate: (info) => {
+          truncation = info;
+        },
+      }),
     );
+    if (truncation !== undefined) {
+      await repairWireJournal(
+        {
+          appendLog: this.appendLogStore,
+          storage: this.storage,
+          log: this.log,
+          telemetry: this.telemetry,
+        },
+        scope,
+        AGENT_WIRE_RECORD_KEY,
+        records,
+        truncation,
+      );
+    }
+    return records;
   }
 
   private async pruneTruncatedForkFiles(