diff --git a/README.md b/README.md index fc92dd4..cf372b1 100644 --- a/README.md +++ b/README.md @@ -171,7 +171,15 @@ Completed conversations are captured automatically: 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. +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. + +OpenCode 2 keeps the 100,000-character formatted request budget without truncating eligible capture text. Oversized batches are sent by turn; oversized turns are split into deterministic message fragments that preserve internal whitespace and Unicode code points. Each received fragment ACK is checkpointed independently, failed fragments remain retryable, and a turn is complete only after all its pieces are accepted. Ordinary batches retain their existing IDs and formatting. OpenCode V1's existing truncation behavior is unchanged. + +Fragment progress uses a version-two ledger. The version-one ledger from 2.0.18 is retained unchanged until explicit session deletion, and only its valid full-turn fingerprints and compaction ACKs are imported. That old format has no credential provenance, so importing its progress assumes an unchanged account at the first upgrade. An old truncated batch ACK is not proof of full coverage, so that first upgrade may repeat its previously accepted prefix while delivering the missing content. Downgrading to a runtime that cannot read fragment progress may also resend confirmed content; retained old records and host transcripts remain recoverable, not a no-replay downgrade guarantee. + +Before sending, OpenCode 2 checkpoints only the selected bounded payloads and their manifest in plugin-private host storage. This adds local retention of capture-policy-selected text, not new redaction or encryption. UnACKed payloads survive compaction, process restart and unsuccessful deletion flushes; recovery drains the stored selection when its original project plugin loads, without backfilling unrelated history. Payloads are garbage-collected only after their ACK progress is durably committed. There is no automatic expiry/drop cutoff or unlimited disk-capacity guarantee: a failed pending checkpoint defers network delivery, and a process exit before that checkpoint succeeds still depends on recoverable host originals. + +Pending records and new ACK ledgers bind the configured API destination, protocol, namespace, project, session and a one-way credential-context fingerprint. Changed or unrecognized provenance/schema fails closed and retains originals; even a same-account key rotation may require restoring the original context to recover pending content. Raw credentials and API URLs are not copied into these records. Late server acceptance before a received/durable ACK remains ambiguous and may be retried. Splitting oversized content adds write operations; production extraction, ordering and billing equivalence are not implied by complete text delivery. ### Keyword Detection diff --git a/package.json b/package.json index ff5dfe8..6cbf706 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode-supermemory", - "version": "2.0.18", + "version": "2.0.19", "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 7f54baf..1ab30d6 100644 --- a/src/services/client.ts +++ b/src/services/client.ts @@ -1,4 +1,5 @@ import Supermemory from "supermemory-legacy"; +import { createHash } from "node:crypto"; import { V5Client } from "./v5-client.js"; import { CONFIG, @@ -9,6 +10,11 @@ import { isConfigured, } from "../config.js"; import { log } from "./logger.js"; +import { + formatConversationContent, + formatConversationMessage, + MAX_CONVERSATION_CHARS, +} from "./conversation.js"; import { mergeListResponses, mergeProfileResponses, @@ -23,7 +29,6 @@ import type { const TIMEOUT_MS = 30000; const TIMEOUT_BACKSTOP_GRACE_MS = 250; const SETTINGS_UPDATE_TIMEOUT_MS = 3_000; -const MAX_CONVERSATION_CHARS = 100_000; const OPENCODE_SOURCE = "opencode"; export interface MemoryRequestOptions { @@ -128,35 +133,14 @@ function withTimeout(promise: Promise, ms: number): Promise { export class SupermemoryClient { private client: Supermemory | V5Client | null = null; + private captureDestination: { baseUrl: string; apiVersion: "legacy" | "v5"; account: string } | undefined; - private formatConversationMessage(message: ConversationMessage): string { - const content = - typeof message.content === "string" - ? message.content - : message.content - .map((part) => - part.type === "text" - ? part.text - : `[image] ${part.imageUrl.url}`, - ) - .join("\n"); - - const trimmed = content.trim(); - if (trimmed.length === 0) { - return `[${message.role}]`; - } - return `[${message.role}] ${trimmed}`; - } - - private formatConversationTranscript( - messages: ConversationMessage[], - ): string { - return messages - .map( - (message, idx) => - `${idx + 1}. ${this.formatConversationMessage(message)}`, - ) - .join("\n"); + getCaptureDestination(): { baseUrl: string; apiVersion: "legacy" | "v5"; account: string } { + const bound = this.captureDestination ??= { + baseUrl: getApiBaseUrl(), apiVersion: getApiVersion(), + account: createHash("sha256").update(SUPERMEMORY_API_KEY ?? "unconfigured").digest("hex"), + }; + return { ...bound }; } private getClient(): Supermemory | V5Client { @@ -164,16 +148,17 @@ export class SupermemoryClient { if (!isConfigured()) { throw new Error("SUPERMEMORY_API_KEY not set"); } + const destination = this.getCaptureDestination(); const legacyClient = new Supermemory({ apiKey: SUPERMEMORY_API_KEY, - baseURL: getApiBaseUrl(), + baseURL: destination.baseUrl, defaultHeaders: { "x-sm-source": OPENCODE_SOURCE }, }); this.client = - getApiVersion() === "v5" + destination.apiVersion === "v5" ? new V5Client( SUPERMEMORY_API_KEY!, - getApiBaseUrl(), + destination.baseUrl, legacyClient.settings, TIMEOUT_MS - TIMEOUT_BACKSTOP_GRACE_MS, ) @@ -644,6 +629,8 @@ export class SupermemoryClient { entityContextByContainerTag?: Record; customId?: string; timeoutMs?: number; + requireComplete?: boolean; + preserveWhitespace?: boolean; }, ) { log("ingestConversation: start", { @@ -666,8 +653,10 @@ export class SupermemoryClient { }; } - const transcript = this.formatConversationTranscript(messages); - const rawContent = `[Conversation ${conversationId}]\n${transcript}`; + const rawContent = formatConversationContent(conversationId, messages, options?.preserveWhitespace); + if (options?.requireComplete && rawContent.length > MAX_CONVERSATION_CHARS) { + return { success: false as const, error: "Conversation exceeds the complete-submission budget" }; + } const content = rawContent.length > MAX_CONVERSATION_CHARS ? `${rawContent.slice(0, MAX_CONVERSATION_CHARS)}\n...[truncated]` @@ -675,7 +664,7 @@ export class SupermemoryClient { let submittedMessageCount = 0; let submittedLength = `[Conversation ${conversationId}]\n`.length; for (const [index, message] of messages.entries()) { - submittedLength += `${index + 1}. ${this.formatConversationMessage(message)}`.length; + submittedLength += `${index + 1}. ${formatConversationMessage(message, options?.preserveWhitespace)}`.length; if (submittedLength > MAX_CONVERSATION_CHARS) break; submittedMessageCount += 1; submittedLength += 1; diff --git a/src/services/conversation.ts b/src/services/conversation.ts new file mode 100644 index 0000000..8ee939a --- /dev/null +++ b/src/services/conversation.ts @@ -0,0 +1,30 @@ +import type { ConversationMessage } from "../types/index.js"; + +export const MAX_CONVERSATION_CHARS = 100_000; + +export function conversationMessageText(message: ConversationMessage): string { + return (typeof message.content === "string" + ? message.content + : message.content.map((part) => part.type === "text" + ? part.text + : `[image] ${part.imageUrl.url}`).join("\n")).trim(); +} + +export function formatConversationMessage( + message: ConversationMessage, + preserveWhitespace = false, +): string { + const content = preserveWhitespace && typeof message.content === "string" + ? message.content + : conversationMessageText(message); + return content.length === 0 ? `[${message.role}]` : `[${message.role}] ${content}`; +} + +export function formatConversationContent( + conversationId: string, + messages: ConversationMessage[], + preserveWhitespace = false, +): string { + return `[Conversation ${conversationId}]\n${messages.map((message, index) => + `${index + 1}. ${formatConversationMessage(message, preserveWhitespace)}`).join("\n")}`; +} diff --git a/src/v2/capture-plan.ts b/src/v2/capture-plan.ts new file mode 100644 index 0000000..52f89bd --- /dev/null +++ b/src/v2/capture-plan.ts @@ -0,0 +1,106 @@ +import { createHash } from "node:crypto"; + +import { getCaptureId, type CaptureBatch, type CaptureTurn } from "../services/capture.js"; +import { + conversationMessageText, + formatConversationContent, + MAX_CONVERSATION_CHARS, +} from "../services/conversation.js"; +import type { ConversationMessage } from "../types/index.js"; + +export interface CapturePiece { + id: string; + acknowledgedId: string; + conversationId: string; + messages: ConversationMessage[]; + startTurn: number; + endTurn: number; + preserveWhitespace: boolean; + metadata: Record; +} + +export interface CaptureGroup { + turnFingerprints: string[]; + pieces: CapturePiece[]; +} + +function sha256(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + +function wholePiece(sessionID: string, batch: CaptureBatch): CapturePiece { + const messages = batch.turns.flatMap((turn) => turn.messages.map((message) => ({ + role: message.role, content: conversationMessageText(message), + }))); + const id = getCaptureId(sessionID, batch); + return { + id, + acknowledgedId: `${id}:${sha256(JSON.stringify(messages))}`, + conversationId: `${sessionID}:${batch.startTurn}-${batch.endTurn}`, + messages, + startTurn: batch.startTurn, + endTurn: batch.endTurn, + preserveWhitespace: false, + metadata: {}, + }; +} + +export function planCapture( + sessionID: string, + batch: CaptureBatch, + positions?: readonly number[], +): CaptureGroup[] { + const whole = wholePiece(sessionID, batch); + if (formatConversationContent(whole.conversationId, whole.messages).length <= MAX_CONVERSATION_CHARS) { + return [{ turnFingerprints: batch.turns.map((turn) => sha256(JSON.stringify(turn))), pieces: [whole] }]; + } + + return batch.turns.map((turn, turnIndex) => { + const position = positions?.[turnIndex] ?? batch.startTurn + turnIndex; + const single = wholePiece(sessionID, { startTurn: position, endTurn: position, turns: [turn] }); + if (formatConversationContent(single.conversationId, single.messages).length <= MAX_CONVERSATION_CHARS) { + return { turnFingerprints: [sha256(JSON.stringify(turn))], pieces: [single] }; + } + + const fingerprint = sha256(JSON.stringify(turn)); + const pieces: CapturePiece[] = []; + for (const [messageIndex, message] of turn.messages.entries()) { + const text = conversationMessageText(message); + let start = 0; + let chunkIndex = 0; + do { + const conversationId = `${sessionID}:${turn.id}:m${messageIndex}:c${chunkIndex}`; + const empty = { role: message.role, content: "" }; + const budget = MAX_CONVERSATION_CHARS - formatConversationContent(conversationId, [empty], true).length - 1; + if (budget < 2) throw new Error("Capture identity exceeds the fragment budget"); + let end = Math.min(start + budget, text.length); + if (end < text.length && + text.charCodeAt(end - 1) >= 0xd800 && text.charCodeAt(end - 1) <= 0xdbff && + text.charCodeAt(end) >= 0xdc00 && text.charCodeAt(end) <= 0xdfff) end -= 1; + const messages = [{ role: message.role, content: text.slice(start, end) }]; + const id = `opencode:capture:${sha256(JSON.stringify([sessionID, turn.id, fingerprint, messageIndex, chunkIndex]))}`; + pieces.push({ + id, + acknowledgedId: `${id}:${sha256(JSON.stringify(messages))}`, + conversationId, + messages, + startTurn: position, + endTurn: position, + preserveWhitespace: true, + metadata: { + captureFragment: true, + captureTurnId: turn.id, + captureTurnFingerprint: fingerprint, + captureMessageIndex: messageIndex, + captureChunkIndex: chunkIndex, + captureChunkStart: start, + captureChunkEnd: end, + }, + }); + start = end; + chunkIndex += 1; + } while (start < text.length); + } + return { turnFingerprints: [fingerprint], pieces }; + }); +} diff --git a/src/v2/capture-queue.ts b/src/v2/capture-queue.ts new file mode 100644 index 0000000..ca13e5e --- /dev/null +++ b/src/v2/capture-queue.ts @@ -0,0 +1,124 @@ +import { createHash } from "node:crypto"; + +import { formatConversationContent, MAX_CONVERSATION_CHARS } from "../services/conversation.js"; +import type { CaptureGroup, CapturePiece } from "./capture-plan.js"; + +export interface PendingCaptureGroup extends CaptureGroup { + id: string; + reason: "cadence" | "session_end"; + pieceIds: string[]; +} + +export interface CaptureQueue { + version: 1; + scope: string; + account: string; + sessionID: string; + deleted: boolean; + groups: PendingCaptureGroup[]; +} + +function sha256(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + +function object(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +function digest(value: unknown): value is string { + return typeof value === "string" && /^[a-f0-9]{64}$/.test(value); +} + +function acknowledgement(value: unknown): value is string { + return typeof value === "string" && /^opencode:capture:[a-f0-9]{64}:[a-f0-9]{64}$/.test(value); +} + +export function parseCaptureAcknowledgements( + value: unknown, + scope: string, + account: string, + version: 1 | 2, +): { captureIds: string[]; turnIds: string[] } | undefined { + if (value === undefined) return; + const keys = version === 2 ? ["version", "scope", "account", "captureIds", "turnIds"] : ["version", "scope", "captureIds", "turnIds"]; + if (!object(value) || value.version !== version || value.scope !== scope || + (version === 2 && value.account !== account) || + !Array.isArray(value.captureIds) || !Array.isArray(value.turnIds) || + !value.captureIds.every((id) => typeof id === "string" && /^opencode:(capture|compaction):[a-f0-9]{64}:[a-f0-9]{64}$/.test(id)) || + !value.turnIds.every(digest) || Object.keys(value).some((key) => !keys.includes(key))) { + throw new Error("Invalid or incompatible capture acknowledgements; originals retained"); + } + return value as unknown as { captureIds: string[]; turnIds: string[] }; +} + +export function queueCaptureGroup(group: CaptureGroup, reason: PendingCaptureGroup["reason"]): PendingCaptureGroup { + const pieceIds = group.pieces.map((piece) => piece.acknowledgedId); + return { ...group, reason, pieceIds, id: sha256(JSON.stringify([group.turnFingerprints, pieceIds])) }; +} + +export function parseCaptureQueue( + value: unknown, + scope: string, + account: string, + sessionID: string, + acknowledged: ReadonlySet, +): CaptureQueue { + const invalid = () => { throw new Error("Invalid or incompatible pending capture queue; originals retained"); }; + if (!/^[A-Za-z0-9_-]{1,1024}$/.test(sessionID)) return invalid(); + if (!object(value) || value.version !== 1 || value.scope !== scope || + value.account !== account || value.sessionID !== sessionID || + typeof value.deleted !== "boolean" || !Array.isArray(value.groups) || + Object.keys(value).some((key) => !["version", "scope", "account", "sessionID", "deleted", "groups"].includes(key))) return invalid(); + const ids = new Set(); + for (const group of value.groups) { + if (!object(group) || !digest(group.id) || ids.has(group.id) || + !["cadence", "session_end"].includes(String(group.reason)) || + !Array.isArray(group.turnFingerprints) || group.turnFingerprints.length === 0 || + !group.turnFingerprints.every(digest) || !Array.isArray(group.pieceIds) || group.pieceIds.length === 0 || + !group.pieceIds.every(acknowledgement) || new Set(group.pieceIds).size !== group.pieceIds.length || + !Array.isArray(group.pieces) || + group.id !== sha256(JSON.stringify([group.turnFingerprints, group.pieceIds])) || + Object.keys(group).some((key) => !["id", "reason", "turnFingerprints", "pieceIds", "pieces"].includes(key))) return invalid(); + ids.add(group.id); + const payloadIds = new Set(); + for (const piece of group.pieces) { + if (!object(piece) || typeof piece.id !== "string" || !/^opencode:capture:[a-f0-9]{64}$/.test(piece.id) || + !acknowledgement(piece.acknowledgedId) || !group.pieceIds.includes(piece.acknowledgedId) || payloadIds.has(piece.acknowledgedId) || + typeof piece.conversationId !== "string" || !piece.conversationId.startsWith(`${sessionID}:`) || + typeof piece.preserveWhitespace !== "boolean" || + !Number.isSafeInteger(piece.startTurn) || Number(piece.startTurn) < 1 || + !Number.isSafeInteger(piece.endTurn) || Number(piece.endTurn) < Number(piece.startTurn) || + !Array.isArray(piece.messages) || piece.messages.length === 0 || + !piece.messages.every((message) => object(message) && + ["user", "assistant", "system", "tool"].includes(String(message.role)) && typeof message.content === "string" && + Object.keys(message).every((key) => key === "role" || key === "content")) || + piece.acknowledgedId !== `${piece.id}:${sha256(JSON.stringify(piece.messages))}` || + !object(piece.metadata) || Object.entries(piece.metadata).some(([key, item]) => { + if (key === "captureFragment") return item !== true; + if (key === "captureTurnFingerprint") return !digest(item); + if (key === "captureTurnId") return typeof item !== "string" || item.length === 0 || item.length > 1024; + return !["captureMessageIndex", "captureChunkIndex", "captureChunkStart", "captureChunkEnd"].includes(key) || + !Number.isSafeInteger(item) || Number(item) < 0; + }) || + Object.keys(piece).some((key) => !["id", "acknowledgedId", "conversationId", "messages", "startTurn", "endTurn", "preserveWhitespace", "metadata"].includes(key))) return invalid(); + const typed = piece as unknown as CapturePiece; + const metadata = typed.metadata; + if (metadata.captureFragment === true) { + const keys = ["captureFragment", "captureTurnId", "captureTurnFingerprint", "captureMessageIndex", "captureChunkIndex", "captureChunkStart", "captureChunkEnd"]; + if (keys.some((key) => !(key in metadata)) || Object.keys(metadata).length !== keys.length || + !/^[A-Za-z0-9_-]{1,1024}$/.test(String(metadata.captureTurnId)) || + !group.turnFingerprints.includes(String(metadata.captureTurnFingerprint)) || + typed.messages.length !== 1 || !typed.preserveWhitespace || + typed.conversationId !== `${sessionID}:${metadata.captureTurnId}:m${metadata.captureMessageIndex}:c${metadata.captureChunkIndex}` || + Number(metadata.captureChunkEnd) - Number(metadata.captureChunkStart) !== typed.messages[0]!.content.length || + typed.id !== `opencode:capture:${sha256(JSON.stringify([sessionID, metadata.captureTurnId, metadata.captureTurnFingerprint, metadata.captureMessageIndex, metadata.captureChunkIndex]))}`) return invalid(); + } else if (Object.keys(metadata).length !== 0 || typed.preserveWhitespace || + !new RegExp(`^${sessionID}:[1-9][0-9]*-[1-9][0-9]*$`).test(typed.conversationId)) return invalid(); + if (formatConversationContent(typed.conversationId, typed.messages, typed.preserveWhitespace).length > MAX_CONVERSATION_CHARS) return invalid(); + payloadIds.add(piece.acknowledgedId); + } + if (!group.pieceIds.every((id) => acknowledged.has(id) || payloadIds.has(id))) return invalid(); + } + return JSON.parse(JSON.stringify(value)) as CaptureQueue; +} diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index 6ea976d..cdc8b2e 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -4,7 +4,7 @@ import { resolve } from "node:path"; import type { Plugin } from "@opencode/plugin"; import type { SessionHooks } from "@opencode/plugin/promise/session"; -import { CONFIG, getApiBaseUrl, getApiVersion, isConfigured, PLUGIN_VERSION } from "../config.js"; +import { CONFIG, getApiBaseUrl, getApiVersion, isConfigured, PLUGIN_VERSION, SUPERMEMORY_API_KEY } from "../config.js"; import { createMemoryActivityReporterFromSink, type MemoryActivityNotice, @@ -14,7 +14,6 @@ import { AUTOMATIC_CAPTURE_TIMEOUT_MS, buildCadenceBatches, buildSessionEndBatch, - getCaptureId, type CaptureBatch, type CaptureTurn, } from "../services/capture.js"; @@ -50,6 +49,8 @@ import { import { getTags, type ResolvedTags } from "../services/tags.js"; import { checkNpmUpdate, type UpdateInfo } from "../services/version-check.js"; import { SUPERMEMORY_RPC } from "../rpc.js"; +import { planCapture } from "./capture-plan.js"; +import { parseCaptureAcknowledgements, parseCaptureQueue, queueCaptureGroup, type PendingCaptureGroup } from "./capture-queue.js"; export type V2Context = Plugin.Context; type SessionContextEvent = SessionHooks["context"]; @@ -113,7 +114,7 @@ type RuntimeMemoryClient = Pick< | "searchMemoriesMany" | "listMemoriesScoped" | "deleteMemory" ->; +> & Partial>; type RuntimeConfig = Pick< typeof CONFIG, @@ -181,7 +182,12 @@ interface SessionState { acknowledgedTurnFingerprints: Set; captureLedgerLoaded?: Promise; captureLedgerKey?: string; + legacyCaptureLedgerKey?: string; captureLedgerDirty: boolean; + pendingCaptures: Map; + pendingCaptureKey?: string; + pendingCaptureAccount?: string; + pendingCaptureDirty: boolean; } interface PendingSummary { @@ -191,6 +197,8 @@ interface PendingSummary { } interface CaptureHandoff { + version?: 2; + account?: string; sessions: Array<{ sessionID: string; directory?: string; @@ -200,7 +208,12 @@ interface CaptureHandoff { captureIds: string[]; turnFingerprints: string[]; ledgerKey?: string; + legacyLedgerKey?: string; dirty: boolean; + pending?: PendingCaptureGroup[]; + pendingKey?: string; + pendingAccount?: string; + pendingDirty?: boolean; }>; summaries: PendingSummary[]; } @@ -474,10 +487,12 @@ export class V2Runtime { readonly #registrations: Registration[] = []; readonly #registrationTasks = new Set>(); readonly #abortController = new AbortController(); + readonly #captureAccount = sha256(SUPERMEMORY_API_KEY ?? "unconfigured"); #emitActivity: ((notice: MemoryActivityNotice) => void) | undefined; #activity: MemoryActivityReporter; #active = true; #cleanup: Promise | undefined; + #recoveryTask: Promise | undefined; constructor( ctx: V2Context, @@ -613,7 +628,14 @@ export class V2Runtime { }); } - if (this.active) this.#startEventSubscription(); + if (this.active) { + this.#startEventSubscription(); + const task = this.#restorePendingSessions(); + this.#recoveryTask = task; + void task.finally(() => { + if (this.#recoveryTask === task) this.#recoveryTask = undefined; + }).catch(() => logger("[capture] pending recovery deferred; originals retained")); + } logger("v2 plugin registered", { tools: [SUPERMEMORY_TOOL_NAME, SUPERMEMORY_RECALL_TOOL_NAME], hooks: ["session.context", "session.compaction", "permission.evaluate", "tool.execute"], @@ -771,6 +793,7 @@ export class V2Runtime { this.#recall.delete(sessionID); if (!state) return; state.deleted = true; + if (state.pendingCaptures.size > 0) state.pendingCaptureDirty = true; if (await this.#finishSession(sessionID)) { this.#states.delete(sessionID); this.#sessionOwners.delete(sessionID); @@ -810,8 +833,11 @@ export class V2Runtime { /** Waits for background capture and summary work; used by tests and shutdown. */ async idle(): Promise { - while (this.#captureInFlight.size > 0 || this.#summaryTasks.size > 0) { - await Promise.allSettled([...this.#captureInFlight.values(), ...this.#summaryTasks.values()]); + while (this.#captureInFlight.size > 0 || this.#summaryTasks.size > 0 || this.#recoveryTask) { + await Promise.allSettled([ + ...this.#captureInFlight.values(), ...this.#summaryTasks.values(), + ...(this.#recoveryTask ? [this.#recoveryTask] : []), + ]); } } @@ -855,6 +881,8 @@ export class V2Runtime { takeCaptureState(): CaptureHandoff { const snapshot: CaptureHandoff = { + version: 2, + account: this.#captureAccount, sessions: [...this.#states].map(([sessionID, state]) => ({ sessionID, directory: state.directory, @@ -864,7 +892,12 @@ export class V2Runtime { captureIds: [...state.acknowledgedCaptureIds], turnFingerprints: [...state.acknowledgedTurnFingerprints], ledgerKey: state.captureLedgerKey, + legacyLedgerKey: state.legacyCaptureLedgerKey, dirty: state.captureLedgerDirty, + pending: [...state.pendingCaptures.values()], + pendingKey: state.pendingCaptureKey, + pendingAccount: state.pendingCaptureAccount, + pendingDirty: state.pendingCaptureDirty, })), summaries: [...this.#pendingSummaries.values()], }; @@ -881,10 +914,16 @@ export class V2Runtime { 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.acknowledgedCaptureIds = new Set(source.captureIds.filter((id) => + snapshot.version === 2 || id.startsWith("opencode:compaction:"))); state.acknowledgedTurnFingerprints = new Set(source.turnFingerprints); - state.captureLedgerKey = source.ledgerKey; + state.captureLedgerKey = snapshot.version === 2 ? source.ledgerKey : undefined; + state.legacyCaptureLedgerKey = snapshot.version === 2 ? source.legacyLedgerKey : source.ledgerKey; state.captureLedgerDirty = source.dirty; + state.pendingCaptures = new Map((source.pending ?? []).map((group) => [group.id, group])); + state.pendingCaptureKey = source.pendingKey; + state.pendingCaptureAccount = source.pendingAccount ?? snapshot.account; + state.pendingCaptureDirty = source.pendingDirty ?? false; } for (const summary of snapshot.summaries) this.#pendingSummaries.set(summary.customId, summary); } @@ -939,6 +978,8 @@ export class V2Runtime { acknowledgedCaptureIds: new Set(), acknowledgedTurnFingerprints: new Set(), captureLedgerDirty: false, + pendingCaptures: new Map(), + pendingCaptureDirty: false, }; this.#states.set(sessionID, state); return state; @@ -1148,107 +1189,154 @@ export class V2Runtime { return state.turns; } - async #saveBatch( + #queueBatch( sessionID: string, state: SessionState, 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; + ): void { + const queued = new Set([...state.pendingCaptures.values()].flatMap((group) => group.turnFingerprints)); + const remaining = batch.turns.filter((turn) => { + const fingerprint = sha256(JSON.stringify(turn)); + return turn.messages.length > 0 && !state.acknowledgedTurnFingerprints.has(fingerprint) && !queued.has(fingerprint); + }); + if (remaining.length === 0) return; + const positions = remaining.map((turn) => batch.startTurn + batch.turns.indexOf(turn)); 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); - const messages = batch.turns.flatMap((turn) => turn.messages); - const acknowledgedId = `${captureId}:${sha256(JSON.stringify(messages))}`; - if (state.acknowledgedCaptureIds.has(acknowledgedId)) return true; - - let result: { success: boolean; error?: string; submittedMessageCount?: number }; - try { - const tags = await this.#resolveSession(sessionID); - result = await this.#deps.memoryClient.ingestConversation( - `${sessionID}:${batch.startTurn}-${batch.endTurn}`, - messages, - [tags.canonical], - { - project: tags.projectName, - sm_project_id: tags.projectId, - sm_scope: "personal", - sm_capture_mode: "automatic", - captureReason: reason, - sessionId: sessionID, - turnStart: batch.startTurn, - turnEnd: batch.endTurn, - }, - { - defaultEntityContext: AGENT_ENTITY_CONTEXT, - customId: captureId, - timeoutMs: AUTOMATIC_CAPTURE_TIMEOUT_MS, - }, - ); - } catch (error) { - result = { - success: false, - error: error instanceof Error ? error.message : String(error), - }; + const groups = planCapture(sessionID, batch, positions); + for (const planned of groups) { + const group = queueCaptureGroup(planned, reason); + if (!state.pendingCaptures.has(group.id)) { + state.pendingCaptures.set(group.id, group); + state.pendingCaptureAccount = this.#captureAccount; + state.pendingCaptureDirty = true; + } } + } - if (result.success) { - 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))); + async #drainCaptureQueue(sessionID: string, state: SessionState): Promise { + await this.#persistCaptureLedger(sessionID, state); + if (!(await this.#persistPendingCaptures(sessionID, state))) return false; + const tags = await this.#resolveSession(sessionID); + let complete = true; + for (const group of [...state.pendingCaptures.values()]) { + const reason = group.reason; + for (const piece of group.pieces) { + if (state.acknowledgedCaptureIds.has(piece.acknowledgedId)) continue; + await this.#persistCaptureLedger(sessionID, state); + if (!(await this.#persistPendingCaptures(sessionID, state))) return false; + if (state.acknowledgedCaptureIds.has(piece.acknowledgedId)) continue; + let result: { success: boolean; error?: string; submittedMessageCount?: number }; + try { + this.#captureDestination(); + result = await this.#deps.memoryClient.ingestConversation( + piece.conversationId, + piece.messages, + [tags.canonical], + { + project: tags.projectName, + sm_project_id: tags.projectId, + sm_scope: "personal", + sm_capture_mode: "automatic", + captureReason: reason, + sessionId: sessionID, + turnStart: piece.startTurn, + turnEnd: piece.endTurn, + ...piece.metadata, + }, + { + defaultEntityContext: AGENT_ENTITY_CONTEXT, + customId: piece.id, + timeoutMs: AUTOMATIC_CAPTURE_TIMEOUT_MS, + requireComplete: true, + preserveWhitespace: piece.preserveWhitespace, + }, + ); + } catch (error) { + result = { success: false, error: error instanceof Error ? error.message : String(error) }; + } + if (result.success && + (result.submittedMessageCount ?? piece.messages.length) === piece.messages.length) { + state.acknowledgedCaptureIds.add(piece.acknowledgedId); + state.captureLedgerDirty = true; + await this.#persistCaptureLedger(sessionID, state); + this.#deps.logger("[capture] conversation piece saved", { + sessionID, reason, captureId: piece.id, + startTurn: piece.startTurn, endTurn: piece.endTurn, + }); + } else { + this.#deps.logger("[capture] failed to save complete conversation piece", { + sessionID, reason, captureId: piece.id, + }); } } - state.captureLedgerDirty = true; await this.#persistCaptureLedger(sessionID, state); - this.#activity.saved(); - this.#deps.logger("[capture] conversation batch saved", { - sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, - }); - return true; + if (!state.captureLedgerDirty) { + const pending = group.pieces.filter((piece) => !state.acknowledgedCaptureIds.has(piece.acknowledgedId)); + if (pending.length !== group.pieces.length) { + group.pieces = pending; + state.pendingCaptureDirty = true; + if (!(await this.#persistPendingCaptures(sessionID, state))) return false; + } + } + if (group.pieceIds.every((id) => state.acknowledgedCaptureIds.has(id))) { + const newlyComplete = group.turnFingerprints.some((fingerprint) => !state.acknowledgedTurnFingerprints.has(fingerprint)); + for (const fingerprint of group.turnFingerprints) state.acknowledgedTurnFingerprints.add(fingerprint); + state.captureLedgerDirty = true; + await this.#persistCaptureLedger(sessionID, state); + if (newlyComplete) this.#activity.saved(); + if (!state.captureLedgerDirty) { + state.pendingCaptures.delete(group.id); + state.pendingCaptureDirty = true; + if (!(await this.#persistPendingCaptures(sessionID, state))) return false; + } + } else { + complete = false; + } } - - this.#deps.logger("[capture] failed to save conversation batch", { - sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, - error: result.error, - }); - return false; + return complete && !state.captureLedgerDirty && !state.pendingCaptureDirty; } async #loadCaptureLedger(sessionID: string, state: SessionState): Promise { + const destination = this.#captureDestination(); if (!this.#ctx.storage) return; state.captureLedgerLoaded ??= (async () => { const tags = await this.#resolveSession(sessionID); + if (tags.canonical !== this.#deps.resolveTags(this.#ctx.location.directory).canonical) { + throw new Error("Capture namespace context changed; pending originals retained"); + } const scope = sha256(JSON.stringify([ - getApiBaseUrl(), getApiVersion(), normalizeDirectory(state.directory!), + destination.baseUrl, destination.apiVersion, 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}`); + const previousKeys = [state.captureLedgerKey, state.legacyCaptureLedgerKey]; + const hasProgress = state.pendingCaptures.size > 0 || state.acknowledgedCaptureIds.size > 0 || + state.acknowledgedTurnFingerprints.size > 0 || state.captureLedgerDirty || state.pendingCaptureDirty || + [...this.#pendingSummaries.values()].some((summary) => summary.sessionID === sessionID); + if (hasProgress && + ((state.pendingCaptureAccount !== undefined && state.pendingCaptureAccount !== this.#captureAccount) || + previousKeys.some((key) => key !== undefined && key !== `capture-acks:v2:${scope}` && key !== `capture-acks:v1:${scope}`))) { + throw new Error("Capture progress destination or credential context changed; originals retained"); + } + state.captureLedgerKey = `capture-acks:v2:${scope}`; + state.legacyCaptureLedgerKey = `capture-acks:v1:${scope}`; + const read = async (key: string, version: 1 | 2) => parseCaptureAcknowledgements( + await this.#ctx.storage.get(key), scope, this.#captureAccount, version, + ); + const current = await read(state.captureLedgerKey, 2); + const saved = current ?? await read(state.legacyCaptureLedgerKey, 1); + if (saved) { + for (const id of saved.captureIds) { + if (current || id.startsWith("opencode:compaction:")) state.acknowledgedCaptureIds.add(id); + } + for (const id of saved.turnIds) state.acknowledgedTurnFingerprints.add(id); + if (!current) state.captureLedgerDirty = true; } - for (const id of saved.captureIds) state.acknowledgedCaptureIds.add(id as string); - for (const id of saved.turnIds) state.acknowledgedTurnFingerprints.add(id as string); + await this.#loadPendingCaptures(sessionID, state, scope); })(); try { await state.captureLedgerLoaded; @@ -1263,22 +1351,145 @@ export class V2Runtime { state.captureLedgerDirty = false; return; } - if (!state.captureLedgerDirty || !state.captureLedgerKey) return; + if (!state.captureLedgerKey) return; + const scope = state.captureLedgerKey.slice("capture-acks:v2:".length); + const saved = parseCaptureAcknowledgements(await this.#ctx.storage.get(state.captureLedgerKey), scope, this.#captureAccount, 2); + const captureIds = new Set(saved?.captureIds ?? []); + const turnIds = new Set(saved?.turnIds ?? []); + if ([...state.acknowledgedCaptureIds].some((id) => !captureIds.has(id)) || + [...state.acknowledgedTurnFingerprints].some((id) => !turnIds.has(id))) state.captureLedgerDirty = true; + for (const id of captureIds) state.acknowledgedCaptureIds.add(id); + for (const id of turnIds) state.acknowledgedTurnFingerprints.add(id); + if (!state.captureLedgerDirty) return; try { await this.#ctx.storage.set(state.captureLedgerKey, { - version: 1, - scope: state.captureLedgerKey.slice("capture-acks:v1:".length), + version: 2, + account: this.#captureAccount, + scope, captureIds: [...state.acknowledgedCaptureIds], turnIds: [...state.acknowledgedTurnFingerprints], }); state.captureLedgerDirty = false; - } catch (error) { + } catch { this.#deps.logger("[capture] acknowledgement persistence failed; retaining in-memory ACKs", { - sessionID, error: String(error), + sessionID, + }); + } + } + + #pendingProjectPrefix(): string { + return `capture-pending:v1:${sha256(normalizeDirectory(this.#ctx.location.directory))}:`; + } + + #captureDestination(): { baseUrl: string; apiVersion: "legacy" | "v5" } { + const current = { baseUrl: getApiBaseUrl(), apiVersion: getApiVersion(), account: this.#captureAccount }; + if (this.#ctx.storage && !this.#deps.memoryClient.getCaptureDestination) { + throw new Error("Capture transport provenance unavailable; originals retained"); + } + const bound = this.#deps.memoryClient.getCaptureDestination?.() ?? current; + if (current.baseUrl !== bound.baseUrl || current.apiVersion !== bound.apiVersion || current.account !== bound.account) { + throw new Error("Capture destination changed; reinitialize the client before retrying; originals retained"); + } + return bound; + } + + async #loadPendingCaptures(sessionID: string, state: SessionState, scope: string): Promise { + const prefix = `${this.#pendingProjectPrefix()}${sha256(sessionID)}:`; + const key = `${prefix}${scope}`; + if (state.pendingCaptures.size > 0 && + ((state.pendingCaptureKey !== undefined && state.pendingCaptureKey !== key) || + (state.pendingCaptureAccount !== undefined && state.pendingCaptureAccount !== this.#captureAccount))) { + throw new Error("Pending capture destination or credential context changed; originals retained"); + } + state.pendingCaptureKey = key; + state.pendingCaptureAccount = this.#captureAccount; + let after: string | undefined; + do { + const page = await this.#ctx.storage.scan({ prefix, after }); + for (const entry of page.entries) { + if (entry.key !== key) throw new Error("Pending capture destination changed; originals retained"); + const saved = parseCaptureQueue(entry.value, scope, this.#captureAccount, sessionID, state.acknowledgedCaptureIds); + state.deleted ||= saved.deleted; + for (const group of saved.groups) { + if (!state.pendingCaptures.has(group.id)) state.pendingCaptures.set(group.id, group); + } + } + after = page.next; + } while (after); + } + + async #persistPendingCaptures(sessionID: string, state: SessionState): Promise { + if (!this.#ctx.storage) { + state.pendingCaptureDirty = false; + return true; + } + if (!state.pendingCaptureKey || !state.captureLedgerKey) return false; + try { + const scope = state.captureLedgerKey.slice("capture-acks:v2:".length); + const progress = parseCaptureAcknowledgements(await this.#ctx.storage.get(state.captureLedgerKey), scope, this.#captureAccount, 2); + const durableIds = new Set(progress?.captureIds ?? []); + const durableTurns = new Set(progress?.turnIds ?? []); + const original = await this.#ctx.storage.get(state.pendingCaptureKey); + if (original !== undefined) { + const saved = parseCaptureQueue(original, scope, this.#captureAccount, sessionID, durableIds); + if (saved.groups.some((group) => !state.pendingCaptures.has(group.id) && + (!group.pieceIds.every((id) => durableIds.has(id)) || + !group.turnFingerprints.every((id) => durableTurns.has(id))))) { + throw new Error("Pending group completion is not durably committed; originals retained"); + } + } + if (!state.pendingCaptureDirty) return true; + if (state.pendingCaptures.size === 0) { + await this.#ctx.storage.remove(state.pendingCaptureKey); + } else { + const value = { + version: 1, + scope, + account: this.#captureAccount, + sessionID, + deleted: state.deleted === true, + groups: [...state.pendingCaptures.values()], + }; + parseCaptureQueue(value, value.scope, value.account, sessionID, durableIds); + await this.#ctx.storage.set(state.pendingCaptureKey, JSON.parse(JSON.stringify(value))); + } + state.pendingCaptureDirty = false; + return true; + } catch { + this.#deps.logger("[capture] pending checkpoint failed; retaining selected content", { + sessionID, }); + return false; } } + async #restorePendingSessions(): Promise { + if (!this.#ctx.storage) return; + const prefix = this.#pendingProjectPrefix(); + let after: string | undefined; + do { + const page = await this.#ctx.storage.scan({ prefix, after }); + for (const entry of page.entries) { + if (!this.active) return; + try { + const value = entry.value as Record | null; + if (!value || typeof value.sessionID !== "string" || value.sessionID.length === 0 || value.sessionID.length > 1024 || + !entry.key.startsWith(`${prefix}${sha256(value.sessionID)}:`)) { + throw new Error("Invalid pending capture identity; originals retained"); + } + const sessionID = value.sessionID; + const state = this.#state(sessionID); + state.directory ??= this.#ctx.location.directory; + state.tags ??= this.#deps.resolveTags(this.#ctx.location.directory); + await this.#finishSession(sessionID, { queuedOnly: true, forgetWhenComplete: true }); + } catch { + this.#deps.logger("[capture] stored pending recovery deferred; originals retained"); + } + } + after = page.next; + } while (after); + } + async #captureCadence(sessionID: string): Promise { const state = this.#state(sessionID); await this.#loadCaptureLedger(sessionID, state); @@ -1288,8 +1499,9 @@ export class V2Runtime { turns, this.#deps.config.captureEveryNTurns, )) { - await this.#saveBatch(sessionID, state, batch, "cadence"); + this.#queueBatch(sessionID, state, batch, "cadence"); } + await this.#drainCaptureQueue(sessionID, state); } async #captureSessionEnd(sessionID: string): Promise { @@ -1297,34 +1509,53 @@ export class V2Runtime { 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, )) { - if (!(await this.#saveBatch(sessionID, state, batch, "cadence"))) complete = false; + this.#queueBatch(sessionID, state, batch, "cadence"); } const finalBatch = buildSessionEndBatch( turns, this.#deps.config.captureEveryNTurns, ); if (finalBatch) { - if (!(await this.#saveBatch(sessionID, state, finalBatch, "session_end"))) complete = false; + this.#queueBatch(sessionID, state, finalBatch, "session_end"); } - return complete && !state.captureLedgerDirty; + const complete = await this.#drainCaptureQueue(sessionID, state); + return complete && !state.captureLedgerDirty && !state.pendingCaptureDirty && state.pendingCaptures.size === 0; } - async #finishSession(sessionID: string, allowInactive = false): Promise { - await this.#retryPendingSummaries(sessionID, allowInactive); + async #finishSession( + sessionID: string, + options: { allowInactive?: boolean; queuedOnly?: boolean; forgetWhenComplete?: boolean } = {}, + ): Promise { + await this.#retryPendingSummaries(sessionID, options.allowInactive); let complete = false; await this.#runCaptureExclusive(sessionID, async () => { - if (!(await this.#captureSessionEnd(sessionID))) return; + if (options.queuedOnly) { + const state = this.#state(sessionID); + await this.#loadCaptureLedger(sessionID, state); + await this.#persistCaptureLedger(sessionID, state); + if (!(await this.#drainCaptureQueue(sessionID, state))) return; + } else 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) { + const scope = state.captureLedgerKey.slice("capture-acks:v2:".length); + parseCaptureAcknowledgements(await this.#ctx.storage.get(state.captureLedgerKey), scope, this.#captureAccount, 2); + if (state.legacyCaptureLedgerKey) { + parseCaptureAcknowledgements(await this.#ctx.storage.get(state.legacyCaptureLedgerKey), scope, this.#captureAccount, 1); + } await this.#ctx.storage.remove(state.captureLedgerKey); + if (state.legacyCaptureLedgerKey) await this.#ctx.storage.remove(state.legacyCaptureLedgerKey); } complete = true; + if (options.forgetWhenComplete && this.#states.get(sessionID) === state && + state.pendingCaptures.size === 0 && !state.captureLedgerDirty && !state.pendingCaptureDirty) { + this.#states.delete(sessionID); + this.#recall.delete(sessionID); + } }); return complete; } @@ -1336,7 +1567,7 @@ export class V2Runtime { ]); const unfinished = new Set(sessions); await Promise.all([...sessions].map(async (sessionID) => { - if (await this.#finishSession(sessionID, allowInactive)) unfinished.delete(sessionID); + if (await this.#finishSession(sessionID, { allowInactive })) unfinished.delete(sessionID); })); return unfinished; } @@ -1350,10 +1581,9 @@ export class V2Runtime { this.#captureInFlight.set(sessionID, next); try { await next; - } catch (error) { + } catch { this.#deps.logger("[capture] background capture failed", { sessionID, - error: String(error), }); } finally { if (this.#captureInFlight.get(sessionID) === next) { diff --git a/src/version.ts b/src/version.ts index 9051940..7e0ff38 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.18"; +export const PLUGIN_VERSION = "2.0.19";