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