From 89df44feca8c2e8d2b80ec3769697bfcbf006625 Mon Sep 17 00:00:00 2001 From: KRISTHIAN MANUEL Date: Fri, 16 Oct 2026 18:48:25 -0600 Subject: [PATCH] fix(memory): support the OpenCode V2 plugin API and session_v2 schema OpenCode V2 renamed its conversation tables (session/message/part -> session_v2/session_message, parts inlined into data.content) and replaced the V1 plugin API, so the installed plugin failed to load and both the hook capture path and the desktop scan ignored the V2 database entirely. - AgentSourceCore: new opencode-v2-source mappers project V2 rows onto the shared turn parser (sequential parent grouping, inline content parts, completion-timestamp finish signal) - opencode db-readers (Memory + App/backend): fall back to the V2 tables when the V1 tables are absent, in both the scan reader and the bridge hook reader - plugin template: dual entrypoint. The V1 helper import is now lazy so the file loads under V2, and a default export carries setup() for V2 (tools via ctx.tool.transform, prompt-hook recall injection, event stream capture) plus server() for V1 (>= 1.18.29) - tests: mapper unit tests plus V2-schema SQLite reader cases Verified live against OpenCode V2 + Memory 2.1.4 (see #654). Closes #654 --- AgentSourceCore/src/index.ts | 1 + AgentSourceCore/src/opencode-v2-source.ts | 141 +++++++ .../src/tests/opencode-v2-source.test.ts | 157 +++++++ .../agent-source/opencode/db-reader.ts | 39 +- .../templates/memmy-opencode-plugin.ts | 394 +++++++++++++++++- .../skill-writer/workspace-bridge/runtime.ts | 64 ++- .../adapters/opencode/db-reader.ts | 39 +- .../templates/memmy-opencode-plugin.ts | 394 +++++++++++++++++- .../integration/workspace-bridge/runtime.ts | 64 ++- .../agent-source-opencode-reader.test.ts | 110 +++++ 10 files changed, 1381 insertions(+), 22 deletions(-) create mode 100644 AgentSourceCore/src/opencode-v2-source.ts create mode 100644 AgentSourceCore/src/tests/opencode-v2-source.test.ts diff --git a/AgentSourceCore/src/index.ts b/AgentSourceCore/src/index.ts index cd3020e3e..93a29737d 100644 --- a/AgentSourceCore/src/index.ts +++ b/AgentSourceCore/src/index.ts @@ -4,6 +4,7 @@ export * from "./cursor-source-turn.js"; export * from "./claude-code-source-turn.js"; export * from "./openclaw-source-turn.js"; export * from "./opencode-source-turn.js"; +export * from "./opencode-v2-source.js"; export * from "./hermes-source-turn.js"; export * from "./deepseek-source-turn.js"; export * from "./deepseek-session-files.js"; diff --git a/AgentSourceCore/src/opencode-v2-source.ts b/AgentSourceCore/src/opencode-v2-source.ts new file mode 100644 index 000000000..7d6b3c505 --- /dev/null +++ b/AgentSourceCore/src/opencode-v2-source.ts @@ -0,0 +1,141 @@ +import type { OpencodeSession } from "./opencode-source-turn.js"; + +/** + * OpenCode V2 renamed its conversation store: `session`/`message`/`part` became + * `session_v2`/`session_message`, message parts moved inline into `data.content`, + * and the assistant `parentID` link was dropped. These mappers project V2 rows onto + * the V1-shaped view the shared turn parser already consumes, so both schemas share + * one turn pipeline. Rows are fetched by the caller; only the projection lives here. + */ + +/** One `session_v2` row. */ +export interface OpencodeV2SessionRow { + id: string; + parentId: string | null; + directory: string | null; + agent?: string | null; + /** Raw JSON column; `{ messageID }` marks the revert boundary like V1. */ + revert?: string | null; +} + +/** One `session_message` row. `type` carries the role (`user`, `assistant`, ...). */ +export interface OpencodeV2MessageRow { + id: string; + type: string; + seq: number; + /** Parsed `data` JSON. */ + data: unknown; +} + +export function toOpencodeV2Session(row: OpencodeV2SessionRow): OpencodeSession { + return { + id: String(row.id), + parentId: row.parentId == null ? null : String(row.parentId), + directory: row.directory == null ? null : String(row.directory), + agent: typeof row.agent === "string" ? row.agent : null, + revertMessageId: opencodeV2RevertMessageId(row.revert) + }; +} + +/** + * Maps one session's `session_message` rows onto the message view the shared turn + * parser expects. V2 has no assistant `parentID`, so turns are grouped sequentially: + * every assistant row belongs to the latest preceding user row, which reproduces the + * V1 parent buckets from `seq` order. Rows that are neither user nor assistant are + * bookkeeping (`idle`, `model-switched`, `system`, `synthetic`) and are dropped. + */ +export function toOpencodeV2MessageView( + rows: readonly OpencodeV2MessageRow[] +): Array<{ id: string; data: Record | undefined }> { + const view: Array<{ id: string; data: Record | undefined }> = []; + let parentId = ""; + for (const row of rows) { + const data = isRecord(row.data) ? row.data : {}; + if (row.type === "user") { + parentId = row.id; + view.push({ + id: row.id, + data: { + role: "user", + ...(typeof data.agent === "string" ? { agent: data.agent } : {}), + time: isRecord(data.time) ? data.time : undefined + } + }); + continue; + } + if (row.type !== "assistant" || !parentId) continue; + const time = isRecord(data.time) ? data.time : {}; + const completed = typeof time.completed === "number"; + view.push({ + id: row.id, + data: { + role: "assistant", + parentID: parentId, + ...(typeof data.agent === "string" ? { agent: data.agent } : {}), + ...(data.summary === true ? { summary: true } : {}), + ...(isRecord(data.error) ? { error: data.error } : {}), + // V2 has no `finish`; a persisted completion timestamp is the stop signal. + ...(completed ? { finish: "stop" } : {}), + time + } + }); + } + return view; +} + +/** + * Maps one `session_message` row onto the part view. V2 inlines message content in + * `data.content`; user text lives directly in `data.text`. Reasoning parts have no V1 + * counterpart and are dropped. + */ +export function toOpencodeV2MessageParts( + row: OpencodeV2MessageRow +): Array<{ id: string; data: Record | undefined }> { + const data = isRecord(row.data) ? row.data : {}; + if (row.type === "user") { + const text = typeof data.text === "string" ? data.text : ""; + return text ? [{ id: row.id + ":text", data: { type: "text", text } }] : []; + } + if (row.type !== "assistant") return []; + const content = Array.isArray(data.content) ? data.content : []; + const parts: Array<{ id: string; data: Record | undefined }> = []; + for (const [index, item] of content.entries()) { + if (!isRecord(item)) continue; + if (item.type === "text" && typeof item.text === "string") { + parts.push({ id: row.id + ":" + index, data: { type: "text", text: item.text } }); + } else if (item.type === "tool") { + parts.push({ + id: row.id + ":" + index, + data: { + type: "tool", + callID: typeof item.id === "string" ? item.id : undefined, + tool: typeof item.name === "string" ? item.name : "tool", + ...(isRecord(item.state) ? { state: item.state } : {}), + ...(isRecord(item.metadata) ? { metadata: item.metadata } : {}) + } + }); + } else if (item.type === "compaction") { + parts.push({ id: row.id + ":" + index, data: { type: "compaction" } }); + } + } + return parts; +} + +function opencodeV2RevertMessageId(value: unknown): string | null { + const parsed = typeof value === "string" ? parseLoose(value) : value; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return null; + const messageId = (parsed as { messageID?: unknown }).messageID; + return typeof messageId === "string" && messageId ? messageId : null; +} + +function parseLoose(value: string): unknown { + try { + return JSON.parse(value); + } catch { + return undefined; + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/AgentSourceCore/src/tests/opencode-v2-source.test.ts b/AgentSourceCore/src/tests/opencode-v2-source.test.ts new file mode 100644 index 000000000..7732a86e5 --- /dev/null +++ b/AgentSourceCore/src/tests/opencode-v2-source.test.ts @@ -0,0 +1,157 @@ +import { describe, expect, it } from "vitest"; +import { + readOpencodeSourceTurn, + toOpencodeV2MessageParts, + toOpencodeV2MessageView, + toOpencodeV2Session, + type OpencodeSource, + type OpencodeV2MessageRow +} from "../index.js"; + +describe("OpenCode V2 row mappers", () => { + it("maps a session_v2 row onto the shared session shape", () => { + expect(toOpencodeV2Session({ + id: "ses_1", + parentId: null, + directory: "/work/project", + agent: "build", + revert: "{\"messageID\":\"msg_9\"}" + })).toEqual({ + id: "ses_1", + parentId: null, + directory: "/work/project", + agent: "build", + revertMessageId: "msg_9" + }); + expect(toOpencodeV2Session({ id: "ses_2", parentId: "ses_1", directory: null }).revertMessageId).toBeNull(); + }); + + it("groups assistants under the latest preceding user row and drops bookkeeping rows", () => { + const rows: OpencodeV2MessageRow[] = [ + { id: "a0", type: "assistant", seq: 0, data: { agent: "build", time: { created: 0 }, content: [] } }, + { id: "m1", type: "model-switched", seq: 1, data: {} }, + { id: "u1", type: "user", seq: 2, data: { text: "first", time: { created: 1 } } }, + { id: "a1", type: "assistant", seq: 3, data: { agent: "build", time: { created: 2, completed: 3 }, content: [] } }, + { id: "i1", type: "idle", seq: 4, data: {} }, + { id: "a2", type: "assistant", seq: 5, data: { agent: "build", time: { created: 4, completed: 5 }, content: [] } }, + { id: "u2", type: "user", seq: 6, data: { text: "second", time: { created: 6 } } }, + { id: "a3", type: "assistant", seq: 7, data: { agent: "build", time: { created: 7, completed: 8 }, content: [] } } + ]; + const view = toOpencodeV2MessageView(rows); + expect(view.map((row) => row.id)).toEqual(["u1", "a1", "a2", "u2", "a3"]); + expect(view[1].data).toMatchObject({ role: "assistant", parentID: "u1", agent: "build", finish: "stop" }); + expect(view[4].data).toMatchObject({ role: "assistant", parentID: "u2" }); + expect(view[1].data?.finish).toBe("stop"); + }); + + it("marks streaming assistants without a completion timestamp as unfinished", () => { + const view = toOpencodeV2MessageView([ + { id: "u1", type: "user", seq: 1, data: { text: "q", time: { created: 1 } } }, + { id: "a1", type: "assistant", seq: 2, data: { agent: "build", time: { created: 2 }, content: [] } } + ]); + expect(view[1].data?.finish).toBeUndefined(); + }); + + it("synthesizes a text part for user rows and maps inline content for assistant rows", () => { + expect(toOpencodeV2MessageParts({ id: "u1", type: "user", seq: 1, data: { text: "hello" } })) + .toEqual([{ id: "u1:text", data: { type: "text", text: "hello" } }]); + expect(toOpencodeV2MessageParts({ id: "u2", type: "user", seq: 2, data: {} })).toEqual([]); + const parts = toOpencodeV2MessageParts({ + id: "a1", + type: "assistant", + seq: 3, + data: { + content: [ + { type: "reasoning", text: "internal" }, + { type: "text", text: "done" }, + { type: "tool", id: "call_1", name: "read", state: { status: "completed", input: { path: "/x" }, output: "ok" } }, + { type: "tool", name: "unnamed" }, + { type: "compaction" } + ] + } + }); + expect(parts).toEqual([ + { id: "a1:1", data: { type: "text", text: "done" } }, + { + id: "a1:2", + data: { + type: "tool", + callID: "call_1", + tool: "read", + state: { status: "completed", input: { path: "/x" }, output: "ok" } + } + }, + { id: "a1:3", data: { type: "tool", callID: undefined, tool: "unnamed" } }, + { id: "a1:4", data: { type: "compaction" } } + ]); + }); + + it("feeds the shared turn parser: a completed V2 turn captures with paired tools", async () => { + const source = createSource({ + main: [ + userRow("u1", "Inspect the configuration and report the result.", 1_000), + assistantRow("a1", "build", 2_000, 3_000, [{ type: "tool", id: "call_1", name: "read", state: { status: "completed", input: { path: "config" }, output: "Configuration verified." } }]), + assistantRow("a2", "build", 4_000, 5_000, [{ type: "text", text: "The configuration was verified." }]) + ] + }); + const result = await readOpencodeSourceTurn(source, { conversationId: "main", turnId: "u1" }); + expect(result.turn).toBeDefined(); + expect(result.turn?.query).toBe("Inspect the configuration and report the result."); + expect(result.turn?.answer).toBe("The configuration was verified."); + expect(result.turn?.profileId).toBe("build"); + expect(result.turn?.toolCalls).toHaveLength(1); + expect(result.turn?.toolCalls[0]).toMatchObject({ name: "read", status: "completed" }); + expect(result.turn?.completionEvidence).toBe("assistant_completed:a2"); + }); + + it("feeds the shared turn parser: an unfinished last assistant is left for a later attempt", async () => { + const source = createSource({ + main: [ + userRow("u1", "still running", 1_000), + assistantRow("a1", "build", 2_000, undefined, [{ type: "text", text: "partial" }]) + ] + }); + const result = await readOpencodeSourceTurn(source, { conversationId: "main", turnId: "u1" }); + expect(result.turn).toBeNull(); + expect(result.reason).toBe("turn_incomplete"); + }); +}); + +function userRow(id: string, text: string, created: number): OpencodeV2MessageRow { + return { id, type: "user", seq: created, data: { text, time: { created } } }; +} + +function assistantRow( + id: string, + agent: string, + created: number, + completed: number | undefined, + content: unknown[] +): OpencodeV2MessageRow { + return { + id, + type: "assistant", + seq: created, + data: { + agent, + time: completed === undefined ? { created } : { created, completed }, + content + } + }; +} + +function createSource(sessions: Record): OpencodeSource { + const rowsByMessage = new Map(); + for (const rows of Object.values(sessions)) { + for (const row of rows) rowsByMessage.set(row.id, row); + } + return { + sessions: () => Object.keys(sessions).map((id) => + toOpencodeV2Session({ id, parentId: null, directory: "/work/project", agent: "build" })), + messages: (sessionId) => toOpencodeV2MessageView(sessions[sessionId] ?? []), + parts: (messageId) => { + const row = rowsByMessage.get(messageId); + return row ? toOpencodeV2MessageParts(row) : []; + } + }; +} diff --git a/App/backend/src/adapters/outbound/agent-source/opencode/db-reader.ts b/App/backend/src/adapters/outbound/agent-source/opencode/db-reader.ts index 791a8748d..240c5ec08 100644 --- a/App/backend/src/adapters/outbound/agent-source/opencode/db-reader.ts +++ b/App/backend/src/adapters/outbound/agent-source/opencode/db-reader.ts @@ -2,8 +2,13 @@ import { DatabaseSync } from "node:sqlite"; import { readOpencodeSessions, + toOpencodeV2MessageParts, + toOpencodeV2MessageView, + toOpencodeV2Session, type OpencodeSession, type OpencodeSource, + type OpencodeV2MessageRow, + type OpencodeV2SessionRow, type RawSourceMessage } from "@memmy/agent-source-core"; @@ -41,7 +46,14 @@ export async function* streamOpencodeDatabase(path: string, signal?: AbortSignal /** A database without the conversation tables is not a conversation store. */ function createSource(db: DatabaseSync): OpencodeSource | null { - if (!hasTable(db, "session") || !hasTable(db, "message") || !hasTable(db, "part")) return null; + if (hasTable(db, "session") && hasTable(db, "message") && hasTable(db, "part")) { + return createOpencodeV1Source(db); + } + // OpenCode V2 renamed the conversation tables and inlined message parts. + return createOpencodeV2Source(db); +} + +function createOpencodeV1Source(db: DatabaseSync): OpencodeSource { const columns = tableColumns(db, "session"); const sessions = db.prepare(`SELECT id, parent_id AS parentId, directory${columns.has("agent") ? ", agent" : ""}${columns.has("revert") ? ", revert" : ""} FROM session ORDER BY time_created ASC, id ASC`); const messages = db.prepare("SELECT id, data FROM message WHERE session_id = ? ORDER BY time_created ASC, id ASC"); @@ -55,6 +67,31 @@ function createSource(db: DatabaseSync): OpencodeSource | null { }; } +/** Projects OpenCode V2 (`session_v2`/`session_message`) rows through the shared V2 mappers. */ +function createOpencodeV2Source(db: DatabaseSync): OpencodeSource | null { + if (!hasTable(db, "session_v2") || !hasTable(db, "session_message")) return null; + const sessions = db.prepare("SELECT id, parent_id AS parentId, directory, agent, revert FROM session_v2 ORDER BY time_created ASC, id ASC"); + const messages = db.prepare("SELECT id, type, seq, data FROM session_message WHERE session_id = ? ORDER BY seq ASC, id ASC"); + const messageById = db.prepare("SELECT id, type, seq, data FROM session_message WHERE id = ?"); + const toRow = (row: { id: string; type: string; seq: number; data: string }): OpencodeV2MessageRow => ({ + id: row.id, + type: row.type, + seq: Number(row.seq), + data: parseJson(row.data) + }); + return { + sessions: () => (sessions.all() as unknown as Array>) + .map((row) => toOpencodeV2Session(row as unknown as OpencodeV2SessionRow)), + messages: (sessionId) => toOpencodeV2MessageView( + (messages.all(sessionId) as unknown as Array<{ id: string; type: string; seq: number; data: string }>).map(toRow) + ), + parts: (messageId) => { + const row = messageById.get(messageId) as unknown as { id: string; type: string; seq: number; data: string } | undefined; + return row ? toOpencodeV2MessageParts(toRow(row)) : []; + } + }; +} + function toOpencodeSession(row: Record): OpencodeSession { return { id: String(row.id), diff --git a/App/backend/src/adapters/outbound/skill-writer/templates/memmy-opencode-plugin.ts b/App/backend/src/adapters/outbound/skill-writer/templates/memmy-opencode-plugin.ts index 9ace951b2..e29645dc8 100644 --- a/App/backend/src/adapters/outbound/skill-writer/templates/memmy-opencode-plugin.ts +++ b/App/backend/src/adapters/outbound/skill-writer/templates/memmy-opencode-plugin.ts @@ -5,7 +5,6 @@ export function renderMemmyOpencodePlugin(): string { return String.raw`import { readFile } from "node:fs/promises"; import { homedir } from "node:os"; import { join } from "node:path"; -import { tool } from "@opencode-ai/plugin"; import { closeRuntimeSession, completeSourceTurn, @@ -27,6 +26,9 @@ const RESUME_CONTEXT_MAX_CHARS = 24000; const TOOL_OUTPUT_MAX_CHARS = 12000; export const MemmyMemoryPlugin = async ({ client, directory, worktree }) => { + // Loaded lazily: OpenCode V1 provides this package at runtime, while V2 never runs + // this path and a static import would fail resolution before the default export loads. + const { tool } = await import("@opencode-ai/plugin"); const sessionCache = new Map(); const l3InjectOnce = new Map(); const pendingTurns = new Map(); @@ -1046,6 +1048,396 @@ function hashText(value) { function normalizeText(value) { return typeof value === "string" ? value.trim() : ""; } + +// --- OpenCode V2 entrypoint ------------------------------------------------- +// +// OpenCode V2 calls setup() and never runs the V1 function above; V1 (>= 1.18.29) +// calls server() instead. The default export below carries both, so one installed +// file supports either generation. The V2 plugin context APIs are feature-detected +// and every Memmy failure degrades to a log line so a turn is never interrupted. + +export default { + id: "memmy-memory", + async server(context) { + return MemmyMemoryPlugin(context); + }, + async setup(context) { + const directory = context && context.location && typeof context.location.directory === "string" + ? context.location.directory + : null; + const sessionCache = new Map(); + const pendingTurns = new Map(); + const pendingResumeSelections = new Map(); + const latestRequests = new Map(); + const l3InjectOnce = new Map(); + const firstTurnSeen = new Set(); + const captureJobs = new Set(); + const controller = new AbortController(); + + function log(level, message, extra) { + try { + const line = "[memmy-memory] " + message + (extra ? " " + JSON.stringify(extra) : ""); + if (level === "warn" || level === "error") { + console.error(line); + } else { + console.log(line); + } + } catch { + // Logging must never interrupt an OpenCode turn. + } + } + + async function ensureSession(externalSessionId, agent) { + const profileId = normalizeText(agent) || "main"; + const cacheKey = externalSessionId + ":" + profileId; + const cached = sessionCache.get(cacheKey); + if (cached) { + return cached; + } + const opened = await openRuntimeSession({ + configUrl: CONFIG_URL, + source: SOURCE, + adapterId: "memmy-opencode-plugin", + profileId, + sessionKey: "opencode-memory-" + externalSessionId, + workspaceRoot: directory, + transition: "allow_legacy_rollover" + }); + if (!opened) throw new Error("Memmy session unavailable"); + sessionCache.set(cacheKey, opened); + return opened; + } + + async function beginTurn(sessionID, messageID, agent, query) { + const cleanQuery = sanitizeUserText(query); + if (!cleanQuery) { + return ""; + } + latestRequests.set(sessionID, cleanQuery); + let recalledContext = ""; + try { + const memmy = await createMemmyClient(); + const runtimeSession = await ensureSession(sessionID, agent); + const sessionId = runtimeSession.sessionId; + const requestedTurnId = normalizeText(messageID); + const turn = await startRuntimeTurn( + runtimeSession, + requestedTurnId || "opencode-turn-" + hashText([sessionId, cleanQuery, String(Date.now())].join("\u0000")), + cleanQuery + ); + const turnId = normalizeText(turn && turn.turnId) || requestedTurnId || + "opencode-fallback-" + hashText([sessionId, sessionID, cleanQuery].join("\u0000")); + pendingTurns.set(sessionID, { + externalSessionId: sessionID, + sessionId, + turnId, + sourceMemoryIds: Array.isArray(turn && turn.sourceMemoryIds) ? turn.sourceMemoryIds : undefined, + userMessageId: requestedTurnId || turnId, + profileId: normalizeText(agent) || "main" + }); + recalledContext = normalizeText(turn && turn.injectedContext && turn.injectedContext.markdown); + } catch (error) { + log("warn", "Memmy recall failed", { error: formatError(error), sessionID }); + } + return recalledContext; + } + + function queueTurnCompletion(sessionID) { + const pending = pendingTurns.get(sessionID); + if (!pending) { + return; + } + pendingTurns.delete(sessionID); + const job = completeTurn(pending) + .catch((error) => { + log("warn", "Memmy turn capture failed", { error: formatError(error), sessionID }); + }) + .finally(() => { + captureJobs.delete(job); + }); + captureJobs.add(job); + } + + // Capture reads the turn back from the OpenCode database through the same bridge + // reader the offline scan uses, so both channels resolve one turn to one identity. + async function completeTurn(pending) { + const conversationId = normalizeText(pending.externalSessionId); + const turnId = normalizeText(pending.userMessageId); + if (!conversationId || !turnId) { + return; + } + const parsed = await readOpencodeHookSourceTurn({ conversationId, turnId }); + if (!parsed.turn) { + log("warn", "Memmy turn capture skipped", { reason: parsed.reason || "identity_unresolved", sessionID: conversationId }); + return; + } + const result = await completeSourceTurn({ + configUrl: CONFIG_URL, + turn: parsed.turn, + sessionId: normalizeText(pending.sessionId) || undefined, + sourceMemoryIds: pending.sourceMemoryIds, + profileId: normalizeText(parsed.turn.profileId) || normalizeText(pending.profileId) || "main", + adapterId: "memmy-opencode-plugin" + }); + const status = normalizeText(result && result.status); + if (status !== "stored" && status !== "existing" && status !== "rejected") { + log("warn", "Memmy turn capture failed", { reason: normalizeText(result && result.reason) || status || "unexpected_response", sessionID: conversationId }); + } + } + + async function handleResumeSearch(sessionID, query, setText) { + if (!query) { + setText(renderCommandResult("Usage: /memmy-resume ")); + return; + } + if (query === "cancel") { + pendingResumeSelections.delete(sessionID); + setText(renderCommandResult("Memmy resume selection cancelled.")); + return; + } + try { + const memmy = await createMemmyClient(); + const result = await memmy.post("/api/v1/memory/search", { + query, + layers: ["L1"], + limit: SEARCH_LIMIT, + verbose: true + }); + const candidates = await buildEpisodeCandidates(memmy, result); + pendingResumeSelections.set(sessionID, { createdAt: Date.now(), candidates }); + setText(renderCommandResult(formatResumeSearchResult(query, candidates))); + } catch (error) { + setText(renderCommandResult("Memmy resume search failed: " + formatError(error))); + } + } + + async function handleResumeSelection(sessionID, messageID, agent, selection, setText) { + const selected = resolveResumeSelection(pendingResumeSelections.get(sessionID), selection); + if (!selected) { + setText(renderCommandResult("No active Memmy resume selection. Run /memmy-resume first.")); + return; + } + try { + const memmy = await createMemmyClient(); + const detail = await memmy.get("/api/v1/memory/" + encodeURIComponent(selected.episodeId)); + pendingResumeSelections.delete(sessionID); + const context = buildResumeContext(selected, detail); + const title = normalizeText(detail && detail.title) || normalizeText(selected.title) || selected.episodeId; + const query = "Continue Memmy episode " + selected.episodeId + ": " + title; + const recalled = await beginTurn(sessionID, messageID, agent, query); + const packet = [ + context ? "Selected episode context:\n" + context : "", + recalled ? "Additional relevant memory:\n" + recalled : "" + ].filter(Boolean).join("\n\n"); + setText(renderMemmyContextPacket(packet || context, "resume", query)); + } catch (error) { + setText(renderCommandResult("Memmy resume selection failed: " + formatError(error))); + } + } + + function agentFromEvent(event) { + const candidate = event && (event.agent || (event.prompt && event.prompt.agent)); + return typeof candidate === "string" ? candidate : ""; + } + + await context.tool.transform((editor) => { + editor.add({ + name: "memmy_memory_search", + description: "Search Memmy local memory for relevant facts, preferences, decisions, procedures, and prior tasks.", + input: { + type: "object", + properties: { + query: { type: "string" }, + layers: { type: "array", items: { type: "string", enum: ["L1", "L2", "L3", "Skill"] } } + }, + required: ["query"], + additionalProperties: false + }, + async execute(args, toolContext) { + const memmy = await createMemmyClient(); + const result = await memmy.post("/api/v1/memory/search", { + query: normalizeText(args && args.query), + layers: Array.isArray(args && args.layers) ? args.layers : undefined + }); + return { content: renderMemmyContextPacket( + formatSearchResult(result), + "tool_search", + latestRequests.get(toolContext.sessionID) || normalizeText(args && args.query) + ) }; + } + }); + editor.add({ + name: "memmy_memory_get", + description: "Read one Memmy memory detail by id.", + input: { + type: "object", + properties: { id: { type: "string" } }, + required: ["id"], + additionalProperties: false + }, + async execute(args, toolContext) { + const id = normalizeText(args && args.id); + if (!id) { + throw new Error("Missing required parameter: id"); + } + const memmy = await createMemmyClient(); + const result = await memmy.get("/api/v1/memory/" + encodeURIComponent(id)); + return { content: renderMemmyContextPacket( + formatMemoryDetail(result), + "tool_get", + latestRequests.get(toolContext.sessionID) || "Read Memmy memory " + id + ) }; + } + }); + editor.add({ + name: "memmy_memory_add", + description: "Write an important durable fact, preference, decision, reusable procedure, or unresolved follow-up into Memmy.", + input: { + type: "object", + properties: { + content: { type: "string" }, + title: { type: "string" }, + tags: { type: "array", items: { type: "string" } }, + layer: { type: "string", enum: ["L1", "L2", "L3", "Skill"] } + }, + required: ["content"], + additionalProperties: false + }, + async execute(args, toolContext) { + const content = sanitizeCaptureText(args && args.content); + if (!content) { + throw new Error("Missing required parameter: content"); + } + const memmy = await createMemmyClient(); + const sessionId = (await ensureSession(toolContext.sessionID, toolContext.agent)).sessionId; + const result = await memmy.post("/api/v1/memory/add", { + content, + title: normalizeText(args && args.title) || undefined, + tags: Array.isArray(args && args.tags) ? args.tags.filter((item) => typeof item === "string") : undefined, + layer: normalizeText(args && args.layer) || "L1", + sessionId + }); + return { content: "Stored Memmy memory " + normalizeText(result && result.id) + ": " + + (normalizeText(result && result.summary) || content) }; + } + }); + }); + + await context.session.hook("prompt", async (event) => { + try { + const prompt = event && event.prompt; + if (!prompt || typeof prompt.text !== "string") { + return; + } + const sessionID = normalizeText(event.sessionID); + const messageID = normalizeText(event.messageID); + const rawPrompt = prompt.text; + + // A previous turn in this session may not have been captured yet. + queueTurnCompletion(sessionID); + + const commandArguments = parseResumeCommandArguments(rawPrompt); + const selection = parseResumeSelection(commandArguments === null ? rawPrompt : commandArguments); + const hasPendingSelection = Boolean(resolveResumeSelection(pendingResumeSelections.get(sessionID), selection)); + const setText = (text) => { prompt.text = text; }; + const agent = agentFromEvent(event); + + if (selection && (commandArguments !== null || hasPendingSelection)) { + await handleResumeSelection(sessionID, messageID, agent, selection, setText); + return; + } + if (commandArguments !== null) { + await handleResumeSearch(sessionID, normalizeText(commandArguments), setText); + return; + } + + if (!firstTurnSeen.has(sessionID)) { + firstTurnSeen.add(sessionID); + try { + const runtimeSession = await ensureSession(sessionID, agent); + const loaded = await loadRuntimeL3(runtimeSession); + if (loaded && loaded.additionalContext) l3InjectOnce.set(sessionID, normalizeText(loaded.additionalContext)); + } catch (error) { + log("warn", "Memmy L3 preload failed", { error: formatError(error), sessionID }); + } + } + const l3Context = l3InjectOnce.get(sessionID) || ""; + l3InjectOnce.delete(sessionID); + const recalled = await beginTurn(sessionID, messageID, agent, rawPrompt); + const packet = [ + l3Context ? "Additional relevant memory:\n" + l3Context : "", + recalled ? "Additional relevant memory:\n" + recalled : "" + ].filter(Boolean).join("\n\n"); + if (packet) { + setText(renderMemmyContextPacket(packet, "turn_start", rawPrompt)); + } + } catch (error) { + log("warn", "Memmy prompt hook failed", { error: formatError(error) }); + } + }); + + void (async () => { + try { + for await (const event of context.event.subscribe({ signal: controller.signal })) { + const type = event && event.type; + const properties = event && event.properties && typeof event.properties === "object" + ? event.properties + : (event && event.data && typeof event.data === "object" ? event.data : {}); + if (type === "session.compacted") { + const sessionID = normalizeText(properties.sessionID || properties.id); + for (const key of [...sessionCache.keys()]) { + if (key !== sessionID + ":main" && !key.startsWith(sessionID + ":")) continue; + const runtimeSession = sessionCache.get(key); + await notifyRuntimeBoundary(runtimeSession, "token_compaction").catch(() => undefined); + const loaded = await loadRuntimeL3(runtimeSession).catch(() => null); + if (loaded && loaded.additionalContext) l3InjectOnce.set(sessionID, normalizeText(loaded.additionalContext)); + } + continue; + } + if (type === "session.deleted") { + const sessionID = normalizeText(properties.sessionID || properties.id); + queueTurnCompletion(sessionID); + for (const key of [...sessionCache.keys()]) { + if (!key.startsWith(sessionID + ":")) continue; + const runtimeSession = sessionCache.get(key); + sessionCache.delete(key); + await closeRuntimeSession(runtimeSession).catch(() => undefined); + } + l3InjectOnce.delete(sessionID); + firstTurnSeen.delete(sessionID); + continue; + } + if (type === "session.idle") { + queueTurnCompletion(normalizeText(properties.sessionID || properties.id)); + } + } + } catch (error) { + if (!controller.signal.aborted) { + log("warn", "Memmy event stream failed", { error: formatError(error) }); + } + } + })(); + + log("info", "Memmy plugin active (OpenCode V2)", { directory }); + + return () => { + controller.abort(); + for (const sessionID of [...pendingTurns.keys()]) { + queueTurnCompletion(sessionID); + } + // Snapshot before clearing: the close loop runs after the last capture settles. + const sessionsToClose = [...sessionCache.values()]; + void Promise.allSettled([...captureJobs]).then(() => { + for (const session of sessionsToClose) { + void closeRuntimeSession(session).catch(() => undefined); + } + }); + sessionCache.clear(); + l3InjectOnce.clear(); + firstTurnSeen.clear(); + }; + } +}; `; } diff --git a/App/backend/src/adapters/outbound/skill-writer/workspace-bridge/runtime.ts b/App/backend/src/adapters/outbound/skill-writer/workspace-bridge/runtime.ts index 32b881c67..32944fc09 100644 --- a/App/backend/src/adapters/outbound/skill-writer/workspace-bridge/runtime.ts +++ b/App/backend/src/adapters/outbound/skill-writer/workspace-bridge/runtime.ts @@ -9,9 +9,14 @@ import { readDeepseekHarnessSourceTurn, readOpenclawSourceTurn, readOpencodeSourceTurn, + toOpencodeV2MessageParts, + toOpencodeV2MessageView, + toOpencodeV2Session, type CursorVscdbSource, type OpenclawTranscriptSource, type OpencodeSource, + type OpencodeV2MessageRow, + type OpencodeV2SessionRow, type SourceTurn } from "@memmy/agent-source-core"; export { readClaudeCodeSourceTurn, readCodexSourceTurn } from "@memmy/agent-source-core"; @@ -412,10 +417,6 @@ export async function readOpencodeHookSourceTurn(input: { return { turn: null, reason: "source_store_unavailable" }; } try { - const columns = new Set((db.prepare("PRAGMA table_info(session)").all() as Array<{ name: string }>).map((row) => row.name)); - const sessions = db.prepare(`SELECT id, parent_id AS parentId, directory${columns.has("agent") ? ", agent" : ""}${columns.has("revert") ? ", revert" : ""} FROM session WHERE id = ?`); - const messages = db.prepare("SELECT id, data FROM message WHERE session_id = ? ORDER BY time_created ASC, id ASC"); - const parts = db.prepare("SELECT id, data FROM part WHERE message_id = ? ORDER BY time_created ASC, id ASC"); const parse = (value: unknown): unknown => { if (typeof value !== "string") return undefined; try { @@ -424,8 +425,31 @@ export async function readOpencodeHookSourceTurn(input: { return undefined; } }; - const source: OpencodeSource = { - sessions: () => (sessions.all(input.conversationId) as unknown as Array>).map((row) => { + const source = createOpencodeHookSource(db, input.conversationId, parse); + if (!source) return { turn: null, reason: "source_store_unavailable" }; + return await readOpencodeSourceTurn(source, input); + } finally { + db.close(); + } +} + +/** + * One row-access layer per OpenCode schema generation. V1 keeps part rows and the + * assistant `parentID` link; V2 inlines content in `session_message` and drops the + * link, so its rows are projected onto the shared turn parser through the V2 mappers. + */ +function createOpencodeHookSource( + db: NodeSqliteDatabase, + conversationId: string, + parse: (value: unknown) => unknown +): OpencodeSource | null { + if (hasSqliteTable(db, "session") && hasSqliteTable(db, "message") && hasSqliteTable(db, "part")) { + const columns = new Set((db.prepare("PRAGMA table_info(session)").all() as Array<{ name: string }>).map((row) => row.name)); + const sessions = db.prepare(`SELECT id, parent_id AS parentId, directory${columns.has("agent") ? ", agent" : ""}${columns.has("revert") ? ", revert" : ""} FROM session WHERE id = ?`); + const messages = db.prepare("SELECT id, data FROM message WHERE session_id = ? ORDER BY time_created ASC, id ASC"); + const parts = db.prepare("SELECT id, data FROM part WHERE message_id = ? ORDER BY time_created ASC, id ASC"); + return { + sessions: () => (sessions.all(conversationId) as unknown as Array>).map((row) => { const revert = typeof row.revert === "string" ? parse(row.revert) : row.revert; const messageId = revert && typeof revert === "object" && !Array.isArray(revert) ? (revert as { messageID?: unknown }).messageID @@ -443,10 +467,32 @@ export async function readOpencodeHookSourceTurn(input: { parts: (messageId) => (parts.all(messageId) as unknown as Array<{ id: string; data: string }>) .map((row) => ({ id: row.id, data: parse(row.data) })) }; - return await readOpencodeSourceTurn(source, input); - } finally { - db.close(); } + if (!hasSqliteTable(db, "session_v2") || !hasSqliteTable(db, "session_message")) return null; + const sessions = db.prepare("SELECT id, parent_id AS parentId, directory, agent, revert FROM session_v2 WHERE id = ?"); + const messages = db.prepare("SELECT id, type, seq, data FROM session_message WHERE session_id = ? ORDER BY seq ASC, id ASC"); + const messageById = db.prepare("SELECT id, type, seq, data FROM session_message WHERE id = ?"); + const toRow = (row: { id: string; type: string; seq: number; data: string }): OpencodeV2MessageRow => ({ + id: row.id, + type: row.type, + seq: Number(row.seq), + data: parse(row.data) + }); + return { + sessions: () => (sessions.all(conversationId) as unknown as Array>) + .map((row) => toOpencodeV2Session(row as unknown as OpencodeV2SessionRow)), + messages: (sessionId) => toOpencodeV2MessageView( + (messages.all(sessionId) as unknown as Array<{ id: string; type: string; seq: number; data: string }>).map(toRow) + ), + parts: (messageId) => { + const row = messageById.get(messageId) as unknown as { id: string; type: string; seq: number; data: string } | undefined; + return row ? toOpencodeV2MessageParts(toRow(row)) : []; + } + }; +} + +function hasSqliteTable(db: NodeSqliteDatabase, tableName: string): boolean { + return Boolean(db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?").get(tableName)); } /** diff --git a/Memory/src/agent-source/adapters/opencode/db-reader.ts b/Memory/src/agent-source/adapters/opencode/db-reader.ts index cf5257670..e308807a6 100644 --- a/Memory/src/agent-source/adapters/opencode/db-reader.ts +++ b/Memory/src/agent-source/adapters/opencode/db-reader.ts @@ -2,8 +2,13 @@ import Database from "better-sqlite3"; import { readOpencodeSessions, + toOpencodeV2MessageParts, + toOpencodeV2MessageView, + toOpencodeV2Session, type OpencodeSession, type OpencodeSource, + type OpencodeV2MessageRow, + type OpencodeV2SessionRow, type RawSourceMessage } from "@memmy/agent-source-core"; @@ -41,7 +46,14 @@ export async function* streamOpencodeDatabase(path: string, signal?: AbortSignal /** A database without the conversation tables is not a conversation store. */ function createSource(db: Database.Database): OpencodeSource | null { - if (!hasTable(db, "session") || !hasTable(db, "message") || !hasTable(db, "part")) return null; + if (hasTable(db, "session") && hasTable(db, "message") && hasTable(db, "part")) { + return createOpencodeV1Source(db); + } + // OpenCode V2 renamed the conversation tables and inlined message parts. + return createOpencodeV2Source(db); +} + +function createOpencodeV1Source(db: Database.Database): OpencodeSource { const columns = tableColumns(db, "session"); const sessions = db.prepare(`SELECT id, parent_id AS parentId, directory${columns.has("agent") ? ", agent" : ""}${columns.has("revert") ? ", revert" : ""} FROM session ORDER BY time_created ASC, id ASC`); const messages = db.prepare("SELECT id, data FROM message WHERE session_id = ? ORDER BY time_created ASC, id ASC"); @@ -55,6 +67,31 @@ function createSource(db: Database.Database): OpencodeSource | null { }; } +/** Projects OpenCode V2 (`session_v2`/`session_message`) rows through the shared V2 mappers. */ +function createOpencodeV2Source(db: Database.Database): OpencodeSource | null { + if (!hasTable(db, "session_v2") || !hasTable(db, "session_message")) return null; + const sessions = db.prepare("SELECT id, parent_id AS parentId, directory, agent, revert FROM session_v2 ORDER BY time_created ASC, id ASC"); + const messages = db.prepare("SELECT id, type, seq, data FROM session_message WHERE session_id = ? ORDER BY seq ASC, id ASC"); + const messageById = db.prepare("SELECT id, type, seq, data FROM session_message WHERE id = ?"); + const toRow = (row: { id: string; type: string; seq: number; data: string }): OpencodeV2MessageRow => ({ + id: row.id, + type: row.type, + seq: Number(row.seq), + data: parseJson(row.data) + }); + return { + sessions: () => (sessions.all() as Array>) + .map((row) => toOpencodeV2Session(row as unknown as OpencodeV2SessionRow)), + messages: (sessionId) => toOpencodeV2MessageView( + (messages.all(sessionId) as Array<{ id: string; type: string; seq: number; data: string }>).map(toRow) + ), + parts: (messageId) => { + const row = messageById.get(messageId) as { id: string; type: string; seq: number; data: string } | undefined; + return row ? toOpencodeV2MessageParts(toRow(row)) : []; + } + }; +} + function toOpencodeSession(row: Record): OpencodeSession { return { id: String(row.id), diff --git a/Memory/src/agent-source/integration/templates/memmy-opencode-plugin.ts b/Memory/src/agent-source/integration/templates/memmy-opencode-plugin.ts index 9ace951b2..e29645dc8 100644 --- a/Memory/src/agent-source/integration/templates/memmy-opencode-plugin.ts +++ b/Memory/src/agent-source/integration/templates/memmy-opencode-plugin.ts @@ -5,7 +5,6 @@ export function renderMemmyOpencodePlugin(): string { return String.raw`import { readFile } from "node:fs/promises"; import { homedir } from "node:os"; import { join } from "node:path"; -import { tool } from "@opencode-ai/plugin"; import { closeRuntimeSession, completeSourceTurn, @@ -27,6 +26,9 @@ const RESUME_CONTEXT_MAX_CHARS = 24000; const TOOL_OUTPUT_MAX_CHARS = 12000; export const MemmyMemoryPlugin = async ({ client, directory, worktree }) => { + // Loaded lazily: OpenCode V1 provides this package at runtime, while V2 never runs + // this path and a static import would fail resolution before the default export loads. + const { tool } = await import("@opencode-ai/plugin"); const sessionCache = new Map(); const l3InjectOnce = new Map(); const pendingTurns = new Map(); @@ -1046,6 +1048,396 @@ function hashText(value) { function normalizeText(value) { return typeof value === "string" ? value.trim() : ""; } + +// --- OpenCode V2 entrypoint ------------------------------------------------- +// +// OpenCode V2 calls setup() and never runs the V1 function above; V1 (>= 1.18.29) +// calls server() instead. The default export below carries both, so one installed +// file supports either generation. The V2 plugin context APIs are feature-detected +// and every Memmy failure degrades to a log line so a turn is never interrupted. + +export default { + id: "memmy-memory", + async server(context) { + return MemmyMemoryPlugin(context); + }, + async setup(context) { + const directory = context && context.location && typeof context.location.directory === "string" + ? context.location.directory + : null; + const sessionCache = new Map(); + const pendingTurns = new Map(); + const pendingResumeSelections = new Map(); + const latestRequests = new Map(); + const l3InjectOnce = new Map(); + const firstTurnSeen = new Set(); + const captureJobs = new Set(); + const controller = new AbortController(); + + function log(level, message, extra) { + try { + const line = "[memmy-memory] " + message + (extra ? " " + JSON.stringify(extra) : ""); + if (level === "warn" || level === "error") { + console.error(line); + } else { + console.log(line); + } + } catch { + // Logging must never interrupt an OpenCode turn. + } + } + + async function ensureSession(externalSessionId, agent) { + const profileId = normalizeText(agent) || "main"; + const cacheKey = externalSessionId + ":" + profileId; + const cached = sessionCache.get(cacheKey); + if (cached) { + return cached; + } + const opened = await openRuntimeSession({ + configUrl: CONFIG_URL, + source: SOURCE, + adapterId: "memmy-opencode-plugin", + profileId, + sessionKey: "opencode-memory-" + externalSessionId, + workspaceRoot: directory, + transition: "allow_legacy_rollover" + }); + if (!opened) throw new Error("Memmy session unavailable"); + sessionCache.set(cacheKey, opened); + return opened; + } + + async function beginTurn(sessionID, messageID, agent, query) { + const cleanQuery = sanitizeUserText(query); + if (!cleanQuery) { + return ""; + } + latestRequests.set(sessionID, cleanQuery); + let recalledContext = ""; + try { + const memmy = await createMemmyClient(); + const runtimeSession = await ensureSession(sessionID, agent); + const sessionId = runtimeSession.sessionId; + const requestedTurnId = normalizeText(messageID); + const turn = await startRuntimeTurn( + runtimeSession, + requestedTurnId || "opencode-turn-" + hashText([sessionId, cleanQuery, String(Date.now())].join("\u0000")), + cleanQuery + ); + const turnId = normalizeText(turn && turn.turnId) || requestedTurnId || + "opencode-fallback-" + hashText([sessionId, sessionID, cleanQuery].join("\u0000")); + pendingTurns.set(sessionID, { + externalSessionId: sessionID, + sessionId, + turnId, + sourceMemoryIds: Array.isArray(turn && turn.sourceMemoryIds) ? turn.sourceMemoryIds : undefined, + userMessageId: requestedTurnId || turnId, + profileId: normalizeText(agent) || "main" + }); + recalledContext = normalizeText(turn && turn.injectedContext && turn.injectedContext.markdown); + } catch (error) { + log("warn", "Memmy recall failed", { error: formatError(error), sessionID }); + } + return recalledContext; + } + + function queueTurnCompletion(sessionID) { + const pending = pendingTurns.get(sessionID); + if (!pending) { + return; + } + pendingTurns.delete(sessionID); + const job = completeTurn(pending) + .catch((error) => { + log("warn", "Memmy turn capture failed", { error: formatError(error), sessionID }); + }) + .finally(() => { + captureJobs.delete(job); + }); + captureJobs.add(job); + } + + // Capture reads the turn back from the OpenCode database through the same bridge + // reader the offline scan uses, so both channels resolve one turn to one identity. + async function completeTurn(pending) { + const conversationId = normalizeText(pending.externalSessionId); + const turnId = normalizeText(pending.userMessageId); + if (!conversationId || !turnId) { + return; + } + const parsed = await readOpencodeHookSourceTurn({ conversationId, turnId }); + if (!parsed.turn) { + log("warn", "Memmy turn capture skipped", { reason: parsed.reason || "identity_unresolved", sessionID: conversationId }); + return; + } + const result = await completeSourceTurn({ + configUrl: CONFIG_URL, + turn: parsed.turn, + sessionId: normalizeText(pending.sessionId) || undefined, + sourceMemoryIds: pending.sourceMemoryIds, + profileId: normalizeText(parsed.turn.profileId) || normalizeText(pending.profileId) || "main", + adapterId: "memmy-opencode-plugin" + }); + const status = normalizeText(result && result.status); + if (status !== "stored" && status !== "existing" && status !== "rejected") { + log("warn", "Memmy turn capture failed", { reason: normalizeText(result && result.reason) || status || "unexpected_response", sessionID: conversationId }); + } + } + + async function handleResumeSearch(sessionID, query, setText) { + if (!query) { + setText(renderCommandResult("Usage: /memmy-resume ")); + return; + } + if (query === "cancel") { + pendingResumeSelections.delete(sessionID); + setText(renderCommandResult("Memmy resume selection cancelled.")); + return; + } + try { + const memmy = await createMemmyClient(); + const result = await memmy.post("/api/v1/memory/search", { + query, + layers: ["L1"], + limit: SEARCH_LIMIT, + verbose: true + }); + const candidates = await buildEpisodeCandidates(memmy, result); + pendingResumeSelections.set(sessionID, { createdAt: Date.now(), candidates }); + setText(renderCommandResult(formatResumeSearchResult(query, candidates))); + } catch (error) { + setText(renderCommandResult("Memmy resume search failed: " + formatError(error))); + } + } + + async function handleResumeSelection(sessionID, messageID, agent, selection, setText) { + const selected = resolveResumeSelection(pendingResumeSelections.get(sessionID), selection); + if (!selected) { + setText(renderCommandResult("No active Memmy resume selection. Run /memmy-resume first.")); + return; + } + try { + const memmy = await createMemmyClient(); + const detail = await memmy.get("/api/v1/memory/" + encodeURIComponent(selected.episodeId)); + pendingResumeSelections.delete(sessionID); + const context = buildResumeContext(selected, detail); + const title = normalizeText(detail && detail.title) || normalizeText(selected.title) || selected.episodeId; + const query = "Continue Memmy episode " + selected.episodeId + ": " + title; + const recalled = await beginTurn(sessionID, messageID, agent, query); + const packet = [ + context ? "Selected episode context:\n" + context : "", + recalled ? "Additional relevant memory:\n" + recalled : "" + ].filter(Boolean).join("\n\n"); + setText(renderMemmyContextPacket(packet || context, "resume", query)); + } catch (error) { + setText(renderCommandResult("Memmy resume selection failed: " + formatError(error))); + } + } + + function agentFromEvent(event) { + const candidate = event && (event.agent || (event.prompt && event.prompt.agent)); + return typeof candidate === "string" ? candidate : ""; + } + + await context.tool.transform((editor) => { + editor.add({ + name: "memmy_memory_search", + description: "Search Memmy local memory for relevant facts, preferences, decisions, procedures, and prior tasks.", + input: { + type: "object", + properties: { + query: { type: "string" }, + layers: { type: "array", items: { type: "string", enum: ["L1", "L2", "L3", "Skill"] } } + }, + required: ["query"], + additionalProperties: false + }, + async execute(args, toolContext) { + const memmy = await createMemmyClient(); + const result = await memmy.post("/api/v1/memory/search", { + query: normalizeText(args && args.query), + layers: Array.isArray(args && args.layers) ? args.layers : undefined + }); + return { content: renderMemmyContextPacket( + formatSearchResult(result), + "tool_search", + latestRequests.get(toolContext.sessionID) || normalizeText(args && args.query) + ) }; + } + }); + editor.add({ + name: "memmy_memory_get", + description: "Read one Memmy memory detail by id.", + input: { + type: "object", + properties: { id: { type: "string" } }, + required: ["id"], + additionalProperties: false + }, + async execute(args, toolContext) { + const id = normalizeText(args && args.id); + if (!id) { + throw new Error("Missing required parameter: id"); + } + const memmy = await createMemmyClient(); + const result = await memmy.get("/api/v1/memory/" + encodeURIComponent(id)); + return { content: renderMemmyContextPacket( + formatMemoryDetail(result), + "tool_get", + latestRequests.get(toolContext.sessionID) || "Read Memmy memory " + id + ) }; + } + }); + editor.add({ + name: "memmy_memory_add", + description: "Write an important durable fact, preference, decision, reusable procedure, or unresolved follow-up into Memmy.", + input: { + type: "object", + properties: { + content: { type: "string" }, + title: { type: "string" }, + tags: { type: "array", items: { type: "string" } }, + layer: { type: "string", enum: ["L1", "L2", "L3", "Skill"] } + }, + required: ["content"], + additionalProperties: false + }, + async execute(args, toolContext) { + const content = sanitizeCaptureText(args && args.content); + if (!content) { + throw new Error("Missing required parameter: content"); + } + const memmy = await createMemmyClient(); + const sessionId = (await ensureSession(toolContext.sessionID, toolContext.agent)).sessionId; + const result = await memmy.post("/api/v1/memory/add", { + content, + title: normalizeText(args && args.title) || undefined, + tags: Array.isArray(args && args.tags) ? args.tags.filter((item) => typeof item === "string") : undefined, + layer: normalizeText(args && args.layer) || "L1", + sessionId + }); + return { content: "Stored Memmy memory " + normalizeText(result && result.id) + ": " + + (normalizeText(result && result.summary) || content) }; + } + }); + }); + + await context.session.hook("prompt", async (event) => { + try { + const prompt = event && event.prompt; + if (!prompt || typeof prompt.text !== "string") { + return; + } + const sessionID = normalizeText(event.sessionID); + const messageID = normalizeText(event.messageID); + const rawPrompt = prompt.text; + + // A previous turn in this session may not have been captured yet. + queueTurnCompletion(sessionID); + + const commandArguments = parseResumeCommandArguments(rawPrompt); + const selection = parseResumeSelection(commandArguments === null ? rawPrompt : commandArguments); + const hasPendingSelection = Boolean(resolveResumeSelection(pendingResumeSelections.get(sessionID), selection)); + const setText = (text) => { prompt.text = text; }; + const agent = agentFromEvent(event); + + if (selection && (commandArguments !== null || hasPendingSelection)) { + await handleResumeSelection(sessionID, messageID, agent, selection, setText); + return; + } + if (commandArguments !== null) { + await handleResumeSearch(sessionID, normalizeText(commandArguments), setText); + return; + } + + if (!firstTurnSeen.has(sessionID)) { + firstTurnSeen.add(sessionID); + try { + const runtimeSession = await ensureSession(sessionID, agent); + const loaded = await loadRuntimeL3(runtimeSession); + if (loaded && loaded.additionalContext) l3InjectOnce.set(sessionID, normalizeText(loaded.additionalContext)); + } catch (error) { + log("warn", "Memmy L3 preload failed", { error: formatError(error), sessionID }); + } + } + const l3Context = l3InjectOnce.get(sessionID) || ""; + l3InjectOnce.delete(sessionID); + const recalled = await beginTurn(sessionID, messageID, agent, rawPrompt); + const packet = [ + l3Context ? "Additional relevant memory:\n" + l3Context : "", + recalled ? "Additional relevant memory:\n" + recalled : "" + ].filter(Boolean).join("\n\n"); + if (packet) { + setText(renderMemmyContextPacket(packet, "turn_start", rawPrompt)); + } + } catch (error) { + log("warn", "Memmy prompt hook failed", { error: formatError(error) }); + } + }); + + void (async () => { + try { + for await (const event of context.event.subscribe({ signal: controller.signal })) { + const type = event && event.type; + const properties = event && event.properties && typeof event.properties === "object" + ? event.properties + : (event && event.data && typeof event.data === "object" ? event.data : {}); + if (type === "session.compacted") { + const sessionID = normalizeText(properties.sessionID || properties.id); + for (const key of [...sessionCache.keys()]) { + if (key !== sessionID + ":main" && !key.startsWith(sessionID + ":")) continue; + const runtimeSession = sessionCache.get(key); + await notifyRuntimeBoundary(runtimeSession, "token_compaction").catch(() => undefined); + const loaded = await loadRuntimeL3(runtimeSession).catch(() => null); + if (loaded && loaded.additionalContext) l3InjectOnce.set(sessionID, normalizeText(loaded.additionalContext)); + } + continue; + } + if (type === "session.deleted") { + const sessionID = normalizeText(properties.sessionID || properties.id); + queueTurnCompletion(sessionID); + for (const key of [...sessionCache.keys()]) { + if (!key.startsWith(sessionID + ":")) continue; + const runtimeSession = sessionCache.get(key); + sessionCache.delete(key); + await closeRuntimeSession(runtimeSession).catch(() => undefined); + } + l3InjectOnce.delete(sessionID); + firstTurnSeen.delete(sessionID); + continue; + } + if (type === "session.idle") { + queueTurnCompletion(normalizeText(properties.sessionID || properties.id)); + } + } + } catch (error) { + if (!controller.signal.aborted) { + log("warn", "Memmy event stream failed", { error: formatError(error) }); + } + } + })(); + + log("info", "Memmy plugin active (OpenCode V2)", { directory }); + + return () => { + controller.abort(); + for (const sessionID of [...pendingTurns.keys()]) { + queueTurnCompletion(sessionID); + } + // Snapshot before clearing: the close loop runs after the last capture settles. + const sessionsToClose = [...sessionCache.values()]; + void Promise.allSettled([...captureJobs]).then(() => { + for (const session of sessionsToClose) { + void closeRuntimeSession(session).catch(() => undefined); + } + }); + sessionCache.clear(); + l3InjectOnce.clear(); + firstTurnSeen.clear(); + }; + } +}; `; } diff --git a/Memory/src/agent-source/integration/workspace-bridge/runtime.ts b/Memory/src/agent-source/integration/workspace-bridge/runtime.ts index 70e25f49e..9b9c2a345 100644 --- a/Memory/src/agent-source/integration/workspace-bridge/runtime.ts +++ b/Memory/src/agent-source/integration/workspace-bridge/runtime.ts @@ -9,9 +9,14 @@ import { readDeepseekHarnessSourceTurn, readOpenclawSourceTurn, readOpencodeSourceTurn, + toOpencodeV2MessageParts, + toOpencodeV2MessageView, + toOpencodeV2Session, type CursorVscdbSource, type OpenclawTranscriptSource, type OpencodeSource, + type OpencodeV2MessageRow, + type OpencodeV2SessionRow, type SourceTurn } from "@memmy/agent-source-core"; export { readClaudeCodeSourceTurn, readCodexSourceTurn } from "@memmy/agent-source-core"; @@ -412,10 +417,6 @@ export async function readOpencodeHookSourceTurn(input: { return { turn: null, reason: "source_store_unavailable" }; } try { - const columns = new Set((db.prepare("PRAGMA table_info(session)").all() as Array<{ name: string }>).map((row) => row.name)); - const sessions = db.prepare(`SELECT id, parent_id AS parentId, directory${columns.has("agent") ? ", agent" : ""}${columns.has("revert") ? ", revert" : ""} FROM session WHERE id = ?`); - const messages = db.prepare("SELECT id, data FROM message WHERE session_id = ? ORDER BY time_created ASC, id ASC"); - const parts = db.prepare("SELECT id, data FROM part WHERE message_id = ? ORDER BY time_created ASC, id ASC"); const parse = (value: unknown): unknown => { if (typeof value !== "string") return undefined; try { @@ -424,8 +425,31 @@ export async function readOpencodeHookSourceTurn(input: { return undefined; } }; - const source: OpencodeSource = { - sessions: () => (sessions.all(input.conversationId) as unknown as Array>).map((row) => { + const source = createOpencodeHookSource(db, input.conversationId, parse); + if (!source) return { turn: null, reason: "source_store_unavailable" }; + return await readOpencodeSourceTurn(source, input); + } finally { + db.close(); + } +} + +/** + * One row-access layer per OpenCode schema generation. V1 keeps part rows and the + * assistant `parentID` link; V2 inlines content in `session_message` and drops the + * link, so its rows are projected onto the shared turn parser through the V2 mappers. + */ +function createOpencodeHookSource( + db: NodeSqliteDatabase, + conversationId: string, + parse: (value: unknown) => unknown +): OpencodeSource | null { + if (hasSqliteTable(db, "session") && hasSqliteTable(db, "message") && hasSqliteTable(db, "part")) { + const columns = new Set((db.prepare("PRAGMA table_info(session)").all() as Array<{ name: string }>).map((row) => row.name)); + const sessions = db.prepare(`SELECT id, parent_id AS parentId, directory${columns.has("agent") ? ", agent" : ""}${columns.has("revert") ? ", revert" : ""} FROM session WHERE id = ?`); + const messages = db.prepare("SELECT id, data FROM message WHERE session_id = ? ORDER BY time_created ASC, id ASC"); + const parts = db.prepare("SELECT id, data FROM part WHERE message_id = ? ORDER BY time_created ASC, id ASC"); + return { + sessions: () => (sessions.all(conversationId) as unknown as Array>).map((row) => { const revert = typeof row.revert === "string" ? parse(row.revert) : row.revert; const messageId = revert && typeof revert === "object" && !Array.isArray(revert) ? (revert as { messageID?: unknown }).messageID @@ -443,10 +467,32 @@ export async function readOpencodeHookSourceTurn(input: { parts: (messageId) => (parts.all(messageId) as unknown as Array<{ id: string; data: string }>) .map((row) => ({ id: row.id, data: parse(row.data) })) }; - return await readOpencodeSourceTurn(source, input); - } finally { - db.close(); } + if (!hasSqliteTable(db, "session_v2") || !hasSqliteTable(db, "session_message")) return null; + const sessions = db.prepare("SELECT id, parent_id AS parentId, directory, agent, revert FROM session_v2 WHERE id = ?"); + const messages = db.prepare("SELECT id, type, seq, data FROM session_message WHERE session_id = ? ORDER BY seq ASC, id ASC"); + const messageById = db.prepare("SELECT id, type, seq, data FROM session_message WHERE id = ?"); + const toRow = (row: { id: string; type: string; seq: number; data: string }): OpencodeV2MessageRow => ({ + id: row.id, + type: row.type, + seq: Number(row.seq), + data: parse(row.data) + }); + return { + sessions: () => (sessions.all(conversationId) as unknown as Array>) + .map((row) => toOpencodeV2Session(row as unknown as OpencodeV2SessionRow)), + messages: (sessionId) => toOpencodeV2MessageView( + (messages.all(sessionId) as unknown as Array<{ id: string; type: string; seq: number; data: string }>).map(toRow) + ), + parts: (messageId) => { + const row = messageById.get(messageId) as unknown as { id: string; type: string; seq: number; data: string } | undefined; + return row ? toOpencodeV2MessageParts(toRow(row)) : []; + } + }; +} + +function hasSqliteTable(db: NodeSqliteDatabase, tableName: string): boolean { + return Boolean(db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?").get(tableName)); } /** diff --git a/Memory/tests/agent-source-opencode-reader.test.ts b/Memory/tests/agent-source-opencode-reader.test.ts index 37e091cc3..4869a3eca 100644 --- a/Memory/tests/agent-source-opencode-reader.test.ts +++ b/Memory/tests/agent-source-opencode-reader.test.ts @@ -31,6 +31,41 @@ describe("OpenCode Memory SQLite reader", () => { }); }); +describe("OpenCode Memory SQLite reader (V2 schema)", () => { + it("captures completed turns from session_v2/session_message and skips subagent branches", async () => { + directory = mkdtempSync(join(tmpdir(), "memmy-opencode-v2-")); + const databasePath = join(directory, "v2.sqlite"); + writeOpencodeV2Database(databasePath); + const main = await selectSourceTurn(readOpencodeDatabase(databasePath), { + conversationId: "ses_main", + turnId: "u1" + }); + expect(main.turn).toBeDefined(); + expect(main.turn?.query).toBe(OPENCODE_QUERY); + expect(main.turn?.answer).toBe(STRUCTURED_OUTPUT.result); + expect(main.turn?.profileId).toBe("build"); + expect(main.turn?.toolCalls[0]).toMatchObject({ name: "read", status: "completed" }); + expect(main.turn?.completionEvidence).toBe("assistant_completed:a2"); + const subagent = await selectSourceTurn(readOpencodeDatabase(databasePath), { + conversationId: "ses_branch", + turnId: "u2" + }); + expect(subagent.turn).toBeNull(); + }); + + it("leaves a turn whose last assistant never completed for a later attempt", async () => { + directory = mkdtempSync(join(tmpdir(), "memmy-opencode-v2-open-")); + const databasePath = join(directory, "v2-open.sqlite"); + writeOpencodeV2Database(databasePath, { openEnded: true }); + const result = await selectSourceTurn(readOpencodeDatabase(databasePath), { + conversationId: "ses_main", + turnId: "u1" + }); + expect(result.turn).toBeNull(); + expect(result.reason).toBe("turn_incomplete"); + }); +}); + const STRUCTURED_OUTPUT = { verified: true, result: "The configuration was verified and documented." }; const OPENCODE_QUERY = "Inspect the configuration and report the verified result."; interface OpencodeTerminalCase { @@ -133,3 +168,78 @@ function writeOpencodeTerminalDatabase( db.close(); } } + +/** Builds an OpenCode V2 database: `session_v2` + `session_message` with inline content. */ +function writeOpencodeV2Database(path: string, options?: { openEnded?: boolean }): void { + const rows: Array<{ id: string; sessionId: string; type: string; seq: number; data: unknown }> = [ + { + id: "u1", + sessionId: "ses_main", + type: "user", + seq: 1, + data: { time: { created: 4_070_944_800_000 }, text: OPENCODE_QUERY, files: [], agents: [] } + }, + { + id: "a1", + sessionId: "ses_main", + type: "assistant", + seq: 2, + data: { + agent: "build", + time: { created: 4_070_944_801_000, completed: 4_070_944_801_500 }, + content: [ + { + type: "tool", + id: "call1", + name: "read", + state: { status: "completed", input: { path: "config" }, output: "Configuration verified." } + } + ] + } + }, + { + id: "a2", + sessionId: "ses_main", + type: "assistant", + seq: 3, + data: { + agent: "build", + ...(options?.openEnded + ? { time: { created: 4_070_944_801_600 } } + : { time: { created: 4_070_944_801_600, completed: 4_070_944_802_000 } }), + content: [{ type: "reasoning", text: "internal" }, { type: "text", text: STRUCTURED_OUTPUT.result }] + } + }, + { id: "i1", sessionId: "ses_main", type: "idle", seq: 4, data: {} }, + { + id: "u2", + sessionId: "ses_branch", + type: "user", + seq: 5, + data: { time: { created: 4_070_944_803_000 }, text: "branch prompt", files: [], agents: [] } + }, + { + id: "a3", + sessionId: "ses_branch", + type: "assistant", + seq: 6, + data: { + agent: "build", + time: { created: 4_070_944_804_000, completed: 4_070_944_805_000 }, + content: [{ type: "text", text: "branch answer" }] + } + } + ]; + const db = new Database(path); + try { + db.exec("CREATE TABLE session_v2(id TEXT, parent_id TEXT, directory TEXT, agent TEXT, revert TEXT, time_created INTEGER); CREATE TABLE session_message(id TEXT, session_id TEXT, type TEXT, seq INTEGER, time_created INTEGER, data TEXT)"); + db.prepare("INSERT INTO session_v2 VALUES(?,?,?,?,?,?)").run("ses_main", null, "/work/project", "build", null, 4_070_944_800_000); + db.prepare("INSERT INTO session_v2 VALUES(?,?,?,?,?,?)").run("ses_branch", "ses_main", "/work/project", "build", null, 4_070_944_801_000); + for (const row of rows) { + db.prepare("INSERT INTO session_message VALUES(?,?,?,?,?,?)") + .run(row.id, row.sessionId, row.type, row.seq, row.seq, JSON.stringify(row.data)); + } + } finally { + db.close(); + } +}