fix: make the auto-capture watermark survive a restart, and bound it when unknown

dist/index.js

@@ -39,7 +39,8 @@ import { createReflectionEventId } from "./src/reflection-event-store.js";
 import { buildReflectionMappedMetadata } from "./src/reflection-mapped-metadata.js";
 import { createMemoryCLI } from "./cli.js";
 import { isNoise } from "./src/noise-filter.js";
-import { normalizeAutoCaptureText } from "./src/auto-capture-cleanup.js";
+import { normalizeAutoCaptureText, capUnknownWatermarkWindow } from "./src/auto-capture-cleanup.js";
+import { loadAutoCaptureWatermarks, saveAutoCaptureWatermarks } from "./src/auto-capture-watermark-store.js";
 // Import smart extraction & lifecycle components
 import { SmartExtractor, createExtractionRateLimiter } from "./src/smart-extractor.js";
 import { compressTexts, estimateConversationValue } from "./src/session-compressor.js";
@@ -830,6 +831,21 @@ function pruneMapIfOver(map, maxEntries) {
             map.delete(key);
     }
 }
+/**
+ * Prune a Set to stay within the given maximum number of entries.
+ * Deletes the oldest (earliest-inserted) values when over the limit.
+ */
+function pruneSetIfOver(set, maxEntries) {
+    if (set.size <= maxEntries)
+        return;
+    const excess = set.size - maxEntries;
+    const iter = set.values();
+    for (let i = 0; i < excess; i++) {
+        const value = iter.next().value;
+        if (value !== undefined)
+            set.delete(value);
+    }
+}
 function isExplicitRememberCommand(text) {
     return AUTO_CAPTURE_EXPLICIT_REMEMBER_RE.test(text.trim());
 }
