diff --git a/dist/index.js b/dist/index.js index baebf86a1..bad216f0a 100644 --- a/dist/index.js +++ b/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::run:` 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:", diff --git a/dist/src/auto-capture-cleanup.js b/dist/src/auto-capture-cleanup.js index 098951ec4..824f30a2f 100644 --- a/dist/src/auto-capture-cleanup.js +++ b/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); +} diff --git a/dist/src/auto-capture-watermark-store.js b/dist/src/auto-capture-watermark-store.js new file mode 100644 index 000000000..4404407d4 --- /dev/null +++ b/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)}`); + } +} diff --git a/index.ts b/index.ts index 75efa8e8f..aca1df189 100644 --- a/index.ts +++ b/index.ts @@ -69,7 +69,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"; @@ -1264,6 +1265,20 @@ function pruneMapIfOver(map: Map, maxEntries: number): void { } } +/** + * 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: Set, maxEntries: number): void { + 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: string): boolean { return AUTO_CAPTURE_EXPLICIT_REMEMBER_RE.test(text.trim()); } @@ -2350,6 +2365,8 @@ interface PluginSingletonState { autoCaptureSeenTextCount: Map; autoCapturePendingIngressTexts: Map; autoCaptureRecentTexts: Map; + autoCapturePayloadShapeLoggedSessions: Set; + autoCaptureSessionIds: Map; } interface DreamingSchedulerState { @@ -2551,9 +2568,11 @@ function _initPluginState(api: OpenClawPluginApi): PluginSingletonState { 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, @@ -2582,6 +2601,8 @@ function _initPluginState(api: OpenClawPluginApi): PluginSingletonState { autoCaptureSeenTextCount, autoCapturePendingIngressTexts, autoCaptureRecentTexts, + autoCapturePayloadShapeLoggedSessions, + autoCaptureSessionIds, }; } @@ -2735,8 +2756,20 @@ const memoryLanceDBProPlugin = { 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: string, value: number): Promise => { + autoCaptureSeenTextCount.set(key, value); + pruneMapIfOver(autoCaptureSeenTextCount, AUTO_CAPTURE_MAP_MAX_ENTRIES); + await saveAutoCaptureWatermarks(resolvedDbPath, autoCaptureSeenTextCount, (message) => api.logger.debug(message)); + }; + warnForDisabledChannelPlugin( (api as OpenClawPluginApi & { config?: unknown }).config, api.logger, @@ -3768,6 +3801,8 @@ const memoryLanceDBProPlugin = { // as if they were conversation. Same guard convention as the sibling // reflection injection hooks. const hookSessionKey = ctx?.sessionKey || (event as any).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}'`, @@ -3866,17 +3901,60 @@ const memoryLanceDBProPlugin = { 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: number | null = 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; @@ -3893,7 +3971,6 @@ const memoryLanceDBProPlugin = { 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}`, @@ -4000,7 +4077,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( + await persistAutoCaptureWatermark( sessionKey, pendingIngressTexts.length > 0 ? 0 : eligibleTexts.length, ); @@ -4025,6 +4102,19 @@ const memoryLanceDBProPlugin = { 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); + } } } diff --git a/package.json b/package.json index dfeb2daa5..313603887 100644 --- a/package.json +++ b/package.json @@ -33,7 +33,7 @@ "skills/**/*.md" ], "scripts": { - "test": "node test/embedder-error-hints.test.mjs && node --test test/embedder-max-input-chars.test.mjs && node test/cjk-recursion-regression.test.mjs && node test/extraction-prompt-structural-noise.test.mjs && node test/i18n-memory-triggers.test.mjs && node test/migrate-legacy-schema.test.mjs && node --test test/config-session-strategy-migration.test.mjs && node --test test/scope-access-undefined.test.mjs && node --test test/reflection-bypass-hook.test.mjs && node --test test/smart-extractor-scope-filter.test.mjs && node --test test/store-empty-scope-filter.test.mjs && node --test test/recall-text-cleanup.test.mjs && node test/update-consistency-lancedb.test.mjs && node --test test/strip-envelope-metadata.test.mjs && node test/cli-smoke.mjs && node test/functional-e2e.mjs && node --test test/per-agent-auto-recall.test.mjs && node test/retriever-rerank-regression.mjs && node test/smart-memory-lifecycle.mjs && node test/smart-extractor-branches.mjs && node --test test/smart-extractor-noise-gating.test.mjs && node test/memory-capability-runtime.test.mjs && node --test test/startup-health-diagnostics.test.mjs && node test/corpus-indexer.test.mjs && node --test test/regex-fallback-bulk-store.test.mjs && node test/plugin-manifest-regression.mjs && node --test test/dreaming-engine.test.mjs && node --test test/session-summary-before-reset.test.mjs && node --test test/sync-plugin-version.test.mjs && node test/smart-metadata-v2.mjs && node test/vector-search-cosine.test.mjs && node test/context-support-e2e.mjs && node test/temporal-facts.test.mjs && node test/memory-update-supersede.test.mjs && node test/memory-update-metadata-refresh.test.mjs && node test/memory-upgrader-diagnostics.test.mjs && node --test test/llm-api-key-client.test.mjs && node --test test/llm-oauth-client.test.mjs && node --test test/cli-oauth-login.test.mjs && node --test test/workflow-fork-guards.test.mjs && node --test test/clawteam-scope.test.mjs && node --test test/cross-process-lock.test.mjs && node --test test/preference-slots.test.mjs && node test/is-latest-auto-supersede.test.mjs && node --test test/temporal-awareness.test.mjs && node --test test/command-reflection-guard.test.mjs && node --test test/tier1-counters.test.mjs && node --test test/startup-check-timeout.test.mjs && node --test test/memory-subsession-prompt-hooks.test.mjs && node --test test/read-consistency-interval.test.mjs && node --test test/reflection-distiller-hook-skip.test.mjs && node --test test/register-scope-dedup.test.mjs && node --test test/raw-run-distiller-hooks.test.mjs && node --test test/autocapture-watermark-reset.test.mjs && node --test test/autocapture-internal-session-guard.test.mjs && node --test test/memory-categories-storage-map.test.mjs && node --test test/delete-invalidate-reflection-caches.test.mjs", + "test": "node test/embedder-error-hints.test.mjs && node --test test/embedder-max-input-chars.test.mjs && node test/cjk-recursion-regression.test.mjs && node test/extraction-prompt-structural-noise.test.mjs && node test/i18n-memory-triggers.test.mjs && node test/migrate-legacy-schema.test.mjs && node --test test/config-session-strategy-migration.test.mjs && node --test test/scope-access-undefined.test.mjs && node --test test/reflection-bypass-hook.test.mjs && node --test test/smart-extractor-scope-filter.test.mjs && node --test test/store-empty-scope-filter.test.mjs && node --test test/recall-text-cleanup.test.mjs && node test/update-consistency-lancedb.test.mjs && node --test test/strip-envelope-metadata.test.mjs && node test/cli-smoke.mjs && node test/functional-e2e.mjs && node --test test/per-agent-auto-recall.test.mjs && node test/retriever-rerank-regression.mjs && node test/smart-memory-lifecycle.mjs && node test/smart-extractor-branches.mjs && node --test test/smart-extractor-noise-gating.test.mjs && node test/memory-capability-runtime.test.mjs && node --test test/startup-health-diagnostics.test.mjs && node test/corpus-indexer.test.mjs && node --test test/regex-fallback-bulk-store.test.mjs && node test/plugin-manifest-regression.mjs && node --test test/dreaming-engine.test.mjs && node --test test/session-summary-before-reset.test.mjs && node --test test/sync-plugin-version.test.mjs && node test/smart-metadata-v2.mjs && node test/vector-search-cosine.test.mjs && node test/context-support-e2e.mjs && node test/temporal-facts.test.mjs && node test/memory-update-supersede.test.mjs && node test/memory-update-metadata-refresh.test.mjs && node test/memory-upgrader-diagnostics.test.mjs && node --test test/llm-api-key-client.test.mjs && node --test test/llm-oauth-client.test.mjs && node --test test/cli-oauth-login.test.mjs && node --test test/workflow-fork-guards.test.mjs && node --test test/clawteam-scope.test.mjs && node --test test/cross-process-lock.test.mjs && node --test test/preference-slots.test.mjs && node test/is-latest-auto-supersede.test.mjs && node --test test/temporal-awareness.test.mjs && node --test test/command-reflection-guard.test.mjs && node --test test/tier1-counters.test.mjs && node --test test/startup-check-timeout.test.mjs && node --test test/memory-subsession-prompt-hooks.test.mjs && node --test test/read-consistency-interval.test.mjs && node --test test/reflection-distiller-hook-skip.test.mjs && node --test test/register-scope-dedup.test.mjs && node --test test/raw-run-distiller-hooks.test.mjs && node --test test/autocapture-watermark-reset.test.mjs && node --test test/autocapture-internal-session-guard.test.mjs && node --test test/memory-categories-storage-map.test.mjs && node --test test/delete-invalidate-reflection-caches.test.mjs && node --test test/auto-capture-watermark-store.test.mjs && node --test test/autocapture-watermark-restart-survival.test.mjs && node --test test/auto-capture-unknown-watermark-window.test.mjs && node --test test/autocapture-unknown-watermark-injection.test.mjs && node --test test/autocapture-payload-shape-observability.test.mjs", "test:cli-smoke": "node scripts/run-ci-tests.mjs --group cli-smoke", "test:core-regression": "node scripts/run-ci-tests.mjs --group core-regression", "test:storage-and-schema": "node scripts/run-ci-tests.mjs --group storage-and-schema", diff --git a/scripts/ci-test-manifest.mjs b/scripts/ci-test-manifest.mjs index c71de5313..2024cc3e9 100644 --- a/scripts/ci-test-manifest.mjs +++ b/scripts/ci-test-manifest.mjs @@ -1,35 +1,35 @@ -export const CI_TEST_GROUPS = [ - "cli-smoke", - "core-regression", - "storage-and-schema", - "llm-clients-and-auth", - "packaging-and-workflow", -]; - +export const CI_TEST_GROUPS = [ + "cli-smoke", + "core-regression", + "storage-and-schema", + "llm-clients-and-auth", + "packaging-and-workflow", +]; + export const CI_TEST_MANIFEST = [ { group: "llm-clients-and-auth", runner: "node", file: "test/embedder-error-hints.test.mjs" }, { group: "llm-clients-and-auth", runner: "node", file: "test/embedder-max-input-chars.test.mjs", args: ["--test"] }, { group: "llm-clients-and-auth", runner: "node", file: "test/cjk-recursion-regression.test.mjs" }, - { group: "storage-and-schema", runner: "node", file: "test/migrate-legacy-schema.test.mjs" }, - { group: "storage-and-schema", runner: "node", file: "test/config-session-strategy-migration.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/scope-access-undefined.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/reflection-bypass-hook.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/smart-extractor-scope-filter.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/store-empty-scope-filter.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/store-timestamp-normalization.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/storage-path-normalization.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/storage-maintenance.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/read-consistency-interval.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/fts-index-fold.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/store-list-stats-projection-fallback.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/recall-text-cleanup.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/category-filter-normalization.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/update-consistency-lancedb.test.mjs" }, - { group: "core-regression", runner: "node", file: "test/strip-envelope-metadata.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/auto-recall-timeout.test.mjs", args: ["--test"] }, - { group: "cli-smoke", runner: "node", file: "test/import-markdown/import-markdown.test.mjs", args: ["--test"] }, - { group: "cli-smoke", runner: "node", file: "test/cli-smoke.mjs" }, - { group: "cli-smoke", runner: "node", file: "test/functional-e2e.mjs" }, + { group: "storage-and-schema", runner: "node", file: "test/migrate-legacy-schema.test.mjs" }, + { group: "storage-and-schema", runner: "node", file: "test/config-session-strategy-migration.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/scope-access-undefined.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/reflection-bypass-hook.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/smart-extractor-scope-filter.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/store-empty-scope-filter.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/store-timestamp-normalization.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/storage-path-normalization.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/storage-maintenance.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/read-consistency-interval.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/fts-index-fold.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/store-list-stats-projection-fallback.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/recall-text-cleanup.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/category-filter-normalization.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/update-consistency-lancedb.test.mjs" }, + { group: "core-regression", runner: "node", file: "test/strip-envelope-metadata.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/auto-recall-timeout.test.mjs", args: ["--test"] }, + { group: "cli-smoke", runner: "node", file: "test/import-markdown/import-markdown.test.mjs", args: ["--test"] }, + { group: "cli-smoke", runner: "node", file: "test/cli-smoke.mjs" }, + { group: "cli-smoke", runner: "node", file: "test/functional-e2e.mjs" }, { group: "storage-and-schema", runner: "node", file: "test/per-agent-auto-recall.test.mjs", args: ["--test"] }, { group: "core-regression", runner: "node", file: "test/retriever-rerank-regression.mjs" }, { group: "core-regression", runner: "node", file: "test/retriever-neighbor-enrichment.test.mjs", args: ["--test"] }, @@ -41,80 +41,85 @@ export const CI_TEST_MANIFEST = [ { group: "core-regression", runner: "node", file: "test/dreaming-engine.test.mjs", args: ["--test"] }, { group: "core-regression", runner: "node", file: "test/memory-governance-tools.test.mjs", args: ["--test"] }, { group: "core-regression", runner: "node", file: "test/corpus-indexer.test.mjs" }, - { group: "packaging-and-workflow", runner: "node", file: "test/plugin-manifest-regression.mjs" }, - { group: "packaging-and-workflow", runner: "node", file: "test/openclaw-twitter-source-recipe.test.mjs", args: ["--test"] }, - { group: "packaging-and-workflow", runner: "node", file: "test/package-runtime.test.mjs" }, - { group: "packaging-and-workflow", runner: "node", file: "test/release-readiness.test.mjs" }, - { group: "core-regression", runner: "node", file: "test/session-summary-before-reset.test.mjs", args: ["--test"] }, + { group: "packaging-and-workflow", runner: "node", file: "test/plugin-manifest-regression.mjs" }, + { group: "packaging-and-workflow", runner: "node", file: "test/openclaw-twitter-source-recipe.test.mjs", args: ["--test"] }, + { group: "packaging-and-workflow", runner: "node", file: "test/package-runtime.test.mjs" }, + { group: "packaging-and-workflow", runner: "node", file: "test/release-readiness.test.mjs" }, + { group: "core-regression", runner: "node", file: "test/session-summary-before-reset.test.mjs", args: ["--test"] }, { group: "core-regression", runner: "node", file: "test/self-improvement-reset-note.test.mjs", args: ["--test"] }, { group: "core-regression", runner: "node", file: "test/self-improvement.test.mjs", args: ["--test"] }, { group: "packaging-and-workflow", runner: "node", file: "test/sync-plugin-version.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/smart-metadata-v2.mjs" }, - { group: "storage-and-schema", runner: "node", file: "test/vector-search-cosine.test.mjs" }, - { group: "core-regression", runner: "node", file: "test/context-support-e2e.mjs" }, + { group: "core-regression", runner: "node", file: "test/smart-metadata-v2.mjs" }, + { group: "storage-and-schema", runner: "node", file: "test/vector-search-cosine.test.mjs" }, + { group: "core-regression", runner: "node", file: "test/context-support-e2e.mjs" }, { group: "core-regression", runner: "node", file: "test/temporal-facts.test.mjs" }, { group: "core-regression", runner: "node", file: "test/memory-fact-query.test.mjs", args: ["--test"] }, { group: "core-regression", runner: "node", file: "test/memory-update-supersede.test.mjs" }, - { group: "llm-clients-and-auth", runner: "node", file: "test/memory-upgrader-diagnostics.test.mjs" }, - { group: "llm-clients-and-auth", runner: "node", file: "test/llm-api-key-client.test.mjs", args: ["--test"] }, - { group: "llm-clients-and-auth", runner: "node", file: "test/llm-oauth-client.test.mjs", args: ["--test"] }, - { group: "llm-clients-and-auth", runner: "node", file: "test/cli-oauth-login.test.mjs", args: ["--test"] }, - { group: "packaging-and-workflow", runner: "node", file: "test/workflow-fork-guards.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/clawteam-scope.test.mjs", args: ["--test"] }, + { group: "llm-clients-and-auth", runner: "node", file: "test/memory-upgrader-diagnostics.test.mjs" }, + { group: "llm-clients-and-auth", runner: "node", file: "test/llm-api-key-client.test.mjs", args: ["--test"] }, + { group: "llm-clients-and-auth", runner: "node", file: "test/llm-oauth-client.test.mjs", args: ["--test"] }, + { group: "llm-clients-and-auth", runner: "node", file: "test/cli-oauth-login.test.mjs", args: ["--test"] }, + { group: "packaging-and-workflow", runner: "node", file: "test/workflow-fork-guards.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/clawteam-scope.test.mjs", args: ["--test"] }, { group: "storage-and-schema", runner: "node", file: "test/cross-process-lock.test.mjs", args: ["--test"] }, { group: "storage-and-schema", runner: "node", file: "test/redis-lock.test.mjs", args: ["--test"] }, { group: "core-regression", runner: "node", file: "test/lock-stress-test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/lock-release-on-error.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/preference-slots.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/is-latest-auto-supersede.test.mjs" }, - { group: "core-regression", runner: "node", file: "test/temporal-awareness.test.mjs", args: ["--test"] }, - // Issue #598 regression tests - { group: "core-regression", runner: "node", file: "test/store-serialization.test.mjs" }, - { group: "core-regression", runner: "node", file: "test/mmr-tiny.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/access-tracker-retry.test.mjs" }, - { group: "core-regression", runner: "node", file: "test/embedder-cache.test.mjs" }, - // Issue #629 batch embedding fix - { group: "llm-clients-and-auth", runner: "node", file: "test/embedder-ollama-batch-routing.test.mjs" }, - // Issue #665 bulkStore tests - // Issue #690 cross-call batch accumulator tests - { group: "storage-and-schema", runner: "node", file: "test/issue-690-cross-call-batch.test.mjs", args: ["--test"] }, - // Issue #665 bulkStore tests (from upstream) - { group: "storage-and-schema", runner: "node", file: "test/bulk-store.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/bulk-store-edge-cases.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/smart-extractor-bulk-store.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/smart-extractor-bulk-store-edge-cases.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/store-importance-normalization.test.mjs", args: ["--test"] }, - // Issue #680 regression tests (from upstream) - { group: "core-regression", runner: "node", file: "test/memory-reflection-issue680-tdd.test.mjs", args: ["--test"] }, - // Issue #606 SDK migration Bug 2 regression tests - { group: "core-regression", runner: "node", file: "test/issue606_sdk-migration.test.mjs" }, - // PR #713 inference regression tests - inferProviderFromBaseURL + model fallback - { group: "core-regression", runner: "node", file: "test/infer-provider-from-baseurl.test.mjs" }, - // Issue #736 recall governance - isRecallUsed() unit tests - { group: "core-regression", runner: "node", file: "test/is-recall-used.test.mjs", args: ["--test"] }, - // Issue #492 agentId validation tests - { group: "core-regression", runner: "node", file: "test/agentid-validation.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/command-reflection-guard.test.mjs", args: ["--test"] }, - // Tier 1 memory counter fix - { group: "core-regression", runner: "node", file: "test/tier1-counters.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/memory-subsession-prompt-hooks.test.mjs", args: ["--test"] }, - // Reflection distiller sub-session must not receive auto-recall/injected blocks - { group: "core-regression", runner: "node", file: "test/reflection-distiller-hook-skip.test.mjs", args: ["--test"] }, - // register() re-registration hardening (scope cache-miss log/handler dedup) + { group: "core-regression", runner: "node", file: "test/lock-release-on-error.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/preference-slots.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/is-latest-auto-supersede.test.mjs" }, + { group: "core-regression", runner: "node", file: "test/temporal-awareness.test.mjs", args: ["--test"] }, + // Issue #598 regression tests + { group: "core-regression", runner: "node", file: "test/store-serialization.test.mjs" }, + { group: "core-regression", runner: "node", file: "test/mmr-tiny.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/access-tracker-retry.test.mjs" }, + { group: "core-regression", runner: "node", file: "test/embedder-cache.test.mjs" }, + // Issue #629 batch embedding fix + { group: "llm-clients-and-auth", runner: "node", file: "test/embedder-ollama-batch-routing.test.mjs" }, + // Issue #665 bulkStore tests + // Issue #690 cross-call batch accumulator tests + { group: "storage-and-schema", runner: "node", file: "test/issue-690-cross-call-batch.test.mjs", args: ["--test"] }, + // Issue #665 bulkStore tests (from upstream) + { group: "storage-and-schema", runner: "node", file: "test/bulk-store.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/bulk-store-edge-cases.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/smart-extractor-bulk-store.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/smart-extractor-bulk-store-edge-cases.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/store-importance-normalization.test.mjs", args: ["--test"] }, + // Issue #680 regression tests (from upstream) + { group: "core-regression", runner: "node", file: "test/memory-reflection-issue680-tdd.test.mjs", args: ["--test"] }, + // Issue #606 SDK migration Bug 2 regression tests + { group: "core-regression", runner: "node", file: "test/issue606_sdk-migration.test.mjs" }, + // PR #713 inference regression tests - inferProviderFromBaseURL + model fallback + { group: "core-regression", runner: "node", file: "test/infer-provider-from-baseurl.test.mjs" }, + // Issue #736 recall governance - isRecallUsed() unit tests + { group: "core-regression", runner: "node", file: "test/is-recall-used.test.mjs", args: ["--test"] }, + // Issue #492 agentId validation tests + { group: "core-regression", runner: "node", file: "test/agentid-validation.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/command-reflection-guard.test.mjs", args: ["--test"] }, + // Tier 1 memory counter fix + { group: "core-regression", runner: "node", file: "test/tier1-counters.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/memory-subsession-prompt-hooks.test.mjs", args: ["--test"] }, + // Reflection distiller sub-session must not receive auto-recall/injected blocks + { group: "core-regression", runner: "node", file: "test/reflection-distiller-hook-skip.test.mjs", args: ["--test"] }, + // register() re-registration hardening (scope cache-miss log/handler dedup) { group: "core-regression", runner: "node", file: "test/register-scope-dedup.test.mjs", args: ["--test"] }, - // Reflection distiller sub-run must request raw-run semantics (skip foreign before_prompt_build hooks) + // Reflection distiller sub-run must request raw-run semantics (skip foreign before_prompt_build hooks) { group: "core-regression", runner: "node", file: "test/raw-run-distiller-hooks.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/autocapture-watermark-reset.test.mjs", args: ["--test"] }, - { group: "core-regression", runner: "node", file: "test/autocapture-internal-session-guard.test.mjs", args: ["--test"] }, - { group: "storage-and-schema", runner: "node", file: "test/memory-categories-storage-map.test.mjs", args: ["--test"] }, - // Delete/delete-bulk must synchronously invalidate in-process reflection read caches - { group: "core-regression", runner: "node", file: "test/delete-invalidate-reflection-caches.test.mjs", args: ["--test"] }, -]; - -export function getEntriesForGroup(group) { - if (!CI_TEST_GROUPS.includes(group)) { - throw new Error(`Unknown CI test group: ${group}`); - } - - return CI_TEST_MANIFEST.filter((entry) => entry.group === group); -} + { group: "core-regression", runner: "node", file: "test/autocapture-watermark-reset.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/autocapture-internal-session-guard.test.mjs", args: ["--test"] }, + { group: "storage-and-schema", runner: "node", file: "test/memory-categories-storage-map.test.mjs", args: ["--test"] }, + // Delete/delete-bulk must synchronously invalidate in-process reflection read caches + { group: "core-regression", runner: "node", file: "test/delete-invalidate-reflection-caches.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/auto-capture-watermark-store.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/autocapture-watermark-restart-survival.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/auto-capture-unknown-watermark-window.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/autocapture-unknown-watermark-injection.test.mjs", args: ["--test"] }, + { group: "core-regression", runner: "node", file: "test/autocapture-payload-shape-observability.test.mjs", args: ["--test"] }, +]; + +export function getEntriesForGroup(group) { + if (!CI_TEST_GROUPS.includes(group)) { + throw new Error(`Unknown CI test group: ${group}`); + } + + return CI_TEST_MANIFEST.filter((entry) => entry.group === group); +} diff --git a/src/auto-capture-cleanup.ts b/src/auto-capture-cleanup.ts index c5c00b7b0..98a854391 100644 --- a/src/auto-capture-cleanup.ts +++ b/src/auto-capture-cleanup.ts @@ -151,3 +151,27 @@ export function normalizeAutoCaptureText( if (shouldSkipMessage?.(role, normalized)) 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: string[], + batchSize: number, + maxChars: number, +): string[] { + 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); +} diff --git a/src/auto-capture-watermark-store.ts b/src/auto-capture-watermark-store.ts new file mode 100644 index 000000000..1f3ed43d0 --- /dev/null +++ b/src/auto-capture-watermark-store.ts @@ -0,0 +1,65 @@ +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: string): string { + 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: string): Map { + const map = new Map(); + try { + const raw = readFileSync(watermarkFilePath(dbPath), "utf8"); + const parsed = JSON.parse(raw) as unknown; + if (parsed && typeof parsed === "object") { + for (const [key, value] of Object.entries(parsed as Record)) { + 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: string, + map: Map, + onWarning?: (message: string) => void, +): Promise { + try { + const { writeFile, mkdir } = await import("node:fs/promises"); + const file = watermarkFilePath(dbPath); + await mkdir(dirname(file), { recursive: true }); + const snapshot: Record = {}; + 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)}`); + } +} diff --git a/test/auto-capture-unknown-watermark-window.test.mjs b/test/auto-capture-unknown-watermark-window.test.mjs new file mode 100644 index 000000000..53da8be4c --- /dev/null +++ b/test/auto-capture-unknown-watermark-window.test.mjs @@ -0,0 +1,54 @@ +/** + * Unit coverage for capUnknownWatermarkWindow -- the bounding function that + * keeps a genuinely-unknown watermark (first-ever run, or persisted state + * lost) from ingesting an entire history-carrying transcript in one + * extraction call. See test/autocapture-unknown-watermark-injection.test.mjs + * for the full hook-level behavior this unblocks. + * + * Fixtures are entirely synthetic; no real fleet data. + */ + +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import jitiFactory from "jiti"; + +const testDir = path.dirname(fileURLToPath(import.meta.url)); +const jiti = jitiFactory(import.meta.url, { interopDefault: true }); +const { capUnknownWatermarkWindow } = jiti(path.resolve(testDir, "..", "src", "auto-capture-cleanup.ts")); + +describe("capUnknownWatermarkWindow", () => { + it("caps a large history down to the most recent batchSize texts", () => { + const texts = Array.from({ length: 40 }, (_, i) => `synthetic turn ${i + 1}`); + const result = capUnknownWatermarkWindow(texts, 2, 8000); + assert.deepEqual(result, ["synthetic turn 39", "synthetic turn 40"]); + }); + + it("leaves a short history untouched when it already fits within batchSize", () => { + const texts = ["alpha", "bravo"]; + const result = capUnknownWatermarkWindow(texts, 4, 8000); + assert.deepEqual(result, ["alpha", "bravo"]); + }); + + it("trims further from the front of the window when even batchSize texts exceed maxChars", () => { + const texts = ["a".repeat(3000), "b".repeat(3000), "c".repeat(3000)]; + // batchSize=3 would include all three (9000 chars), but maxChars=5000 + // only leaves room for the last two (6000 -- still over) then just the + // last one (3000, under budget). + const result = capUnknownWatermarkWindow(texts, 3, 5000); + assert.deepEqual(result, ["c".repeat(3000)]); + }); + + it("never returns an empty window, even if the single most recent text alone exceeds maxChars", () => { + const texts = ["short one", "x".repeat(20000)]; + const result = capUnknownWatermarkWindow(texts, 2, 100); + assert.deepEqual(result, ["x".repeat(20000)]); + }); + + it("treats a batchSize smaller than 1 as 1 (always keeps at least the latest text)", () => { + const texts = ["alpha", "bravo", "charlie"]; + const result = capUnknownWatermarkWindow(texts, 0, 8000); + assert.deepEqual(result, ["charlie"]); + }); +}); diff --git a/test/auto-capture-watermark-store.test.mjs b/test/auto-capture-watermark-store.test.mjs new file mode 100644 index 000000000..20bce71d5 --- /dev/null +++ b/test/auto-capture-watermark-store.test.mjs @@ -0,0 +1,99 @@ +/** + * Unit coverage for the auto-capture watermark's on-disk persistence + * (restart-survivability for `autoCaptureSeenTextCount` -- see + * test/autocapture-watermark-restart-survival.test.mjs for the full + * hook-level behavior this unblocks). + * + * Fixtures are entirely synthetic; no real fleet data. + */ + +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync, writeFileSync, readFileSync, mkdirSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import jitiFactory from "jiti"; + +const testDir = path.dirname(fileURLToPath(import.meta.url)); +const jiti = jitiFactory(import.meta.url, { interopDefault: true }); +const { + loadAutoCaptureWatermarks, + saveAutoCaptureWatermarks, +} = jiti(path.resolve(testDir, "..", "src", "auto-capture-watermark-store.ts")); + +describe("auto-capture watermark persistence", () => { + let workspaceDir; + let dbPath; + + beforeEach(() => { + workspaceDir = mkdtempSync(path.join(tmpdir(), "autocapture-watermark-store-")); + dbPath = path.join(workspaceDir, "db"); + mkdirSync(dbPath, { recursive: true }); + }); + + afterEach(() => { + rmSync(workspaceDir, { recursive: true, force: true }); + }); + + it("returns an empty map when no watermark file exists yet", () => { + const map = loadAutoCaptureWatermarks(dbPath); + assert.equal(map.size, 0); + }); + + it("returns an empty map when the watermark file is malformed JSON", () => { + writeFileSync(path.join(workspaceDir, ".auto-capture-watermark.json"), "{not valid json", "utf8"); + const map = loadAutoCaptureWatermarks(dbPath); + assert.equal(map.size, 0); + }); + + it("skips non-numeric entries but keeps valid ones", () => { + writeFileSync( + path.join(workspaceDir, ".auto-capture-watermark.json"), + JSON.stringify({ "agent:a:main": 3, "agent:b:main": "not-a-number", "agent:c:main": null }), + "utf8", + ); + const map = loadAutoCaptureWatermarks(dbPath); + assert.equal(map.size, 1); + assert.equal(map.get("agent:a:main"), 3); + }); + + it("round-trips a saved map through load", async () => { + const original = new Map([ + ["agent:agent-one:webchat", 1], + ["agent:agent-two:main", 4], + ]); + await saveAutoCaptureWatermarks(dbPath, original); + const reloaded = loadAutoCaptureWatermarks(dbPath); + assert.equal(reloaded.size, 2); + assert.equal(reloaded.get("agent:agent-one:webchat"), 1); + assert.equal(reloaded.get("agent:agent-two:main"), 4); + }); + + it("writes the watermark file next to the LanceDB dir, not inside it (matches the compaction-state.json convention)", async () => { + await saveAutoCaptureWatermarks(dbPath, new Map([["agent:a:main", 2]])); + const sidecarPath = path.join(workspaceDir, ".auto-capture-watermark.json"); + const raw = readFileSync(sidecarPath, "utf8"); + assert.deepEqual(JSON.parse(raw), { "agent:a:main": 2 }); + }); + + it("creates the parent directory on first save (fresh dbPath, nothing on disk yet)", async () => { + const freshDbPath = path.join(workspaceDir, "not-yet-created", "db"); + await saveAutoCaptureWatermarks(freshDbPath, new Map([["agent:a:main", 1]])); + const reloaded = loadAutoCaptureWatermarks(freshDbPath); + assert.equal(reloaded.get("agent:a:main"), 1); + }); + + it("save never throws when the parent directory is unwritable", async () => { + const unwritableParent = path.join(workspaceDir, "unwritable"); + mkdirSync(unwritableParent, { mode: 0o444 }); + const bogusDbPath = path.join(unwritableParent, "nested", "db"); + const warnings = []; + await assert.doesNotReject( + saveAutoCaptureWatermarks(bogusDbPath, new Map([["agent:a:main", 1]]), (msg) => warnings.push(msg)), + ); + if (process.getuid && process.getuid() !== 0) { + assert.ok(warnings.length > 0, "expected a warning callback on a genuine write failure"); + } + }); +}); diff --git a/test/autocapture-payload-shape-observability.test.mjs b/test/autocapture-payload-shape-observability.test.mjs new file mode 100644 index 000000000..75bb97d61 --- /dev/null +++ b/test/autocapture-payload-shape-observability.test.mjs @@ -0,0 +1,288 @@ +/** + * Regression coverage for the auto-capture payload-shape INFO line. + * + * The one debug line that reveals whether a session's agent_end payload is + * delta-only (small, roughly constant message count per turn) or + * history-carrying (grows every turn) was previously only written at DEBUG, + * which is invisible in the file logger used in production. Diagnosing a + * stuck watermark meant reconstructing this from indirect evidence across + * hours of unrelated log lines. + * + * This suite pins a compact INFO-level line, emitted once per session per + * process (not once per turn -- that would just be a different kind of log + * spam), carrying exactly the fields needed to read the payload shape and + * the watermark's gate decision at a glance: messages, eligible, + * previousSeen, cumulative, fired. + * + * Fixtures are entirely synthetic; no real fleet data. + */ + +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import jitiFactory from "jiti"; + +const testDir = path.dirname(fileURLToPath(import.meta.url)); +const pluginSdkStubPath = path.resolve(testDir, "helpers", "openclaw-plugin-sdk-stub.mjs"); +const jiti = jitiFactory(import.meta.url, { + interopDefault: true, + alias: { + "openclaw/plugin-sdk": pluginSdkStubPath, + }, +}); + +const pluginModule = jiti("../index.ts"); +const memoryLanceDBProPlugin = pluginModule.default || pluginModule; +const resetRegistration = pluginModule.resetRegistration ?? (() => {}); +const { NoisePrototypeBank } = jiti("../src/noise-prototypes.ts"); +NoisePrototypeBank.prototype.isNoise = () => false; + +const EMBEDDING_DIMENSIONS = 64; + +function hashToIndex(text, dims) { + let h = 0; + for (let i = 0; i < text.length; i++) { + h = (h * 31 + text.charCodeAt(i)) >>> 0; + } + return h % dims; +} + +function oneHot(text) { + const v = new Array(EMBEDDING_DIMENSIONS).fill(0); + v[hashToIndex(text || "", EMBEDDING_DIMENSIONS)] = 1; + return v; +} + +function createEmbeddingServer() { + return http.createServer(async (req, res) => { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + const payload = JSON.parse(Buffer.concat(chunks).toString("utf8")); + const inputs = Array.isArray(payload.input) ? payload.input : [payload.input]; + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + object: "list", + data: inputs.map((input, index) => ({ object: "embedding", index, embedding: oneHot(String(input)) })), + model: payload.model || "mock-embedding-model", + usage: { prompt_tokens: 0, total_tokens: 0 }, + })); + }); +} + +function createLlmServer() { + let calls = 0; + return http.createServer(async (req, res) => { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + calls += 1; + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion", + created: 1, + model: "mock-memory-model", + choices: [{ + index: 0, + finish_reason: "stop", + message: { + role: "assistant", + content: JSON.stringify({ + memories: [{ + category: "preferences", + abstract: `Synthetic preference marker number ${calls}`, + overview: `## Preference\n- Marker ${calls}`, + content: `User stated synthetic preference marker number ${calls}.`, + }], + }), + }, + }], + })); + }); +} + +function createPluginApiHarness({ pluginConfig, resolveRoot }) { + const eventHandlers = new Map(); + const logs = { info: [], warn: [], debug: [] }; + const api = { + pluginConfig, + resolvePath(target) { + if (typeof target !== "string") return target; + if (path.isAbsolute(target)) return target; + return path.join(resolveRoot, target); + }, + logger: { + info(message) { logs.info.push(String(message)); }, + warn(message) { logs.warn.push(String(message)); }, + debug(message) { logs.debug.push(String(message)); }, + }, + registerTool() {}, + registerCli() {}, + registerService() {}, + on(eventName, handler, meta) { + const list = eventHandlers.get(eventName) || []; + list.push({ handler, meta }); + eventHandlers.set(eventName, list); + }, + registerHook(eventName, handler, opts) { + const list = eventHandlers.get(eventName) || []; + list.push({ handler, meta: opts }); + eventHandlers.set(eventName, list); + }, + }; + return { api, eventHandlers, logs }; +} + +function getAutoCaptureHook(eventHandlers) { + const hooks = eventHandlers.get("agent_end") || []; + assert.ok(hooks.length >= 1, "expected at least one agent_end handler"); + return hooks[0].handler; +} + +async function fireAgentEnd(hook, messages, ctx) { + hook({ success: true, messages }, ctx); + const run = hook.__lastRun; + assert.ok(run && typeof run.then === "function", "expected a background capture run"); + await run; +} + +function userMessages(...texts) { + return texts.map((text) => ({ role: "user", content: text })); +} + +function payloadShapeLines(logs) { + return logs.info.filter((line) => line.includes("auto-capture payload shape")); +} + +describe("auto-capture payload-shape INFO line", () => { + let workspaceDir; + let embeddingServer; + let llmServer; + let embeddingPort; + let llmPort; + + beforeEach(async () => { + workspaceDir = mkdtempSync(path.join(tmpdir(), "autocapture-payload-shape-")); + embeddingServer = createEmbeddingServer(); + llmServer = createLlmServer(); + await new Promise((resolve) => embeddingServer.listen(0, "127.0.0.1", resolve)); + await new Promise((resolve) => llmServer.listen(0, "127.0.0.1", resolve)); + embeddingPort = embeddingServer.address().port; + llmPort = llmServer.address().port; + resetRegistration(); + }); + + afterEach(async () => { + resetRegistration(); + await new Promise((resolve) => embeddingServer.close(resolve)); + await new Promise((resolve) => llmServer.close(resolve)); + rmSync(workspaceDir, { recursive: true, force: true }); + }); + + function buildPluginConfig(extractMinMessages) { + return { + dbPath: path.join(workspaceDir, "db"), + autoCapture: true, + autoRecall: false, + smartExtraction: true, + extractMinMessages, + extractionThrottle: { skipLowValue: false, maxExtractionsPerHour: 200 }, + sessionCompression: { enabled: false }, + selfImprovement: { enabled: false, beforeResetNote: false, ensureLearningFiles: false }, + embedding: { + apiKey: "test-api-key", + model: "mock-embedding-model", + baseURL: `http://127.0.0.1:${embeddingPort}/v1`, + dimensions: EMBEDDING_DIMENSIONS, + }, + llm: { + apiKey: "test-api-key", + model: "mock-memory-model", + baseURL: `http://127.0.0.1:${llmPort}`, + }, + }; + } + + it("logs one INFO line on the first turn with messages/eligible/previousSeen/cumulative/fired, then stays silent on later turns in the same process", async () => { + const pluginConfig = buildPluginConfig(4); + const ctx = { sessionKey: "agent:agent-one:webchat", agentId: "agent-one" }; + const harness = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + + // Turn 1: 2 texts, below minMessages(4) -- gate does not fire yet. + await fireAgentEnd(hook, userMessages("alpha turn content", "bravo turn content"), ctx); + + let lines = payloadShapeLines(harness.logs); + assert.equal(lines.length, 1, "exactly one payload-shape INFO line after the first turn"); + assert.match(lines[0], /messages=2/); + assert.match(lines[0], /eligible=2/); + assert.match(lines[0], /previousSeen=0/); + assert.match(lines[0], /cumulative=2/); + assert.match(lines[0], /fired=no/); + + // Turn 2: 2 more texts, cumulative should now cross minMessages(4) and + // fire -- but the INFO line itself must NOT repeat (rate-limited to once + // per session per process). + await fireAgentEnd(hook, userMessages("charlie turn content", "delta turn content"), ctx); + + lines = payloadShapeLines(harness.logs); + assert.equal(lines.length, 1, "no additional payload-shape INFO line on a later turn in the same process"); + }); + + it("logs again for a different session in the same process (rate limit is per-session, not global)", async () => { + const pluginConfig = buildPluginConfig(4); + const harness = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + + await fireAgentEnd(hook, userMessages("alpha turn content"), { sessionKey: "agent:agent-one:webchat", agentId: "agent-one" }); + await fireAgentEnd(hook, userMessages("zulu turn content"), { sessionKey: "agent:agent-two:main", agentId: "agent-two" }); + + const lines = payloadShapeLines(harness.logs); + assert.equal(lines.length, 2, "each distinct session gets its own first-turn payload-shape line"); + assert.ok(lines.some((l) => l.includes("agent:agent-one:webchat"))); + assert.ok(lines.some((l) => l.includes("agent:agent-two:main"))); + }); + + it("logs again after a simulated restart (rate limit is per-process, not persisted)", async () => { + const pluginConfig = buildPluginConfig(4); + const ctx = { sessionKey: "agent:agent-one:webchat", agentId: "agent-one" }; + + const harness1 = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness1.api); + const hook1 = getAutoCaptureHook(harness1.eventHandlers); + await fireAgentEnd(hook1, userMessages("alpha turn content"), ctx); + assert.equal(payloadShapeLines(harness1.logs).length, 1); + + resetRegistration(); + const harness2 = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness2.api); + const hook2 = getAutoCaptureHook(harness2.eventHandlers); + await fireAgentEnd(hook2, userMessages("bravo turn content"), ctx); + + assert.equal( + payloadShapeLines(harness2.logs).length, + 1, + "a fresh process must re-emit the payload-shape line for a session it hasn't seen yet this run", + ); + }); + + it("reports fired=yes once cumulative reaches minMessages", async () => { + const pluginConfig = buildPluginConfig(2); + const ctx = { sessionKey: "agent:agent-two:main", agentId: "agent-two" }; + const harness = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + + await fireAgentEnd(hook, userMessages("alpha turn content", "bravo turn content"), ctx); + + const lines = payloadShapeLines(harness.logs); + assert.equal(lines.length, 1); + assert.match(lines[0], /cumulative=2/); + assert.match(lines[0], /fired=yes/); + }); +}); diff --git a/test/autocapture-unknown-watermark-injection.test.mjs b/test/autocapture-unknown-watermark-injection.test.mjs new file mode 100644 index 000000000..efb957637 --- /dev/null +++ b/test/autocapture-unknown-watermark-injection.test.mjs @@ -0,0 +1,289 @@ +/** + * Regression coverage for bounded injection when a session's watermark is + * genuinely unknown (first-ever run, or persisted state lost -- see + * test/autocapture-watermark-restart-survival.test.mjs for the + * restart-survivability half of this story) and the agent_end payload is + * history-carrying with far more than one batch's worth of messages. + * + * Without a cap, `previousSeenCount === 0` takes the unsliced default + * (`newTexts = eligibleTexts`), so the entire transcript is handed to + * extraction in one call -- expensive, and floods the extraction with + * mostly-stale content. This suite proves the injected window is capped to + * the most recent batch, the watermark jumps straight to the full length + * (the older prefix is forfeited, not queued for a later turn), and the + * very next turn behaves as a normal small delta. + * + * Fixtures are entirely synthetic; no real fleet data. + */ + +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import jitiFactory from "jiti"; + +const testDir = path.dirname(fileURLToPath(import.meta.url)); +const pluginSdkStubPath = path.resolve(testDir, "helpers", "openclaw-plugin-sdk-stub.mjs"); +const jiti = jitiFactory(import.meta.url, { + interopDefault: true, + alias: { + "openclaw/plugin-sdk": pluginSdkStubPath, + }, +}); + +const pluginModule = jiti("../index.ts"); +const memoryLanceDBProPlugin = pluginModule.default || pluginModule; +const resetRegistration = pluginModule.resetRegistration ?? (() => {}); +const { NoisePrototypeBank } = jiti("../src/noise-prototypes.ts"); +NoisePrototypeBank.prototype.isNoise = () => false; + +const EMBEDDING_DIMENSIONS = 64; + +function hashToIndex(text, dims) { + let h = 0; + for (let i = 0; i < text.length; i++) { + h = (h * 31 + text.charCodeAt(i)) >>> 0; + } + return h % dims; +} + +function oneHot(text) { + const v = new Array(EMBEDDING_DIMENSIONS).fill(0); + v[hashToIndex(text || "", EMBEDDING_DIMENSIONS)] = 1; + return v; +} + +function createEmbeddingServer() { + return http.createServer(async (req, res) => { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + const payload = JSON.parse(Buffer.concat(chunks).toString("utf8")); + const inputs = Array.isArray(payload.input) ? payload.input : [payload.input]; + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + object: "list", + data: inputs.map((input, index) => ({ object: "embedding", index, embedding: oneHot(String(input)) })), + model: payload.model || "mock-embedding-model", + usage: { prompt_tokens: 0, total_tokens: 0 }, + })); + }); +} + +function createLlmServer(extractionPrompts) { + let calls = 0; + return http.createServer(async (req, res) => { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + const payload = JSON.parse(Buffer.concat(chunks).toString("utf8")); + const prompt = String(payload.messages?.map((m) => m.content).join("\n") ?? ""); + if (prompt.includes("## Recent Conversation")) { + extractionPrompts.push(prompt); + } + calls += 1; + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion", + created: 1, + model: "mock-memory-model", + choices: [{ + index: 0, + finish_reason: "stop", + message: { + role: "assistant", + content: JSON.stringify({ + memories: [{ + category: "preferences", + abstract: `Synthetic preference marker number ${calls}`, + overview: `## Preference\n- Marker ${calls}`, + content: `User stated synthetic preference marker number ${calls}.`, + }], + }), + }, + }], + })); + }); +} + +function createPluginApiHarness({ pluginConfig, resolveRoot }) { + const eventHandlers = new Map(); + const logs = { info: [], warn: [], debug: [] }; + const api = { + pluginConfig, + resolvePath(target) { + if (typeof target !== "string") return target; + if (path.isAbsolute(target)) return target; + return path.join(resolveRoot, target); + }, + logger: { + info(message) { logs.info.push(String(message)); }, + warn(message) { logs.warn.push(String(message)); }, + debug(message) { logs.debug.push(String(message)); }, + }, + registerTool() {}, + registerCli() {}, + registerService() {}, + on(eventName, handler, meta) { + const list = eventHandlers.get(eventName) || []; + list.push({ handler, meta }); + eventHandlers.set(eventName, list); + }, + registerHook(eventName, handler, opts) { + const list = eventHandlers.get(eventName) || []; + list.push({ handler, meta: opts }); + eventHandlers.set(eventName, list); + }, + }; + return { api, eventHandlers, logs }; +} + +function getAutoCaptureHook(eventHandlers) { + const hooks = eventHandlers.get("agent_end") || []; + assert.ok(hooks.length >= 1, "expected at least one agent_end handler"); + return hooks[0].handler; +} + +async function fireAgentEnd(hook, messages, ctx) { + hook({ success: true, messages }, ctx); + const run = hook.__lastRun; + assert.ok(run && typeof run.then === "function", "expected a background capture run"); + await run; +} + +function userMessages(...texts) { + return texts.map((text) => ({ role: "user", content: text })); +} + +describe("bounded injection when the watermark is unknown and history is large", () => { + let workspaceDir; + let embeddingServer; + let llmServer; + let extractionPrompts; + + beforeEach(async () => { + workspaceDir = mkdtempSync(path.join(tmpdir(), "autocapture-unknown-watermark-")); + extractionPrompts = []; + embeddingServer = createEmbeddingServer(); + llmServer = createLlmServer(extractionPrompts); + await new Promise((resolve) => embeddingServer.listen(0, "127.0.0.1", resolve)); + await new Promise((resolve) => llmServer.listen(0, "127.0.0.1", resolve)); + resetRegistration(); + }); + + afterEach(async () => { + resetRegistration(); + await new Promise((resolve) => embeddingServer.close(resolve)); + await new Promise((resolve) => llmServer.close(resolve)); + rmSync(workspaceDir, { recursive: true, force: true }); + }); + + it("caps a 40-turn history to the batch window, forfeits the rest, and resumes as a normal delta next turn", async () => { + const embeddingPort = embeddingServer.address().port; + const llmPort = llmServer.address().port; + const harness = createPluginApiHarness({ + resolveRoot: workspaceDir, + pluginConfig: { + dbPath: path.join(workspaceDir, "db"), + autoCapture: true, + autoRecall: false, + smartExtraction: true, + extractMinMessages: 2, + extractMaxChars: 8000, + extractionThrottle: { skipLowValue: false, maxExtractionsPerHour: 200 }, + sessionCompression: { enabled: false }, + selfImprovement: { enabled: false, beforeResetNote: false, ensureLearningFiles: false }, + embedding: { + apiKey: "test-api-key", + model: "mock-embedding-model", + baseURL: `http://127.0.0.1:${embeddingPort}/v1`, + dimensions: EMBEDDING_DIMENSIONS, + }, + llm: { + apiKey: "test-api-key", + model: "mock-memory-model", + baseURL: `http://127.0.0.1:${llmPort}`, + }, + }, + }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + const ctx = { sessionKey: "agent:agent-one:webchat", agentId: "agent-one" }; + + // A 40-message history arrives in a single agent_end call on a session + // whose watermark has never been established (fresh Map, nothing + // persisted) -- e.g. a long-lived session picked up for the first time + // after the plugin was installed, or a persisted-state wipe. + const history = Array.from({ length: 40 }, (_, i) => `synthetic turn ${i + 1} content here`); + await fireAgentEnd(hook, userMessages(...history), ctx); + + assert.equal(extractionPrompts.length, 1, "the oversized history must still fire exactly one extraction"); + const firstPrompt = extractionPrompts[0]; + // Only the capped window (last extractMinMessages=2 texts) may appear. + assert.ok(firstPrompt.includes("synthetic turn 39 content here")); + assert.ok(firstPrompt.includes("synthetic turn 40 content here")); + for (let i = 1; i <= 38; i++) { + assert.ok( + !firstPrompt.includes(`synthetic turn ${i} content here`), + `forfeited turn ${i} must not appear in the capped extraction input`, + ); + } + + // Second call: one more new message. The watermark must have jumped to + // the FULL prior length (40, not just the 2-text window), so this turn + // is a normal small delta -- not another oversized dump, and not a + // re-read of the forfeited prefix. + await fireAgentEnd(hook, userMessages(...history, "synthetic turn 41 content here"), ctx); + assert.equal(extractionPrompts.length, 2, "the next turn's single new message must fire a normal delta extraction"); + const secondPrompt = extractionPrompts[1]; + assert.ok(secondPrompt.includes("synthetic turn 41 content here")); + for (let i = 1; i <= 40; i++) { + assert.ok( + !secondPrompt.includes(`synthetic turn ${i} content here`), + `turn 2 must not re-read forfeited or already-extracted turn ${i}`, + ); + } + }); + + it("does not cap a session with a small, normal amount of history even with an unknown watermark", async () => { + const embeddingPort = embeddingServer.address().port; + const llmPort = llmServer.address().port; + const harness = createPluginApiHarness({ + resolveRoot: workspaceDir, + pluginConfig: { + dbPath: path.join(workspaceDir, "db"), + autoCapture: true, + autoRecall: false, + smartExtraction: true, + extractMinMessages: 4, + extractMaxChars: 8000, + extractionThrottle: { skipLowValue: false, maxExtractionsPerHour: 200 }, + sessionCompression: { enabled: false }, + selfImprovement: { enabled: false, beforeResetNote: false, ensureLearningFiles: false }, + embedding: { + apiKey: "test-api-key", + model: "mock-embedding-model", + baseURL: `http://127.0.0.1:${embeddingPort}/v1`, + dimensions: EMBEDDING_DIMENSIONS, + }, + llm: { + apiKey: "test-api-key", + model: "mock-memory-model", + baseURL: `http://127.0.0.1:${llmPort}`, + }, + }, + }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + const ctx = { sessionKey: "agent:agent-two:main", agentId: "agent-two" }; + + // Only 3 texts total, at or under minMessages(4) -- a totally normal + // fresh session, not the "large unknown history" scenario. Must not be + // capped or forfeit anything, and must not even fire yet. + const smallHistory = ["alpha turn content", "bravo turn content", "charlie turn content"]; + await fireAgentEnd(hook, userMessages(...smallHistory), ctx); + assert.equal(extractionPrompts.length, 0, "a small fresh history under minMessages must simply defer, not fire"); + }); +}); diff --git a/test/autocapture-watermark-reset.test.mjs b/test/autocapture-watermark-reset.test.mjs index fe662fb21..8d36c9048 100644 --- a/test/autocapture-watermark-reset.test.mjs +++ b/test/autocapture-watermark-reset.test.mjs @@ -259,3 +259,154 @@ describe("auto-capture watermark after successful extraction (history flow)", () } }); }); + +describe("auto-capture watermark invalidation on session renewal", () => { + let workspaceDir; + let embeddingServer; + let llmServer; + let extractionPrompts; + + beforeEach(async () => { + workspaceDir = mkdtempSync(path.join(tmpdir(), "autocapture-renewal-")); + extractionPrompts = []; + embeddingServer = createEmbeddingServer(); + llmServer = createLlmServer(extractionPrompts); + await new Promise((resolve) => embeddingServer.listen(0, "127.0.0.1", resolve)); + await new Promise((resolve) => llmServer.listen(0, "127.0.0.1", resolve)); + resetRegistration(); + }); + + afterEach(async () => { + resetRegistration(); + await new Promise((resolve) => embeddingServer.close(resolve)); + await new Promise((resolve) => llmServer.close(resolve)); + rmSync(workspaceDir, { recursive: true, force: true }); + }); + + function makeConfig() { + return { + dbPath: path.join(workspaceDir, "db"), + autoCapture: true, + autoRecall: false, + smartExtraction: true, + extractMinMessages: 2, + extractionThrottle: { skipLowValue: false, maxExtractionsPerHour: 200 }, + sessionCompression: { enabled: false }, + selfImprovement: { enabled: false, beforeResetNote: false, ensureLearningFiles: false }, + embedding: { + apiKey: "test-api-key", + model: "mock-embedding-model", + baseURL: `http://127.0.0.1:${embeddingServer.address().port}/v1`, + dimensions: EMBEDDING_DIMENSIONS, + }, + llm: { + apiKey: "test-api-key", + model: "mock-memory-model", + baseURL: `http://127.0.0.1:${llmServer.address().port}`, + }, + }; + } + + it("resets the watermark when the sessionId rotates under a stable sessionKey (stale-HIGH swallow case)", async () => { + const harness = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig: makeConfig() }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + + // Session A: two texts, extraction fires, watermark now counts them. + await fireAgentEnd(hook, userMessages(...TURN_1_TEXTS), { + sessionKey: "agent:dave:main", + agentId: "dave", + sessionId: "session-aaa", + }); + assert.equal(extractionPrompts.length, 1, "session A must extract"); + + // Renewal: same key, NEW sessionId, and the fresh session's history is + // the same SIZE as the stale cursor. Without invalidation the payload + // reads as already-seen and is silently swallowed. + await fireAgentEnd(hook, userMessages(...TURN_2_TEXTS), { + sessionKey: "agent:dave:main", + agentId: "dave", + sessionId: "session-bbb", + }); + assert.equal(extractionPrompts.length, 2, "the renewed session's first payload must extract, not be swallowed"); + assert.ok( + extractionPrompts[1].includes(TURN_2_TEXTS[0]) && extractionPrompts[1].includes(TURN_2_TEXTS[1]), + "the renewed session's own texts must be extracted", + ); + assert.ok( + harness.logs.info.some((m) => m.includes("watermark reset") && m.includes("session renewed")), + "the reset must be visible at the standard log level", + ); + }); + + it("does not reset while the sessionId stays stable", async () => { + const harness = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig: makeConfig() }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + const ctx = { sessionKey: "agent:dave:main", agentId: "dave", sessionId: "session-aaa" }; + + await fireAgentEnd(hook, userMessages(...TURN_1_TEXTS), ctx); + await fireAgentEnd(hook, userMessages(...TURN_1_TEXTS, ...TURN_2_TEXTS), ctx); + + assert.equal(extractionPrompts.length, 2, "normal delta flow is unchanged"); + assert.ok( + !extractionPrompts[1].includes(TURN_1_TEXTS[0]), + "the stable-session delta must not re-read extracted history", + ); + assert.ok( + !harness.logs.info.some((m) => m.includes("session renewed")), + "no renewal reset may fire for a stable sessionId", + ); + }); + + it("does not reset when the hook context carries no sessionId", async () => { + const harness = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig: makeConfig() }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + const ctx = { sessionKey: "agent:dave:main", agentId: "dave" }; + + await fireAgentEnd(hook, userMessages(...TURN_1_TEXTS), ctx); + await fireAgentEnd(hook, userMessages(...TURN_1_TEXTS, ...TURN_2_TEXTS), ctx); + + assert.equal(extractionPrompts.length, 2); + assert.ok( + !harness.logs.info.some((m) => m.includes("session renewed")), + "a missing renewal signal must never trigger a reset", + ); + }); + + it("records without resetting on the first sighting after a restart (restart survival preserved)", async () => { + const config = makeConfig(); + const first = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig: config }); + memoryLanceDBProPlugin.register(first.api); + const firstHook = getAutoCaptureHook(first.eventHandlers); + const ctx = { sessionKey: "agent:dave:main", agentId: "dave", sessionId: "session-aaa" }; + + await fireAgentEnd(firstHook, userMessages(...TURN_1_TEXTS), ctx); + assert.equal(extractionPrompts.length, 1); + + // Restart: fresh process state, same dbPath. The persisted watermark + // rehydrates; the in-memory sessionId map starts empty and must record + // the (unchanged) sessionId without resetting anything. + resetRegistration(); + const second = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig: config }); + memoryLanceDBProPlugin.register(second.api); + const secondHook = getAutoCaptureHook(second.eventHandlers); + + await fireAgentEnd(secondHook, userMessages(...TURN_1_TEXTS), ctx); + // The persisted watermark must rehydrate (previousSeen carries across the + // restart). Known separate limitation, deliberately NOT pinned here: the + // in-memory flow detector also resets, so the first post-restart history + // payload is re-counted as new and re-extracts once (dedup absorbs it); + // that is the restart-reclassification sibling of the delta-fed stall, + // tracked on the watermark family ticket. + assert.ok( + [...second.logs.info, ...second.logs.debug].some((m) => m.includes("previousSeen=2")), + "the persisted watermark must rehydrate after a restart", + ); + assert.ok( + !second.logs.info.some((m) => m.includes("session renewed")), + "a restart must never masquerade as a session renewal", + ); + }); +}); diff --git a/test/autocapture-watermark-restart-survival.test.mjs b/test/autocapture-watermark-restart-survival.test.mjs new file mode 100644 index 000000000..ccd294734 --- /dev/null +++ b/test/autocapture-watermark-restart-survival.test.mjs @@ -0,0 +1,398 @@ +/** + * Regression coverage for restart-survivability of the auto-capture watermark + * (`autoCaptureSeenTextCount`). + * + * `autoCaptureSeenTextCount` is an in-memory Map. Any plugin/process restart + * wipes it. For a session that banks partial progress toward + * `extractMinMessages` (via the ingress accumulator path, where a + * below-threshold turn's tentative advance is intentionally kept rather than + * rolled back — see the "below-threshold turns are deferred" comment this + * suite pins in the second describe block below), a restart silently forgets + * that progress. If the session's `agent_end` payload is delta-only (only the + * newest turn, not the full transcript), there is no way to recover the lost + * count from the payload itself, so the session needs strictly more new + * messages after the restart than it would have needed without one. + * + * This suite proves the watermark survives a restart by persisting it + * alongside the LanceDB store and rehydrating on the next `register()` call, + * using the same "simulate a restart" technique as the rest of this test + * family: `resetRegistration()` wipes the in-process singleton (including + * every in-memory Map), then `register()` is called again against the same + * `dbPath` — exactly what happens when the host process restarts and the + * plugin reinitializes against its existing on-disk state. + * + * Fixtures are entirely synthetic; no real fleet data. + */ + +import { describe, it, beforeEach, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import jitiFactory from "jiti"; + +const testDir = path.dirname(fileURLToPath(import.meta.url)); +const pluginSdkStubPath = path.resolve(testDir, "helpers", "openclaw-plugin-sdk-stub.mjs"); +const jiti = jitiFactory(import.meta.url, { + interopDefault: true, + alias: { + "openclaw/plugin-sdk": pluginSdkStubPath, + }, +}); + +const pluginModule = jiti("../index.ts"); +const memoryLanceDBProPlugin = pluginModule.default || pluginModule; +const resetRegistration = pluginModule.resetRegistration ?? (() => {}); +// The embedding mock below returns one-hot vectors, which can land arbitrary +// texts near noise prototypes; force the bank off for determinism. +const { NoisePrototypeBank } = jiti("../src/noise-prototypes.ts"); +NoisePrototypeBank.prototype.isNoise = () => false; + +const EMBEDDING_DIMENSIONS = 64; + +function hashToIndex(text, dims) { + let h = 0; + for (let i = 0; i < text.length; i++) { + h = (h * 31 + text.charCodeAt(i)) >>> 0; + } + return h % dims; +} + +function oneHot(text) { + const v = new Array(EMBEDDING_DIMENSIONS).fill(0); + v[hashToIndex(text || "", EMBEDDING_DIMENSIONS)] = 1; + return v; +} + +function createEmbeddingServer() { + return http.createServer(async (req, res) => { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + const payload = JSON.parse(Buffer.concat(chunks).toString("utf8")); + const inputs = Array.isArray(payload.input) ? payload.input : [payload.input]; + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + object: "list", + data: inputs.map((input, index) => ({ object: "embedding", index, embedding: oneHot(String(input)) })), + model: payload.model || "mock-embedding-model", + usage: { prompt_tokens: 0, total_tokens: 0 }, + })); + }); +} + +/** + * LLM mock: records every extract-candidates prompt, answers each with one + * distinct memory (distinct abstracts embed to distinct one-hot vectors, so + * dedup never matches and every extraction creates). + */ +function createLlmServer(extractionPrompts) { + let calls = 0; + return http.createServer(async (req, res) => { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + const payload = JSON.parse(Buffer.concat(chunks).toString("utf8")); + const prompt = String(payload.messages?.map((m) => m.content).join("\n") ?? ""); + if (prompt.includes("## Recent Conversation")) { + extractionPrompts.push(prompt); + } + calls += 1; + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion", + created: 1, + model: "mock-memory-model", + choices: [{ + index: 0, + finish_reason: "stop", + message: { + role: "assistant", + content: JSON.stringify({ + memories: [{ + category: "preferences", + abstract: `Synthetic preference marker number ${calls}`, + overview: `## Preference\n- Marker ${calls}`, + content: `User stated synthetic preference marker number ${calls}.`, + }], + }), + }, + }], + })); + }); +} + +function createPluginApiHarness({ pluginConfig, resolveRoot }) { + const eventHandlers = new Map(); + const logs = { info: [], warn: [], debug: [] }; + const api = { + pluginConfig, + resolvePath(target) { + if (typeof target !== "string") return target; + if (path.isAbsolute(target)) return target; + return path.join(resolveRoot, target); + }, + logger: { + info(message) { logs.info.push(String(message)); }, + warn(message) { logs.warn.push(String(message)); }, + debug(message) { logs.debug.push(String(message)); }, + }, + registerTool() {}, + registerCli() {}, + registerService() {}, + on(eventName, handler, meta) { + const list = eventHandlers.get(eventName) || []; + list.push({ handler, meta }); + eventHandlers.set(eventName, list); + }, + registerHook(eventName, handler, opts) { + const list = eventHandlers.get(eventName) || []; + list.push({ handler, meta: opts }); + eventHandlers.set(eventName, list); + }, + }; + return { api, eventHandlers, logs }; +} + +function getAutoCaptureHook(eventHandlers) { + const hooks = eventHandlers.get("agent_end") || []; + assert.ok(hooks.length >= 1, "expected at least one agent_end handler"); + return hooks[0].handler; +} + +async function fireAgentEnd(hook, messages, ctx) { + hook({ success: true, messages }, ctx); + const run = hook.__lastRun; + assert.ok(run && typeof run.then === "function", "expected a background capture run"); + await run; +} + +function fireMessageReceived(eventHandlers, content, ctx) { + const hooks = eventHandlers.get("message_received") || []; + assert.ok(hooks.length >= 1, "expected at least one message_received handler"); + hooks[0].handler({ content, from: ctx.from || "user" }, ctx); +} + +function userMessages(...texts) { + return texts.map((text) => ({ role: "user", content: text })); +} + +function buildPluginConfig({ workspaceDir, embeddingPort, llmPort, extractMinMessages }) { + return { + dbPath: path.join(workspaceDir, "db"), + autoCapture: true, + autoRecall: false, + smartExtraction: true, + extractMinMessages, + extractionThrottle: { skipLowValue: false, maxExtractionsPerHour: 200 }, + sessionCompression: { enabled: false }, + selfImprovement: { enabled: false, beforeResetNote: false, ensureLearningFiles: false }, + embedding: { + apiKey: "test-api-key", + model: "mock-embedding-model", + baseURL: `http://127.0.0.1:${embeddingPort}/v1`, + dimensions: EMBEDDING_DIMENSIONS, + }, + llm: { + apiKey: "test-api-key", + model: "mock-memory-model", + baseURL: `http://127.0.0.1:${llmPort}`, + }, + }; +} + +describe("auto-capture watermark survives a simulated process restart", () => { + let workspaceDir; + let embeddingServer; + let llmServer; + let embeddingPort; + let llmPort; + let extractionPrompts; + + beforeEach(async () => { + workspaceDir = mkdtempSync(path.join(tmpdir(), "autocapture-restart-")); + extractionPrompts = []; + embeddingServer = createEmbeddingServer(); + llmServer = createLlmServer(extractionPrompts); + await new Promise((resolve) => embeddingServer.listen(0, "127.0.0.1", resolve)); + await new Promise((resolve) => llmServer.listen(0, "127.0.0.1", resolve)); + embeddingPort = embeddingServer.address().port; + llmPort = llmServer.address().port; + resetRegistration(); + }); + + afterEach(async () => { + resetRegistration(); + await new Promise((resolve) => embeddingServer.close(resolve)); + await new Promise((resolve) => llmServer.close(resolve)); + rmSync(workspaceDir, { recursive: true, force: true }); + }); + + it("an ingress-fed, delta-only session resumes counting from where it left off after a restart and fires once cumulative crosses minMessages", async () => { + const pluginConfig = buildPluginConfig({ workspaceDir, embeddingPort, llmPort, extractMinMessages: 2 }); + const ctx = { sessionKey: "agent:agent-one:webchat", agentId: "agent-one", channelId: "webchat" }; + + // --- "process 1" --- + const harness1 = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness1.api); + const hook1 = getAutoCaptureHook(harness1.eventHandlers); + + // Turn 1: exactly one message arrives via the ingress path (message_received), + // and agent_end's own payload is delta-only (just this turn's exchange). + // cumulative = 0 + 1 = 1 < minMessages(2) -> deferred, not fired. + fireMessageReceived(harness1.eventHandlers, "I keep my synthetic dotfiles in a bare repo named quartz.", ctx); + await fireAgentEnd( + hook1, + userMessages("I keep my synthetic dotfiles in a bare repo named quartz."), + ctx, + ); + assert.equal(extractionPrompts.length, 0, "turn 1 alone must not cross minMessages yet"); + + // --- simulated restart: fresh singleton, fresh in-memory Maps, same dbPath --- + resetRegistration(); + const harness2 = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness2.api); + const hook2 = getAutoCaptureHook(harness2.eventHandlers); + + // Turn 2 (post-restart): exactly one MORE new message, still delta-only. + // Without restart-survivability the watermark would have been wiped back + // to 0 and this turn alone (cumulative=1) would defer again -- forever, + // since every subsequent turn contributes the same single message. With + // the persisted watermark rehydrated to 1, this turn's contribution + // brings cumulative to 2 >= minMessages(2), and extraction fires. + fireMessageReceived(harness2.eventHandlers, "My preferred terminal font is a synthetic monospace called Duckspace.", ctx); + await fireAgentEnd( + hook2, + userMessages("My preferred terminal font is a synthetic monospace called Duckspace."), + ctx, + ); + + assert.equal( + extractionPrompts.length, + 1, + "post-restart turn must fire once the persisted watermark's cumulative count reaches minMessages", + ); + }); + + it("a history-carrying session's slice cursor survives a restart without re-extracting already-consumed history", async () => { + const pluginConfig = buildPluginConfig({ workspaceDir, embeddingPort, llmPort, extractMinMessages: 4 }); + const ctx = { sessionKey: "agent:agent-two:main", agentId: "agent-two" }; + + const TURN_1_TEXTS = [ + "I keep my synthetic dotfiles in a bare repository named quartz.", + "My preferred terminal font is a synthetic monospace called Duckspace.", + ]; + const TURN_2_TEXTS = [ + "For synthetic backups I rotate three encrypted drives weekly.", + "My synthetic editor theme of choice is called Marmalade Night.", + ]; + const TURN_3_TEXTS = [ + "My synthetic standing desk motor brand is called Elevar.", + "I label synthetic backup drives with constellation names.", + ]; + + // --- "process 1" --- + const harness1 = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness1.api); + const hook1 = getAutoCaptureHook(harness1.eventHandlers); + + // Turn 1: history so far = 2 texts, cumulative=2 < minMessages(4) -> deferred. + await fireAgentEnd(hook1, userMessages(...TURN_1_TEXTS), ctx); + assert.equal(extractionPrompts.length, 0, "turn 1 alone must not cross minMessages yet"); + + // Turn 2: history so far = 4 texts, cumulative=4 >= minMessages(4) -> fires, + // consuming all 4 (nothing was extracted before this point). + await fireAgentEnd(hook1, userMessages(...TURN_1_TEXTS, ...TURN_2_TEXTS), ctx); + assert.equal(extractionPrompts.length, 1, "turn 2 must cross minMessages and fire"); + + // --- simulated restart: fresh singleton, fresh in-memory Maps, same dbPath --- + resetRegistration(); + const harness2 = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness2.api); + const hook2 = getAutoCaptureHook(harness2.eventHandlers); + + // Turn 3 (post-restart): history so far = 6 texts (4 already consumed + 2 new). + // Without restart-survivability the watermark would have been wiped to 0, + // and this turn would re-extract ALL 6 texts (re-reading the 4 that were + // already extracted pre-restart). With the persisted cursor rehydrated to + // 4, this turn must see only the 2 new texts. + await fireAgentEnd(hook2, userMessages(...TURN_1_TEXTS, ...TURN_2_TEXTS, ...TURN_3_TEXTS), ctx); + assert.equal(extractionPrompts.length, 2, "post-restart turn must fire on the delta only"); + + const thirdPrompt = extractionPrompts[1]; + assert.ok( + thirdPrompt.includes(TURN_3_TEXTS[0]) && thirdPrompt.includes(TURN_3_TEXTS[1]), + "post-restart extraction must see the new texts", + ); + for (const alreadyExtracted of [...TURN_1_TEXTS, ...TURN_2_TEXTS]) { + assert.ok( + !thirdPrompt.includes(alreadyExtracted), + `post-restart extraction must not re-read already-extracted history: ${alreadyExtracted.slice(0, 40)}`, + ); + } + }); +}); + +describe("auto-capture watermark rollback-on-skip (pinned so restart-survivability work cannot silently drop it)", () => { + let workspaceDir; + let embeddingServer; + let llmServer; + let extractionPrompts; + + beforeEach(async () => { + workspaceDir = mkdtempSync(path.join(tmpdir(), "autocapture-rollback-")); + extractionPrompts = []; + embeddingServer = createEmbeddingServer(); + llmServer = createLlmServer(extractionPrompts); + await new Promise((resolve) => embeddingServer.listen(0, "127.0.0.1", resolve)); + await new Promise((resolve) => llmServer.listen(0, "127.0.0.1", resolve)); + resetRegistration(); + }); + + afterEach(async () => { + resetRegistration(); + await new Promise((resolve) => embeddingServer.close(resolve)); + await new Promise((resolve) => llmServer.close(resolve)); + rmSync(workspaceDir, { recursive: true, force: true }); + }); + + it("below-threshold history-flow turns roll the cursor back so a later turn's slice still includes their content", async () => { + const embeddingPort = embeddingServer.address().port; + const llmPort = llmServer.address().port; + const pluginConfig = buildPluginConfig({ workspaceDir, embeddingPort, llmPort, extractMinMessages: 6 }); + const ctx = { sessionKey: "agent:agent-two:main", agentId: "agent-two" }; + + const harness = createPluginApiHarness({ resolveRoot: workspaceDir, pluginConfig }); + memoryLanceDBProPlugin.register(harness.api); + const hook = getAutoCaptureHook(harness.eventHandlers); + + const TURN_1 = ["Synthetic marker alpha.", "Synthetic marker bravo."]; + const TURN_2 = ["Synthetic marker charlie.", "Synthetic marker delta."]; + const TURN_3 = ["Synthetic marker echo.", "Synthetic marker foxtrot."]; + + // Turn 1: cumulative=2 < 6 -> deferred; cursor rolls back to 0 (a no-op, + // since previousSeenCount was already 0). + await fireAgentEnd(hook, userMessages(...TURN_1), ctx); + assert.equal(extractionPrompts.length, 0); + + // Turn 2: history so far = 4 texts. If the cursor had NOT rolled back and + // instead advanced to a tentative 2, this turn would slice off only the 2 + // new texts (cumulative=2+2=4, still <6, still deferred, but now tracking + // 4). Either way still deferred here -- the meaningful assertion is turn 3. + await fireAgentEnd(hook, userMessages(...TURN_1, ...TURN_2), ctx); + assert.equal(extractionPrompts.length, 0); + + // Turn 3: history so far = 6 texts, cumulative=6 >= 6 -> fires. Because the + // cursor was rolled back (not tentatively advanced) on every deferred + // turn, this extraction's input is the FULL 6-text window -- proving no + // content from turns 1-2 was silently forfeited while deferred. + await fireAgentEnd(hook, userMessages(...TURN_1, ...TURN_2, ...TURN_3), ctx); + assert.equal(extractionPrompts.length, 1, "turn 3 must cross minMessages and fire"); + + const prompt = extractionPrompts[0]; + for (const text of [...TURN_1, ...TURN_2, ...TURN_3]) { + assert.ok(prompt.includes(text), `deferred-turn content must survive to the firing extraction: ${text}`); + } + }); +});