test(v2): simplify storage contract setup

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:*",

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:*",

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 "./storage/session"
+import { SessionStorageSql } from "./storage/session-sql"
 
 export const Delivery = Schema.Literals(["immediate", "deferred"]).annotate({
   identifier: "Session.Delivery",
@@ -73,40 +71,17 @@ export interface Interface {
     workspaceID?: WorkspaceID
   }) => Effect.Effect<Info>
   readonly get: (sessionID: SessionID) => Effect.Effect<Info, NotFoundError>
-  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<Info[], never>
-  readonly messages: (input: {
-    sessionID: SessionID
-    limit?: number
-    order?: "asc" | "desc"
-    cursor?: {
-      id: SessionMessage.ID
-      time: number
-      direction: "previous" | "next"
-    }
-  }) => Effect.Effect<SessionMessage.Message[], never>
-  readonly context: (sessionID: SessionID) => Effect.Effect<SessionMessage.Message[], never>
+  readonly list: (input: SessionStorage.SessionListInput) => Effect.Effect<Info[]>
+  readonly messages: (input: SessionStorage.MessageListInput) => Effect.Effect<SessionMessage.Message[]>
+  readonly context: (sessionID: SessionID) => Effect.Effect<SessionMessage.Message[]>
   readonly prompt: (input: {
     id?: EventV2.ID
     sessionID: SessionID
     prompt: Prompt
     delivery?: Delivery
-  }) => Effect.Effect<SessionMessage.User, never>
-  readonly shell: (input: { id?: EventV2.ID; sessionID: SessionID; command: string }) => Effect.Effect<void, never>
-  readonly skill: (input: { id?: EventV2.ID; sessionID: SessionID; skill: string }) => Effect.Effect<void, never>
+  }) => Effect.Effect<SessionMessage.User>
+  readonly shell: (input: { id?: EventV2.ID; sessionID: SessionID; command: string }) => Effect.Effect<void>
+  readonly skill: (input: { id?: EventV2.ID; sessionID: SessionID; skill: string }) => Effect.Effect<void>
   readonly subagent: (input: {
     id?: EventV2.ID
     parentID: SessionID
@@ -114,10 +89,10 @@ export interface Interface {
     agent: string
     model?: ModelV2.Ref
   }) => Effect.Effect<void, NotFoundError>
-  readonly switchAgent: (input: { sessionID: SessionID; agent: string }) => Effect.Effect<void, never>
-  readonly switchModel: (input: { sessionID: SessionID; model: ModelV2.Ref }) => Effect.Effect<void, never>
-  readonly compact: (sessionID: SessionID) => Effect.Effect<void, never>
-  readonly wait: (sessionID: SessionID) => Effect.Effect<void, never>
+  readonly switchAgent: (input: { sessionID: SessionID; agent: string }) => Effect.Effect<void>
+  readonly switchModel: (input: { sessionID: SessionID; model: ModelV2.Ref }) => Effect.Effect<void>
+  readonly compact: (sessionID: SessionID) => Effect.Effect<void>
+  readonly wait: (sessionID: SessionID) => Effect.Effect<void>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@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"

packages/opencode/src/v2/storage/database.ts

@@ -0,0 +1,37 @@
+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"
+
+const makeDatabase = EffectDrizzleSqlite.makeWithDefaults()
+type DatabaseShape = Effect.Success<typeof makeDatabase>
+
+export class Service extends Context.Service<Service, DatabaseShape>()("@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(() => {
+    // 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())
+  }),
+)
+
+export const defaultLayer = layer
+
+export * as StorageDatabase from "./database"

packages/opencode/src/v2/storage/session-memory.ts

@@ -0,0 +1,101 @@
+import { DateTime, Effect, Layer } from "effect"
+import { SessionMessage } from "@opencode-ai/core/session-message"
+import { SessionStorage } from "./session"
+
+export const make = (state: {
+  readonly sessions: Map<string, SessionStorage.SessionRow>
+  readonly messages: Map<string, SessionMessage.Message[]>
+}) =>
+  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)
+      }),
+  })
+
+const storageLayer = Layer.sync(SessionStorage.Service, () => {
+  return make({
+    sessions: new Map(),
+    messages: new Map(),
+  })
+})
+
+export const layer = storageLayer
+
+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 "./session-memory"

packages/opencode/src/v2/storage/session-sql.ts

