1import { assert, describe, it } from "@effect/vitest";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import {
4 CommandId,
5 MessageId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 ProviderThreadId,
9 ProviderTurnId,
10 ThreadId,
11 TurnItemId,
12 type OrchestrationV2ProviderSession,
13 type OrchestrationV2ProviderThread,
14 type OrchestrationV2ThreadProjection,
15} from "@t3tools/contracts";
16import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
17import * as DateTime from "effect/DateTime";
18import * as Effect from "effect/Effect";
19import * as FileSystem from "effect/FileSystem";
20import * as Layer from "effect/Layer";
21import * as Logger from "effect/Logger";
22import * as Path from "effect/Path";
23import * as PubSub from "effect/PubSub";
24import * as References from "effect/References";
25import * as Ref from "effect/Ref";
26import * as Stream from "effect/Stream";
27import * as SqlClient from "effect/sql/SqlClient";
28
29import { runMigrations } from "../../persistence/Migrations.ts";
30import * as SqlitePersistence from "../../persistence/Sqlite.ts";
31import Migration0042 from "../../persistence/Migrations/042_ProjectionThreadLinkedPullRequest.ts";
32import Migration0043 from "../../persistence/Migrations/043_ProjectionThreadsUnsettledAt.ts";
33import Migration0044 from "../../persistence/Migrations/044_ClearAutomaticProjectModelDefaults.ts";
34import Migration0045 from "../../persistence/Migrations/045_ProjectionProjectsAutoPull.ts";
35import Migration0046 from "../../persistence/Migrations/046_RepairAutomaticSettlementTimestamps.ts";
36import Migration0047 from "../../persistence/Migrations/047_ProjectionProjectIcon.ts";
37import Migration0048 from "../../persistence/Migrations/048_ProjectionThreadBranchPullRequest.ts";
38import Migration0049 from "../../persistence/Migrations/049_ProjectionThreadsActiveOrderKey.ts";
39import { CodexProviderCapabilitiesV2 } from "../Adapters/CodexAdapterV2.ts";
40import * as EffectWorker from "../EffectWorker.ts";
41import * as EventSink from "../EventSink.ts";
42import * as EventStore from "../EventStore.ts";
43import * as LegacyV1ThreadImporter from "./LegacyV1ThreadImporter.ts";
44import * as Orchestrator from "../Orchestrator.ts";
45import * as ProjectionMaintenance from "../ProjectionMaintenance.ts";
46import * as ProjectionStore from "../ProjectionStore.ts";
47import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts";
48import * as ProviderReplayHarness from "../testkit/ProviderReplayHarness.ts";
49import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
50import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
51
52const PROJECT_ID = "project:cutover";
53const ACTIVE_THREAD = "thread:cutover:active";
54const ARCHIVED_THREAD = "thread:cutover:archived";
55const PINNED_THREAD = "thread:cutover:pinned";
56const SNOOZED_THREAD = "thread:cutover:snoozed";
57const DELETED_THREAD = "thread:cutover:deleted";
58const INTERRUPTED_THREAD = "thread:cutover:interrupted";
59const LONG_THREAD = "thread:cutover:long";
60const ALL_THREADS = [
61 ACTIVE_THREAD,
62 ARCHIVED_THREAD,
63 PINNED_THREAD,
64 SNOOZED_THREAD,
65 DELETED_THREAD,
66 INTERRUPTED_THREAD,
67 LONG_THREAD,
68] as const;
69
70const EARLIEST_MARKER = "EARLIEST_IMPORT_MARKER";
71const LATEST_MARKER = "LATEST_IMPORT_MARKER";
72const CONTINUATION_PROMPT = "Continue the migrated thread.";
73const CONTINUATION_RESPONSE = "codex continuation response";
74
75const driver = ProviderDriverKind.make("codex");
76const instanceId = ProviderInstanceId.make("codex");
77const modelSelectionJson = '{"instanceId":"codex","model":"gpt-5.4"}';
78const codexModelSelection = {
79 instanceId,
80 model: "gpt-5.4",
81};
82
83/**
84 * A V1 database as it exists on disk before a V2 server first opens it: schema
85 * through migration 40 plus the 42-49 tail. Slot 41 carries a site-local
86 * `ThreadSummaryTimeline` migration, matching production databases where local
87 * builds recorded extra names under the shared id sequence. The V2 runner only
88 * applies migrations past the recorded maximum id, so the cutover in this test
89 * applies 050, 051 and 052 on top of the untouched copy — the same path the
90 * real upgrade takes.
91 */
92const seedV1Database = (fixturePath: string, workspace: string) =>
93 Effect.scoped(
94 Effect.gen(function* () {
95 const sql = yield* SqlClient.SqlClient;
96 yield* sql`PRAGMA busy_timeout = 5000;`;
97 yield* sql`PRAGMA foreign_keys = ON;`;
98 yield* sql`PRAGMA journal_mode = WAL;`;
99 yield* runMigrations({ toMigrationInclusive: 40 });
100 yield* sql`
101 INSERT INTO effect_sql_migrations (migration_id, name)
102 VALUES (41, 'ThreadSummaryTimeline')
103 `;
104 yield* sql`
105 CREATE TABLE thread_summary_timeline_entries (
106 entry_id TEXT PRIMARY KEY,
107 thread_id TEXT NOT NULL,
108 payload_json TEXT NOT NULL
109 )
110 `;
111 const tailMigrations = [
112 [42, "ProjectionThreadLinkedPullRequest", Migration0042],
113 [43, "ProjectionThreadsUnsettledAt", Migration0043],
114 [44, "ClearAutomaticProjectModelDefaults", Migration0044],
115 [45, "ProjectionProjectsAutoPull", Migration0045],
116 [46, "RepairAutomaticSettlementTimestamps", Migration0046],
117 [47, "ProjectionProjectIcon", Migration0047],
118 [48, "ProjectionThreadBranchPullRequest", Migration0048],
119 [49, "ProjectionThreadsActiveOrderKey", Migration0049],
120 ] as const;
121 for (const [id, name, migration] of tailMigrations) {
122 yield* migration;
123 yield* sql`
124 INSERT INTO effect_sql_migrations (migration_id, name)
125 VALUES (${id}, ${name})
126 `;
127 }
128
129 yield* sql`
130 INSERT INTO projection_projects (
131 project_id, title, workspace_root, scripts_json, created_at, updated_at, deleted_at
132 ) VALUES (
133 ${PROJECT_ID},
134 'Cutover project',
135 ${workspace},
136 '[]',
137 '2026-01-01T00:00:00.000Z',
138 '2026-01-10T00:00:00.000Z',
139 NULL
140 )
141 `;
142
143 const insertThread = (input: {
144 readonly threadId: string;
145 readonly title: string;
146 readonly createdAt: string;
147 readonly updatedAt: string;
148 readonly archivedAt?: string;
149 readonly snoozedAt?: string;
150 readonly snoozedUntil?: string;
151 readonly pinnedAt?: string;
152 readonly pinOrderKey?: string;
153 readonly deletedAt?: string;
154 readonly worktreePath?: string;
155 readonly linkedPullRequestJson?: string;
156 }) =>
157 sql`
158 INSERT INTO projection_threads (
159 thread_id, project_id, title, model_selection_json, runtime_mode,
160 interaction_mode, branch, worktree_path, latest_turn_id,
161 created_at, updated_at, archived_at, settled_override, settled_at,
162 unsettled_at, snoozed_until, snoozed_at, pinned_at, pin_order_key,
163 linked_pull_request_json, deleted_at
164 ) VALUES (
165 ${input.threadId},
166 ${PROJECT_ID},
167 ${input.title},
168 ${modelSelectionJson},
169 'full-access',
170 'default',
171 'main',
172 ${input.worktreePath ?? null},
173 NULL,
174 ${input.createdAt},
175 ${input.updatedAt},
176 ${input.archivedAt ?? null},
177 NULL,
178 NULL,
179 NULL,
180 ${input.snoozedUntil ?? null},
181 ${input.snoozedAt ?? null},
182 ${input.pinnedAt ?? null},
183 ${input.pinOrderKey ?? null},
184 ${input.linkedPullRequestJson ?? null},
185 ${input.deletedAt ?? null}
186 )
187 `;
188
189 const insertMessage = (input: {
190 readonly messageId: string;
191 readonly threadId: string;
192 readonly role: "user" | "assistant";
193 readonly text: string;
194 readonly createdAt: string;
195 readonly isStreaming?: boolean;
196 readonly attachmentsJson?: string;
197 }) =>
198 sql`
199 INSERT INTO projection_thread_messages (
200 message_id, thread_id, turn_id, role, text, attachments_json,
201 is_streaming, created_at, updated_at
202 ) VALUES (
203 ${input.messageId},
204 ${input.threadId},
205 NULL,
206 ${input.role},
207 ${input.text},
208 ${input.attachmentsJson ?? "[]"},
209 ${input.isStreaming === true ? 1 : 0},
210 ${input.createdAt},
211 ${input.createdAt}
212 )
213 `;
214
215 yield* insertThread({
216 threadId: ACTIVE_THREAD,
217 title: "Active conversation",
218 createdAt: "2026-01-01T00:00:00.000Z",
219 updatedAt: "2026-01-05T00:00:00.000Z",
220 linkedPullRequestJson:
221 '{"projectId":"project:cutover","repository":"pingdotgg/t3code","number":9100,"url":"https://github.com/pingdotgg/t3code/pull/9100"}',
222 });
223 yield* insertMessage({
224 messageId: "message:cutover:active:1",
225 threadId: ACTIVE_THREAD,
226 role: "user",
227 text: "First question",
228 createdAt: "2026-01-01T01:00:00.000Z",
229 });
230 yield* insertMessage({
231 messageId: "message:cutover:active:2",
232 threadId: ACTIVE_THREAD,
233 role: "assistant",
234 text: "First answer",
235 createdAt: "2026-01-02T01:00:00.000Z",
236 attachmentsJson:
237 '[{"type":"image","id":"att-1","name":"shot.png","mimeType":"image/png","sizeBytes":1234}]',
238 });
239 yield* insertMessage({
240 messageId: "message:cutover:active:3",
241 threadId: ACTIVE_THREAD,
242 role: "user",
243 text: "Follow-up question",
244 createdAt: "2026-01-03T01:00:00.000Z",
245 });
246 yield* insertMessage({
247 messageId: "message:cutover:active:4",
248 threadId: ACTIVE_THREAD,
249 role: "assistant",
250 text: "Final answer",
251 createdAt: "2026-01-05T01:00:00.000Z",
252 });
253
254 yield* insertThread({
255 threadId: ARCHIVED_THREAD,
256 title: "Archived conversation",
257 createdAt: "2026-01-01T00:00:00.000Z",
258 updatedAt: "2026-01-06T00:00:00.000Z",
259 archivedAt: "2026-01-06T00:00:00.000Z",
260 });
261 yield* insertMessage({
262 messageId: "message:cutover:archived:1",
263 threadId: ARCHIVED_THREAD,
264 role: "user",
265 text: "Archived question",
266 createdAt: "2026-01-01T02:00:00.000Z",
267 });
268 yield* insertMessage({
269 messageId: "message:cutover:archived:2",
270 threadId: ARCHIVED_THREAD,
271 role: "assistant",
272 text: "Archived answer",
273 createdAt: "2026-01-02T02:00:00.000Z",
274 });
275
276 yield* insertThread({
277 threadId: PINNED_THREAD,
278 title: "Pinned conversation",
279 createdAt: "2026-01-01T00:00:00.000Z",
280 updatedAt: "2026-01-07T00:00:00.000Z",
281 pinnedAt: "2026-01-03T00:00:00.000Z",
282 pinOrderKey: "a0",
283 });
284 yield* insertMessage({
285 messageId: "message:cutover:pinned:1",
286 threadId: PINNED_THREAD,
287 role: "user",
288 text: "Pinned question",
289 createdAt: "2026-01-01T03:00:00.000Z",
290 });
291 yield* insertMessage({
292 messageId: "message:cutover:pinned:2",
293 threadId: PINNED_THREAD,
294 role: "assistant",
295 text: "Pinned answer",
296 createdAt: "2026-01-02T03:00:00.000Z",
297 });
298
299 yield* insertThread({
300 threadId: SNOOZED_THREAD,
301 title: "Snoozed conversation",
302 createdAt: "2026-01-01T00:00:00.000Z",
303 updatedAt: "2026-01-08T00:00:00.000Z",
304 snoozedAt: "2026-01-08T00:00:00.000Z",
305 snoozedUntil: "2026-02-01T00:00:00.000Z",
306 });
307 yield* insertMessage({
308 messageId: "message:cutover:snoozed:1",
309 threadId: SNOOZED_THREAD,
310 role: "user",
311 text: "Snoozed question",
312 createdAt: "2026-01-01T04:00:00.000Z",
313 });
314 yield* insertMessage({
315 messageId: "message:cutover:snoozed:2",
316 threadId: SNOOZED_THREAD,
317 role: "assistant",
318 text: "Snoozed answer part one",
319 createdAt: "2026-01-02T04:00:00.000Z",
320 });
321 yield* insertMessage({
322 messageId: "message:cutover:snoozed:3",
323 threadId: SNOOZED_THREAD,
324 role: "assistant",
325 text: "Snoozed answer part two",
326 createdAt: "2026-01-03T04:00:00.000Z",
327 });
328
329 yield* insertThread({
330 threadId: DELETED_THREAD,
331 title: "Deleted conversation",
332 createdAt: "2026-01-01T00:00:00.000Z",
333 updatedAt: "2026-01-09T00:00:00.000Z",
334 deletedAt: "2026-01-09T00:00:00.000Z",
335 });
336 yield* insertMessage({
337 messageId: "message:cutover:deleted:1",
338 threadId: DELETED_THREAD,
339 role: "user",
340 text: "Deleted question",
341 createdAt: "2026-01-01T05:00:00.000Z",
342 });
343 yield* insertMessage({
344 messageId: "message:cutover:deleted:2",
345 threadId: DELETED_THREAD,
346 role: "assistant",
347 text: "Deleted answer",
348 createdAt: "2026-01-02T05:00:00.000Z",
349 });
350
351 yield* insertThread({
352 threadId: INTERRUPTED_THREAD,
353 title: "Interrupted conversation",
354 createdAt: "2026-01-01T00:00:00.000Z",
355 updatedAt: "2026-01-09T01:00:00.000Z",
356 });
357 yield* insertMessage({
358 messageId: "message:cutover:interrupted:1",
359 threadId: INTERRUPTED_THREAD,
360 role: "user",
361 text: "Interrupted question",
362 createdAt: "2026-01-01T06:00:00.000Z",
363 });
364 yield* insertMessage({
365 messageId: "message:cutover:interrupted:2",
366 threadId: INTERRUPTED_THREAD,
367 role: "assistant",
368 text: "Partial answer",
369 createdAt: "2026-01-02T06:00:00.000Z",
370 isStreaming: true,
371 });
372
373 yield* insertThread({
374 threadId: LONG_THREAD,
375 title: "Long conversation",
376 createdAt: "2026-01-01T00:00:00.000Z",
377 updatedAt: "2026-01-10T00:00:00.000Z",
378 worktreePath: workspace,
379 });
380 const filler = "history detail ".repeat(800); // ~12k chars per message
381 yield* insertMessage({
382 messageId: "message:cutover:long:1",
383 threadId: LONG_THREAD,
384 role: "user",
385 text: `${EARLIEST_MARKER} ${filler}`,
386 createdAt: "2026-01-01T07:00:00.000Z",
387 });
388 yield* insertMessage({
389 messageId: "message:cutover:long:2",
390 threadId: LONG_THREAD,
391 role: "assistant",
392 text: `${filler} middle answer`,
393 createdAt: "2026-01-02T07:00:00.000Z",
394 });
395 yield* insertMessage({
396 messageId: "message:cutover:long:3",
397 threadId: LONG_THREAD,
398 role: "user",
399 text: `${filler} middle question`,
400 createdAt: "2026-01-03T07:00:00.000Z",
401 });
402 yield* insertMessage({
403 messageId: "message:cutover:long:4",
404 threadId: LONG_THREAD,
405 role: "assistant",
406 text: `${filler} ${LATEST_MARKER}`,
407 createdAt: "2026-01-04T07:00:00.000Z",
408 });
409
410 yield* sql`PRAGMA wal_checkpoint(TRUNCATE);`;
411 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: fixturePath }))),
412 );
413
414interface CapturedTurn {
415 readonly threadId: ThreadId;
416 readonly providerThreadId: ProviderThreadId;
417 readonly text: string;
418}
419
420const unimplemented = (detail: string) =>
421 Effect.fail(new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail }));
422
423const makeCodexAdapter = (capturedTurns: Ref.Ref<ReadonlyArray<CapturedTurn>>) =>
424 ({
425 instanceId,
426 driver,
427 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
428 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
429 openSession: (sessionInput) =>
430 Effect.gen(function* () {
431 const events = yield* PubSub.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
432 const now = yield* DateTime.now;
433 const providerSession: OrchestrationV2ProviderSession = {
434 id: sessionInput.providerSessionId,
435 driver,
436 providerInstanceId: instanceId,
437 status: "ready",
438 cwd: sessionInput.runtimePolicy.cwd ?? process.cwd(),
439 model: codexModelSelection.model,
440 capabilities: CodexProviderCapabilitiesV2,
441 createdAt: now,
442 updatedAt: now,
443 lastError: null,
444 };
445 return {
446 instanceId,
447 driver,
448 providerSessionId: sessionInput.providerSessionId,
449 providerSession,
450 events: Stream.fromPubSub(events),
451 ensureThread: (threadInput) =>
452 Effect.gen(function* () {
453 const createdAt = yield* DateTime.now;
454 const nativeThreadId = `${driver}:${threadInput.threadId}`;
455 return {
456 id: ProviderThreadId.make(`provider-thread:${nativeThreadId}`),
457 driver,
458 providerInstanceId: instanceId,
459 providerSessionId: sessionInput.providerSessionId,
460 appThreadId: threadInput.threadId,
461 ownerNodeId: null,
462 nativeThreadRef: {
463 driver,
464 nativeId: nativeThreadId,
465 strength: "strong",
466 },
467 nativeConversationHeadRef: null,
468 status: "idle",
469 firstRunOrdinal: null,
470 lastRunOrdinal: null,
471 handoffIds: [],
472 forkedFrom: null,
473 createdAt,
474 updatedAt: createdAt,
475 } satisfies OrchestrationV2ProviderThread;
476 }),
477 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
478 startTurn: (turnInput) =>
479 Effect.gen(function* () {
480 yield* Ref.update(capturedTurns, (turns) => [
481 ...turns,
482 {
483 threadId: turnInput.threadId,
484 providerThreadId: turnInput.providerThread.id,
485 text: turnInput.message.text,
486 },
487 ]);
488 const eventTime = yield* DateTime.now;
489 const providerTurnId = ProviderTurnId.make(
490 `provider-turn:${turnInput.threadId}:${turnInput.runOrdinal}`,
491 );
492 yield* PubSub.publishAll(events, [
493 {
494 type: "provider_turn.updated",
495 driver,
496 providerTurn: {
497 id: providerTurnId,
498 providerThreadId: turnInput.providerThread.id,
499 nodeId: turnInput.rootNodeId,
500 runAttemptId: turnInput.attemptId,
501 nativeTurnRef: {
502 driver,
503 nativeId: `native-turn:${turnInput.threadId}:${turnInput.runOrdinal}`,
504 strength: "strong",
505 },
506 ordinal: turnInput.runOrdinal,
507 status: "completed",
508 startedAt: eventTime,
509 completedAt: eventTime,
510 },
511 },
512 {
513 type: "turn_item.updated",
514 driver,
515 turnItem: {
516 id: TurnItemId.make(
517 `turn-item:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`,
518 ),
519 threadId: turnInput.threadId,
520 runId: turnInput.runId,
521 nodeId: turnInput.rootNodeId,
522 providerThreadId: turnInput.providerThread.id,
523 providerTurnId,
524 nativeItemRef: null,
525 parentItemId: null,
526 ordinal: turnInput.runOrdinal * 100 + 1,
527 status: "completed",
528 title: null,
529 startedAt: eventTime,
530 completedAt: eventTime,
531 updatedAt: eventTime,
532 type: "assistant_message",
533 messageId: MessageId.make(
534 `message:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`,
535 ),
536 text: CONTINUATION_RESPONSE,
537 streaming: false,
538 },
539 },
540 {
541 type: "turn.terminal",
542 driver,
543 providerThreadId: turnInput.providerThread.id,
544 providerTurnId,
545 runOrdinal: turnInput.runOrdinal,
546 status: "completed",
547 failure: null,
548 threadDisposition: "reusable",
549 },
550 ] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>);
551 }),
552 steerTurn: () => Effect.void,
553 interruptTurn: () => Effect.void,
554 respondToRuntimeRequest: () => Effect.void,
555 readThreadSnapshot: () => unimplemented("readThreadSnapshot unused in cutover test"),
556 rollbackThread: () => unimplemented("rollbackThread unused in cutover test"),
557 forkThread: () => unimplemented("forkThread unused in cutover test"),
558 };
559 }),
560 }) satisfies ProviderAdapter.ProviderAdapterV2["Service"];
561
562const waitForIdle = Effect.fn("LegacyV1Cutover.waitForIdle")(function* (threadId: ThreadId) {
563 const orchestrator = yield* Orchestrator.OrchestratorV2;
564 for (let attempt = 0; attempt < 1_000; attempt += 1) {
565 const projection = yield* orchestrator.getThreadProjection(threadId);
566 if (
567 projection.runs.every(
568 (run) => !["queued", "starting", "running", "waiting"].includes(run.status),
569 )
570 ) {
571 return projection;
572 }
573 yield* Effect.sleep("5 millis");
574 }
575 return yield* Effect.die(new Error("Cutover test timed out waiting for idle"));
576});
577
578const layerBoot = (input: {
579 readonly name: string;
580 readonly dbPath: string;
581 readonly workspace: string;
582 readonly capturedTurns: Ref.Ref<ReadonlyArray<CapturedTurn>>;
583}) => {
584 const layerDatabase = SqlitePersistence.layerFromPath(input.dbPath).pipe(
585 Layer.provide(NodeServices.layer),
586 );
587 const layerEventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(layerDatabase));
588 const layerProjectionStoreProvided = ProjectionStore.layer.pipe(
589 Layer.provideMerge(layerDatabase),
590 );
591 const layerStoresProvided = Layer.mergeAll(
592 layerDatabase,
593 layerEventStoreProvided,
594 layerProjectionStoreProvided,
595 );
596 const layerEventSinkProvided = EventSink.layer.pipe(Layer.provide(layerStoresProvided));
597 const layerImporterProvided = LegacyV1ThreadImporter.layer.pipe(
598 Layer.provide(Layer.mergeAll(layerStoresProvided, layerEventSinkProvided)),
599 );
600 const layerMaintenanceProvided = ProjectionMaintenance.layer.pipe(
601 Layer.provide(layerStoresProvided),
602 );
603 const layerOrchestratorProvided = ProviderReplayHarness.layerWithRegistry(
604 {
605 name: input.name,
606 runtimePolicyOverride: {
607 cwd: input.workspace,
608 approvalPolicy: "never",
609 sandboxPolicy: {
610 type: "readOnly",
611 access: { type: "fullAccess" },
612 networkAccess: false,
613 },
614 },
615 },
616 ProviderAdapterRegistry.layerSingle(makeCodexAdapter(input.capturedTurns)),
617 { databaseLayer: layerDatabase },
618 );
619 return Layer.mergeAll(
620 layerStoresProvided,
621 layerEventSinkProvided,
622 layerImporterProvided,
623 layerMaintenanceProvided,
624 layerOrchestratorProvided,
625 );
626};
627
628interface CapturedLog {
629 readonly message: unknown;
630 readonly annotations: Readonly<Record<string, unknown>>;
631}
632
633const makeCapturingLogger = (logs: CapturedLog[]) =>
634 Logger.make(({ fiber, message }) => {
635 logs.push({
636 message,
637 annotations: fiber.getRef(References.CurrentLogAnnotations),
638 });
639 });
640
641const messageOrdinals = (projection: OrchestrationV2ThreadProjection) =>
642 projection.turnItems
643 .filter(
644 (
645 item,
646 ): item is Extract<typeof item, { readonly type: "user_message" | "assistant_message" }> =>
647 item.type === "user_message" || item.type === "assistant_message",
648 )
649 .map((item) => [String(item.messageId), item.ordinal, item.status] as const);
650
651describe("orchestration v2 legacy v1 cutover", () => {
652 it.live(
653 "migrates an untouched v1 database copy through shell import, lazy transcripts, continuation and restart",
654 () =>
655 Effect.scoped(
656 Effect.gen(function* () {
657 const fs = yield* FileSystem.FileSystem;
658 const path = yield* Path.Path;
659 const workspace = yield* checkpointWorkspace("legacy-v1-cutover");
660 const stateDir = yield* fs.makeTempDirectory({ prefix: "t3-v1-cutover-state-" });
661 const fixturePath = path.join(stateDir, "v1-source.sqlite");
662 const copyPath = path.join(stateDir, "userdata", "state.sqlite");
663 yield* fs.makeDirectory(path.join(stateDir, "userdata"), { recursive: true });
664
665 yield* seedV1Database(fixturePath, workspace);
666 yield* fs.copyFile(fixturePath, copyPath);
667
668 const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]);
669 const longThreadId = ThreadId.make(LONG_THREAD);
670
671 // First boot: the real file-backed stack migrates the copied database.
672 const boot1Logs: CapturedLog[] = [];
673 const firstBoot = yield* Effect.scoped(
674 Effect.gen(function* () {
675 const sql = yield* SqlClient.SqlClient;
676 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
677 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
678 const projections = yield* ProjectionStore.ProjectionStoreV2;
679 const orchestrator = yield* Orchestrator.OrchestratorV2;
680
681 assert.equal(yield* importer.pendingThreadCount, ALL_THREADS.length);
682 const shellImport = yield* importer.reconcileShells;
683 assert.deepStrictEqual(shellImport, {
684 importedThreadCount: ALL_THREADS.length,
685 importedMessageCount: 14,
686 });
687 // A repeated shell reconcile in the same boot is a no-op.
688 assert.deepStrictEqual(yield* importer.reconcileShells, {
689 importedThreadCount: 0,
690 importedMessageCount: 0,
691 });
692 assert.isTrue((yield* maintenance.rebuild).valid);
693
694 const shellSnapshot = yield* projections.getShellSnapshot();
695 const visibleShellIds = [
696 ...shellSnapshot.threads.map((thread) => thread.id),
697 ...shellSnapshot.archivedThreads.map((thread) => thread.id),
698 ];
699 assert.notInclude(visibleShellIds, ThreadId.make(DELETED_THREAD));
700 assert.include(
701 shellSnapshot.archivedThreads.map((thread) => thread.id),
702 ThreadId.make(ARCHIVED_THREAD),
703 );
704 const pinnedShell = shellSnapshot.threads.find(
705 (thread) => thread.id === PINNED_THREAD,
706 );
707 assert.equal(pinnedShell?.pinOrderKey, "a0");
708
709 // Shells carry only the last user + latest message previews until a
710 // transcript is hydrated on demand.
711 const activeShell = yield* projections.getThreadProjection(
712 ThreadId.make(ACTIVE_THREAD),
713 );
714 assert.equal(activeShell.thread.historyOrigin, "v1_import");
715 assert.isNull(activeShell.thread.activeProviderThreadId);
716 assert.deepStrictEqual(
717 activeShell.messages.map((message) => message.id),
718 ["message:cutover:active:3", "message:cutover:active:4"],
719 );
720 assert.isEmpty(activeShell.providerThreads);
721 // The import contract is messages + metadata only: no checkpoints,
722 // tool items, or native sessions are presented as migrated.
723 assert.isEmpty(
724 activeShell.turnItems.filter(
725 (item) => item.type !== "user_message" && item.type !== "assistant_message",
726 ),
727 );
728
729 const deletedShell = yield* projections.getThreadProjection(
730 ThreadId.make(DELETED_THREAD),
731 );
732 assert.isNotNull(deletedShell.thread.deletedAt);
733 assert.equal(
734 deletedShell.thread.deletedAt?.toString(),
735 DateTime.makeUnsafe("2026-01-09T00:00:00.000Z").toString(),
736 );
737
738 // Lazy transcript import on first read. The snoozed thread stays
739 // unhydrated until after the restart to prove laziness survives a
740 // process boundary.
741 for (const threadId of [
742 ACTIVE_THREAD,
743 ARCHIVED_THREAD,
744 PINNED_THREAD,
745 DELETED_THREAD,
746 INTERRUPTED_THREAD,
747 LONG_THREAD,
748 ]) {
749 yield* importer.ensureTranscript(ThreadId.make(threadId));
750 }
751 assert.deepStrictEqual(
752 yield* importer.ensureTranscript(ThreadId.make(ACTIVE_THREAD)),
753 { importedThreadCount: 0, importedMessageCount: 0 },
754 );
755
756 const active = yield* projections.getThreadProjection(ThreadId.make(ACTIVE_THREAD));
757 assert.deepStrictEqual(
758 active.messages.map((message) => message.id),
759 [
760 "message:cutover:active:1",
761 "message:cutover:active:2",
762 "message:cutover:active:3",
763 "message:cutover:active:4",
764 ],
765 );
766 assert.deepStrictEqual(messageOrdinals(active), [
767 ["message:cutover:active:1", 1, "completed"],
768 ["message:cutover:active:2", 2, "completed"],
769 ["message:cutover:active:3", 3, "completed"],
770 ["message:cutover:active:4", 4, "completed"],
771 ]);
772 assert.deepStrictEqual(
773 active.messages.find((message) => message.id === "message:cutover:active:2")
774 ?.attachments,
775 [
776 {
777 type: "image",
778 id: "att-1",
779 name: "shot.png",
780 mimeType: "image/png",
781 sizeBytes: 1234,
782 },
783 ],
784 );
785 assert.equal(active.thread.linkedPullRequest?.number, 9100);
786 assert.deepStrictEqual(
787 active.thread.pullRequests?.map((pullRequest) => pullRequest.number),
788 [9100],
789 );
790
791 const archived = yield* projections.getThreadProjection(
792 ThreadId.make(ARCHIVED_THREAD),
793 );
794 assert.isNotNull(archived.thread.archivedAt);
795 const pinned = yield* projections.getThreadProjection(ThreadId.make(PINNED_THREAD));
796 assert.isNotNull(pinned.thread.pinnedAt);
797 assert.equal(pinned.thread.pinOrderKey, "a0");
798 const snoozed = yield* projections.getThreadProjection(ThreadId.make(SNOOZED_THREAD));
799 assert.equal(
800 snoozed.thread.snoozedUntil?.toString(),
801 DateTime.makeUnsafe("2026-02-01T00:00:00.000Z").toString(),
802 );
803 // Still preview-only: its transcript was never requested, so the
804 // middle message between the two shell previews is absent.
805 assert.deepStrictEqual(
806 snoozed.messages.map((message) => message.id),
807 ["message:cutover:snoozed:1", "message:cutover:snoozed:3"],
808 );
809 const snoozedImportRow = yield* sql<{
810 readonly transcript_imported_at: string | null;
811 }>`
812 SELECT transcript_imported_at
813 FROM orchestration_v2_legacy_imports
814 WHERE thread_id = ${SNOOZED_THREAD}
815 `;
816 assert.isNull(snoozedImportRow[0]?.transcript_imported_at);
817
818 const interrupted = yield* projections.getThreadProjection(
819 ThreadId.make(INTERRUPTED_THREAD),
820 );
821 assert.deepStrictEqual(messageOrdinals(interrupted), [
822 ["message:cutover:interrupted:1", 1, "completed"],
823 ["message:cutover:interrupted:2", 2, "interrupted"],
824 ]);
825
826 // First continuation of a migrated thread: a fresh provider session
827 // receives the newest transcript suffix inside the 32k handoff.
828 yield* orchestrator.dispatch({
829 type: "message.dispatch",
830 createdBy: "user",
831 creationSource: "web",
832 commandId: CommandId.make("command:cutover:continue"),
833 threadId: longThreadId,
834 messageId: MessageId.make("message:cutover:long:continuation"),
835 text: CONTINUATION_PROMPT,
836 attachments: [],
837 dispatchMode: { type: "start_immediately" },
838 });
839 const continued = yield* waitForIdle(longThreadId);
840
841 const runs = continued.runs.filter((run) => run.ordinal >= 1);
842 assert.equal(runs.at(-1)?.status, "completed");
843 assert.equal(continued.providerThreads.length, 1);
844 assert.isNotNull(continued.providerThreads[0]?.nativeThreadRef);
845 assert.equal(continued.contextHandoffs.length, 1);
846 const handoff = continued.contextHandoffs[0]!;
847 assert.equal(handoff.strategy, "manual_context");
848 assert.isAtMost(handoff.summaryText.length, 32_000);
849 assert.include(handoff.summaryText, LATEST_MARKER);
850 assert.notInclude(handoff.summaryText, EARLIEST_MARKER);
851 const turns = yield* Ref.get(capturedTurns);
852 assert.equal(turns.length, 1);
853 assert.include(turns[0]!.text, "Context handoff (manual_context):");
854 assert.include(turns[0]!.text, `User message:\n${CONTINUATION_PROMPT}`);
855
856 // The next continuation reuses the provider thread without a new
857 // handoff: the imported context is only reissued until a v2 run
858 // completes.
859 yield* orchestrator.dispatch({
860 type: "message.dispatch",
861 createdBy: "user",
862 creationSource: "web",
863 commandId: CommandId.make("command:cutover:continue:again"),
864 threadId: longThreadId,
865 messageId: MessageId.make("message:cutover:long:continuation:2"),
866 text: "One more.",
867 attachments: [],
868 dispatchMode: { type: "start_immediately" },
869 });
870 const continuedAgain = yield* waitForIdle(longThreadId);
871 assert.equal(continuedAgain.contextHandoffs.length, 1);
872 assert.equal((yield* Ref.get(capturedTurns)).length, 2);
873
874 // Drain the outbox worker so the persisted event count is settled
875 // before the restart boot re-reads it.
876 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
877 yield* worker.drain();
878
879 const migrationEventCount = yield* sql<{ readonly count: number }>`
880 SELECT COUNT(*) AS count
881 FROM orchestration_events
882 WHERE application_event_version = 2
883 AND event_id LIKE 'migration:v1:%'
884 `;
885 const importRows = yield* sql<{
886 readonly thread_id: string;
887 readonly transcript_imported_at: string | null;
888 }>`
889 SELECT thread_id, transcript_imported_at
890 FROM orchestration_v2_legacy_imports
891 ORDER BY thread_id
892 `;
893 const legacyMessageCount = yield* sql<{ readonly count: number }>`
894 SELECT COUNT(*) AS count FROM projection_thread_messages
895 `;
896 const legacyThreadCount = yield* sql<{ readonly count: number }>`
897 SELECT COUNT(*) AS count FROM projection_threads
898 `;
899 const recordedMigration41 = yield* sql<{ readonly name: string }>`
900 SELECT name FROM effect_sql_migrations WHERE migration_id = 41
901 `;
902 const authSessionColumns = yield* sql<{ readonly name: string }>`
903 PRAGMA table_info(auth_sessions)
904 `;
905 return {
906 migrationEventCount: migrationEventCount[0]?.count ?? 0,
907 importRows,
908 legacyMessageCount: legacyMessageCount[0]?.count ?? 0,
909 legacyThreadCount: legacyThreadCount[0]?.count ?? 0,
910 longProjection: continuedAgain,
911 migration41Name: recordedMigration41[0]?.name ?? null,
912 authSessionColumnNames: authSessionColumns.map((column) => column.name),
913 };
914 }).pipe(
915 Effect.provide(
916 layerBoot({
917 name: "legacy-v1-cutover-first",
918 dbPath: copyPath,
919 workspace,
920 capturedTurns,
921 }),
922 ),
923 Effect.provideService(
924 Logger.CurrentLoggers,
925 new Set([makeCapturingLogger(boot1Logs)]),
926 ),
927 ),
928 );
929
930 // The copied database recorded a site-local migration under id 41, so
931 // the migrator skipped this build's AuthSessionClientConnection by
932 // id. The divergence is surfaced at startup while the rest of the
933 // cutover still runs.
934 const divergenceLog = boot1Logs.find((log) =>
935 String(log.message).includes("migration history diverges"),
936 );
937 assert.deepStrictEqual(divergenceLog?.annotations.divergent, [
938 "41:ThreadSummaryTimeline (this build: AuthSessionClientConnection)",
939 ]);
940 assert.equal(firstBoot.migration41Name, "ThreadSummaryTimeline");
941 // The skipped migration's columns never landed; the schema gap is
942 // what the startup warning points at.
943 assert.notInclude(firstBoot.authSessionColumnNames, "client_surface");
944 assert.notInclude(firstBoot.authSessionColumnNames, "client_app_version");
945
946 assert.equal(firstBoot.importRows.length, ALL_THREADS.length);
947 const unhydratedRows = firstBoot.importRows.filter(
948 (row) => row.transcript_imported_at === null,
949 );
950 assert.deepStrictEqual(
951 unhydratedRows.map((row) => row.thread_id),
952 [SNOOZED_THREAD],
953 );
954
955 // Restart: a fresh runtime on the same file repeats nothing, and the
956 // still-unhydrated snoozed transcript lands lazily after the restart.
957 const boot2Logs: CapturedLog[] = [];
958 yield* Effect.scoped(
959 Effect.gen(function* () {
960 const sql = yield* SqlClient.SqlClient;
961 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
962 const projections = yield* ProjectionStore.ProjectionStoreV2;
963
964 assert.equal(yield* importer.pendingThreadCount, 1);
965 assert.deepStrictEqual(yield* importer.reconcileShells, {
966 importedThreadCount: 0,
967 importedMessageCount: 0,
968 });
969 assert.deepStrictEqual(
970 yield* importer.ensureTranscript(ThreadId.make(SNOOZED_THREAD)),
971 { importedThreadCount: 1, importedMessageCount: 1 },
972 );
973 const snoozed = yield* projections.getThreadProjection(ThreadId.make(SNOOZED_THREAD));
974 assert.deepStrictEqual(
975 snoozed.messages.map((message) => message.id),
976 [
977 "message:cutover:snoozed:1",
978 "message:cutover:snoozed:2",
979 "message:cutover:snoozed:3",
980 ],
981 );
982 assert.equal(yield* importer.pendingThreadCount, 0);
983 for (const threadId of ALL_THREADS) {
984 assert.deepStrictEqual(yield* importer.ensureTranscript(ThreadId.make(threadId)), {
985 importedThreadCount: 0,
986 importedMessageCount: 0,
987 });
988 }
989 const migrationEventCount = yield* sql<{ readonly count: number }>`
990 SELECT COUNT(*) AS count
991 FROM orchestration_events
992 WHERE application_event_version = 2
993 AND event_id LIKE 'migration:v1:%'
994 `;
995 // The snoozed transcript hydration above appends its own two
996 // migration events (message + turn item); nothing else repeats.
997 assert.equal(migrationEventCount[0]?.count, firstBoot.migrationEventCount + 2);
998
999 const restarted = yield* projections.getThreadProjection(longThreadId);
1000 assert.equal(restarted.thread.historyOrigin, "v1_import");
1001 assert.equal(restarted.contextHandoffs.length, 1);
1002 assert.isNotNull(restarted.thread.activeProviderThreadId);
1003 assert.deepStrictEqual(
1004 restarted.runs.map((run) => run.status),
1005 firstBoot.longProjection.runs.map((run) => run.status),
1006 );
1007
1008 // The v1 projection tables remain the untouched, read-only
1009 // recovery source after migration and restart.
1010 const legacyMessageCount = yield* sql<{ readonly count: number }>`
1011 SELECT COUNT(*) AS count FROM projection_thread_messages
1012 `;
1013 const legacyThreadCount = yield* sql<{ readonly count: number }>`
1014 SELECT COUNT(*) AS count FROM projection_threads
1015 `;
1016 assert.equal(legacyMessageCount[0]?.count, firstBoot.legacyMessageCount);
1017 assert.equal(legacyThreadCount[0]?.count, firstBoot.legacyThreadCount);
1018 const quickCheck = yield* sql<{ readonly quick_check: string }>`
1019 PRAGMA quick_check
1020 `;
1021 assert.deepStrictEqual(
1022 quickCheck.map((row) => row.quick_check),
1023 ["ok"],
1024 );
1025 }).pipe(
1026 Effect.provide(
1027 layerBoot({
1028 name: "legacy-v1-cutover-restart",
1029 dbPath: copyPath,
1030 workspace,
1031 capturedTurns,
1032 }),
1033 ),
1034 Effect.provideService(
1035 Logger.CurrentLoggers,
1036 new Set([makeCapturingLogger(boot2Logs)]),
1037 ),
1038 ),
1039 );
1040
1041 // The recorded-name divergence persists across restarts; the warning
1042 // fires again so it cannot be missed between upgrades.
1043 assert.isTrue(
1044 boot2Logs.some((log) => String(log.message).includes("migration history diverges")),
1045 );
1046 }).pipe(Effect.provide(NodeServices.layer)),
1047 ),
1048 );
1049});