Compare commits

...
3 Commits
36 changed files with 761 additions and 707 deletions
+10 -9
View File
@@ -1032,16 +1032,17 @@ export function createData(config: CreateDataInput) {
if (store.session.info[event.data.sessionID]) {
setStore("session", "info", event.data.sessionID, "revert", undefined)
}
// The projector also deletes inbox items enqueued at or after the boundary without a cancel event.
setStore(
"session",
"pending",
event.data.sessionID,
(store.session.pending[event.data.sessionID] ?? []).filter((item) => item.id < event.data.to),
)
// Inbox ordering is server-owned; IDs do not encode delivery order.
result.session.pending.invalidate(event.data.sessionID)
if (store.session.pending[event.data.sessionID]?.length)
refresh(() => result.session.pending.sync(event.data.sessionID))
message.update(event.data.sessionID, (draft, index) => {
const position = draft.findIndex((item) => item.id >= event.data.to)
if (position === -1) return
const position = draft.findIndex((item) => item.id === event.data.to)
if (position === -1) {
result.session.message.invalidate(event.data.sessionID)
refresh(() => result.session.message.sync(event.data.sessionID))
return
}
for (const item of draft.splice(position)) index.delete(item.id)
})
return
+147 -6
View File
@@ -1,8 +1,8 @@
{
"version": "7",
"dialect": "sqlite",
"id": "be60f352-8da1-40e1-8d70-dc41121cfbc5",
"prevIds": ["3fb67508-0196-4bae-b2bd-c08ece7583fd"],
"id": "685a01ff-9b49-4789-a8a6-f1b5779a866c",
"prevIds": ["be60f352-8da1-40e1-8d70-dc41121cfbc5"],
"ddl": [
{
"name": "account_state",
@@ -72,6 +72,10 @@
"name": "session_v2",
"entityType": "tables"
},
{
"name": "timeline",
"entityType": "tables"
},
{
"name": "workspace",
"entityType": "tables"
@@ -940,6 +944,16 @@
"entityType": "columns",
"table": "session_message"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "timeline_id",
"entityType": "columns",
"table": "session_message"
},
{
"type": "text",
"notNull": true,
@@ -1070,6 +1084,16 @@
"entityType": "columns",
"table": "session_v2"
},
{
"type": "text",
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "timeline_id",
"entityType": "columns",
"table": "session_v2"
},
{
"type": "text",
"notNull": true,
@@ -1420,6 +1444,36 @@
"entityType": "columns",
"table": "session_v2"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "id",
"entityType": "columns",
"table": "timeline"
},
{
"type": "text",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "base_id",
"entityType": "columns",
"table": "timeline"
},
{
"type": "integer",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "base_seq",
"entityType": "columns",
"table": "timeline"
},
{
"type": "text",
"notNull": false,
@@ -1588,13 +1642,13 @@
"table": "session_inbox"
},
{
"columns": ["session_id"],
"tableTo": "session_v2",
"columns": ["timeline_id"],
"tableTo": "timeline",
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
"name": "fk_session_message_session_id_session_v2_id_fk",
"name": "fk_session_message_timeline_id_timeline_id_fk",
"entityType": "fks",
"table": "session_message"
},
@@ -1609,6 +1663,17 @@
"entityType": "fks",
"table": "session_pending"
},
{
"columns": ["timeline_id"],
"tableTo": "timeline",
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "NO ACTION",
"nameExplicit": false,
"name": "fk_session_v2_timeline_id_timeline_id_fk",
"entityType": "fks",
"table": "session_v2"
},
{
"columns": ["project_id"],
"tableTo": "project",
@@ -1620,6 +1685,17 @@
"entityType": "fks",
"table": "session_v2"
},
{
"columns": ["base_id"],
"tableTo": "timeline",
"columnsTo": ["id"],
"onUpdate": "NO ACTION",
"onDelete": "NO ACTION",
"nameExplicit": false,
"name": "fk_timeline_base_id_timeline_id_fk",
"entityType": "fks",
"table": "timeline"
},
{
"columns": ["project_id"],
"tableTo": "project",
@@ -1757,6 +1833,13 @@
"table": "session_v2",
"entityType": "pks"
},
{
"columns": ["id"],
"nameExplicit": false,
"name": "timeline_pk",
"table": "timeline",
"entityType": "pks"
},
{
"columns": ["id"],
"nameExplicit": false,
@@ -1877,13 +1960,71 @@
"isExpression": false
}
],
"isUnique": true,
"isUnique": false,
"where": null,
"origin": "manual",
"name": "session_message_session_seq_idx",
"entityType": "indexes",
"table": "session_message"
},
{
"columns": [
{
"value": "timeline_id",
"isExpression": false
},
{
"value": "seq",
"isExpression": false
}
],
"isUnique": true,
"where": null,
"origin": "manual",
"name": "session_message_timeline_seq_idx",
"entityType": "indexes",
"table": "session_message"
},
{
"columns": [
{
"value": "timeline_id",
"isExpression": false
},
{
"value": "type",
"isExpression": false
},
{
"value": "seq",
"isExpression": false
}
],
"isUnique": false,
"where": null,
"origin": "manual",
"name": "session_message_timeline_type_seq_idx",
"entityType": "indexes",
"table": "session_message"
},
{
"columns": [
{
"value": "timeline_id",
"isExpression": false
},
{
"value": "seq",
"isExpression": false
}
],
"isUnique": false,
"where": "((\"session_message\".\"type\" = 'assistant' AND json_extract(\"session_message\".\"data\", '$.time.completed') IS NULL)\n OR (\"session_message\".\"type\" IN ('shell', 'compaction') AND json_extract(\"session_message\".\"data\", '$.status') = 'running'))",
"origin": "manual",
"name": "session_message_unsettled_idx",
"entityType": "indexes",
"table": "session_message"
},
{
"columns": [
{
+2
View File
@@ -45,6 +45,7 @@ import m42 from "./migration/20260812181746_session_inbox.js"
import m43 from "./migration/20260812213948_worktree.js"
import m44 from "./migration/20260819222447_session_viewed_state.js"
import m45 from "./migration/20260823191254_nullable_workspace_binding.js"
import m46 from "./migration/20260906003536_timeline.js"
export const migrations = [
m00,
@@ -93,4 +94,5 @@ export const migrations = [
m43,
m44,
m45,
m46,
] satisfies DatabaseMigration.Migration[]
@@ -0,0 +1,126 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration.js"
// Each existing Session, including copied forks, starts with an independent root.
// Rebuilds preserve IDs, payloads, sequence numbers and timestamps. Session dependents
// are evacuated inside this transaction so engines with mandatory FK cascades are safe.
const migration: DatabaseMigration.Migration = {
id: "20260906003536_timeline",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`CREATE TABLE \`timeline\` (
\`id\` text PRIMARY KEY,
\`base_id\` text,
\`base_seq\` integer,
CONSTRAINT \`fk_timeline_base_id_timeline_id_fk\` FOREIGN KEY (\`base_id\`) REFERENCES \`timeline\`(\`id\`)
);`)
// Use the small Session table to map existing messages to independent roots.
// The replacement table below enforces NOT NULL on the backfilled IDs.
yield* tx.run(`ALTER TABLE session_v2 ADD timeline_id text`)
yield* tx.run(`UPDATE session_v2 SET timeline_id = 'tml_' || lower(hex(randomblob(16)))`)
yield* tx.run(`INSERT INTO timeline (id) SELECT timeline_id FROM session_v2`)
yield* tx.run(`CREATE TABLE \`__new_session_message\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`timeline_id\` text NOT NULL,
\`type\` text NOT NULL,
\`seq\` integer NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`data\` text NOT NULL,
CONSTRAINT \`fk_session_message_timeline_id_timeline_id_fk\` FOREIGN KEY (\`timeline_id\`) REFERENCES \`timeline\`(\`id\`) ON DELETE CASCADE
);`)
yield* tx.run(
`INSERT INTO \`__new_session_message\` (\`id\`, \`session_id\`, \`timeline_id\`, \`type\`, \`seq\`, \`time_created\`, \`time_updated\`, \`data\`) SELECT \`id\`, \`session_id\`, (SELECT timeline_id FROM session_v2 WHERE session_v2.id = session_message.session_id), \`type\`, \`seq\`, \`time_created\`, \`time_updated\`, \`data\` FROM \`session_message\``,
)
yield* tx.run(`DROP TABLE \`session_message\``)
yield* tx.run(`ALTER TABLE \`__new_session_message\` RENAME TO \`session_message\``)
yield* tx.run(`CREATE TABLE __timeline_instruction_entry AS SELECT * FROM instruction_entry`)
yield* tx.run(`DELETE FROM instruction_entry`)
yield* tx.run(`CREATE TABLE __timeline_instruction_state AS SELECT * FROM instruction_state`)
yield* tx.run(`DELETE FROM instruction_state`)
yield* tx.run(`CREATE TABLE __timeline_session_inbox AS SELECT * FROM session_inbox`)
yield* tx.run(`DELETE FROM session_inbox`)
yield* tx.run(`CREATE TABLE __timeline_session_pending AS SELECT * FROM session_pending`)
yield* tx.run(`DELETE FROM session_pending`)
yield* tx.run(`CREATE TABLE \`__new_session_v2\` (
\`id\` text PRIMARY KEY,
\`timeline_id\` text NOT NULL,
\`project_id\` text NOT NULL,
\`workspace_id\` text,
\`parent_id\` text,
\`fork_session_id\` text,
\`fork_boundary\` text,
\`slug\` text NOT NULL,
\`directory\` text NOT NULL,
\`path\` text,
\`title\` text,
\`version\` text NOT NULL,
\`share_url\` text,
\`summary_additions\` integer,
\`summary_deletions\` integer,
\`summary_files\` integer,
\`summary_diffs\` text,
\`metadata\` text,
\`cost\` real DEFAULT 0 NOT NULL,
\`tokens_input\` integer DEFAULT 0 NOT NULL,
\`tokens_output\` integer DEFAULT 0 NOT NULL,
\`tokens_reasoning\` integer DEFAULT 0 NOT NULL,
\`tokens_cache_read\` integer DEFAULT 0 NOT NULL,
\`tokens_cache_write\` integer DEFAULT 0 NOT NULL,
\`revert\` text,
\`permission\` text,
\`agent\` text,
\`model\` text,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`time_idle\` integer,
\`time_viewed\` integer,
\`idle_outcome\` text,
\`time_compacting\` integer,
\`time_archived\` integer,
\`time_suspended\` integer,
\`resume_attempts\` integer DEFAULT 0 NOT NULL,
CONSTRAINT \`fk_session_v2_timeline_id_timeline_id_fk\` FOREIGN KEY (\`timeline_id\`) REFERENCES \`timeline\`(\`id\`),
CONSTRAINT \`fk_session_v2_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE
);`)
yield* tx.run(
`INSERT INTO \`__new_session_v2\` (\`id\`, \`timeline_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_idle\`, \`time_viewed\`, \`idle_outcome\`, \`time_compacting\`, \`time_archived\`, \`time_suspended\`, \`resume_attempts\`) SELECT \`id\`, \`timeline_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_idle\`, \`time_viewed\`, \`idle_outcome\`, \`time_compacting\`, \`time_archived\`, \`time_suspended\`, \`resume_attempts\` FROM \`session_v2\``,
)
yield* tx.run(`DROP TABLE \`session_v2\``)
yield* tx.run(`ALTER TABLE \`__new_session_v2\` RENAME TO \`session_v2\``)
yield* tx.run(`INSERT INTO instruction_entry SELECT * FROM __timeline_instruction_entry`)
yield* tx.run(`DROP TABLE __timeline_instruction_entry`)
yield* tx.run(`INSERT INTO instruction_state SELECT * FROM __timeline_instruction_state`)
yield* tx.run(`DROP TABLE __timeline_instruction_state`)
yield* tx.run(`INSERT INTO session_inbox SELECT * FROM __timeline_session_inbox`)
yield* tx.run(`DROP TABLE __timeline_session_inbox`)
yield* tx.run(`INSERT INTO session_pending SELECT * FROM __timeline_session_pending`)
yield* tx.run(`DROP TABLE __timeline_session_pending`)
yield* tx.run(`CREATE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_message_timeline_seq_idx\` ON \`session_message\` (\`timeline_id\`,\`seq\`);`,
)
yield* tx.run(
`CREATE INDEX \`session_message_timeline_type_seq_idx\` ON \`session_message\` (\`timeline_id\`,\`type\`,\`seq\`);`,
)
yield* tx.run(`CREATE INDEX \`session_message_unsettled_idx\` ON \`session_message\` (\`timeline_id\`,\`seq\`) WHERE (("session_message"."type" = 'assistant' AND json_extract("session_message"."data", '$.time.completed') IS NULL)
OR ("session_message"."type" IN ('shell', 'compaction') AND json_extract("session_message"."data", '$.status') = 'running'));`)
yield* tx.run(
`CREATE INDEX \`session_message_session_type_seq_idx\` ON \`session_message\` (\`session_id\`,\`type\`,\`seq\`);`,
)
yield* tx.run(
`CREATE INDEX \`session_message_session_time_created_id_idx\` ON \`session_message\` (\`session_id\`,\`time_created\`,\`id\`);`,
)
yield* tx.run(`CREATE INDEX \`session_message_time_created_idx\` ON \`session_message\` (\`time_created\`);`)
yield* tx.run(`CREATE INDEX \`session_v2_project_idx\` ON \`session_v2\` (\`project_id\`);`)
yield* tx.run(`CREATE INDEX \`session_v2_workspace_idx\` ON \`session_v2\` (\`workspace_id\`);`)
yield* tx.run(`CREATE INDEX \`session_v2_parent_idx\` ON \`session_v2\` (\`parent_id\`);`)
yield* tx.run(
`CREATE INDEX \`session_v2_time_suspended_idx\` ON \`session_v2\` (\`time_suspended\`) WHERE "session_v2"."time_suspended" is not null;`,
)
})
},
}
export default migration
+21 -2
View File
@@ -158,12 +158,13 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
CREATE TABLE \`session_message\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`timeline_id\` text NOT NULL,
\`type\` text NOT NULL,
\`seq\` integer NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`data\` text NOT NULL,
CONSTRAINT \`fk_session_message_session_id_session_v2_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session_v2\`(\`id\`) ON DELETE CASCADE
CONSTRAINT \`fk_session_message_timeline_id_timeline_id_fk\` FOREIGN KEY (\`timeline_id\`) REFERENCES \`timeline\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
@@ -181,6 +182,7 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
yield* tx.run(`
CREATE TABLE \`session_v2\` (
\`id\` text PRIMARY KEY,
\`timeline_id\` text NOT NULL,
\`project_id\` text NOT NULL,
\`workspace_id\` text,
\`parent_id\` text,
@@ -216,9 +218,18 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
\`time_archived\` integer,
\`time_suspended\` integer,
\`resume_attempts\` integer DEFAULT 0 NOT NULL,
CONSTRAINT \`fk_session_v2_timeline_id_timeline_id_fk\` FOREIGN KEY (\`timeline_id\`) REFERENCES \`timeline\`(\`id\`),
CONSTRAINT \`fk_session_v2_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`timeline\` (
\`id\` text PRIMARY KEY,
\`base_id\` text,
\`base_seq\` integer,
CONSTRAINT \`fk_timeline_base_id_timeline_id_fk\` FOREIGN KEY (\`base_id\`) REFERENCES \`timeline\`(\`id\`)
);
`)
yield* tx.run(`
CREATE TABLE \`workspace\` (
\`id\` text PRIMARY KEY,
@@ -249,9 +260,17 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
yield* tx.run(
`CREATE UNIQUE INDEX \`session_inbox_session_enqueued_seq_idx\` ON \`session_inbox\` (\`session_id\`,\`enqueued_seq\`);`,
)
yield* tx.run(`CREATE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`,
`CREATE UNIQUE INDEX \`session_message_timeline_seq_idx\` ON \`session_message\` (\`timeline_id\`,\`seq\`);`,
)
yield* tx.run(
`CREATE INDEX \`session_message_timeline_type_seq_idx\` ON \`session_message\` (\`timeline_id\`,\`type\`,\`seq\`);`,
)
yield* tx.run(`
CREATE INDEX \`session_message_unsettled_idx\` ON \`session_message\` (\`timeline_id\`,\`seq\`) WHERE (("session_message"."type" = 'assistant' AND json_extract("session_message"."data", '$.time.completed') IS NULL)
OR ("session_message"."type" IN ('shell', 'compaction') AND json_extract("session_message"."data", '$.status') = 'running'));
`)
yield* tx.run(
`CREATE INDEX \`session_message_session_type_seq_idx\` ON \`session_message\` (\`session_id\`,\`type\`,\`seq\`);`,
)
+13 -305
View File
@@ -10,10 +10,9 @@ import { KVTable } from "../kv/sql.js"
import { EventSequenceTable } from "../event/sql.js"
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"
import { Timeline } from "../session/timeline.js"
export type SourceMessage = {
readonly id: string
@@ -88,10 +87,6 @@ type RunResult = {
readonly status: "completed"
}
type Options = {
readonly nextDatabasePath?: string
}
type MigrationState = { readonly phase: "sessions"; readonly cursor?: string } | { readonly phase: "completed" }
type RuntimeState =
@@ -99,118 +94,6 @@ type RuntimeState =
| { 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 NextColumns<A> = Record<keyof A, "required" | "nullable" | { readonly fallback: keyof A & string }>
const NEXT_PROJECT_COLUMNS = {
id: "required",
worktree: "required",
vcs: "nullable",
name: "nullable",
icon_url: "nullable",
icon_url_override: { fallback: "icon_url" },
icon_color: "nullable",
time_created: "required",
time_updated: "required",
time_initialized: "nullable",
sandboxes: "required",
commands: "nullable",
} satisfies NextColumns<NextProject>
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
}
const NEXT_SESSION_COLUMNS = {
id: "required",
project_id: "required",
workspace_id: "nullable",
parent_id: "nullable",
fork_session_id: "nullable",
fork_boundary: "nullable",
slug: "required",
directory: "required",
path: "nullable",
title: "nullable",
version: "required",
share_url: "nullable",
summary_additions: "nullable",
summary_deletions: "nullable",
summary_files: "nullable",
summary_diffs: "nullable",
metadata: "nullable",
cost: "required",
tokens_input: "required",
tokens_output: "required",
tokens_reasoning: "required",
tokens_cache_read: "required",
tokens_cache_write: "required",
revert: "nullable",
permission: "nullable",
agent: "nullable",
model: "nullable",
time_created: "required",
time_updated: "required",
time_compacting: "nullable",
time_archived: "nullable",
time_suspended: "nullable",
} satisfies NextColumns<NextSession>
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 EVENT_DELETE_BATCH_SIZE = 1_000
@@ -513,7 +396,7 @@ function updateProgress(progress: Progress) {
if (runtimeState.status === "running") runtimeState = { status: "running", progress }
}
export function run(options: Options = {}): Effect.Effect<RunResult, never, Database.Service | Global.Service> {
export function run(): Effect.Effect<RunResult, never, Database.Service | Global.Service> {
return lock.withPermit(
Effect.gen(function* () {
const db = (yield* Database.Service).db
@@ -546,7 +429,6 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
}),
)
.pipe(Effect.orDie)
const sourceTotal = yield* countNextSessions(nextPath(options, global.data))
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 =
@@ -554,12 +436,7 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
? ((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, global.data), (completed) => {
updateProgress({ label: "Migrating sessions", numerator: migrated + completed, denominator })
})
updateProgress({ label: "Migrating sessions", numerator: migrated + sourceTotal, denominator })
updateProgress({ label: "Migrating sessions", numerator: migrated, denominator: legacyTotal })
const projects = new Set(
(yield* db.all<{ id: string }>(sql`SELECT id FROM project`)).map((project) => project.id),
)
@@ -589,15 +466,21 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
sessionID: nextID.id,
projectID: nextID.project_id,
})
const existing = yield* tx
.select({ timelineID: SessionTable.timeline_id })
.from(SessionTable)
.where(eq(SessionTable.id, SessionSchema.ID.make(nextID.id)))
.get()
const timelineID = existing?.timelineID ?? (yield* Timeline.create(tx))
yield* tx.run(sql`
INSERT OR IGNORE INTO session_v2 (
id, project_id, workspace_id, parent_id, slug, directory, path, title, version, share_url,
id, timeline_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,
id, ${timelineID}, ${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
@@ -627,6 +510,7 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
.values({
id: SessionMessage.ID.make(message.id),
session_id: SessionSchema.ID.make(message.session_id),
timeline_id: next.timeline_id,
type: message.type,
seq: message.seq,
time_created: message.time_created,
@@ -657,7 +541,7 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
progress: {
label: "Migrating sessions",
numerator: (runtimeState.progress.numerator ?? 0) + 1,
denominator,
denominator: legacyTotal,
},
}
yield* Effect.yieldNow
@@ -681,182 +565,6 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
)
}
function nextPath(options: Options, data: string) {
if (options.nextDatabasePath) return options.nextDatabasePath
if (process.env.OPENCODE_DB === ":memory:") return undefined
return path.join(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
}),
)
}
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(
selectNextRows<NextProject>(source, "project", NEXT_PROJECT_COLUMNS).map((project) => [project.id, project]),
)
const sessions = selectNextRows<NextSession>(source, "session", NEXT_SESSION_COLUMNS)
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
.insert(SessionMessageTable)
.values({
id: SessionMessage.ID.make(message.id),
session_id: SessionSchema.ID.make(message.session_id),
type: message.type as SessionMessage.Type,
seq: message.seq,
time_created: message.time_created,
time_updated: message.time_updated,
data: sql`${message.data}`,
})
.run(),
)
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 selectNextRows<A>(source: SQLiteDatabase, table: "project" | "session", definition: NextColumns<A>) {
const columns = new Set(
source
.query<{ name: string }, [string]>("SELECT name FROM pragma_table_info(?)")
.all(table)
.map((column) => column.name),
)
const missing = Object.entries(definition)
.filter(([column, strategy]) => strategy === "required" && !columns.has(column))
.map(([column]) => column)
if (missing.length)
throw new Error(`Incompatible opencode-next.db: ${table} is missing required columns: ${missing.join(", ")}`)
const projection = Object.entries(definition).map(([column, strategy]) => {
if (columns.has(column)) return `"${column}"`
if (
typeof strategy === "object" &&
strategy !== null &&
"fallback" in strategy &&
typeof strategy.fallback === "string" &&
columns.has(strategy.fallback)
)
return `"${strategy.fallback}" AS "${column}"`
return `NULL AS "${column}"`
})
return source
.query<A, []>(`SELECT ${projection.join(", ")} FROM "${table}"${table === "session" ? ' ORDER BY "id" DESC' : ""}`)
.all()
}
function row(
source: SourceMessage,
message: {
+7 -14
View File
@@ -4,7 +4,7 @@ export * from "./session/schema.js"
import { Effect, Layer, Schema, Context, Stream } from "effect"
import { LLMClient } from "@opencode-ai/ai"
import { ListAnchor } from "@opencode-ai/schema/session"
import { and, desc, eq } from "drizzle-orm"
import { eq } from "drizzle-orm"
import { Project } from "./project.js"
import { Model } from "@opencode-ai/schema/model"
import { Location } from "./location.js"
@@ -63,6 +63,7 @@ import { Job } from "./job.js"
import type { Command } from "./command.js"
import { SessionEnvironment } from "./session/environment.js"
import { InstructionEntry } from "./session/instruction-entry.js"
import { Timeline } from "./session/timeline.js"
// get project -> project.locations
//
@@ -292,19 +293,11 @@ const layer = Layer.effect(
}),
fork: Effect.fn("Session.fork")(function* (input) {
const parent = yield* result.get(input.sessionID)
const boundary = yield* db
.select({ id: SessionMessageTable.id })
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, input.sessionID),
input.boundary.type === "before" ? eq(SessionMessageTable.id, input.boundary.messageID) : undefined,
),
)
.orderBy(desc(SessionMessageTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
const ranges = yield* Timeline.forSession(db, input.sessionID)
const [boundary] = yield* Timeline.rows(db, ranges, {
where: input.boundary.type === "before" ? eq(SessionMessageTable.id, input.boundary.messageID) : undefined,
limit: 1,
})
if (!boundary && input.boundary.type === "before")
return yield* new MessageNotFoundError({
sessionID: input.sessionID,
+26 -36
View File
@@ -1,4 +1,4 @@
import { and, asc, desc, eq, gte, sql } from "drizzle-orm"
import { and, eq, gte, sql } from "drizzle-orm"
import { Effect, Schema } from "effect"
import { Database } from "../database/database.js"
import { MessageDecodeError } from "./error.js"
@@ -7,26 +7,25 @@ import { SessionSchema } from "./schema.js"
import { Instructions } from "../instructions/index.js"
import { InstructionState } from "./instruction-state.js"
import { SessionMessageTable } from "./sql.js"
import { Timeline } from "./timeline.js"
type DatabaseService = Database.Interface["db"]
const decode = Schema.decodeUnknownEffect(SessionMessage.Info)
export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
return yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
eq(SessionMessageTable.type, "compaction"),
sql`json_extract(${SessionMessageTable.data}, '$.status') = 'completed'`,
),
)
.orderBy(desc(SessionMessageTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
export const latestCompaction = Effect.fnUntraced(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
ranges?: readonly Timeline.Range[],
) {
const [row] = yield* Timeline.rows(db, ranges ?? (yield* Timeline.forSession(db, sessionID)), {
where: and(
eq(SessionMessageTable.type, "compaction"),
sql`json_extract(${SessionMessageTable.data}, '$.status') = 'completed'`,
),
limit: 1,
})
return row
})
export const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =>
@@ -41,19 +40,12 @@ export const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) =
)
const messageEntries = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
const compaction = yield* latestCompaction(db, sessionID)
const rows = yield* db
.select()
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
compaction ? gte(SessionMessageTable.seq, compaction.seq) : undefined,
),
)
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
const ranges = yield* Timeline.forSession(db, sessionID)
const compaction = yield* latestCompaction(db, sessionID, ranges)
const rows = yield* Timeline.rows(db, ranges, {
where: compaction ? gte(SessionMessageTable.seq, compaction.seq) : undefined,
order: "asc",
})
return yield* Effect.forEach(rows, (row) =>
decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))),
)
@@ -112,13 +104,11 @@ export const firstUserMessage = Effect.fn("SessionHistory.firstUserMessage")(fun
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
const row = yield* db
.select()
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "user")))
.orderBy(asc(SessionMessageTable.seq))
.get()
.pipe(Effect.orDie)
const [row] = yield* Timeline.rows(db, yield* Timeline.forSession(db, sessionID), {
where: eq(SessionMessageTable.type, "user"),
order: "asc",
limit: 1,
})
if (!row) return undefined
const message = yield* decodeMessageRow(row).pipe(Effect.orElseSucceed(() => undefined))
return message?.type === "user" ? message : undefined
+48 -65
View File
@@ -1,6 +1,6 @@
export * as SessionProjector from "./projector.js"
import { and, asc, desc, eq, gt, gte, inArray, isNull, lt, lte, or, sql } from "drizzle-orm"
import { and, desc, eq, gt, gte, inArray, isNull, lt, lte, not, or, sql } from "drizzle-orm"
import { DateTime, Effect, Layer, Schema, Stream } from "effect"
import path from "path"
import { Database } from "../database/database.js"
@@ -14,7 +14,7 @@ import { SessionMessageUpdater } from "./message-updater.js"
import { SessionInbox } from "./inbox.js"
import { Workspace } from "@opencode-ai/schema/workspace"
import { InstructionState } from "./instruction-state.js"
import { SessionInboxTable, SessionMessageTable, SessionTable } from "./sql.js"
import { SessionInboxTable, SessionMessageTable, SessionTable, unsettled } from "./sql.js"
import { InstructionEntry } from "./instruction-entry.js"
import { Slug } from "../util/slug.js"
import { FSUtil } from "@opencode-ai/util/fs-util"
@@ -24,6 +24,7 @@ import { Project } from "@opencode-ai/schema/project"
import { AbsolutePath, RelativePath } from "../schema.js"
import type { SessionSchema } from "./schema.js"
import { ProjectTable } from "../project/sql.js"
import { Timeline } from "./timeline.js"
type DatabaseService = Database.Interface["db"]
type MessageEvent = Exclude<
@@ -115,39 +116,30 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.pipe(Effect.orDie)
if (!parent) return yield* Effect.die(new Error(`Fork parent session not found: ${event.data.parentID}`))
const boundary = yield* db
.select({ seq: SessionMessageTable.seq })
.select()
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, event.data.parentID),
eq(SessionMessageTable.id, event.data.boundary.messageID),
),
)
.where(eq(SessionMessageTable.id, event.data.boundary.messageID))
.get()
.pipe(Effect.orDie)
if (!boundary)
return yield* Effect.die(new Error(`Fork boundary message not found: ${event.data.boundary.messageID}`))
const copied = yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, event.data.parentID),
event.data.boundary.type === "before"
? lt(SessionMessageTable.seq, boundary.seq)
: lte(SessionMessageTable.seq, boundary.seq),
),
)
.orderBy(desc(SessionMessageTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
const ranges = yield* Timeline.ranges(db, boundary.timeline_id)
const end = boundary.seq + (event.data.boundary.type === "through" ? 1 : 0)
const [copied] = yield* Timeline.rows(db, ranges, { where: lt(SessionMessageTable.seq, end), limit: 1 })
const copiedSeq = copied?.seq
const [active] = yield* Timeline.rows(db, ranges, {
where: and(lt(SessionMessageTable.seq, end), unsettled(SessionMessageTable)),
order: "asc",
limit: 1,
})
const base = yield* Timeline.prefix(db, ranges, active?.seq ?? end)
const timelineID = yield* Timeline.create(db, base)
const stored = yield* db
.insert(SessionTable)
.values({
id: event.data.sessionID,
timeline_id: timelineID,
parent_id: null,
fork_session_id: event.data.parentID,
fork_boundary: event.data.boundary,
@@ -179,26 +171,19 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
if (event.data.instructionEntries)
yield* InstructionEntry.initialize(db, event.data.sessionID, event.data.instructionEntries, event.created)
let cursor = -1
while (copiedSeq !== undefined) {
const rows = yield* db
.select()
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, event.data.parentID),
gt(SessionMessageTable.seq, cursor),
lt(SessionMessageTable.seq, copiedSeq + 1),
// Terminal events for active projections stay on the parent, so forks copy only settled history.
sql`${SessionMessageTable.type} != 'assistant' or json_extract(${SessionMessageTable.data}, '$.time.completed') is not null`,
sql`${SessionMessageTable.type} != 'shell' or json_extract(${SessionMessageTable.data}, '$.status') != 'running'`,
sql`${SessionMessageTable.type} != 'compaction' or json_extract(${SessionMessageTable.data}, '$.status') != 'running'`,
),
)
.orderBy(asc(SessionMessageTable.seq))
.limit(ForkBatchSize)
.all()
.pipe(Effect.orDie)
// Active forks only copy the settled suffix after the first omitted message;
// the preceding immutable prefix is shared.
let cursor = (active?.seq ?? end) - 1
while (cursor < (copiedSeq ?? -1)) {
const rows = yield* Timeline.rows(db, ranges, {
where: and(
gt(SessionMessageTable.seq, cursor),
lt(SessionMessageTable.seq, end),
not(unsettled(SessionMessageTable)),
),
order: "asc",
limit: ForkBatchSize,
})
if (rows.length === 0) break
yield* db
@@ -207,6 +192,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
rows.map((row) => ({
id: SessionMessage.ID.make(`${SessionMessage.ID.fromEvent(event.id)}_${row.seq}`),
session_id: event.data.sessionID,
timeline_id: timelineID,
type: row.type,
seq: row.seq,
time_created: row.time_created,
@@ -301,7 +287,10 @@ function run(db: DatabaseService, event: MessageEvent) {
.select()
.from(SessionMessageTable)
.where(
and(eq(SessionMessageTable.session_id, event.data.sessionID), eq(SessionMessageTable.type, "assistant")),
and(
eq(SessionMessageTable.timeline_id, Timeline.current(event.data.sessionID)),
eq(SessionMessageTable.type, "assistant"),
),
)
.orderBy(desc(SessionMessageTable.seq))
.limit(1)
@@ -359,7 +348,7 @@ function run(db: DatabaseService, event: MessageEvent) {
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, event.data.sessionID),
eq(SessionMessageTable.timeline_id, Timeline.current(event.data.sessionID)),
eq(SessionMessageTable.type, "compaction"),
sql`json_extract(${SessionMessageTable.data}, '$.status') = 'running'`,
),
@@ -390,6 +379,7 @@ function insertMessage(db: DatabaseService, event: SessionEvent.DurableEvent, me
.values({
id: SessionMessage.ID.make(id),
session_id: event.data.sessionID,
timeline_id: Timeline.current(event.data.sessionID),
type,
seq: event.durable.seq,
time_created: DateTime.toEpochMillis(message.time.created),
@@ -436,10 +426,12 @@ const layer = Layer.effectDiscard(
const db = (yield* Database.Service).db
yield* bus.project(SessionEvent.Created, (event) =>
Effect.gen(function* () {
const timelineID = yield* Timeline.create(db)
const stored = yield* db
.insert(SessionTable)
.values({
id: event.data.sessionID,
timeline_id: timelineID,
project_id: event.data.projectID,
workspace_id: event.data.location.workspaceID ? Workspace.ID.make(event.data.location.workspaceID) : null,
parent_id: event.data.parentID,
@@ -539,7 +531,10 @@ const layer = Layer.effectDiscard(
}),
)
yield* bus.project(SessionEvent.Deleted, (event) =>
db.delete(SessionTable).where(eq(SessionTable.id, event.data.sessionID)).run().pipe(Effect.orDie),
Effect.gen(function* () {
yield* db.delete(SessionTable).where(eq(SessionTable.id, event.data.sessionID)).run().pipe(Effect.orDie)
yield* Timeline.collect(db)
}),
)
yield* bus.project(SessionEvent.AgentSelected, (event) =>
Effect.gen(function* () {
@@ -720,20 +715,14 @@ const layer = Layer.effectDiscard(
)
yield* bus.project(SessionEvent.RevertEvent.Committed, (event) =>
Effect.gen(function* () {
const boundary = yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(eq(SessionMessageTable.session_id, event.data.sessionID), eq(SessionMessageTable.id, event.data.to)),
)
.get()
.pipe(Effect.orDie)
const boundary = yield* Timeline.find(db, event.data.sessionID, event.data.to)
if (!boundary) return yield* Effect.die(new Error(`Revert boundary message not found: ${event.data.to}`))
const base = yield* Timeline.prefix(db, yield* Timeline.ranges(db, boundary.timeline_id), boundary.seq)
const timelineID = yield* Timeline.create(db, base)
yield* db
.delete(SessionMessageTable)
.where(
and(eq(SessionMessageTable.session_id, event.data.sessionID), gte(SessionMessageTable.seq, boundary.seq)),
)
.update(SessionTable)
.set({ timeline_id: timelineID, revert: null, time_updated: event.created })
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
yield* db
@@ -746,12 +735,6 @@ const layer = Layer.effectDiscard(
)
.run()
.pipe(Effect.orDie)
yield* db
.update(SessionTable)
.set({ revert: null, time_updated: event.created })
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
yield* InstructionState.reset(db, event.data.sessionID)
}),
)
+7 -20
View File
@@ -1,6 +1,6 @@
export * as SessionRevert from "./revert.js"
import { and, asc, eq, gt } from "drizzle-orm"
import { and, eq, gt } from "drizzle-orm"
import { Effect, Schema } from "effect"
import { Database } from "../database/database.js"
import { Bus } from "../bus.js"
@@ -12,6 +12,7 @@ import { MessageNotFoundError } from "./error.js"
import { SessionMessage } from "./message.js"
import { SessionSchema } from "./schema.js"
import { SessionMessageTable } from "./sql.js"
import { Timeline } from "./timeline.js"
export { MessageNotFoundError }
@@ -84,26 +85,12 @@ export const commit = Effect.fn("SessionRevert.commit")(function* (bus: Bus.Inte
})
const plan = Effect.fn("SessionRevert.plan")(function* (db: Database.Interface["db"], input: BoundaryInput) {
const boundary = yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.messageID)))
.get()
.pipe(Effect.orDie)
const boundary = yield* Timeline.find(db, input.sessionID, input.messageID)
if (!boundary) return yield* new MessageNotFoundError(input)
const rows = yield* db
.select()
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, input.sessionID),
eq(SessionMessageTable.type, "assistant"),
gt(SessionMessageTable.seq, boundary.seq),
),
)
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
const rows = yield* Timeline.rows(db, yield* Timeline.forSession(db, input.sessionID), {
where: and(eq(SessionMessageTable.type, "assistant"), gt(SessionMessageTable.seq, boundary.seq)),
order: "asc",
})
const decode = Schema.decodeUnknownEffect(SessionMessage.Info)
const files = new Map<RelativePath, Snapshot.ID>()
for (const row of rows) {
+2 -1
View File
@@ -17,6 +17,7 @@ import { SessionMessage } from "../message.js"
import { SessionSchema } from "../schema.js"
import { SessionStore } from "../store.js"
import { SessionMessageTable } from "../sql.js"
import { Timeline } from "../timeline.js"
import { SessionTitle } from "../title.js"
import { DrainResult, Service, type Interface } from "./index.js"
import { Snapshot } from "../../snapshot.js"
@@ -293,7 +294,7 @@ const layer = Layer.effect(
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
eq(SessionMessageTable.timeline_id, Timeline.current(sessionID)),
eq(SessionMessageTable.type, "compaction"),
sql`json_extract(${SessionMessageTable.data}, '$.status') = 'running'`,
),
+1 -1
View File
@@ -58,7 +58,7 @@ export const make = Effect.fn("Session.make")(function* () {
return session
})
const message = Effect.fn("Session.message")(function* (sessionID: SessionSchema.ID, messageID: SessionMessage.ID) {
const stored = yield* store.message(messageID)
const stored = yield* store.message(messageID, sessionID)
return stored?.sessionID === sessionID ? stored.message : undefined
})
const updateMessage = Effect.fn("Session.updateMessage")(function* (
+39 -6
View File
@@ -1,5 +1,14 @@
import { sqliteTable, text, integer, index, primaryKey, real, uniqueIndex } from "drizzle-orm/sqlite-core"
import { sql } from "drizzle-orm"
import {
sqliteTable,
text,
integer,
index,
primaryKey,
real,
uniqueIndex,
type AnySQLiteColumn,
} from "drizzle-orm/sqlite-core"
import { sql, type SQLWrapper } from "drizzle-orm"
import { directoryColumn, pathColumn } from "../database/path.js"
import { ProjectTable } from "../project/sql.js"
import type { SessionMessage } from "./message.js"
@@ -15,14 +24,27 @@ import type { Session } from "@opencode-ai/schema/session"
import type { CompactionPayload, MovePayload, SyntheticPayload, UserPayload } from "@opencode-ai/schema/session-inbox"
import type { RevertV1 } from "@opencode-ai/schema/session-revert"
import type { Schema } from "effect"
import type { Timeline } from "./timeline.js"
type DistributiveOmit<T, K extends PropertyKey> = T extends unknown ? Omit<T, K> : never
type SessionMessageData = DistributiveOmit<(typeof SessionMessage.Info)["Encoded"], "type" | "id">
export const TimelineTable = sqliteTable("timeline", {
id: text().$type<Timeline.ID>().primaryKey(),
base_id: text()
.$type<Timeline.ID>()
.references((): AnySQLiteColumn => TimelineTable.id),
base_seq: integer(),
})
export const SessionTable = sqliteTable(
"session_v2",
{
id: text().$type<SessionSchema.ID>().primaryKey(),
timeline_id: text()
.$type<Timeline.ID>()
.notNull()
.references(() => TimelineTable.id),
project_id: text()
.$type<Project.ID>()
.notNull()
@@ -80,23 +102,34 @@ export const SessionMessageTable = sqliteTable(
"session_message",
{
id: text().$type<SessionMessage.ID>().primaryKey(),
session_id: text()
.$type<SessionSchema.ID>()
// Provenance survives deletion of the originating session while forks still reference its history.
session_id: text().$type<SessionSchema.ID>().notNull(),
timeline_id: text()
.$type<Timeline.ID>()
.notNull()
.references(() => SessionTable.id, { onDelete: "cascade" }),
.references(() => TimelineTable.id, { onDelete: "cascade" }),
type: text().$type<SessionMessage.Type>().notNull(),
seq: integer().notNull(),
...Timestamps,
data: text({ mode: "json" }).notNull().$type<SessionMessageData>(),
},
(table) => [
uniqueIndex("session_message_session_seq_idx").on(table.session_id, table.seq),
// A restored Session ID may have retained messages in another timeline.
index("session_message_session_seq_idx").on(table.session_id, table.seq),
uniqueIndex("session_message_timeline_seq_idx").on(table.timeline_id, table.seq),
index("session_message_timeline_type_seq_idx").on(table.timeline_id, table.type, table.seq),
index("session_message_unsettled_idx").on(table.timeline_id, table.seq).where(unsettled(table)),
index("session_message_session_type_seq_idx").on(table.session_id, table.type, table.seq),
index("session_message_session_time_created_id_idx").on(table.session_id, table.time_created, table.id),
index("session_message_time_created_idx").on(table.time_created),
],
)
export function unsettled(table: { type: SQLWrapper; data: SQLWrapper }) {
return sql`((${table.type} = 'assistant' AND json_extract(${table.data}, '$.time.completed') IS NULL)
OR (${table.type} IN ('shell', 'compaction') AND json_extract(${table.data}, '$.status') = 'running'))`
}
export const SessionPendingTable = sqliteTable(
"session_pending",
{
+18 -29
View File
@@ -13,6 +13,7 @@ import { SessionMessage } from "./message.js"
import { Session } from "@opencode-ai/schema/session"
import { SessionMessageTable, SessionTable } from "./sql.js"
import { fromRow } from "./info.js"
import { Timeline } from "./timeline.js"
const ListInputBase = {
workspaceID: Workspace.ID.pipe(Schema.optional),
@@ -56,6 +57,7 @@ export interface Interface {
readonly context: (sessionID: Session.ID) => Effect.Effect<SessionMessage.Info[], MessageDecodeError>
readonly message: (
messageID: SessionMessage.ID,
sessionID?: Session.ID,
) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.Info } | undefined>
/**
* Top-level Sessions holding an execution claim. Recoverable background
@@ -140,49 +142,36 @@ const layer = Layer.effect(
const direction = input.cursor?.direction ?? "next"
const requestedOrder = input.order ?? "desc"
const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder
const anchor = input.cursor
? yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.cursor.id)),
)
.get()
.pipe(Effect.orDie)
: undefined
const anchor = input.cursor ? yield* Timeline.find(db, input.sessionID, input.cursor.id) : undefined
if (input.cursor && !anchor) return []
const boundary = anchor
? order === "asc"
? gt(SessionMessageTable.seq, anchor.seq)
: lt(SessionMessageTable.seq, anchor.seq)
: undefined
const where = boundary
? and(eq(SessionMessageTable.session_id, input.sessionID), boundary)
: eq(SessionMessageTable.session_id, input.sessionID)
const query = db
.select()
.from(SessionMessageTable)
.where(where)
.orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq))
const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
Effect.orDie,
)
const rows = yield* Timeline.rows(db, yield* Timeline.forSession(db, input.sessionID), {
where: boundary,
order,
limit: input.limit,
})
return yield* Effect.forEach(
direction === "previous" ? rows.toReversed() : rows,
SessionHistory.decodeMessageRow,
)
}),
context: Effect.fn("SessionStore.context")((sessionID) => SessionHistory.load(db, sessionID)),
message: Effect.fn("SessionStore.message")(function* (messageID) {
const row = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, messageID))
.get()
.pipe(Effect.orDie)
message: Effect.fn("SessionStore.message")(function* (messageID, sessionID) {
const row = sessionID
? yield* Timeline.find(db, sessionID, messageID)
: yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, messageID))
.get()
.pipe(Effect.orDie)
return row
? {
sessionID: Session.ID.make(row.session_id),
sessionID: sessionID ?? Session.ID.make(row.session_id),
message: yield* SessionHistory.decodeMessageRow(row).pipe(Effect.orDie),
}
: undefined
+123
View File
@@ -0,0 +1,123 @@
export * as Timeline from "./timeline.js"
import { and, asc, desc, eq, lt, sql, type SQL } from "drizzle-orm"
import { Effect, Schema } from "effect"
import { Session } from "@opencode-ai/schema/session"
import { SessionMessage } from "@opencode-ai/schema/session-message"
import { statics } from "@opencode-ai/schema/schema"
import type { Database } from "../database/database.js"
import { SessionMessageTable, SessionTable, TimelineTable } from "./sql.js"
export const ID = Schema.String.check(Schema.isStartsWith("tml_")).pipe(
Schema.brand("Timeline.ID"),
statics((schema) => ({ create: () => schema.make(`tml_${crypto.randomUUID()}`) })),
)
export type ID = typeof ID.Type
type DB = Omit<Database.Interface["db"], "$client">
export type Position = { readonly id: ID; readonly seq: number }
export type Range = { readonly id: ID; readonly end: number | null }
export const create = Effect.fn("Timeline.create")(function* (db: DB, base?: Position) {
const id = ID.create()
yield* db.insert(TimelineTable).values({ id, base_id: base?.id, base_seq: base?.seq }).run().pipe(Effect.orDie)
return id
})
export const current = (sessionID: Session.ID) => sql`(SELECT timeline_id FROM ${SessionTable} WHERE id = ${sessionID})`
/** Resolve ancestry once, then read each physical range using (timeline_id, seq). */
export const ranges = Effect.fn("Timeline.ranges")(function* (db: DB, id: ID) {
return yield* db
.all<Range>(
sql`
WITH RECURSIVE lineage(id, base_id, base_seq, end, depth) AS (
SELECT id, base_id, base_seq, NULL, 0 FROM timeline WHERE id = ${id}
UNION ALL
SELECT base.id, base.base_id, base.base_seq,
CASE WHEN lineage.end IS NULL THEN lineage.base_seq
ELSE min(lineage.end, lineage.base_seq) END,
lineage.depth + 1
FROM timeline AS base JOIN lineage ON base.id = lineage.base_id
)
SELECT id, end FROM lineage ORDER BY depth ASC
`,
)
.pipe(Effect.orDie)
})
export const forSession = Effect.fn("Timeline.forSession")(function* (db: DB, sessionID: Session.ID) {
const session = yield* db
.select({ id: SessionTable.timeline_id })
.from(SessionTable)
.where(eq(SessionTable.id, sessionID))
.get()
.pipe(Effect.orDie)
return session ? yield* ranges(db, session.id) : []
})
export const includes = (ranges: readonly Range[], row: { timeline_id: ID; seq: number }) =>
ranges.some((range) => range.id === row.timeline_id && (range.end === null || row.seq < range.end))
export const find = Effect.fn("Timeline.find")(function* (db: DB, sessionID: Session.ID, messageID: SessionMessage.ID) {
const row = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, messageID))
.get()
.pipe(Effect.orDie)
if (!row) return undefined
return includes(yield* forSession(db, sessionID), row) ? row : undefined
})
export const rows = Effect.fn("Timeline.rows")(function* (
db: DB,
ranges: readonly Range[],
input: { readonly where?: SQL; readonly order?: "asc" | "desc"; readonly limit?: number } = {},
) {
const result: (typeof SessionMessageTable.$inferSelect)[] = []
for (const range of input.order === "asc" ? ranges.toReversed() : ranges) {
if (input.limit !== undefined && result.length >= input.limit) break
const query = db
.select()
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.timeline_id, range.id),
range.end === null ? undefined : lt(SessionMessageTable.seq, range.end),
input.where,
),
)
.orderBy(input.order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq))
result.push(
...(yield* (input.limit === undefined ? query.all() : query.limit(input.limit - result.length).all()).pipe(
Effect.orDie,
)),
)
}
return result
})
/** Root at the physical owner of the last retained message, skipping empty intermediate timelines. */
export const prefix = Effect.fn("Timeline.prefix")(function* (db: DB, ranges: readonly Range[], end: number) {
const [last] = yield* rows(db, ranges, { where: lt(SessionMessageTable.seq, end), limit: 1 })
return last ? { id: last.timeline_id, seq: last.seq + 1 } : undefined
})
/** Session deletion releases a head; references from surviving histories keep their storage alive. */
export const collect = Effect.fn("Timeline.collect")(function* (db: DB) {
yield* db
.run(
sql`
WITH RECURSIVE retained(id) AS (
SELECT timeline_id FROM session_v2
UNION
SELECT timeline.base_id FROM timeline JOIN retained ON timeline.id = retained.id
WHERE timeline.base_id IS NOT NULL
)
DELETE FROM timeline WHERE id NOT IN (SELECT id FROM retained)
`,
)
.pipe(Effect.orDie)
})
+22 -2
View File
@@ -3,7 +3,7 @@ export * as SessionTransfer from "./transfer.js"
import { SessionTransfer } from "@opencode-ai/schema/session-transfer"
import { Tool } from "@opencode-ai/schema/tool"
import { Skill } from "@opencode-ai/schema/skill"
import { eq } from "drizzle-orm"
import { eq, inArray } from "drizzle-orm"
import { Clock, Context, DateTime, Effect, Layer, Schema } from "effect"
import { map } from "effect/Array"
import path from "path"
@@ -21,6 +21,7 @@ import { SessionEvent } from "./event.js"
import { SessionMessage } from "./message.js"
import { SessionProjector } from "./projector.js"
import { SessionMessageTable, SessionTable } from "./sql.js"
import { Timeline } from "./timeline.js"
export const Data = SessionTransfer.Data
export type Data = SessionTransfer.Data
@@ -74,12 +75,31 @@ const layer = Layer.effect(
const project = yield* projects.resolve(input.location.directory)
yield* upsertProject(db, project).pipe(Effect.orDie)
const importedAt = yield* Clock.currentTimeMillis
// Related exports may share message IDs. Imports materialize independent
// snapshots, so give colliding rows fresh identities instead of stealing ownership.
const ids = input.data.messages.map((message) => message.id)
const batches = Array.from({ length: Math.ceil(ids.length / 500) }, (_, index) =>
ids.slice(index * 500, (index + 1) * 500),
)
const existing = new Set(
(yield* Effect.forEach(batches, (batch) =>
db
.select({ id: SessionMessageTable.id })
.from(SessionMessageTable)
.where(inArray(SessionMessageTable.id, batch))
.all()
.pipe(Effect.orDie),
))
.flat()
.map((row) => row.id),
)
const messages = input.data.messages.filter(isSettled).map((message, index) => {
const encoded = encodeMessage(message)
const { id: _, type, ...data } = encoded
return {
id: message.id,
id: existing.has(message.id) ? SessionMessage.ID.create() : message.id,
session_id: sessionID,
timeline_id: Timeline.current(sessionID),
type,
seq: index + 1,
time_created: DateTime.toEpochMillis(message.time.created),
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { describe, expect } from "bun:test"
import { Deferred, Effect, Exit, Fiber, Stream } from "effect"
import { eq } from "drizzle-orm"
@@ -37,6 +38,7 @@ const seed = Effect.fn(function* (ref: Location.Ref = a) {
yield* database.db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(database.db),
id,
project_id: Project.ID.global,
directory: ref.directory,
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { describe, expect } from "bun:test"
import { and, asc, eq } from "drizzle-orm"
import { Effect, Schema } from "effect"
@@ -52,6 +53,7 @@ const setup = (sessionID: SessionSchema.ID) =>
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "instruction-state-test",
+2
View File
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { describe, expect } from "bun:test"
import { Cause, Deferred, Effect, Fiber, Layer } from "effect"
import { Agent } from "@opencode-ai/core/agent"
@@ -43,6 +44,7 @@ function setup(rules: Permission.Ruleset = [], sessionID = Session.ID.make("ses_
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "test",
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { expect, test } from "bun:test"
import { LLMClient, LLMEvent, LanguageModel, ToolDefinition, type LLMRequest } from "@opencode-ai/ai"
import { OpenAIChat } from "@opencode-ai/ai/protocols"
@@ -284,6 +285,7 @@ const insertSession = (id: Session.ID, overrides?: Partial<typeof SessionTable.$
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id,
project_id: Project.ID.global,
slug: id,
+7 -5
View File
@@ -500,7 +500,7 @@ describe("Session.create", () => {
}),
)
it.effect("forks a session by replaying a durable fork event into copied projected rows", () =>
it.effect("forks a session through shared projected history", () =>
Effect.gen(function* () {
const session = yield* Session.Service
const bus = yield* Bus.Service
@@ -523,7 +523,7 @@ describe("Session.create", () => {
expect(forked).toMatchObject({ title: "Parent (fork #1)", fork: { sessionID: parent.id } })
expect(forked.parentID).toBeUndefined()
expect(forkContext).toMatchObject([Expected.user("First"), { type: "synthetic", text: "parent note" }])
expect(forkContext.map((message) => message.id)).not.toEqual(parentContext.map((message) => message.id))
expect(forkContext.map((message) => message.id)).toEqual(parentContext.map((message) => message.id))
expect(history).toHaveLength(1)
expect(history[0]).toMatchObject({
type: "session.forked",
@@ -533,8 +533,10 @@ describe("Session.create", () => {
expect(yield* SessionInbox.find(db, forkContext[0].id)).toBeUndefined()
expect(yield* SessionInbox.find(db, forkContext[1].id)).toBeUndefined()
expect(
yield* session.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }),
).toMatchObject({ id: forkContext[0].id, type: "user", payload: { text: "First" } })
yield* Effect.flip(
session.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }),
),
).toMatchObject({ _tag: "Session.PromptConflictError", messageID: forkContext[0].id })
yield* session.prompt({
sessionID: parent.id,
@@ -819,7 +821,7 @@ describe("Session.create", () => {
boundary: { type: "before", messageID: second.id },
})
expect(context).toMatchObject([{ text: "First" }])
expect(context[0]?.id).not.toBe(first.id)
expect(context[0]?.id).toBe(first.id)
expect(history[0]).toMatchObject({
data: { boundary: { type: "before", messageID: second.id } },
})
+4 -1
View File
@@ -1,4 +1,5 @@
import { describe, expect, test } from "bun:test"
import { Timeline } from "@opencode-ai/core/session/timeline"
import { AIError, TransportError } from "@opencode-ai/ai"
import { Database } from "@opencode-ai/core/database/database"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
@@ -1282,11 +1283,13 @@ function seedSessions(
.onConflictDoNothing()
.run()
.pipe(Effect.orDie)
const timelines = yield* Effect.forEach(sessionIDs, () => Timeline.create(database.db))
yield* database.db
.insert(SessionTable)
.values(
sessionIDs.map((id) => ({
sessionIDs.map((id, index) => ({
id,
timeline_id: timelines[index],
project_id: Project.ID.global,
slug: id,
directory: "/project",
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { expect } from "bun:test"
import { LLMClient, LLMEvent, LLMResponse, LanguageModel, ToolDefinition, type LLMRequest } from "@opencode-ai/ai"
import { OpenAIChat } from "@opencode-ai/ai/protocols"
@@ -209,6 +210,7 @@ const setup = Effect.gen(function* () {
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: (yield* projects.resolve(AbsolutePath.make("/project"))).id,
slug: "generate-test",
+2
View File
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { describe, expect } from "bun:test"
import { Effect, Fiber, Schema, Stream } from "effect"
import { Database } from "@opencode-ai/core/database/database"
@@ -106,6 +107,7 @@ describe("Session.log", () => {
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "empty-log",
+24 -5
View File
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { describe, expect } from "bun:test"
import { DateTime, Effect, Fiber, Option, Schema, Stream } from "effect"
import { asc, eq, sql } from "drizzle-orm"
@@ -59,7 +60,15 @@ const assistantRow = (
} = encodeMessage(
SessionMessage.Assistant.make({ id, type: "assistant", agent: build, model, content: [], time, ...usage }),
)
return { id, session_id: sessionID, type, seq, time_created: DateTime.toEpochMillis(time.created), data }
return {
id,
session_id: sessionID,
timeline_id: Timeline.current(sessionID),
type,
seq,
time_created: DateTime.toEpochMillis(time.created),
data,
}
}
const seedSession = (overrides?: Partial<typeof SessionTable.$inferInsert>) =>
@@ -72,6 +81,7 @@ const seedSession = (overrides?: Partial<typeof SessionTable.$inferInsert>) =>
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "test",
@@ -190,9 +200,9 @@ describe("SessionProjector", () => {
sessionID,
to: boundary,
})
expect(
(yield* db.select({ id: SessionMessageTable.id }).from(SessionMessageTable).all()).map((row) => row.id),
).toEqual([earlier])
expect((yield* Timeline.rows(db, yield* Timeline.forSession(db, sessionID))).map((row) => row.id)).toEqual([
earlier,
])
expect(yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get()).toMatchObject({
cost: Money.USD.make(1.25),
tokens_input: 10,
@@ -271,6 +281,7 @@ describe("SessionProjector", () => {
.values({
id: messageID,
session_id: sessionID,
timeline_id: Timeline.current(sessionID),
type: "user",
seq: 0,
data: { text: "valid before corruption", time: { created: 0 } },
@@ -476,7 +487,15 @@ describe("SessionProjector", () => {
const { id: _, type, ...data } = encodeMessage({ id, type: "synthetic", text: "existing", time: { created } })
yield* db
.insert(SessionMessageTable)
.values({ id, session_id: sessionID, type, seq: 0, time_created: 0, data })
.values({
id,
session_id: sessionID,
timeline_id: Timeline.current(sessionID),
type,
seq: 0,
time_created: 0,
data,
})
.run()
const exit = yield* bus
+14 -5
View File
@@ -1,4 +1,5 @@
import { describe, expect } from "bun:test"
import { Timeline } from "@opencode-ai/core/session/timeline"
import { DateTime, Effect, Fiber, Layer, LayerMap, Schema, Stream } from "effect"
import path from "path"
import { pathToFileURL } from "url"
@@ -113,6 +114,7 @@ const setup = Effect.gen(function* () {
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "test",
@@ -165,7 +167,7 @@ const assistantRow = (id: SessionMessage.ID, seq: number) => {
time: { created: DateTime.makeUnsafe(0) },
}),
)
return { id, session_id: sessionID, type, seq, time_created: 0, data }
return { id, session_id: sessionID, timeline_id: Timeline.current(sessionID), type, seq, time_created: 0, data }
}
describe("Session.prompt", () => {
@@ -277,9 +279,7 @@ describe("Session.prompt", () => {
expect((yield* session.get(sessionID)).revert).toBeUndefined()
expect(
(yield* db.select({ id: SessionMessageTable.id }).from(SessionMessageTable).all().pipe(Effect.orDie)).map(
(row) => row.id,
),
(yield* Timeline.rows(db, yield* Timeline.forSession(db, sessionID))).map((row) => row.id),
).not.toContainAnyValues([boundary.id, stale])
expect(yield* SessionInbox.find(db, boundary.id)).toBeUndefined()
}),
@@ -813,6 +813,7 @@ describe("Session.prompt", () => {
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: other,
project_id: Project.ID.global,
slug: "other",
@@ -849,7 +850,15 @@ describe("Session.prompt", () => {
})
yield* db
.insert(SessionMessageTable)
.values({ id: messageID, session_id: sessionID, type, seq: 0, time_created: 0, data })
.values({
id: messageID,
session_id: sessionID,
timeline_id: Timeline.current(sessionID),
type,
seq: 0,
time_created: 0,
data,
})
.run()
.pipe(Effect.orDie)
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { HttpRecorder } from "@opencode-ai/http-recorder"
import { OpenAIChat } from "@opencode-ai/ai/protocols/openai-chat"
import { Auth, LLMClient, type LLMClientService, RequestExecutor } from "@opencode-ai/ai/route"
@@ -204,6 +205,7 @@ describe("SessionRunnerLLM recorded", () => {
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "test",
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { expect, test } from "bun:test"
import { Cause, Deferred, Effect, Exit, Fiber, Schema } from "effect"
import { eq } from "drizzle-orm"
@@ -148,7 +149,14 @@ testEffect(
.run()
yield* database.db
.insert(SessionTable)
.values({ id: sessionID, project_id: Project.ID.global, slug: "publish", directory: "/project", version: "test" })
.values({
timeline_id: yield* Timeline.create(database.db),
id: sessionID,
project_id: Project.ID.global,
slug: "publish",
directory: "/project",
version: "test",
})
.run()
const publisher = createLLMEventPublisher(
{
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { describe, expect, test } from "bun:test"
import {
AIError,
@@ -495,6 +496,7 @@ const insertSession = (id: Session.ID) =>
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id,
project_id: Project.ID.global,
slug: id,
@@ -1465,6 +1467,7 @@ describe("SessionRunnerLLM", () => {
.all()
yield* s.bus.remove(forked.id)
yield* s.db.delete(SessionTable).where(eq(SessionTable.id, forked.id)).run()
yield* Timeline.collect(s.db)
yield* Effect.forEach(
recorded.map((event) => ({
id: event.id,
+37 -9
View File
@@ -1,4 +1,5 @@
import { describe, expect } from "bun:test"
import { Timeline } from "@opencode-ai/core/session/timeline"
import { Agent } from "@opencode-ai/schema/agent"
import { Event } from "@opencode-ai/schema/event"
import { Model } from "@opencode-ai/schema/model"
@@ -44,9 +45,17 @@ describe("SessionStats", () => {
yield* db
.insert(SessionTable)
.values([
{ id: sessionID, project_id: projectID, slug: "root", directory: "/stats", version: "test" },
{
id: sessionID,
timeline_id: yield* Timeline.create(db),
project_id: projectID,
slug: "root",
directory: "/stats",
version: "test",
},
{
id: childID,
timeline_id: yield* Timeline.create(db),
project_id: projectID,
parent_id: sessionID,
slug: "child",
@@ -55,6 +64,7 @@ describe("SessionStats", () => {
},
{
id: forkID,
timeline_id: yield* Timeline.create(db),
project_id: projectID,
fork_session_id: sessionID,
slug: "fork",
@@ -62,8 +72,22 @@ describe("SessionStats", () => {
version: "test",
time_created: Date.UTC(2026, 0, 4),
},
{ id: usageOnlyID, project_id: projectID, slug: "usage", directory: "/stats", version: "test" },
{ id: otherSessionID, project_id: otherProjectID, slug: "other", directory: "/other", version: "test" },
{
id: usageOnlyID,
timeline_id: yield* Timeline.create(db),
project_id: projectID,
slug: "usage",
directory: "/stats",
version: "test",
},
{
id: otherSessionID,
timeline_id: yield* Timeline.create(db),
project_id: otherProjectID,
slug: "other",
directory: "/other",
version: "test",
},
])
.run()
.pipe(Effect.orDie)
@@ -288,12 +312,16 @@ function assistant(
})
}
function messageRow(
sessionID: Session.ID,
seq: number,
message: SessionMessage.Info,
): typeof SessionMessageTable.$inferInsert {
function messageRow(sessionID: Session.ID, seq: number, message: SessionMessage.Info) {
const encoded = encodeMessage(message)
const { id, type, ...data } = encoded
return { id: SessionMessage.ID.make(id), session_id: sessionID, type, seq, time_created: encoded.time.created, data }
return {
id: SessionMessage.ID.make(id),
session_id: sessionID,
timeline_id: Timeline.current(sessionID),
type,
seq,
time_created: encoded.time.created,
data,
}
}
+9 -1
View File
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { expect } from "bun:test"
import { LanguageModel, LLM, LLMEvent } from "@opencode-ai/ai"
import { OpenAIChat } from "@opencode-ai/ai/protocols/openai-chat"
@@ -66,7 +67,14 @@ for (const fixture of [
.run()
yield* db
.insert(SessionTable)
.values({ id: sessionID, project_id: Project.ID.global, slug: "step", directory: "/project", version: "test" })
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "step",
directory: "/project",
version: "test",
})
.run()
const model = SessionRunnerModel.resolved(
LanguageModel.make({ id: "test-model", provider: "test", route: OpenAIChat.route }),
+3
View File
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { beforeEach, expect } from "bun:test"
import { AIError, LLMClient, LLMEvent, LanguageModel, TransportError, type LLMRequest } from "@opencode-ai/ai"
import { OpenAIChat } from "@opencode-ai/ai/protocols"
@@ -139,6 +140,7 @@ const insertSession = (id: Session.ID, title?: string, created?: number, model?:
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id,
project_id: Project.ID.global,
slug: id,
@@ -339,6 +341,7 @@ it.effect("generates a title for an explicitly requested child session", () =>
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
parent_id: Session.ID.make("ses_title_parent"),
@@ -1,3 +1,4 @@
import { Timeline } from "@opencode-ai/core/session/timeline"
import { describe, expect } from "bun:test"
import { asc, eq } from "drizzle-orm"
import { Effect, Schema } from "effect"
@@ -43,6 +44,7 @@ describe("Session tool progress", () => {
yield* db
.insert(SessionTable)
.values({
timeline_id: yield* Timeline.create(db),
id: sessionID,
project_id: Project.ID.global,
slug: "progress",
+7 -179
View File
@@ -8,13 +8,11 @@ import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionSchema } from "@opencode-ai/core/session/schema"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { Project } from "@opencode-ai/core/project"
import { ProjectTable } from "@opencode-ai/core/project/sql"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { Global } from "@opencode-ai/util/global"
import { Effect, Fiber, Layer, Logger, Schedule, Schema, Scope } from "effect"
import { eq, sql } from "drizzle-orm"
import { sql } from "drizzle-orm"
import type { SqlClient } from "effect/unstable/sql/SqlClient"
import { tmpdir } from "./fixture/tmpdir"
import { Timeline } from "@opencode-ai/core/session/timeline"
import path from "path"
const makeDb = EffectDrizzleSqlite.makeWithDefaults()
@@ -32,6 +30,7 @@ const session = (
overrides: Partial<V1Migration.TransformInput["session"]> = {},
): V1Migration.TransformInput["session"] => ({
id: SessionSchema.ID.make("ses_test"),
timeline_id: Timeline.ID.create(),
project_id: Project.ID.global,
workspace_id: null,
parent_id: null,
@@ -831,179 +830,6 @@ describe("V1Migration database workflow", () => {
)
})
test("imports previous V2 sessions and messages containing apostrophes", async () => {
await using tmp = await tmpdir()
const filename = path.join(tmp.path, "opencode-next.db")
const sqlite = await import("bun:sqlite")
const source = new sqlite.Database(filename)
source.run(`
CREATE TABLE project (
id text PRIMARY KEY, worktree text NOT NULL, vcs text, name text, icon_url text, icon_url_override text,
icon_color text, time_created integer NOT NULL, time_updated integer NOT NULL, time_initialized integer,
sandboxes text NOT NULL, commands text
);
CREATE TABLE session (
id text PRIMARY KEY, project_id text NOT NULL, workspace_id text, parent_id text, fork_session_id text,
fork_boundary text, slug text NOT NULL, directory text NOT NULL, path text, title text, version text NOT NULL,
share_url text, summary_additions integer, summary_deletions integer, summary_files integer, summary_diffs text,
metadata text, cost real DEFAULT 0 NOT NULL, tokens_input integer DEFAULT 0 NOT NULL,
tokens_output integer DEFAULT 0 NOT NULL, tokens_reasoning integer DEFAULT 0 NOT NULL,
tokens_cache_read integer DEFAULT 0 NOT NULL, tokens_cache_write integer DEFAULT 0 NOT NULL, revert text,
permission text, agent text, model text, time_created integer NOT NULL, time_updated integer NOT NULL,
time_compacting integer, time_archived integer, time_suspended integer
);
CREATE TABLE session_message (
id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, seq integer NOT NULL,
time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL
);
INSERT INTO project VALUES (
'next-project', 'C:/Users/sewer', 'git', 'Source project', NULL, NULL, NULL, 1, 2, NULL, '[]', NULL
);
INSERT INTO session (
id, project_id, slug, directory, title, version, agent, model, time_created, time_updated
) VALUES
('ses_next', 'next-project', 'next', 'C:/Users/sewer', 'Imported', '2', 'build',
'{"id":"model","providerID":"provider"}', 10, 20),
('ses_existing', 'next-project', 'source-existing', '/tmp/next', 'Source existing', '2', NULL, NULL, 11, 21),
('ses_orphan', 'missing-project', 'orphan', '/tmp/orphan', 'Orphan', '2', NULL, NULL, 12, 22);
INSERT INTO session_message VALUES
('msg_next', 'ses_next', 'user', 4, 12, 13, '{"text":"from next''s history","time":{"created":12}}'),
('msg_source_existing', 'ses_existing', 'user', 2, 12, 13, '{"text":"source","time":{"created":12}}'),
('msg_orphan', 'ses_orphan', 'user', 0, 12, 13, '{"text":"orphan","time":{"created":12}}');
`)
source.close()
await database(
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db.run(sql`
INSERT INTO project (id, worktree, name, time_created, time_updated, sandboxes)
VALUES ('next-project', '/tmp/current', 'Current project', 1, 2, '[]')
`)
yield* db.run(sql`
INSERT INTO session_v2 (id, project_id, slug, directory, title, version, time_created, time_updated)
VALUES ('ses_existing', 'next-project', 'current-existing', '/tmp/current', 'Current existing', '2', 1, 2)
`)
yield* db.run(sql`
INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data)
VALUES ('msg_current_existing', 'ses_existing', 'user', 0, 1, 2, '{"text":"current","time":{"created":1}}')
`)
expect(yield* V1Migration.status()).toEqual({
status: "required",
})
expect(yield* V1Migration.run({ nextDatabasePath: filename })).toEqual({ status: "completed" })
expect(yield* V1Migration.status()).toEqual({
status: "completed",
})
expect(yield* db.get(sql`SELECT title, agent, model FROM session_v2 WHERE id = 'ses_next'`)).toEqual({
title: "Imported",
agent: "build",
model: '{"id":"model","providerID":"provider"}',
})
expect(
yield* db
.select({ directory: SessionTable.directory })
.from(SessionTable)
.where(eq(SessionTable.id, SessionSchema.ID.make("ses_next")))
.get(),
).toEqual({ directory: process.platform === "win32" ? "C:\\Users\\sewer" : "C:/Users/sewer" })
expect(yield* db.all(sql`SELECT id, seq, data FROM session_message WHERE session_id = 'ses_next'`)).toEqual([
{
id: "msg_next",
seq: 4,
data: '{"text":"from next\'s history","time":{"created":12}}',
},
])
expect(yield* db.get(sql`SELECT seq, owner_id FROM event_sequence WHERE aggregate_id = 'ses_next'`)).toEqual({
seq: 4,
owner_id: null,
})
expect(yield* db.get(sql`SELECT title FROM session_v2 WHERE id = 'ses_existing'`)).toEqual({
title: "Current existing",
})
expect(yield* db.all(sql`SELECT id FROM session_message WHERE session_id = 'ses_existing'`)).toEqual([
{ id: "msg_current_existing" },
])
expect(yield* db.get(sql`SELECT project_id FROM session_v2 WHERE id = 'ses_orphan'`)).toEqual({
project_id: "global",
})
expect(yield* db.get(sql`SELECT name, worktree FROM project WHERE id = 'next-project'`)).toEqual({
name: "Current project",
worktree: "/tmp/current",
})
yield* db.run(sql`UPDATE project SET worktree = 'C:/Users/sewer' WHERE id = 'next-project'`)
expect(
yield* db
.select({ worktree: ProjectTable.worktree })
.from(ProjectTable)
.where(eq(ProjectTable.id, Project.ID.make("next-project")))
.get(),
).toEqual({
worktree: AbsolutePath.make(process.platform === "win32" ? "C:\\Users\\sewer" : "C:/Users/sewer"),
})
expect(yield* db.get(sql`SELECT value FROM kv WHERE key = 'migration.v1-v2'`)).toEqual({
value: '{"phase":"completed"}',
})
}),
)
})
test("imports previous V2 databases missing newer nullable columns", async () => {
await using tmp = await tmpdir()
const filename = path.join(tmp.path, "opencode-next.db")
const sqlite = await import("bun:sqlite")
const source = new sqlite.Database(filename)
source.run(`
CREATE TABLE project (
id text PRIMARY KEY, worktree text NOT NULL, vcs text, name text, icon_url text,
time_created integer NOT NULL, time_updated integer NOT NULL, time_initialized integer,
sandboxes text NOT NULL
);
CREATE TABLE session (
id text PRIMARY KEY, project_id text NOT NULL, workspace_id text, parent_id text, fork_session_id text,
slug text NOT NULL, directory text NOT NULL, path text, title text, version text NOT NULL,
share_url text, summary_additions integer, summary_deletions integer, summary_files integer, summary_diffs text,
metadata text, cost real DEFAULT 0 NOT NULL, tokens_input integer DEFAULT 0 NOT NULL,
tokens_output integer DEFAULT 0 NOT NULL, tokens_reasoning integer DEFAULT 0 NOT NULL,
tokens_cache_read integer DEFAULT 0 NOT NULL, tokens_cache_write integer DEFAULT 0 NOT NULL, revert text,
permission text, agent text, model text, time_created integer NOT NULL, time_updated integer NOT NULL,
time_compacting integer, time_archived integer
);
CREATE TABLE session_message (
id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, seq integer NOT NULL,
time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL
);
INSERT INTO project VALUES (
'next-project', '/tmp/next', 'git', 'Source project', 'https://example.com/icon.png', 1, 2, NULL, '[]'
);
INSERT INTO session (
id, project_id, slug, directory, title, version, time_created, time_updated
) VALUES ('ses_next', 'next-project', 'next', '/tmp/next', 'Imported', '2', 10, 20);
`)
source.close()
await database(
Effect.gen(function* () {
const database = yield* Database.Service
expect(yield* V1Migration.run({ nextDatabasePath: filename })).toEqual({ status: "completed" })
expect(
yield* database.db.get(sql`SELECT fork_boundary, time_suspended FROM session_v2 WHERE id = 'ses_next'`),
).toEqual({ fork_boundary: null, time_suspended: null })
expect(
yield* database.db.get(
sql`SELECT icon_url, icon_url_override, icon_color, commands FROM project WHERE id = 'next-project'`,
),
).toEqual({
icon_url: "https://example.com/icon.png",
icon_url_override: "https://example.com/icon.png",
icon_color: null,
commands: null,
})
}),
)
})
test("derives required status from the durable cursor", async () => {
await database(
Effect.gen(function* () {
@@ -1071,8 +897,9 @@ describe("V1Migration database workflow", () => {
yield* db.run(
sql`INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES ('prt_1', ${source.id}, 'ses_test', 1, 2, ${sourcePart.data})`,
)
const timelineID = yield* Timeline.create(db)
yield* db.run(
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('msg_stale', 'ses_test', 'user', 0, 1, 1, '{"text":"stale","time":{"created":1}}')`,
sql`INSERT INTO session_message (id, session_id, timeline_id, type, seq, time_created, time_updated, data) VALUES ('msg_stale', 'ses_test', ${timelineID}, 'user', 0, 1, 1, '{"text":"stale","time":{"created":1}}')`,
)
yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES ('ses_test', 9)`)
yield* db.run(
@@ -1142,8 +969,9 @@ describe("V1Migration database workflow", () => {
yield* db.run(
sql`CREATE TRIGGER fail_b BEFORE UPDATE ON session_v2 WHEN NEW.id = 'ses_b' BEGIN SELECT RAISE(ABORT, 'stop'); END`,
)
const timelineID = yield* Timeline.create(db)
yield* db.run(
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('msg_stale_b', 'ses_b', 'user', 0, 7, 8, '{"text":"stale","time":{"created":7}}')`,
sql`INSERT INTO session_message (id, session_id, timeline_id, type, seq, time_created, time_updated, data) VALUES ('msg_stale_b', 'ses_b', ${timelineID}, 'user', 0, 7, 8, '{"text":"stale","time":{"created":7}}')`,
)
yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq, owner_id) VALUES ('ses_b', 7, 'owner')`)
yield* db.run(
+2 -5
View File
@@ -52,11 +52,8 @@ export function createSessionRows(sessionID: Accessor<string>, onSynced?: (sessi
)
const visible = queued.size === 0 ? messages : messages.filter((message) => !queued.has(message.id))
const boundary = revertBoundary()
const rows = reduceSessionRows(
boundary ? visible.filter((message) => message.id < boundary) : visible,
inputs,
turnTokens(),
)
const cutoff = boundary ? visible.findIndex((message) => message.id === boundary) : -1
const rows = reduceSessionRows(cutoff === -1 ? visible : visible.slice(0, cutoff), inputs, turnTokens())
partitionPending(rows, pendingPermissions())
const position = rows.findIndex((row) => row.type === "message" && inputs.has(row.messageID))
rows.splice(
+14
View File
@@ -1087,6 +1087,19 @@ test("removes committed revert messages from local state", async () => {
const sessionID = "session-revert"
const calls = createFetch((url) => {
if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} })
if (url.pathname === `/api/session/${sessionID}/inbox`)
return json({
data: [
{
id: "msg_001",
sessionID,
type: "user",
payload: { text: "msg_001" },
delivery: "steer",
timeCreated: 0,
},
],
})
}, events)
let data!: ReturnType<typeof useData>
@@ -1132,6 +1145,7 @@ test("removes committed revert messages from local state", async () => {
expect(data.session.message.get(sessionID, "msg_002")).toBeUndefined()
expect(data.session.message.get(sessionID, "msg_003")).toBeUndefined()
// The projector also drops inbox items enqueued at or after the boundary, without a cancel event.
await wait(() => data.session.pending.list(sessionID).length === 1)
expect(data.session.pending.list(sessionID).map((item) => item.id)).toEqual(["msg_001"])
expect(data.session.input.list(sessionID)).toEqual(["msg_001"])
expect(data.session.input.has(sessionID, "msg_002")).toBe(false)