@@ -0,0 +1,164 @@
+import { SessionMessageTable, SessionTable } from "@/session/session.sql"
+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 { 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 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* StorageDatabase.Service
+
+    const get: SessionStorage.Interface["get"] = Effect.fn("SessionStorageSql.get")(function* (sessionID) {
+      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"
+      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 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* 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"
+      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* 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* 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* 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(StorageDatabase.defaultLayer.pipe(Layer.orDie)))
+
+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 "./session-sql"

packages/opencode/src/v2/storage/session.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>()("StorageError", {
+  message: Schema.String,
+  cause: Schema.Defect,
+}) {}
+
+export class SessionRow extends Schema.Class<SessionRow>("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<SessionRow | undefined, StorageError>
+  readonly list: (input: SessionListInput) => Effect.Effect<SessionRow[], StorageError>
+  readonly messages: (input: MessageListInput) => Effect.Effect<SessionMessage.Message[], StorageError>
+  readonly context: (sessionID: SessionID) => Effect.Effect<SessionMessage.Message[], StorageError>
+}
+
+export function pageOrder(order: SortOrder, direction: PageDirection) {
+  if (direction !== "previous") return order
+  return order === "asc" ? "desc" : "asc"
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/v2/session/Storage") {}
+
+export * as SessionStorage from "./session"

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/storage/session-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(),

packages/opencode/test/v2/session-storage.test.ts

@@ -0,0 +1,364 @@
+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 { 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"
+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 path from "path"
+import { tmpdir } from "../fixture/fixture"
+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 sessionD = SessionID.make("ses_storage_d")
+const encodeMessage = Schema.encodeSync(SessionMessage.Message)
+const memoryState = {
+  sessions: new Map(),
+  messages: new Map(),
+}
+const memoryLayer = Layer.sync(SessionStorage.Service, () => SessionStorageMemory.make(memoryState))
+
+interface Seeds<R> {
+  readonly reset: Effect.Effect<void, never, R>
+  readonly project: Effect.Effect<void, never, R>
+  readonly session: (input: {
+    id: SessionID
+    title: string
+    directory?: string
+    path: string
+    updated: number
+  }) => Effect.Effect<void, never, R>
+  readonly userMessage: (input: { id: SessionMessage.ID; text: string; time: number }) => Effect.Effect<void, never, R>
+  readonly compaction: (input: {
+    id: SessionMessage.ID
+    summary: string
+    time: number
+  }) => Effect.Effect<void, never, R>
+}
+
+function sessionStorageContract<R, E>(name: string, layer: Layer.Layer<SessionStorage.Service | R, E>, seed: Seeds<R>) {
+  const it = testEffect(layer)
+
+  const setup = Effect.gen(function* () {
+    yield* seed.reset
+    yield* Effect.addFinalizer(() => seed.reset)
+    yield* seed.project
+  })
+
+  it.effect(`${name}: 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",
+        path: "docs",
+        updated: 3000,
+      })
+      yield* seed.session({
+        id: sessionD,
+        title: "Delta",
+        directory: "/tmp/other-project",
+        path: "other",
+        updated: 4000,
+      })
+
+      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({ directory: "/tmp/project-session-storage", 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, 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({
+          directory: "/tmp/project-session-storage",
+          order: "asc",
+          cursor: { id: sessionC, time: 3000, direction: "previous" },
+        })).map((row) => row.id),
+      ).toEqual([sessionA, sessionB])
+    }),
+  )
+
+  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 })
+      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(`${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 })
+      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 testDatabaseLayer = Layer.unwrap(
+  Effect.gen(function* () {
+    const dir = yield* Effect.acquireRelease(
+      Effect.promise(() => tmpdir()),
+      (dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
+    )
+    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.path, "storage.db"))))
+  }),
+)
+
+const sqlLayer = SessionStorageSql.layer.pipe(Layer.provideMerge(testDatabaseLayer))
+
+const sqlSeeds: Seeds<StorageDatabase.Service> = {
+  reset: resetSqlSeeds(),
+  project: seedProject(),
+  session: seedSession,
+  userMessage: seedUserMessage,
+  compaction: seedCompaction,
+}
+
+sessionStorageContract("SessionStorageSql", sqlLayer, sqlSeeds)
+
+const memorySeeds: Seeds<never> = {
+  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", memoryLayer, memorySeeds)
+
+function seedProject() {
+  return Effect.gen(function* () {
+    const db = yield* StorageDatabase.Service
+    yield* db
+      .insert(ProjectTable)
+      .values({
+        id: projectID,
+        worktree: "/tmp/project-session-storage",
+        time_created: 1,
+        time_updated: 1,
+        sandboxes: [],
+      })
+      .onConflictDoNothing()
+      .run()
+      .pipe(Effect.orDie)
+  })
+}
+
+function resetSqlSeeds() {
+  return Effect.gen(function* () {
+    const db = yield* StorageDatabase.Service
+    yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionA)).run().pipe(Effect.orDie)
+    yield* db
+      .delete(SessionTable)
+      .where(
+        or(
+          eq(SessionTable.id, sessionA),
+          eq(SessionTable.id, sessionB),
+          eq(SessionTable.id, sessionC),
+          eq(SessionTable.id, sessionD),
+        ),
+      )
+      .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 }) {
+  return Effect.gen(function* () {
+    const db = yield* StorageDatabase.Service
+    yield* 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()
+      .pipe(Effect.orDie)
+  })
+}
+
+function seedUserMessage(input: { id: SessionMessage.ID; text: string; time: number }) {
+  const encoded = encodeMessage(makeUserMessage(input))
+  const { id: _, type: __, ...data } = encoded
+  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
+  return 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,
+) {
+  return Effect.gen(function* () {
+    const db = yield* StorageDatabase.Service
+    yield* db
+      .insert(SessionMessageTable)
+      .values({
+        id,
+        session_id: sessionA,
+        type,
+        time_created: time,
+        time_updated: time,
+        data,
+      })
+      .run()
+      .pipe(Effect.orDie)
+  })
+}
+
+function appendMemoryMessage(message: SessionMessage.Message) {
+  const current = memoryState.messages.get(sessionA) ?? []
+  current.push(message)
+  memoryState.messages.set(sessionA, current)
+}

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<SessionRow | undefined, StorageError>
+  readonly list: (input: SessionListInput) => Effect.Effect<SessionRow[], StorageError>
+  readonly messages: (input: SessionMessageListInput) => Effect.Effect<SessionMessageRow[], StorageError>
+  readonly context: (sessionID: SessionID) => Effect.Effect<SessionMessageRow[], StorageError>
+}
+```
+
+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<SessionRow | undefined, StorageError>
+  readonly list: (input: SessionListInput) => Effect.Effect<SessionRow[], StorageError>
+  readonly messages: (input: SessionMessageListInput) => Effect.Effect<SessionMessageRow[], StorageError>
+  readonly context: (sessionID: SessionID) => Effect.Effect<SessionMessageRow[], StorageError>
+}
+```
+
+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<void, StorageError>
+  readonly removeAggregate: (aggregateID: string) => Effect.Effect<void, StorageError>
+  readonly appendProjected: <A, E>(input: {
+    definition: SyncDefinition
+    aggregateID: string
+    data: unknown
+    project: Effect.Effect<A, E, Transaction>
+    publish: Effect.Effect<void>
+  }) => Effect.Effect<A, E | StorageError>
+  readonly fence: (aggregateIDs?: string[]) => Effect.Effect<Record<string, number>, StorageError>
+  readonly history: (input: { since?: Record<string, number> }) => Effect.Effect<StoredEvent[], StorageError>
+  readonly replay: (aggregateID: string) => Effect.Effect<StoredEvent[], StorageError>
+}
+```
+
+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<SessionRow | undefined, StorageError>
+  readonly list: (input: SessionListInput) => Effect.Effect<SessionRow[], StorageError>
+  readonly listGlobal: (
+    input: GlobalSessionListInput,
+  ) => Effect.Effect<Array<{ session: SessionRow; project?: ProjectSummary }>, StorageError>
+  readonly children: (parentID: SessionID) => Effect.Effect<SessionRow[], StorageError>
+  readonly insert: (row: SessionInsert) => Effect.Effect<void, StorageError>
+  readonly patch: (id: SessionID, patch: SessionPatch) => Effect.Effect<SessionRow | undefined, StorageError>
+  readonly delete: (id: SessionID) => Effect.Effect<void, StorageError>
+  readonly incrementUsage: (id: SessionID, usage: UsageDelta) => Effect.Effect<void, StorageError>
+}
+```
+
+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<Array<{ message: MessageRow; parts: PartRow[] }>, StorageError>
+  readonly get: (input: {
+    sessionID: SessionID
+    messageID: MessageID
+  }) => Effect.Effect<MessageRow | undefined, StorageError>
+  readonly parts: (messageID: MessageID) => Effect.Effect<PartRow[], StorageError>
+  readonly getPart: (input: {
+    sessionID: SessionID
+    messageID: MessageID
+    partID: PartID
+  }) => Effect.Effect<PartRow | undefined, StorageError>
+  readonly upsertMessage: (row: MessageInsert) => Effect.Effect<void, StorageError | ForeignKeyError>
+  readonly upsertPart: (row: PartInsert) => Effect.Effect<{ previous?: PartRow }, StorageError | ForeignKeyError>
+  readonly deleteMessage: (input: {
+    sessionID: SessionID
+    messageID: MessageID
+  }) => Effect.Effect<PartRow[], StorageError>
+  readonly deletePart: (input: {
+    sessionID: SessionID
+    partID: PartID
+  }) => Effect.Effect<PartRow | undefined, StorageError>
+}
+```
+
+The delete/update methods expose previous rows where callers need usage compensation.
+
+### `SessionMessageStore`
+
+```ts
+interface SessionMessageStore {
+  readonly list: (input: SessionMessageListInput) => Effect.Effect<SessionMessageRow[], StorageError>
+  readonly context: (sessionID: SessionID) => Effect.Effect<SessionMessageRow[], StorageError>
+  readonly currentByType: (input: {
+    sessionID: SessionID
+    type: SessionMessage.Type
+  }) => Effect.Effect<SessionMessageRow[], StorageError>
+  readonly append: (row: SessionMessageInsert) => Effect.Effect<void, StorageError>
+  readonly updateData: (input: {
+    id: SessionMessage.ID
+    sessionID: SessionID
+    type: SessionMessage.Type
+    data: SessionMessageData
+  }) => Effect.Effect<void, StorageError>
+}
+```
+
+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/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.
+- 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.