fix(server,web): make projection rebuilds restore full history

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -2911,6 +2911,200 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
     }),
   );
 
+  it.effect("retains checkpointless turns and their messages across a revert", () =>
+    Effect.gen(function* () {
+      const projectionPipeline = yield* OrchestrationProjectionPipeline;
+      const eventStore = yield* OrchestrationEventStore;
+      const sql = yield* SqlClient.SqlClient;
+      const threadId = ThreadId.make("thread-revert-turns");
+      const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
+        eventStore
+          .append(event)
+          .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
+      const sessionSet = (
+        suffix: string,
+        occurredAt: string,
+        status: "running" | "ready",
+        activeTurnId: string | null,
+      ) =>
+        appendAndProject({
+          type: "thread.session-set",
+          eventId: EventId.make(`evt-rt-${suffix}`),
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          occurredAt,
+          commandId: CommandId.make(`cmd-rt-${suffix}`),
+          causationEventId: null,
+          correlationId: CorrelationId.make(`cmd-rt-${suffix}`),
+          metadata: {},
+          payload: {
+            threadId,
+            session: {
+              threadId,
+              status,
+              providerName: "claude",
+              runtimeMode: "full-access",
+              activeTurnId: activeTurnId === null ? null : TurnId.make(activeTurnId),
+              lastError: null,
+              updatedAt: occurredAt,
+            },
+          },
+        });
+      const messageSent = (suffix: string, occurredAt: string, messageId: string, turnId: string) =>
+        appendAndProject({
+          type: "thread.message-sent",
+          eventId: EventId.make(`evt-rt-${suffix}`),
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          occurredAt,
+          commandId: CommandId.make(`cmd-rt-${suffix}`),
+          causationEventId: null,
+          correlationId: CorrelationId.make(`cmd-rt-${suffix}`),
+          metadata: {},
+          payload: {
+            threadId,
+            messageId: MessageId.make(messageId),
+            role: "assistant" as const,
+            text: messageId,
+            turnId: TurnId.make(turnId),
+            streaming: false,
+            createdAt: occurredAt,
+            updatedAt: occurredAt,
+          },
+        });
+      const diffCompleted = (
+        suffix: string,
+        occurredAt: string,
+        turnId: string,
+        checkpointTurnCount: number,
+        assistantMessageId: string,
+      ) =>
+        appendAndProject({
+          type: "thread.turn-diff-completed",
+          eventId: EventId.make(`evt-rt-${suffix}`),
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          occurredAt,
+          commandId: CommandId.make(`cmd-rt-${suffix}`),
+          causationEventId: null,
+          correlationId: CorrelationId.make(`cmd-rt-${suffix}`),
+          metadata: {},
+          payload: {
+            threadId,
+            turnId: TurnId.make(turnId),
+            checkpointTurnCount,
+            checkpointRef: CheckpointRef.make(
+              `refs/t3/checkpoints/thread-revert-turns/turn/${checkpointTurnCount}`,
+            ),
+            status: "ready" as const,
+            files: [],
+            assistantMessageId: MessageId.make(assistantMessageId),
+            completedAt: occurredAt,
+          },
+        });
+
+      yield* appendAndProject({
+        type: "project.created",
+        eventId: EventId.make("evt-rt-1"),
+        aggregateKind: "project",
+        aggregateId: ProjectId.make("project-revert-turns"),
+        occurredAt: "2026-02-26T13:00:00.000Z",
+        commandId: CommandId.make("cmd-rt-1"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-rt-1"),
+        metadata: {},
+        payload: {
+          projectId: ProjectId.make("project-revert-turns"),
+          title: "Project Revert Turns",
+          workspaceRoot: "/tmp/project-revert-turns",
+          defaultModelSelection: null,
+          scripts: [],
+          createdAt: "2026-02-26T13:00:00.000Z",
+          updatedAt: "2026-02-26T13:00:00.000Z",
+        },
+      });
+      yield* appendAndProject({
+        type: "thread.created",
+        eventId: EventId.make("evt-rt-2"),
+        aggregateKind: "thread",
+        aggregateId: threadId,
+        occurredAt: "2026-02-26T13:00:01.000Z",
+        commandId: CommandId.make("cmd-rt-2"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-rt-2"),
+        metadata: {},
+        payload: {
+          threadId,
+          projectId: ProjectId.make("project-revert-turns"),
+          title: "Thread Revert Turns",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          runtimeMode: "full-access",
+          branch: null,
+          worktreePath: null,
+          createdAt: "2026-02-26T13:00:01.000Z",
+          updatedAt: "2026-02-26T13:00:01.000Z",
+        },
+      });
+
+      // A turn recorded while checkpointing was unavailable: it settles
+      // without ever getting a checkpoint_turn_count.
+      yield* sessionSet("3", "2026-02-26T13:00:02.000Z", "running", "turn-early");
+      yield* messageSent("4", "2026-02-26T13:00:03.000Z", "assistant-early", "turn-early");
+      yield* sessionSet("5", "2026-02-26T13:00:04.000Z", "ready", null);
+
+      // A checkpointed turn at the revert baseline, then one past it.
+      yield* sessionSet("6", "2026-02-26T13:00:05.000Z", "running", "turn-mid");
+      yield* diffCompleted("7", "2026-02-26T13:00:06.000Z", "turn-mid", 1, "assistant-mid");
+      yield* messageSent("8", "2026-02-26T13:00:06.500Z", "assistant-mid", "turn-mid");
+      yield* sessionSet("9", "2026-02-26T13:00:07.000Z", "ready", null);
+      yield* sessionSet("10", "2026-02-26T13:00:08.000Z", "running", "turn-late");
+      yield* diffCompleted("11", "2026-02-26T13:00:09.000Z", "turn-late", 2, "assistant-late");
+      yield* messageSent("12", "2026-02-26T13:00:09.500Z", "assistant-late", "turn-late");
+      yield* sessionSet("13", "2026-02-26T13:00:10.000Z", "ready", null);
+
+      yield* appendAndProject({
+        type: "thread.reverted",
+        eventId: EventId.make("evt-rt-14"),
+        aggregateKind: "thread",
+        aggregateId: threadId,
+        occurredAt: "2026-02-26T13:00:11.000Z",
+        commandId: CommandId.make("cmd-rt-14"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-rt-14"),
+        metadata: {},
+        payload: {
+          threadId,
+          turnCount: 1,
+        },
+      });
+
+      // The post-baseline checkpointed turn is reverted; the checkpointless
+      // turn survives so its message stays reachable to turn-anchored
+      // pagination.
+      const turnRows = yield* sql<{ readonly turnId: string | null }>`
+        SELECT turn_id AS "turnId"
+        FROM projection_turns
+        WHERE thread_id = ${threadId}
+        ORDER BY turn_id ASC
+      `;
+      assert.deepEqual(turnRows, [{ turnId: "turn-early" }, { turnId: "turn-mid" }]);
+
+      const messageRowsAfterRevert = yield* sql<{ readonly messageId: string }>`
+        SELECT message_id AS "messageId"
+        FROM projection_thread_messages
+        WHERE thread_id = ${threadId}
+        ORDER BY message_id ASC
+      `;
+      assert.deepEqual(messageRowsAfterRevert, [
+        { messageId: "assistant-early" },
+        { messageId: "assistant-mid" },
+      ]);
+    }),
+  );
+
   it.effect("excludes a completed retraction message from SQLite projection", () =>
     Effect.gen(function* () {
       const projectionPipeline = yield* OrchestrationProjectionPipeline;

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -1466,11 +1466,17 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           const existingTurns = yield* projectionTurnRepository.listByThreadId({
             threadId: event.payload.threadId,
           });
+          // Same denylist as the message/activity/plan projectors: turns
+          // without checkpoint evidence (e.g. recorded while checkpointing
+          // was unavailable) must survive a revert, or their retained
+          // messages become unreachable to turn-anchored pagination.
+          const revertedTurnIds = yield* getRevertedTurnIds({
+            threadId: event.payload.threadId,
+            baselineTurnCount: event.payload.turnCount,
+            retractionTurnId: event.payload.retraction?.turnId ?? null,
+          });
           const keptTurns = existingTurns.filter(
-            (turn) =>
-              turn.turnId !== null &&
-              turn.checkpointTurnCount !== null &&
-              turn.checkpointTurnCount <= event.payload.turnCount,
+            (turn) => turn.turnId !== null && !revertedTurnIds.has(turn.turnId),
           );
           yield* projectionTurnRepository.deleteByThreadId({
             threadId: event.payload.threadId,
@@ -1858,7 +1864,11 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
       // Match live event-major ordering so a revert sees turn/session evidence
       // before later projectors apply that same event's trims.
-      yield* Stream.runForEach(eventStore.readFromSequence(firstSequence), (event) =>
+      // The full backlog, not the default 1k page: a projection rebuild (migration
+      // 045) resets the cursor to 0 and needs every event replayed.
+      yield* Stream.runForEach(
+        eventStore.readFromSequence(firstSequence, Number.MAX_SAFE_INTEGER),
+        (event) =>
         Effect.forEach(
           projectors,
           (projector) =>

apps/server/src/persistence/Migrations.ts

@@ -58,6 +58,7 @@ import Migration0042 from "./Migrations/042_ProjectionTurnDispatchOwnership.ts";
 import Migration0043 from "./Migrations/043_ProjectionManagedWorktrees.ts";
 import Migration0044 from "./Migrations/044_CleanupCompletedRetractionMessages.ts";
 import Migration0045 from "./Migrations/045_RebuildProjectionsFromEvents.ts";
+import Migration0046 from "./Migrations/046_RebuildProjectionsWithRetainedTurns.ts";
 
 /**
  * Migration loader with all migrations defined inline.
@@ -115,6 +116,7 @@ export const migrationEntries = [
   [43, "ProjectionManagedWorktrees", Migration0043],
   [44, "CleanupCompletedRetractionMessages", Migration0044],
   [45, "RebuildProjectionsFromEvents", Migration0045],
+  [46, "RebuildProjectionsWithRetainedTurns", Migration0046],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts

@@ -54,7 +54,9 @@ layer("045_RebuildProjectionsFromEvents", (it) => {
 
       yield* sql`DROP TABLE projection_thread_proposed_plans`;
       yield* rebuildProjectionsFromEvents;
-      assert.deepEqual(migrationManifest.at(-1), [45, "RebuildProjectionsFromEvents"]);
+      // The manifest must end on a full rebuild so boot replays every event
+      // through the current projectors after the wipe.
+      assert.deepEqual(migrationManifest.at(-1), [46, "RebuildProjectionsWithRetainedTurns"]);
     }),
   );
 });

apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts

@@ -0,0 +1,9 @@
+import Migration0045 from "./045_RebuildProjectionsFromEvents.ts";
+
+// Re-run migration 045's wipe-and-replay. The first rebuild ran through a
+// turns projector whose thread.reverted handler kept only checkpointed turns,
+// deleting checkpointless turns and stranding their messages outside
+// turn-anchored pagination. The projector now shares the revert denylist with
+// the other projectors, so replaying the event history again rebuilds those
+// turn rows.
+export default Migration0045;

apps/web/src/connection/storage.ts

@@ -63,8 +63,10 @@ const StoredShellSnapshotJson = Schema.fromJsonString(StoredShellSnapshot);
 // a warm cache resuming via `afterSequence` would render the ghost messages
 // forever. Any server-side row surgery needs a bump here to reach clients.
 // v6 pairs with server migration 045's full projection rebuild.
+// v7 pairs with server migration 046, which rebuilds again now that the
+// turns projector retains checkpointless turns across replayed reverts.
 const StoredThreadSnapshot = Schema.Struct({
-  schemaVersion: Schema.Literal(6),
+  schemaVersion: Schema.Literal(7),
   environmentId: EnvironmentId,
   threadId: ThreadId,
   snapshot: OrchestrationThreadDetailSnapshot,
@@ -572,7 +574,7 @@ export const connectionStorageLayer = Layer.effectContext(
       saveThread: (environmentId, snapshot) =>
         Effect.gen(function* () {
           const encoded = yield* encodeStoredThreadSnapshot({
-            schemaVersion: 6,
+            schemaVersion: 7,
             environmentId,
             threadId: snapshot.thread.id,
             snapshot,