Open History to see the last commit
1import { assert, it } from "@effect/vitest";
2import {
3 ComposerContextId,
4 EventId,
5 ProjectId,
6 ProviderInstanceId,
7 type OrchestrationMessageContext,
8 ThreadId,
9} from "@t3tools/contracts";
10import * as DateTime from "effect/DateTime";
11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";
13import * as Layer from "effect/Layer";
14import * as SqlClient from "effect/sql/SqlClient";
15import * as Statement from "effect/sql/Statement";
16
17import { listLinkedPullRequestThreads } from "../../pullRequest/linkedThreads.ts";
18import * as EventSink from "../EventSink.ts";
19import * as EventStore from "../EventStore.ts";
20import * as LegacyV1ThreadImporter from "./LegacyV1ThreadImporter.ts";
21import * as ProjectionReplay from "../testkit/ProjectionReplay.ts";
22import * as ProjectionStore from "../ProjectionStore.ts";
23
24import { layerReaderDatabase, traceSqlRouting } from "../../persistence/testkit/SqliteRouting.ts";
25
26const layerDatabase = layerReaderDatabase;
27const layerEventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(layerDatabase));
28const layerProjectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(layerDatabase));
29const layerStoresProvided = Layer.mergeAll(
30 layerDatabase,
31 layerEventStoreProvided,
32 layerProjectionStoreProvided,
33);
34const layerEventSinkProvided = EventSink.layer.pipe(Layer.provide(layerStoresProvided));
35const layerImporterProvided = LegacyV1ThreadImporter.layer.pipe(
36 Layer.provide(Layer.mergeAll(layerStoresProvided, layerEventSinkProvided)),
37);
38const layerProjectionReplayProvided = ProjectionReplay.layer.pipe(
39 Layer.provide(layerStoresProvided),
40);
41const layerTest = Layer.mergeAll(
42 layerStoresProvided,
43 layerEventSinkProvided,
44 layerImporterProvided,
45 layerProjectionReplayProvided,
46);
47
48const seedPendingTranscripts = Effect.gen(function* () {
49 const sql = yield* SqlClient.SqlClient;
50 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
51 const threadIds = [ThreadId.make("thread:first"), ThreadId.make("thread:later")];
52 yield* sql`
53 INSERT INTO projection_projects (
54 project_id, title, workspace_root, scripts_json, created_at, updated_at
55 ) VALUES (
56 'project:background', 'Legacy project', '/tmp/legacy-background', '[]',
57 '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z'
58 )
59 `;
60 for (const threadId of threadIds) {
61 yield* sql`
62 INSERT INTO projection_threads (
63 thread_id, project_id, title, model_selection_json, runtime_mode,
64 interaction_mode, created_at, updated_at
65 ) VALUES (
66 ${threadId}, 'project:background', 'Legacy thread',
67 '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', 'default',
68 '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z'
69 )
70 `;
71 yield* sql`
72 INSERT INTO projection_thread_messages (
73 message_id, thread_id, role, text, is_streaming, created_at, updated_at
74 ) VALUES
75 (${`${threadId}:old`}, ${threadId}, 'user', 'Old question', 0,
76 '2026-01-01T01:00:00.000Z', '2026-01-01T01:00:00.000Z'),
77 (${`${threadId}:latest`}, ${threadId}, 'user', 'Latest question', 0,
78 '2026-01-02T01:00:00.000Z', '2026-01-02T01:00:00.000Z')
79 `;
80 }
81 yield* importer.reconcileShells;
82});
83
84it.effect("routes shell selectors to readers while imports stay writable", () =>
85 Effect.gen(function* () {
86 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
87 const sql = yield* SqlClient.SqlClient;
88 const { queries, trace } = traceSqlRouting();
89 yield* seedPendingTranscripts.pipe(trace);
90 const selectors = queries.filter(
91 ({ query }) =>
92 query.includes("FROM projection_threads AS thread") ||
93 query.includes("FROM orchestration_v2_legacy_imports AS legacy_import") ||
94 query.includes("FROM projection_thread_messages AS message"),
95 );
96 assert.lengthOf(selectors, 6);
97 assert.deepEqual([...new Set(selectors.map((row) => row.connection))], ["reader"]);
98 assert.deepEqual(yield* importer.reconcileShells, {
99 importedThreadCount: 0,
100 importedMessageCount: 0,
101 });
102 yield* sql.withTransaction(
103 Effect.gen(function* () {
104 yield* sql`
105 INSERT INTO projection_threads (
106 thread_id, project_id, title, model_selection_json, runtime_mode,
107 interaction_mode, created_at, updated_at
108 ) VALUES (
109 'thread:uncommitted', 'project:background', 'Uncommitted legacy title',
110 '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', 'default',
111 '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z'
112 )
113 `;
114 yield* sql`
115 INSERT INTO projection_thread_messages (
116 message_id, thread_id, role, text, is_streaming, created_at, updated_at
117 ) VALUES (
118 'message:uncommitted', 'thread:uncommitted', 'user', 'Uncommitted preview', 0,
119 '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z'
120 )
121 `;
122 yield* sql`
123 UPDATE orchestration_v2_projection_threads
124 SET payload_json = json_remove(payload_json, '$.pinnedAt')
125 WHERE thread_id = 'thread:first'
126 `;
127 yield* sql`
128 UPDATE projection_threads SET pinned_at = '2026-01-01T00:00:00.000Z'
129 WHERE thread_id = 'thread:first'
130 `;
131 const result = yield* importer.reconcileShells;
132 assert.deepEqual(result, { importedThreadCount: 2, importedMessageCount: 1 });
133 const projections = yield* ProjectionStore.ProjectionStoreV2;
134 const repaired = yield* projections.getThreadProjection(ThreadId.make("thread:first"));
135 assert.equal(DateTime.formatIso(repaired.thread.pinnedAt!), "2026-01-01T00:00:00.000Z");
136 const imported = yield* projections.getThreadProjection(
137 ThreadId.make("thread:uncommitted"),
138 );
139 assert.equal(imported.messages[0]?.text, "Uncommitted preview");
140 assert.equal(
141 (yield* projections.getThreadProjection(ThreadId.make("thread:uncommitted"))).thread
142 .title,
143 "Uncommitted legacy title",
144 );
145 }),
146 );
147 }).pipe(Effect.provide(layerTest)),
148);
149
150it.effect(
151 "keeps hydrating later transcripts after a defect and leaves the failed thread pending",
152 () =>
153 Effect.gen(function* () {
154 const sql = yield* SqlClient.SqlClient;
155 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
156 const projections = yield* ProjectionStore.ProjectionStoreV2;
157 yield* seedPendingTranscripts;
158
159 yield* importer.importPendingTranscripts.pipe(
160 Effect.provideService(Statement.CurrentTransformer, (statement) => {
161 const [query, params] = statement.compile();
162 return query.includes("FROM projection_thread_messages") &&
163 params.includes("thread:first")
164 ? Effect.die("transcript read failed")
165 : Effect.succeed(statement);
166 }),
167 );
168
169 const later = yield* projections.getThreadProjection(ThreadId.make("thread:later"));
170 assert.deepStrictEqual(
171 later.messages.map((message) => message.text),
172 ["Old question", "Latest question"],
173 );
174 assert.deepStrictEqual(
175 yield* sql`
176 SELECT thread_id, transcript_imported_at, last_error FROM orchestration_v2_legacy_imports
177 WHERE transcript_imported_at IS NULL OR last_error IS NOT NULL
178 `,
179 [
180 {
181 thread_id: "thread:first",
182 transcript_imported_at: null,
183 last_error: "Transcript hydration failed; retry on next open.",
184 },
185 ],
186 );
187 }).pipe(Effect.provide(layerTest)),
188);
189
190// regression: #15920 one bad legacy message stopped the server from starting
191it.effect("propagates interruption from transcript reads and retry-marker writes", () =>
192 Effect.gen(function* () {
193 const sql = yield* SqlClient.SqlClient;
194 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
195 yield* seedPendingTranscripts;
196 const interrupted = [];
197 for (const phase of ["transcript", "retry-marker"]) {
198 const exit = yield* importer.importPendingTranscripts.pipe(
199 Effect.provideService(Statement.CurrentTransformer, (statement) => {
200 const [query] = statement.compile();
201 const readingTranscript = query.includes("FROM projection_thread_messages");
202 const writingRetryMarker = query.includes("SET last_error =");
203 if (phase === "retry-marker" && readingTranscript) {
204 return Effect.die("transcript read failed");
205 }
206 if (phase === "transcript" ? readingTranscript : writingRetryMarker) {
207 return Effect.interrupt;
208 }
209 return Effect.succeed(statement);
210 }),
211 Effect.exit,
212 );
213 interrupted.push(Exit.hasInterrupts(exit));
214 }
215 assert.deepStrictEqual(interrupted, [true, true]);
216 assert.deepStrictEqual(
217 yield* sql`
218 SELECT thread_id FROM orchestration_v2_legacy_imports
219 WHERE transcript_imported_at IS NOT NULL OR last_error IS NOT NULL
220 `,
221 [],
222 );
223 }).pipe(Effect.provide(layerTest)),
224);
225
226// regression: #30813 a malformed V1 timestamp stopped the server from starting
227it.effect("imports legacy rows with malformed timestamps and keeps the archive marker", () =>
228 Effect.gen(function* () {
229 const sql = yield* SqlClient.SqlClient;
230 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
231 const projections = yield* ProjectionStore.ProjectionStoreV2;
232 const malformed = "2026-07-13T14:13:00.3NZ";
233 yield* sql`
234 INSERT INTO projection_projects (
235 project_id, title, workspace_root, scripts_json, created_at, updated_at
236 ) VALUES (
237 'project:malformed', 'Legacy project', '/tmp/legacy-malformed', '[]',
238 '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z'
239 )
240 `;
241 yield* sql`
242 INSERT INTO projection_threads (
243 thread_id, project_id, title, model_selection_json, runtime_mode,
244 interaction_mode, created_at, updated_at, archived_at
245 ) VALUES (
246 'thread:malformed', 'project:malformed', 'Legacy thread',
247 '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', 'default',
248 '2026-07-13T14:00:00.000Z', ${malformed}, ${malformed}
249 )
250 `;
251 yield* sql`
252 INSERT INTO projection_thread_messages (
253 message_id, thread_id, role, text, is_streaming, created_at, updated_at
254 ) VALUES (
255 'message:malformed', 'thread:malformed', 'user', 'Question', 0,
256 ${malformed}, '2026-07-13T14:13:00.000Z'
257 )
258 `;
259
260 assert.deepStrictEqual(yield* importer.reconcileShells, {
261 importedThreadCount: 1,
262 importedMessageCount: 1,
263 });
264 const projection = yield* projections.getThreadProjection(ThreadId.make("thread:malformed"));
265 assert.isNotNull(projection.thread.archivedAt);
266 assert.deepStrictEqual(
267 projection.messages.map((message) => message.text),
268 ["Question"],
269 );
270 }).pipe(Effect.provide(layerTest)),
271);
272
273it.effect("caches the absent legacy transcript for a native V2 thread", () =>
274 Effect.gen(function* () {
275 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
276 const eventSink = yield* EventSink.EventSinkV2;
277 const projections = yield* ProjectionStore.ProjectionStoreV2;
278 const threadId = ThreadId.make("thread:v2-born");
279 const providerInstanceId = ProviderInstanceId.make("codex");
280 const now = DateTime.makeUnsafe("2026-01-01T00:00:00.000Z");
281 yield* eventSink.write({
282 events: [
283 {
284 id: EventId.make("event:v2-born"),
285 type: "thread.created",
286 threadId,
287 occurredAt: now,
288 payload: {
289 id: threadId,
290 projectId: ProjectId.make("project:v2-born"),
291 title: "Native V2 thread",
292 providerInstanceId,
293 modelSelection: { instanceId: providerInstanceId, model: "gpt-6" },
294 createdBy: "user",
295 creationSource: "web",
296 runtimeMode: "full-access",
297 interactionMode: "default",
298 branch: null,
299 worktreePath: null,
300 activeProviderThreadId: null,
301 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
302 forkedFrom: null,
303 createdAt: now,
304 updatedAt: now,
305 archivedAt: null,
306 settledOverride: null,
307 settledAt: null,
308 lastVisitedAt: null,
309 deletedAt: null,
310 },
311 },
312 ],
313 });
314 const before = yield* projections.getThreadProjection(threadId);
315 let legacyLookups = 0;
316 for (let read = 0; read < 3; read++) {
317 const summary = yield* importer.ensureTranscript(threadId).pipe(
318 Effect.provideService(Statement.CurrentTransformer, (statement) => {
319 const [query, parameters] = statement.compile();
320 if (
321 query.includes("FROM orchestration_v2_legacy_imports") &&
322 parameters[0] === threadId
323 ) {
324 legacyLookups++;
325 }
326 return Effect.succeed(statement);
327 }),
328 );
329 assert.deepStrictEqual(summary, { importedThreadCount: 0, importedMessageCount: 0 });
330 }
331 assert.equal(legacyLookups, 1);
332 assert.deepStrictEqual(yield* projections.getThreadProjection(threadId), before);
333 }).pipe(Effect.provide(layerTest)),
334);
335
336it.layer(layerTest)("LegacyV1ThreadImporter", (it) => {
337 it.effect("imports lightweight shells, hydrates transcripts, and remains idempotent", () =>
338 Effect.gen(function* () {
339 const sql = yield* SqlClient.SqlClient;
340 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
341 const maintenance = yield* ProjectionReplay.ProjectionReplay;
342 const projections = yield* ProjectionStore.ProjectionStoreV2;
343 const eventSink = yield* EventSink.EventSinkV2;
344 const threadId = ThreadId.make("thread:legacy-import");
345
346 yield* sql`
347 INSERT INTO projection_projects (
348 project_id,
349 title,
350 workspace_root,
351 default_model_selection_json,
352 scripts_json,
353 created_at,
354 updated_at,
355 deleted_at
356 ) VALUES (
357 'project:legacy-import',
358 'Legacy project',
359 '/tmp/legacy-project',
360 '{"instanceId":"codex","model":"gpt-5.4"}',
361 '[]',
362 '2026-01-01T00:00:00.000Z',
363 '2026-01-04T00:00:00.000Z',
364 NULL
365 )
366 `;
367 yield* sql`
368 INSERT INTO projection_threads (
369 thread_id,
370 project_id,
371 title,
372 model_selection_json,
373 runtime_mode,
374 interaction_mode,
375 branch,
376 worktree_path,
377 latest_turn_id,
378 created_at,
379 updated_at,
380 archived_at,
381 settled_override,
382 settled_at,
383 unsettled_at,
384 snoozed_until,
385 snoozed_at,
386 pinned_at,
387 pin_order_key,
388 linked_pull_request_json,
389 deleted_at
390 ) VALUES (
391 ${threadId},
392 'project:legacy-import',
393 'Migrated conversation',
394 '{"instanceId":"codex","model":"gpt-5.4"}',
395 'full-access',
396 'default',
397 ' main ',
398 ' /tmp/legacy-project ',
399 NULL,
400 '2026-01-01T00:00:00.000Z',
401 '2026-01-04T00:00:00.000Z',
402 NULL,
403 NULL,
404 NULL,
405 '2026-01-03T12:00:00.000Z',
406 '2026-02-01T00:00:00.000Z',
407 '2026-01-04T00:00:00.000Z',
408 '2026-01-02T00:00:00.000Z',
409 'm',
410 '{"projectId":"project:legacy-import","repository":"pingdotgg/t3code","number":9000,"url":"https://github.com/pingdotgg/t3code/pull/9000"}',
411 NULL
412 )
413 `;
414 yield* sql`
415 INSERT INTO projection_thread_messages (
416 message_id,
417 thread_id,
418 turn_id,
419 role,
420 text,
421 attachments_json,
422 is_streaming,
423 created_at,
424 updated_at
425 ) VALUES
426 (
427 'message:legacy:1',
428 ${threadId},
429 NULL,
430 'user',
431 'First question',
432 '[]',
433 0,
434 '2026-01-01T01:00:00.000Z',
435 '2026-01-01T01:00:00.000Z'
436 ),
437 (
438 'message:legacy:2',
439 ${threadId},
440 NULL,
441 'assistant',
442 'First answer',
443 '[]',
444 0,
445 '2026-01-02T01:00:00.000Z',
446 '2026-01-02T01:00:00.000Z'
447 ),
448 (
449 'message:legacy:3',
450 ${threadId},
451 NULL,
452 'user',
453 'Follow-up question',
454 '[]',
455 0,
456 '2026-01-03T01:00:00.000Z',
457 '2026-01-03T01:00:00.000Z'
458 ),
459 (
460 'message:legacy:4',
461 ${threadId},
462 NULL,
463 'assistant',
464 'Partial answer',
465 '[]',
466 1,
467 '2026-01-04T01:00:00.000Z',
468 '2026-01-04T01:00:00.000Z'
469 )
470 `;
471
472 const validContext = {
473 version: 1,
474 records: [
475 {
476 version: 1,
477 contextId: ComposerContextId.make("ctx_1"),
478 kind: "skill",
479 label: "$test",
480 name: "test",
481 },
482 ],
483 } satisfies OrchestrationMessageContext;
484 yield* sql`
485 UPDATE projection_thread_messages
486 SET context_json = CASE message_id
487 WHEN 'message:legacy:2' THEN '{broken-json'
488 WHEN 'message:legacy:3' THEN '{"version":2,"records":[]}'
489 ELSE ${JSON.stringify(validContext)}
490 END
491 WHERE thread_id = ${threadId}
492 `;
493 const sourceMessages = yield* sql`
494 SELECT * FROM projection_thread_messages WHERE thread_id = ${threadId} ORDER BY message_id
495 `;
496
497 yield* sql`
498 INSERT INTO projection_thread_pull_requests (
499 thread_id, host, repository, number, url, source, linked_at, snapshot_json
500 ) VALUES
501 (${threadId}, 'github.com', 'pingdotgg/t3code', 9002,
502 'https://github.com/pingdotgg/t3code/pull/9002', 'created',
503 '2026-01-03T00:00:00.000Z',
504 '{"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"}'),
505 (${threadId}, 'github.com', 'pingdotgg/t3code', 9003,
506 'https://github.com/pingdotgg/t3code/pull/9003', 'manual',
507 '2026-01-04T00:00:00.000Z', NULL)
508 `;
509
510 assert.equal(yield* importer.pendingThreadCount, 1);
511 const shellImport = yield* importer.reconcileShells;
512 assert.equal(yield* importer.pendingThreadCount, 1);
513 assert.deepStrictEqual(shellImport, {
514 importedThreadCount: 1,
515 importedMessageCount: 2,
516 });
517 const shellEventCount = yield* sql<{ readonly count: number }>`
518 SELECT COUNT(*) AS count
519 FROM orchestration_events
520 WHERE application_event_version = 2
521 AND aggregate_kind = 'thread'
522 AND stream_id = ${threadId}
523 `;
524 assert.equal(shellEventCount[0]?.count, 6);
525
526 assert.isTrue((yield* maintenance.verify).valid);
527 const shellProjection = yield* projections.getThreadProjection(threadId);
528 assert.equal(shellProjection.thread.historyOrigin, "v1_import");
529 assert.equal(shellProjection.thread.branch, "main");
530 assert.equal(shellProjection.thread.worktreePath, "/tmp/legacy-project");
531 assert.deepEqual(
532 shellProjection.thread.pinnedAt,
533 DateTime.makeUnsafe("2026-01-02T00:00:00.000Z"),
534 );
535 assert.equal(shellProjection.thread.pinOrderKey, "m");
536 assert.deepEqual(
537 shellProjection.thread.snoozedUntil,
538 DateTime.makeUnsafe("2026-02-01T00:00:00.000Z"),
539 );
540 assert.deepEqual(
541 shellProjection.thread.unsettledAt,
542 DateTime.makeUnsafe("2026-01-03T12:00:00.000Z"),
543 );
544 assert.equal(shellProjection.thread.linkedPullRequest?.number, 9000);
545 assert.deepStrictEqual(
546 (shellProjection.thread.pullRequests ?? []).map((link) => link.number),
547 [9002, 9003, 9000],
548 );
549 assert.equal(shellProjection.thread.pullRequests?.[0]?.snapshot?.title, "Second PR");
550 assert.deepStrictEqual(
551 (yield* listLinkedPullRequestThreads({
552 host: "github.com",
553 repository: "pingdotgg/t3code",
554 number: 9002,
555 })).threads.map((thread) => thread.id),
556 [threadId],
557 );
558 const shellSnapshot = yield* projections.getShellSnapshot();
559 assert.equal(
560 shellSnapshot.threads.find((thread) => thread.id === threadId)?.historyOrigin,
561 "v1_import",
562 );
563 assert.deepStrictEqual(
564 shellProjection.messages.map((message) => message.id),
565 ["message:legacy:3", "message:legacy:4"],
566 );
567 assert.deepStrictEqual(
568 shellProjection.messages.map((message) => [message.text, message.context]),
569 [
570 ["Follow-up question", undefined],
571 ["Partial answer", validContext],
572 ],
573 );
574
575 const renamedAt = DateTime.makeUnsafe("2026-01-05T00:00:00.000Z");
576 yield* eventSink.write({
577 events: [
578 {
579 id: EventId.make("event:legacy-import:metadata-after-shell"),
580 type: "thread.metadata-updated",
581 threadId,
582 providerInstanceId: shellProjection.thread.providerInstanceId,
583 occurredAt: renamedAt,
584 payload: {
585 ...shellProjection.thread,
586 title: "Renamed after shell import",
587 runtimeMode: "approval-required",
588 interactionMode: "plan",
589 archivedAt: renamedAt,
590 settledOverride: "settled",
591 settledAt: renamedAt,
592 updatedAt: renamedAt,
593 },
594 },
595 ],
596 });
597
598 const transcriptImport = yield* importer.importPendingTranscripts;
599 assert.equal(yield* importer.pendingThreadCount, 0);
600 assert.deepStrictEqual(transcriptImport, {
601 importedThreadCount: 1,
602 importedMessageCount: 2,
603 });
604 const projection = yield* projections.getThreadProjection(threadId);
605 assert.equal(projection.thread.title, "Renamed after shell import");
606 assert.isNull(projection.thread.lastVisitedAt);
607 assert.equal(projection.thread.runtimeMode, "approval-required");
608 assert.equal(projection.thread.interactionMode, "plan");
609 assert.deepEqual(projection.thread.archivedAt, renamedAt);
610 assert.equal(projection.thread.settledOverride, "settled");
611 assert.deepEqual(projection.thread.settledAt, renamedAt);
612 assert.deepStrictEqual(
613 projection.messages.map((message) => message.id),
614 ["message:legacy:1", "message:legacy:2", "message:legacy:3", "message:legacy:4"],
615 );
616 const expectedMessages = [
617 ["First question", validContext],
618 ["First answer", undefined],
619 ["Follow-up question", undefined],
620 ["Partial answer", validContext],
621 ];
622 assert.deepStrictEqual(
623 projection.messages.map((message) => [message.text, message.context]),
624 expectedMessages,
625 );
626 assert.deepStrictEqual(
627 yield* sql`
628 SELECT * FROM projection_thread_messages WHERE thread_id = ${threadId} ORDER BY message_id
629 `,
630 sourceMessages,
631 );
632 assert.deepStrictEqual(
633 projection.turnItems
634 .filter(
635 (
636 item,
637 ): item is Extract<
638 (typeof projection.turnItems)[number],
639 { readonly type: "user_message" | "assistant_message" }
640 > => item.type === "user_message" || item.type === "assistant_message",
641 )
642 .map((item) => [item.messageId, item.ordinal, item.status]),
643 [
644 ["message:legacy:1", 1, "completed"],
645 ["message:legacy:2", 2, "completed"],
646 ["message:legacy:3", 3, "completed"],
647 ["message:legacy:4", 4, "interrupted"],
648 ],
649 );
650
651 yield* eventSink.write({
652 events: [
653 {
654 id: EventId.make("event:legacy-import:explicit-unpin"),
655 type: "thread.unpinned",
656 threadId,
657 providerInstanceId: projection.thread.providerInstanceId,
658 occurredAt: renamedAt,
659 payload: { ...projection.thread, pinnedAt: null, pinOrderKey: null },
660 },
661 ],
662 });
663 yield* sql`
664 UPDATE orchestration_v2_projection_threads
665 SET payload_json = json_remove(
666 payload_json,
667 '$.snoozedUntil',
668 '$.snoozedAt',
669 '$.unsettledAt',
670 '$.linkedPullRequest',
671 '$.pullRequests'
672 )
673 WHERE thread_id = ${threadId}
674 `;
675 assert.deepStrictEqual(yield* importer.reconcileShells, {
676 importedThreadCount: 1,
677 importedMessageCount: 0,
678 });
679 const repaired = yield* projections.getThreadProjection(threadId);
680 assert.isNull(repaired.thread.pinnedAt);
681 assert.isNull(repaired.thread.pinOrderKey);
682 assert.deepEqual(
683 repaired.thread.snoozedUntil,
684 DateTime.makeUnsafe("2026-02-01T00:00:00.000Z"),
685 );
686 assert.deepEqual(
687 repaired.thread.unsettledAt,
688 DateTime.makeUnsafe("2026-01-03T12:00:00.000Z"),
689 );
690 assert.equal(repaired.thread.linkedPullRequest?.number, 9000);
691 assert.deepStrictEqual(
692 (repaired.thread.pullRequests ?? []).map((link) => link.number),
693 [9002, 9003, 9000],
694 );
695 assert.equal(repaired.thread.pullRequests?.[0]?.snapshot?.title, "Second PR");
696 const eventCountBeforeRetry = yield* sql<{ readonly count: number }>`
697 SELECT COUNT(*) AS count
698 FROM orchestration_events
699 WHERE application_event_version = 2
700 AND aggregate_kind = 'thread'
701 AND stream_id = ${threadId}
702 `;
703 assert.deepStrictEqual(yield* importer.reconcileShells, {
704 importedThreadCount: 0,
705 importedMessageCount: 0,
706 });
707 assert.deepStrictEqual(yield* importer.ensureTranscript(threadId), {
708 importedThreadCount: 0,
709 importedMessageCount: 0,
710 });
711 const eventCountAfterRetry = yield* sql<{ readonly count: number }>`
712 SELECT COUNT(*) AS count
713 FROM orchestration_events
714 WHERE application_event_version = 2
715 AND aggregate_kind = 'thread'
716 AND stream_id = ${threadId}
717 `;
718 assert.equal(eventCountAfterRetry[0]?.count, eventCountBeforeRetry[0]?.count);
719 }),
720 );
721
722 it.effect("repairs newly added metadata after an earlier metadata repair", () =>
723 Effect.gen(function* () {
724 const sql = yield* SqlClient.SqlClient;
725 const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
726 const maintenance = yield* ProjectionReplay.ProjectionReplay;
727 const projections = yield* ProjectionStore.ProjectionStoreV2;
728 const eventSink = yield* EventSink.EventSinkV2;
729 const threadId = ThreadId.make("thread:legacy-metadata-upgrade");
730 const previousRepairId = EventId.make(`migration:v1:thread:${threadId}:metadata-repair`);
731
732 yield* sql`
733 INSERT INTO projection_projects (
734 project_id, title, workspace_root, scripts_json, created_at, updated_at
735 ) VALUES (
736 'project:legacy-metadata-upgrade',
737 'Legacy project',
738 '/tmp/legacy-metadata-upgrade',
739 '[]',
740 '2026-01-01T00:00:00.000Z',
741 '2026-01-01T00:00:00.000Z'
742 )
743 `;
744 yield* sql`
745 INSERT INTO projection_threads (
746 thread_id, project_id, title, model_selection_json, runtime_mode,
747 interaction_mode, created_at, updated_at, pinned_at, pin_order_key,
748 linked_pull_request_json, branch_pull_request_json, active_order_key
749 ) VALUES (
750 ${threadId},
751 'project:legacy-metadata-upgrade',
752 'Original v1 title',
753 '{"instanceId":"codex","model":"gpt-5.4"}',
754 'full-access',
755 'default',
756 '2026-01-01T00:00:00.000Z',
757 '2026-01-01T00:00:00.000Z',
758 '2026-01-02T00:00:00.000Z',
759 'm',
760 '{"projectId":"project:legacy-metadata-upgrade","repository":"pingdotgg/t3code","number":9000,"url":"https://github.com/pingdotgg/t3code/pull/9000"}',
761 '{"projectId":"project:legacy-metadata-upgrade","repository":"pingdotgg/t3code","number":9001,"url":"https://github.com/pingdotgg/t3code/pull/9001"}',
762 'az'
763 )
764 `;
765 yield* importer.reconcileShells;
766 yield* maintenance.rebuild;
767 const shellProjection = yield* projections.getThreadProjection(threadId);
768 assert.deepStrictEqual(
769 (shellProjection.thread.pullRequests ?? []).map((link) => link.number),
770 [9000],
771 );
772 const previousRepairThread = {
773 ...shellProjection.thread,
774 title: "Renamed in v2",
775 pinnedAt: null,
776 pinOrderKey: null,
777 linkedPullRequest: null,
778 pullRequests: [],
779 };
780 delete previousRepairThread.branchPullRequest;
781 delete previousRepairThread.activeOrderKey;
782 yield* eventSink.write({
783 events: [
784 {
785 id: previousRepairId,
786 type: "thread.metadata-updated",
787 threadId,
788 providerInstanceId: previousRepairThread.providerInstanceId,
789 occurredAt: DateTime.makeUnsafe("2026-01-03T00:00:00.000Z"),
790 payload: previousRepairThread,
791 },
792 ],
793 });
794
795 assert.deepStrictEqual(yield* importer.reconcileShells, {
796 importedThreadCount: 1,
797 importedMessageCount: 0,
798 });
799 const repaired = yield* projections.getThreadProjection(threadId);
800 assert.equal(repaired.thread.title, "Renamed in v2");
801 assert.isNull(repaired.thread.pinnedAt);
802 assert.isNull(repaired.thread.pinOrderKey);
803 assert.isNull(repaired.thread.linkedPullRequest);
804 assert.deepStrictEqual(repaired.thread.pullRequests, []);
805 assert.equal(repaired.thread.branchPullRequest?.number, 9001);
806 assert.equal(repaired.thread.activeOrderKey, "az");
807
808 const eventsBeforeRetry = yield* sql<{ readonly event_id: string }>`
809 SELECT event_id
810 FROM orchestration_events
811 WHERE application_event_version = 2 AND stream_id = ${threadId}
812 ORDER BY sequence
813 `;
814 assert.equal(eventsBeforeRetry.length, 4);
815 assert.isTrue(eventsBeforeRetry.some((event) => event.event_id === previousRepairId));
816 assert.deepStrictEqual(yield* importer.reconcileShells, {
817 importedThreadCount: 0,
818 importedMessageCount: 0,
819 });
820
821 assert.isTrue((yield* maintenance.rebuild).valid);
822 const replayed = yield* projections.getThreadProjection(threadId);
823 assert.deepStrictEqual(replayed.thread, repaired.thread);
824 assert.deepStrictEqual(
825 yield* Effect.gen(function* () {
826 const restartedImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
827 return yield* restartedImporter.reconcileShells;
828 }).pipe(Effect.provide(LegacyV1ThreadImporter.layer)),
829 { importedThreadCount: 0, importedMessageCount: 0 },
830 );
831 const eventsAfterRestart = yield* sql<{ readonly event_id: string }>`
832 SELECT event_id
833 FROM orchestration_events
834 WHERE application_event_version = 2 AND stream_id = ${threadId}
835 ORDER BY sequence
836 `;
837 assert.deepStrictEqual(eventsAfterRestart, eventsBeforeRetry);
838 }),
839 );
840});