mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-12 15:03:24 +00:00
970 lines
38 KiB
TypeScript
970 lines
38 KiB
TypeScript
export * as V1Migration from "./v1-migration"
|
|
|
|
import { Cause, Effect, Layer, Option, Schema, Semaphore } from "effect"
|
|
import { Database } from "./database"
|
|
import { SessionMessageTable, SessionTable } from "../session/sql"
|
|
import { SessionV1 } from "@opencode-ai/schema/session-v1"
|
|
import { SessionMessage } from "../session/message"
|
|
import { SessionSchema } from "../session/schema"
|
|
import { KVTable } from "../kv/sql"
|
|
import { EventSequenceTable, EventTable } from "../event/sql"
|
|
import { eq, sql } from "drizzle-orm"
|
|
import { Global } from "@opencode-ai/util/global"
|
|
import { existsSync } from "node:fs"
|
|
import path from "node:path"
|
|
import type { Database as SQLiteDatabase } from "bun:sqlite"
|
|
import { Project } from "@opencode-ai/schema/project"
|
|
|
|
export type SourceMessage = {
|
|
readonly id: string
|
|
readonly session_id: string
|
|
readonly time_created: number
|
|
readonly time_updated: number
|
|
readonly data: string
|
|
}
|
|
|
|
export type SourcePart = {
|
|
readonly id: string
|
|
readonly message_id: string
|
|
readonly session_id: string
|
|
readonly time_created: number
|
|
readonly time_updated: number
|
|
readonly data: string
|
|
}
|
|
|
|
export type TransformInput = {
|
|
readonly session: typeof SessionTable.$inferSelect
|
|
readonly messages: ReadonlyArray<SourceMessage>
|
|
readonly parts: ReadonlyArray<SourcePart>
|
|
}
|
|
|
|
export type Warning = {
|
|
readonly reason: string
|
|
readonly sessionID: string
|
|
readonly messageID?: string
|
|
readonly partID?: string
|
|
readonly observedType?: string
|
|
}
|
|
|
|
export type TransformResult = {
|
|
readonly messages: ReadonlyArray<{
|
|
readonly id: string
|
|
readonly session_id: string
|
|
readonly type: SessionMessage.Type
|
|
readonly seq: number
|
|
readonly time_created: number
|
|
readonly time_updated: number
|
|
readonly data: Record<string, unknown>
|
|
}>
|
|
readonly session: Pick<
|
|
typeof SessionTable.$inferInsert,
|
|
| "agent"
|
|
| "model"
|
|
| "cost"
|
|
| "tokens_input"
|
|
| "tokens_output"
|
|
| "tokens_reasoning"
|
|
| "tokens_cache_read"
|
|
| "tokens_cache_write"
|
|
| "revert"
|
|
| "time_compacting"
|
|
>
|
|
readonly watermark: number
|
|
readonly warnings: ReadonlyArray<Warning>
|
|
}
|
|
|
|
type Progress = {
|
|
readonly label: string
|
|
readonly numerator?: number
|
|
readonly denominator?: number
|
|
}
|
|
|
|
export type Status =
|
|
| { readonly status: "required" | "completed" }
|
|
| { readonly status: "running"; readonly progress: Progress }
|
|
| { readonly status: "error"; readonly error: string }
|
|
|
|
type RunResult = {
|
|
readonly status: "completed"
|
|
}
|
|
|
|
type Options = {
|
|
readonly nextDatabasePath?: string
|
|
}
|
|
|
|
type MigrationState =
|
|
| { readonly phase: "sessions"; readonly cursor?: string }
|
|
| { readonly phase: "completed" }
|
|
|
|
type RuntimeState =
|
|
| { readonly status: "idle" }
|
|
| { readonly status: "running"; readonly progress: Progress }
|
|
| { readonly status: "error"; readonly error: string }
|
|
|
|
type NextProject = {
|
|
readonly id: string
|
|
readonly worktree: string
|
|
readonly vcs: string | null
|
|
readonly name: string | null
|
|
readonly icon_url: string | null
|
|
readonly icon_url_override: string | null
|
|
readonly icon_color: string | null
|
|
readonly time_created: number
|
|
readonly time_updated: number
|
|
readonly time_initialized: number | null
|
|
readonly sandboxes: string
|
|
readonly commands: string | null
|
|
}
|
|
|
|
type NextSession = {
|
|
readonly id: string
|
|
readonly project_id: string
|
|
readonly workspace_id: string | null
|
|
readonly parent_id: string | null
|
|
readonly fork_session_id: string | null
|
|
readonly fork_boundary: string | null
|
|
readonly slug: string
|
|
readonly directory: string
|
|
readonly path: string | null
|
|
readonly title: string | null
|
|
readonly version: string
|
|
readonly share_url: string | null
|
|
readonly summary_additions: number | null
|
|
readonly summary_deletions: number | null
|
|
readonly summary_files: number | null
|
|
readonly summary_diffs: string | null
|
|
readonly metadata: string | null
|
|
readonly cost: number
|
|
readonly tokens_input: number
|
|
readonly tokens_output: number
|
|
readonly tokens_reasoning: number
|
|
readonly tokens_cache_read: number
|
|
readonly tokens_cache_write: number
|
|
readonly revert: string | null
|
|
readonly permission: string | null
|
|
readonly agent: string | null
|
|
readonly model: string | null
|
|
readonly time_created: number
|
|
readonly time_updated: number
|
|
readonly time_compacting: number | null
|
|
readonly time_archived: number | null
|
|
readonly time_suspended: number | null
|
|
}
|
|
|
|
type NextMessage = {
|
|
readonly id: string
|
|
readonly session_id: string
|
|
readonly type: string
|
|
readonly seq: number
|
|
readonly time_created: number
|
|
readonly time_updated: number
|
|
readonly data: string
|
|
}
|
|
|
|
const lock = Semaphore.makeUnsafe(1)
|
|
const MIGRATION_STATE_KEY = "migration.v1-v2"
|
|
const decodeJson = Schema.decodeUnknownOption(Schema.UnknownFromJsonString)
|
|
const decodeMessage = Schema.decodeUnknownOption(SessionV1.Info)
|
|
const decodePart = Schema.decodeUnknownOption(SessionV1.Part)
|
|
let runtimeState: RuntimeState = { status: "idle" }
|
|
|
|
export function transformSession(input: TransformInput): TransformResult {
|
|
const warnings: Warning[] = []
|
|
const messages = input.messages
|
|
.map((row) => {
|
|
const value = Option.getOrUndefined(decodeJson(row.data))
|
|
const decoded =
|
|
value && typeof value === "object"
|
|
? Option.getOrUndefined(decodeMessage({ ...value, id: row.id, sessionID: row.session_id }))
|
|
: undefined
|
|
if (decoded) return { row, value: decoded }
|
|
warnings.push({ reason: "invalid-message", sessionID: input.session.id, messageID: row.id })
|
|
return undefined
|
|
})
|
|
.filter((item): item is NonNullable<typeof item> => item !== undefined)
|
|
.sort((a, b) => a.row.time_created - b.row.time_created || a.row.id.localeCompare(b.row.id))
|
|
const messageIDs = new Set(input.messages.map((row) => row.id))
|
|
const parts = input.parts
|
|
.map((row) => {
|
|
const value = Option.getOrUndefined(decodeJson(row.data))
|
|
const observedType = value && typeof value === "object" && "type" in value ? String(value.type) : undefined
|
|
if (!messageIDs.has(row.message_id)) {
|
|
warnings.push({
|
|
reason: "orphan-part",
|
|
sessionID: input.session.id,
|
|
messageID: row.message_id,
|
|
partID: row.id,
|
|
observedType,
|
|
})
|
|
return undefined
|
|
}
|
|
const decoded =
|
|
value && typeof value === "object"
|
|
? Option.getOrUndefined(
|
|
decodePart({ ...value, id: row.id, messageID: row.message_id, sessionID: row.session_id }),
|
|
)
|
|
: undefined
|
|
if (decoded) return { row, value: decoded }
|
|
warnings.push({
|
|
reason: "invalid-part",
|
|
sessionID: input.session.id,
|
|
messageID: row.message_id,
|
|
partID: row.id,
|
|
observedType,
|
|
})
|
|
return undefined
|
|
})
|
|
.filter((item): item is NonNullable<typeof item> => item !== undefined)
|
|
.sort((a, b) => a.row.id.localeCompare(b.row.id))
|
|
const byMessage = Map.groupBy(parts, (item) => item.row.message_id)
|
|
const paired = new Set<string>()
|
|
const used = new Set(messages.map((item) => item.row.id))
|
|
const projected = messages
|
|
.flatMap((item) => {
|
|
if (paired.has(item.row.id)) return []
|
|
const owned = byMessage.get(item.row.id)?.map((part) => part.value) ?? []
|
|
if (item.value.role === "user") {
|
|
const compaction = owned.find((part) => part.type === "compaction")
|
|
if (compaction?.type === "compaction") {
|
|
const pairedSummary = messages.find(
|
|
(candidate) =>
|
|
candidate.value.role === "assistant" &&
|
|
candidate.value.parentID === item.row.id &&
|
|
candidate.value.summary,
|
|
)
|
|
if (!pairedSummary || pairedSummary.value.role !== "assistant") return []
|
|
paired.add(pairedSummary.row.id)
|
|
if (pairedSummary.value.error || pairedSummary.value.time.completed === undefined) return []
|
|
const summary = pairedSummary
|
|
const summaryText = (byMessage.get(summary.row.id) ?? [])
|
|
.map((part) => part.value)
|
|
.filter((part) => part.type === "text" && part.text.length > 0)
|
|
.map((part) => (part.type === "text" ? part.text : ""))
|
|
.join("\n\n")
|
|
const tailIndex = compaction.tail_start_id
|
|
? messages.findIndex((candidate) => candidate.row.id === compaction.tail_start_id)
|
|
: -1
|
|
const compactionIndex = messages.findIndex((candidate) => candidate.row.id === item.row.id)
|
|
const tail = tailIndex < 0 ? [] : messages.slice(tailIndex, compactionIndex)
|
|
return [
|
|
row(
|
|
{ ...item.row, time_updated: Math.max(item.row.time_updated, summary.row.time_updated) },
|
|
{
|
|
id: item.row.id,
|
|
type: "compaction",
|
|
status: "completed",
|
|
reason: compaction.auto ? "auto" : "manual",
|
|
summary: summaryText,
|
|
recent: serializeRecent(tail, byMessage),
|
|
time: { created: item.row.time_created },
|
|
},
|
|
),
|
|
]
|
|
}
|
|
const subtasks = owned.filter((part) => part.type === "subtask")
|
|
const visible = owned.filter((part) => part.type === "text" && !part.ignored)
|
|
const files = owned.filter((part) => part.type === "file")
|
|
const agents = owned.filter((part) => part.type === "agent")
|
|
if (subtasks.length > 0 && visible.length === 0 && files.length === 0 && agents.length === 0) return []
|
|
const ordinary = visible.filter((part) => part.type === "text" && !part.synthetic)
|
|
const synthetic = visible.filter((part) => part.type === "text" && part.synthetic)
|
|
const attachments = files.flatMap((part) => (part.type === "file" ? migrateFile(part) : []))
|
|
const unavailable = files.flatMap((part) =>
|
|
part.type === "file" && !part.url.startsWith("data:") ? [unavailableFile(part)] : [],
|
|
)
|
|
const text = owned
|
|
.flatMap((part) => {
|
|
if (part.type === "text" && !part.ignored && !part.synthetic) return [part.text]
|
|
if (part.type === "file" && !part.url.startsWith("data:")) return [unavailableFile(part)]
|
|
return []
|
|
})
|
|
.join("\n\n")
|
|
const agentAttachments = agents.map((part) =>
|
|
part.type === "agent"
|
|
? {
|
|
name: part.name,
|
|
...(part.source
|
|
? { mention: { text: part.source.value, start: part.source.start, end: part.source.end } }
|
|
: {}),
|
|
}
|
|
: { name: "" },
|
|
)
|
|
if (
|
|
ordinary.length === 0 &&
|
|
unavailable.length === 0 &&
|
|
synthetic.length > 0 &&
|
|
attachments.length === 0 &&
|
|
agentAttachments.length === 0
|
|
)
|
|
return [
|
|
row(item.row, {
|
|
id: item.row.id,
|
|
type: "synthetic",
|
|
text: synthetic.map((part) => (part.type === "text" ? part.text : "")).join("\n\n"),
|
|
time: { created: item.row.time_created },
|
|
}),
|
|
]
|
|
const user = row(item.row, {
|
|
id: item.row.id,
|
|
type: "user",
|
|
text,
|
|
...(attachments.length ? { files: attachments } : {}),
|
|
...(agentAttachments.length ? { agents: agentAttachments } : {}),
|
|
time: { created: item.row.time_created },
|
|
})
|
|
if (synthetic.length === 0) return [user]
|
|
return [
|
|
user,
|
|
row(item.row, {
|
|
id: syntheticID(item.row.id, used),
|
|
type: "synthetic",
|
|
text: synthetic.map((part) => (part.type === "text" ? part.text : "")).join("\n\n"),
|
|
time: { created: item.row.time_created },
|
|
}),
|
|
]
|
|
}
|
|
if (item.value.role !== "assistant") return []
|
|
const assistant = item.value
|
|
const parent = messages.find((candidate) => candidate.row.id === assistant.parentID)
|
|
const parentParts = parent ? (byMessage.get(parent.row.id)?.map((part) => part.value) ?? []) : []
|
|
if (
|
|
parentParts.some((part) => part.type === "subtask") &&
|
|
owned.some((part) => part.type === "tool" && part.tool === "task")
|
|
)
|
|
return []
|
|
const content = owned.flatMap((part): Array<Record<string, unknown>> => {
|
|
if (part.type === "text")
|
|
return [{ type: "text", text: part.text, ...(part.metadata ? { state: part.metadata } : {}) }]
|
|
if (part.type === "reasoning")
|
|
return [
|
|
{
|
|
type: "reasoning",
|
|
text: part.text,
|
|
...(part.metadata ? { state: part.metadata } : {}),
|
|
time: { created: part.time.start, ...(part.time.end === undefined ? {} : { completed: part.time.end }) },
|
|
},
|
|
]
|
|
if (part.type !== "tool") return []
|
|
return [migrateTool(part, item.row.time_created)]
|
|
})
|
|
const start =
|
|
owned.flatMap((part) => (part.type === "step-start" && part.snapshot ? [part.snapshot] : []))[0] ??
|
|
owned.flatMap((part) => (part.type === "snapshot" ? [part.snapshot] : []))[0] ??
|
|
owned.flatMap((part) => (part.type === "patch" ? [part.hash] : []))[0]
|
|
const end = owned.flatMap((part) => (part.type === "step-finish" && part.snapshot ? [part.snapshot] : [])).at(-1)
|
|
const snapshotFiles = Array.from(new Set(owned.flatMap((part) => (part.type === "patch" ? part.files : []))))
|
|
const finish = normalizeFinish(assistant.finish)
|
|
return [
|
|
row(item.row, {
|
|
id: item.row.id,
|
|
type: "assistant",
|
|
agent: assistant.agent,
|
|
model: {
|
|
providerID: assistant.providerID,
|
|
id: assistant.modelID,
|
|
variant: assistant.variant ?? "default",
|
|
},
|
|
content,
|
|
...(start || end || snapshotFiles.length
|
|
? {
|
|
snapshot: {
|
|
...(start ? { start } : {}),
|
|
...(end ? { end } : {}),
|
|
...(snapshotFiles.length ? { files: snapshotFiles } : {}),
|
|
},
|
|
}
|
|
: {}),
|
|
...(finish ? { finish } : {}),
|
|
cost: assistant.cost,
|
|
tokens: {
|
|
input: assistant.tokens.input,
|
|
output: assistant.tokens.output,
|
|
reasoning: assistant.tokens.reasoning,
|
|
cache: assistant.tokens.cache,
|
|
},
|
|
...(assistant.error ? { error: migrateError(assistant.error) } : {}),
|
|
time: {
|
|
created: item.row.time_created,
|
|
...(assistant.time.completed === undefined ? {} : { completed: item.row.time_updated }),
|
|
},
|
|
}),
|
|
]
|
|
})
|
|
.map((item, seq) => ({ ...item, seq }))
|
|
const assistants = messages
|
|
.filter((item) => item.value.role === "assistant")
|
|
.map((item) => item.value)
|
|
.filter((item): item is SessionV1.Assistant => item.role === "assistant")
|
|
const latestUser = messages.findLast((item) => {
|
|
if (item.value.role !== "user") return false
|
|
const owned = byMessage.get(item.row.id) ?? []
|
|
if (owned.some((part) => part.value.type === "compaction")) return false
|
|
return !owned.some((part) => part.value.type === "subtask") || !owned.every((part) => part.value.type === "subtask")
|
|
})
|
|
return {
|
|
messages: projected,
|
|
session: {
|
|
agent: input.session.agent ?? (latestUser?.value.role === "user" ? latestUser.value.agent : null),
|
|
model:
|
|
input.session.model ??
|
|
(latestUser?.value.role === "user"
|
|
? {
|
|
id: latestUser.value.model.modelID,
|
|
providerID: latestUser.value.model.providerID,
|
|
variant: latestUser.value.model.variant ?? "default",
|
|
}
|
|
: null),
|
|
cost: assistants.reduce((total, item) => total + item.cost, 0),
|
|
tokens_input: assistants.reduce((total, item) => total + item.tokens.input, 0),
|
|
tokens_output: assistants.reduce((total, item) => total + item.tokens.output, 0),
|
|
tokens_reasoning: assistants.reduce((total, item) => total + item.tokens.reasoning, 0),
|
|
tokens_cache_read: assistants.reduce((total, item) => total + item.tokens.cache.read, 0),
|
|
tokens_cache_write: assistants.reduce((total, item) => total + item.tokens.cache.write, 0),
|
|
revert: null,
|
|
time_compacting: null,
|
|
},
|
|
watermark: projected.length - 1,
|
|
warnings,
|
|
}
|
|
}
|
|
|
|
export function status(): Effect.Effect<Status, never, Database.Service> {
|
|
return Effect.gen(function* () {
|
|
const { db } = yield* Database.Service
|
|
if (!(yield* hasLegacySessions(db))) return { status: "completed" as const }
|
|
const state = yield* readState(db)
|
|
if (runtimeState.status === "running") return runtimeState
|
|
if (runtimeState.status === "error") return runtimeState
|
|
if (state?.phase === "completed") return { status: "completed" as const }
|
|
return { status: "required" as const }
|
|
}).pipe(Effect.orDie)
|
|
}
|
|
|
|
export const layer = Layer.effectDiscard(
|
|
Effect.gen(function* () {
|
|
runtimeState = { status: "running", progress: { label: "Clearing old events" } }
|
|
yield* run().pipe(
|
|
Effect.matchCauseEffect({
|
|
onFailure: (cause) =>
|
|
Effect.sync(() => {
|
|
runtimeState = { status: "error", error: errorText(Cause.squash(cause)) }
|
|
}).pipe(Effect.andThen(Effect.logError("V1 migration failed", { cause }))),
|
|
onSuccess: () =>
|
|
Effect.sync(() => {
|
|
runtimeState = { status: "idle" }
|
|
}),
|
|
}),
|
|
Effect.forkScoped({ startImmediately: true }),
|
|
)
|
|
}),
|
|
)
|
|
|
|
function errorText(input: unknown): string {
|
|
if (!(input instanceof Error)) return String(input)
|
|
const cause = input.cause
|
|
return cause === undefined ? input.message : `${input.message}\nCaused by: ${errorText(cause)}`
|
|
}
|
|
|
|
function updateProgress(progress: Progress) {
|
|
if (runtimeState.status === "running") runtimeState = { status: "running", progress }
|
|
}
|
|
|
|
export function run(options: Options = {}): Effect.Effect<RunResult, never, Database.Service> {
|
|
return lock.withPermit(
|
|
Effect.gen(function* () {
|
|
const { db } = yield* Database.Service
|
|
const state = yield* readState(db)
|
|
if (state?.phase === "completed") return { status: "completed" as const }
|
|
if (!(yield* hasLegacySessions(db))) return { status: "completed" as const }
|
|
const migrate = Effect.gen(function* () {
|
|
const now = Date.now()
|
|
yield* db.run(sql`
|
|
INSERT OR IGNORE INTO project (id, worktree, time_created, time_updated, sandboxes)
|
|
VALUES (${Project.ID.global}, ${path.parse(Global.Path.data).root}, ${now}, ${now}, '[]')
|
|
`)
|
|
if (state === undefined)
|
|
yield* db
|
|
.transaction((tx) =>
|
|
Effect.gen(function* () {
|
|
yield* tx.delete(EventTable).run()
|
|
yield* tx.insert(KVTable).values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions" } }).run()
|
|
}),
|
|
)
|
|
.pipe(Effect.orDie)
|
|
const sourceTotal = yield* countNextSessions(nextPath(options))
|
|
const legacyTotal = (yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM session`))?.value ?? 0
|
|
const cursor = state?.phase === "sessions" ? state.cursor : undefined
|
|
const migrated =
|
|
cursor !== undefined
|
|
? ((yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM session WHERE id >= ${cursor}`))
|
|
?.value ?? 0)
|
|
: 0
|
|
const denominator = sourceTotal + legacyTotal
|
|
updateProgress({ label: "Migrating sessions", numerator: migrated, denominator })
|
|
yield* importNextDatabase(db, nextPath(options), (completed) => {
|
|
updateProgress({ label: "Migrating sessions", numerator: migrated + completed, denominator })
|
|
})
|
|
updateProgress({ label: "Migrating sessions", numerator: migrated + sourceTotal, denominator })
|
|
const projects = new Set((yield* db.all<{ id: string }>(sql`SELECT id FROM project`)).map((project) => project.id))
|
|
while (true) {
|
|
const state = yield* readState(db)
|
|
const cursorValue = state?.phase === "sessions" ? state.cursor : undefined
|
|
const nextID = yield* db.get<{ id: string; project_id: string }>(
|
|
cursorValue === undefined
|
|
? sql`SELECT id, project_id FROM session ORDER BY id DESC LIMIT 1`
|
|
: sql`SELECT id, project_id FROM session WHERE id < ${cursorValue} ORDER BY id DESC LIMIT 1`,
|
|
)
|
|
if (!nextID) break
|
|
yield* db
|
|
.transaction((tx) =>
|
|
Effect.gen(function* () {
|
|
yield* tx
|
|
.insert(KVTable)
|
|
.values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions", cursor: nextID.id } })
|
|
.onConflictDoUpdate({
|
|
target: KVTable.key,
|
|
set: { value: { phase: "sessions", cursor: nextID.id }, time_updated: Date.now() },
|
|
})
|
|
.run()
|
|
const projectID = projects.has(nextID.project_id) ? nextID.project_id : Project.ID.global
|
|
if (projectID !== nextID.project_id)
|
|
yield* Effect.logWarning("Reassigned V1 session with missing project", {
|
|
sessionID: nextID.id,
|
|
projectID: nextID.project_id,
|
|
})
|
|
yield* tx.run(sql`
|
|
INSERT OR IGNORE INTO session_v2 (
|
|
id, project_id, workspace_id, parent_id, slug, directory, path, title, version, share_url,
|
|
summary_additions, summary_deletions, summary_files, summary_diffs, metadata, cost,
|
|
tokens_input, tokens_output, tokens_reasoning, tokens_cache_read, tokens_cache_write,
|
|
revert, permission, agent, model, time_created, time_updated, time_compacting, time_archived
|
|
)
|
|
SELECT
|
|
id, ${projectID}, workspace_id, parent_id, slug, directory, path, title, version, share_url,
|
|
summary_additions, summary_deletions, summary_files, summary_diffs, metadata, cost,
|
|
tokens_input, tokens_output, tokens_reasoning, tokens_cache_read, tokens_cache_write,
|
|
revert, permission, agent, model, time_created, time_updated, time_compacting, time_archived
|
|
FROM session
|
|
WHERE id = ${nextID.id}
|
|
`)
|
|
const next = yield* tx
|
|
.select()
|
|
.from(SessionTable)
|
|
.where(eq(SessionTable.id, SessionSchema.ID.make(nextID.id)))
|
|
.get()
|
|
if (!next) return yield* Effect.die(new Error(`Failed to copy V1 session ${nextID.id}`))
|
|
const sourceMessages = yield* tx.all<SourceMessage>(
|
|
sql`SELECT id, session_id, time_created, time_updated, data FROM message WHERE session_id = ${next.id}`,
|
|
)
|
|
const sourceParts = yield* tx.all<SourcePart>(
|
|
sql`SELECT id, message_id, session_id, time_created, time_updated, data FROM part WHERE session_id = ${next.id}`,
|
|
)
|
|
const transformed = transformSession({ session: next, messages: sourceMessages, parts: sourceParts })
|
|
yield* Effect.forEach(transformed.warnings, (warning) =>
|
|
Effect.logWarning("Skipped V1 migration row", warning),
|
|
)
|
|
yield* tx.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, next.id)).run()
|
|
yield* Effect.forEach(transformed.messages, (message) =>
|
|
tx.run(sql`
|
|
INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data)
|
|
VALUES (${message.id}, ${message.session_id}, ${message.type}, ${message.seq}, ${message.time_created}, ${message.time_updated}, ${JSON.stringify(message.data)})
|
|
`),
|
|
)
|
|
yield* tx
|
|
.update(SessionTable)
|
|
.set({ ...transformed.session, time_updated: next.time_updated })
|
|
.where(eq(SessionTable.id, next.id))
|
|
.run()
|
|
yield* tx
|
|
.insert(EventSequenceTable)
|
|
.values({ aggregate_id: next.id, seq: transformed.watermark })
|
|
.onConflictDoUpdate({
|
|
target: EventSequenceTable.aggregate_id,
|
|
set: { seq: transformed.watermark, owner_id: null },
|
|
})
|
|
.run()
|
|
}),
|
|
)
|
|
.pipe(Effect.orDie)
|
|
if (runtimeState.status === "running")
|
|
runtimeState = {
|
|
status: "running",
|
|
progress: {
|
|
label: "Migrating sessions",
|
|
numerator: (runtimeState.progress.numerator ?? 0) + 1,
|
|
denominator,
|
|
},
|
|
}
|
|
yield* Effect.yieldNow
|
|
}
|
|
yield* db
|
|
.transaction((tx) =>
|
|
Effect.gen(function* () {
|
|
yield* tx
|
|
.insert(KVTable)
|
|
.values({ key: MIGRATION_STATE_KEY, value: { phase: "completed" } })
|
|
.onConflictDoUpdate({
|
|
target: KVTable.key,
|
|
set: { value: { phase: "completed" }, time_updated: Date.now() },
|
|
})
|
|
.run()
|
|
}),
|
|
)
|
|
.pipe(Effect.orDie)
|
|
return { status: "completed" as const }
|
|
})
|
|
return yield* migrate
|
|
}).pipe(Effect.orDie),
|
|
)
|
|
}
|
|
|
|
function nextPath(options: Options) {
|
|
if (options.nextDatabasePath) return options.nextDatabasePath
|
|
if (process.env.OPENCODE_DB === ":memory:") return undefined
|
|
return path.join(Global.Path.data, "opencode-next.db")
|
|
}
|
|
|
|
function openNextDatabase(sourcePath: string) {
|
|
return Effect.acquireRelease(
|
|
Effect.gen(function* () {
|
|
const sqlite = yield* Effect.promise(() => import("bun:sqlite"))
|
|
return new sqlite.Database(sourcePath, { readonly: true, strict: true })
|
|
}),
|
|
(source) => Effect.sync(() => source.close()),
|
|
)
|
|
}
|
|
|
|
function countNextSessions(sourcePath: string | undefined) {
|
|
if (!sourcePath || !existsSync(sourcePath)) return Effect.succeed(0)
|
|
return Effect.scoped(
|
|
Effect.gen(function* () {
|
|
const source = yield* openNextDatabase(sourcePath)
|
|
if (!isNextDatabase(source)) return 0
|
|
return source.query<{ value: number }, []>("SELECT COUNT(*) AS value FROM session").get()?.value ?? 0
|
|
}),
|
|
).pipe(Effect.orElseSucceed(() => 0))
|
|
}
|
|
|
|
function importNextDatabase(
|
|
db: Database.Interface["db"],
|
|
sourcePath: string | undefined,
|
|
onProgress: (completed: number) => void,
|
|
): Effect.Effect<void, unknown> {
|
|
if (!sourcePath || !existsSync(sourcePath)) return Effect.void
|
|
return Effect.scoped(
|
|
Effect.gen(function* () {
|
|
const source = yield* openNextDatabase(sourcePath)
|
|
if (!isNextDatabase(source)) {
|
|
yield* Effect.logWarning("Skipped incompatible opencode-next.db", { path: sourcePath })
|
|
return
|
|
}
|
|
source.run("BEGIN")
|
|
yield* Effect.addFinalizer(() =>
|
|
Effect.sync(() => {
|
|
if (source.inTransaction) source.run("ROLLBACK")
|
|
}),
|
|
)
|
|
const projects = new Map(
|
|
source
|
|
.query<NextProject, []>("SELECT * FROM project")
|
|
.all()
|
|
.map((project) => [project.id, project]),
|
|
)
|
|
const sessions = source.query<NextSession, []>("SELECT * FROM session ORDER BY id DESC").all()
|
|
for (const [index, session] of sessions.entries()) {
|
|
const project = projects.get(session.project_id)
|
|
const projectID = project ? session.project_id : Project.ID.global
|
|
if (!project) {
|
|
yield* Effect.logWarning("Reassigned previous V2 session with missing project", {
|
|
sessionID: session.id,
|
|
projectID: session.project_id,
|
|
})
|
|
}
|
|
const messages = source
|
|
.query<NextMessage, [string]>(
|
|
"SELECT id, session_id, type, seq, time_created, time_updated, data FROM session_message WHERE session_id = ? ORDER BY seq",
|
|
)
|
|
.all(session.id)
|
|
yield* db
|
|
.transaction((tx) =>
|
|
Effect.gen(function* () {
|
|
if (project)
|
|
yield* tx.run(sql`
|
|
INSERT OR IGNORE INTO project (
|
|
id, worktree, vcs, name, icon_url, icon_url_override, icon_color,
|
|
time_created, time_updated, time_initialized, sandboxes, commands
|
|
) VALUES (
|
|
${project.id}, ${project.worktree}, ${project.vcs}, ${project.name}, ${project.icon_url},
|
|
${project.icon_url_override}, ${project.icon_color}, ${project.time_created}, ${project.time_updated},
|
|
${project.time_initialized}, ${project.sandboxes}, ${project.commands}
|
|
)
|
|
`)
|
|
const existing = yield* tx
|
|
.select({ id: SessionTable.id })
|
|
.from(SessionTable)
|
|
.where(eq(SessionTable.id, SessionSchema.ID.make(session.id)))
|
|
.get()
|
|
if (existing) return
|
|
yield* tx.run(sql`
|
|
INSERT INTO session_v2 (
|
|
id, project_id, workspace_id, parent_id, fork_session_id, fork_boundary, slug, directory,
|
|
path, title, version, share_url, summary_additions, summary_deletions, summary_files,
|
|
summary_diffs, metadata, cost, tokens_input, tokens_output, tokens_reasoning, tokens_cache_read,
|
|
tokens_cache_write, revert, permission, agent, model, time_created, time_updated, time_compacting,
|
|
time_archived, time_suspended
|
|
) VALUES (
|
|
${session.id}, ${projectID}, ${session.workspace_id}, ${session.parent_id},
|
|
${session.fork_session_id}, ${session.fork_boundary}, ${session.slug}, ${session.directory},
|
|
${session.path}, ${session.title}, ${session.version}, ${session.share_url},
|
|
${session.summary_additions}, ${session.summary_deletions}, ${session.summary_files},
|
|
${session.summary_diffs}, ${session.metadata}, ${session.cost}, ${session.tokens_input},
|
|
${session.tokens_output}, ${session.tokens_reasoning}, ${session.tokens_cache_read},
|
|
${session.tokens_cache_write}, ${session.revert}, ${session.permission}, ${session.agent},
|
|
${session.model}, ${session.time_created}, ${session.time_updated}, ${session.time_compacting},
|
|
${session.time_archived}, ${session.time_suspended}
|
|
)
|
|
`)
|
|
yield* Effect.forEach(messages, (message) =>
|
|
tx.run(sql`
|
|
INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data)
|
|
VALUES (
|
|
${message.id}, ${message.session_id}, ${message.type}, ${message.seq},
|
|
${message.time_created}, ${message.time_updated}, ${message.data}
|
|
)
|
|
`),
|
|
)
|
|
yield* tx
|
|
.insert(EventSequenceTable)
|
|
.values({ aggregate_id: session.id, seq: messages.at(-1)?.seq ?? -1 })
|
|
.onConflictDoUpdate({
|
|
target: EventSequenceTable.aggregate_id,
|
|
set: { seq: messages.at(-1)?.seq ?? -1, owner_id: null },
|
|
})
|
|
.run()
|
|
}),
|
|
)
|
|
.pipe(Effect.orDie)
|
|
onProgress(index + 1)
|
|
yield* Effect.yieldNow
|
|
}
|
|
source.run("COMMIT")
|
|
}),
|
|
)
|
|
}
|
|
|
|
function isNextDatabase(source: SQLiteDatabase) {
|
|
const tables = new Set(
|
|
source
|
|
.query<{ name: string }, []>("SELECT name FROM sqlite_master WHERE type = 'table'")
|
|
.all()
|
|
.map((table) => table.name),
|
|
)
|
|
return tables.has("project") && tables.has("session") && tables.has("session_message")
|
|
}
|
|
|
|
function row(
|
|
source: SourceMessage,
|
|
message: {
|
|
readonly id: string
|
|
readonly type: SessionMessage.Type
|
|
readonly time: { readonly created: number }
|
|
readonly [key: string]: unknown
|
|
},
|
|
): TransformResult["messages"][number] {
|
|
const { id, type, ...data } = message
|
|
return {
|
|
id,
|
|
session_id: source.session_id,
|
|
type,
|
|
seq: 0,
|
|
time_created: source.time_created,
|
|
time_updated: source.time_updated,
|
|
data,
|
|
}
|
|
}
|
|
|
|
function migrateTool(part: typeof SessionV1.ToolPart.Type, fallback: number) {
|
|
const base = {
|
|
type: "tool" as const,
|
|
id: part.callID,
|
|
name: part.tool,
|
|
...(part.metadata ? { providerState: part.metadata } : {}),
|
|
}
|
|
if (part.state.status === "completed")
|
|
return {
|
|
...base,
|
|
state: {
|
|
status: "completed",
|
|
input: part.state.input,
|
|
content:
|
|
part.state.time.compacted === undefined
|
|
? [
|
|
{ type: "text", text: part.state.output },
|
|
...(part.state.attachments ?? []).map((file) => ({
|
|
type: "file" as const,
|
|
uri: file.url,
|
|
mime: file.mime,
|
|
...(file.filename ? { name: file.filename } : {}),
|
|
})),
|
|
]
|
|
: [{ type: "text", text: "[Old tool result content cleared]" }],
|
|
metadata: part.state.metadata,
|
|
},
|
|
time: { created: part.state.time.start, completed: part.state.time.end },
|
|
}
|
|
if (part.state.status === "error")
|
|
return {
|
|
...base,
|
|
state: {
|
|
status: "error",
|
|
input: part.state.input,
|
|
error: { type: "tool.execution", message: part.state.error },
|
|
...(typeof part.state.metadata?.output === "string"
|
|
? { content: [{ type: "text", text: part.state.metadata.output }] }
|
|
: {}),
|
|
...(part.state.metadata ? { metadata: part.state.metadata } : {}),
|
|
},
|
|
time: { created: part.state.time.start, completed: part.state.time.end },
|
|
}
|
|
return {
|
|
...base,
|
|
state: {
|
|
status: "error",
|
|
input: part.state.input,
|
|
error: { type: "tool.interrupted", message: "Tool execution was interrupted before V2 migration" },
|
|
...(part.state.status === "running" && part.state.metadata ? { metadata: part.state.metadata } : {}),
|
|
},
|
|
time: { created: part.state.status === "running" ? part.state.time.start : fallback },
|
|
}
|
|
}
|
|
|
|
function migrateError(error: NonNullable<(typeof SessionV1.Assistant.Type)["error"]>) {
|
|
const message =
|
|
"message" in error.data
|
|
? error.data.message
|
|
: error.name === "MessageOutputLengthError"
|
|
? "The model exceeded its output limit"
|
|
: error.name
|
|
const type =
|
|
error.name === "ProviderAuthError"
|
|
? "provider.auth"
|
|
: error.name === "ContentFilterError"
|
|
? "provider.content-filter"
|
|
: error.name === "ContextOverflowError"
|
|
? "provider.invalid-request"
|
|
: error.name === "StructuredOutputError" || error.name === "MessageOutputLengthError"
|
|
? "provider.invalid-output"
|
|
: error.name === "MessageAbortedError"
|
|
? "aborted"
|
|
: error.name === "APIError"
|
|
? "provider.error"
|
|
: "unknown"
|
|
return { type, message }
|
|
}
|
|
|
|
function normalizeFinish(finish: string | undefined) {
|
|
if (!finish) return undefined
|
|
return (
|
|
(["stop", "length", "tool-calls", "content-filter", "error", "unknown"] as const).find(
|
|
(value) => value === finish,
|
|
) ?? "unknown"
|
|
)
|
|
}
|
|
|
|
function migrateFile(part: SessionV1.FilePart) {
|
|
if (!part.url.startsWith("data:")) return []
|
|
const comma = part.url.indexOf(",")
|
|
if (comma < 0) return []
|
|
const header = part.url.slice(0, comma)
|
|
const payload = part.url.slice(comma + 1)
|
|
const data = header.endsWith(";base64")
|
|
? Buffer.from(payload, "base64").toString("base64")
|
|
: Buffer.from(decodeURIComponent(payload)).toString("base64")
|
|
return [
|
|
{
|
|
data,
|
|
mime: part.mime,
|
|
source:
|
|
part.source?.type === "resource" ? { type: "uri" as const, uri: part.source.uri } : { type: "inline" as const },
|
|
...(part.filename ? { name: part.filename } : {}),
|
|
...(part.source
|
|
? { mention: { text: part.source.text.value, start: part.source.text.start, end: part.source.text.end } }
|
|
: {}),
|
|
},
|
|
]
|
|
}
|
|
|
|
function unavailableFile(part: SessionV1.FilePart) {
|
|
const label = part.filename ?? (part.source?.type === "resource" ? part.source.uri : part.url)
|
|
return `[Attachment unavailable after migration: ${label} (${part.mime})]`
|
|
}
|
|
|
|
function syntheticID(source: string, used: Set<string>) {
|
|
const prefix = source.slice(0, 16)
|
|
const alphabet = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
|
|
for (let salt = 0; ; salt++) {
|
|
const hex = new Bun.CryptoHasher("sha256").update(`v1-synthetic:${source}${salt ? `:${salt}` : ""}`).digest("hex")
|
|
let value = BigInt(`0x${hex}`)
|
|
let suffix = ""
|
|
while (suffix.length < 14) {
|
|
suffix = alphabet[Number(value % 62n)] + suffix
|
|
value /= 62n
|
|
}
|
|
const id = prefix + suffix
|
|
if (used.has(id)) continue
|
|
used.add(id)
|
|
return id
|
|
}
|
|
}
|
|
|
|
function serializeRecent(
|
|
messages: ReadonlyArray<{ row: SourceMessage; value: typeof SessionV1.Info.Type }>,
|
|
parts: Map<string, Array<{ row: SourcePart; value: typeof SessionV1.Part.Type }>>,
|
|
) {
|
|
return messages
|
|
.flatMap((message) => {
|
|
const owned = parts.get(message.row.id)?.map((part) => part.value) ?? []
|
|
if (message.value.role === "user")
|
|
return [
|
|
`[User]: ${owned
|
|
.filter((part) => part.type === "text" && !part.ignored)
|
|
.map((part) => (part.type === "text" ? part.text : ""))
|
|
.join("\n\n")}`,
|
|
]
|
|
return owned.flatMap((part) =>
|
|
part.type === "text"
|
|
? [`[Assistant]: ${part.text}`]
|
|
: part.type === "reasoning" && part.text
|
|
? [`[Assistant reasoning]: ${part.text}`]
|
|
: [],
|
|
)
|
|
})
|
|
.join("\n\n")
|
|
}
|
|
|
|
function readState(db: Database.Interface["db"]): Effect.Effect<MigrationState | undefined> {
|
|
return db
|
|
.select({ value: KVTable.value })
|
|
.from(KVTable)
|
|
.where(eq(KVTable.key, MIGRATION_STATE_KEY))
|
|
.get()
|
|
.pipe(
|
|
Effect.map((row) => parseState(row?.value)),
|
|
Effect.orDie,
|
|
)
|
|
}
|
|
|
|
function parseState(input: unknown): MigrationState | undefined {
|
|
if (!input || typeof input !== "object" || !("phase" in input)) return
|
|
if (input.phase === "completed") return { phase: "completed" }
|
|
if (input.phase !== "sessions") return
|
|
if (!("cursor" in input) || input.cursor === undefined) return { phase: "sessions" }
|
|
if (typeof input.cursor === "string") return { phase: "sessions", cursor: input.cursor }
|
|
}
|
|
|
|
function hasLegacySessions(db: Database.Interface["db"]) {
|
|
return db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session'`).pipe(
|
|
Effect.map((row) => row !== undefined),
|
|
Effect.orDie,
|
|
)
|
|
}
|