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(); + } +}