diff --git a/src/core/persona/persona-generator.ts b/src/core/persona/persona-generator.ts index cba05901..3b97f3a2 100644 --- a/src/core/persona/persona-generator.ts +++ b/src/core/persona/persona-generator.ts @@ -8,6 +8,7 @@ import path from "node:path"; import { formatForLLM } from "../../utils/time.js"; import { CleanContextRunner } from "../../utils/clean-context-runner.js"; import { CheckpointManager } from "../../utils/checkpoint.js"; +import type { Checkpoint } from "../../utils/checkpoint.js"; import { readSceneIndex } from "../scene/scene-index.js"; import { generateSceneNavigation, stripSceneNavigation } from "../scene/scene-navigation.js"; import { buildPersonaPrompt } from "../prompts/persona-generation.js"; @@ -56,13 +57,18 @@ export class PersonaGenerator { /** * Execute local persona generation without advancing checkpoint. + * `checkpointSnapshot` keeps prompt input and completion coverage on the + * same generation-start view when the caller needs race-safe accounting. */ - async generateLocalPersona(triggerReason?: string): Promise { + async generateLocalPersona( + triggerReason?: string, + checkpointSnapshot?: Checkpoint, + ): Promise { const startMs = Date.now(); this.logger?.debug?.(`${TAG} Starting generation: reason="${triggerReason ?? "none"}"`); const cpManager = new CheckpointManager(this.dataDir); - const cp = await cpManager.read(); + const cp = checkpointSnapshot ?? await cpManager.read(); this.logger?.debug?.(`${TAG} Checkpoint: total_processed=${cp.total_processed}, last_persona_at=${cp.last_persona_at}`); const personaPath = path.join(this.dataDir, "persona.md"); @@ -207,12 +213,16 @@ export class PersonaGenerator { * Backward-compatible wrapper: local generation + checkpoint advance. */ async generate(triggerReason?: string): Promise { - const updated = await this.generateLocalPersona(triggerReason); + const sceneUpdatedThrough = new Date().toISOString(); + const cpManager = new CheckpointManager(this.dataDir); + const baseline = await cpManager.read(); + const updated = await this.generateLocalPersona(triggerReason, baseline); if (!updated) return false; - const cpManager = new CheckpointManager(this.dataDir); - const cp = await cpManager.read(); - await cpManager.markPersonaGenerated(cp.total_processed); + await cpManager.markPersonaGenerated(baseline.total_processed, { + memoriesSinceLastPersona: baseline.memories_since_last_persona, + sceneUpdatedThrough, + }); return true; } } diff --git a/src/utils/checkpoint.test.ts b/src/utils/checkpoint.test.ts new file mode 100644 index 00000000..9347c68e --- /dev/null +++ b/src/utils/checkpoint.test.ts @@ -0,0 +1,84 @@ +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 } from "./checkpoint.js"; + +const tempDirs: string[] = []; + +async function makeTempDir(): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "checkpoint-persona-race-")); + tempDirs.push(dir); + return dir; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); +}); + +describe("CheckpointManager persona generation coverage", () => { + it("preserves memories extracted while persona generation is running", async () => { + const checkpoint = new CheckpointManager(await makeTempDir()); + const sceneUpdatedThrough = "2026-07-12T01:00:00.000Z"; + + await checkpoint.captureAtomically("before", undefined, async () => ({ + maxTimestamp: 100, + messageCount: 3, + })); + await checkpoint.markL1ExtractionComplete("before", 3); + const baseline = await checkpoint.read(); + + await checkpoint.captureAtomically("during", undefined, async () => ({ + maxTimestamp: 200, + messageCount: 2, + })); + await checkpoint.markL1ExtractionComplete("during", 2); + + await checkpoint.markPersonaGenerated(baseline.total_processed, { + memoriesSinceLastPersona: baseline.memories_since_last_persona, + sceneUpdatedThrough, + }); + + const current = await checkpoint.read(); + expect(current.total_processed).toBe(5); + expect(current.last_persona_at).toBe(3); + expect(current.last_persona_time).toBe(sceneUpdatedThrough); + expect(current.total_memories_extracted).toBe(5); + expect(current.memories_since_last_persona).toBe(2); + }); + + it("consumes the full baseline when no new memories arrive", async () => { + const checkpoint = new CheckpointManager(await makeTempDir()); + await checkpoint.markL1ExtractionComplete("session", 4); + const baseline = await checkpoint.read(); + + await checkpoint.markPersonaGenerated(baseline.total_processed, { + memoriesSinceLastPersona: baseline.memories_since_last_persona, + sceneUpdatedThrough: "2026-07-12T01:00:00.000Z", + }); + + expect((await checkpoint.read()).memories_since_last_persona).toBe(0); + }); + + it("keeps the legacy clear-all behavior when coverage is omitted", async () => { + const checkpoint = new CheckpointManager(await makeTempDir()); + await checkpoint.markL1ExtractionComplete("session", 4); + + await checkpoint.markPersonaGenerated(0); + + expect((await checkpoint.read()).memories_since_last_persona).toBe(0); + }); + + it("clamps the remaining counter when cleanup shrinks it below the baseline", async () => { + const checkpoint = new CheckpointManager(await makeTempDir()); + await checkpoint.markL1ExtractionComplete("session", 2); + + await checkpoint.markPersonaGenerated(0, { + memoriesSinceLastPersona: 5, + sceneUpdatedThrough: "2026-07-12T01:00:00.000Z", + }); + + expect((await checkpoint.read()).memories_since_last_persona).toBe(0); + }); +}); diff --git a/src/utils/checkpoint.ts b/src/utils/checkpoint.ts index 301fc0df..2d9b9acc 100644 --- a/src/utils/checkpoint.ts +++ b/src/utils/checkpoint.ts @@ -84,7 +84,9 @@ export interface Checkpoint { last_captured_timestamp: number; /** Total messages processed across all time */ total_processed: number; + /** Total-message watermark covered by the last successful persona generation. */ last_persona_at: number; + /** ISO scene-update watermark covered by the last successful persona generation. */ last_persona_time: string; request_persona_update: boolean; persona_update_reason: string; @@ -144,6 +146,13 @@ export interface CheckpointLogger { warn?(msg: string): void; } +/** Snapshot of the memory counter covered by one persona generation run. */ +export interface PersonaGenerationCoverage { + memoriesSinceLastPersona: number; + /** ISO watermark captured before reading scene data for this generation. */ + sceneUpdatedThrough: string; +} + const noopLogger: CheckpointLogger = { info() {} }; // ============================ @@ -301,11 +310,34 @@ export class CheckpointManager { // Persona methods (L3) // ============================ - async markPersonaGenerated(totalProcessed: number): Promise { + /** + * Mark persona generation complete. + * + * When `coverage` is supplied, only memories present at generation start are + * consumed. L1 memories extracted while the persona model is running remain + * pending for the next trigger. Omitting it preserves the legacy clear-all + * behavior for external callers. + */ + async markPersonaGenerated( + totalProcessed: number, + coverage?: PersonaGenerationCoverage, + ): Promise { await this.mutate((cp) => { cp.last_persona_at = totalProcessed; - cp.last_persona_time = new Date().toISOString(); - cp.memories_since_last_persona = 0; + cp.last_persona_time = coverage && Number.isFinite(Date.parse(coverage.sceneUpdatedThrough)) + ? coverage.sceneUpdatedThrough + : new Date().toISOString(); + if (coverage) { + const coveredMemories = Number.isFinite(coverage.memoriesSinceLastPersona) + ? Math.max(0, Math.trunc(coverage.memoriesSinceLastPersona)) + : 0; + cp.memories_since_last_persona = Math.max( + 0, + cp.memories_since_last_persona - coveredMemories, + ); + } else { + cp.memories_since_last_persona = 0; + } cp.request_persona_update = false; cp.persona_update_reason = ""; }); diff --git a/src/utils/pipeline-factory.ts b/src/utils/pipeline-factory.ts index ce87801c..9b79f130 100644 --- a/src/utils/pipeline-factory.ts +++ b/src/utils/pipeline-factory.ts @@ -628,7 +628,10 @@ export function createL3Runner(opts: { instanceId, llmRunner, }); - const genResult = await generator.generateLocalPersona(reason); + const sceneUpdatedThrough = new Date().toISOString(); + const checkpoint = new CheckpointManager(pluginDataDir, logger); + const baseline = await checkpoint.read(); + const genResult = await generator.generateLocalPersona(reason, baseline); if (!genResult) { logger.info(`${TAG} [L3] Persona generation skipped (no changes)`); return; @@ -638,9 +641,10 @@ export function createL3Runner(opts: { await syncLocalProfilesToStore(pluginDataDir, vectorStore, profileBaseline, logger); } - const checkpoint = new CheckpointManager(pluginDataDir, logger); - const cp = await checkpoint.read(); - await checkpoint.markPersonaGenerated(cp.total_processed); + await checkpoint.markPersonaGenerated(baseline.total_processed, { + memoriesSinceLastPersona: baseline.memories_since_last_persona, + sceneUpdatedThrough, + }); logger.info(`${TAG} [L3] Persona generation succeeded`); }; }