From 444e5ce3fae83d9b797f7c1c15c079871404961f Mon Sep 17 00:00:00 2001 From: Prado-learning Date: Thu, 9 Jul 2026 16:47:09 +0800 Subject: [PATCH] fix:checkpoint,issue #157 --- src/core/tdai-core.ts | 1 + src/utils/checkpoint.test.ts | 155 +++++++++++++++++++++++++++++++++++ src/utils/checkpoint.ts | 80 +++++++++++++++++- 3 files changed, 234 insertions(+), 2 deletions(-) create mode 100644 src/utils/checkpoint.test.ts diff --git a/src/core/tdai-core.ts b/src/core/tdai-core.ts index 9185e51d..b49336ef 100644 --- a/src/core/tdai-core.ts +++ b/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`); diff --git a/src/utils/checkpoint.test.ts b/src/utils/checkpoint.test.ts new file mode 100644 index 00000000..f4f4777e --- /dev/null +++ b/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 { + 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 { + 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 { + 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; + 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); + }); +}); diff --git a/src/utils/checkpoint.ts b/src/utils/checkpoint.ts index 301fc0df..af0098e3 100644 --- a/src/utils/checkpoint.ts +++ b/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>(); +const JSONL_FILE_PATTERN = /\.jsonl$/; /** * Serialize async critical sections per file path. @@ -178,11 +184,40 @@ async function withFileLock(filePath: string, fn: () => Promise): Promise< } } +async function countJsonlNonEmptyLines(dirPath: string, logger: CheckpointLogger): Promise { + 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 { + 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 { + 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; } }); }