fix:checkpoint,issue #157

src/core/tdai-core.ts

@@ -514,6 +514,7 @@ export class TdaiCore {
     this.schedulerStartPromise = (async () => {
       try {
         const checkpoint = new CheckpointManager(this.dataDir, this.logger);
+        await checkpoint.recalibrate({ vectorStore: this.vectorStore });
         const cp = await checkpoint.read();
         scheduler.start(checkpoint.getAllPipelineStates(cp));
         this.logger.debug?.(`${TAG} Scheduler started`);

src/utils/checkpoint.test.ts

@@ -0,0 +1,155 @@
+import fs from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { afterEach, describe, expect, it } from "vitest";
+import { CheckpointManager, type Checkpoint } from "./checkpoint.js";
+import type { IMemoryStore } from "../core/store/types.js";
+
+const tempDirs: string[] = [];
+
+function makeCheckpoint(overrides: Partial<Checkpoint> = {}): Checkpoint {
+  return {
+    last_captured_timestamp: 0,
+    total_processed: 0,
+    last_persona_at: 0,
+    last_persona_time: "",
+    request_persona_update: false,
+    persona_update_reason: "",
+    memories_since_last_persona: 0,
+    scenes_processed: 0,
+    runner_states: {},
+    pipeline_states: {},
+    l0_conversations_count: 0,
+    total_memories_extracted: 0,
+    ...overrides,
+  };
+}
+
+async function makeTempDataDir(): Promise<string> {
+  const dir = await fs.mkdtemp(path.join(os.tmpdir(), "tdai-checkpoint-"));
+  tempDirs.push(dir);
+  return dir;
+}
+
+async function writeJsonlLines(dir: string, fileName: string, count: number): Promise<void> {
+  await fs.mkdir(dir, { recursive: true });
+  const lines = Array.from({ length: count }, (_, i) => JSON.stringify({ id: i }));
+  await fs.writeFile(path.join(dir, fileName), `${lines.join("\n")}\n`, "utf-8");
+}
+
+function makeVectorStore(params: {
+  countL0: () => number | Promise<number>;
+  degraded?: boolean;
+}): IMemoryStore {
+  return {
+    isDegraded: () => params.degraded ?? false,
+    countL0: params.countL0,
+  } as unknown as IMemoryStore;
+}
+
+afterEach(async () => {
+  await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })));
+});
+
+describe("CheckpointManager.recalibrate", () => {
+  it("overwrites drifted counters from real storage without touching split session states", async () => {
+    const dataDir = await makeTempDataDir();
+    const manager = new CheckpointManager(dataDir);
+    const runnerStates = {
+      "session-a": {
+        last_captured_timestamp: 123,
+        last_l1_cursor: 456,
+        last_scene_name: "planning",
+      },
+    };
+    const pipelineStates = {
+      "session-a": {
+        conversation_count: 3,
+        last_extraction_time: "2026-01-01T00:00:00.000Z",
+        last_extraction_updated_time: "2026-01-01T00:01:00.000Z",
+        last_active_time: 789,
+        l2_pending_l1_count: 2,
+        warmup_threshold: 4,
+        l2_last_extraction_time: "2026-01-01T00:02:00.000Z",
+      },
+    };
+
+    await manager.write(makeCheckpoint({
+      total_memories_extracted: 50,
+      l0_conversations_count: 50,
+      runner_states: runnerStates,
+      pipeline_states: pipelineStates,
+    }));
+    await writeJsonlLines(path.join(dataDir, "records"), "2026-01-01.jsonl", 42);
+    await writeJsonlLines(path.join(dataDir, "conversations"), "2026-01-01.jsonl", 9);
+
+    await manager.recalibrate({ vectorStore: makeVectorStore({ countL0: () => 17 }) });
+
+    const cp = await manager.read();
+    expect(cp.total_memories_extracted).toBe(42);
+    expect(cp.l0_conversations_count).toBe(17);
+    expect(cp.runner_states).toEqual(runnerStates);
+    expect(cp.pipeline_states).toEqual(pipelineStates);
+  });
+
+  it("falls back to conversations JSONL when vectorStore countL0 fails", async () => {
+    const dataDir = await makeTempDataDir();
+    const manager = new CheckpointManager(dataDir);
+
+    await manager.write(makeCheckpoint({
+      total_memories_extracted: 50,
+      l0_conversations_count: 50,
+    }));
+    await writeJsonlLines(path.join(dataDir, "records"), "2026-01-01.jsonl", 2);
+    await fs.mkdir(path.join(dataDir, "conversations"), { recursive: true });
+    await fs.writeFile(
+      path.join(dataDir, "conversations", "2026-01-01.jsonl"),
+      '{"id":1}\n\n{"id":2}\r\n   \n{"id":3}\n',
+      "utf-8",
+    );
+
+    await manager.recalibrate({
+      vectorStore: makeVectorStore({
+        countL0: () => {
+          throw new Error("db unavailable");
+        },
+      }),
+    });
+
+    const cp = await manager.read();
+    expect(cp.total_memories_extracted).toBe(2);
+    expect(cp.l0_conversations_count).toBe(3);
+  });
+
+  it("falls back to zero when records and conversations directories do not exist", async () => {
+    const dataDir = await makeTempDataDir();
+    const manager = new CheckpointManager(dataDir);
+
+    await manager.write(makeCheckpoint({
+      total_memories_extracted: 50,
+      l0_conversations_count: 50,
+    }));
+
+    await manager.recalibrate();
+
+    const cp = await manager.read();
+    expect(cp.total_memories_extracted).toBe(0);
+    expect(cp.l0_conversations_count).toBe(0);
+  });
+});
+
+describe("CheckpointManager.captureAtomically", () => {
+  it("increments l0_conversations_count by captured message rows", async () => {
+    const dataDir = await makeTempDataDir();
+    const manager = new CheckpointManager(dataDir);
+
+    await manager.captureAtomically("session-a", undefined, async () => ({
+      maxTimestamp: 123,
+      messageCount: 3,
+    }));
+
+    const cp = await manager.read();
+    expect(cp.total_processed).toBe(3);
+    expect(cp.l0_conversations_count).toBe(3);
+  });
+});

