fix(server): preserve historical latest turn

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

@@ -4,6 +4,8 @@ import {
   CorrelationId,
   EventId,
   MessageId,
+  type OrchestrationEvent,
+  type OrchestrationSessionStatus,
   ProjectId,
   ThreadId,
   TurnId,
@@ -33,6 +35,7 @@ import {
 import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
 import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
 import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
+import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
 import { ServerConfig } from "../../config.ts";
 
 const makeProjectionPipelinePrefixedTestLayer = (prefix: string) =>
@@ -43,6 +46,89 @@ const makeProjectionPipelinePrefixedTestLayer = (prefix: string) =>
     Layer.provideMerge(NodeServices.layer),
   );
 
+const makeProjectionPipelineSnapshotTestLayer = (prefix: string) =>
+  Layer.merge(OrchestrationProjectionPipelineLive, OrchestrationProjectionSnapshotQueryLive).pipe(
+    Layer.provideMerge(OrchestrationEventStoreLive),
+    Layer.provideMerge(RepositoryIdentityResolver.layer),
+    Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix })),
+    Layer.provideMerge(SqlitePersistenceMemory),
+    Layer.provideMerge(NodeServices.layer),
+  );
+
+type ThreadCreatedEvent = Omit<
+  Extract<OrchestrationEvent, { readonly type: "thread.created" }>,
+  "sequence"
+>;
+
+const makeThreadCreatedEvent = (options: {
+  readonly eventId: string;
+  readonly threadId: ThreadId;
+  readonly projectId: ProjectId;
+  readonly title: string;
+  readonly createdAt: string;
+}): ThreadCreatedEvent => ({
+  type: "thread.created",
+  eventId: EventId.make(options.eventId),
+  aggregateKind: "thread",
+  aggregateId: options.threadId,
+  occurredAt: options.createdAt,
+  commandId: CommandId.make(`cmd-${options.eventId}`),
+  causationEventId: null,
+  correlationId: CorrelationId.make(`cmd-${options.eventId}`),
+  metadata: {},
+  payload: {
+    threadId: options.threadId,
+    projectId: options.projectId,
+    title: options.title,
+    modelSelection: {
+      instanceId: ProviderInstanceId.make("codex"),
+      model: "gpt-5-codex",
+    },
+    runtimeMode: "full-access",
+    interactionMode: "default",
+    branch: null,
+    worktreePath: null,
+    createdAt: options.createdAt,
+    updatedAt: options.createdAt,
+  },
+});
+
+type ThreadSessionSetEvent = Omit<
+  Extract<OrchestrationEvent, { readonly type: "thread.session-set" }>,
+  "sequence"
+>;
+
+const makeThreadSessionSetEvent = (options: {
+  readonly eventId: string;
+  readonly threadId: ThreadId;
+  readonly status: OrchestrationSessionStatus;
+  readonly activeTurnId: TurnId | null;
+  readonly updatedAt: string;
+  readonly lastError?: string | null;
+}): ThreadSessionSetEvent => ({
+  type: "thread.session-set",
+  eventId: EventId.make(options.eventId),
+  aggregateKind: "thread",
+  aggregateId: options.threadId,
+  occurredAt: options.updatedAt,
+  commandId: CommandId.make(`cmd-${options.eventId}`),
+  causationEventId: null,
+  correlationId: CorrelationId.make(`cmd-${options.eventId}`),
+  metadata: {},
+  payload: {
+    threadId: options.threadId,
+    session: {
+      threadId: options.threadId,
+      status: options.status,
+      providerName: "codex",
+      runtimeMode: "full-access",
+      activeTurnId: options.activeTurnId,
+      lastError: options.lastError ?? null,
+      updatedAt: options.updatedAt,
+    },
+  },
+});
+
 const exists = (filePath: string) =>
   Effect.gen(function* () {
     const fileSystem = yield* FileSystem.FileSystem;
@@ -387,6 +473,208 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-atta
   },
 );
 
