From 53aa6d3c31d6371816fd62d77180b11f8e71f27a Mon Sep 17 00:00:00 2001 From: Ralph Chang Date: Tue, 28 Apr 2026 11:59:29 +0800 Subject: [PATCH] feat: implement Plan 1 - Critical Stability fixes Wave 1: Storage and Journal Safety - Add frozen cache TTL (1h) and size bounds (50 sessions) - Add pending journal source-aware retention (compaction-only TTL) - Add inter-process file lock with stale recovery - Move processLatestUserMessage to first transform (after isSubAgent guard) Wave 2: Promotion Ownership and Bounded Rejection - Add pendingOwnerSessionID/pendingMessageID metadata - Add owner-aware pending journal clearing - Add explicit/manual bounded retry (max 3 attempts) - Fix session.deleted cleanup idempotency Wave 3: Normalize, Security, and Cache Hardening - Fix load-time write loop (only write on security/migration change) - Add deterministic sort tie-breaker (createdAt -> id) - Add Bearer token redaction - Add processed message cache bounds - Remove priorityWithFreshness dead code Tests: 180 pass, 0 fail --- src/pending-journal.ts | 95 ++++-- src/plugin.ts | 141 ++++++-- src/promotion-accounting.ts | 43 ++- src/storage.ts | 91 +++++- src/types.ts | 17 + src/workspace-memory.ts | 52 +-- tests/pending-journal.test.ts | 208 +++++++++++- tests/plugin.test.ts | 495 ++++++++++++++++++++++++++++- tests/promotion-accounting.test.ts | 19 ++ tests/storage.test.ts | 83 +++++ tests/workspace-memory.test.ts | 213 ++++++++++++- 11 files changed, 1346 insertions(+), 111 deletions(-) create mode 100644 tests/storage.test.ts diff --git a/src/pending-journal.ts b/src/pending-journal.ts index 24f9dd5..b89e2d8 100644 --- a/src/pending-journal.ts +++ b/src/pending-journal.ts @@ -1,4 +1,5 @@ import type { LongTermMemoryEntry, PendingMemoryJournalStore } from "./types.ts"; +import { PROMOTION_RETRY_LIMITS } from "./types.ts"; import { workspaceKey, workspacePendingJournalPath } from "./paths.ts"; import { atomicWriteJSON, readJSON, updateJSON } from "./storage.ts"; @@ -42,7 +43,7 @@ function dedupeByText(entries: LongTermMemoryEntry[]): LongTermMemoryEntry[] { const result: LongTermMemoryEntry[] = []; for (const entry of entries) { - const key = memoryKey(entry); + const key = `${memoryKey(entry)}\u0000${entry.pendingOwnerSessionID ?? ""}`; if (seen.has(key)) continue; seen.add(key); result.push(entry); @@ -67,14 +68,15 @@ function entryTime(entry: LongTermMemoryEntry): number { function isStaleEntry(entry: LongTermMemoryEntry, maxAgeDays: number): boolean { const time = entryTime(entry); - - // If timestamp is 0 (both invalid), treat as stale + + // Invalid timestamps are corruption safety and apply to every source. if (time === 0) return true; - - const ageMs = Date.now() - time; - const maxAgeMs = maxAgeDays * 24 * 60 * 60 * 1000; - - return ageMs > maxAgeMs; + + // TTL policy applies only to compaction candidates. Explicit/manual entries + // represent user intent and should survive age while under the hard cap. + if (entry.source !== "compaction") return false; + + return Date.now() - time > maxAgeDays * 86_400_000; } function applyRetention( @@ -82,23 +84,18 @@ function applyRetention( maxEntries: number, maxAgeDays: number, ): LongTermMemoryEntry[] { - // 1. Dedupe first const deduped = dedupeByText(entries); - - // 2. Remove stale entries const freshEntries = deduped.filter(entry => !isStaleEntry(entry, maxAgeDays)); - - // 3. Sort by entryTime descending (newest first) for cap, using updatedAt then createdAt const sorted = [...freshEntries].sort((a, b) => { - return entryTime(b) - entryTime(a); + const timeDiff = entryTime(b) - entryTime(a); + if (timeDiff !== 0) return timeDiff; + return a.id.localeCompare(b.id); }); - - // 4. Keep maxEntries newest const capped = sorted.slice(0, maxEntries); - - // 5. Restore stable order (oldest-to-newest) for consistency with existing code return capped.sort((a, b) => { - return entryTime(a) - entryTime(b); + const timeDiff = entryTime(a) - entryTime(b); + if (timeDiff !== 0) return timeDiff; + return a.id.localeCompare(b.id); }); } @@ -159,13 +156,69 @@ export async function hasPendingJournalEntries(root: string): Promise { return journal.entries.length > 0; } -export async function clearPendingMemories(root: string, keys?: Set): Promise { +export async function clearPendingMemories( + root: string, + keys?: Set, + options: { ownerSessionID?: string; clearUnowned?: boolean } = {}, +): Promise { await updatePendingJournal(root, store => { if (!keys || keys.size === 0) { store.entries = []; return store; } - store.entries = store.entries.filter(entry => !keys.has(memoryKey(entry))); + + store.entries = store.entries.filter(entry => { + if (!keys.has(memoryKey(entry))) return true; + if (!options.ownerSessionID) return false; + if (entry.pendingOwnerSessionID === options.ownerSessionID) return false; + if (options.clearUnowned && !entry.pendingOwnerSessionID) return false; + return true; + }); return store; }); } + +export async function recordPromotionRejections( + root: string, + keys: Set, + reason: string, + options: { ownerSessionID?: string } = {}, +): Promise> { + const exhausted = new Set(); + if (keys.size === 0) return exhausted; + + await updatePendingJournal(root, store => { + const nowIso = new Date().toISOString(); + const exhaustedEntries = new Set(); + + store.entries = store.entries.map(entry => { + const key = memoryKey(entry); + if (!keys.has(key)) return entry; + if (options.ownerSessionID && entry.pendingOwnerSessionID !== options.ownerSessionID) return entry; + + const promotionAttempts = (entry.promotionAttempts ?? 0) + 1; + const max = entry.source === "manual" + ? PROMOTION_RETRY_LIMITS.maxManualAttempts + : PROMOTION_RETRY_LIMITS.maxExplicitAttempts; + + if (promotionAttempts >= max) { + exhausted.add(key); + exhaustedEntries.add(`${key}\u0000${entry.pendingOwnerSessionID ?? ""}`); + } + + return { + ...entry, + promotionAttempts, + lastPromotionAttemptAt: nowIso, + lastPromotionFailureReason: reason, + }; + }); + + store.entries = store.entries.filter(entry => ( + !exhaustedEntries.has(`${memoryKey(entry)}\u0000${entry.pendingOwnerSessionID ?? ""}`) + )); + return store; + }); + + return exhausted; +} diff --git a/src/plugin.ts b/src/plugin.ts index fa21798..20aa48c 100644 --- a/src/plugin.ts +++ b/src/plugin.ts @@ -43,6 +43,7 @@ import { hasPendingJournalEntries, loadPendingJournal, memoryKey, + recordPromotionRejections, } from "./pending-journal.ts"; import { loadSessionState, @@ -61,6 +62,7 @@ import { pendingTodos, } from "./opencode.ts"; import { accountPendingPromotions } from "./promotion-accounting.ts"; +import { WORKSPACE_MEMORY_CACHE_LIMITS } from "./types.ts"; /** * Build the complete compaction prompt. @@ -203,13 +205,67 @@ export const MemoryV2Plugin: Plugin = async (input) => { // Cache for processed user message IDs (to avoid duplicate processing) const processedUserMessages = new Map>(); + function pruneFrozenWorkspaceMemoryCache(now = Date.now()): void { + for (const [sessionID, cached] of frozenWorkspaceMemoryCache) { + if (now - cached.loadedAt > WORKSPACE_MEMORY_CACHE_LIMITS.frozenTtlMs) { + frozenWorkspaceMemoryCache.delete(sessionID); + } + } + + while (frozenWorkspaceMemoryCache.size > WORKSPACE_MEMORY_CACHE_LIMITS.maxFrozenSessions) { + const oldest = [...frozenWorkspaceMemoryCache.entries()] + .sort((a, b) => a[1].loadedAt - b[1].loadedAt)[0]?.[0]; + if (!oldest) break; + frozenWorkspaceMemoryCache.delete(oldest); + } + } + + function pruneProcessedUserMessagesCache(): void { + for (const [sessionID, messages] of processedUserMessages) { + while (messages.size > WORKSPACE_MEMORY_CACHE_LIMITS.maxProcessedMessagesPerSession) { + const oldest = messages.values().next().value as string | undefined; + if (!oldest) break; + messages.delete(oldest); + } + + if (messages.size === 0) { + processedUserMessages.delete(sessionID); + } + } + + while (processedUserMessages.size > WORKSPACE_MEMORY_CACHE_LIMITS.maxProcessedSessionIDs) { + const oldestSessionID = processedUserMessages.keys().next().value as string | undefined; + if (!oldestSessionID) break; + processedUserMessages.delete(oldestSessionID); + } + } + + function rememberProcessedUserMessage(sessionID: string, messageID: string, processedForSession: Set): void { + processedForSession.add(messageID); + while (processedForSession.size > WORKSPACE_MEMORY_CACHE_LIMITS.maxProcessedMessagesPerSession) { + const oldest = processedForSession.values().next().value as string | undefined; + if (!oldest) break; + processedForSession.delete(oldest); + } + + if (processedUserMessages.has(sessionID)) { + processedUserMessages.delete(sessionID); + } + processedUserMessages.set(sessionID, processedForSession); + pruneProcessedUserMessagesCache(); + } + async function processLatestUserMessage(sessionID: string): Promise { const processedForSession = processedUserMessages.get(sessionID) ?? new Set(); const latestMessage = await latestUserText(client, sessionID); if (!latestMessage?.id || processedForSession.has(latestMessage.id)) return; - const memories = extractExplicitMemories(latestMessage.text); + const memories = extractExplicitMemories(latestMessage.text).map(memory => ({ + ...memory, + pendingOwnerSessionID: sessionID, + pendingMessageID: latestMessage.id, + })); const decisions = memories.filter(memory => memory.type === "decision"); if (memories.length > 0) { @@ -233,19 +289,29 @@ export const MemoryV2Plugin: Plugin = async (input) => { }); } - processedForSession.add(latestMessage.id); - processedUserMessages.set(sessionID, processedForSession); + rememberProcessedUserMessage(sessionID, latestMessage.id, processedForSession); } - async function promotePendingMemories(sessionID?: string): Promise { + async function promotePendingMemories( + sessionID?: string, + options: { includeUnownedJournal?: boolean; includeOwnedJournal?: boolean } = {}, + ): Promise { + const includeUnownedJournal = options.includeUnownedJournal ?? !sessionID; + const includeOwnedJournal = options.includeOwnedJournal ?? Boolean(sessionID); const [journal, sessionState] = await Promise.all([ loadPendingJournal(directory), sessionID ? loadSessionState(directory, sessionID) : Promise.resolve(undefined), ]); + const journalPending = journal.entries.filter(memory => { + if (sessionID && includeOwnedJournal && memory.pendingOwnerSessionID === sessionID) return true; + if (includeUnownedJournal && !memory.pendingOwnerSessionID) return true; + return false; + }); + const pending = [ ...(sessionState?.pendingMemories ?? []), - ...journal.entries, + ...journalPending, ]; if (pending.length === 0) return; @@ -277,16 +343,39 @@ export const MemoryV2Plugin: Plugin = async (input) => { events: updateResult.events, }); + const exhaustedRejectedKeys = await recordPromotionRejections( + directory, + accounting.retryableRejectedKeys, + "rejected_capacity", + { ownerSessionID: sessionID }, + ); + + const sessionRemovalKeys = new Set([ + ...accounting.clearableKeys, + ...exhaustedRejectedKeys, + ]); + if (sessionID) { await updateSessionState(directory, sessionID, state => { - state.pendingMemories = state.pendingMemories.filter(memory => !accounting.clearableKeys.has(memoryKey(memory))); + state.pendingMemories = state.pendingMemories.filter(memory => { + const key = memoryKey(memory); + if (!sessionRemovalKeys.has(key)) return true; + + if (accounting.clearableKeys.has(key)) return false; + if (exhaustedRejectedKeys.has(key)) return false; + + return true; + }); return state; }); clearFrozenWorkspaceMemoryCache(sessionID); } if (accounting.clearableKeys.size > 0) { - await clearPendingMemories(directory, accounting.clearableKeys); + await clearPendingMemories(directory, accounting.clearableKeys, { + ownerSessionID: sessionID, + clearUnowned: !sessionID || includeUnownedJournal === true, + }); } } @@ -324,6 +413,7 @@ export const MemoryV2Plugin: Plugin = async (input) => { renderedPrompt: string; }> { const now = Date.now(); + pruneFrozenWorkspaceMemoryCache(now); const cached = frozenWorkspaceMemoryCache.get(sessionID); // Cache is valid for the current session cache epoch. @@ -336,6 +426,7 @@ export const MemoryV2Plugin: Plugin = async (input) => { const store = await loadWorkspaceMemory(root); const renderedPrompt = renderWorkspaceMemory(store); frozenWorkspaceMemoryCache.set(sessionID, { store, renderedPrompt, loadedAt: now }); + pruneFrozenWorkspaceMemoryCache(now); return { store, renderedPrompt }; } @@ -357,19 +448,23 @@ export const MemoryV2Plugin: Plugin = async (input) => { const { sessionID } = hookInput; if (!sessionID) return; + pruneFrozenWorkspaceMemoryCache(); + pruneProcessedUserMessagesCache(); + // Sub-agents are short-lived - skip memory system if (await isSubAgent(sessionID)) return; - // Before first snapshot in this session, promote durable pending memories from - // prior sessions. Keep this before processing latest user text so current-turn - // explicit memory remains pending (not immediately frozen into system[1]). - if (!frozenWorkspaceMemoryCache.has(sessionID) && await hasPendingJournalEntries(directory)) { - await promotePendingMemories(); - } - - // Process explicit user memory even on no-tool turns. + // Process explicit user memory even on no-tool turns. Keep this after the + // sub-agent guard so child sessions never append to the parent journal. await processLatestUserMessage(sessionID); + // Before first snapshot in this session, promote durable unowned backlog from + // prior sessions. Current-turn owned explicit memory remains pending and only + // appears in hot state for this transform. + if (!frozenWorkspaceMemoryCache.has(sessionID) && await hasPendingJournalEntries(directory)) { + await promotePendingMemories(undefined, { includeUnownedJournal: true, includeOwnedJournal: false }); + } + // Get frozen workspace memory snapshot (loaded and rendered once per session) const workspaceSnapshot = await getFrozenWorkspaceMemorySnapshot(directory, sessionID); @@ -521,7 +616,7 @@ export const MemoryV2Plugin: Plugin = async (input) => { } try { - await promotePendingMemories(sessionID); + await promotePendingMemories(sessionID, { includeUnownedJournal: true }); } catch { // Keep pending memories in session/journal for retry on next event/session. } @@ -532,16 +627,20 @@ export const MemoryV2Plugin: Plugin = async (input) => { if (sessionID) { // Promote pending memories before deleting per-session state. // If promotion fails, leave session state and journal intact. + let promoted = false; try { - await promotePendingMemories(sessionID); + await promotePendingMemories(sessionID, { includeOwnedJournal: true, includeUnownedJournal: false }); + promoted = true; } catch { return; + } finally { + if (promoted) { + frozenWorkspaceMemoryCache.delete(sessionID); + processedUserMessages.delete(sessionID); + sessionParentCache.delete(sessionID); + } } - // Clean up caches - frozenWorkspaceMemoryCache.delete(sessionID); - processedUserMessages.delete(sessionID); - sessionParentCache.delete(sessionID); await rm(await sessionStatePath(directory, sessionID), { force: true }); } } diff --git a/src/promotion-accounting.ts b/src/promotion-accounting.ts index 7f65212..bf2438a 100644 --- a/src/promotion-accounting.ts +++ b/src/promotion-accounting.ts @@ -8,6 +8,7 @@ export type PendingPromotionAccounting = { absorbedKeys: Set; supersededKeys: Set; rejectedKeys: Set; + retryableRejectedKeys: Set; clearableKeys: Set; }; @@ -72,24 +73,38 @@ export function accountPendingPromotions(input: { rejectedKeys.add(key); } + const clearableKeys = new Set([ + ...promotedKeys, + ...absorbedKeys, + ...supersededKeys, + ...input.pending + .filter(memory => { + const terminal = terminalEventByKey.get(memoryKey(memory)); + return memory.source === "compaction" && ( + terminal?.reason === "rejected_capacity" || + terminal?.reason === "rejected_stale" + ); + }) + .map(memory => memoryKey(memory)), + ]); + + const retryableRejectedKeys = new Set( + input.pending + .filter(memory => { + const key = memoryKey(memory); + return rejectedKeys.has(key) && + !clearableKeys.has(key) && + (memory.source === "explicit" || memory.source === "manual"); + }) + .map(memory => memoryKey(memory)), + ); + return { promotedKeys, absorbedKeys, supersededKeys, rejectedKeys, - clearableKeys: new Set([ - ...promotedKeys, - ...absorbedKeys, - ...supersededKeys, - ...input.pending - .filter(memory => { - const terminal = terminalEventByKey.get(memoryKey(memory)); - return memory.source === "compaction" && ( - terminal?.reason === "rejected_capacity" || - terminal?.reason === "rejected_stale" - ); - }) - .map(memory => memoryKey(memory)), - ]), + retryableRejectedKeys, + clearableKeys, }; } diff --git a/src/storage.ts b/src/storage.ts index e75396b..b10af72 100644 --- a/src/storage.ts +++ b/src/storage.ts @@ -1,9 +1,12 @@ import { existsSync } from "fs"; import { randomUUID } from "crypto"; -import { mkdir, readFile, rename, writeFile } from "fs/promises"; +import { mkdir, open, readFile, rename, rm, stat, writeFile } from "fs/promises"; import { dirname } from "path"; const fileLocks = new Map>(); +const LOCK_WAIT_TIMEOUT_MS = 5000; +const LOCK_STALE_MS = 30_000; +const LOCK_ACQUIRE_GRACE_MS = 250; export async function readJSON(path: string, fallback: () => T): Promise { if (!existsSync(path)) return fallback(); @@ -14,6 +17,82 @@ export async function readJSON(path: string, fallback: () => T): Promise { } } +async function readJSONStrict(path: string, fallback: () => T): Promise { + if (!existsSync(path)) return fallback(); + try { + return JSON.parse(await readFile(path, "utf8")) as T; + } catch (error) { + throw new Error(`Invalid JSON in ${path}: ${(error as Error).message}`); + } +} + +async function isLockStale(lockPath: string, now = Date.now()): Promise { + try { + const content = await readFile(lockPath, "utf8"); + const [pidText, createdText] = content.split("\n"); + const pid = Number(pidText); + const createdAt = Number(createdText); + + if (!Number.isFinite(createdAt)) { + return !(await isRecentlyTouched(lockPath, now)); + } + if (now - createdAt > LOCK_STALE_MS) return true; + if (!Number.isInteger(pid) || pid <= 0) { + return !(await isRecentlyTouched(lockPath, now)); + } + + try { + process.kill(pid, 0); + return false; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ESRCH"; + } + } catch { + return true; + } +} + +async function isRecentlyTouched(path: string, now = Date.now()): Promise { + try { + return now - (await stat(path)).mtimeMs <= LOCK_ACQUIRE_GRACE_MS; + } catch { + return false; + } +} + +async function withFileLock(path: string, fn: () => Promise): Promise { + const lockPath = `${path}.lock`; + await mkdir(dirname(path), { recursive: true }); + const started = Date.now(); + + while (true) { + try { + const handle = await open(lockPath, "wx", 0o600); + try { + await handle.writeFile(`${process.pid}\n${Date.now()}\n`, "utf8"); + return await fn(); + } finally { + await handle.close(); + await rm(lockPath, { force: true }); + } + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code !== "EEXIST") throw error; + + if (await isLockStale(lockPath)) { + await rm(lockPath, { force: true }); + continue; + } + + if (Date.now() - started > LOCK_WAIT_TIMEOUT_MS) { + throw new Error(`Timed out waiting for lock ${lockPath}`); + } + + await new Promise(resolve => setTimeout(resolve, 25)); + } + } +} + export async function atomicWriteJSON(path: string, data: unknown): Promise { await mkdir(dirname(path), { recursive: true }); const tmp = `${path}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`; @@ -36,10 +115,12 @@ export async function updateJSON( try { await previous.catch(() => undefined); - const current = await readJSON(path, fallback); - const updated = await updater(current); - await atomicWriteJSON(path, updated); - return updated; + return await withFileLock(path, async () => { + const current = await readJSONStrict(path, fallback); + const updated = await updater(current); + await atomicWriteJSON(path, updated); + return updated; + }); } finally { release(); if (fileLocks.get(path) === queued) { diff --git a/src/types.ts b/src/types.ts index efa170f..0a0c5e1 100644 --- a/src/types.ts +++ b/src/types.ts @@ -15,6 +15,11 @@ export type LongTermMemoryEntry = { staleAfterDays?: number; supersedes?: string[]; tags?: string[]; + pendingOwnerSessionID?: string; + pendingMessageID?: string; + promotionAttempts?: number; + lastPromotionAttemptAt?: string; + lastPromotionFailureReason?: string; }; export type WorkspaceMemoryStore = { @@ -100,3 +105,15 @@ export const HOT_STATE_LIMITS = { maxPendingMemoriesStored: 12, maxPendingMemoriesRendered: 6, } as const; + +export const PROMOTION_RETRY_LIMITS = { + maxExplicitAttempts: 3, + maxManualAttempts: 3, +} as const; + +export const WORKSPACE_MEMORY_CACHE_LIMITS = { + maxFrozenSessions: 50, + maxProcessedSessionIDs: 200, + maxProcessedMessagesPerSession: 50, + frozenTtlMs: 60 * 60 * 1000, +} as const; diff --git a/src/workspace-memory.ts b/src/workspace-memory.ts index 84cdc95..037f093 100644 --- a/src/workspace-memory.ts +++ b/src/workspace-memory.ts @@ -17,6 +17,7 @@ const PIN_PREFIX = String.raw`(\bPIN\b(?:\s*(?:是|=|:|：)\s*|\s+(?![是=:：]) const PASSWORD_PREFIX = String.raw`((?:${PASSWORD_LABELS.source})(?:\s*(?:是|=|:|：)\s*|\s+(?![是=:：])))`; const USERNAME_PREFIX = String.raw`((?:${USERNAME_LABELS.source})(?:\s*(?:是|=|:|：)\s*|\s+(?![是=:：])))`; const SENSITIVE_PREFIX = String.raw`((?:${SENSITIVE_LABELS.source})(?:\s*(?:推|是|=|:|：)\s*|[:：]\s*))`; +const BEARER_PREFIX = String.raw`(Bearer\s+)`; export type MemoryConsolidationReason = | "promoted" @@ -79,30 +80,33 @@ export async function loadWorkspaceMemory(root: string): Promise ({ - ...store, - updatedAt: "", - entries: store.entries.map(entry => ({ - ...entry, - updatedAt: "", - })), - }); + const beforeById = new Map((before.entries ?? []).map(entry => [entry.id, entry])); + for (const afterEntry of after.entries ?? []) { + const beforeEntry = beforeById.get(afterEntry.id); + if (!beforeEntry) continue; + if (beforeEntry.text !== afterEntry.text) return true; + if ((beforeEntry.rationale ?? "") !== (afterEntry.rationale ?? "")) return true; + if (beforeEntry.status !== afterEntry.status) return true; + } - return JSON.stringify(sanitize(before)) !== JSON.stringify(sanitize(after)); + const beforeMigrations = JSON.stringify(before.migrations ?? []); + const afterMigrations = JSON.stringify(after.migrations ?? []); + return beforeMigrations !== afterMigrations; } export async function saveWorkspaceMemory(root: string, store: WorkspaceMemoryStore): Promise { @@ -234,6 +238,11 @@ export function redactCredentials(text: string): string { ); // 4. Standalone sensitive keys/tokens + result = result.replace( + new RegExp(String.raw`${BEARER_PREFIX}(?!token\s*[:=：])[\`'"]?(${SECRET_VALUE})`, "gi"), + "$1[REDACTED]", + ); + result = result.replace( new RegExp(String.raw`${SENSITIVE_PREFIX}[\`'"]?(${SECRET_VALUE})`, "gi"), "$1[REDACTED]", @@ -556,12 +565,14 @@ export function dedupeLongTermEntriesWithAccounting(entries: LongTermMemoryEntry } function compareLongTermMemoryForRetention(a: LongTermMemoryEntry, b: LongTermMemoryEntry): number { - const pA = priorityWithFreshness(a); - const pB = priorityWithFreshness(b); + const pA = priority(a); + const pB = priority(b); if (pB !== pA) return pB - pA; const sourceDiff = sourcePriority(b.source) - sourcePriority(a.source); if (sourceDiff !== 0) return sourceDiff; - return new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime(); + const createdDiff = new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime(); + if (createdDiff !== 0) return createdDiff; + return a.id.localeCompare(b.id); } function priority(entry: LongTermMemoryEntry): number { @@ -576,11 +587,6 @@ function priority(entry: LongTermMemoryEntry): number { return sourceWeight + typeWeight + entry.confidence * 10; } -/** Extended priority including freshness for tie-breaking */ -function priorityWithFreshness(entry: LongTermMemoryEntry): number { - return priority(entry); -} - function wouldFit( lines: string[], nextLine: string, diff --git a/tests/pending-journal.test.ts b/tests/pending-journal.test.ts index f62c9b4..782717e 100644 --- a/tests/pending-journal.test.ts +++ b/tests/pending-journal.test.ts @@ -6,16 +6,20 @@ import { describe, it, beforeEach, afterEach } from "node:test"; import assert from "node:assert"; -import { mkdir, rm } from "fs/promises"; +import { mkdir, mkdtemp as fsMkdtemp, rm } from "fs/promises"; import { tmpdir } from "os"; import { join } from "path"; import { loadPendingJournal, savePendingJournal, appendPendingMemories, + clearPendingMemories, + recordPromotionRejections, + memoryKey, PENDING_JOURNAL_LIMITS, } from "../src/pending-journal.ts"; import type { LongTermMemoryEntry } from "../src/types.ts"; +import { PROMOTION_RETRY_LIMITS } from "../src/types.ts"; describe("pending journal retention", () => { let testDir: string; @@ -193,22 +197,38 @@ describe("pending journal retention", () => { ); }); - it("savePendingJournal prunes stale entries regardless of source", async () => { + it("retains old explicit and manual pending entries while under cap", async () => { const now = new Date(); const staleDate = new Date(now.getTime() - 35 * 24 * 60 * 60 * 1000); const entries: LongTermMemoryEntry[] = [ { - type: "decision", - text: "Stale explicit entry", + id: "explicit_old", + type: "feedback", + text: "Old explicit preference", source: "explicit", + confidence: 1, + status: "active", createdAt: staleDate.toISOString(), updatedAt: staleDate.toISOString(), }, { - type: "decision", - text: "Stale compaction entry", + id: "manual_old", + type: "reference", + text: "Old manual reference", + source: "manual", + confidence: 1, + status: "active", + createdAt: staleDate.toISOString(), + updatedAt: staleDate.toISOString(), + }, + { + id: "compaction_old", + type: "reference", + text: "Old compaction reference", source: "compaction", + confidence: 0.75, + status: "active", createdAt: staleDate.toISOString(), updatedAt: staleDate.toISOString(), }, @@ -223,13 +243,173 @@ describe("pending journal retention", () => { const loaded = await loadPendingJournal(testDir); - // Both explicit and compaction entries past maxAgeDays are pruned - // Retention does not differentiate by source - assert.strictEqual( - loaded.entries.length, - 0, - "Stale entries should be pruned regardless of source" + assert.deepEqual(loaded.entries.map(entry => entry.id), ["explicit_old", "manual_old"]); + }); + + it("clears only entries matching both key and owner when owner is supplied", async () => { + const now = new Date().toISOString(); + await appendPendingMemories(testDir, [ + { + id: "a", + type: "feedback", + text: "Session A preference", + source: "explicit", + confidence: 1, + status: "active", + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "session-a", + }, + { + id: "b", + type: "feedback", + text: "Session B preference", + source: "explicit", + confidence: 1, + status: "active", + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "session-b", + }, + ]); + + await clearPendingMemories( + testDir, + new Set(["feedback:session a preference", "feedback:session b preference"]), + { ownerSessionID: "session-a" }, ); + + const loaded = await loadPendingJournal(testDir); + assert.deepEqual(loaded.entries.map(entry => entry.pendingOwnerSessionID), ["session-b"]); + }); + + it("retains same-key pending entries owned by different sessions", async () => { + const now = new Date().toISOString(); + await appendPendingMemories(testDir, [ + { + id: "same-a", + type: "feedback", + text: "Prefer owner-scoped promotion.", + source: "explicit", + confidence: 1, + status: "active", + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "session-a", + }, + { + id: "same-b", + type: "feedback", + text: "Prefer owner-scoped promotion.", + source: "explicit", + confidence: 1, + status: "active", + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "session-b", + }, + ]); + + const loaded = await loadPendingJournal(testDir); + + assert.deepEqual( + loaded.entries.map(entry => entry.pendingOwnerSessionID).sort(), + ["session-a", "session-b"], + "same memory key must remain separately retryable/clearable per owner", + ); + }); + + it("records bounded promotion rejection attempts and exhausts only matching owner", async () => { + const now = new Date().toISOString(); + const sessionA: LongTermMemoryEntry = { + id: "reject-a", + type: "reference", + text: "Capacity rejected explicit reference.", + source: "explicit", + confidence: 0.1, + status: "active", + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "session-a", + }; + const sessionB: LongTermMemoryEntry = { + ...sessionA, + id: "reject-b", + pendingOwnerSessionID: "session-b", + }; + await appendPendingMemories(testDir, [sessionA, sessionB]); + + for (let attempt = 1; attempt < PROMOTION_RETRY_LIMITS.maxExplicitAttempts; attempt += 1) { + const exhausted = await recordPromotionRejections( + testDir, + new Set([memoryKey(sessionA)]), + "rejected_capacity", + { ownerSessionID: "session-a" }, + ); + + assert.equal(exhausted.size, 0, "entry should not exhaust before the max attempt"); + const loaded = await loadPendingJournal(testDir); + const ownedA = loaded.entries.find(entry => entry.pendingOwnerSessionID === "session-a"); + const ownedB = loaded.entries.find(entry => entry.pendingOwnerSessionID === "session-b"); + assert.equal(ownedA?.promotionAttempts, attempt); + assert.equal(ownedA?.lastPromotionFailureReason, "rejected_capacity"); + assert.equal(ownedB?.promotionAttempts, undefined, + "same-key entry for another owner must not be mutated"); + } + + const exhausted = await recordPromotionRejections( + testDir, + new Set([memoryKey(sessionA)]), + "rejected_capacity", + { ownerSessionID: "session-a" }, + ); + + assert.deepEqual([...exhausted], [memoryKey(sessionA)]); + const loaded = await loadPendingJournal(testDir); + assert.deepEqual(loaded.entries.map(entry => entry.pendingOwnerSessionID), ["session-b"]); + }); + + it("drops invalid timestamp entries for every source as corruption safety", async () => { + await savePendingJournal(testDir, { + version: 1, + workspace: { root: testDir, key: "test" }, + updatedAt: new Date().toISOString(), + entries: [ + { + id: "bad_explicit", + type: "feedback", + text: "Bad explicit timestamp", + source: "explicit", + confidence: 1, + status: "active", + createdAt: "not-a-date", + updatedAt: "also-bad", + }, + { + id: "bad_manual", + type: "reference", + text: "Bad manual timestamp", + source: "manual", + confidence: 1, + status: "active", + createdAt: "", + updatedAt: "", + }, + { + id: "bad_compaction", + type: "reference", + text: "Bad compaction timestamp", + source: "compaction", + confidence: 0.75, + status: "active", + createdAt: "bad", + updatedAt: "bad", + }, + ], + }); + + const loaded = await loadPendingJournal(testDir); + assert.equal(loaded.entries.length, 0); }); it("savePendingJournal uses updatedAt when createdAt is missing", async () => { @@ -275,5 +455,5 @@ describe("pending journal retention", () => { async function mkdtemp(): Promise { const base = join(tmpdir(), "pending-journal-test"); await mkdir(base, { recursive: true }); - return base; -} \ No newline at end of file + return fsMkdtemp(join(base, "case-")); +} diff --git a/tests/plugin.test.ts b/tests/plugin.test.ts index 62574a4..7611aae 100644 --- a/tests/plugin.test.ts +++ b/tests/plugin.test.ts @@ -7,8 +7,9 @@ import { MemoryV2Plugin } from "../src/plugin.ts"; import { loadSessionState, saveSessionState } from "../src/session-state.ts"; import { parseWorkspaceMemoryCandidates } from "../src/extractors.ts"; import type { OpenError } from "../src/types.ts"; +import { PROMOTION_RETRY_LIMITS, WORKSPACE_MEMORY_CACHE_LIMITS } from "../src/types.ts"; import { workspaceMemoryPath, workspacePendingJournalPath } from "../src/paths.ts"; -import { loadPendingJournal, savePendingJournal } from "../src/pending-journal.ts"; +import { loadPendingJournal, savePendingJournal, memoryKey } from "../src/pending-journal.ts"; import { loadWorkspaceMemory, updateWorkspaceMemory } from "../src/workspace-memory.ts"; // Mock client for root session (not a sub-agent) @@ -428,7 +429,7 @@ test("chat system transform reuses frozen rendered workspace snapshot", async () } }); -test("no compaction: explicit memory is promoted on next session start from durable journal", async () => { +test("no compaction: owned explicit memory is not promoted by unrelated next session start", async () => { const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); try { @@ -449,7 +450,113 @@ test("no compaction: explicit memory is promoted on next session start from dura ); const workspacePrompt = output.system.find((part: string) => part.startsWith("Workspace memory")); - assert.match(workspacePrompt ?? "", /Prefer boring cache boundaries/); + assert.doesNotMatch(workspacePrompt ?? "", /Prefer boring cache boundaries/); + + const pending = await loadPendingJournal(tmpDir); + assert.equal(pending.entries.length, 1); + assert.equal(pending.entries[0].pendingOwnerSessionID, "session-without-compaction"); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + +test("explicit memory appended from user message is owned by session and not promoted before current snapshot", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const plugin = await MemoryV2Plugin({ + directory: tmpDir, + client: mockClientWithLatestUser("remember this: Prefer Traditional Chinese.", "msg-a"), + }); + const output = { system: ["base header"] }; + + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "session-a", model: {} }, + output, + ); + + const pending = await loadPendingJournal(tmpDir); + assert.equal(pending.entries.length, 1); + assert.equal(pending.entries[0].pendingOwnerSessionID, "session-a"); + assert.equal(pending.entries[0].pendingMessageID, "msg-a"); + + const workspace = await loadWorkspaceMemory(tmpDir); + assert.equal( + workspace.entries.some(entry => /Prefer Traditional Chinese/.test(entry.text)), + false, + "current-turn explicit memory should remain pending until compaction/promotion", + ); + assert.match(output.system.join("\n"), /Prefer Traditional Chinese/, + "current-turn explicit memory should still appear in hot session state"); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + +test("session promotion does not clear another session's same-key pending journal entry", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const now = new Date().toISOString(); + const pendingA = { + id: "mem_same_key_a", + type: "feedback" as const, + text: "Prefer owner-scoped pending cleanup.", + source: "explicit" as const, + confidence: 1, + status: "active" as const, + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "session-a", + pendingMessageID: "msg-a", + }; + const pendingB = { + ...pendingA, + id: "mem_same_key_b", + pendingOwnerSessionID: "session-b", + pendingMessageID: "msg-b", + }; + + await saveSessionState(tmpDir, { + version: 1, + sessionID: "session-a", + turn: 0, + updatedAt: now, + activeFiles: [], + openErrors: [], + recentDecisions: [], + pendingMemories: [pendingA], + }); + await saveSessionState(tmpDir, { + version: 1, + sessionID: "session-b", + turn: 0, + updatedAt: now, + activeFiles: [], + openErrors: [], + recentDecisions: [], + pendingMemories: [pendingB], + }); + await savePendingJournal(tmpDir, { + version: 1, + workspace: { root: tmpDir, key: "test" }, + updatedAt: now, + entries: [pendingA, pendingB], + }); + + const plugin = await MemoryV2Plugin({ directory: tmpDir, client: mockRootClient() }); + await (plugin as Record)["event"]({ + event: { type: "session.compacted", properties: { sessionID: "session-a" } }, + }); + + const journal = await loadPendingJournal(tmpDir); + assert.equal(journal.entries.length, 1); + assert.equal(journal.entries[0].pendingOwnerSessionID, "session-b", + "session-a promotion must not clear session-b's same-key journal entry"); + + const stateB = await loadSessionState(tmpDir, "session-b"); + assert.equal(stateB.pendingMemories.length, 1); + assert.equal(memoryKey(stateB.pendingMemories[0]), memoryKey(pendingB)); } finally { await rm(tmpDir, { recursive: true, force: true }); } @@ -488,6 +595,76 @@ test("session.deleted promotes pending memories before deleting session state", } }); +test("session.deleted clears caches even when session state file is already gone", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const now = new Date().toISOString(); + await updateWorkspaceMemory(tmpDir, store => { + store.entries.push({ + id: "mem_before_delete_cache", + type: "project", + text: "Workspace memory before delete cleanup.", + source: "compaction", + confidence: 0.9, + status: "active", + createdAt: now, + updatedAt: now, + }); + return store; + }); + + const plugin = await MemoryV2Plugin({ directory: tmpDir, client: mockRootClient() }); + const beforeOutput = { system: ["base header"] }; + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "deleted-missing-state-session", model: {} }, + beforeOutput, + ); + assert.match(beforeOutput.system.join("\n"), /Workspace memory before delete cleanup/); + + const ownedPending = { + id: "mem_delete_owned_journal", + type: "decision" as const, + text: "Owned journal memory promotes during delete cleanup.", + source: "explicit" as const, + confidence: 1, + status: "active" as const, + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "deleted-missing-state-session", + pendingMessageID: "msg-delete-owned", + }; + await savePendingJournal(tmpDir, { + version: 1, + workspace: { root: tmpDir, key: "test" }, + updatedAt: now, + entries: [ownedPending], + }); + + await (plugin as Record)["event"]({ + event: { + type: "session.deleted", + properties: { sessionID: "deleted-missing-state-session" }, + }, + }); + + const pendingAfter = await loadPendingJournal(tmpDir); + assert.equal(pendingAfter.entries.length, 0, + "clearable owned journal entry should be removed even when session state file is absent"); + + const afterOutput = { system: ["base header"] }; + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "deleted-missing-state-session", model: {} }, + afterOutput, + ); + const workspacePrompt = afterOutput.system.find((part: string) => part.startsWith("Workspace memory")); + assert.match(workspacePrompt ?? "", /Owned journal memory promotes during delete cleanup/, + "session.deleted should clear frozen cache after successful promotion"); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + test("duplicate explicit memories dedupe by normalized type and text, not generated id", async () => { const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); @@ -650,18 +827,26 @@ Continue durable memory work. } }); -test("integration: next session promotes prior explicit journal and leaves journal clean", async () => { +test("integration: next session promotes prior unowned journal and leaves journal clean", async () => { const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); try { - const firstPlugin = await MemoryV2Plugin({ - directory: tmpDir, - client: mockClientWithLatestUser("remember this: Cross-session promotion keeps the journal clean.", "msg-cross-session"), + const now = new Date().toISOString(); + await savePendingJournal(tmpDir, { + version: 1, + workspace: { root: tmpDir, key: "test" }, + updatedAt: now, + entries: [{ + id: "mem_unowned_cross_session", + type: "feedback", + text: "Cross-session unowned promotion keeps the journal clean.", + source: "explicit", + confidence: 1, + status: "active", + createdAt: now, + updatedAt: now, + }], }); - await (firstPlugin as Record)["experimental.chat.system.transform"]( - { sessionID: "first-cross-session", model: {} }, - { system: ["base header"] }, - ); assert.equal((await loadPendingJournal(tmpDir)).entries.length, 1); @@ -673,7 +858,7 @@ test("integration: next session promotes prior explicit journal and leaves journ ); assert.equal((await loadPendingJournal(tmpDir)).entries.length, 0); - assert.match(output.system.join("\n"), /Cross-session promotion keeps the journal clean/); + assert.match(output.system.join("\n"), /Cross-session unowned promotion keeps the journal clean/); } finally { await rm(tmpDir, { recursive: true, force: true }); } @@ -751,6 +936,209 @@ test("same-session explicit memory does not mutate frozen system[1]", async () = } }); +test("chat system transform reloads frozen workspace snapshot after cache TTL expires", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + const originalNow = Date.now; + let now = originalNow(); + Date.now = () => now; + + try { + const timestamp = new Date().toISOString(); + await updateWorkspaceMemory(tmpDir, store => { + store.entries.push({ + id: "mem_cache_ttl_old", + type: "project", + text: "Workspace memory before TTL expiry.", + source: "compaction", + confidence: 0.9, + status: "active", + createdAt: timestamp, + updatedAt: timestamp, + }); + return store; + }); + + const plugin = await MemoryV2Plugin({ directory: tmpDir, client: mockRootClient() }); + const output1 = { system: ["base header"] }; + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "ttl-session", model: {} }, + output1, + ); + assert.match(output1.system.join("\n"), /Workspace memory before TTL expiry/); + + await updateWorkspaceMemory(tmpDir, store => { + store.entries.push({ + id: "mem_cache_ttl_new", + type: "project", + text: "Workspace memory after TTL expiry.", + source: "compaction", + confidence: 0.9, + status: "active", + createdAt: timestamp, + updatedAt: timestamp, + }); + return store; + }); + + now += WORKSPACE_MEMORY_CACHE_LIMITS.frozenTtlMs + 1; + + const output2 = { system: ["base header"] }; + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "ttl-session", model: {} }, + output2, + ); + + assert.match(output2.system.join("\n"), /Workspace memory after TTL expiry/); + } finally { + Date.now = originalNow; + await rm(tmpDir, { recursive: true, force: true }); + } +}); + +test("chat system transform evicts oldest frozen snapshots when cache exceeds session limit", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const timestamp = new Date().toISOString(); + await updateWorkspaceMemory(tmpDir, store => { + store.entries.push({ + id: "mem_cache_size_old", + type: "project", + text: "Workspace memory before cache pressure.", + source: "compaction", + confidence: 0.9, + status: "active", + createdAt: timestamp, + updatedAt: timestamp, + }); + return store; + }); + + const plugin = await MemoryV2Plugin({ directory: tmpDir, client: mockRootClient() }); + for (let i = 0; i <= WORKSPACE_MEMORY_CACHE_LIMITS.maxFrozenSessions; i += 1) { + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: `cache-size-session-${i}`, model: {} }, + { system: ["base header"] }, + ); + } + + await updateWorkspaceMemory(tmpDir, store => { + store.entries.push({ + id: "mem_cache_size_new", + type: "project", + text: "Workspace memory after cache pressure.", + source: "compaction", + confidence: 0.9, + status: "active", + createdAt: timestamp, + updatedAt: timestamp, + }); + return store; + }); + + const output = { system: ["base header"] }; + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "cache-size-session-0", model: {} }, + output, + ); + + assert.match(output.system.join("\n"), /Workspace memory after cache pressure/); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + +test("processed user message cache keeps only the latest message IDs per session", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + let latestMessages: Array> = []; + const client = { + session: { + get: async () => ({ data: { parentID: null } }), + messages: async () => ({ data: latestMessages }), + todo: async () => ({ data: [] }), + }, + }; + const plugin = await MemoryV2Plugin({ directory: tmpDir, client }); + + for (let i = 0; i <= WORKSPACE_MEMORY_CACHE_LIMITS.maxProcessedMessagesPerSession; i += 1) { + latestMessages = [{ + info: { role: "user", id: `msg-${i}` }, + parts: [{ type: "text", text: `remember this: Processed cache filler memory ${i}.` }], + }]; + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "processed-cache-session", model: {} }, + { system: ["base header"] }, + ); + } + + latestMessages = [{ + info: { role: "user", id: "msg-0" }, + parts: [{ type: "text", text: "remember this: Evicted processed message id can be reused." }], + }]; + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "processed-cache-session", model: {} }, + { system: ["base header"] }, + ); + + const state = await loadSessionState(tmpDir, "processed-cache-session"); + assert.ok( + state.pendingMemories.some(memory => /Evicted processed message id can be reused/.test(memory.text)), + "oldest processed message id should be evicted and accepted again", + ); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + +test("processed user message cache evicts oldest session ID sets", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const latestBySession = new Map>>(); + const client = { + session: { + get: async () => ({ data: { parentID: null } }), + messages: async ({ path }: { path?: { id?: string } } = {}) => ({ + data: latestBySession.get(path?.id ?? "") ?? [], + }), + todo: async () => ({ data: [] }), + }, + }; + const plugin = await MemoryV2Plugin({ directory: tmpDir, client }); + + for (let i = 0; i <= WORKSPACE_MEMORY_CACHE_LIMITS.maxProcessedSessionIDs; i += 1) { + const sessionID = `processed-session-${i}`; + latestBySession.set(sessionID, [{ + info: { role: "user", id: `msg-${i}` }, + parts: [{ type: "text", text: `remember this: Session cache filler memory ${i}.` }], + }]); + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID, model: {} }, + { system: ["base header"] }, + ); + } + + latestBySession.set("processed-session-0", [{ + info: { role: "user", id: "msg-0" }, + parts: [{ type: "text", text: "remember this: Evicted processed session set can process again." }], + }]); + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "processed-session-0", model: {} }, + { system: ["base header"] }, + ); + + const state = await loadSessionState(tmpDir, "processed-session-0"); + assert.ok( + state.pendingMemories.some(memory => /Evicted processed session set can process again/.test(memory.text)), + "oldest processed session set should be evicted and accept the same message id again", + ); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + test("compaction intentionally refreshes frozen system[1] with promoted memories", async () => { const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); @@ -1116,6 +1504,89 @@ test("session.compacted keeps explicit pending memory rejected by workspace entr } }); +test("explicit capacity rejection records bounded retry metadata", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const now = new Date().toISOString(); + await updateWorkspaceMemory(tmpDir, store => { + for (let i = 0; i < 28; i += 1) { + store.entries.push({ + id: `mem_high_bounded_reject_${i}`, + type: "feedback", + text: `Pinned high priority feedback for bounded rejection ${i}.`, + source: "explicit", + confidence: 1, + status: "active", + createdAt: now, + updatedAt: now, + }); + } + return store; + }); + + const rejectedMemory = { + id: "mem_explicit_bounded_reject", + type: "reference" as const, + text: "Explicit reference should retry only a bounded number of times.", + source: "explicit" as const, + confidence: 0.1, + status: "active" as const, + createdAt: now, + updatedAt: now, + pendingOwnerSessionID: "bounded-reject-session", + pendingMessageID: "msg-bounded-reject", + }; + await saveSessionState(tmpDir, { + version: 1, + sessionID: "bounded-reject-session", + turn: 0, + updatedAt: now, + activeFiles: [], + openErrors: [], + recentDecisions: [], + pendingMemories: [rejectedMemory], + }); + await savePendingJournal(tmpDir, { + version: 1, + workspace: { root: tmpDir, key: "test" }, + updatedAt: now, + entries: [rejectedMemory], + }); + + const plugin = await MemoryV2Plugin({ directory: tmpDir, client: mockRootClient() }); + for (let attempt = 1; attempt < PROMOTION_RETRY_LIMITS.maxExplicitAttempts; attempt += 1) { + await (plugin as Record)["event"]({ + event: { type: "session.compacted", properties: { sessionID: "bounded-reject-session" } }, + }); + + const journal = await loadPendingJournal(tmpDir); + assert.equal(journal.entries.length, 1, + "explicit rejection should not silently clear before retry exhaustion"); + assert.equal(journal.entries[0].promotionAttempts, attempt); + assert.equal(journal.entries[0].lastPromotionFailureReason, "rejected_capacity"); + + const state = await loadSessionState(tmpDir, "bounded-reject-session"); + assert.equal(state.pendingMemories.length, 1, + "hot session state should keep retryable explicit memory visible before exhaustion"); + } + + await (plugin as Record)["event"]({ + event: { type: "session.compacted", properties: { sessionID: "bounded-reject-session" } }, + }); + + const journal = await loadPendingJournal(tmpDir); + assert.equal(journal.entries.length, 0, + "explicit pending journal entry should clear after max retry attempts"); + + const state = await loadSessionState(tmpDir, "bounded-reject-session"); + assert.equal(state.pendingMemories.length, 0, + "explicit hot session state should clear after retry exhaustion"); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + test("session.compacted clears compaction pending memories when all rejected by workspace cap", async () => { const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); diff --git a/tests/promotion-accounting.test.ts b/tests/promotion-accounting.test.ts index cc493b8..930cc77 100644 --- a/tests/promotion-accounting.test.ts +++ b/tests/promotion-accounting.test.ts @@ -209,6 +209,25 @@ test("accountPendingPromotions keeps explicit capacity rejection pending", () => assert.deepEqual([...result.rejectedKeys], [memoryKey(pending[0])]); assert.equal(result.clearableKeys.size, 0); + assert.deepEqual([...result.retryableRejectedKeys], [memoryKey(pending[0])]); +}); + +test("accountPendingPromotions marks manual capacity rejection as retryable", () => { + const pending = [mem("pending_manual_capacity", "Manual reference should retry if capacity rejected.", { + type: "reference", + source: "manual", + })]; + + const result = accountPendingPromotions({ + pending, + before: [], + after: [], + events: [event(pending[0], "rejected_capacity")], + }); + + assert.deepEqual([...result.rejectedKeys], [memoryKey(pending[0])]); + assert.equal(result.clearableKeys.size, 0); + assert.deepEqual([...result.retryableRejectedKeys], [memoryKey(pending[0])]); }); test("accountPendingPromotions clears compaction stale rejection from accounting", () => { diff --git a/tests/storage.test.ts b/tests/storage.test.ts new file mode 100644 index 0000000..9d92018 --- /dev/null +++ b/tests/storage.test.ts @@ -0,0 +1,83 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { existsSync } from "node:fs"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { spawn } from "node:child_process"; +import { updateJSON } from "../src/storage.ts"; + +test("updateJSON serializes concurrent increments", async () => { + const root = await mkdtemp(join(tmpdir(), "wm-storage-")); + try { + const path = join(root, "counter.json"); + await Promise.all(Array.from({ length: 25 }, () => + updateJSON(path, () => ({ count: 0 }), current => ({ count: current.count + 1 })), + )); + + const final = await updateJSON(path, () => ({ count: 0 }), current => current); + assert.equal(final.count, 25); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test("updateJSON does not replace corrupt JSON with fallback", async () => { + const root = await mkdtemp(join(tmpdir(), "wm-storage-corrupt-")); + try { + const path = join(root, "bad.json"); + await writeFile(path, "{not json", "utf8"); + + await assert.rejects( + updateJSON(path, () => ({ ok: true }), current => current), + /Invalid JSON/, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test("updateJSON recovers stale lock files left by crashed process", async () => { + const root = await mkdtemp(join(tmpdir(), "wm-storage-stale-lock-")); + try { + const path = join(root, "locked.json"); + const lockPath = `${path}.lock`; + await writeFile(lockPath, `999999\n0\n`, "utf8"); + + const updated = await updateJSON(path, () => ({ count: 0 }), current => ({ count: current.count + 1 })); + assert.equal(updated.count, 1); + assert.equal(existsSync(lockPath), false, "stale lock file should be removed after update"); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test("updateJSON serializes writes across separate node processes", async () => { + const root = await mkdtemp(join(tmpdir(), "wm-storage-xproc-")); + try { + const path = join(root, "counter.json"); + const worker = ` + import { updateJSON } from ${JSON.stringify(new URL("../src/storage.ts", import.meta.url).href)}; + const path = process.argv[1]; + await Promise.all(Array.from({ length: 20 }, () => updateJSON(path, () => ({ count: 0 }), async current => { + await new Promise(resolve => setTimeout(resolve, Math.floor(Math.random() * 5))); + return { count: current.count + 1 }; + }))); + `; + + await Promise.all(Array.from({ length: 5 }, () => new Promise((resolve, reject) => { + const child = spawn( + process.execPath, + ["--experimental-strip-types", "--input-type=module", "-e", worker, path], + { stdio: "inherit" }, + ); + child.on("exit", code => code === 0 ? resolve() : reject(new Error(`child exited ${code}`))); + child.on("error", reject); + }))); + + const final = await updateJSON(path, () => ({ count: 0 }), current => current); + assert.equal(final.count, 100); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/workspace-memory.test.ts b/tests/workspace-memory.test.ts index 4cda584..615f2cc 100644 --- a/tests/workspace-memory.test.ts +++ b/tests/workspace-memory.test.ts @@ -1,6 +1,6 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; import { join, dirname } from "node:path"; import { tmpdir } from "node:os"; import type { LongTermMemoryEntry, WorkspaceMemoryStore } from "../src/types.ts"; @@ -36,6 +36,10 @@ function entry(id: string, text: string, type: LongTermMemoryEntry["type"] = "de }; } +function sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); +} + /** Create an entry with a createdAt offset from now (negative = in the past) */ function agedEntry( id: string, @@ -249,6 +253,26 @@ test("enforceLongTermLimits respects maxEntries limit", () => { assert.ok(kept.length <= 28, `Should respect maxEntries. Got: ${kept.length}`); }); +test("normalization ordering is deterministic for retention ties", () => { + const createdAt = "2026-04-28T00:00:00.000Z"; + const a = { + ...entry("a", "Durable unique memory A"), + createdAt, + updatedAt: createdAt, + }; + const b = { + ...entry("b", "Durable unique memory B"), + createdAt, + updatedAt: createdAt, + }; + + const first = enforceLongTermLimits([b, a]).map(memory => memory.id); + const second = enforceLongTermLimits([a, b]).map(memory => memory.id); + + assert.deepEqual(first, ["a", "b"]); + assert.deepEqual(second, ["a", "b"]); +}); + test("dedupeLongTermEntriesWithAccounting reports exact duplicates as absorbed", () => { const now = new Date().toISOString(); const lower: LongTermMemoryEntry = { @@ -796,6 +820,21 @@ test("redactCredentials handles generic API keys and tokens", () => { assert.equal(redactCredentials("auth: abc123def"), "auth: [REDACTED]"); }); +test("redactCredentials redacts bearer tokens", () => { + assert.equal( + redactCredentials("Bearer sk-test-123"), + "Bearer [REDACTED]", + ); + assert.equal( + redactCredentials("Authorization: Bearer sk-test-123"), + "Authorization: Bearer [REDACTED]", + ); + assert.equal( + redactCredentials("curl -H 'Authorization: Bearer ghp_secret123'"), + "curl -H 'Authorization: Bearer [REDACTED]'", + ); +}); + test("redactCredentials does not redact benign security-related wording", () => { assert.equal(redactCredentials("token budget is 5200 characters"), "token budget is 5200 characters"); assert.equal(redactCredentials("auth config uses OAuth"), "auth config uses OAuth"); @@ -951,6 +990,178 @@ test("renderWorkspaceMemory excludes superseded entries", () => { assert.doesNotMatch(rendered, /Waves 1-5 已完成/); }); +test("loadWorkspaceMemory does not rewrite an already normalized store", async () => { + const sandbox = await mkdtemp(join(tmpdir(), "wm-normalized-")); + const dataHome = join(sandbox, "xdg-data-home"); + const root = join(sandbox, "workspace"); + const previousXdgDataHome = process.env.XDG_DATA_HOME; + process.env.XDG_DATA_HOME = dataHome; + + try { + await mkdir(root, { recursive: true }); + const now = "2026-04-28T00:00:00.000Z"; + await saveWorkspaceMemory(root, { + version: 1, + workspace: { root, key: "test" }, + limits: { + maxRenderedChars: LONG_TERM_LIMITS.maxRenderedChars, + maxEntries: LONG_TERM_LIMITS.maxEntries, + }, + entries: [ + { + ...entry("normalized-feedback", "Normalized feedback memory", "feedback"), + source: "explicit", + confidence: 1, + createdAt: now, + updatedAt: now, + }, + ], + migrations: [], + updatedAt: now, + }); + + const storePath = await workspaceMemoryPath(root); + const before = (await stat(storePath)).mtimeMs; + await sleep(20); + + await loadWorkspaceMemory(root); + await loadWorkspaceMemory(root); + + const after = (await stat(storePath)).mtimeMs; + assert.equal(after, before, "normalized loads should not touch the store file"); + } finally { + if (previousXdgDataHome === undefined) { + delete process.env.XDG_DATA_HOME; + } else { + process.env.XDG_DATA_HOME = previousXdgDataHome; + } + await rm(sandbox, { recursive: true, force: true }); + } +}); + +test("loadWorkspaceMemory does not persist pure ordering normalization", async () => { + const sandbox = await mkdtemp(join(tmpdir(), "wm-ordering-")); + const dataHome = join(sandbox, "xdg-data-home"); + const root = join(sandbox, "workspace"); + const previousXdgDataHome = process.env.XDG_DATA_HOME; + process.env.XDG_DATA_HOME = dataHome; + + try { + await mkdir(root, { recursive: true }); + const now = "2026-04-28T00:00:00.000Z"; + await saveWorkspaceMemory(root, { + version: 1, + workspace: { root, key: "test" }, + limits: { + maxRenderedChars: LONG_TERM_LIMITS.maxRenderedChars, + maxEntries: LONG_TERM_LIMITS.maxEntries, + }, + entries: [ + { + ...entry("feedback-first", "High priority feedback memory", "feedback"), + source: "explicit", + confidence: 1, + createdAt: now, + updatedAt: now, + }, + { + ...entry("reference-second", "Lower priority reference memory", "reference"), + source: "compaction", + confidence: 0.75, + createdAt: now, + updatedAt: now, + }, + ], + migrations: [], + updatedAt: now, + }); + + const storePath = await workspaceMemoryPath(root); + const canonical = JSON.parse(await readFile(storePath, "utf-8")) as WorkspaceMemoryStore; + await writeFile( + storePath, + JSON.stringify({ ...canonical, entries: [...canonical.entries].reverse() }, null, 2), + "utf-8", + ); + + const before = (await stat(storePath)).mtimeMs; + await sleep(20); + const loaded = await loadWorkspaceMemory(root); + const after = (await stat(storePath)).mtimeMs; + + assert.deepEqual(loaded.entries.map(memory => memory.id), ["feedback-first", "reference-second"]); + assert.equal(after, before, "order-only normalization should not write during load"); + } finally { + if (previousXdgDataHome === undefined) { + delete process.env.XDG_DATA_HOME; + } else { + process.env.XDG_DATA_HOME = previousXdgDataHome; + } + await rm(sandbox, { recursive: true, force: true }); + } +}); + +test("loadWorkspaceMemory persists redaction changes and is stable afterward", async () => { + const sandbox = await mkdtemp(join(tmpdir(), "wm-redact-stable-")); + const dataHome = join(sandbox, "xdg-data-home"); + const root = join(sandbox, "workspace"); + const previousXdgDataHome = process.env.XDG_DATA_HOME; + process.env.XDG_DATA_HOME = dataHome; + + try { + await mkdir(root, { recursive: true }); + const now = "2026-04-28T00:00:00.000Z"; + const unredactedStore: WorkspaceMemoryStore = { + version: 1, + workspace: { root, key: "test" }, + limits: { + maxRenderedChars: LONG_TERM_LIMITS.maxRenderedChars, + maxEntries: LONG_TERM_LIMITS.maxEntries, + }, + entries: [ + { + id: "bearer-secret", + text: "Authorization: Bearer sk-test-123", + rationale: "password: sushi", + type: "reference", + source: "manual", + confidence: 0.9, + status: "active", + createdAt: now, + updatedAt: now, + }, + ], + migrations: ["2026-04-26-p0-cleanup"], + updatedAt: now, + }; + + const storePath = await workspaceMemoryPath(root); + await mkdir(dirname(storePath), { recursive: true }); + await writeFile(storePath, JSON.stringify(unredactedStore, null, 2), "utf-8"); + + const loaded = await loadWorkspaceMemory(root); + assert.equal(loaded.entries[0].text, "Authorization: Bearer [REDACTED]"); + assert.equal(loaded.entries[0].rationale, "password: [REDACTED]"); + + const persistedAfterFirstLoad = await readFile(storePath, "utf-8"); + assert.equal(persistedAfterFirstLoad.includes("sk-test-123"), false); + assert.equal(persistedAfterFirstLoad.includes("sushi"), false); + + const beforeSecondLoad = (await stat(storePath)).mtimeMs; + await sleep(20); + await loadWorkspaceMemory(root); + const afterSecondLoad = (await stat(storePath)).mtimeMs; + assert.equal(afterSecondLoad, beforeSecondLoad, "second load should not rewrite redacted content"); + } finally { + if (previousXdgDataHome === undefined) { + delete process.env.XDG_DATA_HOME; + } else { + process.env.XDG_DATA_HOME = previousXdgDataHome; + } + await rm(sandbox, { recursive: true, force: true }); + } +}); + test("loadWorkspaceMemory normalizes and persists credentials from legacy unredacted store", async () => { const sandbox = await mkdtemp(join(tmpdir(), "wm-redact-")); const dataHome = join(sandbox, "xdg-data-home");