From e2992c1da8a7269532b2c8cebf8343472e2c325e Mon Sep 17 00:00:00 2001 From: BobbyNooby Date: Sun, 27 Sep 2026 18:32:32 +0100 Subject: [PATCH 1/5] fix(v2): scope runtime ownership per project directory --- src/v2/runtime.ts | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index 2fee56b..9b51f25 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -1255,7 +1255,9 @@ export class V2Runtime { } } -const OWNER_KEY = Symbol.for("opencode-supermemory.v2.owner"); +function ownerKeyFor(directory: string | undefined): symbol { + return Symbol.for(`opencode-supermemory.v2.owner:${directory ?? "global"}`); +} interface GlobalOwner { generation: number; @@ -1267,14 +1269,18 @@ function ownerRegistry(): Record { } /** - * Starts the OpenCode 2 runtime. Only one instance is active per process, so a - * hot-reloaded plugin replaces (and cleans up) the previous generation. + * Starts the OpenCode 2 runtime. The OpenCode server instantiates the plugin + * once per project directory in the same process, so ownership is scoped per + * directory: a hot-reloaded plugin replaces (and cleans up) the previous + * generation of the same directory only, while other directories keep their + * runtime active. */ export async function setupV2( ctx: V2Context, options?: Partial, ): Promise<() => void> { const registry = ownerRegistry(); + const OWNER_KEY = ownerKeyFor(ctx.location?.directory); const previous = registry[OWNER_KEY]; previous?.cleanup(); From 017c6b0f3edf822ac366916f10bfd82eb04598ff Mon Sep 17 00:00:00 2001 From: BobbyNooby Date: Sun, 27 Sep 2026 21:30:45 +0100 Subject: [PATCH 2/5] test(v2): cover per-directory runtime ownership --- src/v2/runtime.test.ts | 95 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/src/v2/runtime.test.ts b/src/v2/runtime.test.ts index bb05a5a..489bff5 100644 --- a/src/v2/runtime.test.ts +++ b/src/v2/runtime.test.ts @@ -9,6 +9,7 @@ import { buildTranscriptTurns, buildV2RecallDirective, mergeTurns, + setupV2, SUPERMEMORY_RECALL_TOOL_NAME, SUPERMEMORY_TOOL_NAME, V2Runtime, @@ -464,3 +465,97 @@ describe("OpenCode 2 runtime", () => { h.runtime.cleanup(); }); }); + +function ownershipContext(directory: string): V2Context { + const registration = { dispose: async () => undefined }; + return { + location: { directory }, + tool: { + transform: async () => registration, + hook: async () => registration, + reload: async () => undefined, + list: async () => [], + }, + session: { + hook: async () => registration, + get: async ({ sessionID }: { sessionID: string }) => ({ + id: sessionID, + location: { directory }, + }), + context: async () => [], + }, + permission: { + hook: async () => registration, + }, + rpc: { + register: async () => ({ + ...registration, + events: { emit: async () => undefined }, + }), + }, + event: { + subscribe: () => ({ + [Symbol.asyncIterator]: () => ({ + next: () => new Promise>(() => undefined), + }), + }), + }, + } as unknown as V2Context; +} + +function ownershipDeps(): Partial { + return { + configured: true, + config: { + recallMode: "direct", + injectProfile: false, + captureEveryNTurns: 1, + compactionEnabled: false, + keywordPatterns: [], + maxProjectMemories: 10, + } as V2RuntimeDependencies["config"], + memoryClient: { + searchMemories: async () => ({ success: true, results: [] }), + addMemory: async () => ({ success: true, id: "mem_1" }), + } as unknown as V2RuntimeDependencies["memoryClient"], + executeTool: executeSupermemoryTool, + resolveTags: () => tags, + logger: () => undefined, + checkUpdate: async () => ({ + currentVersion: "2.0.15", + latestVersion: "2.0.15", + updateCommand: "bunx opencode-supermemory@latest install", + }), + }; +} + +function ownerFor(directory: string): { generation: number } | undefined { + return ( + globalThis as unknown as Record< + symbol, + { generation: number } | undefined + > + )[Symbol.for(`opencode-supermemory.v2.owner:${directory}`)]; +} + +describe("OpenCode 2 runtime ownership", () => { + test("each project directory owns its runtime; hot reload replaces only the same directory", async () => { + const cleanupA1 = await setupV2(ownershipContext("/repo-a"), ownershipDeps()); + const cleanupB1 = await setupV2(ownershipContext("/repo-b"), ownershipDeps()); + + expect(ownerFor("/repo-a")?.generation).toBe(1); + expect(ownerFor("/repo-b")?.generation).toBe(1); + + // A same-directory hot reload replaces the previous generation... + const cleanupA2 = await setupV2(ownershipContext("/repo-a"), ownershipDeps()); + expect(ownerFor("/repo-a")?.generation).toBe(2); + // ...but a different directory never disposes another directory's runtime. + expect(ownerFor("/repo-b")?.generation).toBe(1); + + cleanupA2(); + cleanupA1(); + cleanupB1(); + expect(ownerFor("/repo-a")).toBeUndefined(); + expect(ownerFor("/repo-b")).toBeUndefined(); + }); +}); From d108d5f9d1de392785fbba222a5d1b17f4d27df9 Mon Sep 17 00:00:00 2001 From: Ishaan Gupta Date: Sat, 10 Oct 2026 20:11:57 +0530 Subject: [PATCH 3/5] fix(v2): ignore events for sessions in other directories --- src/v2/runtime.test.ts | 133 +++++++++++++++++++++++++++++++++++++++-- src/v2/runtime.ts | 91 ++++++++++++++++++++++++++++ 2 files changed, 219 insertions(+), 5 deletions(-) diff --git a/src/v2/runtime.test.ts b/src/v2/runtime.test.ts index 489bff5..581950b 100644 --- a/src/v2/runtime.test.ts +++ b/src/v2/runtime.test.ts @@ -8,6 +8,7 @@ import { applyInjection, buildTranscriptTurns, buildV2RecallDirective, + isSameDirectory, mergeTurns, setupV2, SUPERMEMORY_RECALL_TOOL_NAME, @@ -16,6 +17,7 @@ import { type RequestMessage, type TranscriptMessage, type V2Context, + type V2Event, type V2RuntimeDependencies, } from "./runtime.js"; @@ -134,12 +136,24 @@ interface Harness { writes: Array<{ customId?: string; timeoutMs?: number; metadata?: Record }>; adds: Array<{ content: string; customId?: string; metadata?: Record }>; emitted: Array<{ kind?: unknown; message?: unknown }>; + sessionLookups: string[]; transcript: TranscriptMessage[]; } +interface HarnessOptions { + /** The plugin instance's location. */ + directory?: string; + /** Directory OpenCode reports for a session; throw to simulate a failed lookup. */ + sessionDirectory?: (sessionID: string) => string; +} + function harness( config: Partial = {}, + options: HarnessOptions = {}, ): Harness { + const directory = options.directory ?? "/repo"; + const sessionDirectory = options.sessionDirectory ?? (() => directory); + const sessionLookups: string[] = []; const tools = new Map(); const hooks: Harness["hooks"] = {}; const queries: string[] = []; @@ -150,7 +164,7 @@ function harness( const registration = { dispose: async () => undefined }; const ctx = { - location: { directory: "/repo" }, + location: { directory }, tool: { transform: async (callback: (editor: unknown) => void) => { callback({ @@ -177,10 +191,10 @@ function harness( hooks[`session.${name}`] = callback; return registration; }, - get: async ({ sessionID }: { sessionID: string }) => ({ - id: sessionID, - location: { directory: "/repo" }, - }), + get: async ({ sessionID }: { sessionID: string }) => { + sessionLookups.push(sessionID); + return { id: sessionID, location: { directory: sessionDirectory(sessionID) } }; + }, context: async () => state.transcript, }, permission: { @@ -282,6 +296,7 @@ function harness( writes, adds, emitted, + sessionLookups, get transcript() { return state.transcript; }, @@ -538,6 +553,114 @@ function ownerFor(directory: string): { generation: number } | undefined { )[Symbol.for(`opencode-supermemory.v2.owner:${directory}`)]; } +describe("OpenCode 2 runtime session ownership", () => { + // OpenCode delivers every location's events to every plugin instance, so two + // runtimes for different directories both receive each event below. + const directories: Record = { + "ses-a": "/repo-a", + "ses-b": "/repo-b", + }; + const lookup = (sessionID: string) => { + const directory = directories[sessionID]; + if (!directory) throw new Error(`Session not found: ${sessionID}`); + return directory; + }; + const pair = () => { + const a = harness({}, { directory: "/repo-a", sessionDirectory: lookup }); + const b = harness({}, { directory: "/repo-b", sessionDirectory: lookup }); + a.transcript = [user("u1", "question 1"), assistant("a1", "answer 1")]; + b.transcript = a.transcript; + return { a, b }; + }; + const broadcast = async (runtimes: V2Runtime[], event: V2Event) => { + for (const runtime of runtimes) await runtime.handleEvent(event); + await Promise.all(runtimes.map((runtime) => runtime.idle())); + }; + + test("captures a session only in the runtime for its directory", async () => { + const { a, b } = pair(); + await a.runtime.register(); + await b.runtime.register(); + + await broadcast([a.runtime, b.runtime], { + id: "e1", + type: "session.execution.succeeded", + data: { sessionID: "ses-a" }, + }); + await broadcast([a.runtime, b.runtime], { + id: "e2", + type: "session.execution.succeeded", + data: { sessionID: "ses-b" }, + }); + + expect(a.writes.map((write) => write.metadata?.sessionId)).toEqual(["ses-a"]); + expect(b.writes.map((write) => write.metadata?.sessionId)).toEqual(["ses-b"]); + a.runtime.cleanup(); + b.runtime.cleanup(); + }); + + test("caches ownership per session and trusts sessions it already serves", async () => { + const { a, b } = pair(); + await a.runtime.register(); + await b.runtime.register(); + + // ses-a reached runtime A through its directory-scoped context hook. + await a.runtime.handleContext({ + sessionID: "ses-a", + messages: [request("u1", "continue the auth flow work from before")], + }); + const lookupsBefore = a.sessionLookups.length; + + for (const id of ["e1", "e2", "e3"]) { + await broadcast([a.runtime, b.runtime], { + id, + type: "session.execution.succeeded", + data: { sessionID: "ses-a" }, + }); + } + + expect(a.sessionLookups.length).toBe(lookupsBefore); + expect(b.sessionLookups).toEqual(["ses-a"]); + expect(b.writes).toHaveLength(0); + expect(a.writes).toHaveLength(1); + a.runtime.cleanup(); + b.runtime.cleanup(); + }); + + test("uses the event location when present and skips sessions it cannot place", async () => { + const { a, b } = pair(); + await a.runtime.register(); + await b.runtime.register(); + + const summary = "Summary ".repeat(20); + await broadcast([a.runtime, b.runtime], { + id: "e1", + type: "session.compaction.ended", + data: { sessionID: "ses-b", text: summary, reason: "auto" }, + location: { directory: "/repo-b/" }, + }); + expect(a.adds).toHaveLength(0); + expect(b.adds).toHaveLength(1); + expect(a.sessionLookups).not.toContain("ses-b"); + + await broadcast([a.runtime, b.runtime], { + id: "e2", + type: "session.execution.succeeded", + data: { sessionID: "ses-unknown" }, + }); + expect(a.writes).toHaveLength(0); + expect(b.writes).toHaveLength(0); + a.runtime.cleanup(); + b.runtime.cleanup(); + }); + + test("compares directories after normalizing them", () => { + expect(isSameDirectory("/repo-a/", "/repo-a")).toBe(true); + expect(isSameDirectory("/repo-a/../repo-a", "/repo-a")).toBe(true); + expect(isSameDirectory("/repo-a", "/repo-ab")).toBe(false); + }); +}); + describe("OpenCode 2 runtime ownership", () => { test("each project directory owns its runtime; hot reload replaces only the same directory", async () => { const cleanupA1 = await setupV2(ownershipContext("/repo-a"), ownershipDeps()); diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index 9b51f25..bb38263 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -1,4 +1,5 @@ import { createHash } from "node:crypto"; +import { resolve } from "node:path"; import type { Plugin } from "@opencode/plugin"; import type { SessionHooks } from "@opencode/plugin/promise/session"; @@ -61,6 +62,19 @@ const INLINE_CODE_PATTERN = /`[^`]+`/g; const SYNTHETIC_METADATA_KEY = "supermemory"; const UPDATE_COMMAND = "bunx opencode-supermemory@latest install"; const MAX_DISPATCHES_PER_SESSION = 8; +const MAX_TRACKED_SESSION_OWNERS = 1_000; + +/** + * Session events this runtime acts on. OpenCode delivers every location's + * events to every plugin instance, so these are only handled for sessions in + * this instance's own directory. + */ +const OWNED_SESSION_EVENTS = new Set([ + "session.execution.succeeded", + "session.execution.interrupted", + "session.deleted", + "session.compaction.ended", +]); const MIN_SUMMARY_CHARS = 100; export const SUPERMEMORY_TOOL_NAME = "supermemory"; @@ -150,6 +164,7 @@ export interface V2Event { id?: string; type: string; data?: Record; + location?: { directory?: string }; } /** Minimal shape of a persisted OpenCode 2 transcript message. */ @@ -198,6 +213,15 @@ function mergeDependencies( }; } +function normalizeDirectory(directory: string): string { + const resolved = resolve(directory).replace(/[\\/]+$/, "") || resolve(directory); + return process.platform === "win32" ? resolved.toLowerCase() : resolved; +} + +export function isSameDirectory(left: string, right: string): boolean { + return normalizeDirectory(left) === normalizeDirectory(right); +} + function sha256(value: string): string { return createHash("sha256").update(value).digest("hex"); } @@ -442,6 +466,8 @@ export class V2Runtime { readonly #pendingSummaries = new Map(); readonly #summaryInFlight = new Set(); readonly #deduper = new EventDeduper(); + readonly #sessionOwners = new Map(); + readonly #ownerLookups = new Map>(); readonly #registrations: Registration[] = []; readonly #abortController = new AbortController(); #emitActivity: ((notice: MemoryActivityNotice) => void) | undefined; @@ -702,6 +728,16 @@ export class V2Runtime { if (!this.active || this.#deduper.hasSeen(event.id)) return; const sessionID = this.#eventSessionID(event); + if ( + sessionID && + OWNED_SESSION_EVENTS.has(event.type) && + !(await this.#ownsSession(sessionID, event)) + ) { + if (event.type === "session.deleted") this.#sessionOwners.delete(sessionID); + return; + } + if (!this.active) return; + if (sessionID && event.type !== "session.compaction.ended") { void this.#retryPendingSummaries(sessionID); } @@ -728,6 +764,7 @@ export class V2Runtime { if (!sessionID) return; const state = this.#states.get(sessionID); this.#recall.delete(sessionID); + this.#sessionOwners.delete(sessionID); if (!state) return; await this.#runCaptureExclusive(sessionID, () => this.#captureSessionEnd(sessionID), @@ -798,6 +835,60 @@ export class V2Runtime { this.#registrations.push(registration); } + /** + * Whether a session belongs to this instance's directory. Hooks and tools + * are already scoped to the directory, so any session with state here is + * ours. Otherwise the event's location decides, and events without one fall + * back to a single session lookup that is cached per session. + */ + async #ownsSession(sessionID: string, event: V2Event): Promise { + const own = this.#ctx.location?.directory; + if (!own || this.#states.has(sessionID)) return true; + + const known = this.#sessionOwners.get(sessionID); + if (known !== undefined) return known; + + const eventDirectory = event.location?.directory; + if (eventDirectory) { + return this.#rememberOwner(sessionID, isSameDirectory(eventDirectory, own)); + } + + const pending = this.#ownerLookups.get(sessionID); + if (pending) return pending; + + const lookup = (async () => { + try { + const session = await this.#ctx.session.get({ sessionID }); + const directory = (session as { location?: { directory?: string } }) + .location?.directory; + if (!directory) return false; + return this.#rememberOwner(sessionID, isSameDirectory(directory, own)); + } catch (error) { + this.#deps.logger("v2 session owner lookup failed; skipping event", { + sessionID, + type: event.type, + error: String(error), + }); + return false; + } finally { + this.#ownerLookups.delete(sessionID); + } + })(); + this.#ownerLookups.set(sessionID, lookup); + return lookup; + } + + #rememberOwner(sessionID: string, owned: boolean): boolean { + this.#sessionOwners.delete(sessionID); + this.#sessionOwners.set(sessionID, owned); + while (this.#sessionOwners.size > MAX_TRACKED_SESSION_OWNERS) { + const oldest = this.#sessionOwners.keys().next().value; + if (oldest === undefined) break; + this.#sessionOwners.delete(oldest); + } + return owned; + } + #state(sessionID: string): SessionState { const existing = this.#states.get(sessionID); if (existing) return existing; From 348b7dfd9b34813d910cdd51a2a477e9a235a021 Mon Sep 17 00:00:00 2001 From: Ishaan Gupta Date: Sat, 10 Oct 2026 20:11:58 +0530 Subject: [PATCH 4/5] fix(v2): flush captures before clearing state on cleanup --- src/v2/runtime.test.ts | 19 +++++++++++++++++++ src/v2/runtime.ts | 15 ++++++++++----- 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/src/v2/runtime.test.ts b/src/v2/runtime.test.ts index 581950b..6f79790 100644 --- a/src/v2/runtime.test.ts +++ b/src/v2/runtime.test.ts @@ -654,6 +654,25 @@ describe("OpenCode 2 runtime session ownership", () => { b.runtime.cleanup(); }); + test("does not re-send saved batches when the runtime shuts down", async () => { + const h = harness({}, { directory: "/repo-a", sessionDirectory: lookup }); + h.transcript = [user("u1", "question 1"), assistant("a1", "answer 1")]; + await h.runtime.register(); + await h.runtime.handleEvent({ + id: "e1", + type: "session.execution.succeeded", + data: { sessionID: "ses-a" }, + }); + await h.runtime.idle(); + expect(h.writes).toHaveLength(1); + + h.transcript = [...h.transcript, user("u2", "question 2"), assistant("a2", "answer 2")]; + h.runtime.cleanup(); + await h.runtime.idle(); + expect(h.writes).toHaveLength(2); + expect(h.writes[1]?.customId).not.toBe(h.writes[0]?.customId); + }); + test("compares directories after normalizing them", () => { expect(isSameDirectory("/repo-a/", "/repo-a")).toBe(true); expect(isSameDirectory("/repo-a/../repo-a", "/repo-a")).toBe(true); diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index bb38263..43bf884 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -815,11 +815,16 @@ export class V2Runtime { this.#active = false; this.#abortController.abort(); - void this.#flushAll(true).catch((error) => { - this.#deps.logger("v2 cleanup capture failed", { error: String(error) }); - }); - this.#states.clear(); - this.#recall.clear(); + // Flush while the session state still records which batches were saved, + // so only unsaved turns are sent; clearing first would re-send them all. + void this.#flushAll(true) + .catch((error) => { + this.#deps.logger("v2 cleanup capture failed", { error: String(error) }); + }) + .finally(() => { + this.#states.clear(); + this.#recall.clear(); + }); for (const registration of this.#registrations.splice(0)) { this.#disposeRegistration(registration); From 23ac306965228dedd4eb62cdad328ab739d01e59 Mon Sep 17 00:00:00 2001 From: Ishaan Gupta Date: Sat, 10 Oct 2026 20:23:42 +0530 Subject: [PATCH 5/5] fix: cap capture retries and remember saved batches across reloads --- src/services/capture.test.ts | 36 ++++++++++++++ src/services/capture.ts | 52 +++++++++++++++++--- src/v2/runtime.test.ts | 94 +++++++++++++++++++++++++++++++++++- src/v2/runtime.ts | 77 ++++++++++++++++++++++++++--- 4 files changed, 242 insertions(+), 17 deletions(-) diff --git a/src/services/capture.test.ts b/src/services/capture.test.ts index 8104e01..585abe5 100644 --- a/src/services/capture.test.ts +++ b/src/services/capture.test.ts @@ -3,6 +3,7 @@ import type { Part } from "@opencode-ai/sdk"; import { AUTOMATIC_CAPTURE_TIMEOUT_MS, + MAX_CAPTURE_ATTEMPTS, buildCadenceBatches, buildCaptureTurns, buildSessionEndBatch, @@ -284,4 +285,39 @@ describe("automatic conversation capture", () => { ]); expect(readAttempts).toBe(3); }); + test("stops re-sending a batch after repeated failed writes", async () => { + let messages = conversation(1); + const writes: string[] = []; + const hook = createCaptureHook( + { directory: "/repo", client: { session: { messages: async () => ({ data: messages }) } } }, + { + canonical: "repo_test__0123456789abcdef", + user: "repo_test__0123456789abcdef", + project: "repo_test__0123456789abcdef", + projectId: "0123456789abcdef", + projectName: "test", + personalReads: [], + projectReads: [], + allReads: [], + }, + { + captureEveryNTurns: 1, + memoryClient: { + ingestConversation: async (_id, _messages, _tags, _metadata, options) => { + writes.push(options?.customId ?? ""); + return { success: false, error: "Timeout after 3250ms" }; + }, + }, + }, + ); + + for (let turn = 1; turn <= 6; turn += 1) { + messages = conversation(turn); + await hook.event({ event: { type: "session.idle", properties: { sessionID: "session-1" } } }); + } + + const firstBatch = writes.filter((id) => id === writes[0]).length; + expect(firstBatch).toBe(MAX_CAPTURE_ATTEMPTS); + expect(writes.length).toBeLessThan(21); + }); }); diff --git a/src/services/capture.ts b/src/services/capture.ts index aaa2af4..7a90699 100644 --- a/src/services/capture.ts +++ b/src/services/capture.ts @@ -10,6 +10,33 @@ import { isFullyPrivate, stripPrivateContent } from "./privacy.js"; import type { ResolvedTags } from "./tags.js"; export const AUTOMATIC_CAPTURE_TIMEOUT_MS = 3_000; +export const MAX_CAPTURE_ATTEMPTS = 3; + +/** + * Bounds how often one batch is re-sent after failed writes. A write that timed + * out on the client may still have reached the server, so retrying it on every + * later turn stores the same conversation again each time. + */ +export class CaptureRetryBudget { + readonly #failures = new Map(); + + constructor(private readonly maxAttempts = MAX_CAPTURE_ATTEMPTS) {} + + /** Records a failed write. Returns true once the batch should stop retrying. */ + recordFailure(captureId: string): boolean { + const failures = (this.#failures.get(captureId) ?? 0) + 1; + if (failures >= this.maxAttempts) { + this.#failures.delete(captureId); + return true; + } + this.#failures.set(captureId, failures); + return false; + } + + recordSuccess(captureId: string): void { + this.#failures.delete(captureId); + } +} interface CaptureMessageInfo { id: string; @@ -220,6 +247,7 @@ export function createCaptureHook( const snapshots = new Map(); const activeSessions = new Set(); const completedCaptureIds = new Set(); + const retries = new CaptureRetryBudget(); const inFlight = new Map>(); async function refreshSnapshot(sessionID: string): Promise { @@ -282,6 +310,7 @@ export function createCaptureHook( if (result.success) { completedCaptureIds.add(captureId); + retries.recordSuccess(captureId); options?.onSaved?.(); log("[capture] conversation batch saved", { sessionID, @@ -292,14 +321,21 @@ export function createCaptureHook( return true; } - log("[capture] failed to save conversation batch", { - sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, - error: result.error, - }); - return false; + const exhausted = retries.recordFailure(captureId); + if (exhausted) completedCaptureIds.add(captureId); + log( + exhausted + ? "[capture] giving up on conversation batch after repeated failures" + : "[capture] failed to save conversation batch", + { + sessionID, + reason, + startTurn: batch.startTurn, + endTurn: batch.endTurn, + error: result.error, + }, + ); + return exhausted; } async function captureCadence( diff --git a/src/v2/runtime.test.ts b/src/v2/runtime.test.ts index 6f79790..d7b1167 100644 --- a/src/v2/runtime.test.ts +++ b/src/v2/runtime.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test"; +import { MAX_CAPTURE_ATTEMPTS } from "../services/capture.js"; import { COMPACTION_CONTEXT_MARKER } from "../services/compaction-prompt.js"; import { executeSupermemoryTool } from "../services/memory-tool.js"; import { DEFAULT_RECALL_DIRECTIVE } from "../services/recall.js"; @@ -145,12 +146,17 @@ interface HarnessOptions { directory?: string; /** Directory OpenCode reports for a session; throw to simulate a failed lookup. */ sessionDirectory?: (sessionID: string) => string; + /** Result of each capture write. */ + writeResult?: () => { success: boolean; error?: string }; + /** Plugin storage shared between runtimes, as OpenCode keeps it across reloads. */ + storage?: Map; } function harness( config: Partial = {}, options: HarnessOptions = {}, ): Harness { + const harnessOptions = options; const directory = options.directory ?? "/repo"; const sessionDirectory = options.sessionDirectory ?? (() => directory); const sessionLookups: string[] = []; @@ -163,8 +169,18 @@ function harness( const state: { transcript: TranscriptMessage[] } = { transcript: [] }; const registration = { dispose: async () => undefined }; + const storage = options.storage; const ctx = { location: { directory }, + ...(storage + ? { + storage: { + get: async (key: string) => storage.get(key), + set: async (key: string, value: unknown) => void storage.set(key, value), + remove: async (key: string) => void storage.delete(key), + }, + } + : {}), tool: { transform: async (callback: (editor: unknown) => void) => { callback({ @@ -251,7 +267,7 @@ function harness( options?: { customId?: string; timeoutMs?: number }, ) => { writes.push({ customId: options?.customId, timeoutMs: options?.timeoutMs, metadata }); - return { success: true }; + return harnessOptions.writeResult?.() ?? { success: true }; }, addMemory: async ( content: string, @@ -673,6 +689,82 @@ describe("OpenCode 2 runtime session ownership", () => { expect(h.writes[1]?.customId).not.toBe(h.writes[0]?.customId); }); + test("stops re-sending a batch after repeated failed writes", async () => { + const h = harness( + {}, + { + directory: "/repo-a", + sessionDirectory: lookup, + // The server stays slow: every write times out on the client. + writeResult: () => ({ success: false, error: "Timeout after 3250ms" }), + }, + ); + await h.runtime.register(); + const turns: TranscriptMessage[] = []; + for (let turn = 1; turn <= 6; turn += 1) { + turns.push(user(`u${turn}`, `question ${turn}`), assistant(`a${turn}`, `answer ${turn}`)); + h.transcript = [...turns]; + await h.runtime.handleEvent({ + id: `e${turn}`, + type: "session.execution.succeeded", + data: { sessionID: "ses-a" }, + }); + await h.runtime.idle(); + } + + const perBatch = new Map(); + for (const write of h.writes) { + perBatch.set(write.customId!, (perBatch.get(write.customId!) ?? 0) + 1); + } + expect(Math.max(...perBatch.values())).toBe(MAX_CAPTURE_ATTEMPTS); + // Unbounded retries would have sent 1 + 2 + ... + 6 = 21 writes. + expect(h.writes.length).toBeLessThan(21); + h.runtime.cleanup(); + }); + + test("a reloaded runtime does not re-send batches saved before the reload", async () => { + const storage = new Map(); + const first = harness({}, { directory: "/repo-a", sessionDirectory: lookup, storage }); + first.transcript = [user("u1", "question 1"), assistant("a1", "answer 1")]; + await first.runtime.register(); + await first.runtime.handleEvent({ + id: "e1", + type: "session.execution.succeeded", + data: { sessionID: "ses-a" }, + }); + await first.runtime.idle(); + first.runtime.cleanup(); + await first.runtime.idle(); + expect(first.writes).toHaveLength(1); + + const second = harness({}, { directory: "/repo-a", sessionDirectory: lookup, storage }); + second.transcript = [ + ...first.transcript, + user("u2", "question 2"), + assistant("a2", "answer 2"), + ]; + await second.runtime.register(); + await second.runtime.handleEvent({ + id: "e2", + type: "session.execution.succeeded", + data: { sessionID: "ses-a" }, + }); + await second.runtime.idle(); + + expect(second.writes).toHaveLength(1); + expect(second.writes[0]?.customId).not.toBe(first.writes[0]?.customId); + + await second.runtime.handleEvent({ + id: "e3", + type: "session.deleted", + data: { sessionID: "ses-a" }, + }); + await second.runtime.idle(); + await Promise.resolve(); + expect([...storage.keys()]).toEqual([]); + second.runtime.cleanup(); + }); + test("compares directories after normalizing them", () => { expect(isSameDirectory("/repo-a/", "/repo-a")).toBe(true); expect(isSameDirectory("/repo-a/../repo-a", "/repo-a")).toBe(true); diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index 43bf884..1925e14 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -12,6 +12,7 @@ import { } from "../services/activity.js"; import { AUTOMATIC_CAPTURE_TIMEOUT_MS, + CaptureRetryBudget, buildCadenceBatches, buildSessionEndBatch, getCaptureId, @@ -63,6 +64,8 @@ const SYNTHETIC_METADATA_KEY = "supermemory"; const UPDATE_COMMAND = "bunx opencode-supermemory@latest install"; const MAX_DISPATCHES_PER_SESSION = 8; const MAX_TRACKED_SESSION_OWNERS = 1_000; +const MAX_SAVED_CAPTURES_PER_SESSION = 500; +const SAVED_CAPTURES_KEY_PREFIX = "capture/saved/"; /** * Session events this runtime acts on. OpenCode delivers every location's @@ -191,6 +194,7 @@ interface SessionState { turns: CaptureTurn[]; turnIndex: Map; completedCaptureIds: Set; + savedCapturesLoaded?: Promise; } interface PendingSummary { @@ -467,6 +471,7 @@ export class V2Runtime { readonly #summaryInFlight = new Set(); readonly #deduper = new EventDeduper(); readonly #sessionOwners = new Map(); + readonly #captureRetries = new CaptureRetryBudget(); readonly #ownerLookups = new Map>(); readonly #registrations: Registration[] = []; readonly #abortController = new AbortController(); @@ -770,6 +775,9 @@ export class V2Runtime { this.#captureSessionEnd(sessionID), ); this.#states.delete(sessionID); + void (this.#ctx as { storage?: V2Context["storage"] }).storage + ?.remove(`${SAVED_CAPTURES_KEY_PREFIX}${sessionID}`) + .catch(() => undefined); return; } @@ -1165,6 +1173,8 @@ export class V2Runtime { if (result.success) { state.completedCaptureIds.add(captureId); + this.#captureRetries.recordSuccess(captureId); + await this.#persistSavedCaptures(sessionID, state); this.#activity.saved(); this.#deps.logger("[capture] conversation batch saved", { sessionID, @@ -1175,18 +1185,68 @@ export class V2Runtime { return true; } - this.#deps.logger("[capture] failed to save conversation batch", { - sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, - error: result.error, - }); - return false; + const exhausted = this.#captureRetries.recordFailure(captureId); + if (exhausted) state.completedCaptureIds.add(captureId); + this.#deps.logger( + exhausted + ? "[capture] giving up on conversation batch after repeated failures" + : "[capture] failed to save conversation batch", + { + sessionID, + reason, + startTurn: batch.startTurn, + endTurn: batch.endTurn, + error: result.error, + }, + ); + return exhausted; + } + + /** + * Saved batches are kept in plugin storage so a reloaded or restarted + * runtime does not send turns that an earlier instance already saved. + * Storage is best effort: when it is unavailable, capture still works. + */ + #loadSavedCaptures(sessionID: string, state: SessionState): Promise { + state.savedCapturesLoaded ??= (async () => { + const storage = (this.#ctx as { storage?: V2Context["storage"] }).storage; + if (!storage) return; + try { + const saved = await storage.get(`${SAVED_CAPTURES_KEY_PREFIX}${sessionID}`); + if (Array.isArray(saved)) { + for (const id of saved) { + if (typeof id === "string") state.completedCaptureIds.add(id); + } + } + } catch (error) { + this.#deps.logger("[capture] could not read saved batches", { + sessionID, + error: String(error), + }); + } + })(); + return state.savedCapturesLoaded; + } + + async #persistSavedCaptures(sessionID: string, state: SessionState): Promise { + const storage = (this.#ctx as { storage?: V2Context["storage"] }).storage; + if (!storage) return; + try { + await storage.set( + `${SAVED_CAPTURES_KEY_PREFIX}${sessionID}`, + [...state.completedCaptureIds].slice(-MAX_SAVED_CAPTURES_PER_SESSION), + ); + } catch (error) { + this.#deps.logger("[capture] could not record saved batch", { + sessionID, + error: String(error), + }); + } } async #captureCadence(sessionID: string): Promise { const state = this.#state(sessionID); + await this.#loadSavedCaptures(sessionID, state); const turns = await this.#refreshTurns(sessionID, state); for (const batch of buildCadenceBatches( turns, @@ -1198,6 +1258,7 @@ export class V2Runtime { async #captureSessionEnd(sessionID: string): Promise { const state = this.#state(sessionID); + await this.#loadSavedCaptures(sessionID, state); const turns = await this.#refreshTurns(sessionID, state); for (const batch of buildCadenceBatches( turns,