From 88c7e433907083856efde759bd5c853bc9b64caf Mon Sep 17 00:00:00 2001 From: Dhravya Date: Mon, 12 Oct 2026 11:33:25 +0000 Subject: [PATCH 1/2] fix(v2): isolate project capture and persist acknowledged batches Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- README.md | 4 + package.json | 2 +- src/services/client.ts | 9 + src/v2/runtime.ts | 383 ++++++++++++++++++++++++++++++++--------- src/version.ts | 2 +- 5 files changed, 319 insertions(+), 81 deletions(-) diff --git a/README.md b/README.md index 19da222..fc92dd4 100644 --- a/README.md +++ b/README.md @@ -169,6 +169,10 @@ Completed conversations are captured automatically: - Synthetic plugin context is excluded and `` content is redacted. - Stable capture IDs retain batch identity across repeated lifecycle events; they do not guarantee exactly-once processing or billing after an ambiguous response. +On OpenCode 2, each directory keeps its own tools and hooks, and only handles events for its sessions. Accepted capture batches and fully submitted turn fingerprints are recorded in host plugin storage, scoped to the API destination, namespace, directory and session. Reloads and restarts suppress those confirmed uploads; a storage read error defers capture rather than treating history as unsaved. Failed writes remain retryable without a permanent attempt cap. + +Older versions kept acceptance only in memory, so the first resume after upgrading has no durable evidence of earlier uploads and may retry historical content. Original host transcripts and configuration are retained; no historical batch is invented as accepted. A timed-out request may also have reached the server: neither stable IDs nor a queued acceptance proves exactly-once processing. If an ACK cannot be persisted, it is retained across same-process replacement, but a process exit before storage recovers still leaves that uncertainty. The existing 100,000-character upload limit remains; truncated turns are not recorded as fully submitted. + ### Keyword Detection Say "remember", "save this", "don't forget" etc. and the agent auto-saves to memory. diff --git a/package.json b/package.json index 67b212c..ff5dfe8 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode-supermemory", - "version": "2.0.17", + "version": "2.0.18", "description": "OpenCode plugin that gives coding agents persistent memory using Supermemory", "type": "module", "main": "dist/index.js", diff --git a/src/services/client.ts b/src/services/client.ts index 0604f7a..7f54baf 100644 --- a/src/services/client.ts +++ b/src/services/client.ts @@ -672,6 +672,14 @@ export class SupermemoryClient { rawContent.length > MAX_CONVERSATION_CHARS ? `${rawContent.slice(0, MAX_CONVERSATION_CHARS)}\n...[truncated]` : rawContent; + let submittedMessageCount = 0; + let submittedLength = `[Conversation ${conversationId}]\n`.length; + for (const [index, message] of messages.entries()) { + submittedLength += `${index + 1}. ${this.formatConversationMessage(message)}`.length; + if (submittedLength > MAX_CONVERSATION_CHARS) break; + submittedMessageCount += 1; + submittedLength += 1; + } const ingestMetadata = { type: "conversation" as const, @@ -734,6 +742,7 @@ export class SupermemoryClient { success: true as const, ...response, storedMemoryIds: savedIds, + submittedMessageCount, }; } } diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index 2fee56b..553a16e 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -1,9 +1,10 @@ import { createHash } from "node:crypto"; +import { resolve } from "node:path"; import type { Plugin } from "@opencode/plugin"; import type { SessionHooks } from "@opencode/plugin/promise/session"; -import { CONFIG, isConfigured, PLUGIN_VERSION } from "../config.js"; +import { CONFIG, getApiBaseUrl, getApiVersion, isConfigured, PLUGIN_VERSION } from "../config.js"; import { createMemoryActivityReporterFromSink, type MemoryActivityNotice, @@ -145,11 +146,11 @@ const DEFAULT_DEPENDENCIES: V2RuntimeDependencies = { checkNpmUpdate("opencode-supermemory", PLUGIN_VERSION, UPDATE_COMMAND), }; -/** Minimal shape of an OpenCode 2 event as read from `ctx.event.subscribe()`. */ export interface V2Event { id?: string; type: string; data?: Record; + location?: { directory?: string }; } /** Minimal shape of a persisted OpenCode 2 transcript message. */ @@ -175,7 +176,11 @@ interface SessionState { dispatchOrder: string[]; turns: CaptureTurn[]; turnIndex: Map; - completedCaptureIds: Set; + acknowledgedCaptureIds: Set; + acknowledgedTurnFingerprints: Set; + captureLedgerLoaded?: Promise; + captureLedgerKey?: string; + captureLedgerDirty: boolean; } interface PendingSummary { @@ -184,6 +189,20 @@ interface PendingSummary { text: string; } +interface CaptureHandoff { + sessions: Array<{ + sessionID: string; + directory?: string; + tags?: ResolvedTags; + turns: CaptureTurn[]; + captureIds: string[]; + turnFingerprints: string[]; + ledgerKey?: string; + dirty: boolean; + }>; + summaries: PendingSummary[]; +} + interface Registration { dispose: () => Promise; } @@ -202,6 +221,11 @@ function sha256(value: string): string { return createHash("sha256").update(value).digest("hex"); } +function normalizeDirectory(directory: string): string { + const normalized = resolve(directory); + return process.platform === "win32" ? normalized.toLowerCase() : normalized; +} + function removeCodeBlocks(text: string): string { return text.replace(CODE_BLOCK_PATTERN, "").replace(INLINE_CODE_PATTERN, ""); } @@ -441,12 +465,17 @@ export class V2Runtime { readonly #captureInFlight = new Map>(); readonly #pendingSummaries = new Map(); readonly #summaryInFlight = new Set(); + readonly #summaryTasks = new Map>(); readonly #deduper = new EventDeduper(); + readonly #sessionOwners = new Map(); + readonly #ownerLookups = new Map>(); readonly #registrations: Registration[] = []; + readonly #registrationTasks = new Set>(); readonly #abortController = new AbortController(); #emitActivity: ((notice: MemoryActivityNotice) => void) | undefined; #activity: MemoryActivityReporter; #active = true; + #cleanup: Promise | undefined; constructor( ctx: V2Context, @@ -471,6 +500,7 @@ export class V2Runtime { } async register(): Promise { + if (!this.active) return; const { logger } = this.#deps; await this.#track( @@ -558,8 +588,10 @@ export class V2Runtime { if (!this.active) return; try { - const rpc = await this.#ctx.rpc.register(SUPERMEMORY_RPC, {}); - this.#registrations.push(rpc); + const pending = this.#ctx.rpc.register(SUPERMEMORY_RPC, {}); + await this.#track(pending); + const rpc = await pending; + if (!this.active) return; this.#emitActivity = (notice) => { void rpc.events .emit("activity", { @@ -682,7 +714,8 @@ export class V2Runtime { } } - applyInjection(event.messages, latestUser, await pending); + const injection = await pending; + if (this.active) applyInjection(event.messages, latestUser, injection); } async handleCompaction(event: { @@ -702,6 +735,11 @@ export class V2Runtime { if (!this.active || this.#deduper.hasSeen(event.id)) return; const sessionID = this.#eventSessionID(event); + if (!sessionID && event.location?.directory && + normalizeDirectory(event.location.directory) !== normalizeDirectory(this.#ctx.location.directory)) return; + if (sessionID && !(await this.#ownsSession(sessionID, event))) return; + if (!this.active) return; + if (sessionID && event.type !== "session.compaction.ended") { void this.#retryPendingSummaries(sessionID); } @@ -729,10 +767,14 @@ export class V2Runtime { const state = this.#states.get(sessionID); this.#recall.delete(sessionID); if (!state) return; - await this.#runCaptureExclusive(sessionID, () => - this.#captureSessionEnd(sessionID), - ); - this.#states.delete(sessionID); + await this.#runCaptureExclusive(sessionID, async () => { + if (!(await this.#captureSessionEnd(sessionID))) return; + if (state.captureLedgerKey) { + await this.#ctx.storage.remove(state.captureLedgerKey); + } + this.#states.delete(sessionID); + this.#sessionOwners.delete(sessionID); + }); return; } @@ -755,8 +797,11 @@ export class V2Runtime { } case "global.disposed": { - await this.#flushAll(); - this.#states.clear(); + await this.idle(); + const unfinished = await this.#flushAll(); + for (const sessionID of this.#states.keys()) { + if (!unfinished.has(sessionID)) this.#states.delete(sessionID); + } this.#recall.clear(); return; } @@ -765,37 +810,115 @@ export class V2Runtime { /** Waits for background capture and summary work; used by tests and shutdown. */ async idle(): Promise { - while (this.#captureInFlight.size > 0 || this.#summaryInFlight.size > 0) { - await Promise.allSettled([...this.#captureInFlight.values()]); - if (this.#summaryInFlight.size > 0) { - await new Promise((resolve) => setTimeout(resolve, 5)); - } + while (this.#captureInFlight.size > 0 || this.#summaryTasks.size > 0) { + await Promise.allSettled([...this.#captureInFlight.values(), ...this.#summaryTasks.values()]); } } - cleanup(): void { - if (!this.#active) return; + cleanup(): Promise { + if (this.#cleanup) return this.#cleanup; this.#active = false; this.#abortController.abort(); - - void this.#flushAll(true).catch((error) => { + const registrations = this.#registrations.splice(0); + this.#cleanup = (async () => { + await Promise.allSettled(this.#registrationTasks); + const disposals = await Promise.allSettled(registrations.map((registration) => registration.dispose())); + for (const result of disposals) { + if (result.status === "rejected") { + this.#deps.logger("v2 registration cleanup failed", { error: String(result.reason) }); + } + } + await this.idle(); + const unfinished = await this.#flushAll(true); + for (const sessionID of this.#states.keys()) { + if (!unfinished.has(sessionID)) this.#states.delete(sessionID); + } + this.#recall.clear(); + })().catch((error) => { this.#deps.logger("v2 cleanup capture failed", { error: String(error) }); }); + return this.#cleanup; + } + + async #track(pending: Promise): Promise { + const task = (async () => { + const registration = await pending; + if (!this.active) { + await registration.dispose(); + return; + } + this.#registrations.push(registration); + })(); + this.#registrationTasks.add(task); + try { await task; } finally { this.#registrationTasks.delete(task); } + } + + takeCaptureState(): CaptureHandoff { + const snapshot: CaptureHandoff = { + sessions: [...this.#states].map(([sessionID, state]) => ({ + sessionID, + directory: state.directory, + tags: state.tags, + turns: state.turns, + captureIds: [...state.acknowledgedCaptureIds], + turnFingerprints: [...state.acknowledgedTurnFingerprints], + ledgerKey: state.captureLedgerKey, + dirty: state.captureLedgerDirty, + })), + summaries: [...this.#pendingSummaries.values()], + }; this.#states.clear(); - this.#recall.clear(); + this.#pendingSummaries.clear(); + return snapshot; + } - for (const registration of this.#registrations.splice(0)) { - this.#disposeRegistration(registration); + inheritCaptureState(snapshot: CaptureHandoff): void { + for (const source of snapshot.sessions) { + const state = this.#state(source.sessionID); + state.directory = source.directory; + state.tags = source.tags; + state.turns = source.turns; + state.turnIndex = new Map(source.turns.map((turn, index) => [turn.id, index])); + state.acknowledgedCaptureIds = new Set(source.captureIds); + state.acknowledgedTurnFingerprints = new Set(source.turnFingerprints); + state.captureLedgerKey = source.ledgerKey; + state.captureLedgerDirty = source.dirty; } + for (const summary of snapshot.summaries) this.#pendingSummaries.set(summary.customId, summary); } - async #track(pending: Promise): Promise { - const registration = await pending; - if (!this.active) { - this.#disposeRegistration(registration); - return; + async #ownsSession(sessionID: string, event: V2Event): Promise { + const directory = event.location?.directory; + if (directory) { + const owned = normalizeDirectory(directory) === normalizeDirectory(this.#ctx.location.directory); + this.#sessionOwners.set(sessionID, owned); + return owned; } - this.#registrations.push(registration); + const known = this.#sessionOwners.get(sessionID); + if (known !== undefined) return known; + const pending = this.#ownerLookups.get(sessionID); + if (pending) return pending; + const lookup = (async () => { + try { + const session = await this.#ctx.session.get({ sessionID }); + const owned = normalizeDirectory(session.location.directory) === + normalizeDirectory(this.#ctx.location.directory); + this.#sessionOwners.set(sessionID, owned); + if (this.#sessionOwners.size > 1024) { + this.#sessionOwners.delete(this.#sessionOwners.keys().next().value!); + } + return owned; + } catch (error) { + this.#deps.logger("v2 session owner lookup failed; skipping event", { + sessionID, error: String(error), + }); + return false; + } finally { + this.#ownerLookups.delete(sessionID); + } + })(); + this.#ownerLookups.set(sessionID, lookup); + return lookup; } #state(sessionID: string): SessionState { @@ -807,7 +930,9 @@ export class V2Runtime { dispatchOrder: [], turns: [], turnIndex: new Map(), - completedCaptureIds: new Set(), + acknowledgedCaptureIds: new Set(), + acknowledgedTurnFingerprints: new Set(), + captureLedgerDirty: false, }; this.#states.set(sessionID, state); return state; @@ -820,20 +945,6 @@ export class V2Runtime { }); } - #disposeRegistration(registration: Registration): void { - try { - void registration.dispose().catch((error) => { - this.#deps.logger("v2 registration cleanup failed", { - error: String(error), - }); - }); - } catch (error) { - this.#deps.logger("v2 registration cleanup failed", { - error: String(error), - }); - } - } - #dispatchKey( messages: RequestMessage[], latestUser: RequestMessage, @@ -866,6 +977,10 @@ export class V2Runtime { if (!directory) { throw new Error(`Unable to resolve directory for OpenCode session ${sessionID}`); } + if (normalizeDirectory(directory) !== normalizeDirectory(this.#ctx.location.directory)) { + throw new Error(`OpenCode session ${sessionID} belongs to another location`); + } + this.#sessionOwners.set(sessionID, true); state.directory = directory; state.tags = this.#deps.resolveTags(directory); return state.tags; @@ -1006,6 +1121,10 @@ export class V2Runtime { async #refreshTurns(sessionID: string, state: SessionState): Promise { try { + const session = await this.#ctx.session.get({ sessionID }); + if (normalizeDirectory(session.location.directory) !== normalizeDirectory(this.#ctx.location.directory)) { + return state.turns; + } const messages = (await this.#ctx.session.context({ sessionID })) as unknown; if (Array.isArray(messages)) { mergeTurns( @@ -1029,15 +1148,20 @@ export class V2Runtime { batch: CaptureBatch, reason: "cadence" | "session_end", ): Promise { + const remaining = batch.turns.filter((turn) => + turn.messages.length > 0 && !state.acknowledgedTurnFingerprints.has(sha256(JSON.stringify(turn)))); + if (remaining.length === 0) return true; + batch = { + startTurn: batch.startTurn + batch.turns.indexOf(remaining[0]!), + endTurn: batch.startTurn + batch.turns.indexOf(remaining.at(-1)!), + turns: remaining, + }; const captureId = getCaptureId(sessionID, batch); - if (state.completedCaptureIds.has(captureId)) return true; const messages = batch.turns.flatMap((turn) => turn.messages); - if (messages.length === 0) { - state.completedCaptureIds.add(captureId); - return true; - } + const acknowledgedId = `${captureId}:${sha256(JSON.stringify(messages))}`; + if (state.acknowledgedCaptureIds.has(acknowledgedId)) return true; - let result: { success: boolean; error?: string }; + let result: { success: boolean; error?: string; submittedMessageCount?: number }; try { const tags = await this.#resolveSession(sessionID); result = await this.#deps.memoryClient.ingestConversation( @@ -1068,7 +1192,17 @@ export class V2Runtime { } if (result.success) { - state.completedCaptureIds.add(captureId); + const submittedMessageCount = result.submittedMessageCount ?? messages.length; + state.acknowledgedCaptureIds.add(acknowledgedId); + let messageCount = 0; + for (const turn of batch.turns) { + messageCount += turn.messages.length; + if (messageCount <= submittedMessageCount) { + state.acknowledgedTurnFingerprints.add(sha256(JSON.stringify(turn))); + } + } + state.captureLedgerDirty = true; + await this.#persistCaptureLedger(sessionID, state); this.#activity.saved(); this.#deps.logger("[capture] conversation batch saved", { sessionID, @@ -1089,8 +1223,60 @@ export class V2Runtime { return false; } + async #loadCaptureLedger(sessionID: string, state: SessionState): Promise { + if (!this.#ctx.storage) return; + state.captureLedgerLoaded ??= (async () => { + const tags = await this.#resolveSession(sessionID); + const scope = sha256(JSON.stringify([ + getApiBaseUrl(), getApiVersion(), normalizeDirectory(state.directory!), + tags.canonical, sessionID, + ])); + state.captureLedgerKey = `capture-acks:v1:${scope}`; + const saved = await this.#ctx.storage.get(state.captureLedgerKey) as Record | undefined; + if (saved === undefined) return; + if (!saved || Array.isArray(saved) || typeof saved !== "object" || + saved.version !== 1 || saved.scope !== scope || + !Array.isArray(saved.captureIds) || !Array.isArray(saved.turnIds) || + !saved.captureIds.every((id) => typeof id === "string" && /^opencode:(capture|compaction):[a-f0-9]{64}:[a-f0-9]{64}$/.test(id)) || + !saved.turnIds.every((id) => typeof id === "string" && /^[a-f0-9]{64}$/.test(id))) { + throw new Error(`Invalid capture acknowledgement ledger for ${sessionID}`); + } + for (const id of saved.captureIds) state.acknowledgedCaptureIds.add(id as string); + for (const id of saved.turnIds) state.acknowledgedTurnFingerprints.add(id as string); + })(); + try { + await state.captureLedgerLoaded; + } catch (error) { + state.captureLedgerLoaded = undefined; + throw error; + } + } + + async #persistCaptureLedger(sessionID: string, state: SessionState): Promise { + if (!this.#ctx.storage) { + state.captureLedgerDirty = false; + return; + } + if (!state.captureLedgerDirty || !state.captureLedgerKey) return; + try { + await this.#ctx.storage.set(state.captureLedgerKey, { + version: 1, + scope: state.captureLedgerKey.slice("capture-acks:v1:".length), + captureIds: [...state.acknowledgedCaptureIds], + turnIds: [...state.acknowledgedTurnFingerprints], + }); + state.captureLedgerDirty = false; + } catch (error) { + this.#deps.logger("[capture] acknowledgement persistence failed; retaining in-memory ACKs", { + sessionID, error: String(error), + }); + } + } + async #captureCadence(sessionID: string): Promise { const state = this.#state(sessionID); + await this.#loadCaptureLedger(sessionID, state); + await this.#persistCaptureLedger(sessionID, state); const turns = await this.#refreshTurns(sessionID, state); for (const batch of buildCadenceBatches( turns, @@ -1100,47 +1286,59 @@ export class V2Runtime { } } - async #captureSessionEnd(sessionID: string): Promise { + async #captureSessionEnd(sessionID: string): Promise { const state = this.#state(sessionID); + await this.#loadCaptureLedger(sessionID, state); + await this.#persistCaptureLedger(sessionID, state); const turns = await this.#refreshTurns(sessionID, state); + let complete = true; for (const batch of buildCadenceBatches( turns, this.#deps.config.captureEveryNTurns, )) { - await this.#saveBatch(sessionID, state, batch, "cadence"); + if (!(await this.#saveBatch(sessionID, state, batch, "cadence"))) complete = false; } const finalBatch = buildSessionEndBatch( turns, this.#deps.config.captureEveryNTurns, ); if (finalBatch) { - await this.#saveBatch(sessionID, state, finalBatch, "session_end"); + if (!(await this.#saveBatch(sessionID, state, finalBatch, "session_end"))) complete = false; } + return complete && !state.captureLedgerDirty; } - async #flushAll(allowInactive = false): Promise { + async #flushAll(allowInactive = false): Promise> { const sessions = [...this.#states.keys()]; + const unfinished = new Set(sessions); const pending = new Set( [...this.#pendingSummaries.values()].map((item) => item.sessionID), ); await Promise.all([ ...sessions.map((sessionID) => this.#runCaptureExclusive(sessionID, () => - this.#captureSessionEnd(sessionID), + this.#captureSessionEnd(sessionID).then((complete) => { + if (complete) unfinished.delete(sessionID); + }), ), ), ...[...pending].map((sessionID) => this.#retryPendingSummaries(sessionID, allowInactive), ), ]); + for (const [sessionID, state] of this.#states) { + if (state.captureLedgerDirty) unfinished.add(sessionID); + } + for (const item of this.#pendingSummaries.values()) unfinished.add(item.sessionID); + return unfinished; } async #runCaptureExclusive( sessionID: string, - task: () => Promise, + task: () => Promise, ): Promise { const previous = this.#captureInFlight.get(sessionID) ?? Promise.resolve(); - const next = previous.catch(() => undefined).then(task); + const next = previous.catch(() => undefined).then(async () => { await task(); }); this.#captureInFlight.set(sessionID, next); try { await next; @@ -1156,7 +1354,18 @@ export class V2Runtime { } } - async #retryPendingSummaries( + #retryPendingSummaries(sessionID: string, allowInactive = false): Promise { + const previous = this.#summaryTasks.get(sessionID); + if (previous) return previous; + const task = this.#savePendingSummaries(sessionID, allowInactive); + this.#summaryTasks.set(sessionID, task); + void task.finally(() => { + if (this.#summaryTasks.get(sessionID) === task) this.#summaryTasks.delete(sessionID); + }).catch(() => undefined); + return task; + } + + async #savePendingSummaries( sessionID: string, allowInactive = false, ): Promise { @@ -1167,8 +1376,10 @@ export class V2Runtime { if (pending.length === 0) return; let tags: ResolvedTags; + const state = this.#state(sessionID); try { tags = await this.#resolveSession(sessionID); + await this.#loadCaptureLedger(sessionID, state); } catch (error) { this.#deps.logger("v2 compaction summary retry deferred", { sessionID, @@ -1178,6 +1389,11 @@ export class V2Runtime { } for (const summary of pending) { + const acknowledgedId = `${summary.customId}:${sha256(summary.text)}`; + if (state.acknowledgedCaptureIds.has(acknowledgedId)) { + this.#pendingSummaries.delete(summary.customId); + continue; + } if (this.#summaryInFlight.has(summary.customId)) continue; this.#summaryInFlight.add(summary.customId); try { @@ -1199,6 +1415,11 @@ export class V2Runtime { }, ); if (result.success) { + await this.#runCaptureExclusive(sessionID, async () => { + state.acknowledgedCaptureIds.add(acknowledgedId); + state.captureLedgerDirty = true; + await this.#persistCaptureLedger(sessionID, state); + }); this.#pendingSummaries.delete(summary.customId); this.#activity.saved(); this.#deps.logger("[compaction] summary saved as memory", { @@ -1255,41 +1476,45 @@ export class V2Runtime { } } -const OWNER_KEY = Symbol.for("opencode-supermemory.v2.owner"); +const OWNER_REGISTRY_KEY = Symbol.for("opencode-supermemory.v2.owners"); interface GlobalOwner { generation: number; - cleanup: () => void; + cleanup: () => Promise; + runtime?: V2Runtime; } -function ownerRegistry(): Record { - return globalThis as unknown as Record; +function ownerRegistry(): Map { + const global = globalThis as unknown as Record | undefined>; + return global[OWNER_REGISTRY_KEY] ??= new Map(); } -/** - * Starts the OpenCode 2 runtime. Only one instance is active per process, so a - * hot-reloaded plugin replaces (and cleans up) the previous generation. - */ +// Replacements finish disposal and capture before the same location can register again. export async function setupV2( ctx: V2Context, options?: Partial, -): Promise<() => void> { +): Promise<() => Promise> { const registry = ownerRegistry(); - const previous = registry[OWNER_KEY]; - previous?.cleanup(); + const directory = normalizeDirectory(ctx.location.directory); + const previous = registry.get(directory); const owner: GlobalOwner = { generation: (previous?.generation ?? 0) + 1, - cleanup: () => undefined, + cleanup: async () => undefined, }; - registry[OWNER_KEY] = owner; - - const runtime = new V2Runtime(ctx, options, () => registry[OWNER_KEY] === owner); - const cleanup = () => { - runtime.cleanup(); - if (registry[OWNER_KEY] === owner) delete registry[OWNER_KEY]; + registry.set(directory, owner); + + const runtime = new V2Runtime(ctx, options, () => registry.get(directory) === owner); + owner.runtime = runtime; + const predecessor = previous?.cleanup() ?? Promise.resolve(); + const cleanup = async () => { + await predecessor; + await runtime.cleanup(); + if (registry.get(directory) === owner && runtime.trackedSessionCount === 0) registry.delete(directory); }; owner.cleanup = cleanup; + await predecessor; + if (previous?.runtime) runtime.inheritCaptureState(previous.runtime.takeCaptureState()); log("v2 plugin init", { generation: owner.generation, @@ -1300,7 +1525,7 @@ export async function setupV2( try { await runtime.register(); } catch (error) { - cleanup(); + await cleanup(); throw error; } diff --git a/src/version.ts b/src/version.ts index 4ee3164..9051940 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1,3 +1,3 @@ // AUTO-GENERATED by scripts/sync-version.mjs — do not edit by hand. // Sourced from package.json "version" so the two cannot drift. -export const PLUGIN_VERSION = "2.0.17"; +export const PLUGIN_VERSION = "2.0.18"; From 54611b9b41552b02481e70588900c367aaeebf66 Mon Sep 17 00:00:00 2001 From: Dhravya Date: Mon, 12 Oct 2026 12:10:45 +0000 Subject: [PATCH 2/2] fix(v2): coordinate deleted-session summary cleanup Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- src/v2/runtime.ts | 82 ++++++++++++++++++++++++++--------------------- 1 file changed, 45 insertions(+), 37 deletions(-) diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index 553a16e..6ea976d 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -169,6 +169,7 @@ interface Injection { interface SessionState { directory?: string; + deleted?: boolean; tags?: ResolvedTags; resolving?: Promise; injectedInitialContext: boolean; @@ -193,6 +194,7 @@ interface CaptureHandoff { sessions: Array<{ sessionID: string; directory?: string; + deleted?: boolean; tags?: ResolvedTags; turns: CaptureTurn[]; captureIds: string[]; @@ -740,9 +742,9 @@ export class V2Runtime { if (sessionID && !(await this.#ownsSession(sessionID, event))) return; if (!this.active) return; - if (sessionID && event.type !== "session.compaction.ended") { - void this.#retryPendingSummaries(sessionID); - } + const summaryWork = sessionID && event.type !== "session.compaction.ended" + ? this.#retryPendingSummaries(sessionID) + : undefined; switch (event.type) { case "session.execution.succeeded": { @@ -764,17 +766,15 @@ export class V2Runtime { case "session.deleted": { if (!sessionID) return; + await summaryWork; const state = this.#states.get(sessionID); this.#recall.delete(sessionID); if (!state) return; - await this.#runCaptureExclusive(sessionID, async () => { - if (!(await this.#captureSessionEnd(sessionID))) return; - if (state.captureLedgerKey) { - await this.#ctx.storage.remove(state.captureLedgerKey); - } + state.deleted = true; + if (await this.#finishSession(sessionID)) { this.#states.delete(sessionID); this.#sessionOwners.delete(sessionID); - }); + } return; } @@ -858,6 +858,7 @@ export class V2Runtime { sessions: [...this.#states].map(([sessionID, state]) => ({ sessionID, directory: state.directory, + deleted: state.deleted, tags: state.tags, turns: state.turns, captureIds: [...state.acknowledgedCaptureIds], @@ -876,6 +877,7 @@ export class V2Runtime { for (const source of snapshot.sessions) { const state = this.#state(source.sessionID); state.directory = source.directory; + state.deleted = source.deleted; state.tags = source.tags; state.turns = source.turns; state.turnIndex = new Map(source.turns.map((turn, index) => [turn.id, index])); @@ -891,8 +893,7 @@ export class V2Runtime { const directory = event.location?.directory; if (directory) { const owned = normalizeDirectory(directory) === normalizeDirectory(this.#ctx.location.directory); - this.#sessionOwners.set(sessionID, owned); - return owned; + return this.#rememberSessionOwner(sessionID, owned); } const known = this.#sessionOwners.get(sessionID); if (known !== undefined) return known; @@ -903,11 +904,7 @@ export class V2Runtime { const session = await this.#ctx.session.get({ sessionID }); const owned = normalizeDirectory(session.location.directory) === normalizeDirectory(this.#ctx.location.directory); - this.#sessionOwners.set(sessionID, owned); - if (this.#sessionOwners.size > 1024) { - this.#sessionOwners.delete(this.#sessionOwners.keys().next().value!); - } - return owned; + return this.#rememberSessionOwner(sessionID, owned); } catch (error) { this.#deps.logger("v2 session owner lookup failed; skipping event", { sessionID, error: String(error), @@ -921,6 +918,15 @@ export class V2Runtime { return lookup; } + #rememberSessionOwner(sessionID: string, owned: boolean): boolean { + this.#sessionOwners.delete(sessionID); + this.#sessionOwners.set(sessionID, owned); + while (this.#sessionOwners.size > 1024) { + this.#sessionOwners.delete(this.#sessionOwners.keys().next().value!); + } + return owned; + } + #state(sessionID: string): SessionState { const existing = this.#states.get(sessionID); if (existing) return existing; @@ -980,7 +986,7 @@ export class V2Runtime { if (normalizeDirectory(directory) !== normalizeDirectory(this.#ctx.location.directory)) { throw new Error(`OpenCode session ${sessionID} belongs to another location`); } - this.#sessionOwners.set(sessionID, true); + this.#rememberSessionOwner(sessionID, true); state.directory = directory; state.tags = this.#deps.resolveTags(directory); return state.tags; @@ -1308,28 +1314,30 @@ export class V2Runtime { return complete && !state.captureLedgerDirty; } + async #finishSession(sessionID: string, allowInactive = false): Promise { + await this.#retryPendingSummaries(sessionID, allowInactive); + let complete = false; + await this.#runCaptureExclusive(sessionID, async () => { + if (!(await this.#captureSessionEnd(sessionID))) return; + if ([...this.#pendingSummaries.values()].some((item) => item.sessionID === sessionID)) return; + const state = this.#state(sessionID); + if (state.deleted && state.captureLedgerKey && this.#ctx.storage) { + await this.#ctx.storage.remove(state.captureLedgerKey); + } + complete = true; + }); + return complete; + } + async #flushAll(allowInactive = false): Promise> { - const sessions = [...this.#states.keys()]; - const unfinished = new Set(sessions); - const pending = new Set( - [...this.#pendingSummaries.values()].map((item) => item.sessionID), - ); - await Promise.all([ - ...sessions.map((sessionID) => - this.#runCaptureExclusive(sessionID, () => - this.#captureSessionEnd(sessionID).then((complete) => { - if (complete) unfinished.delete(sessionID); - }), - ), - ), - ...[...pending].map((sessionID) => - this.#retryPendingSummaries(sessionID, allowInactive), - ), + const sessions = new Set([ + ...this.#states.keys(), + ...[...this.#pendingSummaries.values()].map((item) => item.sessionID), ]); - for (const [sessionID, state] of this.#states) { - if (state.captureLedgerDirty) unfinished.add(sessionID); - } - for (const item of this.#pendingSummaries.values()) unfinished.add(item.sessionID); + const unfinished = new Set(sessions); + await Promise.all([...sessions].map(async (sessionID) => { + if (await this.#finishSession(sessionID, allowInactive)) unfinished.delete(sessionID); + })); return unfinished; }