From 569a5a33858ce77a193ea5504a0ebb86d38e6b3e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 12:29:11 +0000 Subject: [PATCH 1/4] test(server): reproduce #30813 malformed legacy timestamps aborting V1 import These tests reproduce #30813: a legacy V1 thread row with a malformed updated_at/archived_at, or a legacy message with a malformed created_at, makes reconcileShells die with IllegalArgumentError: Invalid date, so the server never reaches readiness. AI-assisted; verified with the tests and tools listed above. --- .../legacy/LegacyV1ThreadImporter.test.ts | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts index a50b3ed3c1910..d6630aaccb1cb 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts @@ -219,6 +219,79 @@ it.effect("propagates interruption from transcript reads and retry-marker writes }).pipe(Effect.provide(layerTest)), ); +const malformedLegacyTimestamp = "2026-07-13T14:13:00.3NZ"; + +const seedLegacyThread = (threadId: ThreadId) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + 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 + ) VALUES ( + ${threadId}, 'project:malformed', 'Legacy thread', + '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', 'default', + '2026-07-13T14:00:00.000Z', '2026-07-13T14:00:00.000Z' + ) + `; + }); + +it.effect("imports a legacy thread whose updated_at and archived_at are malformed", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make("thread:malformed-archived"); + yield* seedLegacyThread(threadId); + yield* sql` + UPDATE projection_threads + SET updated_at = ${malformedLegacyTimestamp}, archived_at = ${malformedLegacyTimestamp} + WHERE thread_id = ${threadId} + `; + + const exit = yield* Effect.exit(importer.reconcileShells); + + assert.deepStrictEqual(exit, Exit.succeed({ importedThreadCount: 1, importedMessageCount: 0 })); + const projection = yield* projections.getThreadProjection(threadId); + assert.isNotNull(projection.thread.archivedAt, "malformed archived_at resurrected the thread"); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("imports a legacy message whose created_at is malformed", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make("thread:malformed-message"); + yield* seedLegacyThread(threadId); + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, role, text, is_streaming, created_at, updated_at + ) VALUES ( + 'message:malformed', ${threadId}, 'user', 'Question', 0, + ${malformedLegacyTimestamp}, '2026-07-13T14:13:00.000Z' + ) + `; + + const exit = yield* Effect.exit(importer.reconcileShells); + + assert.deepStrictEqual(exit, Exit.succeed({ importedThreadCount: 1, importedMessageCount: 1 })); + const projection = yield* projections.getThreadProjection(threadId); + 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; From d361c022651ed84b4137163565231e3349524459 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 12:34:50 +0000 Subject: [PATCH 2/4] fix(server): malformed legacy timestamps no longer block startup A V1 row with an unparseable timestamp (seen in the wild as 2026-07-13T14:13:00.3NZ) made DateTime.makeUnsafe throw inside reconcileShells. That defect escaped the error mapping, and the importer retried the same rows on every start, so the server never became ready. Timestamps now parse with DateTime.make and fall back to a valid time from the same row: thread createdAt falls back to updated_at and then to the import time, updatedAt to createdAt, and a set but malformed nullable marker (archived_at, deleted_at, ...) to updatedAt, so it stays set. Message createdAt and updatedAt fall back to each other and then to the import time. Startup shell import, metadata repair and lazy transcript hydration all share these helpers. The defect-isolation test for transcript hydration used a malformed created_at to inject its defect. It now injects one through Statement.CurrentTransformer. Fixes pingdotgg/t3code#30813 AI-assisted; verified with the tests and tools listed above. --- .../legacy/LegacyV1ThreadImporter.test.ts | 14 +++-- .../legacy/LegacyV1ThreadImporter.ts | 63 +++++++++++-------- 2 files changed, 45 insertions(+), 32 deletions(-) diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts index d6630aaccb1cb..778233f344e1e 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( diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts index 8c134d3b57d71..9fae4eca19cc6 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 now = 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,14 +736,15 @@ const make = Effect.gen(function* () { `, { discard: true }, ); - yield* eventSink.write({ events: batch.flatMap(messageEvents) }); + yield* eventSink.write({ + events: batch.flatMap((message) => messageEvents(message, now)), + }); yield* Effect.yieldNow; } - const now = DateTime.formatIso(yield* DateTime.now); yield* sql` UPDATE orchestration_v2_legacy_imports SET - transcript_imported_at = ${now}, + transcript_imported_at = ${DateTime.formatIso(now)}, imported_message_count = ${messages.length}, last_error = NULL WHERE thread_id = ${threadId} From dad4ea336ec3d8c549cdecd176ee9cbf918ae6c7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 12:44:54 +0000 Subject: [PATCH 3/4] fix(server): keep transcript import marker at completion time AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts index 9fae4eca19cc6..ede38bdd39685 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts @@ -702,7 +702,7 @@ const make = Effect.gen(function* () { confirmedTranscriptThreadIds.add(threadId); return { importedThreadCount: 0, importedMessageCount: 0 }; } - const now = yield* DateTime.now; + 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 }>` @@ -737,14 +737,15 @@ const make = Effect.gen(function* () { { discard: true }, ); yield* eventSink.write({ - events: batch.flatMap((message) => messageEvents(message, now)), + events: batch.flatMap((message) => messageEvents(message, importedAt)), }); yield* Effect.yieldNow; } + const now = DateTime.formatIso(yield* DateTime.now); yield* sql` UPDATE orchestration_v2_legacy_imports SET - transcript_imported_at = ${DateTime.formatIso(now)}, + transcript_imported_at = ${now}, imported_message_count = ${messages.length}, last_error = NULL WHERE thread_id = ${threadId} From 04a000cae086ea1bbb14e0756a455d9a05cb2f85 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 00:07:42 +0000 Subject: [PATCH 4/4] test(server): one regression case for malformed legacy timestamps AI-assisted; verified with the tests and tools listed above. --- .../legacy/LegacyV1ThreadImporter.test.ts | 58 +++++-------------- 1 file changed, 16 insertions(+), 42 deletions(-) diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts index 778233f344e1e..4cea500f6e7b8 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts @@ -223,11 +223,13 @@ it.effect("propagates interruption from transcript reads and retry-marker writes }).pipe(Effect.provide(layerTest)), ); -const malformedLegacyTimestamp = "2026-07-13T14:13:00.3NZ"; - -const seedLegacyThread = (threadId: ThreadId) => +// 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 @@ -239,56 +241,28 @@ const seedLegacyThread = (threadId: ThreadId) => yield* sql` INSERT INTO projection_threads ( thread_id, project_id, title, model_selection_json, runtime_mode, - interaction_mode, created_at, updated_at + interaction_mode, created_at, updated_at, archived_at ) VALUES ( - ${threadId}, 'project:malformed', 'Legacy thread', + 'thread:malformed', 'project:malformed', 'Legacy thread', '{"instanceId":"codex","model":"gpt-5.4"}', 'full-access', 'default', - '2026-07-13T14:00:00.000Z', '2026-07-13T14:00:00.000Z' + '2026-07-13T14:00:00.000Z', ${malformed}, ${malformed} ) `; - }); - -it.effect("imports a legacy thread whose updated_at and archived_at are malformed", () => - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const threadId = ThreadId.make("thread:malformed-archived"); - yield* seedLegacyThread(threadId); - yield* sql` - UPDATE projection_threads - SET updated_at = ${malformedLegacyTimestamp}, archived_at = ${malformedLegacyTimestamp} - WHERE thread_id = ${threadId} - `; - - const exit = yield* Effect.exit(importer.reconcileShells); - - assert.deepStrictEqual(exit, Exit.succeed({ importedThreadCount: 1, importedMessageCount: 0 })); - const projection = yield* projections.getThreadProjection(threadId); - assert.isNotNull(projection.thread.archivedAt, "malformed archived_at resurrected the thread"); - }).pipe(Effect.provide(layerTest)), -); - -it.effect("imports a legacy message whose created_at is malformed", () => - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const threadId = ThreadId.make("thread:malformed-message"); - yield* seedLegacyThread(threadId); yield* sql` INSERT INTO projection_thread_messages ( message_id, thread_id, role, text, is_streaming, created_at, updated_at ) VALUES ( - 'message:malformed', ${threadId}, 'user', 'Question', 0, - ${malformedLegacyTimestamp}, '2026-07-13T14:13:00.000Z' + 'message:malformed', 'thread:malformed', 'user', 'Question', 0, + ${malformed}, '2026-07-13T14:13:00.000Z' ) `; - const exit = yield* Effect.exit(importer.reconcileShells); - - assert.deepStrictEqual(exit, Exit.succeed({ importedThreadCount: 1, importedMessageCount: 1 })); - const projection = yield* projections.getThreadProjection(threadId); + 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"],