@@ -1882,9 +1898,11 @@ function _initPluginState(api) {
     const reflectionByAgentCacheGeneration = { count: 0 };
     const recallHistory = new Map();
     const turnCounter = new Map();
-    const autoCaptureSeenTextCount = new Map();
+    const autoCaptureSeenTextCount = loadAutoCaptureWatermarks(resolvedDbPath);
     const autoCapturePendingIngressTexts = new Map();
     const autoCaptureRecentTexts = new Map();
+    const autoCapturePayloadShapeLoggedSessions = new Set();
+    const autoCaptureSessionIds = new Map();
     return {
         config,
         resolvedDbPath,
@@ -1912,6 +1930,8 @@ function _initPluginState(api) {
         autoCaptureSeenTextCount,
         autoCapturePendingIngressTexts,
         autoCaptureRecentTexts,
+        autoCapturePayloadShapeLoggedSessions,
+        autoCaptureSessionIds,
     };
 }
 export function isAgentOrSessionExcluded(agentId, sessionKey, patterns) {
@@ -2020,7 +2040,16 @@ const memoryLanceDBProPlugin = {
             _registeredApisMap.delete(api); // dual-track rollback: Map un-claim
             throw err;
         }
-        const { config, resolvedDbPath, vectorDim, store, embedder, retriever, canonicalCorpusIndexer, dreamingEngine, dreamingScheduler, scopeManager, migrator, smartExtractor, mdMirror, decayEngine, tierManager, extractionRateLimiter, reflectionErrorStateBySession, reflectionDerivedBySession, reflectionDerivedSuppressionBySession, reflectionByAgentCache, reflectionByAgentCacheGeneration, recallHistory, turnCounter, autoCaptureSeenTextCount, autoCapturePendingIngressTexts, autoCaptureRecentTexts, } = singleton;
+        const { config, resolvedDbPath, vectorDim, store, embedder, retriever, canonicalCorpusIndexer, dreamingEngine, dreamingScheduler, scopeManager, migrator, smartExtractor, mdMirror, decayEngine, tierManager, extractionRateLimiter, reflectionErrorStateBySession, reflectionDerivedBySession, reflectionDerivedSuppressionBySession, reflectionByAgentCache, reflectionByAgentCacheGeneration, recallHistory, turnCounter, autoCaptureSeenTextCount, autoCapturePendingIngressTexts, autoCaptureRecentTexts, autoCapturePayloadShapeLoggedSessions, autoCaptureSessionIds, } = singleton;
+        // issue #417 restart-survivability: every mutation of autoCaptureSeenTextCount
+        // must also go through here so the on-disk watermark never drifts from the
+        // in-memory Map -- a process restart rehydrates from exactly what was last
+        // written here (see loadAutoCaptureWatermarks at construction, above).
+        const persistAutoCaptureWatermark = async (key, value) => {
+            autoCaptureSeenTextCount.set(key, value);
+            pruneMapIfOver(autoCaptureSeenTextCount, AUTO_CAPTURE_MAP_MAX_ENTRIES);
+            await saveAutoCaptureWatermarks(resolvedDbPath, autoCaptureSeenTextCount, (message) => api.logger.debug(message));
+        };
         warnForDisabledChannelPlugin(api.config, api.logger);
         async function sleep(ms, signal) {
             if (signal?.aborted) {
@@ -2873,6 +2902,7 @@ const memoryLanceDBProPlugin = {
                 // as if they were conversation. Same guard convention as the sibling
                 // reflection injection hooks.
                 const hookSessionKey = ctx?.sessionKey || event.sessionKey;
+                const hookSessionId = typeof ctx?.sessionId === "string" && ctx.sessionId.length > 0 ? ctx.sessionId : undefined;
                 if (isInternalReflectionSessionKey(hookSessionKey) || isMemorySubsessionKey(hookSessionKey)) {
                     api.logger.debug(`memory-lancedb-pro: auto-capture skip \u2014 internal memory session '${hookSessionKey}'`);
                     return;
@@ -2953,18 +2983,56 @@ const memoryLanceDBProPlugin = {
                         if (conversationKey) {
                             autoCapturePendingIngressTexts.delete(conversationKey);
                         }
+                        const minMessages = config.extractMinMessages ?? 4;
+                        // Session renewal detection: a rotated sessionId under a stable
+                        // sessionKey (e.g. /reset) means the persisted watermark counts a
+                        // conversation that no longer exists. A stale-HIGH cursor silently
+                        // swallows the fresh session's first messages as already-seen; a
+                        // stale-LOW cursor fires a spurious first-message extraction. Reset
+                        // to zero so the fresh session counts from its own start. The map
+                        // is deliberately in-memory: after a process restart it records
+                        // without resetting, preserving the watermark's restart survival.
+                        if (hookSessionId) {
+                            const recordedSessionId = autoCaptureSessionIds.get(sessionKey);
+                            if (recordedSessionId && recordedSessionId !== hookSessionId) {
+                                await persistAutoCaptureWatermark(sessionKey, 0);
+                                api.logger.info(`memory-lancedb-pro: auto-capture watermark reset for ${sessionKey} (session renewed: ${recordedSessionId.slice(0, 8)} -> ${hookSessionId.slice(0, 8)})`);
+                            }
+                            autoCaptureSessionIds.set(sessionKey, hookSessionId);
+                            pruneMapIfOver(autoCaptureSessionIds, AUTO_CAPTURE_MAP_MAX_ENTRIES);
+                        }
                         const previousSeenCount = autoCaptureSeenTextCount.get(sessionKey) ?? 0;
                         let newTexts = eligibleTexts;
+                        let watermarkAdvanceOverride = null;
                         if (pendingIngressTexts.length > 0) {
                             newTexts = pendingIngressTexts;
                         }
                         else if (previousSeenCount > 0 && eligibleTexts.length > previousSeenCount) {
                             newTexts = eligibleTexts.slice(previousSeenCount);
                         }
+                        else if (previousSeenCount === 0 && eligibleTexts.length > minMessages) {
+                            // issue #417 item 5: a genuinely unknown watermark (first-ever
+                            // run, or persisted state lost) meeting a history-carrying
+                            // payload must not ingest the entire transcript in one call.
+                            // Cap to the most recent batch and forfeit the rest by marking
+                            // it seen below, rather than queueing it for a later turn.
+                            newTexts = capUnknownWatermarkWindow(eligibleTexts, minMessages, config.extractMaxChars ?? 8000);
+                            watermarkAdvanceOverride = eligibleTexts.length;
+                        }
                         // issue #417 Fix #4: cumulative counting — increment by newly observed texts.
-                        const cumulativeCount = previousSeenCount + newTexts.length;
-                        autoCaptureSeenTextCount.set(sessionKey, cumulativeCount);
-                        pruneMapIfOver(autoCaptureSeenTextCount, AUTO_CAPTURE_MAP_MAX_ENTRIES);
+                        const cumulativeCount = watermarkAdvanceOverride ?? (previousSeenCount + newTexts.length);
+                        await persistAutoCaptureWatermark(sessionKey, cumulativeCount);
+                        // Once per session per process (not persisted, not per turn): a
+                        // restart re-emits this, which is exactly when you want to
+                        // re-confirm the payload shape in the new process. This single
+                        // INFO line answers the delta-only-vs-history-carrying question
+                        // and the gate's fire/skip decision without reconstructing them
+                        // from DEBUG-only evidence across unrelated log lines.
+                        if (!autoCapturePayloadShapeLoggedSessions.has(sessionKey)) {
+                            autoCapturePayloadShapeLoggedSessions.add(sessionKey);
+                            pruneSetIfOver(autoCapturePayloadShapeLoggedSessions, AUTO_CAPTURE_MAP_MAX_ENTRIES);
+                            api.logger.info(`memory-lancedb-pro: auto-capture payload shape for agent ${agentId} (sessionKey=${sessionKey}): messages=${event.messages.length}, eligible=${eligibleTexts.length}, previousSeen=${previousSeenCount}, cumulative=${cumulativeCount}, fired=${cumulativeCount >= minMessages ? "yes" : "no"}`);
+                        }
                         const priorRecentTexts = autoCaptureRecentTexts.get(sessionKey) || [];
                         let texts = newTexts;
                         if (texts.length === 1 &&
@@ -2977,7 +3045,6 @@ const memoryLanceDBProPlugin = {
                             autoCaptureRecentTexts.set(sessionKey, nextRecentTexts);
                             pruneMapIfOver(autoCaptureRecentTexts, AUTO_CAPTURE_MAP_MAX_ENTRIES);
                         }
-                        const minMessages = config.extractMinMessages ?? 4;
                         if (skippedAutoCaptureTexts > 0) {
                             api.logger.debug(`memory-lancedb-pro: auto-capture skipped ${skippedAutoCaptureTexts} injected/system text block(s) for agent ${agentId}`);
                         }
@@ -3055,7 +3122,7 @@ const memoryLanceDBProPlugin = {
                                     // turn re-read and re-extract the entire history. Record the
                                     // consumed history length there instead, so the next turn
                                     // only sees the delta.
-                                    autoCaptureSeenTextCount.set(sessionKey, pendingIngressTexts.length > 0 ? 0 : eligibleTexts.length);
+                                    await persistAutoCaptureWatermark(sessionKey, pendingIngressTexts.length > 0 ? 0 : eligibleTexts.length);
                                     return; // Smart extraction handled everything
                                 }
                                 if ((stats.boundarySkipped ?? 0) === 0) {
@@ -3067,6 +3134,19 @@ const memoryLanceDBProPlugin = {
                             }
                             else {
                                 api.logger.debug(`memory-lancedb-pro: auto-capture skipped smart extraction for agent ${agentId} (cumulative=${cumulativeCount} < minMessages=${minMessages}, cleanTexts=${cleanTexts.length})`);
+                                // For history-carrying sessions, roll the cursor back so the
+                                // next turn's slice re-includes these texts in the extraction
+                                // input -- otherwise a run of below-threshold turns would each
+                                // tentatively advance the cursor and the eventual firing turn
+                                // would only see its own last slice, silently forfeiting every
+                                // earlier deferred turn's content. Ingress-fed sessions keep
+                                // their accumulator advance: their per-turn text is not
+                                // recoverable on the next turn by design (the ingress queue
+                                // was already drained), and rolling back would keep the count
+                                // below the threshold forever.
+                                if (pendingIngressTexts.length === 0) {
+                                    await persistAutoCaptureWatermark(sessionKey, previousSeenCount);
+                                }
                             }
                         }
                         api.logger.debug(`memory-lancedb-pro: auto-capture running regex fallback for agent ${agentId}`);
@@ -4066,10 +4146,15 @@ const memoryLanceDBProPlugin = {
                 const now = new Date(params.timestampMs ?? Date.now());
                 const dateStr = now.toISOString().split("T")[0];
                 const timeStr = now.toISOString().split("T")[1].split(".")[0];
+                // Session key/id stay out of `text`: it is the FTS index surface, and
+                // the `simple` tokenizer splits a key like
+                // `agent:main:cron:<uuid>:run:<uuid>` on its punctuation — so every session
+                // summary ends up indexed under `agent`, `main`, `cron`, `run`. A query
+                // mentioning any of those then BM25-matches every session summary in the
+                // store regardless of content. Both ids are already recorded structurally
+                // in metadata below, so provenance is unaffected.
                 const memoryText = [
                     `Session: ${dateStr} ${timeStr} UTC`,
-                    `Session Key: ${params.sessionKey}`,
-                    `Session ID: ${params.sessionId}`,
                     `Source: ${params.source}`,
                     "",
                     "Conversation Summary:",

dist/src/auto-capture-cleanup.js

@@ -118,3 +118,22 @@ export function normalizeAutoCaptureText(role, text, shouldSkipMessage) {
         return null;
     return normalized;
 }
+/**
+ * Bounds the extraction input when a session's watermark is genuinely
+ * unknown (first-ever run, or persisted state lost) and its eligible-text
+ * history is larger than one batch's worth -- ingesting the entire history
+ * in one extraction call risks an oversized, stale-content-heavy prompt.
+ * Caps to the most recent `batchSize` texts, then trims further from the
+ * front of that window if it still exceeds `maxChars`. Always keeps at
+ * least the single most recent text, even if it alone exceeds `maxChars`.
+ */
+export function capUnknownWatermarkWindow(eligibleTexts, batchSize, maxChars) {
+    const window = eligibleTexts.slice(-Math.max(1, batchSize));
+    let start = 0;
+    let totalChars = window.reduce((sum, text) => sum + text.length, 0);
+    while (totalChars > maxChars && start < window.length - 1) {
+        totalChars -= window[start].length;
+        start++;
+    }
+    return window.slice(start);
+}

dist/src/auto-capture-watermark-store.js

@@ -0,0 +1,60 @@
+import { readFileSync } from "node:fs";
+import { dirname, join } from "node:path";
+const AUTO_CAPTURE_WATERMARK_FILE_NAME = ".auto-capture-watermark.json";
+/**
+ * Sidecar file lives next to the LanceDB dir (dirname of dbPath), matching
+ * the existing .compaction-state.json convention -- not inside dbPath itself,
+ * so a stray file never confuses LanceDB's own directory scanning.
+ */
+function watermarkFilePath(dbPath) {
+    return join(dirname(dbPath), AUTO_CAPTURE_WATERMARK_FILE_NAME);
+}
+/**
+ * Rehydrate the auto-capture seen-text watermark from disk. Synchronous
+ * (called once, at plugin init, alongside other startup-time sync fs reads)
+ * and fail-open: a missing or malformed file yields an empty map, which is
+ * exactly a genuinely fresh session's starting state.
+ */
+export function loadAutoCaptureWatermarks(dbPath) {
+    const map = new Map();
+    try {
+        const raw = readFileSync(watermarkFilePath(dbPath), "utf8");
+        const parsed = JSON.parse(raw);
+        if (parsed && typeof parsed === "object") {
+            for (const [key, value] of Object.entries(parsed)) {
+                if (typeof value === "number" && Number.isFinite(value)) {
+                    map.set(key, value);
+                }
+            }
+        }
+    }
+    catch {
+        // Missing or malformed file -- treat as a fresh watermark.
+    }
+    return map;
+}
+/**
+ * Persist a full snapshot of the auto-capture seen-text watermark to disk.
+ * Async and best-effort: never throws, so a write failure (disk full,
+ * permissions) cannot crash or block the auto-capture hook that calls it.
+ * `onWarning` receives a message on failure so the caller can log it.
+ *
+ * Callers are expected to prune the map to its bounded size (see
+ * pruneMapIfOver / AUTO_CAPTURE_MAP_MAX_ENTRIES in index.ts) before calling
+ * this, so the persisted file inherits the same unbounded-growth guard as
+ * the in-memory Map.
+ */
+export async function saveAutoCaptureWatermarks(dbPath, map, onWarning) {
+    try {
+        const { writeFile, mkdir } = await import("node:fs/promises");
+        const file = watermarkFilePath(dbPath);
+        await mkdir(dirname(file), { recursive: true });
+        const snapshot = {};
+        for (const [key, value] of map)
+            snapshot[key] = value;
+        await writeFile(file, JSON.stringify(snapshot), "utf8");
+    }
+    catch (err) {
+        onWarning?.(`memory-lancedb-pro: auto-capture watermark persistence failed: ${String(err)}`);
+    }
+}