diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts index a50b3ed3c1910..4cea500f6e7b8 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts @@ -155,12 +155,16 @@ it.effect( const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; const projections = yield* ProjectionStore.ProjectionStoreV2; yield* seedPendingTranscripts; - yield* sql` - UPDATE projection_thread_messages SET created_at = '0000-not-a-date' - WHERE message_id = 'thread:first:old' - `; - yield* importer.importPendingTranscripts; + yield* importer.importPendingTranscripts.pipe( + Effect.provideService(Statement.CurrentTransformer, (statement) => { + const [query, params] = statement.compile(); + return query.includes("FROM projection_thread_messages") && + params.includes("thread:first") + ? Effect.die("transcript read failed") + : Effect.succeed(statement); + }), + ); const later = yield* projections.getThreadProjection(ThreadId.make("thread:later")); assert.deepStrictEqual( @@ -219,6 +223,53 @@ it.effect("propagates interruption from transcript reads and retry-marker writes }).pipe(Effect.provide(layerTest)), ); +// regression: #30813 a malformed V1 timestamp stopped the server from starting +it.effect("imports legacy rows with malformed timestamps and keeps the archive marker", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const malformed = "2026-07-13T14:13:00.3NZ"; + yield* sql` + INSERT INTO projection_projects ( + project_id, title, workspace_root, scripts_json, created_at, updated_at + ) VALUES ( + 'project:malformed', 'Legacy project', '/tmp/legacy-malformed', '[]', + '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z' + ) + `; + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, + interaction_mode, created_at, updated_at, archived_at + ) VALUES ( + 'thread:malformed', 'project:malformed', 'Legacy thread', + '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', 'default', + '2026-07-13T14:00:00.000Z', ${malformed}, ${malformed} + ) + `; + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, role, text, is_streaming, created_at, updated_at + ) VALUES ( + 'message:malformed', 'thread:malformed', 'user', 'Question', 0, + ${malformed}, '2026-07-13T14:13:00.000Z' + ) + `; + + assert.deepStrictEqual(yield* importer.reconcileShells, { + importedThreadCount: 1, + importedMessageCount: 1, + }); + const projection = yield* projections.getThreadProjection(ThreadId.make("thread:malformed")); + assert.isNotNull(projection.thread.archivedAt); + assert.deepStrictEqual( + projection.messages.map((message) => message.text), + ["Question"], + ); + }).pipe(Effect.provide(layerTest)), +); + it.effect("caches the absent legacy transcript for a native V2 thread", () => Effect.gen(function* () { const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts index 8c134d3b57d71..ede38bdd39685 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts @@ -186,15 +186,19 @@ function settledOverrideFor(value: string | null): OrchestrationV2AppThread["set return value === "settled" || value === "active" ? value : null; } -function dateTime(value: string): DateTime.Utc { - return DateTime.makeUnsafe(value); +// V1 rows can hold unparseable timestamps; fall back to another time rather than fail the import. +function dateTimeOr(value: string | null, fallback: DateTime.Utc): DateTime.Utc { + return value === null ? fallback : Option.getOrElse(DateTime.make(value), () => fallback); } -function nullableDateTime(value: string | null): DateTime.Utc | null { - return value === null ? null : dateTime(value); +// A malformed marker stays set so archived or deleted threads keep that state. +function nullableDateTime(value: string | null, fallback: DateTime.Utc): DateTime.Utc | null { + return value === null ? null : dateTimeOr(value, fallback); } -function importedThread(row: LegacyThreadRow): OrchestrationV2AppThread { +function importedThread(row: LegacyThreadRow, importedAt: DateTime.Utc): OrchestrationV2AppThread { + const createdAt = dateTimeOr(row.created_at, dateTimeOr(row.updated_at, importedAt)); + const updatedAt = dateTimeOr(row.updated_at, createdAt); const threadId = ThreadId.make(row.thread_id); const modelSelection = modelSelectionFor(row); const branch = row.branch?.trim() || null; @@ -234,27 +238,30 @@ function importedThread(row: LegacyThreadRow): OrchestrationV2AppThread { rootThreadId: threadId, }, forkedFrom: null, - createdAt: dateTime(row.created_at), - updatedAt: dateTime(row.updated_at), - archivedAt: nullableDateTime(row.archived_at), + createdAt, + updatedAt, + archivedAt: nullableDateTime(row.archived_at, updatedAt), settledOverride: settledOverrideFor(row.settled_override), - settledAt: nullableDateTime(row.settled_at), - unsettledAt: nullableDateTime(row.unsettled_at), - snoozedUntil: nullableDateTime(row.snoozed_until), - snoozedAt: nullableDateTime(row.snoozed_at), - pinnedAt: nullableDateTime(row.pinned_at), - autoSettleDisabledAt: nullableDateTime(row.auto_settle_disabled_at), + settledAt: nullableDateTime(row.settled_at, updatedAt), + unsettledAt: nullableDateTime(row.unsettled_at, updatedAt), + snoozedUntil: nullableDateTime(row.snoozed_until, updatedAt), + snoozedAt: nullableDateTime(row.snoozed_at, updatedAt), + pinnedAt: nullableDateTime(row.pinned_at, updatedAt), + autoSettleDisabledAt: nullableDateTime(row.auto_settle_disabled_at, updatedAt), pinOrderKey: row.pin_order_key?.trim() || null, lastVisitedAt: null, - deletedAt: nullableDateTime(row.deleted_at), + deletedAt: nullableDateTime(row.deleted_at, updatedAt), }; } -function messageEvents(row: LegacyMessageRow): ReadonlyArray { +function messageEvents( + row: LegacyMessageRow, + importedAt: DateTime.Utc, +): ReadonlyArray { const threadId = ThreadId.make(row.thread_id); const messageId = MessageId.make(row.message_id); - const createdAt = dateTime(row.created_at); - const updatedAt = dateTime(row.updated_at); + const createdAt = dateTimeOr(row.created_at, dateTimeOr(row.updated_at, importedAt)); + const updatedAt = dateTimeOr(row.updated_at, createdAt); const attachments = attachmentsFor(row); const context = contextFor(row); const message: OrchestrationV2ConversationMessage = { @@ -433,7 +440,7 @@ const make = Effect.gen(function* () { }); const reconcileShellsBase = Effect.gen(function* () { - const now = DateTime.formatIso(yield* DateTime.now); + const now = yield* DateTime.now; const repairRows = yield* sql` SELECT thread.thread_id, @@ -482,7 +489,7 @@ const make = Effect.gen(function* () { const decoded = decodeStoredThread(row.payload_json); if (Option.isNone(decoded)) continue; const current = decoded.value; - const legacy = importedThread(row); + const legacy = importedThread(row, now); const legacyPullRequests = legacy.pullRequests ?? []; const repaired: OrchestrationV2AppThread = { ...current, @@ -531,7 +538,7 @@ const make = Effect.gen(function* () { type: "thread.metadata-updated", threadId: repaired.id, providerInstanceId: repaired.providerInstanceId, - occurredAt: dateTime(now), + occurredAt: now, payload: repaired, }, ], @@ -578,7 +585,7 @@ const make = Effect.gen(function* () { let importedThreadCount = repairedThreadCount; let importedMessageCount = 0; for (const row of rows) { - const thread = importedThread(row); + const thread = importedThread(row, now); const previews = yield* listShellMessages(thread.id); const events: Array = [ { @@ -589,7 +596,7 @@ const make = Effect.gen(function* () { occurredAt: thread.createdAt, payload: thread, }, - ...previews.flatMap(messageEvents), + ...previews.flatMap((message) => messageEvents(message, now)), { id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:shell`), type: "thread.metadata-updated", @@ -632,7 +639,7 @@ const make = Effect.gen(function* () { VALUES ( ${thread.id}, ${row.updated_at}, - ${now}, + ${DateTime.formatIso(now)}, NULL, ${previews.length}, NULL @@ -695,6 +702,7 @@ const make = Effect.gen(function* () { confirmedTranscriptThreadIds.add(threadId); return { importedThreadCount: 0, importedMessageCount: 0 }; } + const importedAt = yield* DateTime.now; const messages = yield* listMessages(threadId); // Hint the index to skip V1 events without scanning their payloads. const existingRows = yield* sql<{ readonly event_id: string }>` @@ -728,7 +736,9 @@ const make = Effect.gen(function* () { `, { discard: true }, ); - yield* eventSink.write({ events: batch.flatMap(messageEvents) }); + yield* eventSink.write({ + events: batch.flatMap((message) => messageEvents(message, importedAt)), + }); yield* Effect.yieldNow; } const now = DateTime.formatIso(yield* DateTime.now);