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.scoped(
738 Effect.gen(function* () {
739 const fs = yield* FileSystem.FileSystem;
740 const path = yield* Path.Path;
741 const workspace = yield* checkpointWorkspace("legacy-v1-cutover");
742 const stateDir = yield* fs.makeTempDirectory({ prefix: "t3-v1-cutover-state-" });
743 const fixturePath = path.join(stateDir, "v1-source.sqlite");
744 const copyPath = path.join(stateDir, "userdata", "state.sqlite");
745 yield* fs.makeDirectory(path.join(stateDir, "userdata"), { recursive: true });
746
747 const v1Snapshot = yield* seedV1Database(fixturePath, workspace);
748 const fixtureHashBefore = yield* sha256File(fixturePath);
749 yield* fs.copyFile(fixturePath, copyPath);
750
751 const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]);
752 const longThreadId = ThreadId.make(LONG_THREAD);
753
754 const boot1Logs: CapturedLog[] = [];
755 const firstBoot = yield* Effect.scoped(
756 Effect.gen(function* () {
757 const sql = yield* SqlClient.SqlClient;
758 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
759 const maintenance = yield* ProjectionReplay.ProjectionReplay;
760 const projections = yield* ProjectionStore.ProjectionStoreV2;
761
762 assert.equal(yield* importer.pendingThreadCount, ALL_THREADS.length);
763 const shellImport = yield* importer.reconcileShells;
764 assert.deepStrictEqual(shellImport, {
765 importedThreadCount: ALL_THREADS.length,
766 importedMessageCount: 14,
767 });
768
769 assert.deepStrictEqual(yield* importer.reconcileShells, {
770 importedThreadCount: 0,
771 importedMessageCount: 0,
772 });
773 assert.isTrue((yield* maintenance.rebuild).valid);
774
775 const shellSnapshot = yield* projections.getShellSnapshot();
776 const visibleShellIds = [
777 ...shellSnapshot.threads.map((thread) => thread.id),
778 ...shellSnapshot.archivedThreads.map((thread) => thread.id),
779 ];
780 assert.notInclude(visibleShellIds, ThreadId.make(DELETED_THREAD));
781 assert.include(
782 shellSnapshot.archivedThreads.map((thread) => thread.id),
783 ThreadId.make(ARCHIVED_THREAD),
784 );
785 const pinnedShell = shellSnapshot.threads.find(
786 (thread) => thread.id === PINNED_THREAD,
787 );
788 assert.equal(pinnedShell?.pinOrderKey, "a0");
789
790 const activeShell = yield* projections.getThreadProjection(
791 ThreadId.make(ACTIVE_THREAD),
792 );
793 assert.equal(activeShell.thread.historyOrigin, "v1_import");
794 assert.isNull(activeShell.thread.activeProviderThreadId);
795 assert.deepStrictEqual(
796 activeShell.messages.map((message) => message.id),
797 ["message:cutover:active:3", "message:cutover:active:4"],
798 );
799 assert.isEmpty(activeShell.providerThreads);
800
801 assert.isEmpty(
802 activeShell.turnItems.filter(
803 (item) => item.type !== "user_message" && item.type !== "assistant_message",
804 ),
805 );
806
807 const deletedShell = yield* projections.getThreadProjection(
808 ThreadId.make(DELETED_THREAD),
809 );
810 assert.isNotNull(deletedShell.thread.deletedAt);
811 assert.equal(
812 deletedShell.thread.deletedAt?.toString(),
813 DateTime.makeUnsafe("2026-01-09T00:00:00.000Z").toString(),
814 );
815
816 for (const threadId of [
817 ACTIVE_THREAD,
818 ARCHIVED_THREAD,
819 PINNED_THREAD,
820 DELETED_THREAD,
821 INTERRUPTED_THREAD,
822 LONG_THREAD,
823 ]) {
824 yield* importer.ensureTranscript(ThreadId.make(threadId));
825 }
826 assert.deepStrictEqual(
827 yield* importer.ensureTranscript(ThreadId.make(ACTIVE_THREAD)),
828 { importedThreadCount: 0, importedMessageCount: 0 },
829 );
830
831 const active = yield* projections.getThreadProjection(ThreadId.make(ACTIVE_THREAD));
832 assert.deepStrictEqual(
833 active.messages.map((message) => message.id),
834 [
835 "message:cutover:active:1",
836 "message:cutover:active:2",
837 "message:cutover:active:3",
838 "message:cutover:active:4",
839 ],
840 );
841 assert.deepStrictEqual(messageOrdinals(active), [
842 ["message:cutover:active:1", 1, "completed"],
843 ["message:cutover:active:2", 2, "completed"],
844 ["message:cutover:active:3", 3, "completed"],
845 ["message:cutover:active:4", 4, "completed"],
846 ]);
847 assert.deepStrictEqual(
848 active.messages.find((message) => message.id === "message:cutover:active:2")
849 ?.attachments,
850 [
851 {
852 type: "image",
853 id: "att-1",
854 name: "shot.png",
855 mimeType: "image/png",
856 sizeBytes: 1234,
857 },
858 ],
859 );
860 assert.equal(active.thread.linkedPullRequest?.number, 9100);
861 assert.deepStrictEqual(
862 active.thread.pullRequests?.map((pullRequest) => pullRequest.number),
863 [9100],
864 );
865
866 const archived = yield* projections.getThreadProjection(
867 ThreadId.make(ARCHIVED_THREAD),
868 );
869 assert.isNotNull(archived.thread.archivedAt);
870 const pinned = yield* projections.getThreadProjection(ThreadId.make(PINNED_THREAD));
871 assert.isNotNull(pinned.thread.pinnedAt);
872 assert.equal(pinned.thread.pinOrderKey, "a0");
873 const snoozed = yield* projections.getThreadProjection(ThreadId.make(SNOOZED_THREAD));
874 assert.equal(
875 snoozed.thread.snoozedUntil?.toString(),
876 DateTime.makeUnsafe("2026-02-01T00:00:00.000Z").toString(),
877 );
878
879 assert.deepStrictEqual(
880 snoozed.messages.map((message) => message.id),
881 ["message:cutover:snoozed:1", "message:cutover:snoozed:3"],
882 );
883 const snoozedImportRow = yield* sql<{
884 readonly transcript_imported_at: string | null;
885 }>`
886 SELECT transcript_imported_at
887 FROM orchestration_v2_legacy_imports
888 WHERE thread_id = ${SNOOZED_THREAD}
889 `;
890 assert.isNull(snoozedImportRow[0]?.transcript_imported_at);
891
892 // A failed hydration leaves the legacy rows and pending marker
893 // available for retry in a fresh runtime after restart.
894 yield* sql`
895 CREATE TRIGGER fail_snoozed_hydration
896 BEFORE INSERT ON orchestration_events
897 WHEN NEW.event_id = 'migration:v1:message:message:cutover:snoozed:2'
898 BEGIN
899 SELECT RAISE(ABORT, 'injected cutover hydration failure');
900 END
901 `;
902 const hydrationError = yield* importer
903 .ensureTranscript(ThreadId.make(SNOOZED_THREAD))
904 .pipe(Effect.flip);
905 assert.equal(hydrationError._tag, "LegacyV1ThreadImportError");
906 assert.equal(yield* importer.pendingThreadCount, 1);
907
908 const interrupted = yield* projections.getThreadProjection(
909 ThreadId.make(INTERRUPTED_THREAD),
910 );
911 assert.deepStrictEqual(messageOrdinals(interrupted), [
912 ["message:cutover:interrupted:1", 1, "completed"],
913 ["message:cutover:interrupted:2", 2, "interrupted"],
914 ]);
915
916 const continued = yield* dispatchMessageAwaitingRun({
917 threadId: longThreadId,
918 messageId: "message:cutover:long:continuation",
919 commandId: "command:cutover:continue",
920 text: CONTINUATION_PROMPT,
921 });
922
923 const runs = continued.runs.filter((run) => run.ordinal >= 1);
924 assert.equal(runs.at(-1)?.status, "completed");
925 assert.equal(continued.providerThreads.length, 1);
926 assert.isNotNull(continued.providerThreads[0]?.nativeThreadRef);
927 assert.equal(continued.contextHandoffs.length, 1);
928 const handoff = continued.contextHandoffs[0]!;
929 assert.equal(handoff.strategy, "manual_context");
930 assert.isAtMost(handoff.summaryText.length, 32_000);
931 assert.include(handoff.summaryText, LATEST_MARKER);
932 assert.notInclude(handoff.summaryText, EARLIEST_MARKER);
933 const turns = yield* Ref.get(capturedTurns);
934 assert.equal(turns.length, 1);
935 assert.include(turns[0]!.text, "Context handoff (manual_context):");
936 assert.include(turns[0]!.text, `User message:\n${CONTINUATION_PROMPT}`);
937
938 const continuedAgain = yield* dispatchMessageAwaitingRun({
939 threadId: longThreadId,
940 messageId: "message:cutover:long:continuation:2",
941 commandId: "command:cutover:continue:again",
942 text: "One more.",
943 });
944 assert.equal(continuedAgain.contextHandoffs.length, 1);
945 assert.equal((yield* Ref.get(capturedTurns)).length, 2);
946
947 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
948 yield* worker.drain();
949
950 const migrationEventCount = yield* sql<{ readonly count: number }>`
951 SELECT COUNT(*) AS count
952 FROM orchestration_events
953 WHERE application_event_version = 2
954 AND event_id LIKE 'migration:v1:%'
955 `;
956 const importRows = yield* sql<{
957 readonly thread_id: string;
958 readonly transcript_imported_at: string | null;
959 }>`
960 SELECT thread_id, transcript_imported_at
961 FROM orchestration_v2_legacy_imports
962 ORDER BY thread_id
963 `;
964 const legacyProjects = yield* sql<Record<string, unknown>>`
965 SELECT * FROM projection_projects ORDER BY project_id
966 `;
967 const legacyThreads = yield* sql<Record<string, unknown>>`
968 SELECT * FROM projection_threads ORDER BY thread_id
969 `;
970 const legacyMessages = yield* sql<Record<string, unknown>>`
971 SELECT * FROM projection_thread_messages ORDER BY message_id
972 `;
973 const recordedMigration41 = yield* sql<{ readonly name: string }>`
974 SELECT name FROM effect_sql_migrations WHERE migration_id = 41
975 `;
976 const authSessionColumns = yield* sql<{ readonly name: string }>`
977 PRAGMA table_info(auth_sessions)
978 `;
979 return {
980 migrationEventCount: migrationEventCount[0]?.count ?? 0,
981 importRows,
982 legacyProjects,
983 legacyThreads,
984 legacyMessages,
985 longProjection: continuedAgain,
986 migration41Name: recordedMigration41[0]?.name ?? null,
987 authSessionColumnNames: authSessionColumns.map((column) => column.name),
988 };
989 }).pipe(
990 Effect.provide(
991 layerBoot({
992 name: "legacy-v1-cutover-first",
993 dbPath: copyPath,
994 workspace,
995 capturedTurns,
996 }),
997 ),
998 Effect.provideService(
999 Logger.CurrentLoggers,
1000 new Set([makeCapturingLogger(boot1Logs)]),
1001 ),
1002 ),
1003 );
1004
1005 const divergenceLog = boot1Logs.find((log) =>
1006 String(log.message).includes("migration history diverges"),
1007 );
1008 assert.deepStrictEqual(divergenceLog?.annotations.divergent, [
1009 "41:ThreadSummaryTimeline (this build: AuthSessionClientConnection)",
1010 ]);
1011 assert.equal(firstBoot.migration41Name, "ThreadSummaryTimeline");
1012
1013 assert.notInclude(firstBoot.authSessionColumnNames, "client_surface");
1014 assert.notInclude(firstBoot.authSessionColumnNames, "client_app_version");
1015
1016 // Full row contents of the legacy recovery tables survive migration
1017 // and continuation unchanged, not just their counts.
1018 assert.deepStrictEqual(
1019 restrictToSnapshotColumns(firstBoot.legacyProjects, v1Snapshot.projects),
1020 v1Snapshot.projects.rows,
1021 );
1022 assert.deepStrictEqual(
1023 restrictToSnapshotColumns(firstBoot.legacyThreads, v1Snapshot.threads),
1024 v1Snapshot.threads.rows,
1025 );
1026 assert.deepStrictEqual(
1027 restrictToSnapshotColumns(firstBoot.legacyMessages, v1Snapshot.messages),
1028 v1Snapshot.messages.rows,
1029 );
1030
1031 assert.equal(firstBoot.importRows.length, ALL_THREADS.length);
1032 const unhydratedRows = firstBoot.importRows.filter(
1033 (row) => row.transcript_imported_at === null,
1034 );
1035 assert.deepStrictEqual(
1036 unhydratedRows.map((row) => row.thread_id),
1037 [SNOOZED_THREAD],
1038 );
1039
1040 const boot2Logs: CapturedLog[] = [];
1041 yield* Effect.scoped(
1042 Effect.gen(function* () {
1043 const sql = yield* SqlClient.SqlClient;
1044 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
1045 const projections = yield* ProjectionStore.ProjectionStoreV2;
1046
1047 assert.equal(yield* importer.pendingThreadCount, 1);
1048 assert.deepStrictEqual(yield* importer.reconcileShells, {
1049 importedThreadCount: 0,
1050 importedMessageCount: 0,
1051 });
1052 yield* sql`DROP TRIGGER fail_snoozed_hydration`;
1053 assert.deepStrictEqual(
1054 yield* importer.ensureTranscript(ThreadId.make(SNOOZED_THREAD)),
1055 { importedThreadCount: 1, importedMessageCount: 1 },
1056 );
1057 const snoozed = yield* projections.getThreadProjection(ThreadId.make(SNOOZED_THREAD));
1058 assert.deepStrictEqual(
1059 snoozed.messages.map((message) => message.id),
1060 [
1061 "message:cutover:snoozed:1",
1062 "message:cutover:snoozed:2",
1063 "message:cutover:snoozed:3",
1064 ],
1065 );
1066 assert.equal(yield* importer.pendingThreadCount, 0);
1067 for (const threadId of ALL_THREADS) {
1068 assert.deepStrictEqual(yield* importer.ensureTranscript(ThreadId.make(threadId)), {
1069 importedThreadCount: 0,
1070 importedMessageCount: 0,
1071 });
1072 }
1073 const migrationEventCount = yield* sql<{ readonly count: number }>`
1074 SELECT COUNT(*) AS count
1075 FROM orchestration_events
1076 WHERE application_event_version = 2
1077 AND event_id LIKE 'migration:v1:%'
1078 `;
1079
1080 assert.equal(migrationEventCount[0]?.count, firstBoot.migrationEventCount + 2);
1081
1082 const restarted = yield* projections.getThreadProjection(longThreadId);
1083 assert.equal(restarted.thread.historyOrigin, "v1_import");
1084 assert.equal(restarted.contextHandoffs.length, 1);
1085 assert.isNotNull(restarted.thread.activeProviderThreadId);
1086 assert.deepStrictEqual(
1087 restarted.runs.map((run) => run.status),
1088 firstBoot.longProjection.runs.map((run) => run.status),
1089 );
1090
1091 assert.deepStrictEqual(
1092 restrictToSnapshotColumns(
1093 yield* sql<Record<string, unknown>>`
1094 SELECT * FROM projection_projects ORDER BY project_id
1095 `,
1096 v1Snapshot.projects,
1097 ),
1098 v1Snapshot.projects.rows,
1099 );
1100 assert.deepStrictEqual(
1101 restrictToSnapshotColumns(
1102 yield* sql<Record<string, unknown>>`
1103 SELECT * FROM projection_threads ORDER BY thread_id
1104 `,
1105 v1Snapshot.threads,
1106 ),
1107 v1Snapshot.threads.rows,
1108 );
1109 assert.deepStrictEqual(
1110 restrictToSnapshotColumns(
1111 yield* sql<Record<string, unknown>>`
1112 SELECT * FROM projection_thread_messages ORDER BY message_id
1113 `,
1114 v1Snapshot.messages,
1115 ),
1116 v1Snapshot.messages.rows,
1117 );
1118 const quickCheck = yield* sql<{ readonly quick_check: string }>`
1119 PRAGMA quick_check
1120 `;
1121 assert.deepStrictEqual(
1122 quickCheck.map((row) => row.quick_check),
1123 ["ok"],
1124 );
1125 }).pipe(
1126 Effect.provide(
1127 layerBoot({
1128 name: "legacy-v1-cutover-restart",
1129 dbPath: copyPath,
1130 workspace,
1131 capturedTurns,
1132 }),
1133 ),
1134 Effect.provideService(
1135 Logger.CurrentLoggers,
1136 new Set([makeCapturingLogger(boot2Logs)]),
1137 ),
1138 ),
1139 );
1140
1141 assert.isTrue(
1142 boot2Logs.some((log) => String(log.message).includes("migration history diverges")),
1143 );
1144
1145 // The seeded source file is the untouched recovery copy: migration
1146 // only ever ran against the file copied out of it.
1147 assert.equal(yield* sha256File(fixturePath), fixtureHashBefore);
1148 }).pipe(Effect.provide(NodeServices.layer)),
1149 ),
1150 );
1151});