1import { assert, it } from "@effect/vitest";
2import { EventId, ThreadId } from "@t3tools/contracts";
3import * as DateTime from "effect/DateTime";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Tracer from "effect/Tracer";
7import * as SqlClient from "effect/sql/SqlClient";
8
9import * as SqlitePersistence from "../../persistence/Sqlite.ts";
10import { listLinkedPullRequestThreads } from "../../pullRequest/linkedThreads.ts";
11import * as EventSink from "../EventSink.ts";
12import * as EventStore from "../EventStore.ts";
13import * as LegacyV1ThreadImporter from "./LegacyV1ThreadImporter.ts";
14import * as ProjectionMaintenance from "../ProjectionMaintenance.ts";
15import * as ProjectionStore from "../ProjectionStore.ts";
16
17const layerDatabase = SqlitePersistence.layerMemory;
18const layerEventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(layerDatabase));
19const layerProjectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(layerDatabase));
20const layerStoresProvided = Layer.mergeAll(
21 layerDatabase,
22 layerEventStoreProvided,
23 layerProjectionStoreProvided,
24);
25const layerEventSinkProvided = EventSink.layer.pipe(Layer.provide(layerStoresProvided));
26const layerImporterProvided = LegacyV1ThreadImporter.layer.pipe(
27 Layer.provide(Layer.mergeAll(layerStoresProvided, layerEventSinkProvided)),
28);
29const layerProjectionMaintenanceProvided = ProjectionMaintenance.layer.pipe(
30 Layer.provide(layerStoresProvided),
31);
32const layerTest = Layer.mergeAll(
33 layerStoresProvided,
34 layerEventSinkProvided,
35 layerImporterProvided,
36 layerProjectionMaintenanceProvided,
37);
38
39it.layer(layerTest)("LegacyV1ThreadImporter", (it) => {
40 it.effect("uses the created-thread index for startup migration checks", () =>
41 Effect.gen(function* () {
42 const sql = yield* SqlClient.SqlClient;
43 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
44 const statements: string[] = [];
45 const tracer = Tracer.make({
46 span(options) {
47 const span = new Tracer.NativeSpan(options);
48 const end = span.end.bind(span);
49 span.end = (endTime, exit) => {
50 end(endTime, exit);
51 const query = span.attributes.get("db.query.text");
52 if (
53 typeof query === "string" &&
54 query.includes("FROM projection_threads AS thread") &&
55 query.includes("FROM orchestration_events AS event")
56 ) {
57 statements.push(query);
58 }
59 };
60 return span;
61 },
62 });
63
64 assert.equal(yield* importer.pendingThreadCount.pipe(Effect.withTracer(tracer)), 0);
65 assert.deepStrictEqual(yield* importer.reconcileShells.pipe(Effect.withTracer(tracer)), {
66 importedThreadCount: 0,
67 importedMessageCount: 0,
68 });
69 assert.lengthOf(statements, 2);
70 for (const statement of statements) {
71 const plan = yield* sql.unsafe<{ readonly detail: string }>(
72 `EXPLAIN QUERY PLAN ${statement}`,
73 );
74 assert.match(
75 plan.map((row) => row.detail).join("\n"),
76 /SEARCH event USING INDEX orchestration_events_v2_created_threads_idx \(stream_id=\?\)/,
77 );
78 }
79 }),
80 );
81
82 it.effect("imports lightweight shells, hydrates transcripts, and remains idempotent", () =>
83 Effect.gen(function* () {
84 const sql = yield* SqlClient.SqlClient;
85 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
86 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
87 const projections = yield* ProjectionStore.ProjectionStoreV2;
88 const eventSink = yield* EventSink.EventSinkV2;
89 const threadId = ThreadId.make("thread:legacy-import");
90
91 yield* sql`
92 INSERT INTO projection_projects (
93 project_id,
94 title,
95 workspace_root,
96 default_model_selection_json,
97 scripts_json,
98 created_at,
99 updated_at,
100 deleted_at
101 ) VALUES (
102 'project:legacy-import',
103 'Legacy project',
104 '/tmp/legacy-project',
105 '{"instanceId":"codex","model":"gpt-5.4"}',
106 '[]',
107 '2026-01-01T00:00:00.000Z',
108 '2026-01-04T00:00:00.000Z',
109 NULL
110 )
111 `;
112 yield* sql`
113 INSERT INTO projection_threads (
114 thread_id,
115 project_id,
116 title,
117 model_selection_json,
118 runtime_mode,
119 interaction_mode,
120 branch,
121 worktree_path,
122 latest_turn_id,
123 created_at,
124 updated_at,
125 archived_at,
126 settled_override,
127 settled_at,
128 unsettled_at,
129 snoozed_until,
130 snoozed_at,
131 pinned_at,
132 pin_order_key,
133 linked_pull_request_json,
134 deleted_at
135 ) VALUES (
136 ${threadId},
137 'project:legacy-import',
138 'Migrated conversation',
139 '{"instanceId":"codex","model":"gpt-5.4"}',
140 'full-access',
141 'default',
142 ' main ',
143 ' /tmp/legacy-project ',
144 NULL,
145 '2026-01-01T00:00:00.000Z',
146 '2026-01-04T00:00:00.000Z',
147 NULL,
148 NULL,
149 NULL,
150 '2026-01-03T12:00:00.000Z',
151 '2026-02-01T00:00:00.000Z',
152 '2026-01-04T00:00:00.000Z',
153 '2026-01-02T00:00:00.000Z',
154 'm',
155 '{"projectId":"project:legacy-import","repository":"pingdotgg/t3code","number":9000,"url":"https://github.com/pingdotgg/t3code/pull/9000"}',
156 NULL
157 )
158 `;
159 yield* sql`
160 INSERT INTO projection_thread_messages (
161 message_id,
162 thread_id,
163 turn_id,
164 role,
165 text,
166 attachments_json,
167 is_streaming,
168 created_at,
169 updated_at
170 ) VALUES
171 (
172 'message:legacy:1',
173 ${threadId},
174 NULL,
175 'user',
176 'First question',
177 '[]',
178 0,
179 '2026-01-01T01:00:00.000Z',
180 '2026-01-01T01:00:00.000Z'
181 ),
182 (
183 'message:legacy:2',
184 ${threadId},
185 NULL,
186 'assistant',
187 'First answer',
188 '[]',
189 0,
190 '2026-01-02T01:00:00.000Z',
191 '2026-01-02T01:00:00.000Z'
192 ),
193 (
194 'message:legacy:3',
195 ${threadId},
196 NULL,
197 'user',
198 'Follow-up question',
199 '[]',
200 0,
201 '2026-01-03T01:00:00.000Z',
202 '2026-01-03T01:00:00.000Z'
203 ),
204 (
205 'message:legacy:4',
206 ${threadId},
207 NULL,
208 'assistant',
209 'Partial answer',
210 '[]',
211 1,
212 '2026-01-04T01:00:00.000Z',
213 '2026-01-04T01:00:00.000Z'
214 )
215 `;
216
217 yield* sql`
218 INSERT INTO projection_thread_pull_requests (
219 thread_id, host, repository, number, url, source, linked_at, snapshot_json
220 ) VALUES
221 (${threadId}, 'github.com', 'pingdotgg/t3code', 9002,
222 'https://github.com/pingdotgg/t3code/pull/9002', 'created',
223 '2026-01-03T00:00:00.000Z',
224 '{"state":"open","title":"Second PR","headBranch":"feature-two","baseBranch":"main","isDraft":false,"updatedAt":"2026-01-03T00:00:00.000Z","syncedAt":"2026-01-03T00:00:00.000Z"}'),
225 (${threadId}, 'github.com', 'pingdotgg/t3code', 9003,
226 'https://github.com/pingdotgg/t3code/pull/9003', 'manual',
227 '2026-01-04T00:00:00.000Z', NULL)
228 `;
229
230 assert.equal(yield* importer.pendingThreadCount, 1);
231 const shellImport = yield* importer.reconcileShells;
232 assert.equal(yield* importer.pendingThreadCount, 1);
233 assert.deepStrictEqual(shellImport, {
234 importedThreadCount: 1,
235 importedMessageCount: 2,
236 });
237 const shellEventCount = yield* sql<{ readonly count: number }>`
238 SELECT COUNT(*) AS count
239 FROM orchestration_events
240 WHERE application_event_version = 2
241 AND aggregate_kind = 'thread'
242 AND stream_id = ${threadId}
243 `;
244 assert.equal(shellEventCount[0]?.count, 6);
245
246 assert.isTrue((yield* maintenance.verify).valid);
247 const shellProjection = yield* projections.getThreadProjection(threadId);
248 assert.equal(shellProjection.thread.historyOrigin, "v1_import");
249 assert.equal(shellProjection.thread.branch, "main");
250 assert.equal(shellProjection.thread.worktreePath, "/tmp/legacy-project");
251 assert.deepEqual(
252 shellProjection.thread.pinnedAt,
253 DateTime.makeUnsafe("2026-01-02T00:00:00.000Z"),
254 );
255 assert.equal(shellProjection.thread.pinOrderKey, "m");
256 assert.deepEqual(
257 shellProjection.thread.snoozedUntil,
258 DateTime.makeUnsafe("2026-02-01T00:00:00.000Z"),
259 );
260 assert.deepEqual(
261 shellProjection.thread.unsettledAt,
262 DateTime.makeUnsafe("2026-01-03T12:00:00.000Z"),
263 );
264 assert.equal(shellProjection.thread.linkedPullRequest?.number, 9000);
265 assert.deepStrictEqual(
266 (shellProjection.thread.pullRequests ?? []).map((link) => link.number),
267 [9002, 9003, 9000],
268 );
269 assert.equal(shellProjection.thread.pullRequests?.[0]?.snapshot?.title, "Second PR");
270 assert.deepStrictEqual(
271 (yield* listLinkedPullRequestThreads({
272 host: "github.com",
273 repository: "pingdotgg/t3code",
274 number: 9002,
275 })).threads.map((thread) => thread.id),
276 [threadId],
277 );
278 const shellSnapshot = yield* projections.getShellSnapshot();
279 assert.equal(
280 shellSnapshot.threads.find((thread) => thread.id === threadId)?.historyOrigin,
281 "v1_import",
282 );
283 assert.deepStrictEqual(
284 shellProjection.messages.map((message) => message.id),
285 ["message:legacy:3", "message:legacy:4"],
286 );
287
288 const renamedAt = DateTime.makeUnsafe("2026-01-05T00:00:00.000Z");
289 yield* eventSink.write({
290 events: [
291 {
292 id: EventId.make("event:legacy-import:metadata-after-shell"),
293 type: "thread.metadata-updated",
294 threadId,
295 providerInstanceId: shellProjection.thread.providerInstanceId,
296 occurredAt: renamedAt,
297 payload: {
298 ...shellProjection.thread,
299 title: "Renamed after shell import",
300 runtimeMode: "approval-required",
301 interactionMode: "plan",
302 archivedAt: renamedAt,
303 settledOverride: "settled",
304 settledAt: renamedAt,
305 updatedAt: renamedAt,
306 },
307 },
308 ],
309 });
310
311 const transcriptImport = yield* importer.ensureTranscript(threadId);
312 assert.equal(yield* importer.pendingThreadCount, 0);
313 assert.deepStrictEqual(transcriptImport, {
314 importedThreadCount: 1,
315 importedMessageCount: 2,
316 });
317 const projection = yield* projections.getThreadProjection(threadId);
318 assert.equal(projection.thread.title, "Renamed after shell import");
319 assert.equal(projection.thread.runtimeMode, "approval-required");
320 assert.equal(projection.thread.interactionMode, "plan");
321 assert.deepEqual(projection.thread.archivedAt, renamedAt);
322 assert.equal(projection.thread.settledOverride, "settled");
323 assert.deepEqual(projection.thread.settledAt, renamedAt);
324 assert.deepStrictEqual(
325 projection.messages.map((message) => message.id),
326 ["message:legacy:1", "message:legacy:2", "message:legacy:3", "message:legacy:4"],
327 );
328 assert.deepStrictEqual(
329 projection.turnItems
330 .filter(
331 (
332 item,
333 ): item is Extract<
334 (typeof projection.turnItems)[number],
335 { readonly type: "user_message" | "assistant_message" }
336 > => item.type === "user_message" || item.type === "assistant_message",
337 )
338 .map((item) => [item.messageId, item.ordinal, item.status]),
339 [
340 ["message:legacy:1", 1, "completed"],
341 ["message:legacy:2", 2, "completed"],
342 ["message:legacy:3", 3, "completed"],
343 ["message:legacy:4", 4, "interrupted"],
344 ],
345 );
346
347 yield* eventSink.write({
348 events: [
349 {
350 id: EventId.make("event:legacy-import:explicit-unpin"),
351 type: "thread.unpinned",
352 threadId,
353 providerInstanceId: projection.thread.providerInstanceId,
354 occurredAt: renamedAt,
355 payload: { ...projection.thread, pinnedAt: null, pinOrderKey: null },
356 },
357 ],
358 });
359 yield* sql`
360 UPDATE orchestration_v2_projection_threads
361 SET payload_json = json_remove(
362 payload_json,
363 '$.snoozedUntil',
364 '$.snoozedAt',
365 '$.unsettledAt',
366 '$.linkedPullRequest',
367 '$.pullRequests'
368 )
369 WHERE thread_id = ${threadId}
370 `;
371 assert.deepStrictEqual(yield* importer.reconcileShells, {
372 importedThreadCount: 1,
373 importedMessageCount: 0,
374 });
375 const repaired = yield* projections.getThreadProjection(threadId);
376 assert.isNull(repaired.thread.pinnedAt);
377 assert.isNull(repaired.thread.pinOrderKey);
378 assert.deepEqual(
379 repaired.thread.snoozedUntil,
380 DateTime.makeUnsafe("2026-02-01T00:00:00.000Z"),
381 );
382 assert.deepEqual(
383 repaired.thread.unsettledAt,
384 DateTime.makeUnsafe("2026-01-03T12:00:00.000Z"),
385 );
386 assert.equal(repaired.thread.linkedPullRequest?.number, 9000);
387 assert.deepStrictEqual(
388 (repaired.thread.pullRequests ?? []).map((link) => link.number),
389 [9002, 9003, 9000],
390 );
391 assert.equal(repaired.thread.pullRequests?.[0]?.snapshot?.title, "Second PR");
392 const eventCountBeforeRetry = yield* sql<{ readonly count: number }>`
393 SELECT COUNT(*) AS count
394 FROM orchestration_events
395 WHERE application_event_version = 2
396 AND aggregate_kind = 'thread'
397 AND stream_id = ${threadId}
398 `;
399 assert.deepStrictEqual(yield* importer.reconcileShells, {
400 importedThreadCount: 0,
401 importedMessageCount: 0,
402 });
403 assert.deepStrictEqual(yield* importer.ensureTranscript(threadId), {
404 importedThreadCount: 0,
405 importedMessageCount: 0,
406 });
407 const eventCountAfterRetry = yield* sql<{ readonly count: number }>`
408 SELECT COUNT(*) AS count
409 FROM orchestration_events
410 WHERE application_event_version = 2
411 AND aggregate_kind = 'thread'
412 AND stream_id = ${threadId}
413 `;
414 assert.equal(eventCountAfterRetry[0]?.count, eventCountBeforeRetry[0]?.count);
415 }),
416 );
417
418 it.effect("repairs newly added metadata after an earlier metadata repair", () =>
419 Effect.gen(function* () {
420 const sql = yield* SqlClient.SqlClient;
421 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
422 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
423 const projections = yield* ProjectionStore.ProjectionStoreV2;
424 const eventSink = yield* EventSink.EventSinkV2;
425 const threadId = ThreadId.make("thread:legacy-metadata-upgrade");
426 const previousRepairId = EventId.make(`migration:v1:thread:${threadId}:metadata-repair`);
427
428 yield* sql`
429 INSERT INTO projection_projects (
430 project_id, title, workspace_root, scripts_json, created_at, updated_at
431 ) VALUES (
432 'project:legacy-metadata-upgrade',
433 'Legacy project',
434 '/tmp/legacy-metadata-upgrade',
435 '[]',
436 '2026-01-01T00:00:00.000Z',
437 '2026-01-01T00:00:00.000Z'
438 )
439 `;
440 yield* sql`
441 INSERT INTO projection_threads (
442 thread_id, project_id, title, model_selection_json, runtime_mode,
443 interaction_mode, created_at, updated_at, pinned_at, pin_order_key,
444 linked_pull_request_json, branch_pull_request_json, active_order_key
445 ) VALUES (
446 ${threadId},
447 'project:legacy-metadata-upgrade',
448 'Original v1 title',
449 '{"instanceId":"codex","model":"gpt-5.4"}',
450 'full-access',
451 'default',
452 '2026-01-01T00:00:00.000Z',
453 '2026-01-01T00:00:00.000Z',
454 '2026-01-02T00:00:00.000Z',
455 'm',
456 '{"projectId":"project:legacy-metadata-upgrade","repository":"pingdotgg/t3code","number":9000,"url":"https://github.com/pingdotgg/t3code/pull/9000"}',
457 '{"projectId":"project:legacy-metadata-upgrade","repository":"pingdotgg/t3code","number":9001,"url":"https://github.com/pingdotgg/t3code/pull/9001"}',
458 'az'
459 )
460 `;
461 yield* importer.reconcileShells;
462 yield* maintenance.rebuild;
463 const shellProjection = yield* projections.getThreadProjection(threadId);
464 assert.deepStrictEqual(
465 (shellProjection.thread.pullRequests ?? []).map((link) => link.number),
466 [9000],
467 );
468 const previousRepairThread = {
469 ...shellProjection.thread,
470 title: "Renamed in v2",
471 pinnedAt: null,
472 pinOrderKey: null,
473 linkedPullRequest: null,
474 pullRequests: [],
475 };
476 delete previousRepairThread.branchPullRequest;
477 delete previousRepairThread.activeOrderKey;
478 yield* eventSink.write({
479 events: [
480 {
481 id: previousRepairId,
482 type: "thread.metadata-updated",
483 threadId,
484 providerInstanceId: previousRepairThread.providerInstanceId,
485 occurredAt: DateTime.makeUnsafe("2026-01-03T00:00:00.000Z"),
486 payload: previousRepairThread,
487 },
488 ],
489 });
490
491 assert.deepStrictEqual(yield* importer.reconcileShells, {
492 importedThreadCount: 1,
493 importedMessageCount: 0,
494 });
495 const repaired = yield* projections.getThreadProjection(threadId);
496 assert.equal(repaired.thread.title, "Renamed in v2");
497 assert.isNull(repaired.thread.pinnedAt);
498 assert.isNull(repaired.thread.pinOrderKey);
499 assert.isNull(repaired.thread.linkedPullRequest);
500 assert.deepStrictEqual(repaired.thread.pullRequests, []);
501 assert.equal(repaired.thread.branchPullRequest?.number, 9001);
502 assert.equal(repaired.thread.activeOrderKey, "az");
503
504 const eventsBeforeRetry = yield* sql<{ readonly event_id: string }>`
505 SELECT event_id
506 FROM orchestration_events
507 WHERE application_event_version = 2 AND stream_id = ${threadId}
508 ORDER BY sequence
509 `;
510 assert.equal(eventsBeforeRetry.length, 4);
511 assert.isTrue(eventsBeforeRetry.some((event) => event.event_id === previousRepairId));
512 assert.deepStrictEqual(yield* importer.reconcileShells, {
513 importedThreadCount: 0,
514 importedMessageCount: 0,
515 });
516
517 assert.isTrue((yield* maintenance.rebuild).valid);
518 const replayed = yield* projections.getThreadProjection(threadId);
519 assert.deepStrictEqual(replayed.thread, repaired.thread);
520 assert.deepStrictEqual(
521 yield* Effect.gen(function* () {
522 const restartedImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
523 return yield* restartedImporter.reconcileShells;
524 }).pipe(Effect.provide(LegacyV1ThreadImporter.layer)),
525 { importedThreadCount: 0, importedMessageCount: 0 },
526 );
527 const eventsAfterRestart = yield* sql<{ readonly event_id: string }>`
528 SELECT event_id
529 FROM orchestration_events
530 WHERE application_event_version = 2 AND stream_id = ${threadId}
531 ORDER BY sequence
532 `;
533 assert.deepStrictEqual(eventsAfterRestart, eventsBeforeRetry);
534 }),
535 );
536});