feat(codex): continue provider threads across clients

apps/server/src/auth/RpcAuthorization.ts

@@ -32,6 +32,7 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.serverProbe]: AuthOrchestrationReadScope,
   [WS_METHODS.serverGetConfig]: AuthOrchestrationReadScope,
   [WS_METHODS.serverRefreshProviders]: AuthOrchestrationOperateScope,
+  [WS_METHODS.serverStartProviderThreadSync]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverUpdateProvider]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverUpdateServer]: AuthOrchestrationOperateScope,
   [WS_METHODS.serverUpdateServerWithProgress]: AuthOrchestrationOperateScope,
@@ -89,6 +90,7 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.providerUploadFeedback]: AuthOrchestrationOperateScope,
   [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope,
   [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope,
+  [WS_METHODS.subscribeProviderThreadSync]: AuthOrchestrationReadScope,
   [WS_METHODS.vcsRefreshStatus]: AuthOrchestrationReadScope,
   [WS_METHODS.vcsPull]: AuthOrchestrationOperateScope,
   [WS_METHODS.gitRunStackedAction]: AuthOrchestrationOperateScope,

apps/server/src/bin.test.ts

@@ -17,6 +17,7 @@ import { assert, it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as DateTime from "effect/DateTime";
 import * as Layer from "effect/Layer";
+import * as Stream from "effect/Stream";
 import * as HttpRouter from "effect/unstable/http/HttpRouter";
 import * as HttpServer from "effect/unstable/http/HttpServer";
 import * as HttpApi from "effect/unstable/httpapi/HttpApi";
@@ -31,6 +32,7 @@ import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSna
 import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
 import { OrchestrationLayerLive } from "./orchestration/runtimeLayer.ts";
 import { orchestrationHttpApiLayer } from "./orchestration/http.ts";
+import { ProviderThreadContinuity } from "./provider/Services/ProviderThreadContinuity.ts";
 import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/Sqlite.ts";
 import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts";
 import {
@@ -131,6 +133,18 @@ const withLiveProjectCliServer = <A, E, R>(baseDir: string, run: () => Effect.Ef
         ),
       ),
       Layer.provideMerge(makeProjectPersistenceLayer(config)),
+      Layer.provide(
+        Layer.succeed(
+          ProviderThreadContinuity,
+          ProviderThreadContinuity.of({
+            reconcileThread: () => Effect.succeed(false),
+            requestReconcileThread: () => Effect.void,
+            startSyncAll: Effect.succeed({ status: "idle" }),
+            getSyncStatus: Effect.succeed({ status: "idle" }),
+            streamSyncStatus: Stream.make({ status: "idle" }),
+          }),
+        ),
+      ),
       Layer.provideMerge(
         NodeHttpServer.layer(NodeHttp.createServer, {
           host: "127.0.0.1",

apps/server/src/environment/ServerEnvironment.ts

@@ -154,6 +154,7 @@ export const make = Effect.gen(function* () {
       threadPinReorder: true,
       threadTitleRegeneration: true,
       threadPullRequestLinking: true,
+      providerThreadSync: true,
       ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
       ...(serverSelfUpdate === "boot-service" ? { serverSelfUpdateProgress: true } : {}),
     },

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

@@ -0,0 +1,157 @@
+import {
+  CommandId,
+  CorrelationId,
+  EventId,
+  MessageId,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  TurnId,
+} from "@t3tools/contracts";
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+
+import { ServerConfig } from "../../config.ts";
+import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
+import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
+import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
+import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
+import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
+
+const TestLayer = OrchestrationProjectionPipelineLive.pipe(
+  Layer.provideMerge(OrchestrationEventStoreLive),
+  Layer.provideMerge(
+    ServerConfig.layerTest(process.cwd(), { prefix: "t3-projection-import-test-" }),
+  ),
+  Layer.provideMerge(SqlitePersistenceMemory),
+  Layer.provideMerge(NodeServices.layer),
+);
+
+it.layer(TestLayer)("historical provider message projection", (it) => {
+  it.effect("projects an imported turn as completed without runtime or checkpoint state", () =>
+    Effect.gen(function* () {
+      const pipeline = yield* OrchestrationProjectionPipeline;
+      const eventStore = yield* OrchestrationEventStore;
+      const sql = yield* SqlClient.SqlClient;
+      const projectId = ProjectId.make("project-import");
+      const threadId = ThreadId.make("thread-import");
+      const turnId = TurnId.make("provider-turn-import");
+      const createdAt = "2026-08-26T10:00:00.000Z";
+
+      yield* eventStore.append({
+        type: "project.created",
+        eventId: EventId.make("event-import-project"),
+        aggregateKind: "project",
+        aggregateId: projectId,
+        occurredAt: createdAt,
+        commandId: CommandId.make("command-import-project"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("command-import-project"),
+        metadata: {},
+        payload: {
+          projectId,
+          title: "Imported project",
+          workspaceRoot: "/tmp/imported-project",
+          defaultModelSelection: null,
+          scripts: [],
+          createdAt,
+          updatedAt: createdAt,
+        },
+      });
+      yield* eventStore.append({
+        type: "thread.created",
+        eventId: EventId.make("event-import-thread"),
+        aggregateKind: "thread",
+        aggregateId: threadId,
+        occurredAt: createdAt,
+        commandId: CommandId.make("command-import-thread"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("command-import-thread"),
+        metadata: {},
+        payload: {
+          threadId,
+          projectId,
+          title: "Imported Codex thread",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5.6-sol",
+          },
+          runtimeMode: "full-access",
+          interactionMode: "default",
+          branch: null,
+          worktreePath: null,
+          createdAt,
+          updatedAt: createdAt,
+        },
+      });
+
+      for (const [ordinal, role, text] of [
+        [1, "user", "External prompt"],
+        [2, "assistant", "External response"],
+      ] as const) {
+        const occurredAt = `2026-08-26T10:00:0${ordinal}.000Z`;
+        yield* eventStore.append({
+          type: "thread.message-sent",
+          eventId: EventId.make(`event-import-message-${ordinal}`),
+          aggregateKind: "thread",
+          aggregateId: threadId,
+          occurredAt,
+          commandId: CommandId.make(`provider-import:message-${ordinal}`),
+          causationEventId: null,
+          correlationId: CorrelationId.make(`provider-import:message-${ordinal}`),
+          metadata: {},
+          payload: {
+            threadId,
+            messageId: MessageId.make(`message-import-${ordinal}`),
+            role,
+            text,
+            turnId,
+            streaming: false,
+            historical: true,
+            createdAt: occurredAt,
+            updatedAt: occurredAt,
+          },
+        });
+      }
+
+      yield* pipeline.bootstrap;
+
+      const turns = yield* sql<{
+        readonly state: string;
+        readonly pendingMessageId: string | null;
+        readonly assistantMessageId: string | null;
+      }>`
+        SELECT
+          state,
+          pending_message_id AS "pendingMessageId",
+          assistant_message_id AS "assistantMessageId"
+        FROM projection_turns
+        WHERE thread_id = ${threadId} AND turn_id = ${turnId}
+      `;
+      assert.deepEqual(turns, [
+        {
+          state: "completed",
+          pendingMessageId: null,
+          assistantMessageId: "message-import-2",
+        },
+      ]);
+
+      const threads = yield* sql<{ readonly latestTurnId: string | null }>`
+        SELECT latest_turn_id AS "latestTurnId"
+        FROM projection_threads
+        WHERE thread_id = ${threadId}
+      `;
+      assert.deepEqual(threads, [{ latestTurnId: turnId }]);
+
+      const checkpoints = yield* sql<{ readonly count: number }>`
+        SELECT COUNT(*) AS count
+        FROM projection_turns
+        WHERE thread_id = ${threadId} AND checkpoint_turn_count IS NOT NULL
+      `;
+      assert.deepEqual(checkpoints, [{ count: 0 }]);
+    }),
+  );
+});

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

@@ -133,7 +133,9 @@ function isStalePendingApprovalFailureDetail(detail: string | null): boolean {
 // A full refresh loads all thread history, so skip events that cannot change the summary.
 function shouldRefreshThreadShellSummary(event: OrchestrationEvent): boolean {
   if (event.type === "thread.message-sent") {
-    return event.payload.role === "user";
+    // The turn projector refreshes historical messages after both message and
+    // turn rows exist. Refreshing here runs too early and clears latestTurnId.
+    return event.payload.historical !== true && event.payload.role === "user";
   }
 
   if (event.type !== "thread.activity-appended") {
@@ -809,6 +811,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           }
           yield* projectionThreadRepository.upsert({
             ...existingRow.value,
+            ...(event.payload.projectId !== undefined
+              ? { projectId: event.payload.projectId }
+              : {}),
             ...(event.payload.title !== undefined ? { title: event.payload.title } : {}),
             ...(event.payload.titleRegeneration !== undefined
               ? {
@@ -890,6 +895,12 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           }
           yield* projectionThreadRepository.upsert({
             ...existingRow.value,
+            latestTurnId:
+              event.type === "thread.message-sent" &&
+              event.payload.historical === true &&
+              event.payload.turnId !== null
+                ? event.payload.turnId
+                : existingRow.value.latestTurnId,
             updatedAt: event.occurredAt,
           });
           if (shouldRefreshThreadShellSummary(event)) {
@@ -1327,7 +1338,41 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
         }
 
         case "thread.message-sent": {
-          if (event.payload.turnId === null || event.payload.role !== "assistant") {
+          if (event.payload.turnId === null) {
+            return;
+          }
+          if (event.payload.role === "user") {
+            if (event.payload.historical !== true) {
+              return;
+            }
+            const existingTurn = yield* projectionTurnRepository.getByTurnId({
+              threadId: event.payload.threadId,
+              turnId: event.payload.turnId,
+            });
+            if (Option.isNone(existingTurn)) {
+              yield* projectionTurnRepository.upsertByTurnId({
+                turnId: event.payload.turnId,
+                threadId: event.payload.threadId,
+                // Imported user history is already settled; it is not a queued
+                // message waiting for ProviderCommandReactor to start.
+                pendingMessageId: null,
+                sourceProposedPlanThreadId: null,
+                sourceProposedPlanId: null,
+                assistantMessageId: null,
+                state: "completed",
+                requestedAt: event.payload.createdAt,
+                startedAt: event.payload.createdAt,
+                completedAt: event.payload.updatedAt,
+                checkpointTurnCount: null,
+                checkpointRef: null,
+                checkpointStatus: null,
+                checkpointFiles: [],
+              });
+            }
+            yield* refreshThreadShellSummary(event.payload.threadId);
+            return;
+          }
+          if (event.payload.role !== "assistant") {
             return;
           }
           // A completed assistant message only settles the turn once the
@@ -1364,6 +1409,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
               startedAt: existingTurn.value.startedAt ?? event.payload.createdAt,
               requestedAt: existingTurn.value.requestedAt ?? event.payload.createdAt,
             });
+            if (event.payload.historical === true) {
+              yield* refreshThreadShellSummary(event.payload.threadId);
+            }
             return;
           }
           yield* projectionTurnRepository.upsertByTurnId({
@@ -1382,6 +1430,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             checkpointStatus: null,
             checkpointFiles: [],
           });
+          if (event.payload.historical === true) {
+            yield* refreshThreadShellSummary(event.payload.threadId);
+          }
           return;
         }