src/utils/checkpoint.ts

@@ -27,6 +27,7 @@
 import fs from "node:fs/promises";
 import path from "node:path";
 import { randomBytes } from "node:crypto";
+import type { IMemoryStore } from "../core/store/types.js";
 
 // ============================
 // Types
@@ -144,6 +145,10 @@ export interface CheckpointLogger {
   warn?(msg: string): void;
 }
 
+export interface RecalibrateOptions {
+  vectorStore?: IMemoryStore;
+}
+
 const noopLogger: CheckpointLogger = { info() {} };
 
 // ============================
@@ -154,6 +159,7 @@ const noopLogger: CheckpointLogger = { info() {} };
 // coordinate instance creation.
 
 const fileLocks = new Map<string, Promise<void>>();
+const JSONL_FILE_PATTERN = /\.jsonl$/;
 
 /**
  * Serialize async critical sections per file path.
@@ -178,11 +184,40 @@ async function withFileLock<T>(filePath: string, fn: () => Promise<T>): Promise<
   }
 }
 
+async function countJsonlNonEmptyLines(dirPath: string, logger: CheckpointLogger): Promise<number> {
+  let entries: import("node:fs").Dirent[];
+  try {
+    entries = await fs.readdir(dirPath, { withFileTypes: true });
+  } catch {
+    return 0;
+  }
+
+  let total = 0;
+  for (const entry of entries) {
+    if (!entry.isFile() || !JSONL_FILE_PATTERN.test(entry.name)) continue;
+
+    const filePath = path.join(dirPath, entry.name);
+    try {
+      const raw = await fs.readFile(filePath, "utf-8");
+      total += raw.split(/\r?\n/).filter((line) => line.trim()).length;
+    } catch (err) {
+      logger.warn?.(
+        `[checkpoint] Failed to count JSONL lines in ${filePath}: ` +
+        `${err instanceof Error ? err.message : String(err)}`,
+      );
+    }
+  }
+
+  return total;
+}
+
 export class CheckpointManager {
+  private dataDir: string;
   private filePath: string;
   private logger: CheckpointLogger;
 
   constructor(dataDir: string, logger?: CheckpointLogger) {
+    this.dataDir = dataDir;
     this.filePath = path.join(dataDir, ".metadata", "recall_checkpoint.json");
     this.logger = logger ?? noopLogger;
   }
@@ -293,6 +328,47 @@ export class CheckpointManager {
     return withFileLock(this.filePath, () => this.writeRaw(checkpoint));
   }
 
+  /**
+   * Reconcile drift-prone aggregate counters with the real backing data.
+   *
+   * L1 count is derived from records/*.jsonl because those files are the local
+   * append-only source for persisted memories. L0 prefers VectorStore when
+   * available, then falls back to conversations/*.jsonl.
+   */
+  async recalibrate(opts: RecalibrateOptions = {}): Promise<void> {
+    const totalMemoriesExtracted = await countJsonlNonEmptyLines(
+      path.join(this.dataDir, "records"),
+      this.logger,
+    );
+    const l0ConversationsCount = await this.resolveL0Count(opts.vectorStore);
+
+    const cp = await this.mutate((cp) => {
+      cp.total_memories_extracted = totalMemoriesExtracted;
+      cp.l0_conversations_count = l0ConversationsCount;
+    });
+
+    this.logger.info(
+      `[checkpoint] recalibrated counters: ` +
+      `total_memories_extracted=${cp.total_memories_extracted}, ` +
+      `l0_conversations_count=${cp.l0_conversations_count}`,
+    );
+  }
+
+  private async resolveL0Count(vectorStore?: IMemoryStore): Promise<number> {
+    if (vectorStore && !vectorStore.isDegraded()) {
+      try {
+        return await vectorStore.countL0();
+      } catch (err) {
+        this.logger.warn?.(
+          `[checkpoint] Failed to count L0 via VectorStore; falling back to JSONL: ` +
+          `${err instanceof Error ? err.message : String(err)}`,
+        );
+      }
+    }
+
+    return countJsonlNonEmptyLines(path.join(this.dataDir, "conversations"), this.logger);
+  }
+
   // ============================
   // Public API — mutating (all serialized via file lock)
   // ============================
@@ -479,8 +555,8 @@ export class CheckpointManager {
         // Global stats (aggregate only — not used for filtering)
         cp.last_captured_timestamp = Math.max(cp.last_captured_timestamp, result.maxTimestamp);
         cp.total_processed += result.messageCount;
-        // Increment L0 conversation count (was a separate mutate() call before)
-        cp.l0_conversations_count += 1;
+        // Increment by message rows to match VectorStore/JSONL recalibration.
+        cp.l0_conversations_count += result.messageCount;
       }
     });
   }