agaltsovav/opencode
1
0
Fork 0
Code Issues Pull Requests Actions 99 Packages Projects Releases Wiki Activity

refactor(opencode): remove legacy database wrapper

Browse Source
This commit is contained in:
Dax Raad
2026-05-25 19:26:05 -04:00
parent c3743b26e8
commit 876ce31fef
50 changed files with 502 additions and 1966 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/database/database.ts
+1 -4
View File
@@ -3,7 +3,6 @@ export * as Database from "./database"
import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
import { layer as sqliteLayer } from "#sqlite"
import { Context, Effect, Layer } from "effect"
import { Sqlite } from "./sqlite"
import { Global } from "../global"
import { Flag } from "../flag/flag"
import { isAbsolute, join } from "path"
@@ -15,7 +14,6 @@ type DatabaseShape = Effect.Success<typeof makeDatabase>
export interface Interface {
db: DatabaseShape
drizzle: Sqlite.DrizzleClient
}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/storage/Database") {}
@@ -23,7 +21,6 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/v2
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const drizzle = yield* Sqlite.Drizzle
const db = yield* makeDatabase
yield* db.run("PRAGMA journal_mode = WAL")
@@ -35,7 +32,7 @@ export const layer = Layer.effect(
yield* Effect.log("Applying database migrations")
yield* DatabaseMigration.apply(db)
return { db, drizzle }
return { db }
}).pipe(Effect.orDie),
)
packages/opencode/src/cli/cmd/db.ts
+26 -84
View File
@@ -1,17 +1,14 @@
import type { Argv } from "yargs"
import { spawn } from "child_process"
import { Database } from "@/storage/db"
import { drizzle } from "drizzle-orm/bun-sqlite"
import { Database as BunDatabase } from "bun:sqlite"
import { UI } from "../ui"
import { cmd } from "./cmd"
import { JsonMigration } from "@/storage/json-migration"
import { EOL } from "os"
import { errorMessage } from "../../util/error"
import { Database } from "@opencode-ai/core/database/database"
import { Effect } from "effect"
import { sql } from "drizzle-orm"
import { effectCmd } from "../effect-cmd"
const QueryCommand = cmd({
const QueryCommand = effectCmd({
command: "$0 [query]",
describe: "open an interactive sqlite3 shell or run a query",
instance: false,
builder: (yargs: Argv) => {
return yargs
.positional("query", {
@@ -25,96 +22,41 @@ const QueryCommand = cmd({
describe: "Output format",
})
},
handler: async (args: { query?: string; format: string }) => {
handler: Effect.fn("Cli.db.query")(function* (args: { query?: string; format: string }) {
const query = args.query as string | undefined
if (query) {
const db = new BunDatabase(Database.getPath(), { readonly: true })
try {
const result = db.query(query).all() as Record<string, unknown>[]
if (args.format === "json") {
console.log(JSON.stringify(result, null, 2))
} else if (result.length > 0) {
const keys = Object.keys(result[0])
console.log(keys.join("\t"))
for (const row of result) {
console.log(keys.map((k) => row[k]).join("\t"))
}
}
} catch (err) {
UI.error(errorMessage(err))
process.exit(1)
const { db } = yield* Database.Service
const result = yield* db.all<Record<string, unknown>>(sql.raw(query)).pipe(Effect.orDie)
if (args.format === "json") console.log(JSON.stringify(result, null, 2))
else if (result.length > 0) {
const keys = Object.keys(result[0])
console.log(keys.join("\t"))
for (const row of result) console.log(keys.map((key) => row[key]).join("\t"))
}
db.close()
return
}
const child = spawn("sqlite3", [Database.getPath()], {
const child = spawn("sqlite3", [Database.path()], {
stdio: "inherit",
})
await new Promise((resolve) => child.on("close", resolve))
},
yield* Effect.promise(() => new Promise((resolve) => child.on("close", resolve)))
}),
})
const PathCommand = cmd({
const PathCommand = effectCmd({
command: "path",
describe: "print the database path",
handler: () => {
console.log(Database.getPath())
},
instance: false,
handler: Effect.fn("Cli.db.path")(function* () {
console.log(Database.path())
}),
})
const MigrateCommand = cmd({
command: "migrate",
describe: "migrate JSON data to SQLite (merges with existing data)",
handler: async () => {
const sqlite = new BunDatabase(Database.getPath())
const tty = process.stderr.isTTY
const width = 36
const orange = "\x1b[38;5;214m"
const muted = "\x1b[0;2m"
const reset = "\x1b[0m"
let last = -1
if (tty) process.stderr.write("\x1b[?25l")
try {
const stats = await JsonMigration.run(drizzle({ client: sqlite }), {
progress: (event) => {
const percent = Math.floor((event.current / event.total) * 100)
if (percent === last) return
last = percent
if (tty) {
const fill = Math.round((percent / 100) * width)
const bar = `${"■".repeat(fill)}${"･".repeat(width - fill)}`
process.stderr.write(
`\r${orange}${bar} ${percent.toString().padStart(3)}%${reset} ${muted}${event.current}/${event.total}${reset} `,
)
} else {
process.stderr.write(`sqlite-migration:${percent}${EOL}`)
}
},
})
if (tty) process.stderr.write("\n")
if (tty) process.stderr.write("\x1b[?25h")
else process.stderr.write(`sqlite-migration:done${EOL}`)
UI.println(
`Migration complete: ${stats.projects} projects, ${stats.sessions} sessions, ${stats.messages} messages`,
)
if (stats.errors.length > 0) {
UI.println(`${stats.errors.length} errors occurred during migration`)
}
} catch (err) {
if (tty) process.stderr.write("\x1b[?25h")
UI.error(`Migration failed: ${errorMessage(err)}`)
process.exit(1)
} finally {
sqlite.close()
}
},
})
export const DbCommand = cmd({
export const DbCommand = effectCmd({
command: "db",
describe: "database tools",
instance: false,
builder: (yargs: Argv) => {
return yargs.command(QueryCommand).command(PathCommand).command(MigrateCommand).demandCommand()
return yargs.command(QueryCommand).command(PathCommand).demandCommand()
},
handler: () => {},
handler: Effect.fn("Cli.db")(function* () {}),
})
packages/opencode/src/cli/cmd/import.ts
+33 -35
View File
@@ -3,7 +3,7 @@ import { SessionLegacy } from "@opencode-ai/core/session/legacy"
import { Session } from "@/session/session"
import { MessageV2 } from "../../session/message-v2"
import { CliError, effectCmd } from "../effect-cmd"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { SessionTable, MessageTable, PartTable } from "@opencode-ai/core/session/sql"
import { InstanceRef } from "@/effect/instance-ref"
import { ShareNext } from "@/share/share-next"
@@ -99,6 +99,7 @@ export const ImportCommand = effectCmd({
const runImport = Effect.fn("Cli.import.body")(function* (file: string, ctx: InstanceContext) {
const share = yield* ShareNext.Service
const fs = yield* AppFileSystem.Service
const { db } = yield* Database.Service
let exportData: ExportData | undefined
@@ -176,48 +177,45 @@ const runImport = Effect.fn("Cli.import.body")(function* (file: string, ctx: Ins
path: path.relative(path.resolve(ctx.worktree), ctx.directory).replaceAll("\\", "/"),
}) as Session.Info
const row = Session.toRow(info)
Database.use((db) =>
db
.insert(SessionTable)
.values(row)
.onConflictDoUpdate({
target: SessionTable.id,
set: { project_id: row.project_id, directory: row.directory, path: row.path },
})
.run(),
)
yield* db
.insert(SessionTable)
.values(row)
.onConflictDoUpdate({
target: SessionTable.id,
set: { project_id: row.project_id, directory: row.directory, path: row.path },
})
.run()
.pipe(Effect.orDie)
for (const msg of exportData.messages) {
const msgInfo = decodeMessageInfo(msg.info) as SessionLegacy.Info
const { id, sessionID: _, ...msgData } = msgInfo
Database.use((db) =>
db
.insert(MessageTable)
.values({
id,
session_id: row.id,
time_created: msgInfo.time?.created ?? Date.now(),
data: msgData as never,
})
.onConflictDoNothing()
.run(),
)
yield* db
.insert(MessageTable)
.values({
id,
session_id: row.id,
time_created: msgInfo.time?.created ?? Date.now(),
data: msgData as never,
})
.onConflictDoNothing()
.run()
.pipe(Effect.orDie)
for (const part of msg.parts) {
const partInfo = decodePart(part) as SessionLegacy.Part
const { id: partId, sessionID: _s, messageID, ...partData } = partInfo
Database.use((db) =>
db
.insert(PartTable)
.values({
id: partId,
message_id: messageID,
session_id: row.id,
data: partData,
})
.onConflictDoNothing()
.run(),
)
yield* db
.insert(PartTable)
.values({
id: partId,
message_id: messageID,
session_id: row.id,
data: partData,
})
.onConflictDoNothing()
.run()
.pipe(Effect.orDie)
}
}
packages/opencode/src/cli/cmd/stats.ts
+6 -5
View File
@@ -2,7 +2,7 @@ import { Effect } from "effect"
import { effectCmd } from "../effect-cmd"
import { Session } from "@/session/session"
import { NotFoundError } from "@/storage/storage"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { Project } from "@/project/project"
import { InstanceRef } from "@/effect/instance-ref"
@@ -80,9 +80,10 @@ export const StatsCommand = effectCmd({
}),
})
const getAllSessions = Effect.sync(() =>
Database.use((db) => db.select().from(SessionTable).all()).map((row) => Session.fromRow(row)),
)
const getAllSessions = Effect.fnUntraced(function* () {
const { db } = yield* Database.Service
return (yield* db.select().from(SessionTable).all().pipe(Effect.orDie)).map((row) => Session.fromRow(row))
})
const aggregateSessionStats = Effect.fn("Cli.stats.aggregate")(function* (
days?: number,
@@ -90,7 +91,7 @@ const aggregateSessionStats = Effect.fn("Cli.stats.aggregate")(function* (
currentProject?: Project.Info,
) {
const svc = yield* Session.Service
const sessions = yield* getAllSessions
const sessions = yield* getAllSessions()
const MS_IN_DAY = 24 * 60 * 60 * 1000
const cutoffTime = (() => {
packages/opencode/src/cli/cmd/tui/context/editor-zed.ts
+31 -32
View File
@@ -1,7 +1,10 @@
import { Database } from "bun:sqlite"
import os from "node:os"
import path from "node:path"
import { Option, Schema } from "effect"
import { Effect, Option, Schema } from "effect"
import { sql } from "drizzle-orm"
import { Sqlite } from "@opencode-ai/core/database/sqlite"
import { layer as sqliteLayer } from "@opencode-ai/core/database/sqlite.bun"
import { makeRuntime } from "@opencode-ai/core/effect/runtime"
import { Filesystem } from "@/util/filesystem"
import type { EditorSelection } from "./editor"
@@ -88,12 +91,10 @@ export async function resolveZedSelection(dbPath: string, cwd = process.cwd()):
}
function queryZedActiveEditor(dbPath: string, cwd: string) {
let db: Database | undefined
try {
db = new Database(dbPath, { readonly: true })
const raw = db
.query(
`select
const raw = zedDatabase(dbPath).runSync((db) =>
Effect.succeed(
db.all<ZedEditorRow>(sql.raw(`select
i.kind as item_kind,
e.item_id as editor_id,
i.workspace_id as workspace_id,
@@ -105,9 +106,9 @@ function queryZedActiveEditor(dbPath: string, cwd: string) {
join workspaces w on w.workspace_id = i.workspace_id
left join editors e on e.item_id = i.item_id and e.workspace_id = i.workspace_id
where i.active = 1 and p.active = 1
order by w.timestamp desc`,
)
.all()
order by w.timestamp desc`)),
),
)
const rows = raw.flatMap((row) => {
const parsed = decodeZedEditorRow(row)
@@ -126,24 +127,22 @@ function queryZedActiveEditor(dbPath: string, cwd: string) {
return { type: "row" as const, row }
} catch {
return { type: "unavailable" as const }
} finally {
db?.close()
}
}
function queryZedEditorSelections(dbPath: string, row: ZedActiveEditorRow) {
let db: Database | undefined
try {
db = new Database(dbPath, { readonly: true })
const raw = db
.query(
`select
const raw = zedDatabase(dbPath).runSync((db) =>
Effect.succeed(
db.all<Schema.Schema.Type<typeof ZedSelectionRowSchema>>(
sql`select
start as selection_start,
end as selection_end
from editor_selections
where editor_id = $editorID and workspace_id = $workspaceID`,
)
.all({ $editorID: row.editor_id, $workspaceID: row.workspace_id })
where editor_id = ${row.editor_id} and workspace_id = ${row.workspace_id}`,
),
),
)
const selections = raw.flatMap((selection) => {
const parsed = decodeZedSelectionRow(selection)
@@ -154,33 +153,33 @@ function queryZedEditorSelections(dbPath: string, row: ZedActiveEditorRow) {
return { type: "selections" as const, selections }
} catch {
return { type: "unavailable" as const }
} finally {
db?.close()
}
}
function queryZedEditorContents(dbPath: string, row: ZedActiveEditorRow) {
let db: Database | undefined
try {
db = new Database(dbPath, { readonly: true })
const parsed = decodeZedEditorContents(
db
.query(
`select contents
zedDatabase(dbPath).runSync((db) =>
Effect.succeed(
db.get(
sql`select contents
from editors
where item_id = $editorID and workspace_id = $workspaceID`,
)
.get({ $editorID: row.editor_id, $workspaceID: row.workspace_id }),
where item_id = ${row.editor_id} and workspace_id = ${row.workspace_id}`,
),
),
),
)
if (Option.isNone(parsed)) return { type: "unavailable" as const }
return { type: "contents" as const, contents: parsed.value.contents }
} catch {
return { type: "unavailable" as const }
} finally {
db?.close()
}
}
function zedDatabase(dbPath: string) {
return makeRuntime(Sqlite.Drizzle, sqliteLayer({ filename: dbPath, readonly: true, create: false }))
}
function isZedActiveEditorRow(row: ZedEditorRow): row is ZedActiveEditorRow {
return row.item_kind === "Editor" && row.editor_id != null
}
packages/opencode/src/control-plane/workspace.ts
+12 -11
View File
@@ -9,7 +9,8 @@ import { Project } from "@/project/project"
import { BusEvent } from "@/bus/bus-event"
import { GlobalBus } from "@/bus/global"
import { Auth } from "@/auth"
import { SyncEvent } from "@/sync"
import { EventV2 } from "@opencode-ai/core/event"
import { EventV2Bridge } from "@/event-v2-bridge"
import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
import { AppFileSystem } from "@opencode-ai/core/filesystem"
import * as Log from "@opencode-ai/core/util/log"
@@ -174,7 +175,7 @@ export const layer = Layer.effect(
const session = yield* Session.Service
const prompt = yield* SessionPrompt.Service
const http = yield* HttpClient.HttpClient
const sync = yield* SyncEvent.Service
const events = yield* EventV2Bridge.Service
const vcs = yield* Vcs.Service
const flags = yield* RuntimeFlags.Service
const fs = yield* AppFileSystem.Service
@@ -372,20 +373,20 @@ export const layer = Layer.effect(
})
}
const events = (yield* response.json) as HistoryEvent[]
const history = (yield* response.json) as HistoryEvent[]
log.info("workspace history synced", {
workspaceID: space.id,
events: events.length,
events: history.length,
})
yield* Effect.forEach(
events,
history,
(event) =>
sync
events
.replay(
{
id: event.id,
id: EventV2.ID.make(event.id),
aggregateID: event.aggregate_id,
seq: event.seq,
type: event.type,
@@ -432,11 +433,11 @@ export const layer = Layer.effect(
yield* parseSSE(stream, (evt) =>
Effect.gen(function* () {
if (!evt || typeof evt !== "object" || !("payload" in evt)) return
const payload = evt.payload as { type?: string; syncEvent?: SyncEvent.SerializedEvent }
const payload = evt.payload as { type?: string; syncEvent?: EventV2.SerializedEvent }
if (payload.type === "server.heartbeat") return
if (payload.type === "sync" && payload.syncEvent) {
const failed = yield* sync.replay(payload.syncEvent).pipe(
const failed = yield* events.replay(payload.syncEvent, { publish: true }).pipe(
Effect.as(false),
Effect.catchCause((error) =>
Effect.sync(() => {
@@ -634,7 +635,7 @@ export const layer = Layer.effect(
// "claim" this session so any future events coming from
// the old workspace are ignored
yield* sync.claim(input.sessionID, input.workspaceID ?? previous.projectID)
yield* events.claim(input.sessionID, input.workspaceID ?? previous.projectID)
}
}
@@ -1002,12 +1003,12 @@ export const layer = Layer.effect(
export const defaultLayer = layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(SessionPrompt.defaultLayer),
Layer.provide(Project.defaultLayer),
Layer.provide(Vcs.defaultLayer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2Bridge.defaultLayer),
Layer.provide(FetchHttpClient.layer),
Layer.provide(RuntimeFlags.defaultLayer),
)
packages/opencode/src/data-migration.ts
-161
View File
@@ -1,161 +0,0 @@
import { Context, Effect, Layer } from "effect"
import { Database } from "./storage/db"
import { DataMigrationTable } from "@opencode-ai/core/data-migration.sql"
import * as Log from "@opencode-ai/core/util/log"
import { and, asc, eq, gt, inArray, sql } from "drizzle-orm"
import { MessageTable, SessionTable } from "@opencode-ai/core/session/sql"
import type { SessionID } from "./session/schema"
export type Migration<R = never> = {
name: string
run: Effect.Effect<void, unknown, R>
}
const log = Log.create({ service: "data-migration" })
export interface Interface {}
export class Service extends Context.Service<Service, Interface>()("@opencode/DataMigration") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const migrations: Migration[] = [
{
name: "session_usage_from_messages",
run: Effect.gen(function* () {
type Usage = {
cost: number
tokens: { input: number; output: number; reasoning: number; cache: { read: number; write: number } }
}
for (let cursor: SessionID | undefined, page = 1; ; page++) {
const next = yield* Effect.gen(function* () {
const sessions = yield* Effect.sync(() =>
Database.use((db) =>
db
.select({ id: SessionTable.id })
.from(SessionTable)
.where(cursor ? gt(SessionTable.id, cursor) : undefined)
.orderBy(asc(SessionTable.id))
.limit(100)
.all(),
),
)
if (sessions.length === 0) return
yield* Effect.sync(() =>
Database.transaction((db) => {
const usageBySession = new Map<SessionID, Usage>(
sessions.map((session) => [
session.id,
{ cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } } },
]),
)
for (const row of db
.select({
session_id: MessageTable.session_id,
cost: sql<number>`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.cost'), 0)), 0)`,
tokens_input: sql<number>`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.input'), 0)), 0)`,
tokens_output: sql<number>`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.output'), 0)), 0)`,
tokens_reasoning: sql<number>`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.reasoning'), 0)), 0)`,
tokens_cache_read: sql<number>`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.cache.read'), 0)), 0)`,
tokens_cache_write: sql<number>`coalesce(sum(coalesce(json_extract(${MessageTable.data}, '$.tokens.cache.write'), 0)), 0)`,
})
.from(MessageTable)
.where(
and(
inArray(
MessageTable.session_id,
sessions.map((session) => session.id),
),
sql`json_extract(${MessageTable.data}, '$.role') = 'assistant'`,
),
)
.groupBy(MessageTable.session_id)
.all()) {
const current = usageBySession.get(row.session_id)
if (!current) continue
current.cost = row.cost
current.tokens.input = row.tokens_input
current.tokens.output = row.tokens_output
current.tokens.reasoning = row.tokens_reasoning
current.tokens.cache.read = row.tokens_cache_read
current.tokens.cache.write = row.tokens_cache_write
}
for (const [sessionID, value] of usageBySession) {
db.update(SessionTable)
.set({
cost: value.cost,
tokens_input: value.tokens.input,
tokens_output: value.tokens.output,
tokens_reasoning: value.tokens.reasoning,
tokens_cache_read: value.tokens.cache.read,
tokens_cache_write: value.tokens.cache.write,
time_updated: sql`${SessionTable.time_updated}`,
})
.where(eq(SessionTable.id, sessionID))
.run()
}
}),
)
return sessions.at(-1)?.id
}).pipe(
Effect.withSpan("DataMigration.sessionUsage.page", {
attributes: {
"data_migration.name": "session_usage_from_messages",
"data_migration.page": page,
"data_migration.cursor": cursor ?? "",
},
}),
)
if (!next) return
cursor = next
yield* Effect.sleep("10 millis")
}
}),
},
]
yield* Effect.gen(function* () {
if (migrations.length === 0) return
// Migrations run in a background fiber, so they must be resumable until
// their completion row is written.
for (const migration of migrations) {
const completed = Database.use((db) =>
db
.select({ name: DataMigrationTable.name })
.from(DataMigrationTable)
.where(eq(DataMigrationTable.name, migration.name))
.get(),
)
if (completed) continue
log.info("running data migration", { name: migration.name })
yield* migration.run.pipe(Effect.withSpan("DataMigration", { attributes: { name: migration.name } }))
Database.use((db) =>
db
.insert(DataMigrationTable)
.values({ name: migration.name, time_completed: Date.now() })
.onConflictDoNothing()
.run(),
)
}
}).pipe(
Effect.tapCause((cause) =>
Effect.logError("failed to run data migrations").pipe(Effect.annotateLogs("cause", cause)),
),
Effect.ignore,
Effect.forkScoped,
)
return Service.of({})
}),
)
export const defaultLayer = layer
export * as DataMigration from "./data-migration"
packages/opencode/src/effect/app-runtime.ts
+2 -6
View File
@@ -3,6 +3,7 @@ import { attach } from "./run-service"
import * as Observability from "@opencode-ai/core/effect/observability"
import { AppFileSystem } from "@opencode-ai/core/filesystem"
import { Database } from "@opencode-ai/core/database/database"
import { Bus } from "@/bus"
import { Auth } from "@/auth"
import { Account } from "@/account/account"
@@ -51,17 +52,15 @@ import { PtyTicket } from "@/pty/ticket"
import { Installation } from "@/installation"
import { ShareNext } from "@/share/share-next"
import { SessionShare } from "@/share/session"
import { SyncEvent } from "@/sync"
import { Npm } from "@opencode-ai/core/npm"
import { memoMap } from "@opencode-ai/core/effect/memo-map"
import { DataMigration } from "@/data-migration"
import { BackgroundJob } from "@/background/job"
import { EventV2Bridge } from "@/event-v2-bridge"
import { RuntimeFlags } from "@/effect/runtime-flags"
export const AppLayer = Layer.mergeAll(
Npm.defaultLayer,
AppFileSystem.defaultLayer,
Database.defaultLayer,
Bus.defaultLayer,
Auth.defaultLayer,
Account.defaultLayer,
@@ -111,9 +110,6 @@ export const AppLayer = Layer.mergeAll(
Installation.defaultLayer,
ShareNext.defaultLayer,
SessionShare.defaultLayer,
SyncEvent.defaultLayer,
EventV2Bridge.defaultLayer,
DataMigration.defaultLayer,
).pipe(Layer.provideMerge(InstanceLayer.layer), Layer.provideMerge(Observability.layer))
const rt = ManagedRuntime.make(AppLayer, { memoMap })
packages/opencode/src/event-v2-bridge.ts
+4 -4
View File
@@ -1,5 +1,5 @@
// Temporary V2 bridge: core events are the publish path, but the rest of
// opencode and the HTTP event stream still expect legacy bus/sync payloads.
// opencode and the HTTP event stream still expect legacy bus payloads.
// This layer goes away once consumers subscribe to core EventV2 directly.
import { Bus as ProjectBus } from "@/bus"
import { GlobalBus } from "@/bus/global"
@@ -31,8 +31,8 @@ export const layer = Layer.effect(
})
})
const provideEventLocation = <E, R>(event: EventV2.Payload, effect: Effect.Effect<void, E, R>) => {
return Effect.gen(function* () {
const provideEventLocation = <E, R>(event: EventV2.Payload, effect: Effect.Effect<void, E, R>) =>
Effect.gen(function* () {
const ctx = yield* InstanceRef
if (ctx) return yield* effect
const store = Option.getOrUndefined(yield* Effect.serviceOption(InstanceStore.Service))
@@ -45,7 +45,6 @@ export const layer = Layer.effect(
}),
)
})
}
yield* events.all().pipe(
Stream.runForEach((event) => {
@@ -58,6 +57,7 @@ export const layer = Layer.effect(
}),
Effect.forkScoped,
)
return Service.of(events)
}),
)
packages/opencode/src/index.ts
+5 -4
View File
@@ -30,10 +30,9 @@ import { WebCommand } from "./cli/cmd/web"
import { PrCommand } from "./cli/cmd/pr"
import { SessionCommand } from "./cli/cmd/session"
import { DbCommand } from "./cli/cmd/db"
import path from "path"
import { Global } from "@opencode-ai/core/global"
import { JsonMigration } from "@/storage/json-migration"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { errorMessage } from "./util/error"
import { PluginCommand } from "./cli/cmd/plug"
import { Heap } from "./cli/heap"
@@ -116,7 +115,7 @@ const cli = yargs(args)
run_id: processMetadata.runID,
})
const marker = Database.getPath()
const marker = Database.path()
if (!(await Filesystem.exists(marker))) {
const tty = process.stderr.isTTY
process.stderr.write("Performing one time database migration, may take a few minutes..." + EOL)
@@ -126,8 +125,9 @@ const cli = yargs(args)
const reset = "\x1b[0m"
let last = -1
if (tty) process.stderr.write("\x1b[?25l")
const sqlite = new (await import("bun:sqlite")).Database(marker)
try {
await JsonMigration.run(drizzle({ client: Database.Client().$client }), {
await JsonMigration.run(drizzle({ client: sqlite }), {
progress: (event) => {
const percent = Math.floor((event.current / event.total) * 100)
if (percent === last && event.current !== event.total) return
@@ -145,6 +145,7 @@ const cli = yargs(args)
},
})
} finally {
sqlite.close()
if (tty) process.stderr.write("\x1b[?25h")
else {
process.stderr.write(`sqlite-migration:done${EOL}`)
packages/opencode/src/node.ts
+1 -1
View File
@@ -2,5 +2,5 @@ export { Config } from "@/config/config"
export { Server } from "./server/server"
export { bootstrap } from "./cli/bootstrap"
export * as Log from "@opencode-ai/core/util/log"
export { Database } from "@/storage/db"
export { Database } from "@opencode-ai/core/database/database"
export { JsonMigration } from "@/storage/json-migration"
packages/opencode/src/permission/index.ts
+4 -5
View File
@@ -5,7 +5,7 @@ import { InstanceState } from "@/effect/instance-state"
import { ProjectV2 } from "@opencode-ai/core/project"
import { MessageID, SessionID } from "@/session/schema"
import { PermissionTable } from "@opencode-ai/core/session/sql"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { eq } from "drizzle-orm"
import * as Log from "@opencode-ai/core/util/log"
import { Wildcard } from "@opencode-ai/core/util/wildcard"
@@ -145,11 +145,10 @@ export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const { db } = yield* Database.Service
const state = yield* InstanceState.make<State>(
Effect.fn("Permission.state")(function* (ctx) {
const row = Database.use((db) =>
db.select().from(PermissionTable).where(eq(PermissionTable.project_id, ctx.project.id)).get(),
)
const row = yield* db.select().from(PermissionTable).where(eq(PermissionTable.project_id, ctx.project.id)).get().pipe(Effect.orDie)
const state = {
pending: new Map<PermissionID, PendingEntry>(),
approved: [...(row?.data ?? [])],
@@ -307,6 +306,6 @@ export function disabled(tools: string[], ruleset: Ruleset): Set<string> {
return PermissionV2.disabled(tools, ruleset)
}
export const defaultLayer = layer.pipe(Layer.provide(Bus.layer))
export const defaultLayer = layer.pipe(Layer.provide(Database.defaultLayer), Layer.provide(Bus.layer))
export * as Permission from "."
packages/opencode/src/server/routes/instance/httpapi/api.ts
+2 -4
View File
@@ -1,7 +1,6 @@
import { Schema } from "effect"
import { HttpApi } from "effect/unstable/httpapi"
import { BusEvent } from "@/bus/bus-event"
import { SyncEvent } from "@/sync"
import { ConfigApi } from "./groups/config"
import { ControlApi } from "./groups/control"
import { EventApi } from "./groups/event"
@@ -23,9 +22,8 @@ import { V2Api } from "./groups/v2"
import { Authorization } from "./middleware/authorization"
import { SchemaErrorMiddleware } from "./middleware/schema-error"
// SSE event schemas built from the BusEvent/SyncEvent registries.
// SSE event schemas built from the BusEvent and EventV2 registries.
const EventSchema = Schema.Union(BusEvent.effectPayloads()).annotate({ identifier: "Event" })
const SyncEventSchemas = SyncEvent.effectPayloads()
export const RootHttpApi = HttpApi.make("opencode-root")
.addHttpApi(ControlApi)
@@ -56,7 +54,7 @@ export const OpenCodeHttpApi = HttpApi.make("opencode")
.addHttpApi(EventApi)
.addHttpApi(InstanceHttpApi)
.addHttpApi(PtyConnectApi)
.annotate(HttpApi.AdditionalSchemas, [EventSchema, ...SyncEventSchemas])
.annotate(HttpApi.AdditionalSchemas, [EventSchema])
export type RootHttpApiType = typeof RootHttpApi
export type InstanceHttpApiType = typeof InstanceHttpApi
packages/opencode/src/server/routes/instance/httpapi/groups/global.ts
+1 -2
View File
@@ -1,6 +1,5 @@
import { Config } from "@/config/config"
import { BusEvent } from "@/bus/bus-event"
import { SyncEvent } from "@/sync"
import "@/server/event"
import { Schema } from "effect"
import { HttpApi, HttpApiEndpoint, HttpApiError, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
@@ -15,7 +14,7 @@ const GlobalEventSchema = Schema.Struct({
directory: Schema.String,
project: Schema.optional(Schema.String),
workspace: Schema.optional(Schema.String),
payload: Schema.Union([...BusEvent.effectPayloads(), ...SyncEvent.effectPayloads()]),
payload: Schema.Union(BusEvent.effectPayloads()),
}).annotate({ identifier: "GlobalEvent" })
export const GlobalUpgradeInput = Schema.Struct({
packages/opencode/src/server/routes/instance/httpapi/handlers/sync.ts
+26 -25
View File
@@ -1,8 +1,9 @@
import { Workspace } from "@/control-plane/workspace"
import * as InstanceState from "@/effect/instance-state"
import { Session } from "@/session/session"
import { Database } from "@/storage/db"
import { SyncEvent } from "@/sync"
import { Database } from "@opencode-ai/core/database/database"
import { EventV2 } from "@opencode-ai/core/event"
import { EventV2Bridge } from "@/event-v2-bridge"
import { EventTable } from "@opencode-ai/core/event/sql"
import { asc } from "drizzle-orm"
import { and } from "drizzle-orm"
@@ -23,7 +24,8 @@ export const syncHandlers = HttpApiBuilder.group(InstanceHttpApi, "sync", (handl
const workspace = yield* Workspace.Service
const session = yield* Session.Service
const scope = yield* Scope.Scope
const sync = yield* SyncEvent.Service
const events = yield* EventV2Bridge.Service
const { db } = yield* Database.Service
const start = Effect.fn("SyncHttpApi.start")(function* () {
yield* workspace
@@ -33,27 +35,27 @@ export const syncHandlers = HttpApiBuilder.group(InstanceHttpApi, "sync", (handl
})
const replay = Effect.fn("SyncHttpApi.replay")(function* (ctx: { payload: typeof ReplayPayload.Type }) {
const events: SyncEvent.SerializedEvent[] = ctx.payload.events.map((event) => ({
id: event.id,
const payload: EventV2.SerializedEvent[] = ctx.payload.events.map((event) => ({
id: EventV2.ID.make(event.id),
aggregateID: event.aggregateID,
seq: event.seq,
type: event.type,
data: { ...event.data },
}))
const source = events[0].aggregateID
const source = payload[0].aggregateID
log.info("sync replay requested", {
sessionID: source,
events: events.length,
first: events[0]?.seq,
last: events.at(-1)?.seq,
events: payload.length,
first: payload[0]?.seq,
last: payload.at(-1)?.seq,
directory: ctx.payload.directory,
})
yield* sync.replayAll(events)
yield* events.replayAll(payload)
log.info("sync replay complete", {
sessionID: source,
events: events.length,
first: events[0]?.seq,
last: events.at(-1)?.seq,
events: payload.length,
first: payload[0]?.seq,
last: payload.at(-1)?.seq,
})
return { sessionID: source }
})
@@ -74,18 +76,17 @@ export const syncHandlers = HttpApiBuilder.group(InstanceHttpApi, "sync", (handl
const history = Effect.fn("SyncHttpApi.history")(function* (ctx: { payload: typeof HistoryPayload.Type }) {
const exclude = Object.entries(ctx.payload)
return Database.use((db) =>
db
.select()
.from(EventTable)
.where(
exclude.length > 0
? not(or(...exclude.map(([id, seq]) => and(eq(EventTable.aggregate_id, id), lte(EventTable.seq, seq))))!)
: undefined,
)
.orderBy(asc(EventTable.seq))
.all(),
)
return yield* db
.select()
.from(EventTable)
.where(
exclude.length > 0
? not(or(...exclude.map(([id, seq]) => and(eq(EventTable.aggregate_id, id), lte(EventTable.seq, seq))))!)
: undefined,
)
.orderBy(asc(EventTable.seq))
.all()
.pipe(Effect.orDie)
})
return handlers.handle("start", start).handle("replay", replay).handle("steal", steal).handle("history", history)
packages/opencode/src/server/routes/instance/httpapi/server.ts
+2 -4
View File
@@ -49,7 +49,6 @@ import { EventV2Bridge } from "@/event-v2-bridge"
import { Database } from "@opencode-ai/core/database/database"
import { Skill } from "@/skill"
import { Snapshot } from "@/snapshot"
import { SyncEvent } from "@/sync"
import { ToolRegistry } from "@/tool/registry"
import { lazy } from "@/util/lazy"
import { Vcs } from "@/project/vcs"
@@ -192,12 +191,12 @@ export function createRoutes(
corsVaryFix,
fenceLayer,
cors(corsOptions),
Database.defaultLayer,
Account.defaultLayer,
Agent.defaultLayer,
Auth.defaultLayer,
Command.defaultLayer,
Config.defaultLayer,
Database.defaultLayer,
File.defaultLayer,
FileWatcher.defaultLayer,
Format.defaultLayer,
@@ -225,7 +224,6 @@ export function createRoutes(
SessionSummary.defaultLayer,
ShareNext.defaultLayer,
Snapshot.defaultLayer,
SyncEvent.defaultLayer,
EventV2Bridge.defaultLayer,
Skill.defaultLayer,
Todo.defaultLayer,
@@ -233,7 +231,7 @@ export function createRoutes(
Vcs.defaultLayer,
Workspace.defaultLayer,
Worktree.appLayer,
Bus.layer,
Bus.defaultLayer,
AppFileSystem.defaultLayer,
FetchHttpClient.layer,
HttpServer.layerServices,
packages/opencode/src/server/shared/fence.ts
+12 -9
View File
@@ -1,25 +1,28 @@
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { inArray } from "drizzle-orm"
import { EventSequenceTable } from "@opencode-ai/core/event/sql"
import { Workspace } from "@/control-plane/workspace"
import type { WorkspaceV2 } from "@opencode-ai/core/workspace"
import * as Log from "@opencode-ai/core/util/log"
import { Effect } from "effect"
import { makeRuntime } from "@/effect/run-service"
export const HEADER = "x-opencode-sync"
export type State = Record<string, number>
const log = Log.create({ service: "fence" })
const runtime = makeRuntime(Database.Service, Database.defaultLayer)
export function load(ids?: string[]) {
const rows = Database.use((db) => {
if (!ids?.length) {
return db.select().from(EventSequenceTable).all()
}
return runtime.runSync(({ db }) =>
Effect.gen(function* () {
const rows = yield* (ids?.length
? db.select().from(EventSequenceTable).where(inArray(EventSequenceTable.aggregate_id, ids)).all()
: db.select().from(EventSequenceTable).all()
).pipe(Effect.orDie)
return db.select().from(EventSequenceTable).where(inArray(EventSequenceTable.aggregate_id, ids)).all()
})
return Object.fromEntries(rows.map((row) => [row.aggregate_id, row.seq]))
return Object.fromEntries(rows.map((row) => [row.aggregate_id, row.seq]))
}),
)
}
export function diff(prev: State, next: State) {
packages/opencode/src/session/prompt.ts
+1 -1
View File
@@ -1652,7 +1652,6 @@ export const defaultLayer = Layer.suspend(() =>
Layer.provide(Image.defaultLayer),
Layer.provide(
Layer.mergeAll(
EventV2Bridge.defaultLayer,
Agent.defaultLayer,
Database.defaultLayer,
SystemPrompt.defaultLayer,
@@ -1661,6 +1660,7 @@ export const defaultLayer = Layer.suspend(() =>
Bus.layer,
CrossSpawnSpawner.defaultLayer,
RuntimeFlags.defaultLayer,
EventV2Bridge.defaultLayer,
),
),
),
packages/opencode/src/session/session.ts
+4 -4
View File
@@ -10,7 +10,7 @@ import type { ProviderMetadata, Usage } from "@opencode-ai/llm"
import { InstallationVersion } from "@opencode-ai/core/installation/version"
import { Database } from "@opencode-ai/core/database/database"
import { makeRuntime } from "@opencode-ai/core/effect/runtime"
import { EventV2 } from "@opencode-ai/core/event"
import { EventV2Bridge } from "@/event-v2-bridge"
import { SessionV2 } from "@opencode-ai/core/session"
import { NotFoundError } from "@/storage/storage"
@@ -510,7 +510,7 @@ export type Patch = Omit<Partial<Info>, "time" | "share" | "summary" | "revert"
export const layer: Layer.Layer<
Service,
never,
BackgroundJob.Service | Bus.Service | Storage.Service | RuntimeFlags.Service | Database.Service | EventV2.Service
BackgroundJob.Service | Bus.Service | Storage.Service | RuntimeFlags.Service | Database.Service | EventV2Bridge.Service
> = Layer.effect(
Service,
Effect.gen(function* () {
@@ -518,7 +518,7 @@ export const layer: Layer.Layer<
const database = yield* Database.Service
const background = yield* BackgroundJob.Service
const bus = yield* Bus.Service
const events = yield* EventV2.Service
const events = yield* EventV2Bridge.Service
const storage = yield* Storage.Service
const flags = yield* RuntimeFlags.Service
@@ -935,7 +935,7 @@ export const defaultLayer = layer.pipe(
Layer.provide(Bus.layer),
Layer.provide(Storage.defaultLayer),
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2.defaultLayer),
Layer.provide(EventV2Bridge.defaultLayer),
Layer.provide(SessionV2.defaultLayer),
Layer.provide(RuntimeFlags.defaultLayer),
)
packages/opencode/src/storage/db.bun.ts
-12
View File
@@ -1,12 +0,0 @@
import { Database } from "bun:sqlite"
import { Sqlite } from "@opencode-ai/core/database/sqlite"
import { layer } from "@opencode-ai/core/database/sqlite.bun"
import { makeRuntime } from "@opencode-ai/core/effect/runtime"
import { drizzle } from "drizzle-orm/bun-sqlite"
import { Effect } from "effect"
export function init(path: string) {
const runtime = makeRuntime(Sqlite.Native, layer({ filename: path }))
const native = runtime.runSync((native) => Effect.succeed(native)) as Database
return drizzle({ client: native })
}
packages/opencode/src/storage/db.node.ts
-12
View File
@@ -1,12 +0,0 @@
import { DatabaseSync } from "node:sqlite"
import { Sqlite } from "@opencode-ai/core/database/sqlite"
import { layer } from "@opencode-ai/core/database/sqlite.node"
import { makeRuntime } from "@opencode-ai/core/effect/runtime"
import { drizzle } from "drizzle-orm/node-sqlite"
import { Effect } from "effect"
export function init(path: string) {
const runtime = makeRuntime(Sqlite.Native, layer({ filename: path }))
const native = runtime.runSync((native) => Effect.succeed(native)) as DatabaseSync
return drizzle({ client: native })
}
packages/opencode/src/storage/db.ts
-115
View File
@@ -1,115 +0,0 @@
import { type SQLiteTransaction } from "drizzle-orm/sqlite-core"
import type { TablesRelationalConfig } from "drizzle-orm/relations"
export * from "drizzle-orm"
import { LocalContext } from "@/util/local-context"
import * as Log from "@opencode-ai/core/util/log"
import { NamedError } from "@opencode-ai/core/util/error"
import { EffectBridge } from "@/effect/bridge"
import { Effect, Schema } from "effect"
import { Database } from "@opencode-ai/core/database/database"
import { makeRuntime } from "@opencode-ai/core/effect/runtime"
export const NotFoundError = NamedError.create("NotFoundError", {
message: Schema.String,
})
const log = Log.create({ service: "db" })
const runtime = makeRuntime(Database.Service, Database.defaultLayer)
const database = await runtime.runPromise((db) => Effect.succeed(db))
export const getPath = () => Database.path()
export type Transaction = SQLiteTransaction<"sync", void, Record<string, unknown>, TablesRelationalConfig>
type Client = Database.Interface["drizzle"]
let client: Client | undefined
let loaded = false
export const Client = Object.assign(
(): Client => {
if (loaded) return client as Client
const dbPath = getPath()
log.info("opening database", { path: dbPath })
const db = database.drizzle
db.run("PRAGMA journal_mode = WAL")
db.run("PRAGMA synchronous = NORMAL")
db.run("PRAGMA busy_timeout = 5000")
db.run("PRAGMA cache_size = -64000")
db.run("PRAGMA foreign_keys = ON")
db.run("PRAGMA wal_checkpoint(PASSIVE)")
client = db
loaded = true
return db
},
{
reset: () => {
loaded = false
client = undefined
},
loaded: () => loaded,
},
)
export function close() {
if (!Client.loaded()) return
Client.reset()
}
export type TxOrDb = Transaction | Client
const ctx = LocalContext.create<{
tx: TxOrDb
effects: (() => void | Promise<void>)[]
}>("database")
export function use<T>(callback: (trx: TxOrDb) => T): T {
try {
return callback(ctx.use().tx)
} catch (err) {
if (err instanceof LocalContext.NotFound) {
const effects: (() => void | Promise<void>)[] = []
const result = ctx.provide({ effects, tx: Client() }, () => callback(Client()))
for (const effect of effects) effect()
return result
}
throw err
}
}
export function effect(fn: () => any | Promise<any>) {
const bound = EffectBridge.bind(fn)
try {
ctx.use().effects.push(bound)
} catch {
bound()
}
}
type NotPromise<T> = T extends Promise<any> ? never : T
export function transaction<T>(
callback: (tx: TxOrDb) => NotPromise<T>,
options?: {
behavior?: "deferred" | "immediate" | "exclusive"
},
): NotPromise<T> {
try {
return callback(ctx.use().tx)
} catch (err) {
if (err instanceof LocalContext.NotFound) {
const effects: (() => void | Promise<void>)[] = []
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()
return result as NotPromise<T>
}
throw err
}
}
export * as Database from "./db"
packages/opencode/src/sync/index.ts
-411
View File
@@ -1,411 +0,0 @@
// Legacy sync event system. It should stay unaware of core EventV2 execution;
// the only temporary V2 coupling here is exposing versioned core event schemas
// in effectPayloads() so existing HTTP/SDK schema generation remains stable.
// Remove that registry read when event schemas are generated from core directly.
import { Database } from "@/storage/db"
import { eq } from "drizzle-orm"
import { GlobalBus } from "@/bus/global"
import { Bus as ProjectBus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
import { EventID } from "./schema"
import { Context, Effect, Layer, Schema as EffectSchema } from "effect"
import type { DeepMutable } from "@opencode-ai/core/schema"
import { EventV2 } from "@opencode-ai/core/event"
import { serviceUse } from "@opencode-ai/core/effect/service-use"
import { InstanceState } from "@/effect/instance-state"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { EffectBridge } from "@/effect/bridge"
// Keep `Event["data"]` mutable because projectors mutate the persisted shape
// when writing to the database. Bus payloads (`Properties`) stay readonly —
// subscribers only read.
export type Definition<
Type extends string = string,
Schema extends EffectSchema.Top = EffectSchema.Top,
BusSchema extends EffectSchema.Top = Schema,
> = {
type: Type
version: number
aggregate: string
schema: Schema
// Bus event payload schema. Defaults to `schema` unless `busSchema` was
// passed at definition time (see `session.updated`, whose projector
// expands the persisted data to a `{ sessionID, info }` bus payload).
properties: BusSchema
}
export type Event<Def extends Definition = Definition> = {
id: string
seq: number
aggregateID: string
data: DeepMutable<EffectSchema.Schema.Type<Def["schema"]>>
}
export type Properties<Def extends Definition = Definition> = EffectSchema.Schema.Type<Def["properties"]>
export type SerializedEvent<Def extends Definition = Definition> = Event<Def> & { type: string }
type ProjectorFunc = (db: Database.TxOrDb, data: unknown, event: Event) => void
type ConvertEvent = (type: string, data: Event["data"]) => unknown | Promise<unknown>
export interface Interface {
readonly run: <Def extends Definition>(
def: Def,
data: Event<Def>["data"],
options?: { publish?: boolean },
) => Effect.Effect<void>
readonly replay: (event: SerializedEvent, options?: { publish: boolean; ownerID?: string }) => Effect.Effect<void>
readonly replayAll: (
events: SerializedEvent[],
options?: { publish: boolean; ownerID?: string },
) => Effect.Effect<string | undefined>
readonly remove: (aggregateID: string) => Effect.Effect<void>
readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SyncEvent") {}
export const layer = Layer.effect(Service)(
Effect.gen(function* () {
const flags = yield* RuntimeFlags.Service
const bus = yield* ProjectBus.Service
const replay: Interface["replay"] = Effect.fn("SyncEvent.replay")(function* (event, options) {
const def = registry.get(event.type)
if (!def) {
throw new Error(`Unknown event type: ${event.type}`)
}
const row = Database.use((db) =>
db
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
.from(EventSequenceTable)
.where(eq(EventSequenceTable.aggregate_id, event.aggregateID))
.get(),
)
const latest = row?.seq ?? -1
if (event.seq <= latest) return
if (row?.ownerID && row.ownerID !== options?.ownerID) {
return
}
const expected = latest + 1
if (event.seq !== expected) {
throw new Error(
`Sequence mismatch for aggregate "${event.aggregateID}": expected ${expected}, got ${event.seq}`,
)
}
const publish = !!options?.publish
// Bridge captures handler-fiber refs (InstanceRef/WorkspaceRef) and the
// full Effect context, so the forked publish + GlobalBus emit run with
// the right state without a per-call attachWith.
const bridge = yield* EffectBridge.make()
process(def, event, {
bus,
bridge,
publish,
ownerID: options?.ownerID,
experimentalWorkspaces: flags.experimentalWorkspaces,
})
})
const replayAll: Interface["replayAll"] = Effect.fn("SyncEvent.replayAll")(function* (events, options) {
const source = events[0]?.aggregateID
if (!source) return undefined
if (events.some((item) => item.aggregateID !== source)) {
throw new Error("Replay events must belong to the same session")
}
const start = events[0].seq
for (const [i, item] of events.entries()) {
const seq = start + i
if (item.seq !== seq) {
throw new Error(`Replay sequence mismatch at index ${i}: expected ${seq}, got ${item.seq}`)
}
}
for (const item of events) {
yield* replay(item, options)
}
return source
})
const run: Interface["run"] = Effect.fn("SyncEvent.run")(function* (def, data, options) {
const agg = (data as Record<string, string>)[def.aggregate]
// This should never happen: we've enforced it via typescript in
// the definition
if (agg == null) {
throw new Error(`SyncEvent.run: "${def.aggregate}" required but not found: ${JSON.stringify(data)}`)
}
if (def.version !== versions.get(def.type)) {
throw new Error(`SyncEvent.run: running old versions of events is not allowed: ${def.type}`)
}
const { publish = true } = options || {}
const bridge = yield* EffectBridge.make()
// Note that this is an "immediate" transaction which is critical.
// We need to make sure we can safely read and write with nothing
// else changing the data from under us
Database.transaction(
(tx) => {
const id = EventID.ascending()
const row = tx
.select({ seq: EventSequenceTable.seq })
.from(EventSequenceTable)
.where(eq(EventSequenceTable.aggregate_id, agg))
.get()
const seq = row?.seq != null ? row.seq + 1 : 0
const event = { id, seq, aggregateID: agg, data }
process(def, event, { bus, bridge, publish, experimentalWorkspaces: flags.experimentalWorkspaces })
},
{
behavior: "immediate",
},
)
})
const remove: Interface["remove"] = Effect.fn("SyncEvent.remove")(function* (aggregateID) {
Database.transaction((tx) => {
tx.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).run()
tx.delete(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).run()
})
})
const claim: Interface["claim"] = Effect.fn("SyncEvent.claim")((aggregateID, ownerID) =>
Effect.sync(() =>
Database.use((db) =>
db
.update(EventSequenceTable)
.set({ owner_id: ownerID })
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
.run(),
),
),
)
return Service.of({
run,
replay,
replayAll,
remove,
claim,
})
}),
)
export const defaultLayer = layer.pipe(Layer.provide([ProjectBus.defaultLayer, RuntimeFlags.defaultLayer]))
export const use = serviceUse(Service)
export const registry = new Map<string, Definition>()
let projectors: Map<string, ProjectorFunc> | undefined
const versions = new Map<string, number>()
let frozen = false
let convertEvent: ConvertEvent
export function reset() {
frozen = false
projectors = undefined
convertEvent = (_, data) => data
}
export function init(input: { projectors: Array<[Definition, ProjectorFunc]>; convertEvent?: ConvertEvent }) {
projectors = new Map(input.projectors.map(([def, func]) => [versionedType(def.type, def.version), func]))
for (let entry of EventV2.registry.values()) {
if (!entry.sync) continue
register({
type: entry.type,
version: entry.sync.version,
aggregate: entry.sync.aggregate,
properties: entry.data,
schema: entry.data,
})
}
// Install all the latest event defs to the bus. We only ever emit
// latest versions from code, and keep around old versions for
// replaying. Replaying does not go through the bus, and it
// simplifies the bus to only use unversioned latest events
for (let [type, version] of versions.entries()) {
let def = registry.get(versionedType(type, version))!
BusEvent.define(def.type, def.properties)
}
// Freeze the system so it clearly errors if events are defined
// after `init` which would cause bugs
frozen = true
convertEvent = input.convertEvent ?? ((_, data) => data)
}
export function versionedType<A extends string>(type: A): A
export function versionedType<A extends string, B extends number>(type: A, version: B): `${A}/${B}`
export function versionedType(type: string, version?: number) {
return version ? `${type}.${version}` : type
}
export function define<
Type extends string,
Agg extends string,
Schema extends EffectSchema.Top,
BusSchema extends EffectSchema.Top = Schema,
>(input: {
type: Type
version: number
aggregate: Agg
schema: Schema
busSchema?: BusSchema
}): Definition<Type, Schema, BusSchema> {
if (frozen) {
throw new Error("Error defining sync event: sync system has been frozen")
}
const def = {
type: input.type,
version: input.version,
aggregate: input.aggregate,
schema: input.schema,
properties: (input.busSchema ?? input.schema) as BusSchema,
}
register(def)
return def
}
export function project<Def extends Definition>(
def: Def,
func: (db: Database.TxOrDb, data: Event<Def>["data"], event: Event<Def>) => void,
): [Definition, ProjectorFunc] {
return [def, func as ProjectorFunc]
}
function register(def: Definition) {
versions.set(def.type, Math.max(def.version, versions.get(def.type) || 0))
registry.set(versionedType(def.type, def.version), def)
}
function process<Def extends Definition>(
def: Def,
event: Event<Def>,
options: {
bus: ProjectBus.Interface
bridge: EffectBridge.Shape
publish: boolean
ownerID?: string
experimentalWorkspaces: boolean
},
) {
if (projectors == null) {
throw new Error("No projectors available. Call `SyncEvent.init` to install projectors")
}
const projector = projectors.get(versionedType(def.type, def.version))
if (!projector) {
if (!def.type.includes("next")) throw new Error(`Projector not found for event: ${def.type}`)
return
}
Database.transaction((tx) => {
projector(tx, event.data, event)
if (options.experimentalWorkspaces) {
tx.insert(EventSequenceTable)
.values({
aggregate_id: event.aggregateID,
seq: event.seq,
owner_id: options?.ownerID,
})
.onConflictDoUpdate({
target: EventSequenceTable.aggregate_id,
set: { seq: event.seq },
})
.run()
tx.insert(EventTable)
.values({
id: event.id as never,
seq: event.seq,
aggregate_id: event.aggregateID,
type: versionedType(def.type, def.version),
data: event.data as never,
})
.run()
}
Database.effect(() => {
if (!options.publish) return
const result = convertEvent(def.type, event.data)
// The bridge was built inside the caller's fiber so it already carries
// InstanceRef/WorkspaceRef and the full Effect context. Both the bus
// publish and the GlobalBus emit run inside the forked Effect so they
// share the same instance/workspace lookup.
const publish = (data: unknown) =>
options.bridge.fork(
Effect.gen(function* () {
yield* options.bus.publish(def, data as Properties<Def>, { id: event.id })
const instance = yield* InstanceState.context
const workspace = yield* InstanceState.workspaceID
GlobalBus.emit("event", {
directory: instance.directory,
project: instance.project.id,
workspace,
payload: {
type: "sync",
syncEvent: {
type: versionedType(def.type, def.version),
...event,
},
},
})
}),
)
if (result instanceof Promise) {
void result.then(publish)
} else {
publish(result)
}
})
})
}
export function effectPayloads() {
return [
...registry
.entries()
.map(([type, def]) =>
EffectSchema.Struct({
type: EffectSchema.Literal("sync"),
name: EffectSchema.Literal(type),
id: EffectSchema.String,
seq: EffectSchema.Finite,
aggregateID: EffectSchema.Literal(def.aggregate),
data: def.schema,
}).annotate({ identifier: `SyncEvent.${type}` }),
)
.toArray(),
...EventV2.registry
.values()
.filter(
(definition) =>
definition.sync !== undefined && !registry.has(versionedType(definition.type, definition.sync.version)),
)
.map((definition) =>
EffectSchema.Struct({
type: EffectSchema.Literal("sync"),
name: EffectSchema.Literal(versionedType(definition.type, definition.sync!.version)),
id: EffectSchema.String,
seq: EffectSchema.Finite,
aggregateID: EffectSchema.Literal(definition.sync!.aggregate),
data: definition.data,
}).annotate({ identifier: `SyncEvent.${definition.type}` }),
)
.toArray(),
]
}
export * as SyncEvent from "."
packages/opencode/src/worktree/index.ts
+5 -5
View File
@@ -2,7 +2,7 @@ import { Global } from "@opencode-ai/core/global"
import { InstanceLayer } from "@/project/instance-layer"
import { InstanceStore } from "@/project/instance-store"
import { Project } from "@/project/project"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { eq } from "drizzle-orm"
import { ProjectTable } from "@opencode-ai/core/project/sql"
import type { ProjectV2 } from "@opencode-ai/core/project"
@@ -149,7 +149,7 @@ type GitResult = { code: number; text: string; stderr: string }
export const layer: Layer.Layer<
Service,
never,
AppFileSystem.Service | Path.Path | AppProcess.Service | Git.Service | Project.Service | InstanceStore.Service
AppFileSystem.Service | Path.Path | AppProcess.Service | Git.Service | Project.Service | InstanceStore.Service | Database.Service
> = Layer.effect(
Service,
Effect.gen(function* () {
@@ -157,6 +157,7 @@ export const layer: Layer.Layer<
const fs = yield* AppFileSystem.Service
const pathSvc = yield* Path.Path
const appProcess = yield* AppProcess.Service
const { db } = yield* Database.Service
const gitSvc = yield* Git.Service
const project = yield* Project.Service
const store = yield* InstanceStore.Service
@@ -478,9 +479,7 @@ export const layer: Layer.Layer<
directory: string,
input: { projectID: ProjectV2.ID; extra?: string },
) {
const row = yield* Effect.sync(() =>
Database.use((db) => db.select().from(ProjectTable).where(eq(ProjectTable.id, input.projectID)).get()),
)
const row = yield* db.select().from(ProjectTable).where(eq(ProjectTable.id, input.projectID)).get().pipe(Effect.orDie)
const project = row ? Project.fromRow(row) : undefined
const startup = project?.commands?.start?.trim() ?? ""
const ok = yield* runStartScript(directory, startup, "project")
@@ -611,6 +610,7 @@ export const appLayer = layer.pipe(
Layer.provide(Git.defaultLayer),
Layer.provide(AppProcess.defaultLayer),
Layer.provide(Project.defaultLayer),
Layer.provide(Database.defaultLayer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(NodePath.layer),
)
packages/opencode/test/account/repo.test.ts
+7 -6
View File
@@ -1,18 +1,19 @@
import { expect } from "bun:test"
import { Effect, Layer, Option } from "effect"
import { sql } from "drizzle-orm"
import { AccountRepo } from "../../src/account/repo"
import { AccessToken, AccountID, OrgID, RefreshToken } from "../../src/account/schema"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { testEffect } from "../lib/effect"
const truncate = Layer.effectDiscard(
Effect.sync(() => {
const db = Database.Client()
db.run(/*sql*/ `DELETE FROM account_state`)
db.run(/*sql*/ `DELETE FROM account`)
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db.run(sql`DELETE FROM account_state`)
yield* db.run(sql`DELETE FROM account`)
}),
)
).pipe(Layer.provide(Database.defaultLayer))
const it = testEffect(Layer.merge(AccountRepo.defaultLayer, truncate))
packages/opencode/test/account/service.test.ts
+7 -6
View File
@@ -1,5 +1,6 @@
import { expect } from "bun:test"
import { Duration, Effect, Layer, Option, Schema } from "effect"
import { sql } from "drizzle-orm"
import { HttpClient, HttpClientError, HttpClientResponse } from "effect/unstable/http"
import { AccountRepo } from "../../src/account/repo"
@@ -15,16 +16,16 @@ import {
RefreshToken,
UserCode,
} from "../../src/account/schema"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { testEffect } from "../lib/effect"
const truncate = Layer.effectDiscard(
Effect.sync(() => {
const db = Database.Client()
db.run(/*sql*/ `DELETE FROM account_state`)
db.run(/*sql*/ `DELETE FROM account`)
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db.run(sql`DELETE FROM account_state`)
yield* db.run(sql`DELETE FROM account`)
}),
)
).pipe(Layer.provide(Database.defaultLayer))
const it = testEffect(Layer.merge(AccountRepo.defaultLayer, truncate))
packages/opencode/test/control-plane/workspace.test.ts
+112 -100
View File
@@ -10,14 +10,12 @@ import { eq } from "drizzle-orm"
import { AppFileSystem } from "@opencode-ai/core/filesystem"
import * as Log from "@opencode-ai/core/util/log"
import { GlobalBus, type GlobalEvent } from "@/bus/global"
import { Database } from "@/storage/db"
import { Database as CoreDatabase } from "@opencode-ai/core/database/database"
import { Database } from "@opencode-ai/core/database/database"
import { ProjectV2 } from "@opencode-ai/core/project"
import { ProjectTable } from "@opencode-ai/core/project/sql"
import { Session as SessionNs } from "@/session/session"
import { SessionID } from "@/session/schema"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { SyncEvent } from "@/sync"
import { EventSequenceTable } from "@opencode-ai/core/event/sql"
import { resetDatabase } from "../fixture/db"
import { disposeAllInstances, provideTmpdirInstance, requireInstance, TestInstance } from "../fixture/fixture"
@@ -34,6 +32,7 @@ import { SessionPrompt } from "@/session/prompt"
import { Project } from "@/project/project"
import { Vcs } from "@/project/vcs"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { EventV2Bridge } from "@/event-v2-bridge"
void Log.init({ print: false })
@@ -49,11 +48,11 @@ const workspaceLayer = (experimentalWorkspaces: boolean) =>
Workspace.layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(SessionNs.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(SessionPrompt.defaultLayer),
Layer.provide(Project.defaultLayer),
Layer.provide(Vcs.defaultLayer),
Layer.provide(CoreDatabase.defaultLayer),
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2Bridge.defaultLayer),
Layer.provide(FetchHttpClient.layer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces })),
@@ -64,6 +63,7 @@ const testServerLayer = Layer.mergeAll(
NodeHttpServer.layer(Http.createServer, { host: "127.0.0.1", port: 0 }),
workspaceLayer(true),
SessionNs.defaultLayer,
Database.defaultLayer,
)
const it = testEffect(testServerLayer)
@@ -107,7 +107,6 @@ function restoreEnv() {
}
beforeEach(() => {
Database.close()
restoreEnv()
process.env.OPENCODE_EXPERIMENTAL_WORKSPACES = "true"
})
@@ -281,7 +280,7 @@ function workspaceInfo(projectID: ProjectV2.ID, type: string, input?: Partial<Wo
}
function insertWorkspace(info: Workspace.Info) {
Database.use((db) =>
return Database.Service.use(({ db }) =>
db
.insert(WorkspaceTable)
.values({
@@ -294,12 +293,13 @@ function insertWorkspace(info: Workspace.Info) {
project_id: info.projectID,
time_used: info.timeUsed,
})
.run(),
.run()
.pipe(Effect.orDie),
)
}
function insertProject(id: ProjectV2.ID, worktree: string) {
Database.use((db) =>
return Database.Service.use(({ db }) =>
db
.insert(ProjectTable)
.values({
@@ -311,34 +311,37 @@ function insertProject(id: ProjectV2.ID, worktree: string) {
time_updated: Date.now(),
sandboxes: [],
})
.run(),
.run()
.pipe(Effect.orDie),
)
}
function attachSessionToWorkspace(sessionID: SessionID, workspaceID: WorkspaceV2.ID) {
Database.use((db) =>
db.update(SessionTable).set({ workspace_id: workspaceID }).where(eq(SessionTable.id, sessionID)).run(),
return Database.Service.use(({ db }) =>
db.update(SessionTable).set({ workspace_id: workspaceID }).where(eq(SessionTable.id, sessionID)).run().pipe(Effect.orDie),
)
}
function sessionSequence(sessionID: SessionID) {
return Database.use((db) =>
return Database.Service.use(({ db }) =>
db
.select({ seq: EventSequenceTable.seq })
.from(EventSequenceTable)
.where(eq(EventSequenceTable.aggregate_id, sessionID))
.get(),
)?.seq
.get()
.pipe(Effect.orDie, Effect.map((row) => row?.seq)),
)
}
function sessionSequenceOwner(sessionID: SessionID) {
return Database.use((db) =>
return Database.Service.use(({ db }) =>
db
.select({ ownerID: EventSequenceTable.owner_id })
.from(EventSequenceTable)
.where(eq(EventSequenceTable.aggregate_id, sessionID))
.get(),
)?.ownerID
.get()
.pipe(Effect.orDie, Effect.map((row) => row?.ownerID)),
)
}
describe("workspace schemas and exports", () => {
@@ -382,7 +385,7 @@ describe("workspace CRUD", () => {
const instance = yield* requireInstance
const workspace = yield* Workspace.Service
const otherProjectID = ProjectV2.ID.make("project-other")
insertProject(otherProjectID, "/tmp/other")
yield* insertProject(otherProjectID, "/tmp/other")
const a = workspaceInfo(instance.project.id, "manual", {
id: WorkspaceV2.ID.ascending("wrk_a_list"),
branch: "a",
@@ -396,9 +399,9 @@ describe("workspace CRUD", () => {
extra: ["b"],
})
const other = workspaceInfo(otherProjectID, "manual", { id: WorkspaceV2.ID.ascending("wrk_c_list") })
insertWorkspace(b)
insertWorkspace(other)
insertWorkspace(a)
yield* insertWorkspace(b)
yield* insertWorkspace(other)
yield* insertWorkspace(a)
expect(yield* workspace.list(instance.project)).toEqual([a, b])
}),
@@ -580,7 +583,7 @@ describe("workspace CRUD", () => {
name: "existing",
directory: path.join(instance.directory, "existing"),
})
insertWorkspace(existing)
yield* insertWorkspace(existing)
const discovered = {
type,
@@ -765,8 +768,8 @@ describe("workspace CRUD", () => {
const info = yield* workspace.create({ type, branch: null, projectID: instance.project.id, extra: null })
const one = yield* sessionSvc.create({})
const two = yield* sessionSvc.create({})
attachSessionToWorkspace(one.id, info.id)
attachSessionToWorkspace(two.id, info.id)
yield* attachSessionToWorkspace(one.id, info.id)
yield* attachSessionToWorkspace(two.id, info.id)
const removed = yield* workspace.remove(info.id)
@@ -774,10 +777,14 @@ describe("workspace CRUD", () => {
expect(yield* workspace.get(info.id)).toBeUndefined()
expect(recorded.calls.remove).toEqual([info])
expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBeUndefined()
const { db } = yield* Database.Service
expect(
Database.use((db) =>
db.select({ id: SessionTable.id }).from(SessionTable).where(eq(SessionTable.workspace_id, info.id)).all(),
),
yield* db
.select({ id: SessionTable.id })
.from(SessionTable)
.where(eq(SessionTable.workspace_id, info.id))
.all()
.pipe(Effect.orDie),
).toEqual([])
})
},
@@ -804,7 +811,7 @@ describe("workspace CRUD", () => {
},
}).adapter,
)
insertWorkspace(info)
yield* insertWorkspace(info)
expect(yield* workspace.remove(info.id)).toEqual(info)
expect(yield* workspace.get(info.id)).toBeUndefined()
@@ -824,25 +831,27 @@ describe("workspace CRUD", () => {
const targetType = unique("warp-target-local")
const previous = workspaceInfo(instance.project.id, previousType)
const target = workspaceInfo(instance.project.id, targetType)
insertWorkspace(previous)
insertWorkspace(target)
yield* insertWorkspace(previous)
yield* insertWorkspace(target)
registerAdapter(instance.project.id, previousType, localAdapter(path.join(dir, "warp-prev-local")).adapter)
registerAdapter(instance.project.id, targetType, localAdapter(path.join(dir, "warp-target-local")).adapter)
const session = yield* sessionSvc.create({})
attachSessionToWorkspace(session.id, previous.id)
yield* attachSessionToWorkspace(session.id, previous.id)
yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id })
const { db } = yield* Database.Service
expect(
Database.use((db) =>
db
(
yield* db
.select({ workspaceID: SessionTable.workspace_id })
.from(SessionTable)
.where(eq(SessionTable.id, session.id))
.get(),
.get()
.pipe(Effect.orDie)
)?.workspaceID,
).toBe(target.id)
expect(sessionSequenceOwner(session.id)).toBe(target.id)
expect(yield* sessionSequenceOwner(session.id)).toBe(target.id)
})
},
{ git: true },
@@ -867,12 +876,12 @@ describe("workspace CRUD", () => {
const previous = workspaceInfo(instance.project.id, previousType)
const target = workspaceInfo(instance.project.id, targetType)
insertWorkspace(previous)
insertWorkspace(target)
yield* insertWorkspace(previous)
yield* insertWorkspace(target)
registerAdapter(instance.project.id, previousType, localAdapter(previousDir, { createDir: false }).adapter)
registerAdapter(instance.project.id, targetType, localAdapter(targetDir, { createDir: false }).adapter)
const session = yield* sessionSvc.create({})
attachSessionToWorkspace(session.id, previous.id)
yield* attachSessionToWorkspace(session.id, previous.id)
yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id, copyChanges: true })
@@ -893,23 +902,25 @@ describe("workspace CRUD", () => {
const sessionSvc = yield* SessionNs.Service
const previousType = unique("warp-detach-local")
const previous = workspaceInfo(instance.project.id, previousType)
insertWorkspace(previous)
yield* insertWorkspace(previous)
registerAdapter(instance.project.id, previousType, localAdapter(path.join(dir, "warp-detach-local")).adapter)
const session = yield* sessionSvc.create({})
attachSessionToWorkspace(session.id, previous.id)
yield* attachSessionToWorkspace(session.id, previous.id)
yield* workspace.sessionWarp({ workspaceID: null, sessionID: session.id })
const { db } = yield* Database.Service
expect(
Database.use((db) =>
db
(
yield* db
.select({ workspaceID: SessionTable.workspace_id })
.from(SessionTable)
.where(eq(SessionTable.id, session.id))
.get(),
.get()
.pipe(Effect.orDie)
)?.workspaceID,
).toBeNull()
expect(sessionSequenceOwner(session.id)).toBe(instance.project.id)
expect(yield* sessionSequenceOwner(session.id)).toBe(instance.project.id)
})
},
{ git: true },
@@ -926,9 +937,9 @@ describe("workspace CRUD", () => {
const sessionSvc = yield* SessionNs.Service
const previousType = unique("warp-detach-workspace-instance")
const previous = workspaceInfo(projectID, previousType)
insertWorkspace(previous)
yield* insertWorkspace(previous)
const session = yield* sessionSvc.create({})
attachSessionToWorkspace(session.id, previous.id)
yield* attachSessionToWorkspace(session.id, previous.id)
const workspaceProjectID = yield* provideTmpdirInstance(
(workspaceDir) =>
@@ -942,17 +953,19 @@ describe("workspace CRUD", () => {
{ git: true },
)
const { db } = yield* Database.Service
expect(
Database.use((db) =>
db
(
yield* db
.select({ workspaceID: SessionTable.workspace_id })
.from(SessionTable)
.where(eq(SessionTable.id, session.id))
.get(),
.get()
.pipe(Effect.orDie)
)?.workspaceID,
).toBeNull()
expect(sessionSequenceOwner(session.id)).toBe(projectID)
expect(sessionSequenceOwner(session.id)).not.toBe(workspaceProjectID)
expect(yield* sessionSequenceOwner(session.id)).toBe(projectID)
expect(yield* sessionSequenceOwner(session.id)).not.toBe(workspaceProjectID)
}),
{ git: true },
)
@@ -1005,14 +1018,14 @@ describe("workspace CRUD", () => {
const targetType = unique("warp-remote-target")
const previous = workspaceInfo(instance.project.id, previousType)
const target = workspaceInfo(instance.project.id, targetType, { directory: "remote-target-dir" })
insertWorkspace(previous)
insertWorkspace(target)
yield* insertWorkspace(previous)
yield* insertWorkspace(target)
registerAdapter(instance.project.id, previousType, remoteAdapter(`${url}/warp-source`).adapter)
registerAdapter(instance.project.id, targetType, remoteAdapter(`${url}/warp-target`).adapter)
const session = yield* sessionSvc.create({})
attachSessionToWorkspace(session.id, previous.id)
yield* attachSessionToWorkspace(session.id, previous.id)
historySessionID = session.id
historyNextSeq = (sessionSequence(session.id) ?? -1) + 1
historyNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1
yield* workspace.sessionWarp({ workspaceID: target.id, sessionID: session.id, copyChanges: true })
@@ -1042,7 +1055,7 @@ describe("workspace CRUD", () => {
})
expect(calls[4].json).toEqual({ sessionID: session.id })
expect((yield* sessionSvc.get(session.id)).title).toBe("from source history")
expect(sessionSequenceOwner(session.id)).toBe(target.id)
expect(yield* sessionSequenceOwner(session.id)).toBe(target.id)
}),
{ git: true },
)
@@ -1062,8 +1075,8 @@ describe("workspace sync state", () => {
const type = unique("flag-disabled")
const info = workspaceInfo(instance.project.id, type)
const session = yield* sessionSvc.create({})
attachSessionToWorkspace(session.id, info.id)
insertWorkspace(info)
yield* attachSessionToWorkspace(session.id, info.id)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, localAdapter(path.join(dir, "flag-disabled")).adapter)
yield* Effect.promise(() => startWorkspaceSyncingWithFlag(instance.project.id, false))
@@ -1088,12 +1101,10 @@ describe("workspace sync state", () => {
const second = workspaceInfo(projectID, secondType)
yield* Effect.promise(() => fs.mkdir(path.join(dir, "first"), { recursive: true }))
yield* Effect.promise(() => fs.mkdir(path.join(dir, "second"), { recursive: true }))
yield* Effect.sync(() => {
insertWorkspace(first)
insertWorkspace(second)
registerAdapter(projectID, firstType, localAdapter(path.join(dir, "first")).adapter)
registerAdapter(projectID, secondType, localAdapter(path.join(dir, "second")).adapter)
})
yield* insertWorkspace(first)
yield* insertWorkspace(second)
registerAdapter(projectID, firstType, localAdapter(path.join(dir, "first")).adapter)
registerAdapter(projectID, secondType, localAdapter(path.join(dir, "second")).adapter)
yield* Effect.addFinalizer(() =>
Effect.all([workspace.remove(first.id), workspace.remove(second.id)], { discard: true }).pipe(Effect.ignore),
)
@@ -1121,13 +1132,13 @@ describe("workspace sync state", () => {
const sessionSvc = yield* SessionNs.Service
const type = unique("missing-local")
const info = workspaceInfo(instance.project.id, type)
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(
instance.project.id,
type,
localAdapter(path.join(dir, "missing-target"), { createDir: false }).adapter,
)
attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* workspace.startWorkspaceSyncing(instance.project.id)
@@ -1157,9 +1168,9 @@ describe("workspace sync state", () => {
const info = workspaceInfo(instance.project.id, type)
const target = path.join(dir, "dedupe-local")
yield* Effect.promise(() => fs.mkdir(target, { recursive: true }))
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, localAdapter(target).adapter)
attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* workspace.startWorkspaceSyncing(instance.project.id)
yield* workspace.startWorkspaceSyncing(instance.project.id)
@@ -1211,9 +1222,9 @@ describe("workspace sync state", () => {
try {
const type = unique("remote-start")
const info = workspaceInfo(instance.project.id, type)
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sync`).adapter)
attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* workspace.startWorkspaceSyncing(instance.project.id)
yield* eventuallyEffect(
@@ -1265,9 +1276,9 @@ describe("workspace sync state", () => {
const instance = yield* requireInstance
const type = unique("remote-connect-fail")
const info = workspaceInfo(instance.project.id, type)
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, remoteAdapter(`${url}/failed`).adapter)
attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* workspace.startWorkspaceSyncing(instance.project.id)
@@ -1306,9 +1317,9 @@ describe("workspace sync state", () => {
const instance = yield* requireInstance
const type = unique("remote-history-fail")
const info = workspaceInfo(instance.project.id, type)
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, remoteAdapter(`${url}/history-failed`).adapter)
attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* workspace.startWorkspaceSyncing(instance.project.id)
@@ -1364,12 +1375,12 @@ describe("workspace sync state", () => {
try {
const type = unique("history-replay")
const info = workspaceInfo(instance.project.id, type)
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, remoteAdapter(`${url}/history`).adapter)
const session = yield* sessionSvc.create({ title: "before history" })
attachSessionToWorkspace(session.id, info.id)
yield* attachSessionToWorkspace(session.id, info.id)
historySessionID = session.id
historyNextSeq = (sessionSequence(session.id) ?? -1) + 1
historyNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1
yield* workspace.startWorkspaceSyncing(instance.project.id)
@@ -1432,9 +1443,9 @@ describe("workspace sync state", () => {
try {
const type = unique("sse-forward")
const info = workspaceInfo(instance.project.id, type)
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sse-forward`).adapter)
attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* attachSessionToWorkspace((yield* sessionSvc.create({})).id, info.id)
yield* workspace.startWorkspaceSyncing(instance.project.id)
@@ -1514,12 +1525,12 @@ describe("workspace sync state", () => {
try {
const type = unique("sse-sync")
const info = workspaceInfo(instance.project.id, type)
insertWorkspace(info)
yield* insertWorkspace(info)
registerAdapter(instance.project.id, type, remoteAdapter(`${url}/sse-sync`).adapter)
const session = yield* sessionSvc.create({ title: "before sse" })
attachSessionToWorkspace(session.id, info.id)
yield* attachSessionToWorkspace(session.id, info.id)
sseSessionID = session.id
sseNextSeq = (sessionSequence(session.id) ?? -1) + 1
sseNextSeq = ((yield* sessionSequence(session.id)) ?? -1) + 1
yield* workspace.startWorkspaceSyncing(instance.project.id)
@@ -1564,7 +1575,8 @@ describe("workspace waitForSync", () => {
Effect.gen(function* () {
const workspace = yield* Workspace.Service
const sessionID = SessionID.descending("ses_wait_done")
Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 4 }).run())
const { db } = yield* Database.Service
yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 4 }).run().pipe(Effect.orDie)
expect(yield* workspace.waitForSync(WorkspaceV2.ID.ascending("wrk_wait_done"), { [sessionID]: 4 })).toBeUndefined()
expect(
@@ -1581,20 +1593,20 @@ describe("workspace waitForSync", () => {
const workspace = yield* Workspace.Service
const workspaceID = WorkspaceV2.ID.ascending("wrk_wait_event")
const sessionID = SessionID.descending("ses_wait_event")
Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 1 }).run())
const { db } = yield* Database.Service
yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 1 }).run().pipe(Effect.orDie)
yield* Effect.all(
[
workspace.waitForSync(workspaceID, { [sessionID]: 2 }),
Effect.gen(function* () {
yield* Effect.sleep("10 millis")
Database.use((db) =>
db
.update(EventSequenceTable)
.set({ seq: 2 })
.where(eq(EventSequenceTable.aggregate_id, sessionID))
.run(),
)
yield* db
.update(EventSequenceTable)
.set({ seq: 2 })
.where(eq(EventSequenceTable.aggregate_id, sessionID))
.run()
.pipe(Effect.orDie)
GlobalBus.emit("event", { workspace: workspaceID, payload: { type: "anything" } })
}),
],
@@ -1611,20 +1623,20 @@ describe("workspace waitForSync", () => {
const workspace = yield* Workspace.Service
const workspaceID = WorkspaceV2.ID.ascending("wrk_wait_sync_any")
const sessionID = SessionID.descending("ses_wait_sync_any")
Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 0 }).run())
const { db } = yield* Database.Service
yield* db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 0 }).run().pipe(Effect.orDie)
yield* Effect.all(
[
workspace.waitForSync(workspaceID, { [sessionID]: 1 }),
Effect.gen(function* () {
yield* Effect.sleep("10 millis")
Database.use((db) =>
db
.update(EventSequenceTable)
.set({ seq: 1 })
.where(eq(EventSequenceTable.aggregate_id, sessionID))
.run(),
)
yield* db
.update(EventSequenceTable)
.set({ seq: 1 })
.where(eq(EventSequenceTable.aggregate_id, sessionID))
.run()
.pipe(Effect.orDie)
GlobalBus.emit("event", {
workspace: WorkspaceV2.ID.ascending("wrk_other_workspace"),
payload: { type: "sync" },
packages/opencode/test/fixture/db.ts
+2 -3
View File
@@ -1,11 +1,10 @@
import { rm } from "fs/promises"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { disposeAllInstances } from "./fixture"
export async function resetDatabase() {
await disposeAllInstances().catch(() => undefined)
Database.close()
const dbPath = Database.getPath()
const dbPath = Database.path()
await rm(dbPath, { force: true }).catch(() => undefined)
await rm(`${dbPath}-wal`, { force: true }).catch(() => undefined)
await rm(`${dbPath}-shm`, { force: true }).catch(() => undefined)
packages/opencode/test/fixture/workspace.ts
+2 -2
View File
@@ -11,17 +11,17 @@ import { Project } from "../../src/project/project"
import { Vcs } from "../../src/project/vcs"
import { Session } from "../../src/session/session"
import { SessionPrompt } from "../../src/session/prompt"
import { SyncEvent } from "../../src/sync"
import { EventV2Bridge } from "../../src/event-v2-bridge"
export const workspaceLayerWithRuntimeFlags = (overrides: Partial<RuntimeFlags.Info>) =>
Workspace.layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(SessionPrompt.defaultLayer),
Layer.provide(Project.defaultLayer),
Layer.provide(Vcs.defaultLayer),
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2Bridge.defaultLayer),
Layer.provide(FetchHttpClient.layer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(RuntimeFlags.layer(overrides)),
packages/opencode/test/permission/next.test.ts
+2 -1
View File
@@ -3,6 +3,7 @@ import os from "os"
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
import { Bus } from "../../src/bus"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { Database } from "@opencode-ai/core/database/database"
import { Permission } from "../../src/permission"
import { PermissionID } from "../../src/permission/schema"
import { InstanceBootstrap } from "../../src/project/bootstrap-service"
@@ -14,7 +15,7 @@ import { MessageID, SessionID } from "../../src/session/schema"
const bus = Bus.layer
const noopBootstrap = Layer.succeed(InstanceBootstrap.Service, InstanceBootstrap.Service.of({ run: Effect.void }))
const env = Layer.mergeAll(
Permission.layer.pipe(Layer.provide(bus)),
Permission.layer.pipe(Layer.provide(Database.defaultLayer), Layer.provide(bus)),
bus,
CrossSpawnSpawner.defaultLayer,
InstanceStore.defaultLayer.pipe(Layer.provide(noopBootstrap)),
packages/opencode/test/plugin/workspace-adapter.test.ts
+2 -2
View File
@@ -21,12 +21,12 @@ import { Vcs } from "../../src/project/vcs"
import { InstanceState } from "../../src/effect/instance-state"
import { Session } from "../../src/session/session"
import { SessionPrompt } from "../../src/session/prompt"
import { SyncEvent } from "../../src/sync"
import { disposeAllInstances, provideTmpdirInstance } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
import { AccountTest } from "../fake/account"
import { AuthTest } from "../fake/auth"
import { NpmTest } from "../fake/npm"
import { EventV2Bridge } from "../../src/event-v2-bridge"
const configLayer = Config.layer.pipe(
Layer.provide(EffectFlock.defaultLayer),
@@ -46,12 +46,12 @@ const noopBootstrapLayer = Layer.succeed(InstanceBootstrap.Service, InstanceBoot
const workspaceLayer = Workspace.layer.pipe(
Layer.provide(Auth.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(SessionPrompt.defaultLayer),
Layer.provide(Project.defaultLayer),
Layer.provide(Vcs.defaultLayer),
Layer.provide(FetchHttpClient.layer),
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2Bridge.defaultLayer),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(InstanceStore.defaultLayer.pipe(Layer.provide(noopBootstrapLayer))),
Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: true })),
packages/opencode/test/preload.ts
-2
View File
@@ -10,8 +10,6 @@ import { afterAll } from "bun:test"
const dir = path.join(os.tmpdir(), "opencode-test-data-" + process.pid)
await fs.mkdir(dir, { recursive: true })
afterAll(async () => {
const { Database } = await import("../src/storage/db")
Database.close()
const busy = (error: unknown) =>
typeof error === "object" && error !== null && "code" in error && error.code === "EBUSY"
const rm = async (left: number): Promise<void> => {
packages/opencode/test/project/migrate-global.test.ts
+27 -17
View File
@@ -1,6 +1,6 @@
import { describe, expect } from "bun:test"
import { Project } from "@/project/project"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { eq } from "drizzle-orm"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { ProjectTable } from "@opencode-ai/core/project/sql"
@@ -15,7 +15,7 @@ import { testEffect } from "../lib/effect"
void Log.init({ print: false })
const it = testEffect(Layer.mergeAll(Project.defaultLayer, CrossSpawnSpawner.defaultLayer))
const it = testEffect(Layer.mergeAll(Project.defaultLayer, CrossSpawnSpawner.defaultLayer, Database.defaultLayer))
function legacySessionID() {
// Global-session migration covers persisted IDs from before prefixed session IDs.
@@ -24,7 +24,7 @@ function legacySessionID() {
function seed(opts: { id: SessionID; dir: string; project: ProjectV2.ID }) {
const now = Date.now()
Database.use((db) =>
return Database.Service.use(({ db }) =>
db
.insert(SessionTable)
.values({
@@ -37,12 +37,13 @@ function seed(opts: { id: SessionID; dir: string; project: ProjectV2.ID }) {
time_created: now,
time_updated: now,
})
.run(),
.run()
.pipe(Effect.orDie),
)
}
function ensureGlobal() {
Database.use((db) =>
return Database.Service.use(({ db }) =>
db
.insert(ProjectTable)
.values({
@@ -53,7 +54,8 @@ function ensureGlobal() {
sandboxes: [],
})
.onConflictDoNothing()
.run(),
.run()
.pipe(Effect.orDie),
)
}
@@ -72,7 +74,7 @@ describe("migrateFromGlobal", () => {
// 2. Seed a session under "global" with matching directory
const id = legacySessionID()
yield* Effect.sync(() => seed({ id, dir: tmp, project: ProjectV2.ID.global }))
yield* seed({ id, dir: tmp, project: ProjectV2.ID.global })
// 3. Make a commit so the project gets a real ID
yield* Effect.promise(() => $`git commit --allow-empty -m "root"`.cwd(tmp).quiet())
@@ -81,7 +83,9 @@ describe("migrateFromGlobal", () => {
expect(real.id).not.toBe(ProjectV2.ID.global)
// 4. The session should have been migrated to the real project ID
const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get())
const row = yield* Database.Service.use(({ db }) =>
db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie),
)
expect(row).toBeDefined()
expect(row!.project_id).toBe(real.id)
}),
@@ -96,19 +100,21 @@ describe("migrateFromGlobal", () => {
expect(project.id).not.toBe(ProjectV2.ID.global)
// 2. Ensure "global" project row exists (as it would from a prior no-git session)
yield* Effect.sync(() => ensureGlobal())
yield* ensureGlobal()
// 3. Seed a session under "global" with matching directory.
// This simulates a session created before git init that wasn't
// present when the real project row was first created.
const id = legacySessionID()
yield* Effect.sync(() => seed({ id, dir: tmp, project: ProjectV2.ID.global }))
yield* seed({ id, dir: tmp, project: ProjectV2.ID.global })
// 4. Call fromDirectory again — project row already exists,
// so the current code skips migration entirely. This is the bug.
yield* projects.fromDirectory(tmp)
const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get())
const row = yield* Database.Service.use(({ db }) =>
db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie),
)
expect(row).toBeDefined()
expect(row!.project_id).toBe(project.id)
}),
@@ -121,16 +127,18 @@ describe("migrateFromGlobal", () => {
const { project } = yield* projects.fromDirectory(tmp)
expect(project.id).not.toBe(ProjectV2.ID.global)
yield* Effect.sync(() => ensureGlobal())
yield* ensureGlobal()
// Legacy sessions may lack a directory value.
// Without a matching origin directory, they should remain global.
const id = legacySessionID()
yield* Effect.sync(() => seed({ id, dir: "", project: ProjectV2.ID.global }))
yield* seed({ id, dir: "", project: ProjectV2.ID.global })
yield* projects.fromDirectory(tmp)
const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get())
const row = yield* Database.Service.use(({ db }) =>
db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie),
)
expect(row).toBeDefined()
expect(row!.project_id).toBe(ProjectV2.ID.global)
}),
@@ -143,14 +151,16 @@ describe("migrateFromGlobal", () => {
const { project } = yield* projects.fromDirectory(tmp)
expect(project.id).not.toBe(ProjectV2.ID.global)
yield* Effect.sync(() => ensureGlobal())
yield* ensureGlobal()
// Seed a session under "global" but for a DIFFERENT directory
const id = legacySessionID()
yield* Effect.sync(() => seed({ id, dir: "/some/other/dir", project: ProjectV2.ID.global }))
yield* seed({ id, dir: "/some/other/dir", project: ProjectV2.ID.global })
yield* projects.fromDirectory(tmp)
const row = Database.use((db) => db.select().from(SessionTable).where(eq(SessionTable.id, id)).get())
const row = yield* Database.Service.use(({ db }) =>
db.select().from(SessionTable).where(eq(SessionTable.id, id)).get().pipe(Effect.orDie),
)
expect(row).toBeDefined()
// Should remain under "global" — not stolen
expect(row!.project_id).toBe(ProjectV2.ID.global)
packages/opencode/test/server/httpapi-event-diagnostics.test.ts
-279
View File
@@ -1,279 +0,0 @@
// Diagnostic suite for /event SSE delivery.
//
// Each test isolates ONE variable in the publisher chain while keeping the
// subscriber path constant (in-process HttpApi via Server.Default reading the
// SSE body). The pass/fail pattern across tests tells us where the bug lives:
//
// D1 (baseline): publish via Bus.use.publish — mirror of httpapi-event.test.ts
// test 3. Confirms /event SSE delivery works for SOME publish path.
//
// D2: publish N times in quick succession via Bus.use.publish. If the bus
// subscription is acquired correctly there should be no message loss.
//
// D3: publish via SyncEvent.use.run — exercises the same path the HTTP
// handlers use (Session.updatePart → sync.run → bus.publish) without
// the HTTP roundtrip. Tells us whether the sync path itself can deliver
// in-process.
//
// D4: publish via SyncEvent.use.run; subscriber is an in-process Bus
// callback. Confirms pub/sub identity end-to-end without /event SSE.
//
// D5: in-process Bus callback subscriber AND raw /event SSE subscriber
// receive the same publish. If both receive: no bug. If only the
// callback receives: the /event handler has an acquisition race.
//
// D6: same as D5 but the callback subscriber is attached AFTER /event SSE
// subscription is established. Order-of-setup variable.
import { afterEach, describe, expect } from "bun:test"
import { Deferred, Effect, Layer, Schema } from "effect"
import * as Log from "@opencode-ai/core/util/log"
import { Bus } from "../../src/bus"
import { Event as ServerEvent } from "../../src/server/event"
import { Server } from "../../src/server/server"
import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event"
import { MessageV2 } from "../../src/session/message-v2"
import { MessageID, PartID, SessionID } from "../../src/session/schema"
import { SyncEvent } from "../../src/sync"
import { resetDatabase } from "../fixture/db"
import { disposeAllInstances, TestInstance } from "../fixture/fixture"
import { testEffectShared } from "../lib/effect"
void Log.init({ print: false })
const SseEvent = Schema.Struct({
id: Schema.optional(Schema.String),
type: Schema.String,
properties: Schema.Record(Schema.String, Schema.Any),
})
type SseEvent = Schema.Schema.Type<typeof SseEvent>
type BusEvent = { type: string; properties: unknown }
afterEach(async () => {
await disposeAllInstances()
await resetDatabase()
})
const it = testEffectShared(Layer.mergeAll(Bus.defaultLayer, SyncEvent.defaultLayer))
const publishConnected = Bus.use.publish(ServerEvent.Connected, {})
const publishPartUpdated = (partID: ReturnType<typeof PartID.ascending>) => {
const sessionID = SessionID.make(`ses_${Date.now().toString(36)}${Math.random().toString(36).slice(2, 8)}`)
return Bus.use.publish(MessageV2.Event.PartUpdated, {
sessionID,
part: { id: partID, sessionID, messageID: MessageID.ascending(), type: "text", text: "diag" },
time: Date.now(),
})
}
const subscribeAllCallback = (handler: (event: BusEvent) => void) =>
Effect.acquireRelease(Bus.use.subscribeAllCallback(handler), (dispose) => Effect.sync(() => dispose()))
const openEventStream = (directory: string) =>
Effect.gen(function* () {
const response = yield* Effect.promise(async () =>
Server.Default().app.request(EventPaths.event, { headers: { "x-opencode-directory": directory } }),
)
if (!response.body) return yield* Effect.die("missing SSE response body")
const reader = response.body.getReader()
yield* Effect.addFinalizer(() => Effect.promise(() => reader.cancel().catch(() => undefined)))
return reader
})
const decoder = new TextDecoder()
function decodeFrame(value: Uint8Array): SseEvent[] {
return decoder
.decode(value)
.split(/\n\n+/)
.map((part) => part.trim())
.filter((part) => part.length > 0)
.map((part) => Schema.decodeUnknownSync(SseEvent)(JSON.parse(part.replace(/^data: /, ""))))
}
const readNextEvent = (reader: ReadableStreamDefaultReader<Uint8Array>) =>
Effect.promise(() => reader.read()).pipe(
Effect.timeoutOrElse({
duration: "3 seconds",
orElse: () => Effect.fail(new Error("timed out reading SSE chunk")),
}),
Effect.flatMap((result) => {
if (result.done || !result.value) return Effect.fail(new Error("event stream closed"))
const frames = decodeFrame(result.value)
if (frames.length === 0) return Effect.fail(new Error("empty SSE frame"))
return Effect.succeed(frames[0]!)
}),
)
const collectUntilEvent = (reader: ReadableStreamDefaultReader<Uint8Array>, predicate: (event: SseEvent) => boolean) =>
Effect.gen(function* () {
const events: SseEvent[] = []
while (true) {
const event = yield* readNextEvent(reader)
events.push(event)
if (predicate(event)) return events
}
}).pipe(
Effect.timeoutOrElse({
duration: "4 seconds",
orElse: () => Effect.fail(new Error("collectUntil deadline exceeded")),
}),
)
const isPartUpdated = (event: { type: string }) => event.type === MessageV2.Event.PartUpdated.type
describe("/event SSE delivery diagnostics", () => {
// Sanity: baseline same as httpapi-event.test.ts test 3 (already known to pass)
// but explicit about timing — publish happens with NO wait after reading
// server.connected. If this fails we have a deeper problem than just sync.
it.instance(
"D1: delivers a single bus event published right after server.connected",
() =>
Effect.gen(function* () {
const { directory } = yield* TestInstance
const reader = yield* openEventStream(directory)
expect((yield* readNextEvent(reader)).type).toBe("server.connected")
yield* publishConnected
expect((yield* readNextEvent(reader)).type).toBe("server.connected")
}),
{ git: true, config: { formatter: false, lsp: false } },
)
// If D1 passes but D2 fails, we have a queue-drain or partial-loss issue.
it.instance(
"D2: delivers all N bus events published in rapid succession",
() =>
Effect.gen(function* () {
const { directory } = yield* TestInstance
const reader = yield* openEventStream(directory)
expect((yield* readNextEvent(reader)).type).toBe("server.connected")
const N = 5
yield* Effect.replicateEffect(publishConnected, N)
const received = yield* Effect.replicateEffect(readNextEvent(reader), N)
expect(received).toHaveLength(N)
for (const event of received) expect(event.type).toBe("server.connected")
}),
{ git: true, config: { formatter: false, lsp: false } },
)
// The critical test. If D1 passes but this fails, the bus-identity fix is
// incomplete OR the sync.run publish path doesn't reach the same bus
// /event subscribes to, even when both share the memoMap.
it.instance(
"D3: delivers a SyncEvent published via SyncEvent.use.run after server.connected",
() =>
Effect.gen(function* () {
const { directory } = yield* TestInstance
const reader = yield* openEventStream(directory)
expect((yield* readNextEvent(reader)).type).toBe("server.connected")
const partID = PartID.ascending()
yield* publishPartUpdated(partID)
const collected = yield* collectUntilEvent(reader, isPartUpdated)
const updated = collected.find(isPartUpdated)
expect(updated?.properties.part.id).toBe(partID)
}),
{ git: true, config: { formatter: false, lsp: false } },
)
// If D3 passes but D5 (the SDK E2E in httpapi-sdk.test.ts) fails, then the
// bug is specifically in the cross-request / cross-fiber HTTP path, not in
// the publish itself. If D3 also fails, the publish chain is broken.
//
// D4: ensure the publish reaches an in-process Bus subscriber too. Confirms
// pub/sub identity end-to-end without involving /event SSE.
it.instance(
"D4: SyncEvent.use.run publish reaches an in-process Bus callback",
() =>
Effect.gen(function* () {
const received = yield* Deferred.make<BusEvent>()
yield* subscribeAllCallback((event) => {
if (isPartUpdated(event)) Deferred.doneUnsafe(received, Effect.succeed(event))
})
const partID = PartID.ascending()
yield* publishPartUpdated(partID)
const event = yield* Deferred.await(received).pipe(
Effect.timeoutOrElse({
duration: "3 seconds",
orElse: () => Effect.fail(new Error("D4 timed out waiting for callback")),
}),
)
expect(event.type).toBe(MessageV2.Event.PartUpdated.type)
expect(event.properties).toMatchObject({ part: { id: partID } })
}),
{ git: true, config: { formatter: false, lsp: false } },
)
// D5: BOTH subscribers attached simultaneously. Trigger ONE publish via
// SyncEvent.use.run. Both subscribers should receive it. If only one does
// we know exactly which side of the chain is failing.
it.instance(
"D5: same SyncEvent.use.run publish reaches BOTH /event SSE and in-process callback",
() =>
Effect.gen(function* () {
const { directory } = yield* TestInstance
const callbackReceived = yield* Deferred.make<BusEvent>()
yield* subscribeAllCallback((event) => {
if (isPartUpdated(event)) Deferred.doneUnsafe(callbackReceived, Effect.succeed(event))
})
const reader = yield* openEventStream(directory)
expect((yield* readNextEvent(reader)).type).toBe("server.connected")
const partID = PartID.ascending()
yield* publishPartUpdated(partID)
const sseSaw = yield* collectUntilEvent(reader, isPartUpdated).pipe(
Effect.map((events) => events.some(isPartUpdated)),
Effect.catch(() => Effect.succeed(false)),
)
const callbackSaw = yield* Deferred.await(callbackReceived).pipe(
Effect.timeoutOrElse({ duration: "1 second", orElse: () => Effect.succeed(undefined) }),
Effect.map((event) => event !== undefined),
)
// Single assert with the boolean pair so the failure message tells us
// exactly which side broke.
expect({ sseSaw, callbackSaw }).toEqual({ sseSaw: true, callbackSaw: true })
}),
{ git: true, config: { formatter: false, lsp: false } },
)
// D6: same as D5 but the callback subscriber is attached AFTER /event SSE
// subscription is established. If D5 fails and D6 passes, the order of
// subscriber setup is the determining factor.
it.instance(
"D6: /event SSE receives sync.run publish when callback is attached AFTER /event opens",
() =>
Effect.gen(function* () {
const { directory } = yield* TestInstance
const reader = yield* openEventStream(directory)
expect((yield* readNextEvent(reader)).type).toBe("server.connected")
const callbackReceived = yield* Deferred.make<BusEvent>()
yield* subscribeAllCallback((event) => {
if (isPartUpdated(event)) Deferred.doneUnsafe(callbackReceived, Effect.succeed(event))
})
const partID = PartID.ascending()
yield* publishPartUpdated(partID)
const sseSaw = yield* collectUntilEvent(reader, isPartUpdated).pipe(
Effect.map((events) => events.some(isPartUpdated)),
Effect.catch(() => Effect.succeed(false)),
)
const callbackSaw = yield* Deferred.await(callbackReceived).pipe(
Effect.timeoutOrElse({ duration: "1 second", orElse: () => Effect.succeed(undefined) }),
Effect.map((event) => event !== undefined),
)
expect({ sseSaw, callbackSaw }).toEqual({ sseSaw: true, callbackSaw: true })
}),
{ git: true, config: { formatter: false, lsp: false } },
)
})
packages/opencode/test/server/httpapi-experimental.test.ts
+25 -23
View File
@@ -6,7 +6,9 @@ import { Server } from "../../src/server/server"
import { ExperimentalPaths } from "../../src/server/routes/instance/httpapi/groups/experimental"
import { Session } from "@/session/session"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { AccountV2 } from "@opencode-ai/core/account"
import { AccountTable } from "@opencode-ai/core/account/sql"
import * as Log from "@opencode-ai/core/util/log"
import { Worktree } from "../../src/worktree"
import { resetDatabase } from "../fixture/db"
@@ -15,7 +17,7 @@ import { testEffect } from "../lib/effect"
void Log.init({ print: false })
const it = testEffect(Layer.mergeAll(Session.defaultLayer))
const it = testEffect(Layer.mergeAll(Session.defaultLayer, Database.defaultLayer))
const testWorktreeMutations = process.platform === "win32" ? it.instance.skip : it.instance
function app() {
@@ -62,34 +64,34 @@ function waitReady(input: { directory?: string; name?: string }) {
function insertAccount() {
return Effect.acquireRelease(
Effect.sync(() => {
Database.Client()
.$client.prepare(
"INSERT INTO account (id, email, url, access_token, refresh_token, time_created, time_updated) VALUES (?, ?, ?, ?, ?, ?, ?)",
)
.run(
"account-test",
"test@example.com",
"https://console.example.com",
"access",
"refresh",
Date.now(),
Date.now(),
)
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db
.insert(AccountTable)
.values({
id: AccountV2.ID.make("account-test"),
email: "test@example.com",
url: "https://console.example.com",
access_token: AccountV2.AccessToken.make("access"),
refresh_token: AccountV2.RefreshToken.make("refresh"),
time_created: Date.now(),
time_updated: Date.now(),
})
.run()
.pipe(Effect.orDie)
return "account-test"
}),
(id) =>
Effect.sync(() => {
Database.Client().$client.prepare("DELETE FROM account WHERE id = ?").run(id)
}),
Database.Service.use(({ db }) =>
db.delete(AccountTable).where(eq(AccountTable.id, AccountV2.ID.make(id))).run().pipe(Effect.orDie),
),
)
}
function setSessionUpdated(session: Session.Info, updated: number) {
return Effect.sync(() => {
Database.use((db) =>
db.update(SessionTable).set({ time_updated: updated }).where(eq(SessionTable.id, session.id)).run(),
)
return Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db.update(SessionTable).set({ time_updated: updated }).where(eq(SessionTable.id, session.id)).run().pipe(Effect.orDie)
})
}
packages/opencode/test/server/httpapi-instance.test.ts
+1 -1
View File
@@ -17,7 +17,7 @@ import { resetDatabase } from "../fixture/db"
import { tmpdirScoped } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
// Flip the experimental workspaces flag so SyncEvent.run actually writes to
// Flip the experimental workspaces flag so EventV2.run actually writes to
// EventSequenceTable (the source of truth the fence middleware reads). Reset
// the database around the test so per-instance state does not leak between
// runs. resetDatabase() already calls disposeAllInstances(), so we don't
packages/opencode/test/server/httpapi-schema-error-body.test.ts
+17 -19
View File
@@ -1,7 +1,7 @@
import { afterEach, describe, expect } from "bun:test"
import { Effect } from "effect"
import { Effect, Layer } from "effect"
import { eq } from "drizzle-orm"
import * as Database from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { Server } from "../../src/server/server"
import { Session } from "@/session/session"
@@ -14,7 +14,7 @@ import { disposeAllInstances, TestInstance } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
import { ProviderV2 } from "@opencode-ai/core/provider"
const it = testEffect(Session.defaultLayer)
const it = testEffect(Layer.mergeAll(Session.defaultLayer, Database.defaultLayer))
afterEach(async () => {
await disposeAllInstances()
@@ -44,22 +44,20 @@ const seedCorruptStepFinishPart = Effect.gen(function* () {
})
// Schema.Finite still rejects NaN at encode: exact mirror of the corrupt row
// that broke the user's session in the OMO/Windows bug.
yield* Effect.sync(() =>
Database.use((db) =>
db
.update(PartTable)
.set({
data: {
type: "step-finish",
reason: "stop",
cost: 0,
tokens: { input: 0, output: NaN, reasoning: 0, cache: { read: 0, write: 0 } },
} as never, // drizzle's .set() can't narrow the discriminated union
})
.where(eq(PartTable.id, partID))
.run(),
),
)
const { db } = yield* Database.Service
yield* db
.update(PartTable)
.set({
data: {
type: "step-finish",
reason: "stop",
cost: 0,
tokens: { input: 0, output: NaN, reasoning: 0, cache: { read: 0, write: 0 } },
} as never, // drizzle's .set() can't narrow the discriminated union
})
.where(eq(PartTable.id, partID))
.run()
.pipe(Effect.orDie)
return info.id
})
packages/opencode/test/server/httpapi-sdk.test.ts
+1 -1
View File
@@ -642,7 +642,7 @@ describe("HttpApi SDK", () => {
),
)
// Regression: SyncEvent must publish on the same ProjectBus the /event handler
// Regression: EventV2 must publish on the same ProjectBus the /event handler
// subscribes to, AND the /event stream must forward handler ALS/context into the
// body-pump fiber. Drives the full SDK → /event → Session.updatePart → sync.run →
// bus.publish → SDK subscriber path. Goes red if either the publisher uses a
packages/opencode/test/server/httpapi-session.test.ts
+62 -64
View File
@@ -19,7 +19,7 @@ import { SessionPaths } from "../../src/server/routes/instance/httpapi/groups/se
import { Session } from "@/session/session"
import { MessageID, PartID, SessionID, type SessionID as SessionIDType } from "../../src/session/schema"
import { MessageV2 } from "../../src/session/message-v2"
import { Database } from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { ModelV2 } from "@opencode-ai/core/model"
@@ -43,7 +43,9 @@ const instanceStoreLayer = InstanceStore.defaultLayer.pipe(
Layer.succeed(InstanceBootstrapService.Service, InstanceBootstrapService.Service.of({ run: Effect.void })),
),
)
const it = testEffect(Layer.mergeAll(instanceStoreLayer, Project.defaultLayer, Session.defaultLayer, workspaceLayer))
const it = testEffect(
Layer.mergeAll(instanceStoreLayer, Project.defaultLayer, Session.defaultLayer, workspaceLayer, Database.defaultLayer),
)
function app() {
return Server.Default().app
@@ -107,7 +109,7 @@ const createLocalWorkspace = (input: { projectID: Project.Info["id"]; type: stri
)
const insertLegacyAssistantMessage = (sessionID: SessionIDType, time = 1) =>
Effect.sync(() => {
Effect.gen(function* () {
const message = new SessionMessage.Assistant({
id: SessionMessage.ID.create(),
type: "assistant",
@@ -120,76 +122,72 @@ const insertLegacyAssistantMessage = (sessionID: SessionIDType, time = 1) =>
time: { created: DateTime.makeUnsafe(time) },
content: [],
})
Database.use((db) =>
db
.insert(SessionMessageTable)
.values([
{
id: message.id,
session_id: sessionID,
type: message.type,
time_created: time,
data: {
time: { created: time },
agent: message.agent,
model: message.model,
content: message.content,
} as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>,
},
])
.run(),
)
const { db } = yield* Database.Service
yield* db
.insert(SessionMessageTable)
.values([
{
id: message.id,
session_id: sessionID,
type: message.type,
time_created: time,
data: {
time: { created: time },
agent: message.agent,
model: message.model,
content: message.content,
} as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>,
},
])
.run()
.pipe(Effect.orDie)
})
const insertCorruptV2Message = (sessionID: SessionIDType, time = 1) =>
Effect.sync(() =>
Database.use((db) =>
db
.insert(SessionMessageTable)
.values([
{
id: SessionMessage.ID.create(),
session_id: sessionID,
type: "assistant",
time_created: time,
data: {} as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>,
},
])
.run(),
),
)
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db
.insert(SessionMessageTable)
.values([
{
id: SessionMessage.ID.create(),
session_id: sessionID,
type: "assistant",
time_created: time,
data: {} as NonNullable<(typeof SessionMessageTable.$inferInsert)["data"]>,
},
])
.run()
.pipe(Effect.orDie)
})
const setLegacySummaryDiff = (sessionID: SessionIDType) =>
Effect.sync(() =>
Database.use((db) =>
db
.update(SessionTable)
.set({
summary_additions: 1,
summary_deletions: 0,
summary_files: 1,
summary_diffs: [{ additions: 1, deletions: 0 }],
})
.where(eq(SessionTable.id, sessionID))
.run(),
),
)
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db
.update(SessionTable)
.set({
summary_additions: 1,
summary_deletions: 0,
summary_files: 1,
summary_diffs: [{ additions: 1, deletions: 0 }],
})
.where(eq(SessionTable.id, sessionID))
.run()
.pipe(Effect.orDie)
})
const getWorkspaceID = (sessionID: SessionIDType) =>
Effect.sync(() =>
Database.use((db) =>
db
.select({ workspaceID: SessionTable.workspace_id })
.from(SessionTable)
.where(eq(SessionTable.id, sessionID))
.get(),
),
)
Effect.gen(function* () {
const { db } = yield* Database.Service
return yield* db.select({ workspaceID: SessionTable.workspace_id }).from(SessionTable).where(eq(SessionTable.id, sessionID)).get().pipe(Effect.orDie)
})
const clearSessionPath = (sessionID: SessionIDType) =>
Effect.sync(() =>
Database.use((db) => db.update(SessionTable).set({ path: null }).where(eq(SessionTable.id, sessionID)).run()),
)
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db.update(SessionTable).set({ path: null }).where(eq(SessionTable.id, sessionID)).run().pipe(Effect.orDie)
})
function request(path: string, init?: RequestInit) {
return Effect.promise(async () => app().request(path, init))
packages/opencode/test/server/httpapi-workspace-routing.test.ts
+5 -3
View File
@@ -19,6 +19,7 @@ import { WorkspaceV2 } from "@opencode-ai/core/workspace"
import type { WorkspaceAdapter } from "../../src/control-plane/types"
import { Workspace } from "../../src/control-plane/workspace"
import { WorkspaceTable } from "@opencode-ai/core/control-plane/workspace.sql"
import { Database } from "@opencode-ai/core/database/database"
import { Project } from "../../src/project/project"
import { WorkspacePaths } from "../../src/server/routes/instance/httpapi/groups/workspace"
import {
@@ -26,7 +27,6 @@ import {
workspaceRouterMiddleware,
} from "../../src/server/routes/instance/httpapi/middleware/workspace-routing"
import { HEADER as FenceHeader } from "../../src/server/shared/fence"
import { Database } from "../../src/storage/db"
import { resetDatabase } from "../fixture/db"
import { workspaceLayerWithRuntimeFlags } from "../fixture/workspace"
import { tmpdirScoped } from "../fixture/fixture"
@@ -50,6 +50,7 @@ const it = testEffect(
testStateLayer,
NodeHttpServer.layerTest,
NodeServices.layer,
Database.defaultLayer,
Project.defaultLayer,
workspaceLayer,
Socket.layerWebSocketConstructorGlobal,
@@ -160,10 +161,11 @@ const insertRemoteWorkspaceWithoutSync = (input: {
type: string
url: string
}) =>
Effect.sync(() => {
Effect.gen(function* () {
const id = WorkspaceV2.ID.ascending()
registerAdapter(input.projectID, input.type, remoteAdapter(path.join(input.dir, `.${input.type}`), input.url))
Database.use((db) => db.insert(WorkspaceTable).values({ id, type: input.type, project_id: input.projectID }).run())
const { db } = yield* Database.Service
yield* db.insert(WorkspaceTable).values({ id, type: input.type, project_id: input.projectID }).run().pipe(Effect.orDie)
return id
})
packages/opencode/test/server/negative-tokens-regression.test.ts
+17 -17
View File
@@ -6,21 +6,21 @@
// strict `NonNegativeInt` schema then made every load of the message list
// fail to encode, killing Desktop boot for every user with such a row.
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Effect, Layer } from "effect"
import { eq } from "drizzle-orm"
import { Server } from "../../src/server/server"
import { SessionPaths } from "../../src/server/routes/instance/httpapi/groups/session"
import { Session } from "@/session/session"
import { MessageID, PartID } from "../../src/session/schema"
import * as Database from "@/storage/db"
import { Database } from "@opencode-ai/core/database/database"
import { PartTable } from "@opencode-ai/core/session/sql"
import { resetDatabase } from "../fixture/db"
import { TestInstance } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
import { ProviderV2 } from "@opencode-ai/core/provider"
const it = testEffect(Session.defaultLayer)
const it = testEffect(Layer.mergeAll(Session.defaultLayer, Database.defaultLayer))
function seedNegativeTokenSession() {
return Effect.gen(function* () {
@@ -47,20 +47,20 @@ function seedNegativeTokenSession() {
// Bypass the schema with a direct SQL update to install the
// negative `output` value we want to test loading.
Database.use((db) =>
db
.update(PartTable)
.set({
data: {
type: "step-finish",
reason: "stop",
cost: 0,
tokens: { input: 0, output: -42, reasoning: 0, cache: { read: 0, write: 0 } },
} as never,
})
.where(eq(PartTable.id, partID))
.run(),
)
const { db } = yield* Database.Service
yield* db
.update(PartTable)
.set({
data: {
type: "step-finish",
reason: "stop",
cost: 0,
tokens: { input: 0, output: -42, reasoning: 0, cache: { read: 0, write: 0 } },
} as never,
})
.where(eq(PartTable.id, partID))
.run()
.pipe(Effect.orDie)
return info.id
})
packages/opencode/test/server/session-list.test.ts
+2 -4
View File
@@ -1,7 +1,6 @@
import { afterEach, describe, expect } from "bun:test"
import { Effect, Layer } from "effect"
import { Database } from "@opencode-ai/core/database/database"
import { EventV2 } from "@opencode-ai/core/event"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { Session as SessionNs } from "@/session/session"
import * as Log from "@opencode-ai/core/util/log"
@@ -13,9 +12,9 @@ import { eq } from "drizzle-orm"
import { testEffect } from "../lib/effect"
import { Bus } from "@/bus"
import { Storage } from "@/storage/storage"
import { SyncEvent } from "@/sync"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { BackgroundJob } from "@/background/job"
import { EventV2Bridge } from "@/event-v2-bridge"
void Log.init({ print: false })
const it = testEffect(
@@ -24,9 +23,8 @@ const it = testEffect(
SessionNs.layer.pipe(
Layer.provide(Bus.layer),
Layer.provide(Storage.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2.defaultLayer),
Layer.provide(EventV2Bridge.defaultLayer),
Layer.provide(SessionProjector.defaultLayer),
Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: false })),
Layer.provide(BackgroundJob.defaultLayer),
packages/opencode/test/session/compaction.test.ts
+4 -6
View File
@@ -1,6 +1,7 @@
import { afterEach, describe, expect, mock, test } from "bun:test"
import { SessionLegacy } from "@opencode-ai/core/session/legacy"
import { Database } from "@opencode-ai/core/database/database"
import { EventV2Bridge } from "@/event-v2-bridge"
import { APICallError } from "ai"
import { Cause, Deferred, Effect, Exit, Fiber, Layer, Schema } from "effect"
import * as Stream from "effect/Stream"
@@ -29,9 +30,7 @@ import { ProviderTest } from "../fake/provider"
import { testEffect } from "../lib/effect"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { TestConfig } from "../fixture/config"
import { SyncEvent } from "@/sync"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { EventV2Bridge } from "@/event-v2-bridge"
import { LLMEvent, Usage } from "@opencode-ai/llm"
import { ProviderV2 } from "@opencode-ai/core/provider"
@@ -234,22 +233,22 @@ const deps = Layer.mergeAll(
Plugin.defaultLayer,
Bus.layer,
Config.defaultLayer,
SyncEvent.defaultLayer,
RuntimeFlags.layer({ experimentalEventSystem: true }),
EventV2Bridge.defaultLayer,
Database.defaultLayer,
EventV2Bridge.defaultLayer,
)
const env = Layer.mergeAll(
SessionNs.defaultLayer,
Database.defaultLayer,
EventV2Bridge.defaultLayer,
CrossSpawnSpawner.defaultLayer,
SessionCompaction.layer.pipe(Layer.provide(SessionNs.defaultLayer), Layer.provideMerge(deps)),
)
const it = testEffect(env)
const compactionEnv = Layer.mergeAll(SessionNs.defaultLayer, Database.defaultLayer, CrossSpawnSpawner.defaultLayer)
const compactionEnv = Layer.mergeAll(SessionNs.defaultLayer, Database.defaultLayer, EventV2Bridge.defaultLayer, CrossSpawnSpawner.defaultLayer)
const itCompaction = testEffect(compactionEnv)
type CompactionProcessOptions = {
@@ -286,7 +285,6 @@ function compactionProcessLayer(options?: CompactionProcessOptions) {
Layer.provide(status),
Layer.provide(bus),
Layer.provide(options?.config ?? Config.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(RuntimeFlags.layer({ experimentalEventSystem: true })),
Layer.provide(EventV2Bridge.defaultLayer),
)
packages/opencode/test/session/processor-effect.test.ts
+2 -4
View File
@@ -1,6 +1,7 @@
import { NodeFileSystem } from "@effect/platform-node"
import { SessionLegacy } from "@opencode-ai/core/session/legacy"
import { Database } from "@opencode-ai/core/database/database"
import { EventV2Bridge } from "@/event-v2-bridge"
import { expect } from "bun:test"
import { tool } from "ai"
import { Cause, Effect, Exit, Fiber, Layer } from "effect"
@@ -28,9 +29,7 @@ import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { provideTmpdirServer } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
import { raw, reply, TestLLMServer } from "../lib/llm-server"
import { SyncEvent } from "@/sync"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { EventV2Bridge } from "@/event-v2-bridge"
import { ProviderV2 } from "@opencode-ai/core/provider"
void Log.init({ print: false })
@@ -185,9 +184,8 @@ const deps = Layer.mergeAll(
LLM.defaultLayer,
Provider.defaultLayer,
status,
SyncEvent.defaultLayer,
EventV2Bridge.defaultLayer,
Database.defaultLayer,
EventV2Bridge.defaultLayer,
).pipe(Layer.provideMerge(infra))
const env = Layer.mergeAll(
TestLLMServer.layer,
packages/opencode/test/session/prompt.test.ts
+6 -9
View File
@@ -1,6 +1,8 @@
import { NodeFileSystem } from "@effect/platform-node"
import { SessionLegacy } from "@opencode-ai/core/session/legacy"
import { Database as CoreDatabase } from "@opencode-ai/core/database/database"
import { Database } from "@opencode-ai/core/database/database"
import { eq } from "drizzle-orm"
import { EventV2Bridge } from "@/event-v2-bridge"
import { FetchHttpClient } from "effect/unstable/http"
import { expect } from "bun:test"
import { Cause, Deferred, Duration, Effect, Exit, Fiber, Layer } from "effect"
@@ -46,7 +48,6 @@ import { ToolRegistry } from "@/tool/registry"
import { Truncate } from "@/tool/truncate"
import * as Log from "@opencode-ai/core/util/log"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import * as Database from "../../src/storage/db"
import { Ripgrep } from "../../src/file/ripgrep"
import { Format } from "../../src/format"
import { Reference } from "../../src/reference/reference"
@@ -54,9 +55,7 @@ import { RepositoryCache } from "../../src/reference/repository-cache"
import { TestInstance } from "../fixture/fixture"
import { awaitWithTimeout, pollWithTimeout, testEffect } from "../lib/effect"
import { reply, TestLLMServer } from "../lib/llm-server"
import { SyncEvent } from "@/sync"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { EventV2Bridge } from "@/event-v2-bridge"
import { ProviderV2 } from "@opencode-ai/core/provider"
void Log.init({ print: false })
@@ -184,9 +183,8 @@ function makePrompt(input?: { processor?: "blocking" }) {
AppFileSystem.defaultLayer,
BackgroundJob.defaultLayer,
status,
SyncEvent.defaultLayer,
Database.defaultLayer,
EventV2Bridge.defaultLayer,
CoreDatabase.defaultLayer,
).pipe(Layer.provideMerge(infra))
const question = Question.layer.pipe(Layer.provideMerge(deps))
const todo = Todo.layer.pipe(Layer.provideMerge(deps))
@@ -512,9 +510,8 @@ noLLMServer.instance(
const messages = yield* SessionV2.Service.use((session) => session.messages({ sessionID: chat.id })).pipe(
Effect.provide(SessionV2.defaultLayer),
)
const row = Database.use((db) =>
db.select().from(SessionMessageTable).where(Database.eq(SessionMessageTable.session_id, chat.id)).get(),
)
const { db } = yield* Database.Service
const row = yield* db.select().from(SessionMessageTable).where(eq(SessionMessageTable.session_id, chat.id)).get().pipe(Effect.orDie)
expect(messages.find((message) => message.type === "user")).toMatchObject({ type: "user", text: "hello v2" })
expect(typeof row?.data.time.created).toBe("number")
expect(messages).toEqual(
packages/opencode/test/session/session.test.ts
+2 -4
View File
@@ -1,7 +1,6 @@
import { describe, expect } from "bun:test"
import { SessionLegacy } from "@opencode-ai/core/session/legacy"
import { Database } from "@opencode-ai/core/database/database"
import { EventV2 } from "@opencode-ai/core/event"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { Deferred, Effect, Exit, Layer } from "effect"
import { Session as SessionNs } from "@/session/session"
@@ -14,9 +13,9 @@ import { provideInstance, tmpdirScoped } from "../fixture/fixture"
import { testEffect } from "../lib/effect"
import { Bus } from "@/bus"
import { Storage } from "@/storage/storage"
import { SyncEvent } from "@/sync"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { BackgroundJob } from "@/background/job"
import { EventV2Bridge } from "@/event-v2-bridge"
void Log.init({ print: false })
@@ -25,9 +24,8 @@ const it = testEffect(
SessionNs.layer.pipe(
Layer.provide(Bus.layer),
Layer.provide(Storage.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2.defaultLayer),
Layer.provide(EventV2Bridge.defaultLayer),
Layer.provide(SessionProjector.defaultLayer),
Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: false })),
Layer.provide(BackgroundJob.defaultLayer),
packages/opencode/test/session/snapshot-tool-race.test.ts
+2 -4
View File
@@ -31,6 +31,7 @@ import { TestLLMServer } from "../lib/llm-server"
// Same layer setup as prompt-effect.test.ts
import { NodeFileSystem } from "@effect/platform-node"
import { Database } from "@opencode-ai/core/database/database"
import { EventV2Bridge } from "@/event-v2-bridge"
import { Agent as AgentSvc } from "../../src/agent/agent"
import { BackgroundJob } from "@/background/job"
import { Git } from "../../src/git"
@@ -62,9 +63,7 @@ import { Ripgrep } from "../../src/file/ripgrep"
import { Format } from "../../src/format"
import { Reference } from "../../src/reference/reference"
import { RepositoryCache } from "../../src/reference/repository-cache"
import { SyncEvent } from "@/sync"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { EventV2Bridge } from "@/event-v2-bridge"
void Log.init({ print: false })
@@ -132,9 +131,8 @@ function makeHttp() {
AppFileSystem.defaultLayer,
BackgroundJob.defaultLayer,
status,
SyncEvent.defaultLayer,
EventV2Bridge.defaultLayer,
Database.defaultLayer,
EventV2Bridge.defaultLayer,
).pipe(Layer.provideMerge(infra))
const question = Question.layer.pipe(Layer.provideMerge(deps))
const todo = Todo.layer.pipe(Layer.provideMerge(deps))
packages/opencode/test/storage/db.test.ts
-9
View File
@@ -1,9 +0,0 @@
import { describe, expect, it } from "bun:test"
import { Database as CoreDatabase } from "@opencode-ai/core/database/database"
import { Database } from "@/storage/db"
describe("Database.getPath", () => {
it("delegates to the core database path", () => {
expect(Database.getPath()).toBe(CoreDatabase.path())
})
})
packages/opencode/test/sync/index.test.ts
-391
View File
@@ -1,391 +0,0 @@
import { describe, expect, beforeEach, afterAll } from "bun:test"
import { provideTmpdirInstance } from "../fixture/fixture"
import { Deferred, Effect, Layer, Schema } from "effect"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { Bus } from "../../src/bus"
import { GlobalBus, type GlobalEvent } from "../../src/bus/global"
import { SyncEvent } from "../../src/sync"
import { Database, eq } from "@/storage/db"
import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
import { EventV2 } from "@opencode-ai/core/event"
import { MessageID } from "../../src/session/schema"
import { initProjectors } from "../../src/server/projectors"
import { awaitWithTimeout, testEffect } from "../lib/effect"
import { RuntimeFlags } from "@/effect/runtime-flags"
const it = testEffect(
Layer.mergeAll(
SyncEvent.layer.pipe(
Layer.provide(RuntimeFlags.layer({ experimentalWorkspaces: true })),
Layer.provideMerge(Bus.layer),
),
CrossSpawnSpawner.defaultLayer,
),
)
beforeEach(() => {
Database.close()
})
describe("SyncEvent", () => {
function setup() {
SyncEvent.reset()
const Created = SyncEvent.define({
type: "item.created",
version: 1,
aggregate: "id",
schema: Schema.Struct({ id: Schema.String, name: Schema.String }),
})
const Sent = SyncEvent.define({
type: "item.sent",
version: 1,
aggregate: "item_id",
schema: Schema.Struct({ item_id: Schema.String, to: Schema.String }),
})
SyncEvent.init({
projectors: [SyncEvent.project(Created, () => {}), SyncEvent.project(Sent, () => {})],
})
return { Created, Sent }
}
function expectDefect<A, E, R>(effect: Effect.Effect<A, E, R>, pattern: RegExp) {
return Effect.gen(function* () {
const exit = yield* Effect.exit(effect)
if (exit._tag === "Success") throw new Error("Expected effect to fail")
expect(String(exit.cause)).toMatch(pattern)
})
}
afterAll(() => {
SyncEvent.reset()
initProjectors()
})
describe("run", () => {
it.live(
"inserts event row",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
yield* SyncEvent.use.run(Created, { id: "evt_1", name: "first" })
const rows = Database.use((db) => db.select().from(EventTable).all())
expect(rows).toHaveLength(1)
expect(rows[0].type).toBe("item.created.1")
expect(rows[0].aggregate_id).toBe("evt_1")
}),
),
)
it.live(
"increments seq per aggregate",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
yield* SyncEvent.use.run(Created, { id: "evt_1", name: "first" })
yield* SyncEvent.use.run(Created, { id: "evt_1", name: "second" })
const rows = Database.use((db) => db.select().from(EventTable).all())
expect(rows).toHaveLength(2)
expect(rows[1].seq).toBe(rows[0].seq + 1)
}),
),
)
it.live(
"uses custom aggregate field from agg()",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Sent } = setup()
yield* SyncEvent.use.run(Sent, { item_id: "evt_1", to: "james" })
const rows = Database.use((db) => db.select().from(EventTable).all())
expect(rows).toHaveLength(1)
expect(rows[0].aggregate_id).toBe("evt_1")
}),
),
)
it.live(
"emits events",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
const events: Array<{
type: string
properties: { id: string; name: string }
}> = []
let resolve = () => {}
const received = new Promise<void>((done) => {
resolve = done
})
const bus = yield* Bus.Service
const dispose = yield* bus.subscribeAllCallback((event) => {
events.push(event)
resolve()
})
try {
yield* SyncEvent.use.run(Created, { id: "evt_1", name: "test" })
yield* Effect.promise(() => received)
expect(events).toHaveLength(1)
expect(events[0]).toMatchObject({
type: "item.created",
properties: {
id: "evt_1",
name: "test",
},
})
} finally {
dispose()
}
}),
),
)
// Regression for the EffectBridge migration. GlobalBus.emit used to fire
// synchronously inside the Database.effect post-commit callback. After the
// migration it fires inside the forked publish Effect, AFTER bus.publish
// completes. Consumers don't care about microsecond-level ordering, but
// we still need to prove the emit actually fires.
it.live(
"emits sync events to GlobalBus after publishing to ProjectBus",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
// Filter for OUR specific event in the handler so we ignore any
// stray sync events from other tests' lingering forks.
const received = yield* Deferred.make<GlobalEvent>()
const handler = (evt: GlobalEvent) => {
if (evt.payload?.type === "sync" && evt.payload?.syncEvent?.type === "item.created.1") {
Deferred.doneUnsafe(received, Effect.succeed(evt))
}
}
GlobalBus.on("event", handler)
try {
yield* SyncEvent.use.run(Created, { id: "evt_global_1", name: "global" })
const event = yield* awaitWithTimeout(
Deferred.await(received),
"timed out waiting for sync event on GlobalBus",
"2 seconds",
)
expect(event.payload).toMatchObject({
type: "sync",
syncEvent: { type: "item.created.1", data: { id: "evt_global_1", name: "global" } },
})
} finally {
GlobalBus.off("event", handler)
}
}),
),
)
})
describe("replay", () => {
it.live(
"inserts event from external payload",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const id = MessageID.ascending()
yield* SyncEvent.use.replay({
id: "evt_1",
type: "item.created.1",
seq: 0,
aggregateID: id,
data: { id, name: "replayed" },
})
const rows = Database.use((db) => db.select().from(EventTable).all())
expect(rows).toHaveLength(1)
expect(rows[0].aggregate_id).toBe(id)
}),
),
)
it.live(
"throws on sequence mismatch",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const id = MessageID.ascending()
yield* SyncEvent.use.replay({
id: "evt_1",
type: "item.created.1",
seq: 0,
aggregateID: id,
data: { id, name: "first" },
})
yield* expectDefect(
SyncEvent.use.replay({
id: "evt_1",
type: "item.created.1",
seq: 5,
aggregateID: id,
data: { id, name: "bad" },
}),
/Sequence mismatch/,
)
}),
),
)
it.live(
"throws on unknown event type",
provideTmpdirInstance(() =>
Effect.gen(function* () {
yield* expectDefect(
SyncEvent.use.replay({
id: "evt_1",
type: "unknown.event.1",
seq: 0,
aggregateID: "x",
data: {},
}),
/Unknown event type/,
)
}),
),
)
it.live(
"replayAll accepts later chunks after the first batch",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
const id = MessageID.ascending()
const one = yield* SyncEvent.use.replayAll([
{
id: "evt_1",
type: SyncEvent.versionedType(Created.type, Created.version),
seq: 0,
aggregateID: id,
data: { id, name: "first" },
},
{
id: "evt_2",
type: SyncEvent.versionedType(Created.type, Created.version),
seq: 1,
aggregateID: id,
data: { id, name: "second" },
},
])
const two = yield* SyncEvent.use.replayAll([
{
id: "evt_3",
type: SyncEvent.versionedType(Created.type, Created.version),
seq: 2,
aggregateID: id,
data: { id, name: "third" },
},
{
id: "evt_4",
type: SyncEvent.versionedType(Created.type, Created.version),
seq: 3,
aggregateID: id,
data: { id, name: "fourth" },
},
])
expect(one).toBe(id)
expect(two).toBe(id)
const rows = Database.use((db) => db.select().from(EventTable).all())
expect(rows.map((row) => row.seq)).toEqual([0, 1, 2, 3])
}),
),
)
it.live(
"claims unowned event sequence on replay with ownerID",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
const id = MessageID.ascending()
yield* SyncEvent.use.replay(
{
id: "evt_1",
type: SyncEvent.versionedType(Created.type, Created.version),
seq: 0,
aggregateID: id,
data: { id, name: "owned" },
},
{ publish: false, ownerID: "owner-1" },
)
const row = Database.use((db) =>
db
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
.from(EventSequenceTable)
.get(),
)
expect(row).toEqual({ seq: 0, ownerID: "owner-1" })
}),
),
)
it.live(
"ignores replay from a different owner after sequence is claimed",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
const id = MessageID.ascending()
yield* SyncEvent.use.replay(
{
id: "evt_1",
type: SyncEvent.versionedType(Created.type, Created.version),
seq: 0,
aggregateID: id,
data: { id, name: "first" },
},
{ publish: false, ownerID: "owner-1" },
)
yield* SyncEvent.use.replay(
{
id: "evt_2",
type: SyncEvent.versionedType(Created.type, Created.version),
seq: 1,
aggregateID: id,
data: { id, name: "ignored" },
},
{ publish: false, ownerID: "owner-2" },
)
const events = Database.use((db) => db.select().from(EventTable).all())
const sequence = Database.use((db) =>
db
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
.from(EventSequenceTable)
.get(),
)
expect(events).toHaveLength(1)
expect(events[0]?.id).toBe(EventV2.ID.make("evt_1"))
expect(sequence).toEqual({ seq: 0, ownerID: "owner-1" })
}),
),
)
it.live(
"claim updates the event sequence owner",
provideTmpdirInstance(() =>
Effect.gen(function* () {
const { Created } = setup()
const id = MessageID.ascending()
yield* SyncEvent.use.run(Created, { id, name: "claimed" }, { publish: false })
yield* SyncEvent.use.claim(id, "owner-1")
yield* SyncEvent.use.claim(id, "owner-2")
const row = Database.use((db) =>
db
.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
.from(EventSequenceTable)
.where(eq(EventSequenceTable.aggregate_id, id))
.get(),
)
expect(row).toEqual({ seq: 0, ownerID: "owner-2" })
}),
),
)
})
})
specs/storage/remove-opencode-db.md
+17 -33
View File
@@ -25,7 +25,6 @@ Production imports from `packages/opencode/src/storage/db.ts` are concentrated i
- `packages/opencode/src/cli/cmd/import.ts`
- `packages/opencode/src/cli/cmd/stats.ts`
- `packages/opencode/src/control-plane/workspace.ts`
- `packages/opencode/src/data-migration.ts`
- `packages/opencode/src/index.ts`
- `packages/opencode/src/node.ts`
- `packages/opencode/src/permission/index.ts`
@@ -47,6 +46,8 @@ There are 65 direct API/type references in those files. The references fall into
## Group 1: Database Runtime And Startup
Status: Completed. Startup, the public node export, and database CLI tooling no longer import the legacy opencode database wrapper; `packages/opencode/src/storage/db.ts` has been deleted.
Files:
- `packages/opencode/src/storage/db.ts`
@@ -75,6 +76,8 @@ Target shape:
## Group 2: Sync Event Transaction Boundary
Status: Completed. `SyncEvent` and the opencode projector boundary were removed; session/message event projection now lives in core EventV2/projector infrastructure.
Files:
- `packages/opencode/src/sync/index.ts`
@@ -106,6 +109,8 @@ Suggested first step:
## Group 3: Domain Repositories Already Behind Services
Status: Completed. These services no longer import the legacy opencode database wrapper.
Files:
- `packages/opencode/src/account/repo.ts`
@@ -139,6 +144,8 @@ Suggested order:
## Group 4: Session And Message Read Models
Status: Completed. Session/message reads and projector writes have moved off the legacy opencode database wrapper.
Files:
- `packages/opencode/src/session/session.ts`
@@ -176,6 +183,8 @@ Suggested order:
## Group 5: Legacy CLI And One-Off Admin Reads
Status: Completed. Remaining one-off CLI/admin reads and writes now use core database services or domain services instead of the legacy opencode database wrapper.
Files:
- `packages/opencode/src/cli/cmd/import.ts`
@@ -206,46 +215,21 @@ Target shape:
- For HTTP sync reads, move the event log query behind the sync event module.
- For permission and worktree reads, call the permission/project services if available; otherwise add narrow repository methods.
## Group 6: Data Migrations
Files:
- `packages/opencode/src/data-migration.ts`
Current usage:
- Checks `DataMigrationTable` with `Database.use`.
- Runs resumable data migrations with `Database.use` and `Database.transaction`.
- Writes completion rows with `Database.use`.
Why this group is separate:
- Data migrations are database-native by definition and may reasonably stay close to SQL.
- They should not depend on the legacy opencode wrapper, but they do need a stable transaction API and migration completion store.
Target shape:
- Run data migrations through the same Effect database service used by startup/migrations.
- Keep SQL-heavy migration logic local to `data-migration.ts`, but remove callback-style legacy access.
- Ensure migrations still run in a scoped/background fiber and remain resumable.
## Recommended Migration Sequence
1. Replace the legacy runtime seam from Group 1 with an Effect-native database module that exposes path, client/query access, transaction, and after-commit behavior.
2. Port Group 2 sync event transaction semantics to the new module before touching projector bodies.
3. Migrate Group 3 repositories that already hide database access behind service interfaces.
4. Migrate Group 4 session/message reads, then projector write helpers once the new projector transaction type exists.
5. Migrate Group 6 data migrations onto the new database service.
6. Clean up Group 5 one-off reads and CLI/admin commands.
7. Remove drizzle helper re-exports from `@/storage/db` imports by importing operators directly from `drizzle-orm` during each file migration.
8. Delete `packages/opencode/src/storage/db.ts` once `rg "@/storage/db|./storage/db|Database\." packages/opencode/src` no longer finds legacy usages.
All migration groups are complete or superseded. `packages/opencode/src/storage/db.ts` has been deleted.
## Superseded: Data Migrations
Status: Superseded. No opencode data-migration group remains.
The previous opencode `data-migration.ts` service only backfilled session usage from message rows. That work is now covered by core database migration `packages/core/src/database/migration/20260510033149_session_usage.ts`, so there is no separate opencode data-migration group.
## Invariants To Preserve
- Nested reads inside a transaction must use the active transaction, not the root client.
- `SyncEvent.run` sequence allocation must keep immediate transaction behavior.
- Post-commit publish effects must not run before the transaction commits.
- Data migrations must remain resumable and record completion only after successful migration work.
- Existing schema ownership remains in `packages/core/src/**/*.sql.ts`; do not move table definitions back into `packages/opencode`.
## Verification Commands