fix(server): malformed legacy timestamps no longer block startup

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;

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<OrchestrationV2DomainEvent> {
+function messageEvents(
+  row: LegacyMessageRow,
+  importedAt: DateTime.Utc,
+): ReadonlyArray<OrchestrationV2DomainEvent> {
   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<LegacyRepairRow>`
       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<OrchestrationV2DomainEvent> = [
         {
@@ -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);