database-migration.test.ts 53 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152
  1. import { describe, expect, test } from "bun:test"
  2. import { $ } from "bun"
  3. import { fileURLToPath } from "url"
  4. import path from "path"
  5. import { SqliteClient } from "@effect/sql-sqlite-bun"
  6. import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
  7. import { Effect, Layer, Schema } from "effect"
  8. import { eq, inArray, sql } from "drizzle-orm"
  9. import { DatabaseMigration } from "@opencode-ai/core/database/migration"
  10. import { migrations } from "@opencode-ai/core/database/migration.gen"
  11. import sessionUsageMigration from "@opencode-ai/core/database/migration/20260510033149_session_usage"
  12. import normalizeStoragePathsMigration from "@opencode-ai/core/database/migration/20260601010001_normalize_storage_paths"
  13. import sessionMessageProjectionOrderMigration from "@opencode-ai/core/database/migration/20260603040000_session_message_projection_order"
  14. import eventSourcedSessionPendingMigration from "@opencode-ai/core/database/migration/20260604172448_event_sourced_session_input"
  15. import contextEpochAgentMigration from "@opencode-ai/core/database/migration/20260605042240_add_context_epoch_agent"
  16. import simplifyIntegrationCredentialsMigration from "@opencode-ai/core/database/migration/20260611192811_lush_chimera"
  17. import simplifySessionPendingMigration from "@opencode-ai/core/database/migration/20260622202450_simplify_session_input"
  18. import resetSessionEventsMigration from "@opencode-ai/core/database/migration/20260703200000_reset_v2_session_events"
  19. import durableSessionInboxMigration from "@opencode-ai/core/database/migration/20260707010146_durable_session_inbox"
  20. import migratePrelaunchV2StateMigration from "@opencode-ai/core/database/migration/20260707120000_migrate_prelaunch_v2_state"
  21. import genericSessionPendingMigration from "@opencode-ai/core/database/migration/20260709013000_generic_session_input"
  22. import sessionPendingTableMigration from "@opencode-ai/core/database/migration/20260709190621_session_pending_table"
  23. import renameInstructionsMigration from "@opencode-ai/core/database/migration/20260705180000_rename_instructions"
  24. import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork"
  25. import timeSuspendedMigration from "@opencode-ai/core/database/migration/20260709163752_time_suspended"
  26. import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
  27. import { LayerNode } from "@opencode-ai/core/effect/layer-node"
  28. import { EventV2 } from "@opencode-ai/core/event"
  29. import { ProjectV2 } from "@opencode-ai/core/project"
  30. import { ProjectTable } from "@opencode-ai/core/project/sql"
  31. import { AbsolutePath } from "@opencode-ai/core/schema"
  32. import { SessionSchema } from "@opencode-ai/core/session/schema"
  33. import { SessionMessage } from "@opencode-ai/core/session/message"
  34. import { SessionTable } from "@opencode-ai/core/session/sql"
  35. import sessionMetadataMigration from "@opencode-ai/core/database/migration/20260511173437_session-metadata"
  36. import type { SqlClient as SqlClientService } from "effect/unstable/sql/SqlClient"
  37. import { Database } from "@opencode-ai/core/database/database"
  38. import { SessionProjector } from "@opencode-ai/core/session/projector"
  39. import { SessionV1 } from "@opencode-ai/core/v1/session"
  40. import { tmpdir } from "./fixture/tmpdir"
  41. const run = <A, E>(effect: Effect.Effect<A, E, SqlClientService>) =>
  42. Effect.runPromise(
  43. effect.pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", disableWAL: true })), Effect.scoped),
  44. )
  45. const makeDb = EffectDrizzleSqlite.makeWithDefaults()
  46. describe("DatabaseMigration", () => {
  47. test("migrates pre-launch V2 state in place", async () => {
  48. await run(
  49. Effect.gen(function* () {
  50. const db = yield* makeDb
  51. yield* db.run(
  52. sql`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)`,
  53. )
  54. yield* db.run(
  55. sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, prompt text, delivery text, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
  56. )
  57. yield* db.run(
  58. sql`CREATE TABLE event (id text PRIMARY KEY, aggregate_id text NOT NULL, seq integer NOT NULL, created integer NOT NULL, type text NOT NULL, data text NOT NULL)`,
  59. )
  60. yield* db.run(
  61. sql`CREATE TABLE event_sequence (aggregate_id text PRIMARY KEY, seq integer NOT NULL, owner_id text)`,
  62. )
  63. yield* db.run(
  64. sql`CREATE TABLE instruction_checkpoint (session_id text PRIMARY KEY, baseline text NOT NULL, snapshot text NOT NULL, baseline_seq integer NOT NULL)`,
  65. )
  66. const messages = [
  67. ["msg_skill", "skill", { name: "effect", text: "Use Effect", time: { created: 1 } }],
  68. [
  69. "msg_shell",
  70. "shell",
  71. {
  72. shell: { id: "sh_old", command: "pwd", status: "exited", exit: 0, cwd: "/tmp" },
  73. output: { output: "/tmp", cursor: 4, size: 4, truncated: false },
  74. time: { created: 2, completed: 3 },
  75. },
  76. ],
  77. [
  78. "msg_assistant",
  79. "assistant",
  80. {
  81. agent: "build",
  82. model: { id: "model", providerID: "provider" },
  83. content: [
  84. {
  85. type: "tool",
  86. id: "call_old",
  87. name: "read",
  88. provider: "removed",
  89. state: { status: "pending", input: '{"path":"README.md"}', title: "removed" },
  90. time: { created: 3 },
  91. },
  92. ],
  93. time: { created: 3 },
  94. },
  95. ],
  96. [
  97. "msg_failed",
  98. "compaction",
  99. {
  100. status: "failed",
  101. reason: "manual",
  102. summary: "removed",
  103. recent: "removed",
  104. time: { created: 4 },
  105. },
  106. ],
  107. [
  108. "msg_queued",
  109. "compaction",
  110. { status: "queued", reason: "manual", summary: "", recent: "", time: { created: 5 } },
  111. ],
  112. [
  113. "msg_synthetic",
  114. "synthetic",
  115. { sessionID: "ses_test", text: "context", description: "source", time: { created: 6 } },
  116. ],
  117. [
  118. "msg_running",
  119. "compaction",
  120. { status: "running", reason: "auto", summary: "partial", recent: "recent", time: { created: 7 } },
  121. ],
  122. [
  123. "msg_completed",
  124. "compaction",
  125. { status: "completed", reason: "auto", summary: "summary", recent: "recent", time: { created: 8 } },
  126. ],
  127. ] as const
  128. for (const [id, type, data] of messages)
  129. yield* db.run(
  130. sql`INSERT INTO session_message VALUES (${id}, 'ses_test', ${type}, 1, 10, 11, ${JSON.stringify(data)})`,
  131. )
  132. yield* db.run(
  133. sql`INSERT INTO session_input VALUES ('msg_queued', 'ses_test', 'compaction', NULL, NULL, 4, NULL, 5)`,
  134. )
  135. yield* db.run(sql`INSERT INTO event_sequence VALUES ('ses_test', 9, 'owner')`)
  136. yield* db.run(sql`INSERT INTO instruction_checkpoint VALUES ('ses_test', 'baseline', '{"source":"value"}', 7)`)
  137. const events = [
  138. ["evt_skill", 1, 101, "session.skill.activated.1", { sessionID: "ses_test", name: "effect", text: "Use" }],
  139. ["evt_started", 2, 102, "session.compaction.started.1", { sessionID: "ses_test", reason: "auto" }],
  140. ["evt_delta", 3, 103, "session.compaction.delta.1", { sessionID: "ses_test", text: "partial" }],
  141. ["evt_failed", 4, 104, "session.compaction.failed.1", { sessionID: "ses_test" }],
  142. [
  143. "evt_revert",
  144. 5,
  145. 105,
  146. "session.revert.staged.1",
  147. {
  148. sessionID: "ses_test",
  149. revert: {
  150. messageID: "msg_skill",
  151. snapshot: "tree",
  152. diff: "removed",
  153. files: [{ path: "src/a.ts", patch: "@@", additions: 1, deletions: 0, status: "modified" }],
  154. },
  155. },
  156. ],
  157. [
  158. "evt_skill_current",
  159. 6,
  160. 106,
  161. "session.skill.activated.2",
  162. { sessionID: "ses_test", id: "effect-id", name: "Effect", text: "Use" },
  163. ],
  164. ] as const
  165. for (const [id, seq, created, type, data] of events)
  166. yield* db.run(
  167. sql`INSERT INTO event VALUES (${id}, 'ses_test', ${seq}, ${created}, ${type}, ${JSON.stringify(data)})`,
  168. )
  169. yield* DatabaseMigration.applyOnly(db, [migratePrelaunchV2StateMigration])
  170. const rows = yield* db.all<{
  171. id: string
  172. type: string
  173. seq: number
  174. time_created: number
  175. time_updated: number
  176. data: string
  177. }>(sql`SELECT id, type, seq, time_created, time_updated, data FROM session_message ORDER BY id`)
  178. for (const row of rows)
  179. Schema.decodeUnknownSync(SessionMessage.Info)({ ...JSON.parse(row.data), id: row.id, type: row.type })
  180. expect(rows.every((row) => row.seq === 1 && row.time_created === 10 && row.time_updated === 11)).toBe(true)
  181. expect(rows.map((row) => [row.id, JSON.parse(row.data)])).toEqual([
  182. [
  183. "msg_assistant",
  184. expect.objectContaining({
  185. content: [expect.objectContaining({ state: { status: "streaming", input: '{"path":"README.md"}' } })],
  186. }),
  187. ],
  188. ["msg_completed", expect.objectContaining({ status: "completed", summary: "summary", recent: "recent" })],
  189. [
  190. "msg_failed",
  191. {
  192. time: { created: 4 },
  193. status: "failed",
  194. reason: "manual",
  195. error: {
  196. type: "compaction.failed",
  197. message: "Compaction failed before recording an error",
  198. },
  199. },
  200. ],
  201. ["msg_running", expect.objectContaining({ status: "running", summary: "partial", recent: "recent" })],
  202. ["msg_shell", expect.objectContaining({ shellID: "sh_old", command: "pwd", status: "exited", exit: 0 })],
  203. ["msg_skill", { time: { created: 1 }, skill: "effect", name: "effect", text: "Use Effect" }],
  204. ["msg_synthetic", { time: { created: 6 }, text: "context", description: "source" }],
  205. ])
  206. expect(yield* db.get(sql`SELECT * FROM session_input`)).toEqual({
  207. id: "msg_queued",
  208. session_id: "ses_test",
  209. type: "compaction",
  210. prompt: null,
  211. delivery: null,
  212. admitted_seq: 4,
  213. promoted_seq: null,
  214. time_created: 5,
  215. })
  216. const migratedEvents = yield* db.all<{
  217. id: string
  218. aggregate_id: string
  219. seq: number
  220. created: number
  221. type: string
  222. data: string
  223. }>(sql`SELECT * FROM event ORDER BY seq`)
  224. expect(migratedEvents.map((event) => ({ ...event, data: JSON.parse(event.data) }))).toEqual([
  225. {
  226. id: "evt_skill",
  227. aggregate_id: "ses_test",
  228. seq: 1,
  229. created: 101,
  230. type: "session.skill.activated.1",
  231. data: { sessionID: "ses_test", id: "effect", name: "effect", text: "Use" },
  232. },
  233. {
  234. id: "evt_started",
  235. aggregate_id: "ses_test",
  236. seq: 2,
  237. created: 102,
  238. type: "session.compaction.started.1",
  239. data: { sessionID: "ses_test", reason: "auto", recent: "" },
  240. },
  241. {
  242. id: "evt_failed",
  243. aggregate_id: "ses_test",
  244. seq: 4,
  245. created: 104,
  246. type: "session.compaction.failed.1",
  247. data: {
  248. sessionID: "ses_test",
  249. reason: "auto",
  250. error: {
  251. type: "compaction.failed",
  252. message: "Compaction failed before recording an error",
  253. },
  254. },
  255. },
  256. {
  257. id: "evt_revert",
  258. aggregate_id: "ses_test",
  259. seq: 5,
  260. created: 105,
  261. type: "session.revert.staged.1",
  262. data: {
  263. sessionID: "ses_test",
  264. revert: {
  265. messageID: "msg_skill",
  266. snapshot: "tree",
  267. files: [{ file: "src/a.ts", patch: "@@", additions: 1, deletions: 0, status: "modified" }],
  268. },
  269. },
  270. },
  271. {
  272. id: "evt_skill_current",
  273. aggregate_id: "ses_test",
  274. seq: 6,
  275. created: 106,
  276. type: "session.skill.activated.1",
  277. data: { sessionID: "ses_test", id: "effect-id", name: "Effect", text: "Use" },
  278. },
  279. ])
  280. expect(yield* db.get(sql`SELECT * FROM event_sequence`)).toEqual({
  281. aggregate_id: "ses_test",
  282. seq: 9,
  283. owner_id: "owner",
  284. })
  285. expect(yield* db.get(sql`SELECT * FROM instruction_checkpoint`)).toEqual({
  286. session_id: "ses_test",
  287. baseline: "baseline",
  288. snapshot: '{"source":"value"}',
  289. baseline_seq: 7,
  290. })
  291. }),
  292. )
  293. })
  294. test("resets incompatible V2 Session event history", async () => {
  295. await run(
  296. Effect.gen(function* () {
  297. const db = yield* makeDb
  298. yield* db.run(sql`CREATE TABLE session_input (id text PRIMARY KEY)`)
  299. yield* db.run(sql`CREATE TABLE session_message (id text PRIMARY KEY)`)
  300. yield* db.run(sql`CREATE TABLE event (id text PRIMARY KEY)`)
  301. yield* db.run(sql`CREATE TABLE event_sequence (aggregate_id text PRIMARY KEY, seq integer NOT NULL)`)
  302. yield* db.run(sql`INSERT INTO session_input (id) VALUES ('input')`)
  303. yield* db.run(sql`INSERT INTO session_message (id) VALUES ('message')`)
  304. yield* db.run(sql`INSERT INTO event (id) VALUES ('event')`)
  305. yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES ('session', 1)`)
  306. yield* DatabaseMigration.applyOnly(db, [resetSessionEventsMigration])
  307. expect(yield* db.get(sql`SELECT id FROM session_input`)).toBeUndefined()
  308. expect(yield* db.get(sql`SELECT id FROM session_message`)).toBeUndefined()
  309. expect(yield* db.get(sql`SELECT id FROM event`)).toBeUndefined()
  310. expect(yield* db.get(sql`SELECT aggregate_id FROM event_sequence`)).toBeUndefined()
  311. }),
  312. )
  313. })
  314. test("serializes concurrent embedded initialization for one database path", async () => {
  315. await using tmp = await tmpdir()
  316. const filename = path.join(tmp.path, "embedded.sqlite")
  317. const layers = [Database.layerFromPath(filename), Database.layerFromPath(filename)]
  318. await Effect.runPromise(
  319. Effect.all(
  320. layers.map((layer) => Effect.scoped(Layer.build(layer))),
  321. { concurrency: "unbounded" },
  322. ),
  323. )
  324. })
  325. if (process.platform === "linux") {
  326. test("declared schema has no ungenerated migrations", async () => {
  327. const result = await $`bun ${fileURLToPath(new URL("../script/migration.ts", import.meta.url))} --check`
  328. .quiet()
  329. .nothrow()
  330. expect(result.exitCode, result.stderr.toString()).toBe(0)
  331. expect(result.stdout.toString()).toContain("No schema changes, nothing to migrate")
  332. }, 30_000)
  333. }
  334. test("applies tracked migrations to an empty database", async () => {
  335. await run(
  336. Effect.gen(function* () {
  337. const db = yield* makeDb
  338. yield* DatabaseMigration.apply(db)
  339. expect(yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session'`)).toEqual({
  340. name: "session",
  341. })
  342. expect(
  343. yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_input'`),
  344. ).toBeUndefined()
  345. expect(
  346. yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_pending'`),
  347. ).toEqual({ name: "session_pending" })
  348. expect(
  349. yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'instruction_checkpoint'`),
  350. ).toEqual({ name: "instruction_checkpoint" })
  351. expect(
  352. yield* db.get(
  353. sql`SELECT name FROM pragma_table_info('instruction_checkpoint') WHERE name IN ('agent', 'replacement_seq', 'revision')`,
  354. ),
  355. ).toBeUndefined()
  356. expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: migrations.length })
  357. expect(
  358. yield* db.all(
  359. sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_pending_type_delivery_seq_idx', 'session_input_session_pending_compaction_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_pending_session_delivery_seq_idx', 'session_pending_session_compaction_idx', 'session_pending_session_admitted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
  360. ),
  361. ).toEqual([
  362. { name: "event_aggregate_seq_idx" },
  363. { name: "event_aggregate_type_seq_idx" },
  364. { name: "session_message_session_seq_idx" },
  365. { name: "session_message_session_time_created_id_idx" },
  366. { name: "session_message_session_type_seq_idx" },
  367. { name: "session_pending_session_admitted_seq_idx" },
  368. { name: "session_pending_session_compaction_idx" },
  369. { name: "session_pending_session_delivery_seq_idx" },
  370. ])
  371. }),
  372. )
  373. })
  374. test("rejects a non-empty database without a session table", async () => {
  375. await expect(
  376. run(
  377. Effect.gen(function* () {
  378. const db = yield* makeDb
  379. yield* db.run(sql`CREATE TABLE unrelated (id text PRIMARY KEY)`)
  380. yield* DatabaseMigration.apply(db)
  381. }),
  382. ),
  383. ).rejects.toThrow("Database is not empty and has no session table")
  384. })
  385. test("backfills existing Context Epoch rows to the build agent", async () => {
  386. await run(
  387. Effect.gen(function* () {
  388. const db = yield* makeDb
  389. yield* db.run(
  390. sql`CREATE TABLE session_context_epoch (session_id text PRIMARY KEY, baseline text NOT NULL, snapshot text NOT NULL, baseline_seq integer NOT NULL, replacement_seq integer, revision integer DEFAULT 0 NOT NULL)`,
  391. )
  392. yield* db.run(
  393. sql`INSERT INTO session_context_epoch (session_id, baseline, snapshot, baseline_seq) VALUES ('ses_existing', 'baseline', '{}', 0)`,
  394. )
  395. yield* DatabaseMigration.applyOnly(db, [contextEpochAgentMigration])
  396. expect(yield* db.get(sql`SELECT agent FROM session_context_epoch WHERE session_id = 'ses_existing'`)).toEqual({
  397. agent: "build",
  398. })
  399. }),
  400. )
  401. })
  402. test("separates existing fork provenance from subagent hierarchy", async () => {
  403. await run(
  404. Effect.gen(function* () {
  405. const db = yield* makeDb
  406. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, parent_id text)`)
  407. yield* db.run(
  408. sql`CREATE TABLE event (aggregate_id text NOT NULL, seq integer NOT NULL, type text NOT NULL, data text NOT NULL)`,
  409. )
  410. yield* db.run(sql`INSERT INTO session VALUES ('ses_source', NULL), ('ses_fork', 'ses_source')`)
  411. yield* db.run(
  412. sql`INSERT INTO event VALUES ('ses_fork', 0, 'session.forked', '{"sessionID":"ses_fork","parentID":"ses_source","from":"msg_boundary"}')`,
  413. )
  414. yield* DatabaseMigration.applyOnly(db, [addSessionForkMigration])
  415. expect(
  416. yield* db.get(sql`SELECT parent_id, fork_session_id, fork_message_id FROM session WHERE id = 'ses_fork'`),
  417. ).toEqual({
  418. parent_id: null,
  419. fork_session_id: "ses_source",
  420. fork_message_id: "msg_boundary",
  421. })
  422. expect(
  423. yield* db.get(sql`SELECT parent_id, fork_session_id, fork_message_id FROM session WHERE id = 'ses_source'`),
  424. ).toEqual({
  425. parent_id: null,
  426. fork_session_id: null,
  427. fork_message_id: null,
  428. })
  429. }),
  430. )
  431. })
  432. test("does not infer restart continuation from historical shutdown events", async () => {
  433. await run(
  434. Effect.gen(function* () {
  435. const db = yield* makeDb
  436. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
  437. yield* db.run(
  438. sql`CREATE TABLE event (aggregate_id text NOT NULL, seq integer NOT NULL, type text NOT NULL, data text NOT NULL)`,
  439. )
  440. yield* db.run(sql`INSERT INTO session VALUES ('ses_shutdown')`)
  441. yield* db.run(
  442. sql`INSERT INTO event VALUES ('ses_shutdown', 0, 'session.execution.interrupted.1', '{"reason":"shutdown"}')`,
  443. )
  444. yield* DatabaseMigration.applyOnly(db, [timeSuspendedMigration])
  445. expect(yield* db.get(sql`SELECT time_suspended FROM session WHERE id = 'ses_shutdown'`)).toEqual({
  446. time_suspended: null,
  447. })
  448. }),
  449. )
  450. })
  451. test("renames instruction state without losing rows or durable updates", async () => {
  452. await run(
  453. Effect.gen(function* () {
  454. const db = yield* makeDb
  455. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
  456. yield* db.run(
  457. sql`CREATE TABLE session_context_entry (session_id text NOT NULL, key text NOT NULL, value text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, PRIMARY KEY(session_id, key))`,
  458. )
  459. yield* db.run(
  460. sql`CREATE TABLE session_context_epoch (session_id text PRIMARY KEY, baseline text NOT NULL, snapshot text NOT NULL, baseline_seq integer NOT NULL)`,
  461. )
  462. yield* db.run(sql`CREATE TABLE event (type text NOT NULL)`)
  463. yield* db.run(sql`INSERT INTO session_context_entry VALUES ('ses_test', 'plan', '"ready"', 1, 2)`)
  464. yield* db.run(sql`INSERT INTO session_context_epoch VALUES ('ses_test', 'baseline', '{}', 7)`)
  465. yield* db.run(sql`INSERT INTO event VALUES ('session.context.updated.1')`)
  466. yield* DatabaseMigration.applyOnly(db, [renameInstructionsMigration])
  467. expect(yield* db.get(sql`SELECT * FROM instruction_entry`)).toEqual({
  468. session_id: "ses_test",
  469. key: "plan",
  470. value: '"ready"',
  471. time_created: 1,
  472. time_updated: 2,
  473. })
  474. expect(yield* db.get(sql`SELECT * FROM instruction_checkpoint`)).toEqual({
  475. session_id: "ses_test",
  476. baseline: "baseline",
  477. snapshot: "{}",
  478. baseline_seq: 7,
  479. })
  480. expect(yield* db.get(sql`SELECT type FROM event`)).toEqual({ type: "session.instructions.updated.1" })
  481. }),
  482. )
  483. })
  484. test("keeps legacy credential fields nullable", async () => {
  485. await run(
  486. Effect.gen(function* () {
  487. const db = yield* makeDb
  488. yield* db.run(
  489. sql`CREATE TABLE credential (id text PRIMARY KEY, connector_id text NOT NULL, method_id text NOT NULL, label text NOT NULL, value text NOT NULL, active integer DEFAULT false NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL)`,
  490. )
  491. yield* db.run(
  492. sql`CREATE UNIQUE INDEX credential_connector_active_idx ON credential (connector_id) WHERE active = 1`,
  493. )
  494. yield* DatabaseMigration.applyOnly(db, [simplifyIntegrationCredentialsMigration])
  495. yield* db.run(
  496. sql`INSERT INTO credential (id, connector_id, method_id, label, value, active, time_created, time_updated) VALUES ('legacy', 'openai', 'oauth', 'Legacy', '{}', 1, 1, 1)`,
  497. )
  498. yield* db.run(
  499. sql`INSERT INTO credential (id, integration_id, label, value, time_created, time_updated) VALUES ('current', 'anthropic', 'Current', '{}', 2, 2)`,
  500. )
  501. expect(yield* db.get(sql`SELECT connector_id, method_id, active FROM credential WHERE id = 'current'`)).toEqual(
  502. { connector_id: null, method_id: null, active: null },
  503. )
  504. }),
  505. )
  506. })
  507. test("resets beta history and rebuilds event-sourced Session input storage", async () => {
  508. await run(
  509. Effect.gen(function* () {
  510. const db = yield* makeDb
  511. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, workspace_id text)`)
  512. yield* db.run(sql`CREATE TABLE workspace (id text PRIMARY KEY)`)
  513. yield* db.run(sql`CREATE TABLE message (id text PRIMARY KEY)`)
  514. yield* db.run(sql`CREATE TABLE part (id text PRIMARY KEY)`)
  515. yield* db.run(sql`CREATE TABLE event_sequence (aggregate_id text PRIMARY KEY, seq integer NOT NULL)`)
  516. yield* db.run(
  517. sql`CREATE TABLE event (id text PRIMARY KEY, aggregate_id text NOT NULL, seq integer NOT NULL, type text NOT NULL, data text NOT NULL)`,
  518. )
  519. yield* db.run(sql`CREATE INDEX event_aggregate_seq_idx ON event (aggregate_id, seq)`)
  520. yield* db.run(sql`CREATE INDEX event_aggregate_type_seq_idx ON event (aggregate_id, type, seq)`)
  521. yield* db.run(
  522. sql`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)`,
  523. )
  524. yield* db.run(sql`CREATE INDEX session_message_session_seq_idx ON session_message (session_id, seq)`)
  525. yield* db.run(
  526. sql`CREATE TABLE session_input (seq integer PRIMARY KEY AUTOINCREMENT, id text NOT NULL UNIQUE, session_id text NOT NULL, prompt text NOT NULL, delivery text NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
  527. )
  528. yield* db.run(
  529. sql`CREATE INDEX session_input_session_pending_delivery_seq_idx ON session_input (session_id, promoted_seq, delivery, seq)`,
  530. )
  531. yield* db.run(sql`INSERT INTO session (id, workspace_id) VALUES ('session', 'wrk_old')`)
  532. yield* db.run(sql`INSERT INTO workspace (id) VALUES ('wrk_old')`)
  533. yield* db.run(sql`INSERT INTO message (id) VALUES ('message')`)
  534. yield* db.run(sql`INSERT INTO part (id) VALUES ('part')`)
  535. yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES ('session', 0)`)
  536. yield* db.run(
  537. sql`INSERT INTO event (id, aggregate_id, seq, type, data) VALUES ('evt_old', 'session', 0, 'old.1', '{}')`,
  538. )
  539. yield* db.run(
  540. sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('msg_old', 'session', 'user', 0, 1, 1, '{}')`,
  541. )
  542. yield* db.run(
  543. sql`INSERT INTO session_input (id, session_id, prompt, delivery, time_created) VALUES ('msg_pending', 'session', '{}', 'steer', 1)`,
  544. )
  545. yield* DatabaseMigration.applyOnly(db, [eventSourcedSessionPendingMigration])
  546. expect(yield* db.all(sql`SELECT id, workspace_id FROM session`)).toEqual([
  547. { id: "session", workspace_id: null },
  548. ])
  549. expect(yield* db.all(sql`SELECT id FROM workspace`)).toEqual([])
  550. expect(yield* db.all(sql`SELECT id FROM message`)).toEqual([{ id: "message" }])
  551. expect(yield* db.all(sql`SELECT id FROM part`)).toEqual([{ id: "part" }])
  552. expect(yield* db.all(sql`SELECT id FROM event`)).toEqual([])
  553. expect(yield* db.all(sql`SELECT aggregate_id FROM event_sequence`)).toEqual([])
  554. expect(yield* db.all(sql`SELECT id FROM session_message`)).toEqual([])
  555. expect(yield* db.all(sql`SELECT id FROM session_input`)).toEqual([])
  556. expect(
  557. (yield* db.all<{ name: string }>(sql`PRAGMA table_info(session_input)`)).map((column) => column.name),
  558. ).toEqual(["id", "session_id", "prompt", "delivery", "admitted_seq", "promoted_seq", "time_created"])
  559. expect(
  560. (yield* db.all<{ name: string; unique: number }>(sql`PRAGMA index_list(session_message)`)).find(
  561. (index) => index.name === "session_message_session_seq_idx",
  562. ),
  563. ).toMatchObject({ unique: 1 })
  564. expect(
  565. (yield* db.all<{ name: string; unique: number }>(sql`PRAGMA index_list(event)`)).find(
  566. (index) => index.name === "event_aggregate_seq_idx",
  567. ),
  568. ).toMatchObject({ unique: 1 })
  569. expect(
  570. (yield* db.all<{ name: string; unique: number }>(sql`PRAGMA index_list(session_input)`)).filter((index) =>
  571. ["session_input_session_admitted_seq_idx", "session_input_session_promoted_seq_idx"].includes(index.name),
  572. ),
  573. ).toEqual([
  574. expect.objectContaining({ name: "session_input_session_promoted_seq_idx", unique: 1 }),
  575. expect.objectContaining({ name: "session_input_session_admitted_seq_idx", unique: 1 }),
  576. ])
  577. }),
  578. )
  579. })
  580. test("preserves canonical V1 state and restarts its event stream", async () => {
  581. await run(
  582. Effect.gen(function* () {
  583. const db = yield* makeDb
  584. yield* db.run(sql`PRAGMA foreign_keys = ON`)
  585. yield* DatabaseMigration.apply(db)
  586. yield* db.run(
  587. sql`INSERT INTO project (id, worktree, time_created, time_updated, sandboxes) VALUES ('global', '/project', 1, 1, '[]')`,
  588. )
  589. yield* db.run(
  590. sql`INSERT INTO workspace (id, type, project_id, time_used) VALUES ('workspace', 'local', 'global', 1)`,
  591. )
  592. yield* db.run(
  593. sql`INSERT INTO session (id, project_id, workspace_id, slug, directory, title, version, time_created, time_updated) VALUES ('session', 'global', 'workspace', 'session', '/project', 'Before', 'test', 1, 1)`,
  594. )
  595. yield* db.run(
  596. sql`INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES ('message', 'session', 1, 1, '{}')`,
  597. )
  598. yield* db.run(
  599. sql`INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES ('part', 'message', 'session', 1, 1, '{}')`,
  600. )
  601. yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES ('session', 9)`)
  602. yield* db.run(
  603. sql`INSERT INTO event (id, aggregate_id, seq, type, data, created) VALUES ('event', 'session', 9, 'session.updated.1', '{}', 1)`,
  604. )
  605. yield* db.run(
  606. sql`INSERT INTO session_pending (id, session_id, type, data, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'user', '{}', 'steer', 9, 1)`,
  607. )
  608. yield* db.run(
  609. sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('projected', 'session', 'user', 9, 1, 1, '{}')`,
  610. )
  611. yield* db.run(
  612. sql`INSERT INTO instruction_checkpoint (session_id, baseline, snapshot, baseline_seq) VALUES ('session', 'baseline', '{}', 9)`,
  613. )
  614. yield* db.run(sql`ALTER TABLE instruction_checkpoint RENAME TO session_context_epoch`)
  615. // The partial compaction index embeds the qualified table name, so it
  616. // must drop before the historical rename dance and recreate after.
  617. yield* db.run(sql`DROP INDEX session_pending_session_compaction_idx`)
  618. yield* db.run(sql`ALTER TABLE session_pending RENAME TO session_input`)
  619. yield* db.run(sql`DELETE FROM migration WHERE id = ${simplifySessionPendingMigration.id}`)
  620. yield* DatabaseMigration.applyOnly(db, [simplifySessionPendingMigration])
  621. yield* db.run(sql`ALTER TABLE session_context_epoch RENAME TO instruction_checkpoint`)
  622. yield* db.run(sql`ALTER TABLE session_input RENAME TO session_pending`)
  623. yield* db.run(
  624. sql`CREATE UNIQUE INDEX session_pending_session_compaction_idx ON session_pending (session_id) WHERE "session_pending"."type" = 'compaction'`,
  625. )
  626. const database = Layer.succeed(Database.Service, { db })
  627. yield* EventV2.Service.use((service) =>
  628. service.publish(SessionV1.Event.Updated, {
  629. sessionID: SessionSchema.ID.make("session"),
  630. info: {
  631. id: SessionSchema.ID.make("session"),
  632. slug: "session",
  633. projectID: ProjectV2.ID.global,
  634. directory: "/project",
  635. title: "After",
  636. version: "test",
  637. time: { created: 1, updated: 2 },
  638. },
  639. }),
  640. ).pipe(
  641. Effect.provide(
  642. AppNodeBuilder.build(LayerNode.group([EventV2.node, SessionProjector.node]), [[Database.node, database]]),
  643. ),
  644. )
  645. expect(
  646. yield* db.get(sql`
  647. SELECT
  648. (SELECT title FROM session WHERE id = 'session') AS title,
  649. (SELECT workspace_id FROM session WHERE id = 'session') AS workspaceID,
  650. (SELECT COUNT(*) FROM message WHERE id = 'message') AS messages,
  651. (SELECT COUNT(*) FROM part WHERE id = 'part') AS parts,
  652. (SELECT COUNT(*) FROM workspace) AS workspaces,
  653. (SELECT COUNT(*) FROM session_pending) AS sessionInputs,
  654. (SELECT COUNT(*) FROM session_message) AS sessionMessages,
  655. (SELECT COUNT(*) FROM instruction_checkpoint) AS instructionCheckpoints,
  656. (SELECT seq FROM event_sequence WHERE aggregate_id = 'session') AS seq,
  657. (SELECT type FROM event WHERE aggregate_id = 'session') AS eventType
  658. `),
  659. ).toEqual({
  660. title: "After",
  661. workspaceID: null,
  662. messages: 1,
  663. parts: 1,
  664. workspaces: 0,
  665. sessionInputs: 0,
  666. sessionMessages: 0,
  667. instructionCheckpoints: 0,
  668. seq: 0,
  669. eventType: "session.updated.1",
  670. })
  671. }),
  672. )
  673. })
  674. test("preserves admitted prompts while generalizing the durable inbox", async () => {
  675. await run(
  676. Effect.gen(function* () {
  677. const db = yield* makeDb
  678. yield* db.run(
  679. sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL, prompt text NOT NULL, delivery text NOT NULL, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
  680. )
  681. yield* db.run(
  682. sql`INSERT INTO session_input (id, session_id, prompt, delivery, admitted_seq, promoted_seq, time_created) VALUES ('input', 'session', '{"text":"hello"}', 'steer', 4, NULL, 1)`,
  683. )
  684. yield* DatabaseMigration.applyOnly(db, [durableSessionInboxMigration])
  685. expect(
  686. yield* db.all(
  687. sql`SELECT id, type, prompt, delivery, admitted_seq, promoted_seq FROM session_input ORDER BY admitted_seq`,
  688. ),
  689. ).toEqual([
  690. {
  691. id: "input",
  692. type: "prompt",
  693. prompt: '{"text":"hello"}',
  694. delivery: "steer",
  695. admitted_seq: 4,
  696. promoted_seq: null,
  697. },
  698. ])
  699. }),
  700. )
  701. })
  702. test("migrates prompt inbox rows and lifecycle events to generic user input", async () => {
  703. await run(
  704. Effect.gen(function* () {
  705. const db = yield* makeDb
  706. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
  707. yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`)
  708. yield* db.run(
  709. sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, prompt text, delivery text, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
  710. )
  711. yield* db.run(
  712. sql`INSERT INTO session_input (id, session_id, type, prompt, delivery, admitted_seq, promoted_seq, time_created) VALUES ('input', 'session', 'prompt', '{"text":"hello"}', 'queue', 4, NULL, 1)`,
  713. )
  714. yield* db.run(
  715. sql`INSERT INTO session_input (id, session_id, type, prompt, delivery, admitted_seq, promoted_seq, time_created) VALUES ('empty', 'session', 'prompt', NULL, 'steer', 6, NULL, 2)`,
  716. )
  717. yield* db.run(
  718. sql`CREATE TABLE event (id text PRIMARY KEY, aggregate_id text NOT NULL, seq integer NOT NULL, created integer NOT NULL, type text NOT NULL, data text NOT NULL)`,
  719. )
  720. yield* db.run(
  721. sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('admitted', 'session', 4, 1, 'session.prompt.admitted.1', '{"sessionID":"session","inputID":"input","prompt":{"text":"hello"},"delivery":"queue"}')`,
  722. )
  723. yield* db.run(
  724. sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('promoted', 'session', 5, 2, 'session.prompt.promoted.1', '{"sessionID":"session","inputID":"input"}')`,
  725. )
  726. yield* db.run(
  727. sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('empty-admitted', 'session', 6, 2, 'session.prompt.admitted.1', '{"sessionID":"session","inputID":"empty","prompt":null,"delivery":"steer"}')`,
  728. )
  729. yield* db.run(
  730. sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('empty-promoted', 'session', 7, 2, 'session.prompt.promoted.1', '{"sessionID":"session","inputID":"empty"}')`,
  731. )
  732. yield* DatabaseMigration.applyOnly(db, [genericSessionPendingMigration])
  733. expect(yield* db.all(sql`SELECT id, type, data, delivery FROM session_input ORDER BY admitted_seq`)).toEqual([
  734. { id: "input", type: "user", data: '{"text":"hello"}', delivery: "queue" },
  735. ])
  736. expect(yield* db.all(sql`SELECT type, data FROM event ORDER BY seq`)).toEqual([
  737. {
  738. type: "session.input.admitted.1",
  739. data: '{"sessionID":"session","inputID":"input","input":{"type":"user","data":{"text":"hello"},"delivery":"queue"}}',
  740. },
  741. {
  742. type: "session.input.promoted.1",
  743. data: '{"sessionID":"session","inputID":"input"}',
  744. },
  745. ])
  746. }),
  747. )
  748. })
  749. test("replaces the durable inbox with the empty session_pending table", async () => {
  750. await run(
  751. Effect.gen(function* () {
  752. const db = yield* makeDb
  753. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
  754. yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`)
  755. yield* db.run(
  756. sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL REFERENCES session(id) ON DELETE CASCADE, type text NOT NULL, data text NOT NULL, delivery text, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
  757. )
  758. // Interim v2 builds shipped differing index sets on real databases;
  759. // dropping the table removes whatever variant exists.
  760. yield* db.run(
  761. sql`CREATE INDEX session_input_session_pending_type_delivery_seq_idx ON session_input (session_id, promoted_seq, type, delivery, admitted_seq)`,
  762. )
  763. yield* db.run(
  764. sql`INSERT INTO session_input (id, session_id, type, data, delivery, admitted_seq, promoted_seq, time_created) VALUES ('pending', 'session', 'user', '{"text":"hello"}', 'steer', 4, NULL, 1)`,
  765. )
  766. yield* DatabaseMigration.applyOnly(db, [sessionPendingTableMigration])
  767. expect(
  768. yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_input'`),
  769. ).toBeUndefined()
  770. expect(yield* db.all(sql`SELECT id FROM session_pending`)).toEqual([])
  771. expect(
  772. (yield* db.all<{ name: string }>(sql`PRAGMA table_info(session_pending)`)).map((column) => column.name),
  773. ).toEqual(["id", "session_id", "type", "data", "delivery", "admitted_seq", "time_created"])
  774. expect(
  775. (yield* db.all<{ name: string; unique: number }>(sql`PRAGMA index_list(session_pending)`))
  776. .filter((index) => index.name.startsWith("session_"))
  777. .map((index) => ({ name: index.name, unique: index.unique }))
  778. .sort((a, b) => a.name.localeCompare(b.name)),
  779. ).toEqual([
  780. { name: "session_pending_session_admitted_seq_idx", unique: 1 },
  781. { name: "session_pending_session_compaction_idx", unique: 1 },
  782. { name: "session_pending_session_delivery_seq_idx", unique: 0 },
  783. ])
  784. }),
  785. )
  786. })
  787. test("resets incompatible projected Session messages before adding sequence order", async () => {
  788. await run(
  789. Effect.gen(function* () {
  790. const db = yield* makeDb
  791. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
  792. yield* db.run(
  793. sql`CREATE TABLE message (id text PRIMARY KEY, session_id text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL)`,
  794. )
  795. yield* db.run(
  796. sql`CREATE TABLE part (id text PRIMARY KEY, message_id text NOT NULL, session_id text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL)`,
  797. )
  798. yield* db.run(sql`CREATE TABLE event (id text PRIMARY KEY, seq integer NOT NULL)`)
  799. yield* db.run(
  800. sql`CREATE TABLE session_message (id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, time_created integer NOT NULL, time_updated integer NOT NULL, data text NOT NULL)`,
  801. )
  802. yield* db.run(
  803. sql`CREATE INDEX session_message_session_time_created_id_idx ON session_message (session_id, time_created, id)`,
  804. )
  805. yield* db.run(
  806. sql`CREATE INDEX session_message_session_type_time_created_id_idx ON session_message (session_id, type, time_created, id)`,
  807. )
  808. yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`)
  809. yield* db.run(
  810. sql`INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES ('legacy_message', 'session', 1, 1, '{"role":"user"}')`,
  811. )
  812. yield* db.run(
  813. sql`INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES ('legacy_part', 'legacy_message', 'session', 1, 1, '{"type":"text","text":"hello"}')`,
  814. )
  815. yield* db.run(
  816. sql`INSERT INTO session_message (id, session_id, type, time_created, time_updated, data) VALUES ('stale_projection', 'session', 'user', 1, 1, '{}')`,
  817. )
  818. yield* DatabaseMigration.applyOnly(db, [sessionMessageProjectionOrderMigration])
  819. expect(yield* db.all(sql`SELECT id, session_id, data FROM message`)).toEqual([
  820. { id: "legacy_message", session_id: "session", data: '{"role":"user"}' },
  821. ])
  822. expect(yield* db.all(sql`SELECT id, message_id, session_id, data FROM part`)).toEqual([
  823. {
  824. id: "legacy_part",
  825. message_id: "legacy_message",
  826. session_id: "session",
  827. data: '{"type":"text","text":"hello"}',
  828. },
  829. ])
  830. expect(yield* db.all(sql`SELECT id FROM session_message`)).toEqual([])
  831. yield* db.run(
  832. sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('fresh_projection', 'session', 'user', 7, 2, 2, '{}')`,
  833. )
  834. expect(yield* db.get(sql`SELECT id, seq FROM session_message`)).toEqual({ id: "fresh_projection", seq: 7 })
  835. }),
  836. )
  837. })
  838. test("runs session usage backfill in order with schema changes", async () => {
  839. await run(
  840. Effect.gen(function* () {
  841. const db = yield* makeDb
  842. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, time_updated integer NOT NULL)`)
  843. yield* db.run(sql`CREATE TABLE message (id text PRIMARY KEY, session_id text NOT NULL, data text NOT NULL)`)
  844. yield* db.run(sql`INSERT INTO session (id, time_updated) VALUES ('session_1', 1)`)
  845. yield* db.run(
  846. sql`INSERT INTO message (id, session_id, data) VALUES ('message_1', 'session_1', '{"role":"assistant","cost":1.25,"tokens":{"input":2,"output":3,"reasoning":4,"cache":{"read":5,"write":6}}}')`,
  847. )
  848. yield* DatabaseMigration.applyOnly(db, [sessionUsageMigration])
  849. expect(
  850. yield* db.get(
  851. sql`SELECT cost, tokens_input, tokens_output, tokens_reasoning, tokens_cache_read, tokens_cache_write FROM session WHERE id = 'session_1'`,
  852. ),
  853. ).toEqual({
  854. cost: 1.25,
  855. tokens_input: 2,
  856. tokens_output: 3,
  857. tokens_reasoning: 4,
  858. tokens_cache_read: 5,
  859. tokens_cache_write: 6,
  860. })
  861. }),
  862. )
  863. })
  864. test("normalizes Windows storage paths and leaves POSIX paths untouched", async () => {
  865. await run(
  866. Effect.gen(function* () {
  867. const db = yield* makeDb
  868. yield* db.run(sql`CREATE TABLE project (id text PRIMARY KEY, worktree text NOT NULL, sandboxes text NOT NULL)`)
  869. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, directory text NOT NULL, path text)`)
  870. // Windows-shaped rows (drive + backslash) must be normalized.
  871. yield* db.run(
  872. sql`INSERT INTO project (id, worktree, sandboxes) VALUES (${"win"}, ${"C:\\Repo\\Thing"}, ${JSON.stringify([
  873. "C:\\Repo\\Thing\\sandbox",
  874. ])})`,
  875. )
  876. yield* db.run(
  877. sql`INSERT INTO session (id, directory, path) VALUES (${"win"}, ${"C:\\Repo\\Thing\\packages\\api"}, ${"packages\\api"})`,
  878. )
  879. // UNC worktrees and their sandboxes must normalize too (not just drive paths).
  880. yield* db.run(
  881. sql`INSERT INTO project (id, worktree, sandboxes) VALUES (${"unc"}, ${"\\\\server\\share"}, ${JSON.stringify([
  882. "\\\\server\\share\\sandbox",
  883. ])})`,
  884. )
  885. // The "/" worktree sentinel and POSIX paths (including a pathological
  886. // backslash in a POSIX filename) must survive byte-for-byte.
  887. yield* db.run(sql`INSERT INTO project (id, worktree, sandboxes) VALUES (${"global"}, ${"/"}, ${"[]"})`)
  888. yield* db.run(
  889. sql`INSERT INTO session (id, directory, path) VALUES (${"posix"}, ${"/home/me/we\\ird"}, ${"src\\weird"})`,
  890. )
  891. yield* DatabaseMigration.applyOnly(db, [normalizeStoragePathsMigration])
  892. expect(yield* db.get(sql`SELECT worktree, sandboxes FROM project WHERE id = 'win'`)).toEqual({
  893. worktree: "C:/Repo/Thing",
  894. sandboxes: JSON.stringify(["C:/Repo/Thing/sandbox"]),
  895. })
  896. expect(yield* db.get(sql`SELECT directory, path FROM session WHERE id = 'win'`)).toEqual({
  897. directory: "C:/Repo/Thing/packages/api",
  898. path: "packages/api",
  899. })
  900. expect(yield* db.get(sql`SELECT worktree, sandboxes FROM project WHERE id = 'unc'`)).toEqual({
  901. worktree: "//server/share",
  902. sandboxes: JSON.stringify(["//server/share/sandbox"]),
  903. })
  904. expect(yield* db.get(sql`SELECT worktree FROM project WHERE id = 'global'`)).toEqual({ worktree: "/" })
  905. expect(yield* db.get(sql`SELECT directory, path FROM session WHERE id = 'posix'`)).toEqual({
  906. directory: "/home/me/we\\ird",
  907. path: "src\\weird",
  908. })
  909. }),
  910. )
  911. })
  912. test("maps native Windows paths through database columns", async () => {
  913. if (process.platform !== "win32") return
  914. await run(
  915. Effect.gen(function* () {
  916. const db = yield* makeDb
  917. yield* DatabaseMigration.apply(db)
  918. const projectID = ProjectV2.ID.make("codec_project")
  919. const worktree = AbsolutePath.make("C:\\Repo\\Thing")
  920. const sandbox = AbsolutePath.make("C:\\Repo\\Thing\\sandbox")
  921. const directory = "C:\\Repo\\Thing\\packages\\api"
  922. const sessionID = SessionSchema.ID.make("ses_codec")
  923. expect(() =>
  924. Effect.runSync(
  925. db
  926. .insert(ProjectTable)
  927. .values({
  928. id: ProjectV2.ID.make("invalid_path"),
  929. worktree: AbsolutePath.make("not-absolute"),
  930. sandboxes: [],
  931. time_created: 1,
  932. time_updated: 1,
  933. })
  934. .run(),
  935. ),
  936. ).toThrow()
  937. yield* db
  938. .insert(ProjectTable)
  939. .values({
  940. id: projectID,
  941. worktree,
  942. sandboxes: [sandbox],
  943. time_created: 1,
  944. time_updated: 1,
  945. })
  946. .run()
  947. yield* db
  948. .insert(SessionTable)
  949. .values({
  950. id: sessionID,
  951. project_id: projectID,
  952. slug: "codec",
  953. directory,
  954. path: "packages\\api",
  955. title: "Codec",
  956. version: "test",
  957. time_created: 1,
  958. time_updated: 1,
  959. })
  960. .run()
  961. expect(
  962. yield* db.get<{ worktree: string; sandboxes: string }>(
  963. sql`SELECT worktree, sandboxes FROM project WHERE id = ${projectID}`,
  964. ),
  965. ).toEqual({
  966. worktree: "C:/Repo/Thing",
  967. sandboxes: JSON.stringify(["C:/Repo/Thing/sandbox"]),
  968. })
  969. expect(
  970. yield* db.get<{ directory: string; path: string }>(
  971. sql`SELECT directory, path FROM session WHERE id = ${sessionID}`,
  972. ),
  973. ).toEqual({
  974. directory: "C:/Repo/Thing/packages/api",
  975. path: "packages/api",
  976. })
  977. const project = yield* db.select().from(ProjectTable).where(eq(ProjectTable.worktree, worktree)).get()
  978. const session = yield* db.select().from(SessionTable).where(eq(SessionTable.directory, directory)).get()
  979. expect(project?.worktree).toBe(worktree)
  980. expect(project?.sandboxes).toEqual([sandbox])
  981. expect(session?.directory).toBe(directory)
  982. expect(session?.path).toBe("packages/api")
  983. expect((yield* db.select().from(SessionTable).where(eq(SessionTable.path, "packages\\api")).get())?.id).toBe(
  984. sessionID,
  985. )
  986. const moved = AbsolutePath.make("D:\\Moved\\Thing")
  987. const updated = yield* db
  988. .update(ProjectTable)
  989. .set({ worktree: moved, sandboxes: [moved] })
  990. .where(eq(ProjectTable.id, projectID))
  991. .returning()
  992. .get()
  993. expect(updated?.worktree).toBe(moved)
  994. expect(updated?.sandboxes).toEqual([moved])
  995. expect(
  996. yield* db.get<{ worktree: string; sandboxes: string }>(
  997. sql`SELECT worktree, sandboxes FROM project WHERE id = ${projectID}`,
  998. ),
  999. ).toEqual({ worktree: "D:/Moved/Thing", sandboxes: JSON.stringify(["D:/Moved/Thing"]) })
  1000. expect(
  1001. (yield* db
  1002. .select()
  1003. .from(ProjectTable)
  1004. .where(inArray(ProjectTable.worktree, [moved]))
  1005. .get())?.id,
  1006. ).toBe(projectID)
  1007. yield* db.run(sql`UPDATE project SET worktree = ${"not-absolute"} WHERE id = ${projectID}`)
  1008. expect(() =>
  1009. Effect.runSync(db.select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()),
  1010. ).toThrow()
  1011. }),
  1012. )
  1013. })
  1014. test("imports existing drizzle migration state", async () => {
  1015. await run(
  1016. Effect.gen(function* () {
  1017. const db = yield* makeDb
  1018. yield* db.run(
  1019. sql`CREATE TABLE __drizzle_migrations (id INTEGER PRIMARY KEY, hash text NOT NULL, created_at numeric, name text, applied_at TEXT)`,
  1020. )
  1021. yield* db.run(sql`
  1022. INSERT INTO __drizzle_migrations (hash, created_at, name, applied_at)
  1023. VALUES ('hash', 1, '20260127222353_familiar_lady_ursula', ${new Date().toISOString()})
  1024. `)
  1025. yield* DatabaseMigration.applyOnly(db, [])
  1026. expect(yield* db.get(sql`SELECT id FROM migration`)).toEqual({ id: "20260127222353_familiar_lady_ursula" })
  1027. }),
  1028. )
  1029. })
  1030. test("does not replay a migrated session metadata column", async () => {
  1031. await run(
  1032. Effect.gen(function* () {
  1033. const db = yield* makeDb
  1034. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, metadata text)`)
  1035. yield* db.run(
  1036. sql`CREATE TABLE __drizzle_migrations (id INTEGER PRIMARY KEY, hash text NOT NULL, created_at numeric, name text, applied_at TEXT)`,
  1037. )
  1038. yield* db.run(sql`
  1039. INSERT INTO __drizzle_migrations (hash, created_at, name, applied_at)
  1040. VALUES ('hash', 1, '20260511173437_session-metadata', ${new Date().toISOString()})
  1041. `)
  1042. yield* DatabaseMigration.applyOnly(db, [sessionMetadataMigration])
  1043. expect(yield* db.all(sql`SELECT id FROM migration`)).toEqual([{ id: "20260511173437_session-metadata" }])
  1044. }),
  1045. )
  1046. })
  1047. test("accepts the temporary replacement session metadata migration id", async () => {
  1048. await run(
  1049. Effect.gen(function* () {
  1050. const db = yield* makeDb
  1051. yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, metadata text)`)
  1052. yield* db.run(sql`CREATE TABLE migration (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`)
  1053. yield* db.run(sql`INSERT INTO migration (id, time_completed) VALUES ('20260530232709_lovely_romulus', 1)`)
  1054. yield* DatabaseMigration.applyOnly(db, [sessionMetadataMigration])
  1055. expect(yield* db.all(sql`SELECT id FROM migration ORDER BY id`)).toEqual([
  1056. { id: "20260511173437_session-metadata" },
  1057. { id: "20260530232709_lovely_romulus" },
  1058. ])
  1059. }),
  1060. )
  1061. })
  1062. test("skips drizzle import when migration table already has state", async () => {
  1063. await run(
  1064. Effect.gen(function* () {
  1065. const db = yield* makeDb
  1066. yield* db.run(sql`CREATE TABLE migration (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`)
  1067. yield* db.run(sql`INSERT INTO migration (id, time_completed) VALUES ('existing', 1)`)
  1068. yield* db.run(
  1069. sql`CREATE TABLE __drizzle_migrations (id INTEGER PRIMARY KEY, hash text NOT NULL, created_at numeric, name text, applied_at TEXT)`,
  1070. )
  1071. yield* db.run(sql`
  1072. INSERT INTO __drizzle_migrations (hash, created_at, name, applied_at)
  1073. VALUES ('hash', 1, '20260127222353_familiar_lady_ursula', ${new Date().toISOString()})
  1074. `)
  1075. yield* DatabaseMigration.applyOnly(db, [])
  1076. expect(yield* db.all(sql`SELECT id FROM migration ORDER BY id`)).toEqual([{ id: "existing" }])
  1077. }),
  1078. )
  1079. })
  1080. })