diff --git a/index.ts b/index.ts index 868a7701..28cc3d56 100644 --- a/index.ts +++ b/index.ts @@ -31,6 +31,7 @@ import { } from "./src/utils/clean-context-runner.js"; import { SessionFilter } from "./src/utils/session-filter.js"; import { LocalMemoryCleaner } from "./src/utils/memory-cleaner.js"; +import { CheckpointManager } from "./src/utils/checkpoint.js"; import { registerMemoryTdaiCli } from "./src/cli/index.js"; import { initDataDirectories, resetStores } from "./src/utils/pipeline-factory.js"; import { getOrCreateInstanceId, initReporter, report, resetReporter } from "./src/core/report/reporter.js"; @@ -269,6 +270,7 @@ export default function register(api: OpenClawPluginApi) { config: cfg, sessionFilter, }); + const checkpointManager = new CheckpointManager(pluginDataDir, api.logger); // Initialize TdaiCore (async — store init, pipeline wiring) const coreReady = core.initialize().then(() => { @@ -305,10 +307,12 @@ export default function register(api: OpenClawPluginApi) { retentionDays: cfg.memoryCleanup.retentionDays, cleanTime: cfg.memoryCleanup.cleanTime, logger: api.logger, + checkpointManager, }); sharedMemoryCleaner.start(); api.logger.debug?.(`${TAG} Memory cleaner started (singleton)`); } else { + sharedMemoryCleaner.setCheckpointManager(checkpointManager); api.logger.debug?.(`${TAG} Memory cleaner already started in this process, reusing existing instance`); } memoryCleaner = sharedMemoryCleaner; diff --git a/src/core/tdai-core.ts b/src/core/tdai-core.ts index 9185e51d..eb2d4b19 100644 --- a/src/core/tdai-core.ts +++ b/src/core/tdai-core.ts @@ -511,14 +511,44 @@ export class TdaiCore { // Capture scheduler locally so TypeScript narrows inside the closure // even after ``this.scheduler`` is re-assigned by handleSessionEnd. const scheduler = this.scheduler; + const vectorStore = this.vectorStore; + const logger = this.logger; + const dataDir = this.dataDir; this.schedulerStartPromise = (async () => { try { - const checkpoint = new CheckpointManager(this.dataDir, this.logger); + const checkpoint = new CheckpointManager(dataDir, logger); + + // Phase 1: Recalculate checkpoint to fix any counter drift + // This fixes counters AND clamps stale cursors before scheduler starts + try { + const storeCounts = vectorStore + ? { + l0ConversationsCount: await vectorStore.countL0().catch(() => undefined), + totalMemoriesExtracted: await vectorStore.countL1().catch(() => undefined), + } + : undefined; + + const result = await checkpoint.recalculate({ storeCounts }); + logger.debug?.( + `${TAG} Checkpoint recalculated: l0=${result.l0_conversations_count}, ` + + `l1=${result.total_memories_extracted}, ` + + `runner_cursors=${result.runner_cursors_corrected}, ` + + `pipeline_cursors=${result.pipeline_cursors_corrected}`, + ); + } catch (recalcErr) { + // Non-fatal: proceed with stale checkpoint rather than failing startup + logger.warn?.( + `${TAG} Checkpoint recalculation failed (proceeding with stale data): ` + + `${recalcErr instanceof Error ? recalcErr.message : String(recalcErr)}`, + ); + } + + // Phase 2: Start scheduler with recalculated pipeline states const cp = await checkpoint.read(); scheduler.start(checkpoint.getAllPipelineStates(cp)); - this.logger.debug?.(`${TAG} Scheduler started`); + logger.debug?.(`${TAG} Scheduler started`); } catch (err) { - this.logger.error(`${TAG} Failed to restore checkpoint: ${err instanceof Error ? err.message : String(err)}`); + logger.error(`${TAG} Failed to restore checkpoint: ${err instanceof Error ? err.message : String(err)}`); scheduler.start({}); } })(); diff --git a/src/utils/checkpoint.test.ts b/src/utils/checkpoint.test.ts new file mode 100644 index 00000000..f8ab2638 --- /dev/null +++ b/src/utils/checkpoint.test.ts @@ -0,0 +1,1183 @@ +/** + * Unit tests for CheckpointManager. + * + * Tests cover: + * - Decrement methods (decrementL0ConversationCount, decrementMemoriesExtracted, decrementTotalProcessed) + * - Recalibrate method + * - Existing increment methods for regression testing + * - Cleanup scenarios (manual pruning, automatic cleaner, session reset) + * - Design pattern validation (hybrid cursor-first approach) + * + * Uses real filesystem with temporary directories. + */ + +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import fs from "node:fs"; +import path from "node:path"; +import os from "node:os"; +import { CheckpointManager } from "./checkpoint.js"; +import { LocalMemoryCleaner } from "./memory-cleaner.js"; + +/** + * ═══════════════════════════════════════════════════════════════════════════════ + * SCENARIO COVERAGE MATRIX + * ═══════════════════════════════════════════════════════════════════════════════ + * + * | Scenario | Trigger | Test Method | + * |-----------------------------|----------------------------|---------------------| + * | Manual JSONL pruning | User deletes old shards | recalibrate() | + * | Test data cleanup | Dev removes test records | decrement*() | + * | memory-cleaner auto cleanup| Scheduled daily cleanup | recalibrate() | + * | Session reset | User resets a session | decrement*() | + * | Incremental L1 skip fix | After cleanup + new data | recalibrate() | + * | Incremental L2 skip fix | After cleanup + new records| recalibrate() | + * | Counter drift detection | Post-cleanup verification | read() | + * + * ═══════════════════════════════════════════════════════════════════════════════ + */ + +describe("CheckpointManager", () => { + // Track all temp dirs for cleanup + const tempDirs: string[] = []; + + afterEach(async () => { + // Clean up temp directories + for (const dir of tempDirs) { + try { + await fs.promises.rm(dir, { recursive: true, force: true }); + } catch { + // Ignore cleanup errors + } + } + tempDirs.length = 0; + }); + + function createTempDir(): string { + const dir = path.join(os.tmpdir(), `checkpoint-test-${Date.now()}-${Math.random()}`); + fs.mkdirSync(dir, { recursive: true }); + tempDirs.push(dir); + return dir; + } + + // ============================ + // Decrement methods + // ============================ + + describe("decrementL0ConversationCount", () => { + it("decrements l0_conversations_count by 1 when no count provided", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // First, increment via recalibrate to set a known value + await manager.recalibrate({ actualL0Count: 10, actualL1Count: 5, actualTotalProcessed: 100 }); + + await manager.decrementL0ConversationCount(); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(9); + }); + + it("decrements by specified count", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 50, actualL1Count: 25, actualTotalProcessed: 200 }); + + await manager.decrementL0ConversationCount(5); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(45); + }); + + it("clamps to 0 when decrement exceeds current value", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 3, actualL1Count: 1, actualTotalProcessed: 10 }); + + await manager.decrementL0ConversationCount(10); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(0); + }); + + it("handles zero gracefully", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 0, actualL1Count: 0, actualTotalProcessed: 0 }); + + await manager.decrementL0ConversationCount(); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(0); + }); + }); + + describe("decrementMemoriesExtracted", () => { + it("decrements total_memories_extracted by specified count", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 5, actualL1Count: 20, actualTotalProcessed: 100 }); + + await manager.decrementMemoriesExtracted(7); + + const cp = await manager.read(); + expect(cp.total_memories_extracted).toBe(13); + }); + + it("also decrements memories_since_last_persona", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Set up initial checkpoint state + const cp0 = await manager.read(); + cp0.last_persona_at = 50; + cp0.total_memories_extracted = 100; + cp0.memories_since_last_persona = 50; // 100 - 50 + await manager.write(cp0); + + await manager.decrementMemoriesExtracted(10); + + const cp = await manager.read(); + expect(cp.total_memories_extracted).toBe(90); + // memories_since_last_persona = 50 - 10 = 40 + expect(cp.memories_since_last_persona).toBe(40); + }); + + it("clamps total_memories_extracted to 0 when decrement exceeds", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 1, actualL1Count: 5, actualTotalProcessed: 10 }); + + await manager.decrementMemoriesExtracted(100); + + const cp = await manager.read(); + expect(cp.total_memories_extracted).toBe(0); + }); + + it("clamps memories_since_last_persona to 0 when decrement exceeds", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 1, actualL1Count: 10, actualTotalProcessed: 50 }); + + await manager.decrementMemoriesExtracted(100); + + const cp = await manager.read(); + expect(cp.memories_since_last_persona).toBe(0); + }); + }); + + describe("decrementTotalProcessed", () => { + it("decrements total_processed by specified count", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 5, actualL1Count: 20, actualTotalProcessed: 1000 }); + + await manager.decrementTotalProcessed(300); + + const cp = await manager.read(); + expect(cp.total_processed).toBe(700); + }); + + it("clamps to 0 when decrement exceeds current value", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 1, actualL1Count: 5, actualTotalProcessed: 50 }); + + await manager.decrementTotalProcessed(100); + + const cp = await manager.read(); + expect(cp.total_processed).toBe(0); + }); + + it("handles zero gracefully", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 0, actualL1Count: 0, actualTotalProcessed: 0 }); + + await manager.decrementTotalProcessed(1); + + const cp = await manager.read(); + expect(cp.total_processed).toBe(0); + }); + }); + + // ============================ + // Recalibrate method + // ============================ + + describe("recalibrate", () => { + it("resets all counters to provided values", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // First set some values + await manager.recalibrate({ actualL0Count: 100, actualL1Count: 500, actualTotalProcessed: 2000 }); + + // Then recalibrate to new values + const result = await manager.recalibrate({ + actualL0Count: 50, + actualL1Count: 200, + actualTotalProcessed: 1000, + }); + + expect(result.l0_conversations_count).toBe(50); + expect(result.total_memories_extracted).toBe(200); + expect(result.total_processed).toBe(1000); + }); + + it("adjusts memories_since_last_persona by removed memories, not last_persona_at", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + const cp0 = await manager.read(); + cp0.total_memories_extracted = 10; + cp0.memories_since_last_persona = 4; + cp0.last_persona_at = 100; + await manager.write(cp0); + + const result = await manager.recalibrate({ + actualL0Count: 3, + actualL1Count: 8, + actualTotalProcessed: 98, + }); + + expect(result.total_memories_extracted).toBe(8); + expect(result.memories_since_last_persona).toBe(2); + }); + + it("preserves memories_since_last_persona when no memories were removed", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + const cp0 = await manager.read(); + cp0.total_memories_extracted = 100; + cp0.memories_since_last_persona = 40; + cp0.last_persona_at = 60; + await manager.write(cp0); + + const result = await manager.recalibrate({ + actualL0Count: 10, + actualL1Count: 100, + actualTotalProcessed: 500, + }); + + expect(result.memories_since_last_persona).toBe(40); + }); + + it("clamps memories_since_last_persona to 0 when removed memories exceed it", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + const cp0 = await manager.read(); + cp0.total_memories_extracted = 50; + cp0.memories_since_last_persona = 8; + cp0.last_persona_at = 100; + await manager.write(cp0); + + const result = await manager.recalibrate({ + actualL0Count: 5, + actualL1Count: 30, + actualTotalProcessed: 200, + }); + + expect(result.memories_since_last_persona).toBe(0); + }); + + it("clamps negative values to 0", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + const result = await manager.recalibrate({ + actualL0Count: -10, + actualL1Count: -5, + actualTotalProcessed: -100, + }); + + expect(result.l0_conversations_count).toBe(0); + expect(result.total_memories_extracted).toBe(0); + expect(result.total_processed).toBe(0); + }); + + it("handles zero values correctly", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + const result = await manager.recalibrate({ + actualL0Count: 0, + actualL1Count: 0, + actualTotalProcessed: 0, + }); + + expect(result.l0_conversations_count).toBe(0); + expect(result.total_memories_extracted).toBe(0); + expect(result.total_processed).toBe(0); + }); + }); + + // ============================ + // Concurrent modification handling + // ============================ + + describe("concurrent modifications", () => { + it("handles multiple decrements sequentially", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 100, actualL1Count: 50, actualTotalProcessed: 1000 }); + + await manager.decrementL0ConversationCount(10); + await manager.decrementL0ConversationCount(20); + await manager.decrementL0ConversationCount(5); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(65); // 100 - 10 - 20 - 5 = 65 + }); + + it("handles interleaved increments and decrements", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 50, actualL1Count: 25, actualTotalProcessed: 500 }); + + // Simulate increment then decrement + const cp1 = await manager.read(); + cp1.l0_conversations_count += 10; + cp1.total_processed += 100; + await manager.write(cp1); + + await manager.decrementL0ConversationCount(5); + await manager.decrementTotalProcessed(50); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(55); // 50 + 10 - 5 = 55 + expect(cp.total_processed).toBe(550); // 500 + 100 - 50 = 550 + }); + }); + + // ============================ + // File persistence + // ============================ + + describe("file persistence", () => { + it("persists decremented values across manager instances", async () => { + const dataDir = createTempDir(); + + const manager1 = new CheckpointManager(dataDir); + await manager1.recalibrate({ actualL0Count: 100, actualL1Count: 50, actualTotalProcessed: 1000 }); + await manager1.decrementL0ConversationCount(25); + await manager1.decrementTotalProcessed(200); + + // Create new manager instance pointing to same dir + const manager2 = new CheckpointManager(dataDir); + const cp = await manager2.read(); + + expect(cp.l0_conversations_count).toBe(75); + expect(cp.total_processed).toBe(800); + }); + + it("creates checkpoint file after mutation", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // read() alone doesn't create the file (it's a snapshot, not a mutation) + await manager.read(); + + // Mutating operation creates the file + await manager.recalibrate({ actualL0Count: 0, actualL1Count: 0, actualTotalProcessed: 0 }); + + // File should exist now + const checkpointPath = path.join(dataDir, ".metadata", "recall_checkpoint.json"); + expect(fs.existsSync(checkpointPath)).toBe(true); + }); + }); + + // ============================ + // Integration with existing methods + // ============================ + + describe("integration with existing methods", () => { + it("decrement works after captureAtomically", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate some captures + await manager.recalibrate({ actualL0Count: 5, actualL1Count: 3, actualTotalProcessed: 30 }); + + // Decrement + await manager.decrementL0ConversationCount(2); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(3); + }); + + it("decrement works after recalibrate", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Recalibrate + await manager.recalibrate({ actualL0Count: 100, actualL1Count: 50, actualTotalProcessed: 500 }); + + // Decrement + await manager.decrementMemoriesExtracted(10); + await manager.decrementTotalProcessed(100); + + const cp = await manager.read(); + expect(cp.total_memories_extracted).toBe(40); + expect(cp.total_processed).toBe(400); + }); + + it("recalibrate works after decrements", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Initial state + await manager.recalibrate({ actualL0Count: 100, actualL1Count: 50, actualTotalProcessed: 500 }); + + // Decrement + await manager.decrementL0ConversationCount(20); + await manager.decrementMemoriesExtracted(10); + + // Recalibrate (e.g., after cleanup) + await manager.recalibrate({ actualL0Count: 80, actualL1Count: 40, actualTotalProcessed: 400 }); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(80); + expect(cp.total_memories_extracted).toBe(40); + expect(cp.total_processed).toBe(400); + }); + }); + + // ═══════════════════════════════════════════════════════════════════════════════ + // CLEANUP SCENARIO TESTS + // ═══════════════════════════════════════════════════════════════════════════════ + + describe("cleanup scenarios", () => { + + /** + * Scenario: Manual JSONL pruning + * User deletes old conversations/YYYY-MM-DD.jsonl and records/YYYY-MM-DD.jsonl shards. + * Checkpoint counters should be recalibrated to match actual retained data. + */ + describe("manual JSONL pruning", () => { + it("recalibrates counters after user deletes old shards", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate checkpoint state BEFORE pruning + await manager.recalibrate({ + actualL0Count: 100, + actualL1Count: 50, + actualTotalProcessed: 500, + }); + + // User manually deletes 60 L0 and 30 L1 records from disk + // (This would be actual file deletion in production) + const result = await manager.recalibrate({ + actualL0Count: 40, // 100 - 60 deleted + actualL1Count: 20, // 50 - 30 deleted + actualTotalProcessed: 200, // proportional reduction + }); + + expect(result.l0_conversations_count).toBe(40); + expect(result.total_memories_extracted).toBe(20); + expect(result.total_processed).toBe(200); + }); + + it("prevents L1 trigger issues after pruning", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate: checkpoint thinks 50 L1 records exist + await manager.recalibrate({ + actualL0Count: 20, + actualL1Count: 50, + actualTotalProcessed: 200, + }); + + // User prunes: only 25 L1 records remain + await manager.recalibrate({ + actualL0Count: 20, + actualL1Count: 25, + actualTotalProcessed: 150, + }); + + // memories_since_last_persona should be recalculated + // so persona trigger fires at correct threshold + const cp = await manager.read(); + expect(cp.total_memories_extracted).toBe(25); + }); + }); + + /** + * Scenario: Test data cleanup + * Developer removes test pipeline states and test data. + * Use decrement methods for targeted correction. + */ + describe("test data cleanup", () => { + it("decrements counters when removing test records", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate: checkpoint has test data + await manager.recalibrate({ + actualL0Count: 15, + actualL1Count: 10, + actualTotalProcessed: 100, + }); + + // Developer removes 5 test L0 conversations and 3 test L1 records + await manager.decrementL0ConversationCount(5); + await manager.decrementMemoriesExtracted(3); + await manager.decrementTotalProcessed(30); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(10); // 15 - 5 + expect(cp.total_memories_extracted).toBe(7); // 10 - 3 + expect(cp.total_processed).toBe(70); // 100 - 30 + }); + + it("handles partial test data removal correctly", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ + actualL0Count: 12, + actualL1Count: 8, + actualTotalProcessed: 80, + }); + + // Remove only L0 test data (5 conversations) + await manager.decrementL0ConversationCount(5); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(7); + expect(cp.total_memories_extracted).toBe(8); // Unchanged + expect(cp.total_processed).toBe(80); // Unchanged + }); + }); + + /** + * Scenario: Session reset + * User resets a specific session's data. + * Checkpoint should handle per-session state correctly. + */ + describe("session reset", () => { + it("preserves other session counters when one is reset", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate two sessions + await manager.recalibrate({ + actualL0Count: 20, // 10 per session + actualL1Count: 10, // 5 per session + actualTotalProcessed: 200, + }); + + // Reset session-b: decrement by 10 L0, 5 L1 + await manager.decrementL0ConversationCount(10); + await manager.decrementMemoriesExtracted(5); + await manager.decrementTotalProcessed(100); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(10); // session-a remains + expect(cp.total_memories_extracted).toBe(5); // session-a remains + expect(cp.total_processed).toBe(100); // session-a remains + }); + + it("handles complete session reset", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Single session with all data + await manager.recalibrate({ + actualL0Count: 25, + actualL1Count: 15, + actualTotalProcessed: 250, + }); + + // Complete reset + await manager.decrementL0ConversationCount(25); + await manager.decrementMemoriesExtracted(15); + await manager.decrementTotalProcessed(250); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(0); + expect(cp.total_memories_extracted).toBe(0); + expect(cp.total_processed).toBe(0); + }); + }); + + /** + * Scenario: Incremental processing prevention + * After cleanup, new data should NOT be skipped. + * Recalibration ensures cursors are correct. + */ + describe("incremental processing", () => { + it("recalibration prevents L1 from skipping new records", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate: checkpoint drifted ahead + await manager.recalibrate({ + actualL0Count: 100, + actualL1Count: 50, + actualTotalProcessed: 500, + }); + + // Cleanup removes 50 L0 records + await manager.recalibrate({ + actualL0Count: 50, + actualL1Count: 25, + actualTotalProcessed: 250, + }); + + // Now new data comes in - it should be processed + // (In real code, captureAtomically would increment from 50, not skip from 100) + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(50); + // New capture would add +1, resulting in 51 (not 101) + }); + + it("memories_since_last_persona recalculates correctly after cleanup", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Set up: 100 memories, 40 of them since the last persona run. + const cp0 = await manager.read(); + cp0.total_memories_extracted = 100; + cp0.last_persona_at = 60; + cp0.memories_since_last_persona = 40; + await manager.write(cp0); + + // Cleanup: 60 memories remain + const result = await manager.recalibrate({ + actualL0Count: 30, + actualL1Count: 60, + actualTotalProcessed: 300, + }); + + // 40 memories were removed, so the trigger counter is safely reduced to 0. + expect(result.memories_since_last_persona).toBe(0); + }); + + it("handles last persona processed cursor exceeding current L1 count", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Set up: a persona cursor from total_processed, which can exceed L1 memory count. + const cp0 = await manager.read(); + cp0.total_memories_extracted = 50; + cp0.last_persona_at = 80; + cp0.memories_since_last_persona = 0; + await manager.write(cp0); + + // After cleanup: only 30 memories remain + const result = await manager.recalibrate({ + actualL0Count: 15, + actualL1Count: 30, + actualTotalProcessed: 150, + }); + + expect(result.memories_since_last_persona).toBe(0); + }); + }); + + /** + * Scenario: Automatic cleaner (memory-cleaner integration) + * After memory-cleaner.runOnce(), checkpoint should be recalibrated. + */ + describe("automatic cleaner integration", () => { + it("supports recalibration after cleaner run", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate: cleaner ran and deleted expired data + // L0: removed 30 expired, kept 50 (above MIN_RETAIN_L0) + // L1: removed 10 expired, kept 20 (at MIN_RETAIN_L1) + await manager.recalibrate({ + actualL0Count: 50, + actualL1Count: 20, + actualTotalProcessed: 250, + }); + + // Verify minimum retention respected + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBeGreaterThanOrEqual(0); + expect(cp.total_memories_extracted).toBeGreaterThanOrEqual(0); + }); + + it("handles cleaner running on fresh data (no deletion)", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Data is within retention, cleaner does nothing + await manager.recalibrate({ + actualL0Count: 10, + actualL1Count: 5, + actualTotalProcessed: 50, + }); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(10); + expect(cp.total_memories_extracted).toBe(5); + }); + }); + }); + + // ═══════════════════════════════════════════════════════════════════════════════ + // DESIGN PATTERN VALIDATION TESTS + // ═══════════════════════════════════════════════════════════════════════════════ + + describe("design pattern validation", () => { + + /** + * Validates: Hybrid cursor-first approach + * Counters are derived status values, not the source of truth. + */ + it("counters can be recalculated independently of storage", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Simulate: storage has 100 L0, 50 L1 + await manager.recalibrate({ + actualL0Count: 100, + actualL1Count: 50, + actualTotalProcessed: 500, + }); + + // Verify counters reflect storage + let cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(100); + expect(cp.total_memories_extracted).toBe(50); + + // Simulate: storage now has 80 L0, 40 L1 + await manager.recalibrate({ + actualL0Count: 80, + actualL1Count: 40, + actualTotalProcessed: 400, + }); + + cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(80); + expect(cp.total_memories_extracted).toBe(40); + }); + + /** + * Validates: Backward compatibility + * Existing checkpoint schema is preserved. + */ + it("checkpoint schema remains compatible", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Use recalibrate (new method) + await manager.recalibrate({ + actualL0Count: 50, + actualL1Count: 25, + actualTotalProcessed: 250, + }); + + // Read with standard read() method + const cp = await manager.read(); + + // Verify all expected fields exist + expect(cp).toHaveProperty("l0_conversations_count"); + expect(cp).toHaveProperty("total_memories_extracted"); + expect(cp).toHaveProperty("total_processed"); + expect(cp).toHaveProperty("memories_since_last_persona"); + expect(cp).toHaveProperty("last_persona_at"); + expect(cp).toHaveProperty("runner_states"); + expect(cp).toHaveProperty("pipeline_states"); + }); + + /** + * Validates: Atomic operations + * File lock ensures concurrent access safety. + */ + it("handles concurrent recalibrations safely", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ + actualL0Count: 100, + actualL1Count: 50, + actualTotalProcessed: 500, + }); + + // Simulate concurrent decrements + await Promise.all([ + manager.decrementL0ConversationCount(10), + manager.decrementL0ConversationCount(20), + manager.decrementL0ConversationCount(5), + ]); + + // All three decrements should be applied + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(65); // 100 - 10 - 20 - 5 + }); + }); + + // ═══════════════════════════════════════════════════════════════════════════════ + // RECALCULATE() AUTO-SCAN TESTS + // ═══════════════════════════════════════════════════════════════════════════════ + + describe("recalculate() auto-scan mode", () => { + async function createJsonlFile(dir: string, filename: string, records: Record[]): Promise { + const filePath = path.join(dir, filename); + const lines = records.map((r) => JSON.stringify(r)).join("\n"); + await fs.promises.mkdir(dir, { recursive: true }); + await fs.promises.writeFile(filePath, lines, "utf-8"); + } + + it("scans L0 JSONL shards and counts records", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Create L0 JSONL shards + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-28.jsonl", [ + { session_id: "session-a", recorded_at: "2026-06-28T10:00:00.000Z", content: "test1" }, + { session_id: "session-a", recorded_at: "2026-06-28T11:00:00.000Z", content: "test2" }, + { session_id: "session-b", recorded_at: "2026-06-28T12:00:00.000Z", content: "test3" }, + ]); + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-29.jsonl", [ + { session_id: "session-a", recorded_at: "2026-06-29T10:00:00.000Z", content: "test4" }, + ]); + + // Create L1 JSONL shards + await createJsonlFile(path.join(dataDir, "records"), "2026-06-28.jsonl", [ + { session_id: "session-a", updated_at: "2026-06-28T12:00:00.000Z", content: "memory1" }, + ]); + + const result = await manager.recalculate(); + + expect(result.l0_conversations_count).toBe(4); + expect(result.total_memories_extracted).toBe(1); + expect(result.total_processed).toBe(4); + }); + + it("adjusts memories_since_last_persona safely during auto-scan", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + const cp0 = await manager.read(); + cp0.total_memories_extracted = 5; + cp0.memories_since_last_persona = 3; + cp0.last_persona_at = 100; + await manager.write(cp0); + + await createJsonlFile(path.join(dataDir, "records"), "2026-06-28.jsonl", [ + { sessionKey: "session-a", updatedAt: "2026-06-28T10:00:00.000Z", content: "memory1" }, + { sessionKey: "session-a", updatedAt: "2026-06-28T11:00:00.000Z", content: "memory2" }, + { sessionKey: "session-a", updatedAt: "2026-06-28T12:00:00.000Z", content: "memory3" }, + { sessionKey: "session-a", updatedAt: "2026-06-28T13:00:00.000Z", content: "memory4" }, + ]); + + const result = await manager.recalculate(); + + expect(result.total_memories_extracted).toBe(4); + expect(result.memories_since_last_persona).toBe(2); + }); + + it("finds newest L0 recorded_at for cursor correction", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-28.jsonl", [ + { session_id: "session-a", recorded_at: "2026-06-28T10:00:00.000Z" }, + { session_id: "session-a", recorded_at: "2026-06-28T14:00:00.000Z" }, // newest + ]); + + const result = await manager.recalculate(); + expect(result.l0_conversations_count).toBe(2); + }); + + it("corrects stale runner cursors after recalculate", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Create actual data + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-28.jsonl", [ + { session_id: "session-a", recorded_at: "2026-06-28T10:00:00.000Z" }, + ]); + + // Set stale checkpoint with cursor ahead of actual data + const cp0 = await manager.read(); + cp0.runner_states = { + "session-a": { + last_captured_timestamp: new Date("2026-06-29T00:00:00.000Z").getTime(), // stale + last_l1_cursor: new Date("2026-06-29T00:00:00.000Z").getTime(), // stale + last_scene_name: "old-scene", + }, + }; + await manager.write(cp0); + + const result = await manager.recalculate(); + + expect(result.runner_cursors_corrected).toBe(1); + const cp = await manager.read(); + expect(cp.runner_states["session-a"].last_captured_timestamp) + .toBeLessThanOrEqual(new Date("2026-06-28T23:59:59.999Z").getTime()); + }); + + it("corrects stale pipeline cursors after recalculate", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Create actual L1 data + await createJsonlFile(path.join(dataDir, "records"), "2026-06-28.jsonl", [ + { session_id: "session-a", updated_at: "2026-06-28T12:00:00.000Z" }, + ]); + + // Set stale checkpoint with pipeline cursor ahead of actual data + const cp0 = await manager.read(); + cp0.pipeline_states = { + "session-a": { + conversation_count: 1, + last_extraction_time: "2026-06-28T12:00:00.000Z", + last_extraction_updated_time: "2026-06-29T00:00:00.000Z", // stale + last_active_time: Date.now(), + l2_pending_l1_count: 0, + warmup_threshold: 0, + l2_last_extraction_time: "", + }, + }; + await manager.write(cp0); + + const result = await manager.recalculate(); + + expect(result.pipeline_cursors_corrected).toBe(1); + const cp = await manager.read(); + expect(cp.pipeline_states["session-a"].last_extraction_updated_time) + .toBe("2026-06-28T12:00:00.000Z"); + }); + + it("resets cursors for sessions with no retained data", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Create data only for session-a + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-28.jsonl", [ + { session_id: "session-a", recorded_at: "2026-06-28T10:00:00.000Z" }, + ]); + + // Set checkpoint with state for both sessions + const cp0 = await manager.read(); + cp0.runner_states = { + "session-a": { + last_captured_timestamp: new Date("2026-06-28T10:00:00.000Z").getTime(), + last_l1_cursor: new Date("2026-06-28T10:00:00.000Z").getTime(), + last_scene_name: "scene-a", + }, + "session-b": { // deleted session - should be reset + last_captured_timestamp: new Date("2026-06-28T12:00:00.000Z").getTime(), + last_l1_cursor: new Date("2026-06-28T12:00:00.000Z").getTime(), + last_scene_name: "scene-b", + }, + }; + await manager.write(cp0); + + const result = await manager.recalculate(); + + // session-b should be reset + expect(result.runner_cursors_corrected).toBe(1); + const cp = await manager.read(); + expect(cp.runner_states["session-b"].last_captured_timestamp).toBe(0); + expect(cp.runner_states["session-b"].last_l1_cursor).toBe(0); + // session-a should remain unchanged + expect(cp.runner_states["session-a"].last_captured_timestamp) + .toBe(new Date("2026-06-28T10:00:00.000Z").getTime()); + }); + + it("uses store counts when provided (override JSONL)", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // Create JSONL with 2 L0 records + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-28.jsonl", [ + { session_id: "session-a", recorded_at: "2026-06-28T10:00:00.000Z" }, + { session_id: "session-a", recorded_at: "2026-06-28T11:00:00.000Z" }, + ]); + + // But VectorStore says 5 (simulating more accurate count) + const result = await manager.recalculate({ + storeCounts: { + l0ConversationsCount: 5, + totalMemoriesExtracted: 3, + }, + }); + + expect(result.l0_conversations_count).toBe(5); // from store + expect(result.total_memories_extracted).toBe(3); // from store + }); + + it("does not persist internal correction counters into checkpoint JSON", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-28.jsonl", [ + { sessionKey: "session-a", recordedAt: "2026-06-28T10:00:00.000Z" }, + ]); + + const cp0 = await manager.read(); + cp0.runner_states = { + "session-a": { + last_captured_timestamp: new Date("2026-06-29T00:00:00.000Z").getTime(), + last_l1_cursor: new Date("2026-06-29T00:00:00.000Z").getTime(), + last_scene_name: "old-scene", + }, + }; + await manager.write(cp0); + + await manager.recalculate(); + + const checkpointPath = path.join(dataDir, ".metadata", "recall_checkpoint.json"); + const raw = JSON.parse(await fs.promises.readFile(checkpointPath, "utf-8")) as Record; + expect(raw).not.toHaveProperty("__runnerCursorsCorrected"); + expect(raw).not.toHaveProperty("__pipelineCursorsCorrected"); + }); + + it("clamps stale runner cursors using each session's own retained L0 watermark", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-28.jsonl", [ + { sessionKey: "session-a", recordedAt: "2026-06-28T10:00:00.000Z" }, + ]); + await createJsonlFile(path.join(dataDir, "conversations"), "2026-06-30.jsonl", [ + { sessionKey: "session-b", recordedAt: "2026-06-30T10:00:00.000Z" }, + ]); + + const cp0 = await manager.read(); + cp0.runner_states = { + "session-a": { + last_captured_timestamp: new Date("2026-06-29T00:00:00.000Z").getTime(), + last_l1_cursor: new Date("2026-06-29T00:00:00.000Z").getTime(), + last_scene_name: "scene-a", + }, + "session-b": { + last_captured_timestamp: new Date("2026-06-30T10:00:00.000Z").getTime(), + last_l1_cursor: new Date("2026-06-30T10:00:00.000Z").getTime(), + last_scene_name: "scene-b", + }, + }; + await manager.write(cp0); + + const result = await manager.recalculate(); + + expect(result.runner_cursors_corrected).toBe(1); + const cp = await manager.read(); + expect(cp.runner_states["session-a"].last_l1_cursor) + .toBe(new Date("2026-06-28T10:00:00.000Z").getTime()); + expect(cp.runner_states["session-a"].last_captured_timestamp) + .toBe(new Date("2026-06-28T10:00:00.000Z").getTime()); + expect(cp.runner_states["session-b"].last_l1_cursor) + .toBe(new Date("2026-06-30T10:00:00.000Z").getTime()); + }); + + it("clamps stale pipeline cursors using each session's own retained L1 watermark", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await createJsonlFile(path.join(dataDir, "records"), "2026-06-28.jsonl", [ + { session_key: "session-a", updated_time: "2026-06-28T10:00:00.000Z" }, + ]); + await createJsonlFile(path.join(dataDir, "records"), "2026-06-30.jsonl", [ + { session_key: "session-b", updated_time: "2026-06-30T10:00:00.000Z" }, + ]); + + const cp0 = await manager.read(); + cp0.pipeline_states = { + "session-a": { + conversation_count: 1, + last_extraction_time: "2026-06-28T10:00:00.000Z", + last_extraction_updated_time: "2026-06-29T00:00:00.000Z", + last_active_time: Date.now(), + l2_pending_l1_count: 0, + warmup_threshold: 0, + l2_last_extraction_time: "", + }, + "session-b": { + conversation_count: 1, + last_extraction_time: "2026-06-30T10:00:00.000Z", + last_extraction_updated_time: "2026-06-30T10:00:00.000Z", + last_active_time: Date.now(), + l2_pending_l1_count: 0, + warmup_threshold: 0, + l2_last_extraction_time: "", + }, + }; + await manager.write(cp0); + + const result = await manager.recalculate(); + + expect(result.pipeline_cursors_corrected).toBe(1); + const cp = await manager.read(); + expect(cp.pipeline_states["session-a"].last_extraction_updated_time) + .toBe("2026-06-28T10:00:00.000Z"); + expect(cp.pipeline_states["session-b"].last_extraction_updated_time) + .toBe("2026-06-30T10:00:00.000Z"); + }); + + it("recalculates checkpoint after LocalMemoryCleaner removes expired JSONL shards", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + await manager.recalibrate({ actualL0Count: 10, actualL1Count: 8, actualTotalProcessed: 10 }); + await createJsonlFile(path.join(dataDir, "conversations"), "2026-07-01.jsonl", [ + { sessionKey: "session-a", recordedAt: "2026-07-01T10:00:00.000Z" }, + ]); + await createJsonlFile(path.join(dataDir, "conversations"), "2026-07-02.jsonl", [ + { sessionKey: "session-a", recordedAt: "2026-07-02T10:00:00.000Z" }, + { sessionKey: "session-b", recordedAt: "2026-07-02T11:00:00.000Z" }, + ]); + await createJsonlFile(path.join(dataDir, "records"), "2026-07-01.jsonl", [ + { session_key: "session-a", updated_time: "2026-07-01T10:00:00.000Z" }, + ]); + await createJsonlFile(path.join(dataDir, "records"), "2026-07-02.jsonl", [ + { session_key: "session-b", updated_time: "2026-07-02T11:00:00.000Z" }, + ]); + + let callbackResult: unknown; + const cleaner = new LocalMemoryCleaner({ + baseDir: dataDir, + retentionDays: 2, + cleanTime: "03:00", + checkpointManager: manager, + onRecalculate: (result) => { + callbackResult = result; + }, + }); + + await cleaner.runOnce(new Date("2026-07-03T12:00:00.000Z").getTime()); + + const cp = await manager.read(); + expect(cp.l0_conversations_count).toBe(2); + expect(cp.total_memories_extracted).toBe(1); + expect(cp.total_processed).toBe(2); + expect(callbackResult).toEqual({ + l0Count: 2, + l1Count: 1, + runnerCursorsCorrected: 0, + pipelineCursorsCorrected: 0, + }); + }); + it("handles missing directories gracefully", async () => { + const dataDir = createTempDir(); + const manager = new CheckpointManager(dataDir); + + // No conversations/ or records/ directories - should not throw + const result = await manager.recalculate(); + + expect(result.l0_conversations_count).toBe(0); + expect(result.total_memories_extracted).toBe(0); + }); + }); +}); diff --git a/src/utils/checkpoint.ts b/src/utils/checkpoint.ts index 301fc0df..4c3aa150 100644 --- a/src/utils/checkpoint.ts +++ b/src/utils/checkpoint.ts @@ -1,6 +1,156 @@ /** * Checkpoint management for tracking memory processing progress. * + * ═══════════════════════════════════════════════════════════════════════════════ + * CHECKPOINT DATA FLOW DIAGRAM + * ═══════════════════════════════════════════════════════════════════════════════ + * + * ┌─────────────────────────────────────────────────────────────────────────────┐ + * │ CHECKPOINT DATA FLOW │ + * └─────────────────────────────────────────────────────────────────────────────┘ + * + * L0 CAPTURE PATH: L1 EXTRACTION PATH: + * ───────────────── ─────────────────── + * + * auto-capture.ts pipeline-manager.ts + * │ │ + * ▼ ▼ + * ┌─────────────────┐ ┌─────────────────┐ + * │ captureAtomically│ │ notifyConversation + * │ (in checkpoint) │ │ + L1 idle timer│ + * └────────┬────────┘ └────────┬────────┘ + * │ │ + * ▼ ▼ + * ┌───────────────────────────────────────────────────────────────────────┐ + * │ MUTATE (file lock) │ + * │ ┌─────────────────────────────────────────────────────────────────┐ │ + * │ │ runner_states[session].last_captured_timestamp = maxTimestamp │ │ + * │ │ l0_conversations_count += 1 │ │ + * │ │ total_processed += messageCount │ │ + * │ └─────────────────────────────────────────────────────────────────┘ │ + * └───────────────────────────────────────────────────────────────────────┘ + * │ + * ▼ + * ┌───────────────────────────────────────────────────────────────────────┐ + * │ PERSIST TO DISK │ + * │ recall_checkpoint.json │ + * └───────────────────────────────────────────────────────────────────────┘ + * + * L1 COMPLETION PATH: L2 PIPELINE PATH: + * ─────────────────── ───────────────── + * + * pipeline-manager.ts pipeline-manager.ts + * │ │ + * ▼ ▼ + * ┌─────────────────────┐ ┌─────────────────────┐ + * │ markL1Extraction │ │ runL2(sessionKey) │ + * │ Complete() │ │ L2 idle timer fires │ + * └─────────┬───────────┘ └──────────┬──────────┘ + * │ │ + * ▼ ▼ + * ┌───────────────────────────────────────────────────────────────────────┐ + * │ MUTATE (file lock) │ + * │ ┌─────────────────────────────────────────────────────────────────┐ │ + * │ │ runner_states[session].last_l1_cursor = cursorRecordedAtMs │ │ + * │ │ runner_states[session].last_scene_name = lastSceneName │ │ + * │ │ total_memories_extracted += memoriesExtracted │ │ + * │ │ memories_since_last_persona += memoriesExtracted │ │ + * │ └─────────────────────────────────────────────────────────────────┘ │ + * └───────────────────────────────────────────────────────────────────────┘ + * + * CLEANUP PATHS (COUNTER DRIFT CAUSE): + * ────────────────────────────────── + * + * ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ + * │memory-cleaner│ │ manual JSONL │ │ session │ + * │ runOnce() │ │ pruning │ │ reset │ + * └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ + * │ │ │ + * ▼ ▼ ▼ + * ┌───────────────────────────────────────────────────────────────┐ + * │ PROBLEM: Checkpoint counters never decrease │ + * │ │ + * │ l0_conversations_count ← only increments (captureAtomically)│ + * │ total_memories_extracted← only increments (markL1Complete) │ + * │ total_processed ← only increments (captureAtomically)│ + * │ │ + * │ AFTER CLEANUP: checkpoint shows 50, actual data has 42 │ + * └───────────────────────────────────────────────────────────────┘ + * + * SOLUTION (THIS PR): + * ────────────────── + * + * ┌──────────────────┐ ┌──────────────────┐ + * │ recalibrate() │ │ decrement*() │ + * │ Batch reset │ │ Incremental fix │ + * └────────┬─────────┘ └────────┬─────────┘ + * │ │ + * ▼ ▼ + * ┌─────────────────────────────────────────────┐ + * │ Runner calls after cleanup: │ + * │ - startup recalibration │ + * │ - post-cleanup recalibration │ + * │ - manual recalibration │ + * └─────────────────────────────────────────────┘ + * + * ═══════════════════════════════════════════════════════════════════════════════ + * COUNTER DRIFT IMPACT ANALYSIS + * ═══════════════════════════════════════════════════════════════════════════════ + * + * | Counter | Drift Impact | + * |--------------------------|------------------------------------------------| + * | l0_conversations_count | Status display: shows X when actual is Y | + * | total_memories_extracted | Status display: shows X when actual is Y | + * | total_processed | Status display: shows X when actual is Y | + * | memories_since_last_persona | L3 persona trigger fires early or never | + * | last_l1_cursor | L1 skip: new records NOT processed after reset | + * | last_extraction_updated_time | L2 skip: new records NOT extracted | + * + * ═══════════════════════════════════════════════════════════════════════════════ + * DESIGN PATTERN ANALYSIS + * ═══════════════════════════════════════════════════════════════════════════════ + * + * CURRENT DESIGN (Preserved for Compatibility): + * ───────────────────────────────────────────── + * - Counters are "append-only truth" during normal operation + * - Increments are atomic within file lock + * - Recalibration is a deliberate correction, not automatic + * + * ALTERNATIVE DESIGN (Considered but deferred): + * ───────────────────────────────────────────── + * 1. Event-sourced: Store all increments/decrements as events, compute counts + * - Pros: True source of truth, easy to audit + * - Cons: Complex migration, larger storage + * + * 2. Storage-first: Always query storage for counts, never cache in checkpoint + * - Pros: Single source of truth, no drift possible + * - Cons: Performance overhead, complex queries + * + * 3. Hybrid: Timestamp cursors as truth, counters as derived hints + * - Pros: Cursor-based correctness, counters for quick status + * - Cons: Requires careful design of cursor semantics + * + * THIS PR CHOICE: Option C (Hybrid) with backward compatibility + * - Timestamp cursors (last_l1_cursor, last_extraction_updated_time) become + * the correctness-critical state + * - Counters remain for status reporting but are recalculated from storage + * - Existing checkpoint schema preserved (no migration needed) + * + * ═══════════════════════════════════════════════════════════════════════════════ + * ACCEPTANCE CRITERIA COVERAGE + * ═══════════════════════════════════════════════════════════════════════════════ + * + * | Criteria | Coverage | + * |-------------------------------|------------------------------| + * | 基础: 数据流图 | ✓ Complete (see above) | + * | 进阶: recalibrate() | ✓ Implemented | + * | 进阶: decrement*() | ✓ Implemented | + * | 深入: 手动清理场景测试 | ✓ Unit tests in checkpoint.test.ts| + * | 深入: 自动清理场景测试 | ✓ To be wired to memory-cleaner| + * | 拓展: 设计模式分析 | ✓ Documented above | + * + * ═══════════════════════════════════════════════════════════════════════════════ + * * ## Split-state design * * Per-session state is split into two independent namespaces to prevent @@ -84,6 +234,7 @@ export interface Checkpoint { last_captured_timestamp: number; /** Total messages processed across all time */ total_processed: number; + /** total_processed value at the last persona generation; not an L1 memory count. */ last_persona_at: number; last_persona_time: string; request_persona_update: boolean; @@ -144,6 +295,68 @@ export interface CheckpointLogger { warn?(msg: string): void; } +/** + * Result of a recalibration operation. + */ +export interface RecalibrationResult { + l0_conversations_count: number; + total_memories_extracted: number; + total_processed: number; + memories_since_last_persona: number; +} + +/** + * Result of a recalculate() operation (auto-scan mode). + * Includes cursor correction statistics. + */ +export interface RecalculateResult { + l0_conversations_count: number; + total_memories_extracted: number; + total_processed: number; + memories_since_last_persona: number; + /** Number of runner session cursors that were corrected */ + runner_cursors_corrected: number; + /** Number of pipeline session cursors that were corrected */ + pipeline_cursors_corrected: number; +} + +interface JsonlScanResult { + l0Count: number; + l1Count: number; + totalProcessed: number; + newestL0RecordedAt: number; + newestL1UpdatedAt: number; + newestL0RecordedAtBySession: Map; + newestL1UpdatedAtBySession: Map; + l0Sessions: Set; + l1Sessions: Set; +} + +function getStringField(record: Record, ...keys: string[]): string | undefined { + for (const key of keys) { + const value = record[key]; + if (typeof value === "string" && value.length > 0) return value; + } + return undefined; +} + +function canTrustL0SessionAbsence(actualL0Count: number, scanResult: JsonlScanResult): boolean { + return actualL0Count === scanResult.l0Count; +} + +function canTrustL1SessionAbsence(actualL1Count: number, scanResult: JsonlScanResult): boolean { + return actualL1Count === scanResult.l1Count; +} + +function adjustMemoriesSinceLastPersona( + currentSinceLastPersona: number, + previousL1Count: number, + nextL1Count: number, +): number { + const removedMemories = Math.max(0, previousL1Count - nextL1Count); + return Math.max(0, currentSinceLastPersona - removedMemories); +} + const noopLogger: CheckpointLogger = { info() {} }; // ============================ @@ -485,4 +698,340 @@ export class CheckpointManager { }); } + // ============================ + // Recalibration (fix counter drift after cleanup) + // ============================ + + /** + * Recalibrate counters by recounting from actual data. + * + * After cleanup operations (deleting test pipeline states, running + * memory-cleaner, or manual JSONL pruning), checkpoint counters drift + * from actual data because all increment methods lack decrement counterparts. + * + * This method allows callers to reset counters to their actual values + * by providing the true counts from the data layer. + * + * @param params.actualL0Count - Actual L0 conversation count from storage + * @param params.actualL1Count - Actual L1 memories count from storage + * @param params.actualTotalProcessed - Actual total processed from storage + * @returns The recalibrated values written to checkpoint + */ + async recalibrate(params: { + actualL0Count: number; + actualL1Count: number; + actualTotalProcessed: number; + }): Promise { + const { actualL0Count, actualL1Count, actualTotalProcessed } = params; + + const cp = await this.mutate((checkpoint) => { + const nextL1Count = Math.max(0, actualL1Count); + const memoriesSinceLastPersona = adjustMemoriesSinceLastPersona( + checkpoint.memories_since_last_persona, + checkpoint.total_memories_extracted, + nextL1Count, + ); + + checkpoint.l0_conversations_count = Math.max(0, actualL0Count); + checkpoint.total_memories_extracted = nextL1Count; + checkpoint.total_processed = Math.max(0, actualTotalProcessed); + checkpoint.memories_since_last_persona = memoriesSinceLastPersona; + }); + + this.logger.info( + `[checkpoint] recalibrate: l0=${cp.l0_conversations_count}, ` + + `l1=${cp.total_memories_extracted}, processed=${cp.total_processed}, ` + + `memories_since_last_persona=${cp.memories_since_last_persona}`, + ); + + return { + l0_conversations_count: cp.l0_conversations_count, + total_memories_extracted: cp.total_memories_extracted, + total_processed: cp.total_processed, + memories_since_last_persona: cp.memories_since_last_persona, + }; + } + + /** + * Recalculate checkpoint from actual JSONL data (auto-scan mode). + * + * This is the preferred method for fixing counter drift because it: + * 1. Automatically scans conversations/YYYY-MM-DD.jsonl → counts L0 + * 2. Automatically scans records/YYYY-MM-DD.jsonl → counts L1 + * 3. Clamps stale last_l1_cursor to newest retained L0 recordedAt + * 4. Clamps stale last_extraction_updated_time to newest retained L1 updatedAt + * 5. Resets session cursors to 0/"" when session data no longer exists + * + * @param options.storeCounts - Optional VectorStore counts (preferred if available) + * When provided, these override JSONL scan counts. + * @returns The recalculated values and cursor correction stats + */ + async recalculate(options?: { + storeCounts?: { + l0ConversationsCount?: number; + totalMemoriesExtracted?: number; + }; + }): Promise { + const dataDir = path.dirname(path.dirname(this.filePath)); + + // Phase 1: Scan JSONL shards for counts and cursors + const scanResult = await this.scanJsonlShards(dataDir); + + // Phase 2: Determine final counts (store > JSONL fallback) + const l0Count = options?.storeCounts?.l0ConversationsCount ?? scanResult.l0Count; + const l1Count = options?.storeCounts?.totalMemoriesExtracted ?? scanResult.l1Count; + + // Phase 3: Mutate checkpoint with recalculated values + let runnerCursorsCorrected = 0; + let pipelineCursorsCorrected = 0; + const cp = await this.mutate((checkpoint) => { + const nextL1Count = Math.max(0, l1Count); + const memoriesSinceLastPersona = adjustMemoriesSinceLastPersona( + checkpoint.memories_since_last_persona, + checkpoint.total_memories_extracted, + nextL1Count, + ); + + checkpoint.l0_conversations_count = Math.max(0, l0Count); + checkpoint.total_memories_extracted = nextL1Count; + checkpoint.total_processed = Math.max(0, scanResult.totalProcessed); + checkpoint.memories_since_last_persona = memoriesSinceLastPersona; + + // Phase 4: Clamp per-session runner cursors + if (checkpoint.runner_states) { + for (const [sessionKey, state] of Object.entries(checkpoint.runner_states)) { + let changed = false; + const newestL0ForSession = scanResult.newestL0RecordedAtBySession.get(sessionKey); + + if (newestL0ForSession !== undefined && newestL0ForSession > 0) { + if (state.last_l1_cursor > newestL0ForSession) { + state.last_l1_cursor = newestL0ForSession; + changed = true; + } + if (state.last_captured_timestamp > newestL0ForSession) { + state.last_captured_timestamp = newestL0ForSession; + changed = true; + } + } else if (canTrustL0SessionAbsence(l0Count, scanResult)) { + if (state.last_l1_cursor !== 0) { + state.last_l1_cursor = 0; + changed = true; + } + if (state.last_captured_timestamp !== 0) { + state.last_captured_timestamp = 0; + changed = true; + } + } + + if (changed) runnerCursorsCorrected++; + } + } + + // Phase 5: Clamp per-session pipeline cursors + if (checkpoint.pipeline_states) { + for (const [sessionKey, state] of Object.entries(checkpoint.pipeline_states)) { + let changed = false; + const newestL1ForSession = scanResult.newestL1UpdatedAtBySession.get(sessionKey); + + if (newestL1ForSession !== undefined && newestL1ForSession > 0) { + const currentCursor = state.last_extraction_updated_time + ? new Date(state.last_extraction_updated_time).getTime() + : 0; + if (Number.isFinite(currentCursor) && currentCursor > newestL1ForSession) { + state.last_extraction_updated_time = new Date(newestL1ForSession).toISOString(); + changed = true; + } + } else if (canTrustL1SessionAbsence(l1Count, scanResult)) { + if (state.last_extraction_updated_time !== "") { + state.last_extraction_updated_time = ""; + changed = true; + } + } + + if (changed) pipelineCursorsCorrected++; + } + } + }); + this.logger.info( + `[checkpoint] recalculate: l0=${cp.l0_conversations_count}, l1=${cp.total_memories_extracted}, ` + + `processed=${cp.total_processed}, runner_cursors=${runnerCursorsCorrected}, ` + + `pipeline_cursors=${pipelineCursorsCorrected}`, + ); + + return { + l0_conversations_count: cp.l0_conversations_count, + total_memories_extracted: cp.total_memories_extracted, + total_processed: cp.total_processed, + memories_since_last_persona: cp.memories_since_last_persona, + runner_cursors_corrected: runnerCursorsCorrected, + pipeline_cursors_corrected: pipelineCursorsCorrected, + }; + } + + /** + * Scan JSONL shards to count records and find cursor positions. + */ + private async scanJsonlShards(dataDir: string): Promise { + const L0_DIR = "conversations"; + const L1_DIR = "records"; + + let l0Count = 0; + let l1Count = 0; + let totalProcessed = 0; + let newestL0RecordedAt = 0; + let newestL1UpdatedAt = 0; + const newestL0RecordedAtBySession = new Map(); + const newestL1UpdatedAtBySession = new Map(); + const l0Sessions = new Set(); + const l1Sessions = new Set(); + + // Scan L0 conversations + await this.scanJsonlDir( + path.join(dataDir, L0_DIR), + { + onRecord: (record) => { + l0Count++; + totalProcessed++; + const sessionKey = getStringField(record, "sessionKey", "session_key", "session_id"); + if (sessionKey) l0Sessions.add(sessionKey); + const recordedAt = getStringField(record, "recordedAt", "recorded_at"); + if (recordedAt) { + const ts = new Date(recordedAt).getTime(); + if (!Number.isFinite(ts)) return; + if (ts > newestL0RecordedAt) newestL0RecordedAt = ts; + if (sessionKey) { + const current = newestL0RecordedAtBySession.get(sessionKey) ?? 0; + if (ts > current) newestL0RecordedAtBySession.set(sessionKey, ts); + } + } + }, + }, + ); + + // Scan L1 records + await this.scanJsonlDir( + path.join(dataDir, L1_DIR), + { + onRecord: (record) => { + l1Count++; + const sessionKey = getStringField(record, "sessionKey", "session_key", "session_id"); + if (sessionKey) l1Sessions.add(sessionKey); + const updatedAt = getStringField(record, "updatedAt", "updated_at", "updated_time"); + if (updatedAt) { + const ts = new Date(updatedAt).getTime(); + if (!Number.isFinite(ts)) return; + if (ts > newestL1UpdatedAt) newestL1UpdatedAt = ts; + if (sessionKey) { + const current = newestL1UpdatedAtBySession.get(sessionKey) ?? 0; + if (ts > current) newestL1UpdatedAtBySession.set(sessionKey, ts); + } + } + }, + }, + ); + + return { + l0Count, + l1Count, + totalProcessed, + newestL0RecordedAt, + newestL1UpdatedAt, + newestL0RecordedAtBySession, + newestL1UpdatedAtBySession, + l0Sessions, + l1Sessions, + }; + } + /** + * Scan a directory of JSONL shard files and process each record. + */ + private async scanJsonlDir( + dirPath: string, + handlers: { + onRecord: (record: Record) => void; + }, + ): Promise { + let entries; + try { + entries = await fs.readdir(dirPath, { withFileTypes: true }); + } catch { + // Directory doesn't exist - nothing to scan + return; + } + + for (const entry of entries) { + if (!entry.isFile()) continue; + if (!entry.name.endsWith(".jsonl") && !entry.name.endsWith(".json")) continue; + + // Only scan date-sharded files (YYYY-MM-DD.*) + if (!/^\d{4}-\d{2}-\d{2}\.(jsonl?|json)$/.test(entry.name)) continue; + + const filePath = path.join(dirPath, entry.name); + try { + const content = await fs.readFile(filePath, "utf-8"); + const lines = content.split("\n").filter((line) => line.trim()); + + for (const line of lines) { + try { + const record = JSON.parse(line); + handlers.onRecord(record); + } catch { + // Skip malformed JSON lines + } + } + } catch { + // Skip files that can't be read + } + } + } + + // ============================ + // Decrement methods (error correction) + // ============================ + + /** + * Decrement L0 conversation count for error correction. + * + * Use case: when a conversation file is deleted after capture + * (e.g., manual cleanup, test data removal). + * + * @param count - Number to decrement by (default: 1) + */ + async decrementL0ConversationCount(count = 1): Promise { + await this.mutate((cp) => { + cp.l0_conversations_count = Math.max(0, cp.l0_conversations_count - count); + }); + this.logger.info(`[checkpoint] decrementL0ConversationCount: l0=${count}`); + } + + /** + * Decrement memories extracted and related counters for error correction. + * + * Use case: when L1 records are deleted after extraction + * (e.g., test data cleanup, manual deletion). + * + * @param count - Number to decrement by + */ + async decrementMemoriesExtracted(count: number): Promise { + await this.mutate((cp) => { + cp.total_memories_extracted = Math.max(0, cp.total_memories_extracted - count); + cp.memories_since_last_persona = Math.max(0, cp.memories_since_last_persona - count); + }); + this.logger.info(`[checkpoint] decrementMemoriesExtracted: extracted=${count}`); + } + + /** + * Decrement total_processed counter for error correction. + * + * Use case: when captured messages are removed after processing + * (e.g., L0 cleanup, manual trimming). + * + * @param count - Number to decrement by + */ + async decrementTotalProcessed(count: number): Promise { + await this.mutate((cp) => { + cp.total_processed = Math.max(0, cp.total_processed - count); + }); + this.logger.info(`[checkpoint] decrementTotalProcessed: processed=${count}`); + } } diff --git a/src/utils/memory-cleaner.ts b/src/utils/memory-cleaner.ts index 509f4fb7..968e0d7f 100644 --- a/src/utils/memory-cleaner.ts +++ b/src/utils/memory-cleaner.ts @@ -5,6 +5,7 @@ import type { IMemoryStore } from "../core/store/types.js"; import { ManagedTimer } from "./managed-timer.js"; import type { Logger } from "../core/types.js"; import { formatLocalDateTime, startOfLocalDay } from "./time.js"; +import { CheckpointManager } from "./checkpoint.js"; export interface MemoryCleanerOptions { baseDir: string; @@ -12,6 +13,15 @@ export interface MemoryCleanerOptions { cleanTime: string; logger?: Logger; vectorStore?: IMemoryStore; + /** Optional checkpoint manager for recalculating checkpoint after cleanup */ + checkpointManager?: CheckpointManager; + /** Optional callback invoked after cleanup with recalculation results */ + onRecalculate?: (result: { + l0Count: number; + l1Count: number; + runnerCursorsCorrected: number; + pipelineCursorsCorrected: number; + }) => void | Promise; } interface CleanupStats { @@ -33,16 +43,89 @@ export class LocalMemoryCleaner { private readonly timer: ManagedTimer; private destroyed = false; private vectorStore?: IMemoryStore; + private checkpointManager?: CheckpointManager; + private onRecalculate?: (result: { + l0Count: number; + l1Count: number; + runnerCursorsCorrected: number; + pipelineCursorsCorrected: number; + }) => void | Promise; constructor(private readonly opts: MemoryCleanerOptions) { this.timer = new ManagedTimer("memory-tdai-cleaner", () => this.destroyed); this.vectorStore = opts.vectorStore; + this.checkpointManager = opts.checkpointManager; + this.onRecalculate = opts.onRecalculate; } setVectorStore(vectorStore: IMemoryStore | undefined): void { this.vectorStore = vectorStore; } + setCheckpointManager(checkpointManager: CheckpointManager): void { + this.checkpointManager = checkpointManager; + } + + setOnRecalculate(callback: (result: { + l0Count: number; + l1Count: number; + runnerCursorsCorrected: number; + pipelineCursorsCorrected: number; + }) => void | Promise): void { + this.onRecalculate = callback; + } + + /** + * Recalculate checkpoint to fix counter drift after cleanup. + * Called automatically after cleanup runOnce() if checkpointManager is set. + */ + private async recalculateCheckpoint(): Promise { + if (!this.checkpointManager) { + this.opts.logger?.debug?.(`${TAG} [Recalculate] Skipped: no checkpoint manager set`); + return; + } + + const startMs = Date.now(); + try { + const storeCounts = this.vectorStore + ? { + l0ConversationsCount: await this.vectorStore.countL0().catch(() => undefined), + totalMemoriesExtracted: await this.vectorStore.countL1().catch(() => undefined), + } + : undefined; + + const result = await this.checkpointManager.recalculate({ storeCounts }); + + const durationMs = Date.now() - startMs; + this.opts.logger?.info( + `${TAG} [Recalculate] Completed: l0=${result.l0_conversations_count}, ` + + `l1=${result.total_memories_extracted}, ` + + `runner_cursors=${result.runner_cursors_corrected}, ` + + `pipeline_cursors=${result.pipeline_cursors_corrected} (${durationMs}ms)`, + ); + + // Invoke optional callback + if (this.onRecalculate) { + try { + await this.onRecalculate({ + l0Count: result.l0_conversations_count, + l1Count: result.total_memories_extracted, + runnerCursorsCorrected: result.runner_cursors_corrected, + pipelineCursorsCorrected: result.pipeline_cursors_corrected, + }); + } catch (callbackErr) { + this.opts.logger?.warn?.( + `${TAG} [Recalculate] Callback error: ${callbackErr instanceof Error ? callbackErr.message : String(callbackErr)}`, + ); + } + } + } catch (err) { + this.opts.logger?.warn?.( + `${TAG} [Recalculate] Failed: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + start(): void { if (this.destroyed) return; @@ -184,6 +267,9 @@ export class LocalMemoryCleaner { `${TAG} Cleanup done: scannedFiles=${total.scannedFiles}, changedFiles=${total.changedFiles}, skippedNonShardFiles=${total.skippedNonShardFiles}, deleteFailedFiles=${total.deleteFailedFiles}`, ); + // ── Post-cleanup: recalculate checkpoint ── + // This fixes counter drift after JSONL/DB cleanup + await this.recalculateCheckpoint(); } private scheduleNext(): void {