1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
5import * as Path from "effect/Path";
6import * as SqlClient from "effect/sql/SqlClient";
7
8import { runMigrations } from "../src/persistence/Migrations.ts";
9import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
10import { runMigrateDevDb } from "./migrate-dev-db.ts";
11
12const withDatabase = <A, E>(
13 databasePath: string,
14 effect: Effect.Effect<A, E, SqlClient.SqlClient>,
15) => effect.pipe(Effect.provide(NodeSqliteClient.layer({ filename: databasePath })));
16
17/** A migrated source db with one V2 thread per lifecycle state. Only
18 * `stopped-thread` and its fork qualify for the clone. */
19const createFixtureSource = Effect.fn("createMigrateDevDbFixtureSource")(function* (
20 baseDir: string,
21) {
22 const fs = yield* FileSystem.FileSystem;
23 const path = yield* Path.Path;
24 const stateDir = path.join(baseDir, "userdata");
25 const databasePath = path.join(stateDir, "statev2.sqlite");
26 yield* fs.makeDirectory(stateDir, { recursive: true });
27 yield* withDatabase(
28 databasePath,
29 Effect.gen(function* () {
30 const sql = yield* SqlClient.SqlClient;
31 yield* runMigrations();
32
33 yield* sql`INSERT INTO projection_projects
34 (project_id, title, workspace_root, scripts_json, created_at, updated_at, deleted_at)
35 VALUES
36 ('project-kept', 'Kept', '/tmp/kept', '[]', '2026-08-01', '2026-08-01', NULL),
37 ('project-deleted', 'Deleted', '/tmp/deleted', '[]', '2026-08-01', '2026-08-02', '2026-08-02')`;
38
39 const forkPayload =
40 '{"lineage":{"parentThreadId":"stopped-thread","relationshipToParent":"fork","rootThreadId":"stopped-thread"}}';
41 const subagentPayload =
42 '{"lineage":{"parentThreadId":"subagent-parent","relationshipToParent":"subagent","rootThreadId":"subagent-parent"},"forkedFrom":{"type":"node","nodeId":"node-1"}}';
43 // Excluded threads are newer than the kept family, so only the filters
44 // can keep them out of a one-family-per-project clone.
45 const threads = [
46 ["stopped-thread", "project-kept", "completed", "{}", "2026-08-01"],
47 ["fork-thread", "project-kept", "completed", forkPayload, "2026-08-02"],
48 ["running-thread", "project-kept", "running", "{}", "2026-08-05"],
49 ["settled-thread", "project-kept", "completed", '{"settledAt":"2026-08-01"}', "2026-08-05"],
50 [
51 "limit-thread",
52 "project-kept",
53 "completed",
54 '{"limitRecovery":{"autoResume":true}}',
55 "2026-08-05",
56 ],
57 // Its result never reached the parent, so startup would deliver it.
58 ["subagent-parent", "project-kept", "completed", "{}", "2026-08-05"],
59 ["subagent-child", "project-kept", "completed", subagentPayload, "2026-08-05"],
60 ["deleted-project-thread", "project-deleted", "completed", "{}", "2026-08-05"],
61 ] as const;
62 for (const [threadId, projectId, runStatus, payload, updatedAt] of threads) {
63 yield* sql`INSERT INTO orchestration_v2_projection_threads
64 (thread_id, project_id, title, default_provider, runtime_mode, interaction_mode, created_at, updated_at, payload_json)
65 VALUES (${threadId}, ${projectId}, ${threadId}, 'codex', 'full-access', 'default', '2026-08-01', ${updatedAt}, ${payload})`;
66 yield* sql`INSERT INTO orchestration_v2_projection_runs
67 (run_id, thread_id, ordinal, provider, status, requested_at, payload_json)
68 VALUES (${`run-${threadId}`}, ${threadId}, 1, 'codex', ${runStatus}, '2026-08-01', '{}')`;
69 yield* sql`INSERT INTO orchestration_events
70 (event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, actor_kind, payload_json, metadata_json)
71 VALUES (${`event-${threadId}`}, 'thread', ${threadId}, 0, 'thread.created', '2026-08-01', 'user', '{}', '{}')`;
72 }
73 // A provider session shared by two threads names its latest writer.
74 yield* sql`INSERT INTO orchestration_v2_projection_provider_sessions
75 (provider_session_id, thread_id, provider, status, updated_at, payload_json)
76 VALUES ('session-shared', 'running-thread', 'codex', 'stopped', '2026-08-01', '{}')`;
77 yield* sql`INSERT INTO orchestration_v2_projection_provider_session_bindings
78 (provider_session_id, thread_id)
79 VALUES ('session-shared', 'running-thread'), ('session-shared', 'stopped-thread')`;
80 yield* sql`INSERT INTO orchestration_v2_projection_context_transfers
81 (context_transfer_id, source_thread_id, target_thread_id, type, status, updated_at, payload_json)
82 VALUES ('transfer-1', 'settled-thread', 'stopped-thread', 'provider_handoff', 'completed', '2026-08-01', '{}')`;
83 yield* sql`INSERT INTO scheduled_tasks
84 (task_id, title, prompt, enabled, schedule_json, project_id, workspace_strategy_json,
85 model_selection_json, runtime_mode, interaction_mode, created_by, creation_source,
86 created_at, updated_at, last_run_status, run_count)
87 VALUES ('task-1', 'Nightly', 'Run it', 1, '{}', 'project-kept', '{}', '{}',
88 'full-access', 'default', 'user', 'user', '2026-08-01', '2026-08-01', 'never', 0)`;
89 yield* sql`INSERT INTO auth_sessions (session_id, subject, scopes, method, issued_at, expires_at)
90 VALUES ('session-1', 'user', '[]', 'pairing', '2026-08-01', '2027-08-01')`;
91 }),
92 );
93 return databasePath;
94});
95
96it.layer(NodeServices.layer)("migrate-dev-db", (it) => {
97 it.effect("keeps stopped thread families from live projects and clears pending work", () =>
98 Effect.gen(function* () {
99 const fs = yield* FileSystem.FileSystem;
100 const path = yield* Path.Path;
101 const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-src-" });
102 const destDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-dest-" });
103 const source = yield* createFixtureSource(sourceDir);
104
105 const result = yield* runMigrateDevDb(
106 { baseDir: destDir, source, projects: 5, threadsPerProject: 1 },
107 { sharedHome: sourceDir },
108 );
109
110 assert.equal(result.databasePath, path.join(destDir, "userdata", "statev2.sqlite"));
111 const kept = yield* withDatabase(
112 result.databasePath,
113 Effect.gen(function* () {
114 const sql = yield* SqlClient.SqlClient;
115 const threads = yield* sql<{ thread_id: string }>`
116 SELECT thread_id FROM orchestration_v2_projection_threads ORDER BY thread_id`;
117 const events = yield* sql<{ stream_id: string }>`
118 SELECT stream_id FROM orchestration_events ORDER BY stream_id`;
119 const sessions = yield* sql<{ provider_session_id: string }>`
120 SELECT provider_session_id FROM orchestration_v2_projection_provider_sessions`;
121 const [leftovers] = yield* sql<{ auth: number; tasks: number; transfers: number }>`
122 SELECT
123 (SELECT COUNT(*) FROM auth_sessions) AS auth,
124 (SELECT COUNT(*) FROM scheduled_tasks) AS tasks,
125 (SELECT COUNT(*) FROM orchestration_v2_projection_context_transfers) AS transfers`;
126 return { threads, events, sessions, leftovers };
127 }),
128 );
129 assert.deepStrictEqual(
130 kept.threads.map((row) => row.thread_id),
131 ["fork-thread", "stopped-thread"],
132 );
133 assert.deepStrictEqual(
134 kept.events.map((row) => row.stream_id),
135 ["fork-thread", "stopped-thread"],
136 );
137 assert.deepStrictEqual(
138 kept.sessions.map((row) => row.provider_session_id),
139 ["session-shared"],
140 );
141 assert.deepStrictEqual(kept.leftovers, { auth: 0, tasks: 0, transfers: 0 });
142 }),
143 );
144
145 it.effect("never copies a settled thread's rows, and new events append after the source's", () =>
146 Effect.gen(function* () {
147 const fs = yield* FileSystem.FileSystem;
148 const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-slice-" });
149 const destDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-slice-dest-" });
150 const source = yield* createFixtureSource(sourceDir);
151 const [sourceSequence] = yield* withDatabase(
152 source,
153 Effect.gen(function* () {
154 const sql = yield* SqlClient.SqlClient;
155 return yield* sql<{ seq: number }>`
156 SELECT seq FROM sqlite_sequence WHERE name = 'orchestration_events'`;
157 }),
158 );
159
160 // Keep every family, so only the copy can leave the settled one out.
161 const result = yield* runMigrateDevDb(
162 { baseDir: destDir, source, projects: 5, threadsPerProject: 100 },
163 { sharedHome: sourceDir },
164 );
165
166 const copied = yield* withDatabase(
167 result.databasePath,
168 Effect.gen(function* () {
169 const sql = yield* SqlClient.SqlClient;
170 const [settledRows] = yield* sql<{ count: number }>`
171 SELECT
172 (SELECT COUNT(*) FROM orchestration_v2_projection_runs WHERE thread_id = 'settled-thread')
173 + (SELECT COUNT(*) FROM orchestration_events WHERE stream_id = 'settled-thread')
174 AS count`;
175 const [sequence] = yield* sql<{ seq: number }>`
176 SELECT seq FROM sqlite_sequence WHERE name = 'orchestration_events'`;
177 return { settledRows: settledRows?.count, sequence: sequence?.seq };
178 }),
179 );
180 assert.equal(copied.settledRows, 0);
181 assert.equal(copied.sequence, sourceSequence?.seq);
182 }),
183 );
184
185 it.effect("upgrades a source from before the V2 thread tables", () =>
186 Effect.gen(function* () {
187 const fs = yield* FileSystem.FileSystem;
188 const path = yield* Path.Path;
189 const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-v1-" });
190 const destDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-v1-dest-" });
191 const stateDir = path.join(sourceDir, "userdata");
192 const source = path.join(stateDir, "statev2.sqlite");
193 yield* fs.makeDirectory(stateDir, { recursive: true });
194 yield* withDatabase(source, runMigrations({ toMigrationInclusive: 54 }));
195
196 const result = yield* runMigrateDevDb(
197 { baseDir: destDir, source, projects: 5, threadsPerProject: 10 },
198 { sharedHome: sourceDir },
199 );
200 assert.include(result.executedMigrations, "55_OrchestrationV2");
201 }),
202 );
203
204 it.effect("fails loudly on a migration slot collision", () =>
205 Effect.gen(function* () {
206 const fs = yield* FileSystem.FileSystem;
207 const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-slot-" });
208 const destDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-slot-dest-" });
209 const source = yield* createFixtureSource(sourceDir);
210 // Simulate another branch having claimed slot 1 first: the id is
211 // recorded, so this checkout's migration 1 silently never runs.
212 yield* withDatabase(
213 source,
214 Effect.gen(function* () {
215 const sql = yield* SqlClient.SqlClient;
216 yield* sql`UPDATE effect_sql_migrations
217 SET name = 'SomebodyElsesMigration' WHERE migration_id = 1`;
218 }),
219 );
220
221 const error = yield* runMigrateDevDb(
222 { baseDir: destDir, source, projects: 5, threadsPerProject: 10 },
223 { sharedHome: sourceDir },
224 ).pipe(Effect.flip);
225 assert.equal(error._tag, "MigrateDevDbSlotCollisionError");
226 if (error._tag === "MigrateDevDbSlotCollisionError") {
227 assert.equal(error.slot, 1);
228 assert.equal(error.appliedName, "SomebodyElsesMigration");
229 }
230 }),
231 );
232
233 it.effect("refuses while a dev server holds the destination", () =>
234 Effect.gen(function* () {
235 const fs = yield* FileSystem.FileSystem;
236 const path = yield* Path.Path;
237 const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-busy-" });
238 const destDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-busy-dest-" });
239 const source = yield* createFixtureSource(sourceDir);
240 // This test process stands in for a live dev server.
241 const stateDir = path.join(destDir, "userdata");
242 yield* fs.makeDirectory(stateDir, { recursive: true });
243 yield* fs.writeFileString(
244 path.join(stateDir, "server-runtime.json"),
245 `{"version":1,"pid":${process.pid}}`,
246 );
247
248 const error = yield* runMigrateDevDb(
249 { baseDir: destDir, source, projects: 5, threadsPerProject: 10 },
250 { sharedHome: sourceDir },
251 ).pipe(Effect.flip);
252 assert.equal(error._tag, "MigrateDevDbServerRunningError");
253 if (error._tag === "MigrateDevDbServerRunningError") {
254 assert.equal(error.pid, process.pid);
255 }
256 }),
257 );
258
259 it.effect("refuses a source that resolves to a destination path", () =>
260 Effect.gen(function* () {
261 const fs = yield* FileSystem.FileSystem;
262 const path = yield* Path.Path;
263 const sharedDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-overlap-" });
264 const destDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-overlap-dest-" });
265 // A leftover snapshot from a prior failed run, passed as --source: it
266 // must not be deleted before it is read.
267 const leftoverSnapshot = path.join(destDir, "userdata", "statev2.sqlite.migrate-dev-db-tmp");
268 yield* fs.makeDirectory(path.dirname(leftoverSnapshot), { recursive: true });
269 yield* fs.writeFileString(leftoverSnapshot, "not a real db");
270
271 const error = yield* runMigrateDevDb(
272 { baseDir: destDir, source: leftoverSnapshot, projects: 5, threadsPerProject: 10 },
273 { sharedHome: sharedDir },
274 ).pipe(Effect.flip);
275 assert.equal(error._tag, "MigrateDevDbSourceIsDestinationError");
276 assert.equal(yield* fs.exists(leftoverSnapshot), true);
277 }),
278 );
279
280 it.effect("refuses to rebuild the shared home", () =>
281 Effect.gen(function* () {
282 const fs = yield* FileSystem.FileSystem;
283 const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-shared-" });
284 const source = yield* createFixtureSource(sourceDir);
285
286 const error = yield* runMigrateDevDb(
287 { baseDir: sourceDir, source, projects: 5, threadsPerProject: 10 },
288 { sharedHome: sourceDir },
289 ).pipe(Effect.flip);
290 assert.equal(error._tag, "MigrateDevDbSharedHomeError");
291 }),
292 );
293});