From e674c242e1914f25be0e99d4b44fc4f2ab33c860 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 16:39:26 -0400 Subject: [PATCH 01/11] feat(v2): add session storage service --- packages/opencode/src/v2/session.ts | 213 +------- .../opencode/src/v2/session/storage-memory.ts | 102 ++++ .../opencode/src/v2/session/storage-sql.ts | 173 +++++++ packages/opencode/src/v2/session/storage.ts | 100 ++++ packages/opencode/test/session/prompt.test.ts | 2 + .../opencode/test/v2/session-storage.test.ts | 307 ++++++++++++ specs/v2/storage-service.md | 474 ++++++++++++++++++ 7 files changed, 1182 insertions(+), 189 deletions(-) create mode 100644 packages/opencode/src/v2/session/storage-memory.ts create mode 100644 packages/opencode/src/v2/session/storage-sql.ts create mode 100644 packages/opencode/src/v2/session/storage.ts create mode 100644 packages/opencode/test/v2/session-storage.test.ts create mode 100644 specs/v2/storage-service.md diff --git a/packages/opencode/src/v2/session.ts b/packages/opencode/src/v2/session.ts index ec67d1820a..90f344c131 100644 --- a/packages/opencode/src/v2/session.ts +++ b/packages/opencode/src/v2/session.ts @@ -1,9 +1,6 @@ -import { SessionMessageTable, SessionTable } from "@/session/session.sql" import { SessionID } from "@/session/schema" import { WorkspaceID } from "@/control-plane/schema" -import { and, asc, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" -import * as Database from "@/storage/db" -import { Context, DateTime, Effect, Layer, Option, Schema } from "effect" +import { Context, DateTime, Effect, Layer, Schema } from "effect" import { SessionMessage } from "@opencode-ai/core/session-message" import type { Prompt } from "@opencode-ai/core/session-prompt" import { ProjectID } from "@/project/schema" @@ -13,7 +10,8 @@ import { optionalOmitUndefined } from "@opencode-ai/core/schema" import { EventV2 } from "@opencode-ai/core/event" import { EventV2Bridge } from "@/event-v2-bridge" import { ModelV2 } from "@opencode-ai/core/model" -import { ProviderV2 } from "@opencode-ai/core/provider" +import { SessionStorage } from "./session/storage" +import { SessionStorageSql } from "./session/storage-sql" export const Delivery = Schema.Literals(["immediate", "deferred"]).annotate({ identifier: "Session.Delivery", @@ -73,40 +71,17 @@ export interface Interface { workspaceID?: WorkspaceID }) => Effect.Effect readonly get: (sessionID: SessionID) => Effect.Effect - readonly list: (input: { - limit?: number - order?: "asc" | "desc" - directory?: string - path?: string - workspaceID?: WorkspaceID - roots?: boolean - start?: number - search?: string - cursor?: { - id: SessionID - time: number - direction: "previous" | "next" - } - }) => Effect.Effect - readonly messages: (input: { - sessionID: SessionID - limit?: number - order?: "asc" | "desc" - cursor?: { - id: SessionMessage.ID - time: number - direction: "previous" | "next" - } - }) => Effect.Effect - readonly context: (sessionID: SessionID) => Effect.Effect + readonly list: (input: SessionStorage.SessionListInput) => Effect.Effect + readonly messages: (input: SessionStorage.MessageListInput) => Effect.Effect + readonly context: (sessionID: SessionID) => Effect.Effect readonly prompt: (input: { id?: EventV2.ID sessionID: SessionID prompt: Prompt delivery?: Delivery - }) => Effect.Effect - readonly shell: (input: { id?: EventV2.ID; sessionID: SessionID; command: string }) => Effect.Effect - readonly skill: (input: { id?: EventV2.ID; sessionID: SessionID; skill: string }) => Effect.Effect + }) => Effect.Effect + readonly shell: (input: { id?: EventV2.ID; sessionID: SessionID; command: string }) => Effect.Effect + readonly skill: (input: { id?: EventV2.ID; sessionID: SessionID; skill: string }) => Effect.Effect readonly subagent: (input: { id?: EventV2.ID parentID: SessionID @@ -114,10 +89,10 @@ export interface Interface { agent: string model?: ModelV2.Ref }) => Effect.Effect - readonly switchAgent: (input: { sessionID: SessionID; agent: string }) => Effect.Effect - readonly switchModel: (input: { sessionID: SessionID; model: ModelV2.Ref }) => Effect.Effect - readonly compact: (sessionID: SessionID) => Effect.Effect - readonly wait: (sessionID: SessionID) => Effect.Effect + readonly switchAgent: (input: { sessionID: SessionID; agent: string }) => Effect.Effect + readonly switchModel: (input: { sessionID: SessionID; model: ModelV2.Ref }) => Effect.Effect + readonly compact: (sessionID: SessionID) => Effect.Effect + readonly wait: (sessionID: SessionID) => Effect.Effect } export class Service extends Context.Service()("@opencode/v2/Session") {} @@ -126,170 +101,28 @@ export const layer = Layer.effect( Service, Effect.gen(function* () { const events = yield* EventV2Bridge.Service - const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message) - - const decode = (row: typeof SessionMessageTable.$inferSelect) => - decodeMessage({ ...row.data, id: row.id, type: row.type }) - - function fromRow(row: typeof SessionTable.$inferSelect): Info { - return new Info({ - id: SessionID.make(row.id), - projectID: ProjectID.make(row.project_id), - workspaceID: row.workspace_id ? WorkspaceID.make(row.workspace_id) : undefined, - title: row.title, - parentID: row.parent_id ? SessionID.make(row.parent_id) : undefined, - path: row.path ?? "", - agent: row.agent ?? undefined, - model: row.model - ? { - id: ModelV2.ID.make(row.model.id), - providerID: ProviderV2.ID.make(row.model.providerID), - variant: ModelV2.VariantID.make(row.model.variant ?? "default"), - } - : undefined, - cost: row.cost, - tokens: { - input: row.tokens_input, - output: row.tokens_output, - reasoning: row.tokens_reasoning, - cache: { - read: row.tokens_cache_read, - write: row.tokens_cache_write, - }, - }, - time: { - created: DateTime.makeUnsafe(row.time_created), - updated: DateTime.makeUnsafe(row.time_updated), - archived: row.time_archived ? DateTime.makeUnsafe(row.time_archived) : undefined, - }, - }) - } + const storage = yield* SessionStorage.Service const result = Service.of({ create: Effect.fn("V2Session.create")(function* (_input) { - return {} as any + return yield* Effect.die(new Error("V2Session.create is not implemented")) }), get: Effect.fn("V2Session.get")(function* (sessionID) { - const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()) + const row = yield* storage.get(sessionID).pipe(Effect.orDie) if (!row) return yield* new NotFoundError({ sessionID }) - return fromRow(row) + return new Info(row) }), list: Effect.fn("V2Session.list")(function* (input) { - const direction = input.cursor?.direction ?? "next" - let order = input.order ?? "desc" - // This is a load bearing sort, desktop relies on this - const sortColumn = SessionTable.time_updated - // Query the adjacent rows in reverse, then flip them back into the requested order below. - if (direction === "previous" && order === "asc") order = "desc" - if (direction === "previous" && order === "desc") order = "asc" - const conditions: SQL[] = [] - if (input.directory) conditions.push(eq(SessionTable.directory, input.directory)) - if (input.path) - conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!) - if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) - if (input.roots) conditions.push(isNull(SessionTable.parent_id)) - if (input.start) conditions.push(gte(sortColumn, input.start)) - if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) - if (input.cursor) { - conditions.push( - order === "asc" - ? or( - gt(sortColumn, input.cursor.time), - and(eq(sortColumn, input.cursor.time), gt(SessionTable.id, input.cursor.id)), - )! - : or( - lt(sortColumn, input.cursor.time), - and(eq(sortColumn, input.cursor.time), lt(SessionTable.id, input.cursor.id)), - )!, - ) - } - const query = Database.Client() - .select() - .from(SessionTable) - .where(conditions.length > 0 ? and(...conditions) : undefined) - .orderBy( - order === "asc" ? asc(sortColumn) : desc(sortColumn), - order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), - ) - - const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() - return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row)) + return (yield* storage.list(input).pipe(Effect.orDie)).map((row) => new Info(row)) }), messages: Effect.fn("V2Session.messages")(function* (input) { - const direction = input.cursor?.direction ?? "next" - let order = input.order ?? "desc" - // Query the adjacent rows in reverse, then flip them back into the requested order below. - if (direction === "previous" && order === "asc") order = "desc" - if (direction === "previous" && order === "desc") order = "asc" - const boundary = input.cursor - ? order === "asc" - ? or( - gt(SessionMessageTable.time_created, input.cursor.time), - and( - eq(SessionMessageTable.time_created, input.cursor.time), - gt(SessionMessageTable.id, input.cursor.id), - ), - ) - : or( - lt(SessionMessageTable.time_created, input.cursor.time), - and( - eq(SessionMessageTable.time_created, input.cursor.time), - lt(SessionMessageTable.id, input.cursor.id), - ), - ) - : undefined - const where = boundary - ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) - : eq(SessionMessageTable.session_id, input.sessionID) - - const rows = Database.use((db) => { - const query = db - .select() - .from(SessionMessageTable) - .where(where) - .orderBy( - order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), - order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), - ) - const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() - return direction === "previous" ? rows.toReversed() : rows - }) - return rows.map((row) => decode(row)) + return yield* storage.messages(input).pipe(Effect.orDie) }), context: Effect.fn("V2Session.context")(function* (sessionID) { - const rows = Database.use((db) => { - const compaction = db - .select() - .from(SessionMessageTable) - .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) - .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id)) - .limit(1) - .get() - - return db - .select() - .from(SessionMessageTable) - .where( - and( - eq(SessionMessageTable.session_id, sessionID), - compaction - ? or( - gt(SessionMessageTable.time_created, compaction.time_created), - and( - eq(SessionMessageTable.time_created, compaction.time_created), - gte(SessionMessageTable.id, compaction.id), - ), - ) - : undefined, - ), - ) - .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)) - .all() - }) - return rows.map((row) => decode(row)) + return yield* storage.context(sessionID).pipe(Effect.orDie) }), prompt: Effect.fn("V2Session.prompt")(function* (_input) { - return {} as any + return yield* Effect.die(new Error("V2Session.prompt is not implemented")) }), shell: Effect.fn("V2Session.shell")(function* (_input) {}), skill: Effect.fn("V2Session.skill")(function* (_input) {}), @@ -336,6 +169,8 @@ export const layer = Layer.effect( }), ) -export const defaultLayer = layer.pipe(Layer.provide(EventV2Bridge.defaultLayer)) +export const defaultLayer = layer.pipe( + Layer.provide(Layer.mergeAll(EventV2Bridge.defaultLayer, SessionStorageSql.defaultLayer)), +) export * as SessionV2 from "./session" diff --git a/packages/opencode/src/v2/session/storage-memory.ts b/packages/opencode/src/v2/session/storage-memory.ts new file mode 100644 index 0000000000..0ca2bd1c26 --- /dev/null +++ b/packages/opencode/src/v2/session/storage-memory.ts @@ -0,0 +1,102 @@ +import { DateTime, Effect, Layer } from "effect" +import { SessionMessage } from "@opencode-ai/core/session-message" +import { SessionStorage } from "./storage" + +export interface State { + readonly sessions: Map + readonly messages: Map +} + +export const makeState = (): State => ({ + sessions: new Map(), + messages: new Map(), +}) + +export const layer = (state: State = makeState()) => + Layer.succeed( + SessionStorage.Service, + SessionStorage.Service.of({ + get: (sessionID) => Effect.sync(() => state.sessions.get(sessionID)), + list: (input) => + Effect.sync(() => { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const rows = Array.from(state.sessions.values()) + .filter((row) => { + if (input.directory && row.directory !== input.directory) return false + if (input.path && row.path !== input.path && !row.path?.startsWith(`${input.path}/`)) return false + if (input.workspaceID && row.workspaceID !== input.workspaceID) return false + if (input.roots && row.parentID) return false + if (input.start && DateTime.toEpochMillis(row.time.updated) < input.start) return false + if (input.search && !row.title.includes(input.search)) return false + if (!input.cursor) return true + return compareCursor(row.id, DateTime.toEpochMillis(row.time.updated), input.cursor, order) + }) + .toSorted((a, b) => + compareRows( + a.id, + DateTime.toEpochMillis(a.time.updated), + b.id, + DateTime.toEpochMillis(b.time.updated), + order, + ), + ) + const limited = input.limit === undefined ? rows : rows.slice(0, input.limit) + return direction === "previous" ? limited.toReversed() : limited + }), + messages: (input) => + Effect.sync(() => { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const rows = (state.messages.get(input.sessionID) ?? []) + .filter((message) => { + if (!input.cursor) return true + return compareCursor(message.id, DateTime.toEpochMillis(message.time.created), input.cursor, order) + }) + .toSorted((a, b) => + compareRows( + a.id, + DateTime.toEpochMillis(a.time.created), + b.id, + DateTime.toEpochMillis(b.time.created), + order, + ), + ) + const limited = input.limit === undefined ? rows : rows.slice(0, input.limit) + return direction === "previous" ? limited.toReversed() : limited + }), + context: (sessionID) => + Effect.sync(() => { + const messages = (state.messages.get(sessionID) ?? []).toSorted((a, b) => + compareRows( + a.id, + DateTime.toEpochMillis(a.time.created), + b.id, + DateTime.toEpochMillis(b.time.created), + "asc", + ), + ) + const index = messages.findLastIndex((message) => message.type === "compaction") + return index === -1 ? messages : messages.slice(index) + }), + }), + ) + +export const defaultLayer = layer() + +function compareCursor( + id: string, + time: number, + cursor: { readonly id: string; readonly time: number }, + order: SessionStorage.SortOrder, +) { + if (order === "asc") return time > cursor.time || (time === cursor.time && id > cursor.id) + return time < cursor.time || (time === cursor.time && id < cursor.id) +} + +function compareRows(aID: string, aTime: number, bID: string, bTime: number, order: SessionStorage.SortOrder) { + const result = aTime === bTime ? aID.localeCompare(bID) : aTime - bTime + return order === "asc" ? result : -result +} + +export * as SessionStorageMemory from "./storage-memory" diff --git a/packages/opencode/src/v2/session/storage-sql.ts b/packages/opencode/src/v2/session/storage-sql.ts new file mode 100644 index 0000000000..6487e1176a --- /dev/null +++ b/packages/opencode/src/v2/session/storage-sql.ts @@ -0,0 +1,173 @@ +import { SessionMessageTable, SessionTable } from "@/session/session.sql" +import { and, asc, Database, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" +import { SessionMessage } from "@opencode-ai/core/session-message" +import { Effect, Layer, Schema } from "effect" +import { SessionStorage } from "./storage" + +const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message) +const decodeSessionRow = Schema.decodeUnknownSync(SessionStorage.SessionRow) + +export const layer = Layer.effect( + SessionStorage.Service, + Effect.gen(function* () { + const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")((sessionID) => + attempt(() => + Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()), + ).pipe(Effect.map((row) => (row ? fromSessionRow(row) : undefined))), + ) + + const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")((input) => + attempt(() => { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const sortColumn = SessionTable.time_updated + const conditions: SQL[] = [] + if (input.directory) conditions.push(eq(SessionTable.directory, input.directory)) + if (input.path) + conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!) + if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) + if (input.roots) conditions.push(isNull(SessionTable.parent_id)) + if (input.start) conditions.push(gte(sortColumn, input.start)) + if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) + if (input.cursor) conditions.push(sessionCursorBoundary(input.cursor, order)) + + return Database.use((db) => { + const query = db + .select() + .from(SessionTable) + .where(conditions.length > 0 ? and(...conditions) : undefined) + .orderBy( + order === "asc" ? asc(sortColumn) : desc(sortColumn), + order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), + ) + const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() + return direction === "previous" ? rows.toReversed() : rows + }) + }).pipe(Effect.map((rows) => rows.map(fromSessionRow))), + ) + + const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")((input) => + attempt(() => { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const boundary = input.cursor ? messageCursorBoundary(input.cursor, order) : undefined + const where = boundary + ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) + : eq(SessionMessageTable.session_id, input.sessionID) + + return Database.use((db) => { + const query = db + .select() + .from(SessionMessageTable) + .where(where) + .orderBy( + order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), + order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), + ) + const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() + return direction === "previous" ? rows.toReversed() : rows + }) + }).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))), + ) + + const context: SessionStorage.Interface["context"] = Effect.fn("SessionStorageSql.context")((sessionID) => + attempt(() => + Database.use((db) => { + const compaction = db + .select() + .from(SessionMessageTable) + .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) + .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id)) + .limit(1) + .get() + + return db + .select() + .from(SessionMessageTable) + .where( + and( + eq(SessionMessageTable.session_id, sessionID), + compaction + ? or( + gt(SessionMessageTable.time_created, compaction.time_created), + and( + eq(SessionMessageTable.time_created, compaction.time_created), + gte(SessionMessageTable.id, compaction.id), + ), + ) + : undefined, + ), + ) + .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)) + .all() + }), + ).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))), + ) + + return SessionStorage.Service.of({ get, list, messages, context }) + }), +) + +export const defaultLayer = layer + +function attempt(body: () => A) { + return Effect.try({ + try: body, + catch: (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), + }) +} + +function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) { + if (order === "asc") + return or( + gt(SessionTable.time_updated, cursor.time), + and(eq(SessionTable.time_updated, cursor.time), gt(SessionTable.id, cursor.id)), + )! + return or( + lt(SessionTable.time_updated, cursor.time), + and(eq(SessionTable.time_updated, cursor.time), lt(SessionTable.id, cursor.id)), + )! +} + +function messageCursorBoundary(cursor: SessionStorage.MessageCursor, order: SessionStorage.SortOrder) { + if (order === "asc") + return or( + gt(SessionMessageTable.time_created, cursor.time), + and(eq(SessionMessageTable.time_created, cursor.time), gt(SessionMessageTable.id, cursor.id)), + )! + return or( + lt(SessionMessageTable.time_created, cursor.time), + and(eq(SessionMessageTable.time_created, cursor.time), lt(SessionMessageTable.id, cursor.id)), + )! +} + +function fromSessionRow(row: typeof SessionTable.$inferSelect) { + return decodeSessionRow({ + id: row.id, + parentID: row.parent_id ?? undefined, + projectID: row.project_id, + workspaceID: row.workspace_id ?? undefined, + directory: row.directory, + title: row.title, + path: row.path ?? "", + agent: row.agent ?? undefined, + model: row.model ? { ...row.model, variant: row.model.variant ?? "default" } : undefined, + cost: row.cost, + tokens: { + input: row.tokens_input, + output: row.tokens_output, + reasoning: row.tokens_reasoning, + cache: { + read: row.tokens_cache_read, + write: row.tokens_cache_write, + }, + }, + time: { + created: row.time_created, + updated: row.time_updated, + archived: row.time_archived ?? undefined, + }, + }) +} + +export * as SessionStorageSql from "./storage-sql" diff --git a/packages/opencode/src/v2/session/storage.ts b/packages/opencode/src/v2/session/storage.ts new file mode 100644 index 0000000000..c86289d741 --- /dev/null +++ b/packages/opencode/src/v2/session/storage.ts @@ -0,0 +1,100 @@ +import { WorkspaceID } from "@/control-plane/schema" +import { ProjectID } from "@/project/schema" +import { SessionID } from "@/session/schema" +import { V2Schema } from "@opencode-ai/core/v2-schema" +import { SessionMessage } from "@opencode-ai/core/session-message" +import { ModelV2 } from "@opencode-ai/core/model" +import { Context, Effect, Schema } from "effect" + +export const SortOrder = Schema.Literals(["asc", "desc"]).annotate({ + identifier: "SortOrder", +}) +export type SortOrder = typeof SortOrder.Type + +export const PageDirection = Schema.Literals(["previous", "next"]).annotate({ + identifier: "PageDirection", +}) +export type PageDirection = typeof PageDirection.Type + +export class StorageError extends Schema.TaggedErrorClass()("StorageError", { + message: Schema.String, + cause: Schema.Defect, +}) {} + +export class SessionRow extends Schema.Class("SessionRow")({ + id: SessionID, + parentID: Schema.optional(SessionID), + projectID: ProjectID, + workspaceID: Schema.optional(WorkspaceID), + directory: Schema.optional(Schema.String), + path: Schema.optional(Schema.String), + agent: Schema.optional(Schema.String), + model: Schema.optional(ModelV2.Ref), + cost: Schema.Finite, + tokens: Schema.Struct({ + input: Schema.Finite, + output: Schema.Finite, + reasoning: Schema.Finite, + cache: Schema.Struct({ + read: Schema.Finite, + write: Schema.Finite, + }), + }), + time: Schema.Struct({ + created: V2Schema.DateTimeUtcFromMillis, + updated: V2Schema.DateTimeUtcFromMillis, + archived: Schema.optional(V2Schema.DateTimeUtcFromMillis), + }), + title: Schema.String, +}) {} + +export const SessionCursor = Schema.Struct({ + id: SessionID, + time: Schema.Finite, + direction: PageDirection, +}).annotate({ identifier: "SessionCursor" }) +export type SessionCursor = typeof SessionCursor.Type + +export const SessionListInput = Schema.Struct({ + limit: Schema.optional(Schema.Finite), + order: Schema.optional(SortOrder), + directory: Schema.optional(Schema.String), + path: Schema.optional(Schema.String), + workspaceID: Schema.optional(WorkspaceID), + roots: Schema.optional(Schema.Boolean), + start: Schema.optional(Schema.Finite), + search: Schema.optional(Schema.String), + cursor: Schema.optional(SessionCursor), +}).annotate({ identifier: "SessionListInput" }) +export type SessionListInput = typeof SessionListInput.Type + +export const MessageCursor = Schema.Struct({ + id: SessionMessage.ID, + time: Schema.Finite, + direction: PageDirection, +}).annotate({ identifier: "MessageCursor" }) +export type MessageCursor = typeof MessageCursor.Type + +export const MessageListInput = Schema.Struct({ + sessionID: SessionID, + limit: Schema.optional(Schema.Finite), + order: Schema.optional(SortOrder), + cursor: Schema.optional(MessageCursor), +}).annotate({ identifier: "MessageListInput" }) +export type MessageListInput = typeof MessageListInput.Type + +export interface Interface { + readonly get: (sessionID: SessionID) => Effect.Effect + readonly list: (input: SessionListInput) => Effect.Effect + readonly messages: (input: MessageListInput) => Effect.Effect + readonly context: (sessionID: SessionID) => Effect.Effect +} + +export function pageOrder(order: SortOrder, direction: PageDirection) { + if (direction !== "previous") return order + return order === "asc" ? "desc" : "asc" +} + +export class Service extends Context.Service()("@opencode/v2/session/Storage") {} + +export * as SessionStorage from "./storage" diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index ff9ded4d19..ccce5b4015 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -36,6 +36,7 @@ import { SessionRunState } from "../../src/session/run-state" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { SessionStatus } from "../../src/session/status" import { SessionV2 } from "../../src/v2/session" +import { SessionStorageSql } from "../../src/v2/session/storage-sql" import { Skill } from "../../src/skill" import { SystemPrompt } from "../../src/session/system" import { Shell } from "../../src/shell/shell" @@ -507,6 +508,7 @@ noLLMServer.instance( const messages = yield* SessionV2.Service.use((session) => session.messages({ sessionID: chat.id })).pipe( Effect.provide(SessionV2.layer), + Effect.provide(SessionStorageSql.defaultLayer), ) const row = Database.use((db) => db.select().from(SessionMessageTable).where(Database.eq(SessionMessageTable.session_id, chat.id)).get(), diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts new file mode 100644 index 0000000000..8120fb89ea --- /dev/null +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -0,0 +1,307 @@ +import { expect } from "bun:test" +import { ProjectID } from "@/project/schema" +import { ProjectTable } from "@/project/project.sql" +import { SessionID } from "@/session/schema" +import { SessionMessageTable, SessionTable } from "@/session/session.sql" +import { Database } from "@/storage/db" +import { SessionStorage } from "@/v2/session/storage" +import { SessionStorageMemory } from "@/v2/session/storage-memory" +import { SessionStorageSql } from "@/v2/session/storage-sql" +import { EventV2 } from "@opencode-ai/core/event" +import { SessionMessage } from "@opencode-ai/core/session-message" +import { eq, or } from "@/storage/db" +import { DateTime, Effect, Layer, Schema } from "effect" +import { testEffect } from "../lib/effect" + +const projectID = ProjectID.make("project-session-storage") +const sessionA = SessionID.make("ses_storage_a") +const sessionB = SessionID.make("ses_storage_b") +const sessionC = SessionID.make("ses_storage_c") +const encodeMessage = Schema.encodeSync(SessionMessage.Message) +const memoryState = SessionStorageMemory.makeState() + +interface Seeds { + readonly reset: Effect.Effect + readonly project: Effect.Effect + readonly session: (input: { + id: SessionID + title: string + directory?: string + path: string + updated: number + }) => Effect.Effect + readonly userMessage: (input: { id: SessionMessage.ID; text: string; time: number }) => Effect.Effect + readonly compaction: (input: { + id: SessionMessage.ID + summary: string + time: number + }) => Effect.Effect +} + +function sessionStorageContract(name: string, layer: Layer.Layer, seed: Seeds) { + const it = testEffect(layer) + + const setup = Effect.gen(function* () { + yield* seed.reset + yield* Effect.addFinalizer(() => seed.reset) + yield* seed.project + }) + + it.effect("gets and lists sessions with filters and cursors", () => + Effect.gen(function* () { + yield* setup + yield* seed.session({ + id: sessionA, + title: "Alpha", + directory: "/tmp/project-session-storage", + path: "apps/api", + updated: 1000, + }) + yield* seed.session({ + id: sessionB, + title: "Beta", + directory: "/tmp/project-session-storage", + path: "apps/web", + updated: 2000, + }) + yield* seed.session({ + id: sessionC, + title: "Gamma", + directory: "/tmp/other-project", + path: "docs", + updated: 3000, + }) + + const storage = yield* SessionStorage.Service + const found = yield* storage.get(sessionB) + expect(found?.title).toBe("Beta") + expect(found ? DateTime.toEpochMillis(found.time.updated) : undefined).toBe(2000) + + expect((yield* storage.list({ path: "apps", order: "asc" })).map((row) => row.id)).toEqual([sessionA, sessionB]) + expect( + (yield* storage.list({ directory: "/tmp/project-session-storage", order: "asc" })).map((row) => row.id), + ).toEqual([sessionA, sessionB]) + expect( + (yield* storage.list({ order: "asc", cursor: { id: sessionA, time: 1000, direction: "next" } })).map( + (row) => row.id, + ), + ).toEqual([sessionB, sessionC]) + expect( + (yield* storage.list({ order: "asc", cursor: { id: sessionC, time: 3000, direction: "previous" } })).map( + (row) => row.id, + ), + ).toEqual([sessionA, sessionB]) + }), + ) + + it.effect("lists session messages with cursor direction", () => + Effect.gen(function* () { + yield* setup + yield* seed.session({ id: sessionA, title: "Alpha", path: "apps/api", updated: 1000 }) + yield* seed.userMessage({ id: EventV2.ID.make("evt_msg_1"), text: "one", time: 1000 }) + yield* seed.userMessage({ id: EventV2.ID.make("evt_msg_2"), text: "two", time: 2000 }) + yield* seed.userMessage({ id: EventV2.ID.make("evt_msg_3"), text: "three", time: 3000 }) + + const storage = yield* SessionStorage.Service + + expect((yield* storage.messages({ sessionID: sessionA, order: "asc", limit: 2 })).map((row) => row.id)).toEqual([ + EventV2.ID.make("evt_msg_1"), + EventV2.ID.make("evt_msg_2"), + ]) + expect( + (yield* storage.messages({ + sessionID: sessionA, + order: "asc", + cursor: { id: EventV2.ID.make("evt_msg_3"), time: 3000, direction: "previous" }, + })).map((row) => row.id), + ).toEqual([EventV2.ID.make("evt_msg_1"), EventV2.ID.make("evt_msg_2")]) + }), + ) + + it.effect("returns context from the latest compaction boundary", () => + Effect.gen(function* () { + yield* setup + yield* seed.session({ id: sessionA, title: "Alpha", path: "apps/api", updated: 1000 }) + yield* seed.userMessage({ id: EventV2.ID.make("evt_context_1"), text: "before", time: 1000 }) + yield* seed.compaction({ id: EventV2.ID.make("evt_context_2"), summary: "compact", time: 2000 }) + yield* seed.userMessage({ id: EventV2.ID.make("evt_context_3"), text: "after", time: 3000 }) + + const storage = yield* SessionStorage.Service + const context = yield* storage.context(sessionA) + + expect(context.map((message) => message.id)).toEqual([ + EventV2.ID.make("evt_context_2"), + EventV2.ID.make("evt_context_3"), + ]) + expect(context.map((message) => message.type)).toEqual(["compaction", "user"]) + }), + ) +} + +const sqlSeeds: Seeds = { + reset: Effect.sync(resetSqlSeeds), + project: Effect.sync(seedProject), + session: (input) => Effect.sync(() => seedSession(input)), + userMessage: (input) => Effect.sync(() => seedUserMessage(input)), + compaction: (input) => Effect.sync(() => seedCompaction(input)), +} + +sessionStorageContract("SessionStorageSql", SessionStorageSql.defaultLayer, sqlSeeds) + +const memorySeeds: Seeds = { + reset: Effect.sync(() => { + memoryState.sessions.clear() + memoryState.messages.clear() + }), + project: Effect.void, + session: (input) => + Effect.sync(() => { + memoryState.sessions.set(input.id, makeSessionRow(input)) + }), + userMessage: (input) => + Effect.sync(() => { + appendMemoryMessage(makeUserMessage(input)) + }), + compaction: (input) => + Effect.sync(() => { + appendMemoryMessage(makeCompaction(input)) + }), +} + +sessionStorageContract("SessionStorageMemory", SessionStorageMemory.layer(memoryState), memorySeeds) + +function seedProject() { + Database.use((db) => + db + .insert(ProjectTable) + .values({ + id: projectID, + worktree: "/tmp/project-session-storage", + time_created: 1, + time_updated: 1, + sandboxes: [], + }) + .onConflictDoNothing() + .run(), + ) +} + +function resetSqlSeeds() { + Database.use((db) => { + db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run() + db.delete(SessionTable) + .where(or(eq(SessionTable.id, sessionA), eq(SessionTable.id, sessionB), eq(SessionTable.id, sessionC))) + .run() + db.delete(ProjectTable).where(eq(ProjectTable.id, projectID)).run() + }) +} + +function seedSession(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { + Database.use((db) => + db + .insert(SessionTable) + .values({ + id: input.id, + project_id: projectID, + slug: input.title.toLowerCase(), + directory: input.directory ?? "/tmp/project-session-storage", + path: input.path, + title: input.title, + version: "test", + cost: 0, + tokens_input: 0, + tokens_output: 0, + tokens_reasoning: 0, + tokens_cache_read: 0, + tokens_cache_write: 0, + time_created: input.updated, + time_updated: input.updated, + }) + .run(), + ) +} + +function seedUserMessage(input: { id: SessionMessage.ID; text: string; time: number }) { + const encoded = encodeMessage(makeUserMessage(input)) + const { id: _, type: __, ...data } = encoded + seedMessage(input.id, "user", input.time, data) +} + +function seedCompaction(input: { id: SessionMessage.ID; summary: string; time: number }) { + const encoded = encodeMessage(makeCompaction(input)) + const { id: _, type: __, ...data } = encoded + seedMessage(input.id, "compaction", input.time, data) +} + +function makeSessionRow(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { + return new SessionStorage.SessionRow({ + id: input.id, + projectID, + title: input.title, + directory: input.directory ?? "/tmp/project-session-storage", + path: input.path, + cost: 0, + tokens: { + input: 0, + output: 0, + reasoning: 0, + cache: { + read: 0, + write: 0, + }, + }, + time: { + created: DateTime.makeUnsafe(input.updated), + updated: DateTime.makeUnsafe(input.updated), + }, + }) +} + +function makeUserMessage(input: { id: SessionMessage.ID; text: string; time: number }) { + return new SessionMessage.User({ + id: input.id, + type: "user", + text: input.text, + files: [], + agents: [], + references: [], + time: { created: DateTime.makeUnsafe(input.time) }, + }) +} + +function makeCompaction(input: { id: SessionMessage.ID; summary: string; time: number }) { + return new SessionMessage.Compaction({ + id: input.id, + type: "compaction", + reason: "manual", + summary: input.summary, + time: { created: DateTime.makeUnsafe(input.time) }, + }) +} + +function seedMessage( + id: SessionMessage.ID, + type: SessionMessage.Type, + time: number, + data: typeof SessionMessageTable.$inferInsert.data, +) { + Database.use((db) => + db + .insert(SessionMessageTable) + .values({ + id, + session_id: sessionA, + type, + time_created: time, + time_updated: time, + data, + }) + .run(), + ) +} + +function appendMemoryMessage(message: SessionMessage.Message) { + const current = memoryState.messages.get(sessionA) ?? [] + current.push(message) + memoryState.messages.set(sessionA, current) +} diff --git a/specs/v2/storage-service.md b/specs/v2/storage-service.md new file mode 100644 index 0000000000..d1fead2996 --- /dev/null +++ b/specs/v2/storage-service.md @@ -0,0 +1,474 @@ +# V2 Storage Service + +This note inventories the current SQLite/Drizzle query shapes used by opencode and sketches a swappable Effect service boundary for V2 storage. The goal is not to expose a generic SQL abstraction. The goal is to collect the domain operations we actually need behind an Effect service so the implementation can remain Drizzle SQLite today and move to Drizzle Effect SQLite, Effect SQL, a remote store, or a test implementation later. + +## Current Shape + +The current database module is `packages/opencode/src/storage/db.ts`. + +It provides: + +- `Database.Client()` as a lazy singleton Drizzle client. +- `Database.use(callback)` as an ambient callback against the current transaction or global client. +- `Database.transaction(callback, { behavior })` for synchronous SQLite transactions. +- `Database.effect(fn)` for side effects queued until after the surrounding transaction. +- SQLite lifecycle concerns: path selection, PRAGMAs, migrations, close/reset. + +This API is convenient but leaks the concrete database everywhere. Most call sites import Drizzle operators and tables directly, build SQL in feature modules, and depend on synchronous callback execution. + +## Tables In Scope + +The regularly queried tables are: + +- `project`: project identity, worktree, sandbox list, commands, icon metadata. +- `workspace`: workspace records associated with projects. +- `session`: session metadata, hierarchy, workspace/project links, usage totals, archive/revert/permission fields. +- `message`: legacy V2 message rows with JSON payloads. +- `part`: legacy V2 message part rows with JSON payloads. +- `session_message`: new V2 event-derived session message rows. +- `todo`: ordered per-session todo rows. +- `event_sequence`: per-aggregate high-water mark and owner claim. +- `event`: ordered sync/event history rows. +- `account` and `account_state`: auth accounts and singleton active account state. +- `permission`: project-scoped permission rules. +- `session_share`: share records by session. +- `data_migration`: resumable data migration completion markers. + +## Query Shapes + +### Lifecycle And Migrations + +- Open database at the channel/user-selected path. +- Apply SQLite PRAGMAs for WAL, sync mode, busy timeout, cache size, foreign keys, and checkpointing. +- Apply schema migrations from bundled or dev migration files. +- Run data migrations in resumable background fibers. +- Run high-throughput JSON import using bulk inserts and explicit transactions. +- Provide a readonly/raw admin path for `opencode db` diagnostics. + +### Transactions + +- Run synchronous transactions with SQLite behavior options: `deferred`, `immediate`, `exclusive`. +- Preserve transaction context across nested helpers and projectors. +- Queue post-commit side effects for bus/global event publication. +- Need write-lock semantics for event sequencing. `SyncEvent.run` depends on `immediate` to avoid another writer changing the aggregate sequence between read and write. + +### Event Store + +- Read latest sequence and owner by aggregate id. +- Claim an aggregate by updating `event_sequence.owner_id`. +- Remove all sync state for an aggregate by deleting `event_sequence` and `event` rows transactionally. +- Append a projected event: + - read latest sequence inside an immediate transaction, + - run the domain projector, + - upsert `event_sequence`, + - insert `event`, + - publish only after commit. +- Read sync fence state for all aggregates or a supplied aggregate id set. +- Read event history with per-aggregate high-water exclusions. +- Read replay history for one aggregate ordered by sequence. + +### Sessions + +- Get one session by id. +- List sessions with dynamic filters: + - project id, + - workspace id, + - directory, + - path or path prefix, + - root sessions only, + - updated since timestamp, + - title search, + - archived/non-archived, + - cursor pagination by `(time_updated, id)`, + - asc/desc order and previous/next page direction. +- List child sessions by parent id. +- List global sessions, then hydrate project metadata for distinct project ids. +- Create/update/delete sessions through event projectors. +- Update session usage counters using atomic increments. +- Patch sessions with partial row updates and read the updated row. + +### Messages And Parts + +- Page legacy `message` rows by session, descending by `(time_created, id)`, with `limit + 1` pagination. +- Hydrate page results with all `part` rows for returned message ids ordered by `(message_id, id)`. +- Get one message by `(session_id, message_id)`. +- Get one part by `(session_id, message_id, part_id)`. +- List parts by message id ordered by id. +- Upsert messages by id, updating JSON payload on conflict. +- Upsert parts by id, updating JSON payload on conflict. +- Delete messages and parts by session-scoped keys. +- Read previous part usage before update/delete so usage counters can be adjusted. +- Ignore late writes that fail foreign-key constraints when the session/message has already been removed. + +### Session Message Timeline + +- List session messages by session id with cursor pagination by `(time_created, id)`. +- Load context since latest compaction: + - read latest `type = compaction` message, + - read all messages after or at that compaction boundary ordered ascending. +- Read current assistant/compaction/shell messages by session and type, newest first, then apply in-memory predicates. +- Update message JSON data by `(id, session_id, type)`. +- Append session message rows. +- Update session metadata for agent/model switch events. + +### Projects And Workspaces + +- Upsert projects by id. +- List all projects and get one project by id. +- Update project fields and return the updated row. +- Update initialized timestamp. +- Mutate JSON-ish sandbox arrays by read/update-returning. +- Repair global sessions into a discovered project by matching `(project_id = global, directory = worktree)`. +- Workspace CRUD by project and workspace id. +- Read sessions associated with a workspace. +- Read distinct workspaces for a project when needed by control-plane flows. + +### Accounts, Permissions, Shares, Todos + +- Account repository: + - read active singleton state then active account, + - list accounts, + - upsert account, + - upsert singleton active state, + - update tokens, + - transactionally clear state and delete account. +- Permissions: + - load rules by project id during per-instance state initialization. + - persist rules when approval state changes. +- Shares: + - get share by session id, + - upsert share by session id, + - delete share by session id. +- Todos: + - transactionally replace all todos for a session with ordered rows, + - list todos by session ordered by position. + +### Imports And Admin + +- Import sessions/messages/parts idempotently using `onConflictDoNothing`. +- Import sessions with conflict update for project/directory/path. +- Run CLI diagnostics: + - readonly raw SQL query, + - sqlite shell for local SQLite backend, + - print database path. + +## Proposed Boundary + +Use concrete service interfaces at the level higher services actually consume. We do not need a broad abstract hierarchy of sub-interfaces up front. + +For V2, the first useful boundary is a session storage service consumed by `SessionV2.Service`: + +```ts +export interface Interface { + readonly get: (sessionID: SessionID) => Effect.Effect + readonly list: (input: SessionListInput) => Effect.Effect + readonly messages: (input: SessionMessageListInput) => Effect.Effect + readonly context: (sessionID: SessionID) => Effect.Effect +} +``` + +That interface should be shaped by `v2/session.ts`, not by the underlying tables. Internally it can use whatever helpers make sense: Drizzle query builders, transaction helpers, mapper functions, or smaller private modules. Those internals do not need to be stable service boundaries until another higher-level service needs to consume them. + +The important design choice is that public boundaries are domain-level and demand-driven. The implementation may still use Drizzle internally, but V2 session code should stop importing Drizzle tables/operators for normal app behavior. + +## Service Breakdown + +We can safely break this into multiple services when there is a real consumer boundary, but we should not invent a full repository graph before the higher services ask for it. Start with the services that map to product/domain services, then let implementation helpers stay private. + +### Foundation Helper + +`StorageConnection` can exist as an implementation helper that owns the backend and cross-cutting mechanics: + +- open/close lifecycle, +- migrations, +- transaction context, +- transaction behavior (`deferred`, `immediate`, `exclusive`), +- post-commit callbacks, +- backend capabilities such as raw SQL or sqlite shell support. + +This should be the only place that knows whether the implementation is current Drizzle SQLite, future Drizzle Effect SQLite, Effect SQL, or something else. It does not necessarily need to be exposed directly to feature services. + +### Public Services + +Public services should map to higher-level consumers: + +- `SessionV2Storage`: the first public storage service for `v2/session.ts` reads and eventually V2 session writes. +- `SyncEventStorage`: later, if we extract event sequencing/projection out of `sync/index.ts`. +- `StorageAdmin`: later, if CLI diagnostics need a backend-neutral story. +- Smaller stores only when their current owner service needs a swappable dependency. + +Splitting this way is safe because public services share one internal connection/transaction helper. A transaction can compose operations across implementation helpers without each public service opening its own client. + +The unsafe split would be one independent service per table with independent clients/lifecycles. That would make cross-table transactions, event projection, and post-commit publication harder to reason about. + +## V2 Session Today + +`packages/opencode/src/v2/session.ts` is currently a thin experimental service. It mixes domain API shape, row decoding, cursor query construction, and event publishing. + +Implemented methods: + +- `get(sessionID)`: reads `session` by id and maps the row to `SessionV2.Info`. +- `list(input)`: lists sessions from `session` with filters and cursor pagination by `(time_updated, id)`. +- `messages(input)`: lists projected `session_message` rows with cursor pagination by `(time_created, id)` and decodes them to `SessionMessage.Message`. +- `context(sessionID)`: finds the latest compaction message, then returns all `session_message` rows at or after that boundary ordered ascending. +- `switchAgent(input)`: publishes `SessionEvent.AgentSwitched` through `EventV2Bridge`. +- `switchModel(input)`: publishes `SessionEvent.ModelSwitched` through `EventV2Bridge`. +- `subagent(input)`: partially implemented orchestration that calls `create`, `prompt`, `wait`, and `messages`. + +Stubbed or incomplete methods: + +- `create`: currently returns `{}` via `any`. +- `prompt`: currently returns `{}` via `any`. +- `shell`: no-op. +- `skill`: no-op. +- `compact`: no-op. +- `wait`: no-op. + +Current direct storage needs for implemented reads: + +- Get session row by id. +- List sessions with filters: directory, path prefix, workspace id, roots-only, updated-start, title search. +- Cursor sessions by `(time_updated, id)` with `previous`/`next` page semantics. +- List session messages by session id with cursor by `(time_created, id)`. +- Load active context since latest compaction. + +Current write path for implemented commands: + +- V2 session commands publish core `SessionEvent` definitions through `EventV2Bridge`. +- `EventV2Bridge` maps versioned aggregate events to legacy `SyncEvent.run`. +- `SyncEvent.run` applies `session/projectors-next.ts`, which updates `session` and `session_message` transactionally. + +That means V2 session reads and V2 event projection are coupled through storage, but they are two distinct seams. + +## Best First Slice + +The best first storage service to extract is `SessionV2Storage`, scoped to what `v2/session.ts` currently needs, not a general storage layer. + +Reasons: + +- It is central to the new V2 API. +- It has a small current query surface: `get`, `list`, `messages`, `context`. +- It is shaped by the V2 session service API, not by table names. +- It can replace all direct Drizzle reads in `v2/session.ts` without touching legacy `message`/`part` yet. +- It can grow to include V2 writes (`create`, `prompt`, `compact`, etc.) as those methods become real. + +Possible first interface: + +```ts +export interface Interface { + readonly get: (sessionID: SessionID) => Effect.Effect + readonly list: (input: SessionListInput) => Effect.Effect + readonly messages: (input: SessionMessageListInput) => Effect.Effect + readonly context: (sessionID: SessionID) => Effect.Effect +} +``` + +The second slice should be the write side for V2 session behavior once the currently stubbed methods are designed: + +- `create`, +- `prompt`, +- `shell`, +- `skill`, +- `compact`, +- `wait`. + +The third slice should be event sequencing/projection if the V2 write path continues to publish through `EventV2Bridge` and `SyncEvent.run`. That is more valuable but riskier, so doing `SessionV2Storage` first gives us the service pattern before touching event sequencing. + +Recommended sequence: + +1. Add `SessionV2Storage` implemented with current Drizzle SQLite internals. +2. Refactor `V2Session.get`, `V2Session.list`, `V2Session.messages`, and `V2Session.context` to use it. +3. Keep private implementation helpers for session rows, session-message rows, cursor predicates, and row decoding near the storage implementation. +4. Add write methods to `SessionV2Storage` only as `V2Session.create/prompt/shell/skill/compact/wait` become real. +5. Extract `SyncEventStorage` later if event sequencing/projection needs its own swappable boundary. + +## Suggested Sub-Interfaces + +### `EventStore` + +```ts +interface EventStore { + readonly getSequence: ( + aggregateID: string, + ) => Effect.Effect<{ seq: number; ownerID?: string } | undefined, StorageError> + readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect + readonly removeAggregate: (aggregateID: string) => Effect.Effect + readonly appendProjected: (input: { + definition: SyncDefinition + aggregateID: string + data: unknown + project: Effect.Effect + publish: Effect.Effect + }) => Effect.Effect + readonly fence: (aggregateIDs?: string[]) => Effect.Effect, StorageError> + readonly history: (input: { since?: Record }) => Effect.Effect + readonly replay: (aggregateID: string) => Effect.Effect +} +``` + +This is the most important seam. It owns sequence correctness, immediate transaction behavior, and post-commit publication. + +### `SessionStore` + +```ts +interface SessionStore { + readonly get: (id: SessionID) => Effect.Effect + readonly list: (input: SessionListInput) => Effect.Effect + readonly listGlobal: ( + input: GlobalSessionListInput, + ) => Effect.Effect, StorageError> + readonly children: (parentID: SessionID) => Effect.Effect + readonly insert: (row: SessionInsert) => Effect.Effect + readonly patch: (id: SessionID, patch: SessionPatch) => Effect.Effect + readonly delete: (id: SessionID) => Effect.Effect + readonly incrementUsage: (id: SessionID, usage: UsageDelta) => Effect.Effect +} +``` + +This removes query construction from `session.ts` and projectors while preserving the list semantics that clients rely on. + +### `MessageStore` + +```ts +interface MessageStore { + readonly page: ( + input: MessagePageInput, + ) => Effect.Effect<{ rows: MessageRow[]; more: boolean; cursor?: MessageCursor }, StorageError> + readonly hydrate: ( + rows: MessageRow[], + ) => Effect.Effect, StorageError> + readonly get: (input: { + sessionID: SessionID + messageID: MessageID + }) => Effect.Effect + readonly parts: (messageID: MessageID) => Effect.Effect + readonly getPart: (input: { + sessionID: SessionID + messageID: MessageID + partID: PartID + }) => Effect.Effect + readonly upsertMessage: (row: MessageInsert) => Effect.Effect + readonly upsertPart: (row: PartInsert) => Effect.Effect<{ previous?: PartRow }, StorageError | ForeignKeyError> + readonly deleteMessage: (input: { + sessionID: SessionID + messageID: MessageID + }) => Effect.Effect + readonly deletePart: (input: { + sessionID: SessionID + partID: PartID + }) => Effect.Effect +} +``` + +The delete/update methods expose previous rows where callers need usage compensation. + +### `SessionMessageStore` + +```ts +interface SessionMessageStore { + readonly list: (input: SessionMessageListInput) => Effect.Effect + readonly context: (sessionID: SessionID) => Effect.Effect + readonly currentByType: (input: { + sessionID: SessionID + type: SessionMessage.Type + }) => Effect.Effect + readonly append: (row: SessionMessageInsert) => Effect.Effect + readonly updateData: (input: { + id: SessionMessage.ID + sessionID: SessionID + type: SessionMessage.Type + data: SessionMessageData + }) => Effect.Effect +} +``` + +This gives the V2 message updater a storage adapter without exposing Drizzle. + +### Smaller Stores + +- `ProjectStore`: upsert, get, list, patch-returning, set initialized, sandbox mutation, repair global sessions. +- `WorkspaceStore`: create/update/delete/get/list by project, list session ids, distinct project workspaces. +- `AccountStore`: active, list, get row, persist account, persist token, use account/org, remove. +- `PermissionStore`: load and save project rules. +- `ShareStore`: get, upsert, delete by session id. +- `TodoStore`: replace all for session, list by session. +- `MigrationStore`: completed marker get/insert, data migration helpers, JSON import helpers. +- `AdminStore`: path, readonly raw query, local sqlite shell support flag. + +## Transaction Model + +There are two viable implementations. + +### Option A: One Top-Level Service With Fiber-Local Transaction + +The service exposes stores directly. Store methods inspect a fiber-local/current transaction and use it if present. `transaction(effect)` installs the transaction in that context. + +Pros: + +- Closest to current `Database.use` semantics. +- Domain services do not need explicit transaction parameters. +- Projectors can call the same store methods inside and outside transactions. + +Cons: + +- Requires careful implementation of fiber-local context and post-commit queues. +- More ambient than explicit dependency passing. + +### Option B: Explicit `Transaction` Service + +`transaction(effect)` provides a separate `Transaction` context. Projector-only methods require `Transaction` in their environment. + +Pros: + +- More explicit. A method that must be transactional says so in the type. +- Easier to prevent accidental out-of-transaction writes for projectors. + +Cons: + +- More churn at call sites. +- Some methods may need both transactional and non-transactional variants. + +Recommendation: start with Option A for migration ergonomics, but keep the internal implementation structured so projector methods can later move to explicit `Transaction` if needed. + +## Error Model + +Use typed storage errors at the boundary: + +- `StorageError`: unknown backend failure, includes cause. +- `NotFoundError`: domain-specific absence only where absence is exceptional. +- `ConflictError`: uniqueness or stale write conflicts, if exposed. +- `ForeignKeyError`: late event/projector writes where parent rows are already gone. +- `UnsupportedAdminOperation`: non-SQLite backend cannot open sqlite shell or run raw SQL. + +Avoid leaking Drizzle or SQLite error codes above the storage implementation. For current behavior, `ForeignKeyError` should allow projectors to keep ignoring late message/part updates intentionally. + +## Migration Strategy + +1. Add the storage service as a thin wrapper over existing Drizzle SQLite. +2. Move V2 session/session-message read APIs behind `SessionStorage`. This is the smallest reversible slice and establishes the service pattern. +3. Add write methods to `SessionStorage` only as `V2Session.create/prompt/shell/skill/compact/wait` become real. +4. Move event-store operations after the read seam is proven. This is higher value but riskier because it owns sequencing, transactions, and post-commit side effects. +5. Move projector writes behind stores, preserving transaction semantics. +6. Move small repositories: share, todo, permission, account. +7. Leave CLI/admin raw SQL as an explicit `AdminStore` escape hatch. +8. Only after the domain boundary is in place, evaluate replacing the implementation with Drizzle Effect SQLite or another backend. + +## Open Questions + +- Should V2 storage be per instance/workspace via `InstanceState`, or global per data directory like the current DB? +- Should event projection become the only write path for sessions/messages, or should import/migration retain direct write APIs permanently? +- Do we need a remote-capable store soon? If yes, raw admin SQL and SQLite shell must stay backend-specific from day one. +- Should `session_message` replace legacy `message`/`part` for V2 context entirely, or do both need first-class APIs for the medium term? +- Should cursor encoding live in the storage service or stay in domain modules while storage accepts decoded cursor structs? + +## Initial Implementation Target + +The first implementation PR should be small and reversible: + +- Add `packages/opencode/src/v2/session/storage.ts` with the `SessionStorage` service shape. +- Implement SQL and in-memory read backends for `get`, `list`, `messages`, and `context`. +- Refactor `v2/session.ts` to consume `SessionStorage.Service` instead of raw Drizzle calls. +- Keep table schemas and existing migrations unchanged. +- Add generic contract tests that run against both storage implementations. + +That gives us a concrete swappable seam without forcing every event/session/project/account query to move at once. Event-store extraction remains the next broader storage slice. From 110d4091e5450945c0948e0905f1823b5b9cdf7b Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 16:55:07 -0400 Subject: [PATCH 02/11] test(v2): isolate session storage contract --- .../opencode/test/v2/session-storage.test.ts | 42 ++++++++++++++----- 1 file changed, 32 insertions(+), 10 deletions(-) diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 8120fb89ea..165097fdec 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -17,6 +17,7 @@ const projectID = ProjectID.make("project-session-storage") const sessionA = SessionID.make("ses_storage_a") const sessionB = SessionID.make("ses_storage_b") const sessionC = SessionID.make("ses_storage_c") +const sessionD = SessionID.make("ses_storage_d") const encodeMessage = Schema.encodeSync(SessionMessage.Message) const memoryState = SessionStorageMemory.makeState() @@ -67,29 +68,43 @@ function sessionStorageContract(name: string, layer: Layer.Layer row.id)).toEqual([sessionA, sessionB]) expect( - (yield* storage.list({ directory: "/tmp/project-session-storage", order: "asc" })).map((row) => row.id), + (yield* storage.list({ directory: "/tmp/project-session-storage", path: "apps", order: "asc" })).map( + (row) => row.id, + ), ).toEqual([sessionA, sessionB]) expect( - (yield* storage.list({ order: "asc", cursor: { id: sessionA, time: 1000, direction: "next" } })).map( - (row) => row.id, - ), + (yield* storage.list({ directory: "/tmp/project-session-storage", order: "asc" })).map((row) => row.id), + ).toEqual([sessionA, sessionB, sessionC]) + expect( + (yield* storage.list({ + directory: "/tmp/project-session-storage", + order: "asc", + cursor: { id: sessionA, time: 1000, direction: "next" }, + })).map((row) => row.id), ).toEqual([sessionB, sessionC]) expect( - (yield* storage.list({ order: "asc", cursor: { id: sessionC, time: 3000, direction: "previous" } })).map( - (row) => row.id, - ), + (yield* storage.list({ + directory: "/tmp/project-session-storage", + order: "asc", + cursor: { id: sessionC, time: 3000, direction: "previous" }, + })).map((row) => row.id), ).toEqual([sessionA, sessionB]) }), ) @@ -190,7 +205,14 @@ function resetSqlSeeds() { Database.use((db) => { db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run() db.delete(SessionTable) - .where(or(eq(SessionTable.id, sessionA), eq(SessionTable.id, sessionB), eq(SessionTable.id, sessionC))) + .where( + or( + eq(SessionTable.id, sessionA), + eq(SessionTable.id, sessionB), + eq(SessionTable.id, sessionC), + eq(SessionTable.id, sessionD), + ), + ) .run() db.delete(ProjectTable).where(eq(ProjectTable.id, projectID)).run() }) From 479b49ffa6986bc9a0a3cc5b3a7f596d0f83fdd1 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 16:59:59 -0400 Subject: [PATCH 03/11] refactor(v2): make memory storage layer constant --- .../opencode/src/v2/session/storage-memory.ts | 26 ++++++++++------- .../opencode/test/v2/session-storage.test.ts | 28 +++++++++++-------- 2 files changed, 32 insertions(+), 22 deletions(-) diff --git a/packages/opencode/src/v2/session/storage-memory.ts b/packages/opencode/src/v2/session/storage-memory.ts index 0ca2bd1c26..d47d0678b1 100644 --- a/packages/opencode/src/v2/session/storage-memory.ts +++ b/packages/opencode/src/v2/session/storage-memory.ts @@ -1,4 +1,4 @@ -import { DateTime, Effect, Layer } from "effect" +import { Context, DateTime, Effect, Layer } from "effect" import { SessionMessage } from "@opencode-ai/core/session-message" import { SessionStorage } from "./storage" @@ -7,15 +7,18 @@ export interface State { readonly messages: Map } -export const makeState = (): State => ({ +export class StateService extends Context.Service()("@opencode/v2/session/StorageMemoryState") {} + +const stateLayer = Layer.sync(StateService, () => ({ sessions: new Map(), messages: new Map(), -}) +})) -export const layer = (state: State = makeState()) => - Layer.succeed( - SessionStorage.Service, - SessionStorage.Service.of({ +const storageLayer = Layer.effect( + SessionStorage.Service, + Effect.gen(function* () { + const state = yield* StateService + return SessionStorage.Service.of({ get: (sessionID) => Effect.sync(() => state.sessions.get(sessionID)), list: (input) => Effect.sync(() => { @@ -79,10 +82,13 @@ export const layer = (state: State = makeState()) => const index = messages.findLastIndex((message) => message.type === "compaction") return index === -1 ? messages : messages.slice(index) }), - }), - ) + }) + }), +) -export const defaultLayer = layer() +export const layer = storageLayer.pipe(Layer.provideMerge(stateLayer)) + +export const defaultLayer = layer function compareCursor( id: string, diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 165097fdec..262f613d6a 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -19,7 +19,6 @@ const sessionB = SessionID.make("ses_storage_b") const sessionC = SessionID.make("ses_storage_c") const sessionD = SessionID.make("ses_storage_d") const encodeMessage = Schema.encodeSync(SessionMessage.Message) -const memoryState = SessionStorageMemory.makeState() interface Seeds { readonly reset: Effect.Effect @@ -163,27 +162,29 @@ const sqlSeeds: Seeds = { sessionStorageContract("SessionStorageSql", SessionStorageSql.defaultLayer, sqlSeeds) -const memorySeeds: Seeds = { - reset: Effect.sync(() => { +const memorySeeds: Seeds = { + reset: Effect.gen(function* () { + const memoryState = yield* SessionStorageMemory.StateService memoryState.sessions.clear() memoryState.messages.clear() }), project: Effect.void, session: (input) => - Effect.sync(() => { + Effect.gen(function* () { + const memoryState = yield* SessionStorageMemory.StateService memoryState.sessions.set(input.id, makeSessionRow(input)) }), userMessage: (input) => - Effect.sync(() => { - appendMemoryMessage(makeUserMessage(input)) + Effect.gen(function* () { + yield* appendMemoryMessage(makeUserMessage(input)) }), compaction: (input) => - Effect.sync(() => { - appendMemoryMessage(makeCompaction(input)) + Effect.gen(function* () { + yield* appendMemoryMessage(makeCompaction(input)) }), } -sessionStorageContract("SessionStorageMemory", SessionStorageMemory.layer(memoryState), memorySeeds) +sessionStorageContract("SessionStorageMemory", SessionStorageMemory.layer, memorySeeds) function seedProject() { Database.use((db) => @@ -323,7 +324,10 @@ function seedMessage( } function appendMemoryMessage(message: SessionMessage.Message) { - const current = memoryState.messages.get(sessionA) ?? [] - current.push(message) - memoryState.messages.set(sessionA, current) + return Effect.gen(function* () { + const memoryState = yield* SessionStorageMemory.StateService + const current = memoryState.messages.get(sessionA) ?? [] + current.push(message) + memoryState.messages.set(sessionA, current) + }) } From 2743504e606df9337fffc0c7c2a6df455854aaad Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 17:04:40 -0400 Subject: [PATCH 04/11] refactor(v2): simplify memory storage layer --- .../opencode/src/v2/session/storage-memory.ts | 138 +++++++++--------- .../opencode/test/v2/session-storage.test.ts | 32 ++-- 2 files changed, 84 insertions(+), 86 deletions(-) diff --git a/packages/opencode/src/v2/session/storage-memory.ts b/packages/opencode/src/v2/session/storage-memory.ts index d47d0678b1..42f4d9b72d 100644 --- a/packages/opencode/src/v2/session/storage-memory.ts +++ b/packages/opencode/src/v2/session/storage-memory.ts @@ -1,4 +1,4 @@ -import { Context, DateTime, Effect, Layer } from "effect" +import { DateTime, Effect, Layer } from "effect" import { SessionMessage } from "@opencode-ai/core/session-message" import { SessionStorage } from "./storage" @@ -7,86 +7,84 @@ export interface State { readonly messages: Map } -export class StateService extends Context.Service()("@opencode/v2/session/StorageMemoryState") {} - -const stateLayer = Layer.sync(StateService, () => ({ +const makeState = (): State => ({ sessions: new Map(), messages: new Map(), -})) +}) -const storageLayer = Layer.effect( - SessionStorage.Service, - Effect.gen(function* () { - const state = yield* StateService - return SessionStorage.Service.of({ - get: (sessionID) => Effect.sync(() => state.sessions.get(sessionID)), - list: (input) => - Effect.sync(() => { - const direction = input.cursor?.direction ?? "next" - const order = SessionStorage.pageOrder(input.order ?? "desc", direction) - const rows = Array.from(state.sessions.values()) - .filter((row) => { - if (input.directory && row.directory !== input.directory) return false - if (input.path && row.path !== input.path && !row.path?.startsWith(`${input.path}/`)) return false - if (input.workspaceID && row.workspaceID !== input.workspaceID) return false - if (input.roots && row.parentID) return false - if (input.start && DateTime.toEpochMillis(row.time.updated) < input.start) return false - if (input.search && !row.title.includes(input.search)) return false - if (!input.cursor) return true - return compareCursor(row.id, DateTime.toEpochMillis(row.time.updated), input.cursor, order) - }) - .toSorted((a, b) => - compareRows( - a.id, - DateTime.toEpochMillis(a.time.updated), - b.id, - DateTime.toEpochMillis(b.time.updated), - order, - ), - ) - const limited = input.limit === undefined ? rows : rows.slice(0, input.limit) - return direction === "previous" ? limited.toReversed() : limited - }), - messages: (input) => - Effect.sync(() => { - const direction = input.cursor?.direction ?? "next" - const order = SessionStorage.pageOrder(input.order ?? "desc", direction) - const rows = (state.messages.get(input.sessionID) ?? []) - .filter((message) => { - if (!input.cursor) return true - return compareCursor(message.id, DateTime.toEpochMillis(message.time.created), input.cursor, order) - }) - .toSorted((a, b) => - compareRows( - a.id, - DateTime.toEpochMillis(a.time.created), - b.id, - DateTime.toEpochMillis(b.time.created), - order, - ), - ) - const limited = input.limit === undefined ? rows : rows.slice(0, input.limit) - return direction === "previous" ? limited.toReversed() : limited - }), - context: (sessionID) => - Effect.sync(() => { - const messages = (state.messages.get(sessionID) ?? []).toSorted((a, b) => +export const make = (state = makeState()) => + SessionStorage.Service.of({ + get: (sessionID) => Effect.sync(() => state.sessions.get(sessionID)), + list: (input) => + Effect.sync(() => { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const rows = Array.from(state.sessions.values()) + .filter((row) => { + if (input.directory && row.directory !== input.directory) return false + if (input.path && row.path !== input.path && !row.path?.startsWith(`${input.path}/`)) return false + if (input.workspaceID && row.workspaceID !== input.workspaceID) return false + if (input.roots && row.parentID) return false + if (input.start && DateTime.toEpochMillis(row.time.updated) < input.start) return false + if (input.search && !row.title.includes(input.search)) return false + if (!input.cursor) return true + return compareCursor(row.id, DateTime.toEpochMillis(row.time.updated), input.cursor, order) + }) + .toSorted((a, b) => + compareRows( + a.id, + DateTime.toEpochMillis(a.time.updated), + b.id, + DateTime.toEpochMillis(b.time.updated), + order, + ), + ) + const limited = input.limit === undefined ? rows : rows.slice(0, input.limit) + return direction === "previous" ? limited.toReversed() : limited + }), + messages: (input) => + Effect.sync(() => { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const rows = (state.messages.get(input.sessionID) ?? []) + .filter((message) => { + if (!input.cursor) return true + return compareCursor(message.id, DateTime.toEpochMillis(message.time.created), input.cursor, order) + }) + .toSorted((a, b) => compareRows( a.id, DateTime.toEpochMillis(a.time.created), b.id, DateTime.toEpochMillis(b.time.created), - "asc", + order, ), ) - const index = messages.findLastIndex((message) => message.type === "compaction") - return index === -1 ? messages : messages.slice(index) - }), - }) - }), -) + const limited = input.limit === undefined ? rows : rows.slice(0, input.limit) + return direction === "previous" ? limited.toReversed() : limited + }), + context: (sessionID) => + Effect.sync(() => { + const messages = (state.messages.get(sessionID) ?? []).toSorted((a, b) => + compareRows( + a.id, + DateTime.toEpochMillis(a.time.created), + b.id, + DateTime.toEpochMillis(b.time.created), + "asc", + ), + ) + const index = messages.findLastIndex((message) => message.type === "compaction") + return index === -1 ? messages : messages.slice(index) + }), + }) -export const layer = storageLayer.pipe(Layer.provideMerge(stateLayer)) +const storageLayer = Layer.sync(SessionStorage.Service, () => { + const state = makeState() + return make(state) +}) + +export const layer = storageLayer export const defaultLayer = layer diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 262f613d6a..002b926b8b 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -19,6 +19,11 @@ const sessionB = SessionID.make("ses_storage_b") const sessionC = SessionID.make("ses_storage_c") const sessionD = SessionID.make("ses_storage_d") const encodeMessage = Schema.encodeSync(SessionMessage.Message) +const memoryState: SessionStorageMemory.State = { + sessions: new Map(), + messages: new Map(), +} +const memoryLayer = Layer.sync(SessionStorage.Service, () => SessionStorageMemory.make(memoryState)) interface Seeds { readonly reset: Effect.Effect @@ -162,29 +167,27 @@ const sqlSeeds: Seeds = { sessionStorageContract("SessionStorageSql", SessionStorageSql.defaultLayer, sqlSeeds) -const memorySeeds: Seeds = { - reset: Effect.gen(function* () { - const memoryState = yield* SessionStorageMemory.StateService +const memorySeeds: Seeds = { + reset: Effect.sync(() => { memoryState.sessions.clear() memoryState.messages.clear() }), project: Effect.void, session: (input) => - Effect.gen(function* () { - const memoryState = yield* SessionStorageMemory.StateService + Effect.sync(() => { memoryState.sessions.set(input.id, makeSessionRow(input)) }), userMessage: (input) => - Effect.gen(function* () { - yield* appendMemoryMessage(makeUserMessage(input)) + Effect.sync(() => { + appendMemoryMessage(makeUserMessage(input)) }), compaction: (input) => - Effect.gen(function* () { - yield* appendMemoryMessage(makeCompaction(input)) + Effect.sync(() => { + appendMemoryMessage(makeCompaction(input)) }), } -sessionStorageContract("SessionStorageMemory", SessionStorageMemory.layer, memorySeeds) +sessionStorageContract("SessionStorageMemory", memoryLayer, memorySeeds) function seedProject() { Database.use((db) => @@ -324,10 +327,7 @@ function seedMessage( } function appendMemoryMessage(message: SessionMessage.Message) { - return Effect.gen(function* () { - const memoryState = yield* SessionStorageMemory.StateService - const current = memoryState.messages.get(sessionA) ?? [] - current.push(message) - memoryState.messages.set(sessionA, current) - }) + const current = memoryState.messages.get(sessionA) ?? [] + current.push(message) + memoryState.messages.set(sessionA, current) } From 8b38c9b9998dfd0f76d230739b818fcb3fdb4351 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 17:06:06 -0400 Subject: [PATCH 05/11] refactor(v2): inline memory storage state --- .../opencode/src/v2/session/storage-memory.ts | 17 ++++++----------- .../opencode/test/v2/session-storage.test.ts | 2 +- 2 files changed, 7 insertions(+), 12 deletions(-) diff --git a/packages/opencode/src/v2/session/storage-memory.ts b/packages/opencode/src/v2/session/storage-memory.ts index 42f4d9b72d..13f2b147a9 100644 --- a/packages/opencode/src/v2/session/storage-memory.ts +++ b/packages/opencode/src/v2/session/storage-memory.ts @@ -2,17 +2,10 @@ import { DateTime, Effect, Layer } from "effect" import { SessionMessage } from "@opencode-ai/core/session-message" import { SessionStorage } from "./storage" -export interface State { +export const make = (state: { readonly sessions: Map readonly messages: Map -} - -const makeState = (): State => ({ - sessions: new Map(), - messages: new Map(), -}) - -export const make = (state = makeState()) => +}) => SessionStorage.Service.of({ get: (sessionID) => Effect.sync(() => state.sessions.get(sessionID)), list: (input) => @@ -80,8 +73,10 @@ export const make = (state = makeState()) => }) const storageLayer = Layer.sync(SessionStorage.Service, () => { - const state = makeState() - return make(state) + return make({ + sessions: new Map(), + messages: new Map(), + }) }) export const layer = storageLayer diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 002b926b8b..e68730a86c 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -19,7 +19,7 @@ const sessionB = SessionID.make("ses_storage_b") const sessionC = SessionID.make("ses_storage_c") const sessionD = SessionID.make("ses_storage_d") const encodeMessage = Schema.encodeSync(SessionMessage.Message) -const memoryState: SessionStorageMemory.State = { +const memoryState = { sessions: new Map(), messages: new Map(), } From dc55ece38447587a55d28c3228b5706701b1b445 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 17:07:34 -0400 Subject: [PATCH 06/11] refactor(v2): move session storage under storage --- packages/opencode/src/v2/session.ts | 4 ++-- .../storage-memory.ts => storage/session-memory.ts} | 4 ++-- .../v2/{session/storage-sql.ts => storage/session-sql.ts} | 4 ++-- .../src/v2/{session/storage.ts => storage/session.ts} | 2 +- packages/opencode/test/session/prompt.test.ts | 2 +- packages/opencode/test/v2/session-storage.test.ts | 6 +++--- specs/v2/storage-service.md | 2 +- 7 files changed, 12 insertions(+), 12 deletions(-) rename packages/opencode/src/v2/{session/storage-memory.ts => storage/session-memory.ts} (97%) rename packages/opencode/src/v2/{session/storage-sql.ts => storage/session-sql.ts} (98%) rename packages/opencode/src/v2/{session/storage.ts => storage/session.ts} (98%) diff --git a/packages/opencode/src/v2/session.ts b/packages/opencode/src/v2/session.ts index 90f344c131..5b03f4063d 100644 --- a/packages/opencode/src/v2/session.ts +++ b/packages/opencode/src/v2/session.ts @@ -10,8 +10,8 @@ import { optionalOmitUndefined } from "@opencode-ai/core/schema" import { EventV2 } from "@opencode-ai/core/event" import { EventV2Bridge } from "@/event-v2-bridge" import { ModelV2 } from "@opencode-ai/core/model" -import { SessionStorage } from "./session/storage" -import { SessionStorageSql } from "./session/storage-sql" +import { SessionStorage } from "./storage/session" +import { SessionStorageSql } from "./storage/session-sql" export const Delivery = Schema.Literals(["immediate", "deferred"]).annotate({ identifier: "Session.Delivery", diff --git a/packages/opencode/src/v2/session/storage-memory.ts b/packages/opencode/src/v2/storage/session-memory.ts similarity index 97% rename from packages/opencode/src/v2/session/storage-memory.ts rename to packages/opencode/src/v2/storage/session-memory.ts index 13f2b147a9..ecd9d450d5 100644 --- a/packages/opencode/src/v2/session/storage-memory.ts +++ b/packages/opencode/src/v2/storage/session-memory.ts @@ -1,6 +1,6 @@ import { DateTime, Effect, Layer } from "effect" import { SessionMessage } from "@opencode-ai/core/session-message" -import { SessionStorage } from "./storage" +import { SessionStorage } from "./session" export const make = (state: { readonly sessions: Map @@ -98,4 +98,4 @@ function compareRows(aID: string, aTime: number, bID: string, bTime: number, ord return order === "asc" ? result : -result } -export * as SessionStorageMemory from "./storage-memory" +export * as SessionStorageMemory from "./session-memory" diff --git a/packages/opencode/src/v2/session/storage-sql.ts b/packages/opencode/src/v2/storage/session-sql.ts similarity index 98% rename from packages/opencode/src/v2/session/storage-sql.ts rename to packages/opencode/src/v2/storage/session-sql.ts index 6487e1176a..48a6137115 100644 --- a/packages/opencode/src/v2/session/storage-sql.ts +++ b/packages/opencode/src/v2/storage/session-sql.ts @@ -2,7 +2,7 @@ import { SessionMessageTable, SessionTable } from "@/session/session.sql" import { and, asc, Database, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" import { SessionMessage } from "@opencode-ai/core/session-message" import { Effect, Layer, Schema } from "effect" -import { SessionStorage } from "./storage" +import { SessionStorage } from "./session" const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message) const decodeSessionRow = Schema.decodeUnknownSync(SessionStorage.SessionRow) @@ -170,4 +170,4 @@ function fromSessionRow(row: typeof SessionTable.$inferSelect) { }) } -export * as SessionStorageSql from "./storage-sql" +export * as SessionStorageSql from "./session-sql" diff --git a/packages/opencode/src/v2/session/storage.ts b/packages/opencode/src/v2/storage/session.ts similarity index 98% rename from packages/opencode/src/v2/session/storage.ts rename to packages/opencode/src/v2/storage/session.ts index c86289d741..7517b88f77 100644 --- a/packages/opencode/src/v2/session/storage.ts +++ b/packages/opencode/src/v2/storage/session.ts @@ -97,4 +97,4 @@ export function pageOrder(order: SortOrder, direction: PageDirection) { export class Service extends Context.Service()("@opencode/v2/session/Storage") {} -export * as SessionStorage from "./storage" +export * as SessionStorage from "./session" diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index ccce5b4015..e30f89d39d 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -36,7 +36,7 @@ import { SessionRunState } from "../../src/session/run-state" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { SessionStatus } from "../../src/session/status" import { SessionV2 } from "../../src/v2/session" -import { SessionStorageSql } from "../../src/v2/session/storage-sql" +import { SessionStorageSql } from "../../src/v2/storage/session-sql" import { Skill } from "../../src/skill" import { SystemPrompt } from "../../src/session/system" import { Shell } from "../../src/shell/shell" diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index e68730a86c..d03748a048 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -4,9 +4,9 @@ import { ProjectTable } from "@/project/project.sql" import { SessionID } from "@/session/schema" import { SessionMessageTable, SessionTable } from "@/session/session.sql" import { Database } from "@/storage/db" -import { SessionStorage } from "@/v2/session/storage" -import { SessionStorageMemory } from "@/v2/session/storage-memory" -import { SessionStorageSql } from "@/v2/session/storage-sql" +import { SessionStorage } from "@/v2/storage/session" +import { SessionStorageMemory } from "@/v2/storage/session-memory" +import { SessionStorageSql } from "@/v2/storage/session-sql" import { EventV2 } from "@opencode-ai/core/event" import { SessionMessage } from "@opencode-ai/core/session-message" import { eq, or } from "@/storage/db" diff --git a/specs/v2/storage-service.md b/specs/v2/storage-service.md index d1fead2996..d289b1d31c 100644 --- a/specs/v2/storage-service.md +++ b/specs/v2/storage-service.md @@ -465,7 +465,7 @@ Avoid leaking Drizzle or SQLite error codes above the storage implementation. Fo The first implementation PR should be small and reversible: -- Add `packages/opencode/src/v2/session/storage.ts` with the `SessionStorage` service shape. +- Add `packages/opencode/src/v2/storage/session.ts` with the `SessionStorage` service shape. - Implement SQL and in-memory read backends for `get`, `list`, `messages`, and `context`. - Refactor `v2/session.ts` to consume `SessionStorage.Service` instead of raw Drizzle calls. - Keep table schemas and existing migrations unchanged. From dcbd244dd73d39703a4aa33810f4b10825ae6be2 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 20:34:02 -0400 Subject: [PATCH 07/11] refactor(v2): use effect sqlite session storage --- bun.lock | 2 + .../src/effect-sqlite/migrator.ts | 21 ++- packages/effect-drizzle-sqlite/src/index.ts | 2 +- packages/opencode/package.json | 2 + packages/opencode/src/storage/db.ts | 45 ++--- .../opencode/src/v2/storage/session-sql.ts | 169 ++++++++++-------- .../opencode/test/v2/session-storage.test.ts | 62 ++++--- 7 files changed, 181 insertions(+), 122 deletions(-) diff --git a/bun.lock b/bun.lock index 037b72a29b..a34b0cdea1 100644 --- a/bun.lock +++ b/bun.lock @@ -429,12 +429,14 @@ "@clack/prompts": "1.0.0-alpha.1", "@effect/opentelemetry": "catalog:", "@effect/platform-node": "catalog:", + "@effect/sql-sqlite-bun": "catalog:", "@gitlab/opencode-gitlab-auth": "1.3.3", "@lydell/node-pty": "catalog:", "@modelcontextprotocol/sdk": "1.27.1", "@octokit/graphql": "9.0.2", "@octokit/rest": "catalog:", "@openauthjs/openauth": "catalog:", + "@opencode-ai/effect-drizzle-sqlite": "workspace:*", "@opencode-ai/llm": "workspace:*", "@opencode-ai/plugin": "workspace:*", "@opencode-ai/script": "workspace:*", diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts index 6d0d155143..91aacccf37 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts @@ -1,7 +1,8 @@ /* oxlint-disable */ -import type { MigrationConfig } from "drizzle-orm/migrator" +import type { MigrationConfig, MigrationFromJournalConfig, MigrationsJournal } from "drizzle-orm/migrator" import { readMigrationFiles } from "drizzle-orm/migrator" import type { AnyRelations } from "drizzle-orm/relations" +import crypto from "node:crypto" import { migrate as coreMigrate } from "../sqlite-core/effect/session" import type { EffectSQLiteDatabase } from "./driver" @@ -12,3 +13,21 @@ export function migrate( const migrations = readMigrationFiles(config) return coreMigrate(migrations, db.session, config) } + +export function migrateFromJournal( + db: EffectSQLiteDatabase, + journal: MigrationsJournal, + config: Omit = {}, +) { + return coreMigrate( + journal.map((migration) => ({ + sql: migration.sql.split("--> statement-breakpoint"), + bps: true, + folderMillis: migration.timestamp, + hash: crypto.createHash("sha256").update(migration.sql).digest("hex"), + name: migration.name, + })), + db.session, + { migrationsFolder: "", migrationsTable: config.migrationsTable }, + ) +} diff --git a/packages/effect-drizzle-sqlite/src/index.ts b/packages/effect-drizzle-sqlite/src/index.ts index d6606b7d9f..fb5ccd6757 100644 --- a/packages/effect-drizzle-sqlite/src/index.ts +++ b/packages/effect-drizzle-sqlite/src/index.ts @@ -1,6 +1,6 @@ export { EffectLogger } from "drizzle-orm/effect-core" export * from "./effect-sqlite/driver" export * from "./effect-sqlite/session" -export { migrate } from "./effect-sqlite/migrator" +export { migrate, migrateFromJournal } from "./effect-sqlite/migrator" export * as EffectDrizzleSqlite from "." diff --git a/packages/opencode/package.json b/packages/opencode/package.json index 3c6e1076b1..37a2b3869b 100644 --- a/packages/opencode/package.json +++ b/packages/opencode/package.json @@ -96,12 +96,14 @@ "@clack/prompts": "1.0.0-alpha.1", "@effect/opentelemetry": "catalog:", "@effect/platform-node": "catalog:", + "@effect/sql-sqlite-bun": "catalog:", "@gitlab/opencode-gitlab-auth": "1.3.3", "@lydell/node-pty": "catalog:", "@modelcontextprotocol/sdk": "1.27.1", "@octokit/graphql": "9.0.2", "@octokit/rest": "catalog:", "@openauthjs/openauth": "catalog:", + "@opencode-ai/effect-drizzle-sqlite": "workspace:*", "@opencode-ai/llm": "workspace:*", "@opencode-ai/plugin": "workspace:*", "@opencode-ai/script": "workspace:*", diff --git a/packages/opencode/src/storage/db.ts b/packages/opencode/src/storage/db.ts index 06f1f84a9a..55600bd1a4 100644 --- a/packages/opencode/src/storage/db.ts +++ b/packages/opencode/src/storage/db.ts @@ -1,5 +1,6 @@ import { type SQLiteBunDatabase } from "drizzle-orm/bun-sqlite" import { migrate } from "drizzle-orm/bun-sqlite/migrator" +import type { MigrationsJournal } from "drizzle-orm/migrator" import { type SQLiteTransaction } from "drizzle-orm/sqlite-core" export * from "drizzle-orm" import { RuntimeFlags } from "@/effect/runtime-flags" @@ -47,13 +48,19 @@ export type Transaction = SQLiteTransaction<"sync", void> type Client = ReturnType -type Journal = { sql: string; timestamp: number; name: string }[] - -// Drizzle's migrate overloads trigger expensive variance checks here; narrow to the journal overload we actually use. -const migrateFromJournal = migrate as unknown as (db: SQLiteBunDatabase, entries: Journal) => void +export type Journal = MigrationsJournal function applyMigrations(db: SQLiteBunDatabase, entries: Journal) { - migrateFromJournal(db, entries) + migrate(db, entries) +} + +export function migrationJournal(flags: Pick = readRuntimeFlags()) { + const entries = + typeof OPENCODE_MIGRATIONS !== "undefined" + ? OPENCODE_MIGRATIONS + : migrations(path.join(import.meta.dirname, "../../migration")) + if (!flags.skipMigrations) return entries + return entries.map((item) => ({ ...item, sql: "select 1;" })) } function time(tag: string) { @@ -74,17 +81,17 @@ function migrations(dir: string): Journal { .filter((entry) => entry.isDirectory()) .map((entry) => entry.name) - const sql = dirs + const sql: Journal = dirs .map((name) => { const file = path.join(dir, name, "migration.sql") - if (!existsSync(file)) return + if (!existsSync(file)) return undefined return { sql: readFileSync(file, "utf-8"), timestamp: time(name), name, } }) - .filter(Boolean) as Journal + .filter((entry) => entry !== undefined) return sql.sort((a, b) => a.timestamp - b.timestamp) } @@ -94,7 +101,7 @@ let loaded = false export const Client = Object.assign( (flags: DatabaseFlags = readRuntimeFlags()): Client => { - if (loaded) return client as Client + if (loaded && client) return client const dbPath = getPath(flags) log.info("opening database", { path: dbPath }) @@ -109,20 +116,12 @@ export const Client = Object.assign( db.run("PRAGMA wal_checkpoint(PASSIVE)") // Apply schema migrations - const entries = - typeof OPENCODE_MIGRATIONS !== "undefined" - ? OPENCODE_MIGRATIONS - : migrations(path.join(import.meta.dirname, "../../migration")) + const entries = migrationJournal(flags) if (entries.length > 0) { log.info("applying migrations", { count: entries.length, mode: typeof OPENCODE_MIGRATIONS !== "undefined" ? "bundled" : "dev", }) - if (flags.skipMigrations) { - for (const item of entries) { - item.sql = "select 1;" - } - } applyMigrations(db, entries) } @@ -159,19 +158,19 @@ export function use(callback: (trx: TxOrDb) => T): T { if (err instanceof LocalContext.NotFound) { const effects: (() => void | Promise)[] = [] const result = ctx.provide({ effects, tx: Client() }, () => callback(Client())) - for (const effect of effects) effect() + for (const effect of effects) void effect() return result } throw err } } -export function effect(fn: () => any | Promise) { +export function effect(fn: () => void | Promise) { const bound = EffectBridge.bind(fn) try { ctx.use().effects.push(bound) } catch { - bound() + void bound() } } @@ -190,7 +189,9 @@ export function transaction( const effects: (() => void | Promise)[] = [] const txCallback = EffectBridge.bind((tx: TxOrDb) => ctx.provide({ tx, effects }, () => callback(tx))) const result = Client().transaction(txCallback, { behavior: options?.behavior }) - for (const effect of effects) effect() + for (const effect of effects) void effect() + // Drizzle's transaction type does not preserve our NotPromise constraint through the callback wrapper. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion return result as NotPromise } throw err diff --git a/packages/opencode/src/v2/storage/session-sql.ts b/packages/opencode/src/v2/storage/session-sql.ts index 48a6137115..80819ace6a 100644 --- a/packages/opencode/src/v2/storage/session-sql.ts +++ b/packages/opencode/src/v2/storage/session-sql.ts @@ -1,87 +1,110 @@ import { SessionMessageTable, SessionTable } from "@/session/session.sql" -import { and, asc, Database, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" +import { and, asc, Database as LegacyDatabase, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" +import { SqliteClient } from "@effect/sql-sqlite-bun" import { SessionMessage } from "@opencode-ai/core/session-message" -import { Effect, Layer, Schema } from "effect" +import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" +import { Context, Effect, Layer, Schema } from "effect" import { SessionStorage } from "./session" const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message) const decodeSessionRow = Schema.decodeUnknownSync(SessionStorage.SessionRow) +const makeDatabase = EffectDrizzleSqlite.makeWithDefaults() +type DatabaseShape = Effect.Success + +export class Database extends Context.Service()("@opencode/v2/session/StorageSql/Database") {} + +export const databaseLayer = Layer.unwrap( + Effect.sync(() => { + const filename = LegacyDatabase.getPath() + return Layer.effect( + Database, + Effect.gen(function* () { + const db = yield* makeDatabase + yield* db.run("PRAGMA journal_mode = WAL") + yield* db.run("PRAGMA synchronous = NORMAL") + yield* db.run("PRAGMA busy_timeout = 5000") + yield* db.run("PRAGMA cache_size = -64000") + yield* db.run("PRAGMA foreign_keys = ON") + yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") + yield* EffectDrizzleSqlite.migrateFromJournal(db, LegacyDatabase.migrationJournal()) + return db + }), + ).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" }))) + }), +) export const layer = Layer.effect( SessionStorage.Service, Effect.gen(function* () { - const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")((sessionID) => - attempt(() => - Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()), - ).pipe(Effect.map((row) => (row ? fromSessionRow(row) : undefined))), - ) + const db = yield* Database - const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")((input) => - attempt(() => { - const direction = input.cursor?.direction ?? "next" - const order = SessionStorage.pageOrder(input.order ?? "desc", direction) - const sortColumn = SessionTable.time_updated - const conditions: SQL[] = [] - if (input.directory) conditions.push(eq(SessionTable.directory, input.directory)) - if (input.path) - conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!) - if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) - if (input.roots) conditions.push(isNull(SessionTable.parent_id)) - if (input.start) conditions.push(gte(sortColumn, input.start)) - if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) - if (input.cursor) conditions.push(sessionCursorBoundary(input.cursor, order)) + const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")(function* (sessionID) { + const row = yield* attempt(db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()) + return row ? fromSessionRow(row) : undefined + }) - return Database.use((db) => { - const query = db - .select() - .from(SessionTable) - .where(conditions.length > 0 ? and(...conditions) : undefined) - .orderBy( - order === "asc" ? asc(sortColumn) : desc(sortColumn), - order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), - ) - const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() - return direction === "previous" ? rows.toReversed() : rows - }) - }).pipe(Effect.map((rows) => rows.map(fromSessionRow))), - ) + const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")(function* (input) { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const sortColumn = SessionTable.time_updated + const conditions: SQL[] = [] + if (input.directory) conditions.push(eq(SessionTable.directory, input.directory)) + if (input.path) + conditions.push(or(eq(SessionTable.path, input.path), like(SessionTable.path, `${input.path}/%`))!) + if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID)) + if (input.roots) conditions.push(isNull(SessionTable.parent_id)) + if (input.start) conditions.push(gte(sortColumn, input.start)) + if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`)) + if (input.cursor) conditions.push(sessionCursorBoundary(input.cursor, order)) - const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")((input) => - attempt(() => { - const direction = input.cursor?.direction ?? "next" - const order = SessionStorage.pageOrder(input.order ?? "desc", direction) - const boundary = input.cursor ? messageCursorBoundary(input.cursor, order) : undefined - const where = boundary - ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) - : eq(SessionMessageTable.session_id, input.sessionID) + const query = db + .select() + .from(SessionTable) + .where(conditions.length > 0 ? and(...conditions) : undefined) + .orderBy( + order === "asc" ? asc(sortColumn) : desc(sortColumn), + order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), + ) + const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit)) + return (direction === "previous" ? rows.toReversed() : rows).map(fromSessionRow) + }) - return Database.use((db) => { - const query = db - .select() - .from(SessionMessageTable) - .where(where) - .orderBy( - order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), - order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), - ) - const rows = input.limit === undefined ? query.all() : query.limit(input.limit).all() - return direction === "previous" ? rows.toReversed() : rows - }) - }).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))), - ) + const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")(function* (input) { + const direction = input.cursor?.direction ?? "next" + const order = SessionStorage.pageOrder(input.order ?? "desc", direction) + const boundary = input.cursor ? messageCursorBoundary(input.cursor, order) : undefined + const where = boundary + ? and(eq(SessionMessageTable.session_id, input.sessionID), boundary) + : eq(SessionMessageTable.session_id, input.sessionID) + + const query = db + .select() + .from(SessionMessageTable) + .where(where) + .orderBy( + order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), + order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), + ) + const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit)) + return (direction === "previous" ? rows.toReversed() : rows).map((row) => + decodeMessage({ ...row.data, id: row.id, type: row.type }), + ) + }) const context: SessionStorage.Interface["context"] = Effect.fn("SessionStorageSql.context")((sessionID) => - attempt(() => - Database.use((db) => { - const compaction = db + Effect.gen(function* () { + const compaction = yield* attempt( + db .select() .from(SessionMessageTable) .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id)) .limit(1) - .get() + .get(), + ) - return db + const rows = yield* attempt( + db .select() .from(SessionMessageTable) .where( @@ -98,23 +121,25 @@ export const layer = Layer.effect( : undefined, ), ) - .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)) - .all() - }), - ).pipe(Effect.map((rows) => rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })))), + .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)), + ) + + return rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })) + }), ) return SessionStorage.Service.of({ get, list, messages, context }) }), ) -export const defaultLayer = layer +export const defaultLayer = layer.pipe(Layer.provide(databaseLayer.pipe(Layer.orDie))) -function attempt(body: () => A) { - return Effect.try({ - try: body, - catch: (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), - }) +function attempt(effect: Effect.Effect) { + return effect.pipe( + Effect.mapError( + (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), + ), + ) } function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) { diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index d03748a048..7533bf0ce4 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -3,7 +3,6 @@ import { ProjectID } from "@/project/schema" import { ProjectTable } from "@/project/project.sql" import { SessionID } from "@/session/schema" import { SessionMessageTable, SessionTable } from "@/session/session.sql" -import { Database } from "@/storage/db" import { SessionStorage } from "@/v2/storage/session" import { SessionStorageMemory } from "@/v2/storage/session-memory" import { SessionStorageSql } from "@/v2/storage/session-sql" @@ -157,15 +156,17 @@ function sessionStorageContract(name: string, layer: Layer.Layer = { - reset: Effect.sync(resetSqlSeeds), - project: Effect.sync(seedProject), - session: (input) => Effect.sync(() => seedSession(input)), - userMessage: (input) => Effect.sync(() => seedUserMessage(input)), - compaction: (input) => Effect.sync(() => seedCompaction(input)), +const sqlLayer = SessionStorageSql.layer.pipe(Layer.provideMerge(SessionStorageSql.databaseLayer)) + +const sqlSeeds: Seeds = { + reset: resetSqlSeeds(), + project: seedProject(), + session: seedSession, + userMessage: seedUserMessage, + compaction: seedCompaction, } -sessionStorageContract("SessionStorageSql", SessionStorageSql.defaultLayer, sqlSeeds) +sessionStorageContract("SessionStorageSql", sqlLayer, sqlSeeds) const memorySeeds: Seeds = { reset: Effect.sync(() => { @@ -190,8 +191,9 @@ const memorySeeds: Seeds = { sessionStorageContract("SessionStorageMemory", memoryLayer, memorySeeds) function seedProject() { - Database.use((db) => - db + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db .insert(ProjectTable) .values({ id: projectID, @@ -201,14 +203,17 @@ function seedProject() { sandboxes: [], }) .onConflictDoNothing() - .run(), - ) + .run() + .pipe(Effect.orDie) + }) } function resetSqlSeeds() { - Database.use((db) => { - db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run() - db.delete(SessionTable) + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run().pipe(Effect.orDie) + yield* db + .delete(SessionTable) .where( or( eq(SessionTable.id, sessionA), @@ -218,13 +223,15 @@ function resetSqlSeeds() { ), ) .run() - db.delete(ProjectTable).where(eq(ProjectTable.id, projectID)).run() + .pipe(Effect.orDie) + yield* db.delete(ProjectTable).where(eq(ProjectTable.id, projectID)).run().pipe(Effect.orDie) }) } function seedSession(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { - Database.use((db) => - db + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db .insert(SessionTable) .values({ id: input.id, @@ -243,20 +250,21 @@ function seedSession(input: { id: SessionID; title: string; directory?: string; time_created: input.updated, time_updated: input.updated, }) - .run(), - ) + .run() + .pipe(Effect.orDie) + }) } function seedUserMessage(input: { id: SessionMessage.ID; text: string; time: number }) { const encoded = encodeMessage(makeUserMessage(input)) const { id: _, type: __, ...data } = encoded - seedMessage(input.id, "user", input.time, data) + return seedMessage(input.id, "user", input.time, data) } function seedCompaction(input: { id: SessionMessage.ID; summary: string; time: number }) { const encoded = encodeMessage(makeCompaction(input)) const { id: _, type: __, ...data } = encoded - seedMessage(input.id, "compaction", input.time, data) + return seedMessage(input.id, "compaction", input.time, data) } function makeSessionRow(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { @@ -311,8 +319,9 @@ function seedMessage( time: number, data: typeof SessionMessageTable.$inferInsert.data, ) { - Database.use((db) => - db + return Effect.gen(function* () { + const db = yield* SessionStorageSql.Database + yield* db .insert(SessionMessageTable) .values({ id, @@ -322,8 +331,9 @@ function seedMessage( time_updated: time, data, }) - .run(), - ) + .run() + .pipe(Effect.orDie) + }) } function appendMemoryMessage(message: SessionMessage.Message) { From b3d6f931484137d271039272a9cc756c741f9095 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 20:41:57 -0400 Subject: [PATCH 08/11] refactor(v2): share effect sqlite database layer --- .../src/effect-sqlite/migrator.ts | 21 +--- packages/effect-drizzle-sqlite/src/index.ts | 2 +- packages/opencode/src/storage/db.ts | 21 ++-- packages/opencode/src/v2/storage/database.ts | 39 ++++++ .../opencode/src/v2/storage/session-sql.ts | 112 ++++++------------ .../opencode/test/v2/session-storage.test.ts | 13 +- 6 files changed, 97 insertions(+), 111 deletions(-) create mode 100644 packages/opencode/src/v2/storage/database.ts diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts index 91aacccf37..6d0d155143 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/migrator.ts @@ -1,8 +1,7 @@ /* oxlint-disable */ -import type { MigrationConfig, MigrationFromJournalConfig, MigrationsJournal } from "drizzle-orm/migrator" +import type { MigrationConfig } from "drizzle-orm/migrator" import { readMigrationFiles } from "drizzle-orm/migrator" import type { AnyRelations } from "drizzle-orm/relations" -import crypto from "node:crypto" import { migrate as coreMigrate } from "../sqlite-core/effect/session" import type { EffectSQLiteDatabase } from "./driver" @@ -13,21 +12,3 @@ export function migrate( const migrations = readMigrationFiles(config) return coreMigrate(migrations, db.session, config) } - -export function migrateFromJournal( - db: EffectSQLiteDatabase, - journal: MigrationsJournal, - config: Omit = {}, -) { - return coreMigrate( - journal.map((migration) => ({ - sql: migration.sql.split("--> statement-breakpoint"), - bps: true, - folderMillis: migration.timestamp, - hash: crypto.createHash("sha256").update(migration.sql).digest("hex"), - name: migration.name, - })), - db.session, - { migrationsFolder: "", migrationsTable: config.migrationsTable }, - ) -} diff --git a/packages/effect-drizzle-sqlite/src/index.ts b/packages/effect-drizzle-sqlite/src/index.ts index fb5ccd6757..d6606b7d9f 100644 --- a/packages/effect-drizzle-sqlite/src/index.ts +++ b/packages/effect-drizzle-sqlite/src/index.ts @@ -1,6 +1,6 @@ export { EffectLogger } from "drizzle-orm/effect-core" export * from "./effect-sqlite/driver" export * from "./effect-sqlite/session" -export { migrate, migrateFromJournal } from "./effect-sqlite/migrator" +export { migrate } from "./effect-sqlite/migrator" export * as EffectDrizzleSqlite from "." diff --git a/packages/opencode/src/storage/db.ts b/packages/opencode/src/storage/db.ts index 55600bd1a4..6df731b530 100644 --- a/packages/opencode/src/storage/db.ts +++ b/packages/opencode/src/storage/db.ts @@ -48,21 +48,12 @@ export type Transaction = SQLiteTransaction<"sync", void> type Client = ReturnType -export type Journal = MigrationsJournal +type Journal = MigrationsJournal function applyMigrations(db: SQLiteBunDatabase, entries: Journal) { migrate(db, entries) } -export function migrationJournal(flags: Pick = readRuntimeFlags()) { - const entries = - typeof OPENCODE_MIGRATIONS !== "undefined" - ? OPENCODE_MIGRATIONS - : migrations(path.join(import.meta.dirname, "../../migration")) - if (!flags.skipMigrations) return entries - return entries.map((item) => ({ ...item, sql: "select 1;" })) -} - function time(tag: string) { const match = /^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})/.exec(tag) if (!match) return 0 @@ -116,12 +107,20 @@ export const Client = Object.assign( db.run("PRAGMA wal_checkpoint(PASSIVE)") // Apply schema migrations - const entries = migrationJournal(flags) + const entries = + typeof OPENCODE_MIGRATIONS !== "undefined" + ? OPENCODE_MIGRATIONS + : migrations(path.join(import.meta.dirname, "../../migration")) if (entries.length > 0) { log.info("applying migrations", { count: entries.length, mode: typeof OPENCODE_MIGRATIONS !== "undefined" ? "bundled" : "dev", }) + if (flags.skipMigrations) { + for (const item of entries) { + item.sql = "select 1;" + } + } applyMigrations(db, entries) } diff --git a/packages/opencode/src/v2/storage/database.ts b/packages/opencode/src/v2/storage/database.ts new file mode 100644 index 0000000000..47c5a375a6 --- /dev/null +++ b/packages/opencode/src/v2/storage/database.ts @@ -0,0 +1,39 @@ +import { Database as LegacyDatabase } from "@/storage/db" +import { SqliteClient } from "@effect/sql-sqlite-bun" +import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" +import { Context, Effect, Layer } from "effect" +import path from "path" + +const makeDatabase = EffectDrizzleSqlite.makeWithDefaults() +type DatabaseShape = Effect.Success + +export class Service extends Context.Service()("@opencode/v2/storage/Database") {} + +export const layer = Layer.unwrap( + Effect.sync(() => { + const filename = LegacyDatabase.getPath() + return Layer.effect( + Service, + Effect.gen(function* () { + LegacyDatabase.Client() + const db = yield* makeDatabase + yield* db.run("PRAGMA journal_mode = WAL") + yield* db.run("PRAGMA synchronous = NORMAL") + yield* db.run("PRAGMA busy_timeout = 5000") + yield* db.run("PRAGMA cache_size = -64000") + yield* db.run("PRAGMA foreign_keys = ON") + yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") + if (filename === ":memory:") { + yield* EffectDrizzleSqlite.migrate(db, { + migrationsFolder: path.join(import.meta.dirname, "../../../migration"), + }) + } + return db + }), + ).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" }))) + }), +) + +export const defaultLayer = layer + +export * as StorageDatabase from "./database" diff --git a/packages/opencode/src/v2/storage/session-sql.ts b/packages/opencode/src/v2/storage/session-sql.ts index 80819ace6a..cc0ed781eb 100644 --- a/packages/opencode/src/v2/storage/session-sql.ts +++ b/packages/opencode/src/v2/storage/session-sql.ts @@ -1,47 +1,25 @@ import { SessionMessageTable, SessionTable } from "@/session/session.sql" -import { and, asc, Database as LegacyDatabase, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" -import { SqliteClient } from "@effect/sql-sqlite-bun" +import { and, asc, desc, eq, gt, gte, isNull, like, lt, or, type SQL } from "@/storage/db" import { SessionMessage } from "@opencode-ai/core/session-message" -import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" -import { Context, Effect, Layer, Schema } from "effect" +import { Effect, Layer, Schema } from "effect" +import { StorageDatabase } from "./database" import { SessionStorage } from "./session" const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message) const decodeSessionRow = Schema.decodeUnknownSync(SessionStorage.SessionRow) -const makeDatabase = EffectDrizzleSqlite.makeWithDefaults() -type DatabaseShape = Effect.Success - -export class Database extends Context.Service()("@opencode/v2/session/StorageSql/Database") {} - -export const databaseLayer = Layer.unwrap( - Effect.sync(() => { - const filename = LegacyDatabase.getPath() - return Layer.effect( - Database, - Effect.gen(function* () { - const db = yield* makeDatabase - yield* db.run("PRAGMA journal_mode = WAL") - yield* db.run("PRAGMA synchronous = NORMAL") - yield* db.run("PRAGMA busy_timeout = 5000") - yield* db.run("PRAGMA cache_size = -64000") - yield* db.run("PRAGMA foreign_keys = ON") - yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") - yield* EffectDrizzleSqlite.migrateFromJournal(db, LegacyDatabase.migrationJournal()) - return db - }), - ).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" }))) - }), +const mapStorageError = Effect.mapError( + (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), ) export const layer = Layer.effect( SessionStorage.Service, Effect.gen(function* () { - const db = yield* Database + const db = yield* StorageDatabase.Service const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")(function* (sessionID) { - const row = yield* attempt(db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()) + const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get() return row ? fromSessionRow(row) : undefined - }) + }, mapStorageError) const list: SessionStorage.Interface["list"] = Effect.fn("SessionStorageSql.list")(function* (input) { const direction = input.cursor?.direction ?? "next" @@ -65,9 +43,9 @@ export const layer = Layer.effect( order === "asc" ? asc(sortColumn) : desc(sortColumn), order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), ) - const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit)) + const rows = yield* input.limit === undefined ? query : query.limit(input.limit) return (direction === "previous" ? rows.toReversed() : rows).map(fromSessionRow) - }) + }, mapStorageError) const messages: SessionStorage.Interface["messages"] = Effect.fn("SessionStorageSql.messages")(function* (input) { const direction = input.cursor?.direction ?? "next" @@ -85,62 +63,50 @@ export const layer = Layer.effect( order === "asc" ? asc(SessionMessageTable.time_created) : desc(SessionMessageTable.time_created), order === "asc" ? asc(SessionMessageTable.id) : desc(SessionMessageTable.id), ) - const rows = yield* attempt(input.limit === undefined ? query : query.limit(input.limit)) + const rows = yield* input.limit === undefined ? query : query.limit(input.limit) return (direction === "previous" ? rows.toReversed() : rows).map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type }), ) - }) + }, mapStorageError) const context: SessionStorage.Interface["context"] = Effect.fn("SessionStorageSql.context")((sessionID) => Effect.gen(function* () { - const compaction = yield* attempt( - db - .select() - .from(SessionMessageTable) - .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) - .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id)) - .limit(1) - .get(), - ) + const compaction = yield* db + .select() + .from(SessionMessageTable) + .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) + .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id)) + .limit(1) + .get() - const rows = yield* attempt( - db - .select() - .from(SessionMessageTable) - .where( - and( - eq(SessionMessageTable.session_id, sessionID), - compaction - ? or( - gt(SessionMessageTable.time_created, compaction.time_created), - and( - eq(SessionMessageTable.time_created, compaction.time_created), - gte(SessionMessageTable.id, compaction.id), - ), - ) - : undefined, - ), - ) - .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)), - ) + const rows = yield* db + .select() + .from(SessionMessageTable) + .where( + and( + eq(SessionMessageTable.session_id, sessionID), + compaction + ? or( + gt(SessionMessageTable.time_created, compaction.time_created), + and( + eq(SessionMessageTable.time_created, compaction.time_created), + gte(SessionMessageTable.id, compaction.id), + ), + ) + : undefined, + ), + ) + .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id)) return rows.map((row) => decodeMessage({ ...row.data, id: row.id, type: row.type })) - }), + }).pipe(mapStorageError), ) return SessionStorage.Service.of({ get, list, messages, context }) }), ) -export const defaultLayer = layer.pipe(Layer.provide(databaseLayer.pipe(Layer.orDie))) - -function attempt(effect: Effect.Effect) { - return effect.pipe( - Effect.mapError( - (cause) => new SessionStorage.StorageError({ message: "Session storage SQL operation failed", cause }), - ), - ) -} +export const defaultLayer = layer.pipe(Layer.provide(StorageDatabase.defaultLayer.pipe(Layer.orDie))) function sessionCursorBoundary(cursor: SessionStorage.SessionCursor, order: SessionStorage.SortOrder) { if (order === "asc") diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 7533bf0ce4..7083950a1b 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -4,6 +4,7 @@ import { ProjectTable } from "@/project/project.sql" import { SessionID } from "@/session/schema" import { SessionMessageTable, SessionTable } from "@/session/session.sql" import { SessionStorage } from "@/v2/storage/session" +import { StorageDatabase } from "@/v2/storage/database" import { SessionStorageMemory } from "@/v2/storage/session-memory" import { SessionStorageSql } from "@/v2/storage/session-sql" import { EventV2 } from "@opencode-ai/core/event" @@ -156,9 +157,9 @@ function sessionStorageContract(name: string, layer: Layer.Layer = { +const sqlSeeds: Seeds = { reset: resetSqlSeeds(), project: seedProject(), session: seedSession, @@ -192,7 +193,7 @@ sessionStorageContract("SessionStorageMemory", memoryLayer, memorySeeds) function seedProject() { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db .insert(ProjectTable) .values({ @@ -210,7 +211,7 @@ function seedProject() { function resetSqlSeeds() { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run().pipe(Effect.orDie) yield* db .delete(SessionTable) @@ -230,7 +231,7 @@ function resetSqlSeeds() { function seedSession(input: { id: SessionID; title: string; directory?: string; path: string; updated: number }) { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db .insert(SessionTable) .values({ @@ -320,7 +321,7 @@ function seedMessage( data: typeof SessionMessageTable.$inferInsert.data, ) { return Effect.gen(function* () { - const db = yield* SessionStorageSql.Database + const db = yield* StorageDatabase.Service yield* db .insert(SessionMessageTable) .values({ From 71423b9a5825eed2a018b5cc118dcf26f19ed6f1 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 20:49:11 -0400 Subject: [PATCH 09/11] refactor(v2): keep test database setup explicit --- packages/opencode/src/storage/db.ts | 26 ++++++------- packages/opencode/src/v2/storage/database.ts | 38 +++++++++---------- .../opencode/test/v2/session-storage.test.ts | 23 ++++++++++- 3 files changed, 52 insertions(+), 35 deletions(-) diff --git a/packages/opencode/src/storage/db.ts b/packages/opencode/src/storage/db.ts index 6df731b530..06f1f84a9a 100644 --- a/packages/opencode/src/storage/db.ts +++ b/packages/opencode/src/storage/db.ts @@ -1,6 +1,5 @@ import { type SQLiteBunDatabase } from "drizzle-orm/bun-sqlite" import { migrate } from "drizzle-orm/bun-sqlite/migrator" -import type { MigrationsJournal } from "drizzle-orm/migrator" import { type SQLiteTransaction } from "drizzle-orm/sqlite-core" export * from "drizzle-orm" import { RuntimeFlags } from "@/effect/runtime-flags" @@ -48,10 +47,13 @@ export type Transaction = SQLiteTransaction<"sync", void> type Client = ReturnType -type Journal = MigrationsJournal +type Journal = { sql: string; timestamp: number; name: string }[] + +// Drizzle's migrate overloads trigger expensive variance checks here; narrow to the journal overload we actually use. +const migrateFromJournal = migrate as unknown as (db: SQLiteBunDatabase, entries: Journal) => void function applyMigrations(db: SQLiteBunDatabase, entries: Journal) { - migrate(db, entries) + migrateFromJournal(db, entries) } function time(tag: string) { @@ -72,17 +74,17 @@ function migrations(dir: string): Journal { .filter((entry) => entry.isDirectory()) .map((entry) => entry.name) - const sql: Journal = dirs + const sql = dirs .map((name) => { const file = path.join(dir, name, "migration.sql") - if (!existsSync(file)) return undefined + if (!existsSync(file)) return return { sql: readFileSync(file, "utf-8"), timestamp: time(name), name, } }) - .filter((entry) => entry !== undefined) + .filter(Boolean) as Journal return sql.sort((a, b) => a.timestamp - b.timestamp) } @@ -92,7 +94,7 @@ let loaded = false export const Client = Object.assign( (flags: DatabaseFlags = readRuntimeFlags()): Client => { - if (loaded && client) return client + if (loaded) return client as Client const dbPath = getPath(flags) log.info("opening database", { path: dbPath }) @@ -157,19 +159,19 @@ export function use(callback: (trx: TxOrDb) => T): T { if (err instanceof LocalContext.NotFound) { const effects: (() => void | Promise)[] = [] const result = ctx.provide({ effects, tx: Client() }, () => callback(Client())) - for (const effect of effects) void effect() + for (const effect of effects) effect() return result } throw err } } -export function effect(fn: () => void | Promise) { +export function effect(fn: () => any | Promise) { const bound = EffectBridge.bind(fn) try { ctx.use().effects.push(bound) } catch { - void bound() + bound() } } @@ -188,9 +190,7 @@ export function transaction( const effects: (() => void | Promise)[] = [] const txCallback = EffectBridge.bind((tx: TxOrDb) => ctx.provide({ tx, effects }, () => callback(tx))) const result = Client().transaction(txCallback, { behavior: options?.behavior }) - for (const effect of effects) void effect() - // Drizzle's transaction type does not preserve our NotPromise constraint through the callback wrapper. - // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + for (const effect of effects) effect() return result as NotPromise } throw err diff --git a/packages/opencode/src/v2/storage/database.ts b/packages/opencode/src/v2/storage/database.ts index 47c5a375a6..7ac5db016a 100644 --- a/packages/opencode/src/v2/storage/database.ts +++ b/packages/opencode/src/v2/storage/database.ts @@ -2,35 +2,31 @@ import { Database as LegacyDatabase } from "@/storage/db" import { SqliteClient } from "@effect/sql-sqlite-bun" import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" import { Context, Effect, Layer } from "effect" -import path from "path" const makeDatabase = EffectDrizzleSqlite.makeWithDefaults() type DatabaseShape = Effect.Success export class Service extends Context.Service()("@opencode/v2/storage/Database") {} +export const layerForPath = (filename: string) => + Layer.effect( + Service, + Effect.gen(function* () { + const db = yield* makeDatabase + yield* db.run("PRAGMA journal_mode = WAL") + yield* db.run("PRAGMA synchronous = NORMAL") + yield* db.run("PRAGMA busy_timeout = 5000") + yield* db.run("PRAGMA cache_size = -64000") + yield* db.run("PRAGMA foreign_keys = ON") + yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") + return db + }), + ).pipe(Layer.provide(SqliteClient.layer({ filename }))) + export const layer = Layer.unwrap( Effect.sync(() => { - const filename = LegacyDatabase.getPath() - return Layer.effect( - Service, - Effect.gen(function* () { - LegacyDatabase.Client() - const db = yield* makeDatabase - yield* db.run("PRAGMA journal_mode = WAL") - yield* db.run("PRAGMA synchronous = NORMAL") - yield* db.run("PRAGMA busy_timeout = 5000") - yield* db.run("PRAGMA cache_size = -64000") - yield* db.run("PRAGMA foreign_keys = ON") - yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") - if (filename === ":memory:") { - yield* EffectDrizzleSqlite.migrate(db, { - migrationsFolder: path.join(import.meta.dirname, "../../../migration"), - }) - } - return db - }), - ).pipe(Layer.provide(SqliteClient.layer({ filename, disableWAL: filename === ":memory:" }))) + LegacyDatabase.Client() + return layerForPath(LegacyDatabase.getPath()) }), ) diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 7083950a1b..55932e5f47 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -9,8 +9,12 @@ import { SessionStorageMemory } from "@/v2/storage/session-memory" import { SessionStorageSql } from "@/v2/storage/session-sql" import { EventV2 } from "@opencode-ai/core/event" import { SessionMessage } from "@opencode-ai/core/session-message" +import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" import { eq, or } from "@/storage/db" import { DateTime, Effect, Layer, Schema } from "effect" +import fs from "fs/promises" +import os from "os" +import path from "path" import { testEffect } from "../lib/effect" const projectID = ProjectID.make("project-session-storage") @@ -157,7 +161,24 @@ function sessionStorageContract(name: string, layer: Layer.Layer fs.mkdtemp(path.join(os.tmpdir(), "opencode-storage-test-"))), + (dir) => Effect.promise(() => fs.rm(dir, { recursive: true, force: true })), + ) + return Layer.effect( + StorageDatabase.Service, + Effect.gen(function* () { + const db = yield* StorageDatabase.Service + yield* EffectDrizzleSqlite.migrate(db, { migrationsFolder: path.join(import.meta.dirname, "../../migration") }) + return db + }), + ).pipe(Layer.provide(StorageDatabase.layerForPath(path.join(dir, "storage.db")))) + }), +) + +const sqlLayer = SessionStorageSql.layer.pipe(Layer.provideMerge(testDatabaseLayer)) const sqlSeeds: Seeds = { reset: resetSqlSeeds(), From 178840489f3c99b1a47f1aa94ca25585c8dd1e03 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 20:53:20 -0400 Subject: [PATCH 10/11] docs(v2): note storage database bootstrap cleanup --- packages/opencode/src/v2/storage/database.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/opencode/src/v2/storage/database.ts b/packages/opencode/src/v2/storage/database.ts index 7ac5db016a..73f68ad9f8 100644 --- a/packages/opencode/src/v2/storage/database.ts +++ b/packages/opencode/src/v2/storage/database.ts @@ -25,6 +25,8 @@ export const layerForPath = (filename: string) => export const layer = Layer.unwrap( Effect.sync(() => { + // TODO: Extract migration/bootstrap from the legacy Database.Client() so V2 storage + // can ensure the schema exists without opening the old global Drizzle connection. LegacyDatabase.Client() return layerForPath(LegacyDatabase.getPath()) }), From c633d10e744f1977c1776cc040eadf6772426950 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 20 May 2026 20:57:11 -0400 Subject: [PATCH 11/11] test(v2): simplify storage contract setup --- packages/opencode/test/v2/session-storage.test.ts | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/packages/opencode/test/v2/session-storage.test.ts b/packages/opencode/test/v2/session-storage.test.ts index 55932e5f47..10cd1fe2e3 100644 --- a/packages/opencode/test/v2/session-storage.test.ts +++ b/packages/opencode/test/v2/session-storage.test.ts @@ -12,9 +12,8 @@ import { SessionMessage } from "@opencode-ai/core/session-message" import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" import { eq, or } from "@/storage/db" import { DateTime, Effect, Layer, Schema } from "effect" -import fs from "fs/promises" -import os from "os" import path from "path" +import { tmpdir } from "../fixture/fixture" import { testEffect } from "../lib/effect" const projectID = ProjectID.make("project-session-storage") @@ -56,7 +55,7 @@ function sessionStorageContract(name: string, layer: Layer.Layer + it.effect(`${name}: gets and lists sessions with filters and cursors`, () => Effect.gen(function* () { yield* setup yield* seed.session({ @@ -117,7 +116,7 @@ function sessionStorageContract(name: string, layer: Layer.Layer + it.effect(`${name}: lists session messages with cursor direction`, () => Effect.gen(function* () { yield* setup yield* seed.session({ id: sessionA, title: "Alpha", path: "apps/api", updated: 1000 }) @@ -141,7 +140,7 @@ function sessionStorageContract(name: string, layer: Layer.Layer + it.effect(`${name}: returns context from the latest compaction boundary`, () => Effect.gen(function* () { yield* setup yield* seed.session({ id: sessionA, title: "Alpha", path: "apps/api", updated: 1000 }) @@ -164,8 +163,8 @@ function sessionStorageContract(name: string, layer: Layer.Layer fs.mkdtemp(path.join(os.tmpdir(), "opencode-storage-test-"))), - (dir) => Effect.promise(() => fs.rm(dir, { recursive: true, force: true })), + Effect.promise(() => tmpdir()), + (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()), ) return Layer.effect( StorageDatabase.Service, @@ -174,7 +173,7 @@ const testDatabaseLayer = Layer.unwrap( yield* EffectDrizzleSqlite.migrate(db, { migrationsFolder: path.join(import.meta.dirname, "../../migration") }) return db }), - ).pipe(Layer.provide(StorageDatabase.layerForPath(path.join(dir, "storage.db")))) + ).pipe(Layer.provide(StorageDatabase.layerForPath(path.join(dir.path, "storage.db")))) }), )