+it.layer(Layer.fresh(makeProjectionPipelineSnapshotTestLayer("t3-latest-turn-session-set-")))(
+  "OrchestrationProjectionPipeline latest turn",
+  (it) => {
+    it.effect("preserves the completed latest turn when the session becomes ready", () =>
+      Effect.gen(function* () {
+        const projectionPipeline = yield* OrchestrationProjectionPipeline;
+        const snapshotQuery = yield* ProjectionSnapshotQuery;
+        const eventStore = yield* OrchestrationEventStore;
+        const sql = yield* SqlClient.SqlClient;
+        const threadId = ThreadId.make("thread-latest-turn-ready");
+        const turnId = TurnId.make("turn-latest-turn-ready");
+        const createdAt = "2026-07-24T00:00:00.000Z";
+        const startedAt = "2026-07-24T00:00:01.000Z";
+        const completedAt = "2026-07-24T00:01:00.000Z";
+
+        yield* eventStore.append(
+          makeThreadCreatedEvent({
+            eventId: "evt-latest-turn-created",
+            threadId,
+            projectId: ProjectId.make("project-latest-turn"),
+            title: "Latest turn",
+            createdAt,
+          }),
+        );
+        yield* eventStore.append(
+          makeThreadSessionSetEvent({
+            eventId: "evt-latest-turn-running",
+            threadId,
+            status: "running",
+            activeTurnId: turnId,
+            updatedAt: startedAt,
+          }),
+        );
+        yield* eventStore.append(
+          makeThreadSessionSetEvent({
+            eventId: "evt-latest-turn-ready",
+            threadId,
+            status: "ready",
+            activeTurnId: null,
+            updatedAt: completedAt,
+          }),
+        );
+
+        yield* projectionPipeline.bootstrap;
+
+        const rows = yield* sql<{
+          readonly activeTurnId: string | null;
+          readonly latestTurnId: string | null;
+        }>`
+        SELECT
+          sessions.active_turn_id AS "activeTurnId",
+          threads.latest_turn_id AS "latestTurnId"
+        FROM projection_threads threads
+        JOIN projection_thread_sessions sessions
+          ON sessions.thread_id = threads.thread_id
+        WHERE threads.thread_id = ${threadId}
+      `;
+        assert.deepEqual(rows, [{ activeTurnId: null, latestTurnId: turnId }]);
+
+        const shell = yield* snapshotQuery.getThreadShellById(threadId);
+        assert.equal(shell._tag, "Some");
+        if (shell._tag === "Some") {
+          assert.equal(shell.value.session?.activeTurnId, null);
+          assert.equal(shell.value.latestTurn?.turnId, turnId);
+          assert.equal(shell.value.latestTurn?.state, "completed");
+          assert.equal(shell.value.latestTurn?.completedAt, completedAt);
+        }
+      }),
+    );
+
+    it.effect("preserves the latest turn when session statuses have no active turn", () =>
+      Effect.gen(function* () {
+        const projectionPipeline = yield* OrchestrationProjectionPipeline;
+        const eventStore = yield* OrchestrationEventStore;
+        const sql = yield* SqlClient.SqlClient;
+        const threadId = ThreadId.make("thread-latest-turn-statuses");
+        const turnId = TurnId.make("turn-latest-turn-statuses");
+        const statuses = ["idle", "stopped", "interrupted", "error", "starting"] as const;
+
+        yield* eventStore.append(
+          makeThreadCreatedEvent({
+            eventId: "evt-latest-turn-statuses-created",
+            threadId,
+            projectId: ProjectId.make("project-latest-turn-statuses"),
+            title: "Latest turn statuses",
+            createdAt: "2026-07-24T01:00:00.000Z",
+          }),
+        );
+        yield* eventStore.append(
+          makeThreadSessionSetEvent({
+            eventId: "evt-latest-turn-statuses-running",
+            threadId,
+            status: "running",
+            activeTurnId: turnId,
+            updatedAt: "2026-07-24T01:01:00.000Z",
+          }),
+        );
+        yield* projectionPipeline.bootstrap;
+
+        for (const [index, status] of statuses.entries()) {
+          yield* eventStore.append(
+            makeThreadSessionSetEvent({
+              eventId: `evt-latest-turn-statuses-${status}`,
+              threadId,
+              status,
+              activeTurnId: null,
+              updatedAt: `2026-07-24T01:02:0${index}.000Z`,
+              lastError: status === "error" ? "turn failed" : null,
+            }),
+          );
+          yield* projectionPipeline.bootstrap;
+
+          const rows = yield* sql<{
+            readonly activeTurnId: string | null;
+            readonly latestTurnId: string | null;
+            readonly status: string;
+          }>`
+            SELECT
+              sessions.active_turn_id AS "activeTurnId",
+              threads.latest_turn_id AS "latestTurnId",
+              sessions.status
+            FROM projection_threads threads
+            JOIN projection_thread_sessions sessions
+              ON sessions.thread_id = threads.thread_id
+            WHERE threads.thread_id = ${threadId}
+          `;
+          assert.deepEqual(rows, [{ activeTurnId: null, latestTurnId: turnId, status }]);
+        }
+      }),
+    );
+
+    it.effect("keeps no prior turn null and replaces the pointer for new active turns", () =>
+      Effect.gen(function* () {
+        const projectionPipeline = yield* OrchestrationProjectionPipeline;
+        const eventStore = yield* OrchestrationEventStore;
+        const sql = yield* SqlClient.SqlClient;
+        const threadId = ThreadId.make("thread-latest-turn-replacement");
+        const firstTurnId = TurnId.make("turn-latest-turn-first");
+        const secondTurnId = TurnId.make("turn-latest-turn-second");
+        const createdAt = "2026-07-24T02:00:00.000Z";
+
+        yield* eventStore.append(
+          makeThreadCreatedEvent({
+            eventId: "evt-latest-turn-replacement-created",
+            threadId,
+            projectId: ProjectId.make("project-latest-turn-replacement"),
+            title: "Latest turn replacement",
+            createdAt,
+          }),
+        );
+        yield* eventStore.append(
+          makeThreadSessionSetEvent({
+            eventId: "evt-latest-turn-replacement-starting",
+            threadId,
+            status: "starting",
+            activeTurnId: null,
+            updatedAt: "2026-07-24T02:00:01.000Z",
+          }),
+        );
+        yield* projectionPipeline.bootstrap;
+
+        const noTurnRows = yield* sql<{ readonly latestTurnId: string | null }>`
+        SELECT latest_turn_id AS "latestTurnId"
+        FROM projection_threads
+        WHERE thread_id = ${threadId}
+      `;
+        assert.deepEqual(noTurnRows, [{ latestTurnId: null }]);
+
+        for (const [index, turnId] of [firstTurnId, secondTurnId].entries()) {
+          const updatedAt = `2026-07-24T02:00:0${index + 2}.000Z`;
+          yield* eventStore.append(
+            makeThreadSessionSetEvent({
+              eventId: `evt-latest-turn-replacement-running-${index}`,
+              threadId,
+              status: "running",
+              activeTurnId: turnId,
+              updatedAt,
+            }),
+          );
+        }
+        yield* projectionPipeline.bootstrap;
+
+        const replacementRows = yield* sql<{
+          readonly activeTurnId: string | null;
+          readonly latestTurnId: string | null;
+        }>`
+        SELECT
+          sessions.active_turn_id AS "activeTurnId",
+          threads.latest_turn_id AS "latestTurnId"
+        FROM projection_threads threads
+        JOIN projection_thread_sessions sessions
+          ON sessions.thread_id = threads.thread_id
+        WHERE threads.thread_id = ${threadId}
+      `;
+        assert.deepEqual(replacementRows, [
+          { activeTurnId: secondTurnId, latestTurnId: secondTurnId },
+        ]);
+      }),
+    );
+  },
+);
+
 it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
   it.effect(
     "passes explicit empty attachment arrays through the projection pipeline to clear attachments",

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

@@ -809,7 +809,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           }
           yield* projectionThreadRepository.upsert({
             ...existingRow.value,
-            latestTurnId: event.payload.session.activeTurnId,
+            latestTurnId: event.payload.session.activeTurnId ?? existingRow.value.latestTurnId,
             updatedAt: event.occurredAt,
           });
           yield* refreshThreadShellSummary(event.payload.threadId);