diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index c38a71f..a2e0f60 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -96,7 +96,7 @@ Retention monitoring: ### Validation - `npm run typecheck` -- `npm test` — 237 tests passing +- `npm test` — 242 tests passing - `bun scripts/memory-diag.ts health` --- diff --git a/docs/architecture.md b/docs/architecture.md index 0363c8a..52d73fe 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -343,6 +343,10 @@ canonical("Use npm cache for plugins") === "use npm cache for plugins" const workspaceKey = sha256(realpath(workspaceRoot)).slice(0, 16) ``` +### Storage Safety + +All read-modify-write updates go through `updateJSON()`, which combines an in-process promise queue with an on-disk `.lock` file. The lock file uses exclusive creation, heartbeat refreshes while held, stale-lock recovery after 30 seconds, and a 5 second wait timeout for live contention. Direct `readJSON()` is fallback-oriented and does not mutate data except when corrupt JSON is quarantined. + ## Performance Considerations ### Memory Budgets diff --git a/src/plugin.ts b/src/plugin.ts index 7e1137e..468e5cb 100644 --- a/src/plugin.ts +++ b/src/plugin.ts @@ -63,7 +63,7 @@ import { pendingTodos, } from "./opencode.ts"; import { accountPendingPromotions } from "./promotion-accounting.ts"; -import { WORKSPACE_MEMORY_CACHE_LIMITS } from "./types.ts"; +import { type LongTermMemoryEntry, WORKSPACE_MEMORY_CACHE_LIMITS } from "./types.ts"; /** * Build the complete compaction prompt. @@ -162,6 +162,11 @@ function renderTodosForCompaction(todos: Array<{ content: string; status: string return lines.join("\n"); } +function warnMemoryHook(scope: string, error: unknown): void { + const message = error instanceof Error ? error.message : String(error); + console.error(`[memory] ${scope} failed: ${message}`); +} + export const MemoryV2Plugin: Plugin = async (input) => { const { directory, client } = input; @@ -285,6 +290,22 @@ export const MemoryV2Plugin: Plugin = async (input) => { rememberProcessedUserMessage(sessionID, latestMessage.id, processedForSession); } + function dedupePendingPromotionMemories( + memories: LongTermMemoryEntry[], + ): LongTermMemoryEntry[] { + const seen = new Set(); + const deduped: LongTermMemoryEntry[] = []; + + for (const memory of memories) { + const key = memoryKey(memory); + if (seen.has(key)) continue; + seen.add(key); + deduped.push(memory); + } + + return deduped; + } + async function promotePendingMemories( sessionID?: string, options: { includeUnownedJournal?: boolean; includeOwnedJournal?: boolean } = {}, @@ -302,10 +323,10 @@ export const MemoryV2Plugin: Plugin = async (input) => { return false; }); - const pending = [ + const pending = dedupePendingPromotionMemories([ ...(sessionState?.pendingMemories ?? []), ...journalPending, - ]; + ]); if (pending.length === 0) return; let beforeEntries: Awaited>["entries"] = []; @@ -462,38 +483,42 @@ export const MemoryV2Plugin: Plugin = async (input) => { const { sessionID } = hookInput; if (!sessionID) return; - pruneFrozenWorkspaceMemoryCache(); - pruneProcessedUserMessagesCache(); + try { + pruneFrozenWorkspaceMemoryCache(); + pruneProcessedUserMessagesCache(); - // Sub-agents are short-lived - skip memory system - if (await isSubAgent(sessionID)) return; + // Sub-agents are short-lived - skip memory system + if (await isSubAgent(sessionID)) return; - // 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); + // 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 }); - } + // 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); + // Get frozen workspace memory snapshot (loaded and rendered once per session) + const workspaceSnapshot = await getFrozenWorkspaceMemorySnapshot(directory, sessionID); - // Get current hot session state - const sessionState = await loadSessionState(directory, sessionID); + // Get current hot session state + const sessionState = await loadSessionState(directory, sessionID); - // Inject frozen workspace memory snapshot - if (workspaceSnapshot.renderedPrompt) { - output.system.push(workspaceSnapshot.renderedPrompt); - } + // Inject frozen workspace memory snapshot + if (workspaceSnapshot.renderedPrompt) { + output.system.push(workspaceSnapshot.renderedPrompt); + } - // Render and inject hot session state - const hotPrompt = renderHotSessionState(sessionState, directory); - if (hotPrompt) { - output.system.push(hotPrompt); + // Render and inject hot session state + const hotPrompt = renderHotSessionState(sessionState, directory); + if (hotPrompt) { + output.system.push(hotPrompt); + } + } catch (error) { + warnMemoryHook("chat.system.transform", error); } }, @@ -506,50 +531,54 @@ export const MemoryV2Plugin: Plugin = async (input) => { // Sub-agents don't need memory tracking if (await isSubAgent(sessionID)) return; - await updateSessionState(directory, sessionID, state => { - // Track active files from tool usage - if (toolName === "read" || toolName === "edit" || toolName === "write" || toolName === "grep") { - const files = extractActiveFiles( - toolName, - args as Record, - toolOutput ?? "" - ); - for (const { path, action } of files) { - touchActiveFile(state, path, action); - if (action === "edit" || action === "write") { - markErrorsMaybeFixedForFile(state, path, directory); + try { + await updateSessionState(directory, sessionID, state => { + // Track active files from tool usage + if (toolName === "read" || toolName === "edit" || toolName === "write" || toolName === "grep") { + const files = extractActiveFiles( + toolName, + args as Record, + toolOutput ?? "" + ); + for (const { path, action } of files) { + touchActiveFile(state, path, action); + if (action === "edit" || action === "write") { + markErrorsMaybeFixedForFile(state, path, directory); + } } } - } - // Track errors from failed bash commands - if (toolName === "bash") { - const argsRecord = args as Record; - const command: string = typeof argsRecord?.command === "string" - ? argsRecord.command - : ""; - const outputText: string = toolOutput ?? ""; + // Track errors from failed bash commands + if (toolName === "bash") { + const argsRecord = args as Record; + const command: string = typeof argsRecord?.command === "string" + ? argsRecord.command + : ""; + const outputText: string = toolOutput ?? ""; - // Check if command succeeded - clear errors for that category - const exitCode = bashExitCode(hookOutput); - if (typeof exitCode !== "number") { - // Unknown exit status: do not extract and do not clear - } else if (exitCode === 0 && command) { - clearErrorsForSuccessfulCommand(state, command); - } else if (command) { - // Only extract errors for commands with explicit non-zero exit - const errors = extractErrorsFromBash(command, outputText); - for (const error of errors) { - upsertOpenError(state, error); + // Check if command succeeded - clear errors for that category + const exitCode = bashExitCode(hookOutput); + if (typeof exitCode !== "number") { + // Unknown exit status: do not extract and do not clear + } else if (exitCode === 0 && command) { + clearErrorsForSuccessfulCommand(state, command); + } else if (command) { + // Only extract errors for commands with explicit non-zero exit + const errors = extractErrorsFromBash(command, outputText); + for (const error of errors) { + upsertOpenError(state, error); + } } } - } - return state; - }); + return state; + }); - // Process explicit memory from latest user message - // Only process once per message ID - await processLatestUserMessage(sessionID); + // Process explicit memory from latest user message + // Only process once per message ID + await processLatestUserMessage(sessionID); + } catch (error) { + warnMemoryHook("tool.execute.after", error); + } }, /** @@ -567,95 +596,99 @@ export const MemoryV2Plugin: Plugin = async (input) => { const { sessionID } = hookInput; if (!sessionID) return; - // Sub-agents don't need compaction support - if (await isSubAgent(sessionID)) return; + try { + // Sub-agents don't need compaction support + if (await isSubAgent(sessionID)) return; - // Preserve context injected by other plugins that ran before us. - // Setting output.prompt bypasses the default prompt + context join, - // so we must explicitly carry forward any existing output.context. - const otherContext = output.context.filter(Boolean).join("\n\n"); + // Preserve context injected by other plugins that ran before us. + // Setting output.prompt bypasses the default prompt + context join, + // so we must explicitly carry forward any existing output.context. + const otherContext = output.context.filter(Boolean).join("\n\n"); - // Build our private context (workspace memory, hot state, todos) - const contextParts: string[] = []; + // Build our private context (workspace memory, hot state, todos) + const contextParts: string[] = []; // 1. Frozen workspace memory snapshot - const workspaceSnapshot = await getFrozenWorkspaceMemorySnapshot(directory, sessionID); - if (workspaceSnapshot.renderedPrompt) { - contextParts.push(workspaceSnapshot.renderedPrompt); + const workspaceSnapshot = await getFrozenWorkspaceMemorySnapshot(directory, sessionID); + if (workspaceSnapshot.renderedPrompt) { + contextParts.push(workspaceSnapshot.renderedPrompt); + } + + // 2. Hot session state + const sessionState = await loadSessionState(directory, sessionID); + const hotPrompt = renderHotSessionState(sessionState, directory); + if (hotPrompt) { + contextParts.push(hotPrompt); + } + + // 3. Pending todos from OpenCode + const todos = await pendingTodos(client, sessionID); + const todosPrompt = renderTodosForCompaction(todos); + if (todosPrompt) { + contextParts.push(todosPrompt); + } + + // Combine: other plugins' context first, then our private context + const privateContext = [otherContext, ...contextParts] + .filter(Boolean) + .join("\n\n"); + + // Replace the default prompt entirely with our ---free template + output.prompt = buildCompactionPrompt(privateContext); + + // Clear context array since we consumed it into output.prompt. + // Subsequent plugins that set output.prompt will also need to check + // output.context if they want to preserve other plugin contributions. + output.context.length = 0; + } catch (error) { + warnMemoryHook("session.compacting", error); } - - // 2. Hot session state - const sessionState = await loadSessionState(directory, sessionID); - const hotPrompt = renderHotSessionState(sessionState, directory); - if (hotPrompt) { - contextParts.push(hotPrompt); - } - - // 3. Pending todos from OpenCode - const todos = await pendingTodos(client, sessionID); - const todosPrompt = renderTodosForCompaction(todos); - if (todosPrompt) { - contextParts.push(todosPrompt); - } - - // Combine: other plugins' context first, then our private context - const privateContext = [otherContext, ...contextParts] - .filter(Boolean) - .join("\n\n"); - - // Replace the default prompt entirely with our ---free template - output.prompt = buildCompactionPrompt(privateContext); - - // Clear context array since we consumed it into output.prompt. - // Subsequent plugins that set output.prompt will also need to check - // output.context if they want to preserve other plugin contributions. - output.context.length = 0; }, // Handle session events event: async ({ event }) => { if (event.type === "session.compacted") { - const sessionID = sessionIDFromEventProperties(event.properties); - if (!sessionID) return; - - // Sub-agents don't need post-compaction processing - if (await isSubAgent(sessionID)) return; - - // Parse latest compaction summary for memory candidates, stage them into - // durable pending journal, then promote pending memories. - const summary = await latestCompactionSummary(client, sessionID); - const candidates = summary ? parseWorkspaceMemoryCandidates(summary) : []; - if (candidates.length > 0) { - await appendPendingMemories(directory, candidates); - } - try { + const sessionID = sessionIDFromEventProperties(event.properties); + if (!sessionID) return; + + // Sub-agents don't need post-compaction processing + if (await isSubAgent(sessionID)) return; + + // Parse latest compaction summary for memory candidates, stage them into + // durable pending journal, then promote pending memories. + const summary = await latestCompactionSummary(client, sessionID); + const candidates = summary ? parseWorkspaceMemoryCandidates(summary) : []; + if (candidates.length > 0) { + await appendPendingMemories(directory, candidates); + } + await promotePendingMemories(sessionID, { includeUnownedJournal: true }); - } catch { + } catch (error) { // Keep pending memories in session/journal for retry on next event/session. + warnMemoryHook("event.session.compacted", error); } } if (event.type === "session.deleted") { - const sessionID = sessionIDFromEventProperties(event.properties); - if (sessionID) { + try { + const sessionID = sessionIDFromEventProperties(event.properties); + if (sessionID) { // Promote pending memories before deleting per-session state. // If promotion fails, leave session state and journal intact. - let promoted = false; - try { + let promoted = false; await promotePendingMemories(sessionID, { includeOwnedJournal: true, includeUnownedJournal: false }); promoted = true; - } catch { - return; - } finally { if (promoted) { frozenWorkspaceMemoryCache.delete(sessionID); processedUserMessages.delete(sessionID); sessionParentCache.delete(sessionID); } - } - await rm(await sessionStatePath(directory, sessionID), { force: true }); + await rm(await sessionStatePath(directory, sessionID), { force: true }); + } + } catch (error) { + warnMemoryHook("event.session.deleted", error); } } }, diff --git a/src/storage.ts b/src/storage.ts index dadcf53..e4a283f 100644 --- a/src/storage.ts +++ b/src/storage.ts @@ -9,11 +9,32 @@ const LOCK_WAIT_TIMEOUT_MS = 5000; const LOCK_STALE_MS = 30_000; const LOCK_HEARTBEAT_MS = 1_000; +async function quarantineCorruptJSON(path: string): Promise { + const quarantinePath = `${path}.corrupt-${Date.now()}-${process.pid}-${randomUUID()}`; + + try { + await rename(path, quarantinePath); + return quarantinePath; + } catch { + return null; + } +} + export async function readJSON(path: string, fallback: () => T): Promise { if (!existsSync(path)) return fallback(); + try { return JSON.parse(await readFile(path, "utf8")) as T; - } catch { + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const quarantinePath = await quarantineCorruptJSON(path); + + if (quarantinePath) { + console.error(`[memory] invalid JSON in ${path}; quarantined to ${quarantinePath}: ${message}`); + } else { + console.error(`[memory] invalid JSON in ${path}; using fallback without quarantine: ${message}`); + } + return fallback(); } } diff --git a/tests/plugin.test.ts b/tests/plugin.test.ts index 46749cf..42b6bcb 100644 --- a/tests/plugin.test.ts +++ b/tests/plugin.test.ts @@ -1,6 +1,6 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import { MemoryV2Plugin } from "../src/plugin.ts"; @@ -450,6 +450,33 @@ test("chat system transform reuses frozen rendered workspace snapshot", async () } }); +test("chat system transform degrades gracefully when workspace memory JSON is corrupt", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const path = await workspaceMemoryPath(tmpDir); + await mkdir(dirname(path), { recursive: true }); + await writeFile(path, "{ invalid json", "utf8"); + + const plugin = await MemoryV2Plugin({ + directory: tmpDir, + client: mockRootClient(), + }); + const output = { system: ["base header"] }; + + await assert.doesNotReject(async () => { + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "corrupt-json-session", model: {} }, + output, + ); + }); + + assert.deepEqual(output.system, ["base header"]); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + test("no compaction: owned explicit memory is not promoted by unrelated next session start", async () => { const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); @@ -876,6 +903,47 @@ test("integration: explicit memory flows from user message through pending journ } }); +test("session.compacted promotes first-time explicit memory without self-reinforcement", async () => { + const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); + + try { + const plugin = await MemoryV2Plugin({ + directory: tmpDir, + client: mockClientWithLatestUser( + "remember this: Prefer no self reinforcement during first promotion.", + "msg-no-self-reinforce", + ), + }); + + await (plugin as Record)["experimental.chat.system.transform"]( + { sessionID: "no-self-reinforce-session", model: {} }, + { system: ["base header"] }, + ); + + await (plugin as Record)["event"]({ + event: { + type: "session.compacted", + properties: { sessionID: "no-self-reinforce-session" }, + }, + }); + + const workspace = await loadWorkspaceMemory(tmpDir); + const promoted = workspace.entries.find(entry => + /Prefer no self reinforcement/.test(entry.text) + ); + + assert.ok(promoted, "explicit memory should be promoted"); + assert.equal(promoted.reinforcementCount ?? 0, 0); + assert.equal(promoted.lastReinforcedSessionID, undefined); + + const state = await loadSessionState(tmpDir, "no-self-reinforce-session"); + assert.equal(state.pendingMemories.length, 0); + assert.equal((await loadPendingJournal(tmpDir)).entries.length, 0); + } finally { + await rm(tmpDir, { recursive: true, force: true }); + } +}); + test("integration: compaction candidate flows through journal promotion and clears pending journal", async () => { const tmpDir = await mkdtemp(join(tmpdir(), "memory-plugin-test-")); diff --git a/tests/storage.test.ts b/tests/storage.test.ts index 9d92018..d6f0e3b 100644 --- a/tests/storage.test.ts +++ b/tests/storage.test.ts @@ -1,11 +1,11 @@ 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 { mkdtemp, readdir, 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"; +import { readJSON, updateJSON } from "../src/storage.ts"; test("updateJSON serializes concurrent increments", async () => { const root = await mkdtemp(join(tmpdir(), "wm-storage-")); @@ -37,6 +37,25 @@ test("updateJSON does not replace corrupt JSON with fallback", async () => { } }); +test("readJSON quarantines corrupt JSON and returns fallback", async () => { + const dir = await mkdtemp(join(tmpdir(), "wm-storage-corrupt-")); + const path = join(dir, "store.json"); + + try { + await writeFile(path, "{ invalid json", "utf8"); + + const loaded = await readJSON(path, () => ({ ok: true })); + + assert.deepEqual(loaded, { ok: true }); + + const files = await readdir(dir); + assert.equal(files.includes("store.json"), false); + assert.equal(files.some(file => file.startsWith("store.json.corrupt-")), true); + } finally { + await rm(dir, { 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 { @@ -81,3 +100,43 @@ test("updateJSON serializes writes across separate node processes", async () => await rm(root, { recursive: true, force: true }); } }); + +test("updateJSON waits for a live cross-process lock and preserves both updates", async () => { + const root = await mkdtemp(join(tmpdir(), "wm-storage-live-lock-")); + 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 updateJSON(path, () => ({ count: 0, order: [] }), async current => { + await new Promise(resolve => setTimeout(resolve, 250)); + return { count: current.count + 1, order: [...current.order, "child"] }; + }); + `; + + const child = spawn( + process.execPath, + ["--experimental-strip-types", "--input-type=module", "-e", worker, path], + { stdio: "inherit" }, + ); + + await new Promise(resolve => setTimeout(resolve, 50)); + + await updateJSON(path, () => ({ count: 0, order: [] as string[] }), current => ({ + count: current.count + 1, + order: [...current.order, "parent"], + })); + + await new Promise((resolve, reject) => { + child.on("exit", code => code === 0 ? resolve() : reject(new Error(`child exited ${code}`))); + child.on("error", reject); + }); + + const final = await updateJSON(path, () => ({ count: 0, order: [] as string[] }), current => current); + assert.equal(final.count, 2); + assert.deepEqual(new Set(final.order), new Set(["child", "parent"])); + assert.equal(existsSync(`${path}.lock`), false); + } finally { + await rm(root, { recursive: true, force: true }); + } +});