feat(codex): sync persisted threads into T3

apps/server/src/orchestration/decider.titleRegeneration.test.ts

@@ -46,6 +46,48 @@ const readModel: OrchestrationReadModel = {
 };
 
 it.layer(NodeServices.layer)("title regeneration decider", (it) => {
+  it.effect("applies a title update while the expected title is still current", () =>
+    Effect.gen(function* () {
+      const result = yield* decideOrchestrationCommand({
+        command: {
+          type: "thread.meta.update",
+          commandId: CommandId.make("cmd-title-update-current"),
+          threadId: ThreadId.make("thread-1"),
+          title: "Provider title",
+          expectedTitle: "Manual title",
+        },
+        readModel,
+      });
+      const event = Array.isArray(result) ? result[0] : result;
+
+      expect(event.type).toBe("thread.meta-updated");
+      if (event.type === "thread.meta-updated") {
+        expect(event.payload.title).toBe("Provider title");
+      }
+    }),
+  );
+
+  it.effect("does not replace a title that changed after the caller sampled it", () =>
+    Effect.gen(function* () {
+      const result = yield* decideOrchestrationCommand({
+        command: {
+          type: "thread.meta.update",
+          commandId: CommandId.make("cmd-title-update-stale"),
+          threadId: ThreadId.make("thread-1"),
+          title: "Provider title",
+          expectedTitle: "Previous provider title",
+        },
+        readModel,
+      });
+      const event = Array.isArray(result) ? result[0] : result;
+
+      expect(event.type).toBe("thread.meta-updated");
+      if (event.type === "thread.meta-updated") {
+        expect(event.payload.title).toBeUndefined();
+      }
+    }),
+  );
+
   it.effect("preserves updatedAt for a stale completion", () =>
     Effect.gen(function* () {
       const result = yield* decideOrchestrationCommand({

apps/server/src/orchestration/decider.ts

@@ -805,6 +805,12 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
         thread.branch !== command.expectedBranch
           ? thread.branch
           : command.branch;
+      const title =
+        command.title !== undefined &&
+        command.expectedTitle !== undefined &&
+        thread.title !== command.expectedTitle
+          ? undefined
+          : command.title;
       const occurredAt = yield* nowIso;
       return {
         ...(yield* withEventBase({
@@ -816,7 +822,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
         type: "thread.meta-updated",
         payload: {
           threadId: command.threadId,
-          ...(command.title !== undefined ? { title: command.title } : {}),
+          ...(title !== undefined ? { title } : {}),
           ...(command.regenerateTitle === true
             ? {
                 regenerateTitle: true as const,
@@ -827,7 +833,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
                 },
               }
             : {}),
-          ...(command.title !== undefined && thread.titleRegeneration != null
+          ...(title !== undefined && thread.titleRegeneration != null
             ? { titleRegeneration: null }
             : {}),
           ...(command.modelSelection !== undefined

apps/server/src/provider/Layers/CodexAdapter.ts

@@ -28,11 +28,13 @@ import {
 } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 import * as Crypto from "effect/Crypto";
+import * as DateTime from "effect/DateTime";
 import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
 import * as Queue from "effect/Queue";
 import * as Schema from "effect/Schema";
+import * as Option from "effect/Option";
 import * as Scope from "effect/Scope";
 import * as Stream from "effect/Stream";
 import { ChildProcessSpawner } from "effect/unstable/process";
@@ -52,6 +54,7 @@ import {
   type ProviderAdapterError,
 } from "../Errors.ts";
 import { type CodexAdapterShape } from "../Services/CodexAdapter.ts";
+import type { ProviderThreadSummary } from "../Services/ProviderAdapter.ts";
 import { resolveAttachmentPath } from "../../attachmentStore.ts";
 import { ServerConfig } from "../../config.ts";
 import {
@@ -62,6 +65,8 @@ import {
   type CodexSessionRuntimeOptions,
   type CodexSessionRuntimeShape,
 } from "./CodexSessionRuntime.ts";
+import { listAllCodexThreads, makeCodexAppServerConnection } from "./CodexAppServerConnection.ts";
+import { buildCodexInitializeParams } from "./CodexProvider.ts";
 import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
 import { resolveCodexLaunchArgs } from "./codexLaunchArgs.ts";
 const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError);
@@ -149,6 +154,27 @@ function trimText(value: string | undefined | null): string | undefined {
   return trimmed && trimmed.length > 0 ? trimmed : undefined;
 }
 
+function codexTimestampToIso(seconds: number): string {
+  return DateTime.make(seconds * 1_000).pipe(
+    Option.map(DateTime.formatIso),
+    Option.getOrElse(() => DateTime.formatIso(DateTime.makeUnsafe(0))),
+  );
+}
+
+function toProviderThreadSummary(
+  thread: EffectCodexSchema.V2ThreadListResponse["data"][number],
+): ProviderThreadSummary {
+  return {
+    providerThreadId: thread.id,
+    cwd: thread.cwd,
+    title: trimText(thread.name),
+    preview: trimText(thread.preview),
+    branch: trimText(thread.gitInfo?.branch),
+    createdAt: codexTimestampToIso(thread.createdAt),
+    updatedAt: codexTimestampToIso(thread.updatedAt),
+  };
+}
+
 const FATAL_CODEX_STDERR_SNIPPETS = ["failed to connect to websocket"];
 
 function isFatalCodexProcessStderrMessage(message: string): boolean {
@@ -1950,6 +1976,34 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
       { concurrency: 1 },
     );
 
+  const listThreads: NonNullable<CodexAdapterShape["listThreads"]> = () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const { client } = yield* makeCodexAppServerConnection({
+          binaryPath: codexConfig.binaryPath,
+          cwd: process.cwd(),
+          launchArgs: resolveCodexLaunchArgs(codexConfig.launchArgs, options?.environment),
+          ...(options?.environment ? { environment: options.environment } : {}),
+          ...(codexConfig.homePath ? { homePath: codexConfig.homePath } : {}),
+        });
+        yield* client.request("initialize", buildCodexInitializeParams());
+        yield* client.notify("initialized", undefined);
+        const threads = yield* listAllCodexThreads(client);
+        return threads.filter((thread) => !thread.ephemeral).map(toProviderThreadSummary);
+      }).pipe(
+        Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner),
+        Effect.mapError(
+          (cause) =>
+            new ProviderAdapterRequestError({
+              provider: PROVIDER,
+              method: "thread/list",
+              detail: "Failed to list persisted Codex threads.",
+              cause,
+            }),
+        ),
+      ),
+    );
+
   const hasSession: CodexAdapterShape["hasSession"] = (threadId) =>
     Effect.succeed(Boolean(sessions.get(threadId) && !sessions.get(threadId)?.stopped));
 
@@ -1981,6 +2035,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
     respondToUserInput,
     stopSession,
     listSessions,
+    listThreads,
     hasSession,
     stopAll,
     get streamEvents() {

apps/server/src/provider/Layers/CodexAppServerConnection.test.ts

@@ -0,0 +1,93 @@
+// @effect-diagnostics nodeBuiltinImport:off
+import * as NodeAssert from "node:assert/strict";
+import { it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import type * as CodexRpc from "effect-codex-app-server/rpc";
+import type * as CodexSchema from "effect-codex-app-server/schema";
+
+import { listAllCodexThreads, type CodexThreadListClient } from "./CodexAppServerConnection.ts";
+
+type CodexThread = CodexSchema.V2ThreadListResponse["data"][number];
+type ThreadListParams = CodexRpc.ClientRequestParamsByMethod["thread/list"];
+
+function thread(id: string, updatedAt: number): CodexThread {
+  return {
+    cliVersion: "1.0.0",
+    createdAt: updatedAt - 10,
+    cwd: "/workspace/project",
+    ephemeral: false,
+    id,
+    modelProvider: "openai",
+    preview: `Preview ${id}`,
+    sessionId: `session-${id}`,
+    source: "cli",
+    status: { type: "notLoaded" },
+    turns: [],
+    updatedAt,
+  };
+}
+
+it.effect("lists every Codex thread/list page newest-first and de-duplicates page overlaps", () =>
+  Effect.gen(function* () {
+    const requests: ThreadListParams[] = [];
+    const first = thread("thread-1", 200);
+    const second = thread("thread-2", 100);
+    const client: CodexThreadListClient = {
+      request: (_method, params) =>
+        Effect.sync(() => {
+          requests.push(params);
+          return params.cursor === undefined
+            ? { data: [first], nextCursor: "cursor-1" }
+            : { data: [first, second], nextCursor: null };
+        }),
+    };
+
+    const result = yield* listAllCodexThreads(client);
+
+    NodeAssert.deepEqual(
+      result.map((entry) => entry.id),
+      ["thread-1", "thread-2"],
+    );
+    NodeAssert.deepEqual(requests, [
+      {
+        archived: false,
+        limit: 100,
+        sortKey: "updated_at",
+        sortDirection: "desc",
+        useStateDbOnly: false,
+      },
+      {
+        archived: false,
+        limit: 100,
+        sortKey: "updated_at",
+        sortDirection: "desc",
+        useStateDbOnly: false,
+        cursor: "cursor-1",
+      },
+    ]);
+  }),
+);
+
+it.effect("stops safely when Codex repeats a thread/list cursor", () =>
+  Effect.gen(function* () {
+    let requestCount = 0;
+    const client: CodexThreadListClient = {
+      request: () =>
+        Effect.sync(() => {
+          requestCount += 1;
+          return {
+            data: [thread(`thread-${requestCount}`, requestCount)],
+            nextCursor: "repeated-cursor",
+          };
+        }),
+    };
+
+    const result = yield* listAllCodexThreads(client);
+
+    NodeAssert.equal(requestCount, 2);
+    NodeAssert.deepEqual(
+      result.map((entry) => entry.id),
+      ["thread-1", "thread-2"],
+    );
+  }),
+);