diff --git a/src/services/capture.test.ts b/src/services/capture.test.ts deleted file mode 100644 index 8104e01..0000000 --- a/src/services/capture.test.ts +++ /dev/null @@ -1,287 +0,0 @@ -import { describe, expect, test } from "bun:test"; -import type { Part } from "@opencode-ai/sdk"; - -import { - AUTOMATIC_CAPTURE_TIMEOUT_MS, - buildCadenceBatches, - buildCaptureTurns, - buildSessionEndBatch, - createCaptureHook, - getCaptureId, - type SessionMessage, -} from "./capture.js"; -import { SupermemoryClient } from "./client.js"; -import type { ResolvedTags } from "./tags.js"; - -function textPart( - messageID: string, - text: string, - synthetic = false, -): Part { - return { - id: `part-${messageID}`, - sessionID: "session-1", - messageID, - type: "text", - text, - synthetic, - }; -} - -function user(id: string, text: string): SessionMessage { - return { - info: { id, role: "user", sessionID: "session-1" }, - parts: [textPart(id, text)], - }; -} - -function assistant(id: string, text: string): SessionMessage { - return { - info: { - id, - role: "assistant", - sessionID: "session-1", - finish: "stop", - }, - parts: [textPart(id, text)], - }; -} - -function conversation(turnCount: number): SessionMessage[] { - return Array.from({ length: turnCount }, (_, index) => { - const turn = index + 1; - return [ - user(`user-${turn}`, `question ${turn}`), - assistant(`assistant-${turn}`, `answer ${turn}`), - ]; - }).flat(); -} - -describe("automatic conversation capture", () => { - test("builds fixed cadence batches and a final remainder", () => { - const turns = buildCaptureTurns(conversation(7)); - const cadence = buildCadenceBatches(turns, 3); - const sessionEnd = buildSessionEndBatch(turns, 3); - - expect(cadence.map((batch) => [batch.startTurn, batch.endTurn])).toEqual([ - [1, 3], - [4, 6], - ]); - expect(sessionEnd && [sessionEnd.startTurn, sessionEnd.endTurn]).toEqual([ - 7, 7, - ]); - }); - - test("uses a stable capture ID within the API length limit", () => { - const batch = buildCadenceBatches( - [ - { - id: `msg_${"a".repeat(48)}`, - messages: [{ role: "user", content: "test" }], - }, - ], - 1, - )[0]!; - const sessionID = `ses_${"b".repeat(48)}`; - const captureId = getCaptureId(sessionID, batch); - - expect(captureId).toBe(getCaptureId(sessionID, batch)); - expect(captureId.length).toBeLessThanOrEqual(100); - expect(captureId).not.toContain(sessionID); - }); - - test("excludes synthetic context and protects private turns", () => { - const messages = [ - { - ...user("user-1", "real prompt"), - parts: [ - textPart("user-1", "injected recall", true), - textPart("user-1", "real prompt"), - ], - }, - assistant("assistant-1", "real answer"), - user("user-2", "secret prompt"), - assistant("assistant-2", "secret-derived answer"), - user("user-3", "token secret"), - assistant("assistant-3", "safe answer"), - ]; - - const turns = buildCaptureTurns(messages); - expect(turns[0]?.messages).toEqual([ - { role: "user", content: "real prompt" }, - { role: "assistant", content: "real answer" }, - ]); - expect(turns[1]?.messages).toEqual([]); - expect(turns[2]?.messages[0]).toEqual({ - role: "user", - content: "token [REDACTED]", - }); - }); - - test("captures on the third idle and flushes the final remainder once", async () => { - let messages = conversation(1); - const writes: Array<{ - conversationId: string; - metadata?: Record; - customId?: string; - timeoutMs?: number; - }> = []; - const ctx = { - directory: "/repo", - client: { - session: { - messages: async () => ({ data: messages }), - }, - }, - }; - const tags: ResolvedTags = { - canonical: "repo_test__0123456789abcdef", - user: "repo_test__0123456789abcdef", - project: "repo_test__0123456789abcdef", - projectId: "0123456789abcdef", - projectName: "test", - personalReads: [], - projectReads: [], - allReads: [], - }; - const hook = createCaptureHook(ctx, tags, { - captureEveryNTurns: 3, - memoryClient: { - ingestConversation: async ( - conversationId, - _conversationMessages, - _containerTags, - metadata, - options, - ) => { - writes.push({ - conversationId, - metadata, - customId: options?.customId, - timeoutMs: options?.timeoutMs, - }); - return { success: true }; - }, - }, - }); - - await hook.event({ - event: { - type: "session.idle", - properties: { sessionID: "session-1" }, - }, - }); - messages = conversation(3); - await hook.event({ - event: { - type: "session.idle", - properties: { sessionID: "session-1" }, - }, - }); - await hook.event({ - event: { - type: "session.idle", - properties: { sessionID: "session-1" }, - }, - }); - - expect(writes).toHaveLength(1); - expect(writes[0]?.metadata?.captureReason).toBe("cadence"); - expect(writes[0]?.timeoutMs).toBe(AUTOMATIC_CAPTURE_TIMEOUT_MS); - - messages = conversation(4); - await hook.event({ - event: { - type: "session.idle", - properties: { sessionID: "session-1" }, - }, - }); - await hook.event({ - event: { - type: "session.deleted", - properties: { info: { id: "session-1" } }, - }, - }); - await hook.event({ - event: { - type: "session.deleted", - properties: { info: { id: "session-1" } }, - }, - }); - - expect(writes).toHaveLength(2); - expect(writes[1]?.metadata?.captureReason).toBe("session_end"); - expect(writes[0]?.customId).not.toBe(writes[1]?.customId); - }); - - test("retains terminal capture state after failure and retries with bounded SDK options", async () => { - const sdkOptions: Array<{ timeout?: number; maxRetries?: number }> = []; - let attempts = 0; - let readAttempts = 0; - const memoryClient = new SupermemoryClient(); - ( - memoryClient as unknown as { - client: { - memories: { - add: ( - payload: unknown, - options?: { timeout?: number; maxRetries?: number }, - ) => Promise<{ id: string }>; - }; - }; - } - ).client = { - memories: { - add: async (_payload, options) => { - sdkOptions.push(options ?? {}); - attempts += 1; - if (attempts === 1) throw new Error("temporary capture failure"); - return { id: "memory-1" }; - }, - }, - }; - - const hook = createCaptureHook( - { - directory: "/repo", - client: { - session: { - messages: async () => { - readAttempts += 1; - if (readAttempts === 1) { - throw new Error("temporary transcript read failure"); - } - return { data: conversation(1) }; - }, - }, - }, - }, - { - canonical: "repo_test__0123456789abcdef", - user: "repo_test__0123456789abcdef", - project: "repo_test__0123456789abcdef", - projectId: "0123456789abcdef", - projectName: "test", - personalReads: [], - projectReads: [], - allReads: [], - }, - { captureEveryNTurns: 0, memoryClient }, - ); - - for (let attempt = 0; attempt < 4; attempt += 1) { - await hook.event({ - event: { - type: "session.deleted", - properties: { info: { id: "session-1" } }, - }, - }); - } - - expect(sdkOptions).toEqual([ - { timeout: AUTOMATIC_CAPTURE_TIMEOUT_MS, maxRetries: 0 }, - { timeout: AUTOMATIC_CAPTURE_TIMEOUT_MS, maxRetries: 0 }, - ]); - expect(readAttempts).toBe(3); - }); -}); diff --git a/src/services/capture.ts b/src/services/capture.ts index aaa2af4..3298f10 100644 --- a/src/services/capture.ts +++ b/src/services/capture.ts @@ -1,15 +1,15 @@ -import { createHash } from "node:crypto"; import type { Part } from "@opencode-ai/sdk"; -import { CONFIG } from "../config.js"; -import type { ConversationMessage } from "../types/index.js"; import { supermemoryClient } from "./client.js"; import { AGENT_ENTITY_CONTEXT } from "./entity-context.js"; import { log } from "./logger.js"; -import { isFullyPrivate, stripPrivateContent } from "./privacy.js"; import type { ResolvedTags } from "./tags.js"; - -export const AUTOMATIC_CAPTURE_TIMEOUT_MS = 3_000; +import { + captureSession, + SESSION_CAPTURE_TIMEOUT_MS, + sessionMetadata, + type SessionWriter, +} from "./session.js"; interface CaptureMessageInfo { id: string; @@ -24,17 +24,6 @@ export interface SessionMessage { parts?: Part[]; } -export interface CaptureTurn { - id: string; - messages: ConversationMessage[]; -} - -export interface CaptureBatch { - startTurn: number; - endTurn: number; - turns: CaptureTurn[]; -} - interface CaptureContext { directory: string; client: { @@ -49,180 +38,21 @@ interface CaptureContext { }; } -interface ConversationWriter { - ingestConversation: ( - conversationId: string, - messages: ConversationMessage[], - containerTags: string[], - metadata?: Record, - options?: { - defaultEntityContext?: string; - customId?: string; - timeoutMs?: number; - }, - ) => Promise<{ success: boolean; error?: string }>; -} - export interface CaptureOptions { - captureEveryNTurns?: number; - memoryClient?: ConversationWriter; + writer?: SessionWriter; onSaved?: () => void; } -function extractText(parts: Part[] | undefined): string { - const text = (parts ?? []) - .filter( - (part): part is Part & { type: "text"; text: string } => - part.type === "text" && - part.synthetic !== true && - part.ignored !== true && - typeof part.text === "string", - ) - .map((part) => part.text) - .join("\n") - .trim(); - - if (!text || isFullyPrivate(text)) return ""; - return stripPrivateContent(text).trim(); -} - -function isFinalAssistantMessage(info: CaptureMessageInfo): boolean { - return ( - info.role === "assistant" && - info.summary !== true && - typeof info.finish === "string" && - info.finish.length > 0 && - info.finish !== "tool-calls" - ); -} - -export function buildCaptureTurns(messages: SessionMessage[]): CaptureTurn[] { - const turns: CaptureTurn[] = []; - let current: - | { - id: string; - messages: ConversationMessage[]; - fullyPrivate: boolean; - complete: boolean; - } - | undefined; - - const finishCurrent = () => { - if (current?.complete) { - turns.push({ - id: current.id, - messages: current.fullyPrivate ? [] : current.messages, - }); - } - current = undefined; - }; - - for (const message of messages) { - const { info } = message; - - if (info.role === "user") { - finishCurrent(); - const rawText = (message.parts ?? []) - .filter( - (part): part is Part & { type: "text"; text: string } => - part.type === "text" && - part.synthetic !== true && - part.ignored !== true && - typeof part.text === "string", - ) - .map((part) => part.text) - .join("\n") - .trim(); - const text = extractText(message.parts); - current = { - id: info.id, - messages: text ? [{ role: "user", content: text }] : [], - fullyPrivate: rawText.length > 0 && isFullyPrivate(rawText), - complete: false, - }; - continue; - } - - if (!current || info.role !== "assistant" || info.summary === true) { - continue; - } - - const text = extractText(message.parts); - if (text && !current.fullyPrivate) { - current.messages.push({ role: "assistant", content: text }); - } - if (isFinalAssistantMessage(info)) { - current.complete = true; - } - } - - finishCurrent(); - return turns; -} - -export function buildCadenceBatches( - turns: CaptureTurn[], - captureEveryNTurns: number, -): CaptureBatch[] { - if (captureEveryNTurns <= 0) return []; - - const batches: CaptureBatch[] = []; - const completeBatchCount = Math.floor(turns.length / captureEveryNTurns); - for (let index = 0; index < completeBatchCount; index += 1) { - const start = index * captureEveryNTurns; - const end = start + captureEveryNTurns; - batches.push({ - startTurn: start + 1, - endTurn: end, - turns: turns.slice(start, end), - }); - } - return batches; -} - -export function buildSessionEndBatch( - turns: CaptureTurn[], - captureEveryNTurns: number, -): CaptureBatch | null { - if (turns.length === 0) return null; - - const remainder = - captureEveryNTurns > 0 ? turns.length % captureEveryNTurns : turns.length; - if (remainder === 0) return null; - - const start = turns.length - remainder; - return { - startTurn: start + 1, - endTurn: turns.length, - turns: turns.slice(start), - }; -} - -export function getCaptureId( - sessionID: string, - batch: CaptureBatch, -): string { - const firstTurn = batch.turns[0]?.id ?? String(batch.startTurn); - const lastTurn = batch.turns.at(-1)?.id ?? String(batch.endTurn); - const fingerprint = `${sessionID}:${firstTurn}:${lastTurn}`; - const digest = createHash("sha256").update(fingerprint).digest("hex"); - return `opencode:capture:${digest}`; -} - export function createCaptureHook( ctx: CaptureContext, tags: ResolvedTags, options?: CaptureOptions, ) { - const captureEveryNTurns = - options?.captureEveryNTurns ?? CONFIG.captureEveryNTurns; - const memoryClient = options?.memoryClient ?? supermemoryClient; - const snapshots = new Map(); + const writer = options?.writer ?? supermemoryClient; const activeSessions = new Set(); - const completedCaptureIds = new Set(); const inFlight = new Map>(); - async function refreshSnapshot(sessionID: string): Promise { + async function readMessages(sessionID: string): Promise { const response = await ctx.client.session.messages({ path: { id: sessionID }, query: { directory: ctx.directory }, @@ -230,111 +60,30 @@ export function createCaptureHook( if (!Array.isArray(response) && !response.data) { throw new Error(`Unable to read OpenCode session ${sessionID}`); } - const rawMessages = Array.isArray(response) ? response : response.data ?? []; - const turns = buildCaptureTurns(rawMessages); - snapshots.set(sessionID, turns); - activeSessions.add(sessionID); - return turns; + return Array.isArray(response) ? response : (response.data ?? []); } - async function saveBatch( - sessionID: string, - batch: CaptureBatch, - reason: "cadence" | "session_end", - ): Promise { - const captureId = getCaptureId(sessionID, batch); - if (completedCaptureIds.has(captureId)) return true; - - const messages = batch.turns.flatMap((turn) => turn.messages); - if (messages.length === 0) { - completedCaptureIds.add(captureId); - return true; - } - - let result: { success: boolean; error?: string }; + async function capture(sessionID: string): Promise { + let messages: SessionMessage[]; try { - result = await 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, - }, - ); + messages = await readMessages(sessionID); } catch (error) { - result = { - success: false, - error: error instanceof Error ? error.message : String(error), - }; - } - - if (result.success) { - completedCaptureIds.add(captureId); - options?.onSaved?.(); - log("[capture] conversation batch saved", { + log("[capture] failed to read session transcript", { sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, + error: String(error), }); - return true; + return false; } - - log("[capture] failed to save conversation batch", { - sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, - error: result.error, + return captureSession({ + writer, + namespace: tags.canonical, + sessionId: sessionID, + messages, + metadata: sessionMetadata(tags.projectName, tags.projectId), + supportingContext: AGENT_ENTITY_CONTEXT, + timeoutMs: SESSION_CAPTURE_TIMEOUT_MS, + onSaved: options?.onSaved, }); - return false; - } - - async function captureCadence( - sessionID: string, - turns: CaptureTurn[], - ): Promise { - let complete = true; - for (const batch of buildCadenceBatches(turns, captureEveryNTurns)) { - if (!(await saveBatch(sessionID, batch, "cadence"))) { - complete = false; - } - } - return complete; - } - - async function captureSessionEnd(sessionID: string): Promise { - let turns = snapshots.get(sessionID); - if (!turns) { - try { - turns = await refreshSnapshot(sessionID); - } catch (error) { - log("[capture] failed to read terminal session", { - sessionID, - error: String(error), - }); - return false; - } - } - - const cadenceComplete = await captureCadence(sessionID, turns); - const finalBatch = buildSessionEndBatch(turns, captureEveryNTurns); - const finalComplete = finalBatch - ? await saveBatch(sessionID, finalBatch, "session_end") - : true; - return cadenceComplete && finalComplete; } async function runExclusive( @@ -367,32 +116,18 @@ export function createCaptureHook( const sessionID = props?.sessionID as string | undefined; if (!sessionID) return; activeSessions.add(sessionID); - await runExclusive(sessionID, async () => { - try { - const turns = await refreshSnapshot(sessionID); - await captureCadence(sessionID, turns); - } catch (error) { - log("[capture] failed to process idle session", { - sessionID, - error: String(error), - }); - } + await capture(sessionID); }); return; } if (event.type === "session.deleted") { - const sessionInfo = props?.info as { id?: string } | undefined; - const sessionID = sessionInfo?.id; + const sessionID = (props?.info as { id?: string } | undefined)?.id; if (!sessionID) return; activeSessions.add(sessionID); - await runExclusive(sessionID, async () => { - if (await captureSessionEnd(sessionID)) { - snapshots.delete(sessionID); - activeSessions.delete(sessionID); - } + if (await capture(sessionID)) activeSessions.delete(sessionID); }); return; } @@ -401,10 +136,7 @@ export function createCaptureHook( await Promise.all( [...activeSessions].map((sessionID) => runExclusive(sessionID, async () => { - if (await captureSessionEnd(sessionID)) { - snapshots.delete(sessionID); - activeSessions.delete(sessionID); - } + if (await capture(sessionID)) activeSessions.delete(sessionID); }), ), ); diff --git a/src/services/client.ts b/src/services/client.ts index 0604f7a..b46adaf 100644 --- a/src/services/client.ts +++ b/src/services/client.ts @@ -736,6 +736,63 @@ export class SupermemoryClient { storedMemoryIds: savedIds, }; } + + async addSession( + namespace: string, + raw: string, + options: { + sessionId: string; + generation?: number; + offset?: number; + metadata?: Record; + supportingContext?: string; + timeoutMs?: number; + }, + ): Promise { + if (!SUPERMEMORY_API_KEY) throw new Error("SUPERMEMORY_API_KEY not set"); + const baseUrl = getApiBaseUrl().replace(/\/+$/, ""); + const metadata: Record = { + sm_source: OPENCODE_SOURCE, + sm_client: OPENCODE_SOURCE, + sm_plugin_version: PLUGIN_VERSION, + ...(options.metadata ?? {}), + }; + if (options.generation != null) { + metadata.sm_session_generation = options.generation; + } + if (options.offset != null) metadata.sm_session_offset = options.offset; + const body: { + content: string; + id: string; + metadata: Record; + supportingContext?: string; + } = { content: raw, id: options.sessionId, metadata }; + if (options.supportingContext) { + body.supportingContext = options.supportingContext; + } + + const response = await fetch( + `${baseUrl}/ns/${encodeURIComponent(namespace)}/document`, + { + method: "POST", + headers: { + Authorization: `Bearer ${SUPERMEMORY_API_KEY}`, + "Content-Type": "application/json", + "x-sm-source": OPENCODE_SOURCE, + "x-sm-session": OPENCODE_SOURCE, + }, + body: JSON.stringify(body), + signal: AbortSignal.timeout(options.timeoutMs ?? TIMEOUT_MS), + }, + ); + if (!response.ok) { + const text = await response.text().catch(() => ""); + throw Object.assign( + new Error(`Supermemory session ${response.status}: ${text.slice(0, 200)}`), + { status: response.status }, + ); + } + } } export const supermemoryClient = new SupermemoryClient(); diff --git a/src/services/session.ts b/src/services/session.ts new file mode 100644 index 0000000..3059181 --- /dev/null +++ b/src/services/session.ts @@ -0,0 +1,209 @@ +import { createHash } from "node:crypto"; +import { + existsSync, + mkdirSync, + readFileSync, + renameSync, + writeFileSync, +} from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; + +import { log as defaultLog } from "./logger.js"; + +export const SESSION_CAPTURE_TIMEOUT_MS = 25_000; + +function sessionDir(): string { + const override = process.env.SUPERMEMORY_SESSION_DIR?.trim(); + return override || join(homedir(), ".supermemory-opencode", "session"); +} + +export interface SessionCursor { + count: number; + generation: number; + headHash: string; +} + +export interface SessionDelta { + raw: string; + generation: number; + offset: number; + count: number; + commit: () => void; +} + +export type SessionMetadata = Record; + +export interface SessionWriter { + addSession: ( + namespace: string, + raw: string, + options: { + sessionId: string; + generation: number; + offset: number; + metadata?: SessionMetadata; + supportingContext?: string; + timeoutMs?: number; + }, + ) => Promise; +} + +export function sessionMetadata( + projectName: string, + projectId: string, +): SessionMetadata { + return { + type: "session_turn", + project: projectName, + sm_project_id: projectId, + sm_scope: "personal", + sm_capture_mode: "automatic", + timestamp: new Date().toISOString(), + }; +} + +function sanitizeSegment(value: string): string { + return value.replace(/[^a-zA-Z0-9._-]/g, "_"); +} + +function cursorPath(sessionId: string): string { + return join(sessionDir(), `${sanitizeSegment(sessionId)}.json`); +} + +function ensureDir(): void { + const dir = sessionDir(); + if (!existsSync(dir)) mkdirSync(dir, { recursive: true }); +} + +export function readCursor(sessionId: string): SessionCursor { + try { + const cursor = JSON.parse( + readFileSync(cursorPath(sessionId), "utf8"), + ) as Partial; + if ( + typeof cursor.count === "number" && + typeof cursor.generation === "number" + ) { + return { + count: cursor.count, + generation: cursor.generation, + headHash: typeof cursor.headHash === "string" ? cursor.headHash : "", + }; + } + } catch {} + return { count: 0, generation: 0, headHash: "" }; +} + +function writeCursor(sessionId: string, cursor: SessionCursor): void { + ensureDir(); + const path = cursorPath(sessionId); + const temp = `${path}.${process.pid}.tmp`; + writeFileSync(temp, JSON.stringify(cursor)); + renameSync(temp, path); +} + +function messageId(message: unknown): string { + if (message && typeof message === "object") { + const record = message as { id?: unknown; info?: { id?: unknown } }; + if (typeof record.info?.id === "string") return record.info.id; + if (typeof record.id === "string") return record.id; + } + return ""; +} + +function hashPrefix(ids: string[], count: number): string { + if (count <= 0) return ""; + return createHash("sha256") + .update(ids.slice(0, count).join("\n")) + .digest("hex"); +} + +// OpenCode has no append-only transcript file: the host returns the whole +// message history each call and compaction rewrites its head. The message count +// is the append offset; a shorter list or a changed already-sent prefix means the +// history was rewritten, so the generation is bumped and the delta restarts at 0. +export function buildSessionDelta( + sessionId: string, + messages: readonly unknown[], +): SessionDelta | null { + const previous = readCursor(sessionId); + const ids = messages.map(messageId); + let offset = previous.count; + let generation = previous.generation; + + const consumedHash = hashPrefix(ids, Math.min(previous.count, ids.length)); + if ( + messages.length < previous.count || + (previous.headHash !== "" && previous.headHash !== consumedHash) + ) { + offset = 0; + generation = previous.generation + 1; + } + + if (messages.length <= offset) return null; + + const raw = `${messages + .slice(offset) + .map((message) => JSON.stringify(message)) + .join("\n")}\n`; + const count = messages.length; + const headHash = hashPrefix(ids, count); + + return { + raw, + generation, + offset, + count, + commit: () => writeCursor(sessionId, { count, generation, headHash }), + }; +} + +export async function captureSession(options: { + writer: SessionWriter; + namespace: string; + sessionId: string; + messages: readonly unknown[]; + metadata?: SessionMetadata; + supportingContext?: string; + timeoutMs?: number; + onSaved?: () => void; + logger?: typeof defaultLog; +}): Promise { + const { writer, namespace, sessionId, messages } = options; + const logger = options.logger ?? defaultLog; + if (!namespace) return false; + + const delta = buildSessionDelta(sessionId, messages); + if (!delta) return true; + + try { + await writer.addSession(namespace, delta.raw, { + sessionId, + generation: delta.generation, + offset: delta.offset, + metadata: options.metadata, + supportingContext: options.supportingContext, + timeoutMs: options.timeoutMs ?? SESSION_CAPTURE_TIMEOUT_MS, + }); + } catch (error) { + logger("[session] failed to send session session", { + sessionId, + generation: delta.generation, + offset: delta.offset, + error: error instanceof Error ? error.message : String(error), + }); + return false; + } + + delta.commit(); + options.onSaved?.(); + logger("[session] session session sent", { + sessionId, + generation: delta.generation, + offset: delta.offset, + count: delta.count, + bytes: delta.raw.length, + }); + return true; +} diff --git a/src/v2/runtime.test.ts b/src/v2/runtime.test.ts index bb05a5a..99d2525 100644 --- a/src/v2/runtime.test.ts +++ b/src/v2/runtime.test.ts @@ -6,14 +6,11 @@ import { DEFAULT_RECALL_DIRECTIVE } from "../services/recall.js"; import type { ResolvedTags } from "../services/tags.js"; import { applyInjection, - buildTranscriptTurns, buildV2RecallDirective, - mergeTurns, SUPERMEMORY_RECALL_TOOL_NAME, SUPERMEMORY_TOOL_NAME, V2Runtime, type RequestMessage, - type TranscriptMessage, type V2Context, type V2RuntimeDependencies, } from "./runtime.js"; @@ -29,18 +26,6 @@ const tags: ResolvedTags = { allReads: [], }; -function user(id: string, text: string): TranscriptMessage { - return { id, type: "user", text }; -} - -function assistant( - id: string, - text: string, - finish: string = "stop", -): TranscriptMessage { - return { id, type: "assistant", finish, content: [{ type: "text", text }] }; -} - function request(id: string, text: string): RequestMessage { return { id, @@ -55,49 +40,6 @@ function texts(message: RequestMessage): string[] { ); } -describe("OpenCode 2 transcript capture", () => { - test("groups completed turns and protects private content", () => { - const turns = buildTranscriptTurns([ - { id: "sys", type: "system", text: "injected" }, - user("u1", "real prompt"), - assistant("a1", "calling a tool", "tool-calls"), - assistant("a2", "real answer"), - user("u2", "secret prompt"), - assistant("a3", "secret answer"), - user("u3", "token secret"), - assistant("a4", "safe answer"), - user("u4", "still streaming"), - ]); - - expect(turns.map((turn) => turn.id)).toEqual(["u1", "u2", "u3"]); - expect(turns[0]?.messages).toEqual([ - { role: "user", content: "real prompt" }, - { role: "assistant", content: "calling a tool" }, - { role: "assistant", content: "real answer" }, - ]); - expect(turns[1]?.messages).toEqual([]); - expect(turns[2]?.messages[0]).toEqual({ - role: "user", - content: "token [REDACTED]", - }); - }); - - test("keeps turn positions stable when the transcript is compacted", () => { - const existing = mergeTurns( - [], - new Map(), - buildTranscriptTurns([user("u1", "one"), assistant("a1", "1")]), - ); - const index = new Map([["u1", 0]]); - mergeTurns( - existing, - index, - buildTranscriptTurns([user("u2", "two"), assistant("a2", "2")]), - ); - expect(existing.map((turn) => turn.id)).toEqual(["u1", "u2"]); - }); -}); - describe("OpenCode 2 request injection", () => { test("points the advisory directive at the recall helper", () => { const directive = buildV2RecallDirective(DEFAULT_RECALL_DIRECTIVE); @@ -133,7 +75,7 @@ interface Harness { writes: Array<{ customId?: string; timeoutMs?: number; metadata?: Record }>; adds: Array<{ content: string; customId?: string; metadata?: Record }>; emitted: Array<{ kind?: unknown; message?: unknown }>; - transcript: TranscriptMessage[]; + transcript: unknown[]; } function harness( @@ -145,7 +87,7 @@ function harness( const writes: Harness["writes"] = []; const adds: Harness["adds"] = []; const emitted: Harness["emitted"] = []; - const state: { transcript: TranscriptMessage[] } = { transcript: [] }; + const state: { transcript: unknown[] } = { transcript: [] }; const registration = { dispose: async () => undefined }; const ctx = { @@ -255,7 +197,6 @@ function harness( config: { recallMode: "direct", injectProfile: true, - captureEveryNTurns: 1, compactionEnabled: true, keywordPatterns: ["remember"], maxProjectMemories: 10, @@ -284,7 +225,7 @@ function harness( get transcript() { return state.transcript; }, - set transcript(value: TranscriptMessage[]) { + set transcript(value: unknown[]) { state.transcript = value; }, }; @@ -375,77 +316,6 @@ describe("OpenCode 2 runtime", () => { h.runtime.cleanup(); }); - test("captures completed turns idempotently and saves compaction summaries", async () => { - const h = harness(); - await h.runtime.register(); - - h.transcript = [user("u1", "question 1"), assistant("a1", "answer 1")]; - await h.runtime.handleEvent({ - id: "e1", - type: "session.execution.succeeded", - data: { sessionID: "s1" }, - }); - await h.runtime.idle(); - await h.runtime.handleEvent({ - id: "e2", - type: "session.execution.succeeded", - data: { sessionID: "s1" }, - }); - await h.runtime.idle(); - expect(h.writes).toHaveLength(1); - expect(h.writes[0]?.customId).toStartWith("opencode:capture:"); - expect(h.writes[0]?.timeoutMs).toBe(3_000); - expect(h.writes[0]?.metadata).toMatchObject({ - captureReason: "cadence", - sm_capture_mode: "automatic", - }); - - // Compaction truncates the transcript; the earlier turn keeps its slot. - h.transcript = [ - { id: "c1", type: "compaction" }, - user("u2", "question 2"), - assistant("a2", "answer 2"), - ]; - await h.runtime.handleEvent({ - id: "e3", - type: "session.execution.succeeded", - data: { sessionID: "s1" }, - }); - await h.runtime.idle(); - expect(h.writes).toHaveLength(2); - expect(h.writes[1]?.customId).not.toBe(h.writes[0]?.customId); - - const summary = "Summary ".repeat(20); - await h.runtime.handleEvent({ - id: "e4", - type: "session.compaction.ended", - data: { sessionID: "s1", text: summary, reason: "auto" }, - }); - await h.runtime.handleEvent({ - id: "e4", - type: "session.compaction.ended", - data: { sessionID: "s1", text: summary, reason: "auto" }, - }); - expect(h.adds).toHaveLength(1); - expect(h.adds[0]?.customId).toStartWith("opencode:compaction:"); - expect(h.adds[0]?.metadata).toMatchObject({ sm_capture_mode: "compaction" }); - - const system: Array<{ type: "text"; text: string }> = []; - await h.runtime.handleCompaction({ sessionID: "s1", system }); - expect(system[0]?.text).toContain(COMPACTION_CONTEXT_MARKER); - expect(system[0]?.text).toContain("Build: bun run build"); - - await h.runtime.handleEvent({ - id: "e5", - type: "session.deleted", - data: { sessionID: "s1" }, - }); - expect(h.runtime.trackedSessionCount).toBe(0); - expect(h.writes).toHaveLength(2); - expect(h.emitted.filter((event) => event.kind === "saved")).toHaveLength(3); - h.runtime.cleanup(); - }); - test("reports tool-driven recall activity", async () => { const h = harness(); await h.runtime.register(); diff --git a/src/v2/runtime.ts b/src/v2/runtime.ts index 2fee56b..5af1ac3 100644 --- a/src/v2/runtime.ts +++ b/src/v2/runtime.ts @@ -9,15 +9,12 @@ import { type MemoryActivityNotice, type MemoryActivityReporter, } from "../services/activity.js"; -import { - AUTOMATIC_CAPTURE_TIMEOUT_MS, - buildCadenceBatches, - buildSessionEndBatch, - getCaptureId, - type CaptureBatch, - type CaptureTurn, -} from "../services/capture.js"; import { supermemoryClient, type SupermemoryClient } from "../services/client.js"; +import { + captureSession, + SESSION_CAPTURE_TIMEOUT_MS, + sessionMetadata, +} from "../services/session.js"; import { createCompactionPrompt, fitProjectMemories, @@ -38,7 +35,6 @@ import { type SupermemoryToolArgs, } from "../services/memory-tool.js"; import { RECALL_PERMISSION } from "../services/opencode-permissions.js"; -import { isFullyPrivate, stripPrivateContent } from "../services/privacy.js"; import { buildDirectRecallResult, buildRecallDirective, @@ -62,6 +58,7 @@ const SYNTHETIC_METADATA_KEY = "supermemory"; const UPDATE_COMMAND = "bunx opencode-supermemory@latest install"; const MAX_DISPATCHES_PER_SESSION = 8; const MIN_SUMMARY_CHARS = 100; +const SUMMARY_SAVE_TIMEOUT_MS = 3_000; export const SUPERMEMORY_TOOL_NAME = "supermemory"; export const SUPERMEMORY_RECALL_TOOL_NAME = "supermemory_recall"; @@ -105,7 +102,7 @@ export const SUPERMEMORY_RECALL_INPUT = { type RuntimeMemoryClient = Pick< SupermemoryClient, | "addMemory" - | "ingestConversation" + | "addSession" | "getProfileScoped" | "searchMemoriesForRecall" | "searchMemoriesScoped" @@ -118,7 +115,6 @@ type RuntimeConfig = Pick< typeof CONFIG, | "recallMode" | "injectProfile" - | "captureEveryNTurns" | "compactionEnabled" | "keywordPatterns" | "maxProjectMemories" @@ -152,15 +148,6 @@ export interface V2Event { data?: Record; } -/** Minimal shape of a persisted OpenCode 2 transcript message. */ -export interface TranscriptMessage { - id: string; - type: string; - text?: string; - finish?: string; - content?: ReadonlyArray<{ type: string; text?: string }>; -} - interface Injection { start: string[]; end: string[]; @@ -173,9 +160,6 @@ interface SessionState { injectedInitialContext: boolean; dispatches: Map>; dispatchOrder: string[]; - turns: CaptureTurn[]; - turnIndex: Map; - completedCaptureIds: Set; } interface PendingSummary { @@ -294,105 +278,6 @@ export function applyInjection( } } -function sanitizeCaptureText(text: string): string { - if (!text || isFullyPrivate(text)) return ""; - return stripPrivateContent(text).trim(); -} - -function isFinalAssistant(message: TranscriptMessage): boolean { - return ( - typeof message.finish === "string" && - message.finish.length > 0 && - message.finish !== "tool-calls" - ); -} - -function assistantText(message: TranscriptMessage): string { - return (message.content ?? []) - .filter((part) => part.type === "text" && typeof part.text === "string") - .map((part) => part.text as string) - .join("\n") - .trim(); -} - -/** - * Groups an OpenCode 2 transcript into completed user/assistant turns using - * the same rules as the V1 capture hook: synthetic and system messages are - * ignored, `` content is redacted, and a fully private prompt hides - * its whole turn. - */ -export function buildTranscriptTurns( - messages: ReadonlyArray, -): CaptureTurn[] { - const turns: CaptureTurn[] = []; - let current: - | { - id: string; - messages: CaptureTurn["messages"]; - fullyPrivate: boolean; - complete: boolean; - } - | undefined; - - const finish = () => { - if (current?.complete) { - turns.push({ - id: current.id, - messages: current.fullyPrivate ? [] : current.messages, - }); - } - current = undefined; - }; - - for (const message of messages) { - if (message.type === "user") { - finish(); - const rawText = (message.text ?? "").trim(); - const text = sanitizeCaptureText(rawText); - current = { - id: message.id, - messages: text ? [{ role: "user", content: text }] : [], - fullyPrivate: rawText.length > 0 && isFullyPrivate(rawText), - complete: false, - }; - continue; - } - - if (!current || message.type !== "assistant") continue; - - const text = sanitizeCaptureText(assistantText(message)); - if (text && !current.fullyPrivate) { - current.messages.push({ role: "assistant", content: text }); - } - if (isFinalAssistant(message)) current.complete = true; - } - - finish(); - return turns; -} - -/** - * Merges freshly read turns into the session's cumulative turn list. The - * transcript OpenCode returns may shrink after compaction, so turn positions - * (and therefore capture IDs) are anchored to the first time a turn was seen. - */ -export function mergeTurns( - existing: CaptureTurn[], - index: Map, - incoming: CaptureTurn[], -): CaptureTurn[] { - for (const turn of incoming) { - const position = index.get(turn.id); - if (position === undefined) { - index.set(turn.id, existing.length); - existing.push(turn); - } else { - existing[position] = turn; - } - } - return existing; -} - export class EventDeduper { readonly #limit: number; readonly #seen = new Set(); @@ -585,7 +470,6 @@ export class V2Runtime { hooks: ["session.context", "session.compaction", "permission.evaluate", "tool.execute"], activity: this.#emitActivity ? "rpc" : "log-only", recallMode: this.#deps.config.recallMode, - captureEveryNTurns: this.#deps.config.captureEveryNTurns, compactionEnabled: this.#deps.config.compactionEnabled, }); } @@ -710,7 +594,7 @@ export class V2Runtime { case "session.execution.succeeded": { if (!sessionID) return; void this.#runCaptureExclusive(sessionID, () => - this.#captureCadence(sessionID), + this.#captureSession(sessionID), ); return; } @@ -719,7 +603,7 @@ export class V2Runtime { if (!sessionID || event.data?.reason !== "shutdown") return; if (!this.#states.has(sessionID)) return; void this.#runCaptureExclusive(sessionID, () => - this.#captureSessionEnd(sessionID), + this.#captureSession(sessionID), ); return; } @@ -730,7 +614,7 @@ export class V2Runtime { this.#recall.delete(sessionID); if (!state) return; await this.#runCaptureExclusive(sessionID, () => - this.#captureSessionEnd(sessionID), + this.#captureSession(sessionID), ); this.#states.delete(sessionID); return; @@ -805,9 +689,6 @@ export class V2Runtime { injectedInitialContext: false, dispatches: new Map(), dispatchOrder: [], - turns: [], - turnIndex: new Map(), - completedCaptureIds: new Set(), }; this.#states.set(sessionID, state); return state; @@ -1004,118 +885,47 @@ export class V2Runtime { } } - async #refreshTurns(sessionID: string, state: SessionState): Promise { + async #sessionMessages(sessionID: string): Promise { + const messages = (await this.#ctx.session.context({ sessionID })) as unknown; + return Array.isArray(messages) ? messages : []; + } + + async #captureSession(sessionID: string): Promise { + this.#state(sessionID); + + let tags: ResolvedTags; try { - const messages = (await this.#ctx.session.context({ sessionID })) as unknown; - if (Array.isArray(messages)) { - mergeTurns( - state.turns, - state.turnIndex, - buildTranscriptTurns(messages as TranscriptMessage[]), - ); - } + tags = await this.#resolveSession(sessionID); } catch (error) { - this.#deps.logger("v2 capture transcript read failed; using cached turns", { + this.#deps.logger("v2 session capture skipped; no session directory", { sessionID, error: String(error), }); - } - return state.turns; - } - - async #saveBatch( - sessionID: string, - state: SessionState, - batch: CaptureBatch, - reason: "cadence" | "session_end", - ): Promise { - 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; + return; } - let result: { success: boolean; error?: string }; + let messages: unknown[]; 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, - }, - ); + messages = await this.#sessionMessages(sessionID); } catch (error) { - result = { - success: false, - error: error instanceof Error ? error.message : String(error), - }; - } - - if (result.success) { - state.completedCaptureIds.add(captureId); - this.#activity.saved(); - this.#deps.logger("[capture] conversation batch saved", { + this.#deps.logger("v2 session transcript read failed", { sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, + error: String(error), }); - return true; + return; } - this.#deps.logger("[capture] failed to save conversation batch", { - sessionID, - reason, - startTurn: batch.startTurn, - endTurn: batch.endTurn, - error: result.error, + await captureSession({ + writer: this.#deps.memoryClient, + namespace: tags.canonical, + sessionId: sessionID, + messages, + metadata: sessionMetadata(tags.projectName, tags.projectId), + supportingContext: AGENT_ENTITY_CONTEXT, + timeoutMs: SESSION_CAPTURE_TIMEOUT_MS, + onSaved: () => this.#activity.saved(), + logger: this.#deps.logger, }); - return false; - } - - async #captureCadence(sessionID: string): Promise { - const state = this.#state(sessionID); - const turns = await this.#refreshTurns(sessionID, state); - for (const batch of buildCadenceBatches( - turns, - this.#deps.config.captureEveryNTurns, - )) { - await this.#saveBatch(sessionID, state, batch, "cadence"); - } - } - - async #captureSessionEnd(sessionID: string): Promise { - const state = this.#state(sessionID); - const turns = await this.#refreshTurns(sessionID, state); - for (const batch of buildCadenceBatches( - turns, - this.#deps.config.captureEveryNTurns, - )) { - await this.#saveBatch(sessionID, state, batch, "cadence"); - } - const finalBatch = buildSessionEndBatch( - turns, - this.#deps.config.captureEveryNTurns, - ); - if (finalBatch) { - await this.#saveBatch(sessionID, state, finalBatch, "session_end"); - } } async #flushAll(allowInactive = false): Promise { @@ -1126,7 +936,7 @@ export class V2Runtime { await Promise.all([ ...sessions.map((sessionID) => this.#runCaptureExclusive(sessionID, () => - this.#captureSessionEnd(sessionID), + this.#captureSession(sessionID), ), ), ...[...pending].map((sessionID) => @@ -1195,7 +1005,7 @@ export class V2Runtime { { customId: summary.customId, entityContext: AGENT_ENTITY_CONTEXT, - timeoutMs: AUTOMATIC_CAPTURE_TIMEOUT_MS, + timeoutMs: SUMMARY_SAVE_TIMEOUT_MS, }, ); if (result.success) {