1import {
2 threadPullRequestKeysEqual,
3 threadPullRequestsOf,
4} from "../../sourceControl/threadPullRequests.ts";
5import {
6 ChatAttachment,
7 OrchestrationMessageContext,
8 DEFAULT_MODEL,
9 EventId,
10 MessageId,
11 ModelSelection,
12 type OrchestrationV2AppThread,
13 OrchestrationV2AppThreadJson,
14 type OrchestrationV2ConversationMessage,
15 type OrchestrationV2DomainEvent,
16 type OrchestrationV2TurnItem,
17 ProjectId,
18 ProviderInstanceId,
19 ThreadId,
20 ThreadLinkedPullRequest,
21 ThreadPullRequestLink,
22 TurnItemId,
23} from "@t3tools/contracts";
24import * as KeyedLock from "@t3tools/shared/KeyedLock";
25import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
26import * as Cause from "effect/Cause";
27import * as Context from "effect/Context";
28import * as DateTime from "effect/DateTime";
29import * as Effect from "effect/Effect";
30import * as Layer from "effect/Layer";
31import * as Option from "effect/Option";
32import * as Schema from "effect/Schema";
33import * as SqlClient from "effect/sql/SqlClient";
34
35import * as EventSink from "../EventSink.ts";
36import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid";
37
38const IMPORT_EVENT_PREFIX = "migration:v1";
39const TRANSCRIPT_EVENT_BATCH_SIZE = 100;
40
41interface LegacyThreadRow {
42 readonly thread_id: string;
43 readonly project_id: string;
44 readonly title: string;
45 readonly model_selection_json: string | null;
46 readonly runtime_mode: string;
47 readonly interaction_mode: string;
48 readonly branch: string | null;
49 readonly worktree_path: string | null;
50 readonly created_at: string;
51 readonly updated_at: string;
52 readonly archived_at: string | null;
53 readonly settled_override: string | null;
54 readonly settled_at: string | null;
55 readonly unsettled_at: string | null;
56 readonly snoozed_until: string | null;
57 readonly snoozed_at: string | null;
58 readonly pinned_at: string | null;
59 readonly auto_settle_disabled_at: string | null;
60 readonly pin_order_key: string | null;
61 readonly pull_requests_json: string;
62 readonly linked_pull_request_json: string | null;
63 readonly branch_pull_request_json: string | null;
64 readonly active_order_key: string | null;
65 readonly deleted_at: string | null;
66}
67
68interface LegacyRepairRow extends LegacyThreadRow {
69 readonly payload_json: string;
70}
71
72interface LegacyMessageRow {
73 readonly message_id: string;
74 readonly thread_id: string;
75 readonly role: "user" | "assistant";
76 readonly text: string;
77 readonly attachments_json: string | null;
78 readonly context_json?: string | null;
79 readonly is_streaming: number;
80 readonly created_at: string;
81 readonly updated_at: string;
82 readonly ordinal: number;
83}
84
85interface LegacyImportRow {
86 readonly thread_id: string;
87 readonly transcript_imported_at: string | null;
88}
89
90export interface LegacyV1ImportSummary {
91 readonly importedThreadCount: number;
92 readonly importedMessageCount: number;
93}
94
95export class LegacyV1ThreadImportError extends Schema.TaggedError<LegacyV1ThreadImportError>()(
96 "LegacyV1ThreadImportError",
97 {
98 operation: Schema.String,
99 threadId: Schema.optional(ThreadId),
100 cause: Schema.optional(Schema.Defect()),
101 },
102) {
103 override get message(): string {
104 return this.threadId === undefined
105 ? `Failed to ${this.operation} legacy v1 threads.`
106 : `Failed to ${this.operation} legacy v1 thread ${this.threadId}.`;
107 }
108}
109
110export interface LegacyV1ThreadImporterShape {
111 readonly pendingThreadCount: Effect.Effect<number, LegacyV1ThreadImportError>;
112 readonly reconcileShells: Effect.Effect<LegacyV1ImportSummary, LegacyV1ThreadImportError>;
113 readonly ensureTranscript: (
114 threadId: ThreadId,
115 ) => Effect.Effect<LegacyV1ImportSummary, LegacyV1ThreadImportError>;
116 readonly importPendingTranscripts: Effect.Effect<LegacyV1ImportSummary, never>;
117}
118
119export class LegacyV1ThreadImporter extends Context.Service<
120 LegacyV1ThreadImporter,
121 LegacyV1ThreadImporterShape
122>()("t3/orchestration-v2/legacy/LegacyV1ThreadImporter") {}
123
124const decodeModelSelection = Schema.decodeUnknownOption(ModelSelection);
125const decodeAttachments = Schema.decodeUnknownOption(Schema.Array(ChatAttachment));
126const decodePullRequests = Schema.decodeUnknownOption(Schema.Array(ThreadPullRequestLink));
127const decodeLinkedPullRequest = Schema.decodeUnknownOption(ThreadLinkedPullRequest);
128const decodeStoredThread = Schema.decodeUnknownOption(
129 Schema.fromJsonString(OrchestrationV2AppThreadJson),
130);
131const decodeMessageContext = Schema.decodeUnknownOption(OrchestrationMessageContext);
132
133function parseJson(json: string): unknown {
134 try {
135 return JSON.parse(json);
136 } catch {
137 return undefined;
138 }
139}
140
141function modelSelectionFor(row: LegacyThreadRow) {
142 const decoded =
143 row.model_selection_json === null
144 ? Option.none()
145 : decodeModelSelection(parseJson(row.model_selection_json));
146 return Option.getOrElse(decoded, () => ({
147 instanceId: ProviderInstanceId.make("codex"),
148 model: DEFAULT_MODEL,
149 }));
150}
151
152function attachmentsFor(row: LegacyMessageRow) {
153 if (row.attachments_json === null) return [];
154 return Option.getOrElse(decodeAttachments(parseJson(row.attachments_json)), () => []);
155}
156
157function contextFor(row: LegacyMessageRow) {
158 if (!row.context_json) return undefined;
159 return Option.getOrUndefined(decodeMessageContext(parseJson(row.context_json)));
160}
161
162function linkedPullRequestFor(row: LegacyThreadRow) {
163 if (row.linked_pull_request_json === null) return null;
164 return Option.getOrNull(decodeLinkedPullRequest(parseJson(row.linked_pull_request_json)));
165}
166
167function branchPullRequestFor(row: LegacyThreadRow) {
168 if (row.branch_pull_request_json === null) return null;
169 return Option.getOrNull(decodeLinkedPullRequest(parseJson(row.branch_pull_request_json)));
170}
171
172function runtimeModeFor(value: string): OrchestrationV2AppThread["runtimeMode"] {
173 return value === "approval-required" ||
174 value === "auto-accept-edits" ||
175 value === "auto" ||
176 value === "full-access"
177 ? value
178 : "full-access";
179}
180
181function interactionModeFor(value: string): OrchestrationV2AppThread["interactionMode"] {
182 return value === "plan" ? "plan" : "default";
183}
184
185function settledOverrideFor(value: string | null): OrchestrationV2AppThread["settledOverride"] {
186 return value === "settled" || value === "active" ? value : null;
187}
188
189// V1 rows can hold unparseable timestamps; fall back to another time rather than fail the import.
190function dateTimeOr(value: string | null, fallback: DateTime.Utc): DateTime.Utc {
191 return value === null ? fallback : Option.getOrElse(DateTime.make(value), () => fallback);
192}
193
194// A malformed marker stays set so archived or deleted threads keep that state.
195function nullableDateTime(value: string | null, fallback: DateTime.Utc): DateTime.Utc | null {
196 return value === null ? null : dateTimeOr(value, fallback);
197}
198
199function importedThread(row: LegacyThreadRow, importedAt: DateTime.Utc): OrchestrationV2AppThread {
200 const createdAt = dateTimeOr(row.created_at, dateTimeOr(row.updated_at, importedAt));
201 const updatedAt = dateTimeOr(row.updated_at, createdAt);
202 const threadId = ThreadId.make(row.thread_id);
203 const modelSelection = modelSelectionFor(row);
204 const branch = row.branch?.trim() || null;
205 const worktreePath = row.worktree_path?.trim() || null;
206 const pullRequests = Option.getOrElse(
207 decodePullRequests(parseJson(row.pull_requests_json)),
208 () => [],
209 );
210 const linkedPullRequest = linkedPullRequestFor(row);
211 const legacyLink = threadPullRequestsOf({ linkedPullRequest })[0];
212 const importedPullRequests =
213 legacyLink !== undefined &&
214 !pullRequests.some((link) => threadPullRequestKeysEqual(link, legacyLink))
215 ? [...pullRequests, legacyLink]
216 : pullRequests;
217 return {
218 createdBy: "system",
219 creationSource: "server",
220 id: threadId,
221 projectId: ProjectId.make(row.project_id),
222 title: row.title.trim() === "" ? "Untitled thread" : row.title,
223 providerInstanceId: modelSelection.instanceId,
224 modelSelection,
225 runtimeMode: runtimeModeFor(row.runtime_mode),
226 interactionMode: interactionModeFor(row.interaction_mode),
227 branch,
228 worktreePath,
229 linkedPullRequest,
230 pullRequests: importedPullRequests,
231 branchPullRequest: branchPullRequestFor(row),
232 activeOrderKey: row.active_order_key?.trim() || null,
233 activeProviderThreadId: null,
234 historyOrigin: "v1_import",
235 lineage: {
236 parentThreadId: null,
237 relationshipToParent: null,
238 rootThreadId: threadId,
239 },
240 forkedFrom: null,
241 createdAt,
242 updatedAt,
243 archivedAt: nullableDateTime(row.archived_at, updatedAt),
244 settledOverride: settledOverrideFor(row.settled_override),
245 settledAt: nullableDateTime(row.settled_at, updatedAt),
246 unsettledAt: nullableDateTime(row.unsettled_at, updatedAt),
247 snoozedUntil: nullableDateTime(row.snoozed_until, updatedAt),
248 snoozedAt: nullableDateTime(row.snoozed_at, updatedAt),
249 pinnedAt: nullableDateTime(row.pinned_at, updatedAt),
250 autoSettleDisabledAt: nullableDateTime(row.auto_settle_disabled_at, updatedAt),
251 pinOrderKey: row.pin_order_key?.trim() || null,
252 lastVisitedAt: null,
253 deletedAt: nullableDateTime(row.deleted_at, updatedAt),
254 };
255}
256
257function messageEvents(
258 row: LegacyMessageRow,
259 importedAt: DateTime.Utc,
260): ReadonlyArray<OrchestrationV2DomainEvent> {
261 const threadId = ThreadId.make(row.thread_id);
262 const messageId = MessageId.make(row.message_id);
263 const createdAt = dateTimeOr(row.created_at, dateTimeOr(row.updated_at, importedAt));
264 const updatedAt = dateTimeOr(row.updated_at, createdAt);
265 const attachments = attachmentsFor(row);
266 const context = contextFor(row);
267 const message: OrchestrationV2ConversationMessage = {
268 createdBy: row.role === "user" ? "user" : "agent",
269 creationSource: "server",
270 id: messageId,
271 threadId,
272 runId: null,
273 nodeId: null,
274 role: row.role,
275 text: row.text,
276 ...(context !== undefined ? { context } : {}),
277 attachments,
278 streaming: false,
279 createdAt,
280 updatedAt,
281 };
282 const baseTurnItem = {
283 id: TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${row.message_id}`),
284 threadId,
285 runId: null,
286 nodeId: null,
287 providerThreadId: null,
288 providerTurnId: null,
289 nativeItemRef: null,
290 parentItemId: null,
291 ordinal: row.ordinal,
292 status: row.is_streaming === 1 ? ("interrupted" as const) : ("completed" as const),
293 title: null,
294 startedAt: createdAt,
295 completedAt: updatedAt,
296 updatedAt,
297 };
298 const turnItem: OrchestrationV2TurnItem =
299 row.role === "user"
300 ? {
301 ...baseTurnItem,
302 createdBy: "user",
303 creationSource: "server",
304 type: "user_message",
305 messageId,
306 inputIntent: "turn_start",
307 text: row.text,
308 ...(context !== undefined ? { context } : {}),
309 attachments,
310 }
311 : {
312 ...baseTurnItem,
313 type: "assistant_message",
314 messageId,
315 text: row.text,
316 ...(context !== undefined ? { context } : {}),
317 streaming: false,
318 };
319 return [
320 {
321 id: EventId.make(`${IMPORT_EVENT_PREFIX}:message:${row.message_id}`),
322 type: "message.updated",
323 threadId,
324 occurredAt: updatedAt,
325 payload: message,
326 },
327 {
328 id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${row.message_id}`),
329 type: "turn-item.updated",
330 threadId,
331 occurredAt: updatedAt,
332 payload: turnItem,
333 },
334 ];
335}
336
337function chunks<A>(items: ReadonlyArray<A>, size: number): Array<ReadonlyArray<A>> {
338 const result: Array<ReadonlyArray<A>> = [];
339 for (let index = 0; index < items.length; index += size) {
340 result.push(items.slice(index, index + size));
341 }
342 return result;
343}
344
345const make = Effect.gen(function* () {
346 const sql = yield* SqlClient.SqlClient;
347 const eventSink = yield* EventSink.EventSinkV2;
348 const transcriptImports = yield* KeyedLock.make<ThreadId>();
349
350 const listMessages = (threadId: ThreadId) =>
351 sql<LegacyMessageRow>`
352 SELECT
353 message_id,
354 thread_id,
355 role,
356 text,
357 attachments_json,
358 context_json,
359 is_streaming,
360 created_at,
361 updated_at,
362 ROW_NUMBER() OVER (
363 PARTITION BY thread_id
364 ORDER BY created_at ASC, message_id ASC
365 ) AS ordinal
366 FROM projection_thread_messages
367 WHERE thread_id = ${threadId}
368 AND role IN ('user', 'assistant')
369 ORDER BY created_at ASC, message_id ASC
370 `.pipe(NodeSqliteClient.readOnly);
371
372 const listShellMessages = Effect.fn("LegacyV1ThreadImporter.listShellMessages")(function* (
373 threadId: ThreadId,
374 ) {
375 const latest = yield* sql<LegacyMessageRow>`
376 SELECT
377 message.message_id,
378 message.thread_id,
379 message.role,
380 message.text,
381 message.attachments_json,
382 message.context_json,
383 message.is_streaming,
384 message.created_at,
385 message.updated_at,
386 (
387 SELECT COUNT(*)
388 FROM projection_thread_messages AS earlier
389 WHERE earlier.thread_id = message.thread_id
390 AND earlier.role IN ('user', 'assistant')
391 AND (
392 earlier.created_at < message.created_at
393 OR (
394 earlier.created_at = message.created_at
395 AND earlier.message_id <= message.message_id
396 )
397 )
398 ) AS ordinal
399 FROM projection_thread_messages AS message
400 WHERE message.thread_id = ${threadId}
401 AND message.role IN ('user', 'assistant')
402 ORDER BY message.created_at DESC, message.message_id DESC
403 LIMIT 1
404 `.pipe(NodeSqliteClient.readOnly);
405 const latestUser = yield* sql<LegacyMessageRow>`
406 SELECT
407 message.message_id,
408 message.thread_id,
409 message.role,
410 message.text,
411 message.attachments_json,
412 message.context_json,
413 message.is_streaming,
414 message.created_at,
415 message.updated_at,
416 (
417 SELECT COUNT(*)
418 FROM projection_thread_messages AS earlier
419 WHERE earlier.thread_id = message.thread_id
420 AND earlier.role IN ('user', 'assistant')
421 AND (
422 earlier.created_at < message.created_at
423 OR (
424 earlier.created_at = message.created_at
425 AND earlier.message_id <= message.message_id
426 )
427 )
428 ) AS ordinal
429 FROM projection_thread_messages AS message
430 WHERE message.thread_id = ${threadId}
431 AND message.role = 'user'
432 ORDER BY message.created_at DESC, message.message_id DESC
433 LIMIT 1
434 `.pipe(NodeSqliteClient.readOnly);
435 return [latestUser[0], latest[0]].filter(
436 (message, index, selected): message is LegacyMessageRow =>
437 message !== undefined &&
438 selected.findIndex((candidate) => candidate?.message_id === message.message_id) === index,
439 );
440 });
441
442 const reconcileShellsBase = Effect.gen(function* () {
443 const now = yield* DateTime.now;
444 const repairRows = yield* sql<LegacyRepairRow>`
445 SELECT
446 thread.thread_id,
447 thread.project_id,
448 thread.title,
449 thread.model_selection_json,
450 thread.runtime_mode,
451 thread.interaction_mode,
452 thread.branch,
453 thread.worktree_path,
454 thread.created_at,
455 thread.updated_at,
456 thread.archived_at,
457 thread.settled_override,
458 thread.settled_at,
459 thread.unsettled_at,
460 thread.snoozed_until,
461 thread.snoozed_at,
462 thread.pinned_at,
463 thread.auto_settle_disabled_at,
464 thread.pin_order_key,
465 (SELECT json_group_array(json_object('host', pr.host, 'repository', pr.repository, 'number', pr.number, 'url', pr.url, 'source', pr.source, 'linkedAt', pr.linked_at, 'snapshot', json(pr.snapshot_json), 'stack', json(pr.stack_json))) FROM projection_thread_pull_requests pr WHERE pr.thread_id = thread.thread_id) AS pull_requests_json,
466 thread.linked_pull_request_json,
467 thread.branch_pull_request_json,
468 thread.active_order_key,
469 thread.deleted_at,
470 projection.payload_json
471 FROM orchestration_v2_legacy_imports AS legacy_import
472 INNER JOIN projection_threads AS thread
473 ON thread.thread_id = legacy_import.thread_id
474 INNER JOIN orchestration_v2_projection_threads AS projection
475 ON projection.thread_id = legacy_import.thread_id
476 WHERE json_type(projection.payload_json, '$.pinnedAt') IS NULL
477 OR json_type(projection.payload_json, '$.pinOrderKey') IS NULL
478 OR json_type(projection.payload_json, '$.snoozedUntil') IS NULL
479 OR json_type(projection.payload_json, '$.snoozedAt') IS NULL
480 OR json_type(projection.payload_json, '$.unsettledAt') IS NULL
481 OR json_type(projection.payload_json, '$.linkedPullRequest') IS NULL
482 OR json_type(projection.payload_json, '$.pullRequests') IS NULL
483 OR json_type(projection.payload_json, '$.branchPullRequest') IS NULL
484 OR json_type(projection.payload_json, '$.activeOrderKey') IS NULL
485 ORDER BY thread.created_at ASC, thread.thread_id ASC
486 `.pipe(NodeSqliteClient.readOnly);
487 let repairedThreadCount = 0;
488 for (const row of repairRows) {
489 const decoded = decodeStoredThread(row.payload_json);
490 if (Option.isNone(decoded)) continue;
491 const current = decoded.value;
492 const legacy = importedThread(row, now);
493 const legacyPullRequests = legacy.pullRequests ?? [];
494 const repaired: OrchestrationV2AppThread = {
495 ...current,
496 pinnedAt: current.pinnedAt === undefined ? legacy.pinnedAt : current.pinnedAt,
497 autoSettleDisabledAt:
498 current.autoSettleDisabledAt === undefined
499 ? legacy.autoSettleDisabledAt
500 : current.autoSettleDisabledAt,
501 pinOrderKey: current.pinOrderKey === undefined ? legacy.pinOrderKey : current.pinOrderKey,
502 snoozedUntil:
503 current.snoozedUntil === undefined ? legacy.snoozedUntil : current.snoozedUntil,
504 snoozedAt: current.snoozedAt === undefined ? legacy.snoozedAt : current.snoozedAt,
505 unsettledAt: current.unsettledAt === undefined ? legacy.unsettledAt : current.unsettledAt,
506 linkedPullRequest:
507 current.linkedPullRequest === undefined
508 ? legacy.linkedPullRequest
509 : current.linkedPullRequest,
510 pullRequests:
511 current.pullRequests === undefined
512 ? current.linkedPullRequest === null
513 ? []
514 : legacyPullRequests.length > 0
515 ? legacyPullRequests
516 : threadPullRequestsOf({
517 linkedPullRequest:
518 current.linkedPullRequest === undefined
519 ? legacy.linkedPullRequest
520 : current.linkedPullRequest,
521 })
522 : current.pullRequests,
523 branchPullRequest:
524 current.branchPullRequest === undefined
525 ? legacy.branchPullRequest
526 : current.branchPullRequest,
527 activeOrderKey:
528 current.activeOrderKey === undefined ? legacy.activeOrderKey : current.activeOrderKey,
529 };
530 // Later schema additions can require another repair for the same thread.
531 const repairId = yield* randomUuidV4;
532 yield* eventSink.write({
533 events: [
534 {
535 id: EventId.make(
536 `${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:metadata-repair:${repairId}`,
537 ),
538 type: "thread.metadata-updated",
539 threadId: repaired.id,
540 providerInstanceId: repaired.providerInstanceId,
541 occurredAt: now,
542 payload: repaired,
543 },
544 ],
545 });
546 repairedThreadCount += 1;
547 }
548 const rows = yield* sql<LegacyThreadRow>`
549 SELECT
550 thread.thread_id,
551 thread.project_id,
552 thread.title,
553 thread.model_selection_json,
554 thread.runtime_mode,
555 thread.interaction_mode,
556 thread.branch,
557 thread.worktree_path,
558 thread.created_at,
559 thread.updated_at,
560 thread.archived_at,
561 thread.settled_override,
562 thread.settled_at,
563 thread.unsettled_at,
564 thread.snoozed_until,
565 thread.snoozed_at,
566 thread.pinned_at,
567 thread.auto_settle_disabled_at,
568 thread.pin_order_key,
569 (SELECT json_group_array(json_object('host', pr.host, 'repository', pr.repository, 'number', pr.number, 'url', pr.url, 'source', pr.source, 'linkedAt', pr.linked_at, 'snapshot', json(pr.snapshot_json), 'stack', json(pr.stack_json))) FROM projection_thread_pull_requests pr WHERE pr.thread_id = thread.thread_id) AS pull_requests_json,
570 thread.linked_pull_request_json,
571 thread.branch_pull_request_json,
572 thread.active_order_key,
573 thread.deleted_at
574 FROM projection_threads AS thread
575 WHERE NOT EXISTS (
576 SELECT 1
577 FROM orchestration_events AS event INDEXED BY orchestration_events_v2_created_threads_idx
578 WHERE event.application_event_version = 2
579 AND event.aggregate_kind = 'thread'
580 AND event.stream_id = thread.thread_id
581 AND event.event_type = 'thread.created'
582 )
583 ORDER BY thread.created_at ASC, thread.thread_id ASC
584 `.pipe(NodeSqliteClient.readOnly);
585 let importedThreadCount = repairedThreadCount;
586 let importedMessageCount = 0;
587 for (const row of rows) {
588 const thread = importedThread(row, now);
589 const previews = yield* listShellMessages(thread.id);
590 const events: Array<OrchestrationV2DomainEvent> = [
591 {
592 id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:created`),
593 type: "thread.created",
594 threadId: thread.id,
595 providerInstanceId: thread.providerInstanceId,
596 occurredAt: thread.createdAt,
597 payload: thread,
598 },
599 ...previews.flatMap((message) => messageEvents(message, now)),
600 {
601 id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:shell`),
602 type: "thread.metadata-updated",
603 threadId: thread.id,
604 providerInstanceId: thread.providerInstanceId,
605 occurredAt: thread.updatedAt,
606 payload: thread,
607 },
608 ];
609 yield* sql.withTransaction(
610 Effect.gen(function* () {
611 yield* Effect.forEach(
612 previews,
613 (message) =>
614 sql`
615 INSERT INTO orchestration_v2_turn_item_positions (
616 thread_id,
617 turn_item_id,
618 ordinal
619 )
620 VALUES (
621 ${thread.id},
622 ${TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${message.message_id}`)},
623 ${message.ordinal}
624 )
625 ON CONFLICT(thread_id, turn_item_id) DO NOTHING
626 `,
627 { discard: true },
628 );
629 yield* eventSink.write({ events });
630 yield* sql`
631 INSERT INTO orchestration_v2_legacy_imports (
632 thread_id,
633 source_updated_at,
634 shell_imported_at,
635 transcript_imported_at,
636 imported_message_count,
637 last_error
638 )
639 VALUES (
640 ${thread.id},
641 ${row.updated_at},
642 ${DateTime.formatIso(now)},
643 NULL,
644 ${previews.length},
645 NULL
646 )
647 ON CONFLICT(thread_id) DO NOTHING
648 `;
649 }),
650 );
651 importedThreadCount += 1;
652 importedMessageCount += previews.length;
653 }
654 return { importedThreadCount, importedMessageCount };
655 });
656
657 const reconcileShells = reconcileShellsBase.pipe(
658 Effect.mapError((cause) => new LegacyV1ThreadImportError({ operation: "import", cause })),
659 );
660
661 const pendingThreadCount = sql<{ readonly count: number }>`
662 SELECT COUNT(*) AS count
663 FROM (
664 SELECT thread.thread_id
665 FROM projection_threads AS thread
666 WHERE NOT EXISTS (
667 SELECT 1
668 FROM orchestration_events AS event INDEXED BY orchestration_events_v2_created_threads_idx
669 WHERE event.application_event_version = 2
670 AND event.aggregate_kind = 'thread'
671 AND event.stream_id = thread.thread_id
672 AND event.event_type = 'thread.created'
673 )
674 UNION
675 SELECT legacy_import.thread_id
676 FROM orchestration_v2_legacy_imports AS legacy_import
677 WHERE legacy_import.transcript_imported_at IS NULL
678 )
679 `.pipe(
680 NodeSqliteClient.readOnly,
681 Effect.map((rows) => rows[0]?.count ?? 0),
682 Effect.mapError(
683 (cause) => new LegacyV1ThreadImportError({ operation: "inspect pending", cause }),
684 ),
685 );
686
687 // Both results are final for this process: transcript_imported_at never resets, and only startup shell reconciliation creates legacy rows.
688 const confirmedTranscriptThreadIds = new Set<ThreadId>();
689
690 const ensureTranscriptBase = (threadId: ThreadId) =>
691 transcriptImports.withLock(
692 threadId,
693 Effect.gen(function* () {
694 const imports = yield* sql<LegacyImportRow>`
695 SELECT thread_id, transcript_imported_at
696 FROM orchestration_v2_legacy_imports
697 WHERE thread_id = ${threadId}
698 LIMIT 1
699 `.pipe(NodeSqliteClient.readOnly);
700 const imported = imports[0];
701 if (imported === undefined || imported.transcript_imported_at !== null) {
702 confirmedTranscriptThreadIds.add(threadId);
703 return { importedThreadCount: 0, importedMessageCount: 0 };
704 }
705 const importedAt = yield* DateTime.now;
706 const messages = yield* listMessages(threadId);
707 // Hint the index to skip V1 events without scanning their payloads.
708 const existingRows = yield* sql<{ readonly event_id: string }>`
709 SELECT event_id
710 FROM orchestration_events INDEXED BY idx_orchestration_events_agent_stream_sequence
711 WHERE application_event_version = 2
712 AND aggregate_kind = 'thread'
713 AND stream_id = ${threadId}
714 AND event_id LIKE ${`${IMPORT_EVENT_PREFIX}:message:%`}
715 `.pipe(NodeSqliteClient.readOnly);
716 const existing = new Set(existingRows.map((row) => row.event_id));
717 const missing = messages.filter(
718 (message) => !existing.has(`${IMPORT_EVENT_PREFIX}:message:${message.message_id}`),
719 );
720 for (const batch of chunks(missing, TRANSCRIPT_EVENT_BATCH_SIZE / 2)) {
721 yield* Effect.forEach(
722 batch,
723 (message) =>
724 sql`
725 INSERT INTO orchestration_v2_turn_item_positions (
726 thread_id,
727 turn_item_id,
728 ordinal
729 )
730 VALUES (
731 ${threadId},
732 ${TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${message.message_id}`)},
733 ${message.ordinal}
734 )
735 ON CONFLICT(thread_id, turn_item_id) DO NOTHING
736 `,
737 { discard: true },
738 );
739 yield* eventSink.write({
740 events: batch.flatMap((message) => messageEvents(message, importedAt)),
741 });
742 yield* Effect.yieldNow;
743 }
744 const now = DateTime.formatIso(yield* DateTime.now);
745 yield* sql`
746 UPDATE orchestration_v2_legacy_imports
747 SET
748 transcript_imported_at = ${now},
749 imported_message_count = ${messages.length},
750 last_error = NULL
751 WHERE thread_id = ${threadId}
752 `;
753 confirmedTranscriptThreadIds.add(threadId);
754 return {
755 importedThreadCount: 1,
756 importedMessageCount: missing.length,
757 };
758 }),
759 );
760
761 const ensureTranscript = (threadId: ThreadId) =>
762 confirmedTranscriptThreadIds.has(threadId)
763 ? Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 })
764 : ensureTranscriptBase(threadId).pipe(
765 Effect.mapError(
766 (cause) =>
767 new LegacyV1ThreadImportError({
768 operation: "hydrate transcript for",
769 threadId,
770 cause,
771 }),
772 ),
773 );
774
775 const importPendingTranscripts = Effect.gen(function* () {
776 const rows = yield* sql<LegacyImportRow>`
777 SELECT thread_id, transcript_imported_at
778 FROM orchestration_v2_legacy_imports
779 WHERE transcript_imported_at IS NULL
780 ORDER BY shell_imported_at ASC, thread_id ASC
781 `.pipe(NodeSqliteClient.readOnly);
782 let importedThreadCount = 0;
783 let importedMessageCount = 0;
784 for (const row of rows) {
785 const result = yield* ensureTranscript(ThreadId.make(row.thread_id)).pipe(
786 Effect.catchCauseIf(
787 (cause) => !Cause.hasInterrupts(cause),
788 (cause) =>
789 Effect.logWarning("Failed to hydrate migrated v1 thread transcript", {
790 threadId: row.thread_id,
791 cause,
792 }).pipe(
793 Effect.andThen(() =>
794 sql`
795 UPDATE orchestration_v2_legacy_imports
796 SET last_error = 'Transcript hydration failed; retry on next open.'
797 WHERE thread_id = ${row.thread_id}
798 `.pipe(
799 Effect.catchCauseIf(
800 (cause) => !Cause.hasInterrupts(cause),
801 () => Effect.void,
802 ),
803 ),
804 ),
805 Effect.as({ importedThreadCount: 0, importedMessageCount: 0 }),
806 ),
807 ),
808 );
809 importedThreadCount += result.importedThreadCount;
810 importedMessageCount += result.importedMessageCount;
811 yield* Effect.yieldNow;
812 }
813 return { importedThreadCount, importedMessageCount };
814 }).pipe(
815 Effect.catchCause((cause) =>
816 Cause.hasInterrupts(cause)
817 ? Effect.failCause(Cause.fromReasons<never>(cause.reasons.filter(Cause.isInterruptReason)))
818 : Effect.logWarning("Legacy v1 transcript background import stopped", { cause }).pipe(
819 Effect.as({ importedThreadCount: 0, importedMessageCount: 0 }),
820 ),
821 ),
822 );
823
824 return LegacyV1ThreadImporter.of({
825 pendingThreadCount,
826 reconcileShells,
827 ensureTranscript,
828 importPendingTranscripts,
829 });
830});
831
832export const layer: Layer.Layer<
833 LegacyV1ThreadImporter,
834 never,
835 EventSink.EventSinkV2 | SqlClient.SqlClient
836> = Layer.effect(LegacyV1ThreadImporter, make);