From 9a0ad73259413da67bd19de1adc7bb5eb984bc02 Mon Sep 17 00:00:00 2001 From: seeb1337 <63622047+seeb1337@users.noreply.github.com> Date: Sat, 8 Aug 2026 23:08:40 +0100 Subject: [PATCH 1/3] feat(codex): sync persisted threads --- .../src/provider/Layers/CodexAdapter.ts | 55 ++ .../Layers/CodexAppServerConnection.test.ts | 93 +++ .../Layers/CodexAppServerConnection.ts | 131 ++++ .../provider/Layers/CodexSessionRuntime.ts | 60 +- .../Layers/ProviderThreadDiscovery.test.ts | 327 ++++++++++ .../Layers/ProviderThreadDiscovery.ts | 580 ++++++++++++++++++ .../src/provider/Services/ProviderAdapter.ts | 22 + apps/server/src/server.ts | 18 +- apps/web/src/components/ChatView.tsx | 7 +- 9 files changed, 1240 insertions(+), 53 deletions(-) create mode 100644 apps/server/src/provider/Layers/CodexAppServerConnection.test.ts create mode 100644 apps/server/src/provider/Layers/CodexAppServerConnection.ts create mode 100644 apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts create mode 100644 apps/server/src/provider/Layers/ProviderThreadDiscovery.ts diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 6b99bf52b1e0..f458cc235e75 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/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 = () => + 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: cause.message, + 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() { diff --git a/apps/server/src/provider/Layers/CodexAppServerConnection.test.ts b/apps/server/src/provider/Layers/CodexAppServerConnection.test.ts new file mode 100644 index 000000000000..bd5b59c664d8 --- /dev/null +++ b/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"], + ); + }), +); diff --git a/apps/server/src/provider/Layers/CodexAppServerConnection.ts b/apps/server/src/provider/Layers/CodexAppServerConnection.ts new file mode 100644 index 000000000000..d33b56364565 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexAppServerConnection.ts @@ -0,0 +1,131 @@ +import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Scope from "effect/Scope"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import * as CodexClient from "effect-codex-app-server/client"; +import * as CodexErrors from "effect-codex-app-server/errors"; +import type * as CodexRpc from "effect-codex-app-server/rpc"; +import type * as EffectCodexSchema from "effect-codex-app-server/schema"; + +import { expandHomePath } from "../../pathExpansion.ts"; +import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; + +const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const; + +export interface CodexAppServerConnectionOptions { + readonly binaryPath: string; + readonly cwd: string; + readonly environment?: NodeJS.ProcessEnv; + readonly homePath?: string; + readonly appServerArgs?: ReadonlyArray; + readonly launchArgs?: string; +} + +export interface CodexThreadListClient { + readonly request: ( + method: "thread/list", + payload: CodexRpc.ClientRequestParamsByMethod["thread/list"], + ) => Effect.Effect< + CodexRpc.ClientRequestResponsesByMethod["thread/list"], + CodexErrors.CodexAppServerError + >; +} + +/** + * Spawn and connect to a Codex app-server process in the caller's scope. + * Session runtimes and one-shot discovery calls share this path so CODEX_HOME, + * environment handling, process termination, and protocol setup cannot drift. + */ +export const makeCodexAppServerConnection = Effect.fn("makeCodexAppServerConnection")(function* ( + options: CodexAppServerConnectionOptions, +) { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const scope = yield* Scope.Scope; + const resolvedHomePath = options.homePath ? expandHomePath(options.homePath) : undefined; + const environment = { + ...options.environment, + ...(resolvedHomePath ? { CODEX_HOME: resolvedHomePath } : {}), + }; + const extendEnv = options.environment === undefined; + const spawnCommand = yield* resolveSpawnCommand( + options.binaryPath, + codexSessionAppServerArgs(options.appServerArgs, options.launchArgs), + { env: environment, extendEnv }, + ); + const child = yield* spawner + .spawn( + ChildProcess.make(spawnCommand.command, spawnCommand.args, { + cwd: options.cwd, + env: environment, + extendEnv, + forceKillAfter: CODEX_APP_SERVER_FORCE_KILL_AFTER, + shell: spawnCommand.shell, + }), + ) + .pipe( + Effect.provideService(Scope.Scope, scope), + Effect.mapError( + (cause) => + new CodexErrors.CodexAppServerSpawnError({ + command: `${options.binaryPath} app-server`, + cause, + }), + ), + ); + + const clientContext = yield* CodexClient.layerChildProcess(child).pipe( + Layer.build, + Effect.provideService(Scope.Scope, scope), + ); + const client = yield* Effect.service(CodexClient.CodexAppServerClient).pipe( + Effect.provide(clientContext), + ); + + return { child, client } as const; +}); + +const THREAD_LIST_PAGE_SIZE = 100; + +/** Fetch every page from Codex's persisted thread catalog, newest first. */ +export const listAllCodexThreads = Effect.fn("listAllCodexThreads")(function* ( + client: CodexThreadListClient, +) { + const threads: EffectCodexSchema.V2ThreadListResponse["data"][number][] = []; + const threadIds = new Set(); + const requestedCursors = new Set(); + let cursor: string | undefined; + + while (true) { + if (cursor !== undefined) { + if (requestedCursors.has(cursor)) { + yield* Effect.logWarning("Codex thread/list returned a repeated pagination cursor", { + cursor, + }); + break; + } + requestedCursors.add(cursor); + } + + const page = yield* client.request("thread/list", { + archived: false, + limit: THREAD_LIST_PAGE_SIZE, + sortKey: "updated_at", + sortDirection: "desc", + useStateDbOnly: false, + ...(cursor !== undefined ? { cursor } : {}), + }); + + for (const thread of page.data) { + if (threadIds.has(thread.id)) continue; + threadIds.add(thread.id); + threads.push(thread); + } + + const nextCursor = page.nextCursor ?? undefined; + if (nextCursor === undefined) break; + cursor = nextCursor; + } + + return threads; +}); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 58c012bd63ea..9791ac18f7a0 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -16,28 +16,24 @@ import { ThreadId, TurnId, } from "@t3tools/contracts"; -import { resolveSpawnCommand } from "@t3tools/shared/shell"; import { normalizeModelSlug } from "@t3tools/shared/model"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; -import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; -import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; -import * as CodexClient from "effect-codex-app-server/client"; +import { ChildProcessSpawner } from "effect/unstable/process"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { buildCodexInitializeParams } from "./CodexProvider.ts"; -import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; -import { expandHomePath } from "../../pathExpansion.ts"; +import { makeCodexAppServerConnection } from "./CodexAppServerConnection.ts"; import { buildCodexDeveloperInstructions } from "../CodexDeveloperInstructions.ts"; const decodeV2TurnStartResponse = Schema.decodeUnknownEffect(EffectCodexSchema.V2TurnStartResponse); @@ -51,7 +47,6 @@ const BENIGN_ERROR_LOG_SNIPPETS = [ "state db missing rollout path for thread", "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back", ]; -const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const; const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ "not found", "missing thread", @@ -846,7 +841,6 @@ export const makeCodexSessionRuntime = ( ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto | Scope.Scope > => Effect.gen(function* () { - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const runtimeScope = yield* Scope.Scope; const crypto = yield* Crypto.Crypto; const events = yield* Queue.unbounded(); @@ -859,48 +853,14 @@ export const makeCodexSessionRuntime = ( const collabChildLiveTurnsRef = yield* Ref.make(new Map()); const closedRef = yield* Ref.make(false); - // `~` is not shell-expanded when env vars are set via - // `child_process.spawn`; `expandHomePath` lets a configured - // `CODEX_HOME=~/.codex_work` reach codex as an absolute path. - const resolvedHomePath = options.homePath ? expandHomePath(options.homePath) : undefined; - const env = { - ...options.environment, - ...(resolvedHomePath ? { CODEX_HOME: resolvedHomePath } : {}), - }; - const extendEnv = options.environment === undefined; - const appServerArgs = codexSessionAppServerArgs(options.appServerArgs, options.launchArgs); - const spawnCommand = yield* resolveSpawnCommand(options.binaryPath, appServerArgs, { - env, - extendEnv, - }); - const child = yield* spawner - .spawn( - ChildProcess.make(spawnCommand.command, spawnCommand.args, { - cwd: options.cwd, - env, - extendEnv, - forceKillAfter: CODEX_APP_SERVER_FORCE_KILL_AFTER, - shell: spawnCommand.shell, - }), - ) - .pipe( - Effect.provideService(Scope.Scope, runtimeScope), - Effect.mapError( - (cause) => - new CodexErrors.CodexAppServerSpawnError({ - command: `${options.binaryPath} app-server`, - cause, - }), - ), - ); - - const clientContext = yield* CodexClient.layerChildProcess(child).pipe( - Layer.build, - Effect.provideService(Scope.Scope, runtimeScope), - ); - const client = yield* Effect.service(CodexClient.CodexAppServerClient).pipe( - Effect.provide(clientContext), - ); + const { child, client } = yield* makeCodexAppServerConnection({ + binaryPath: options.binaryPath, + cwd: options.cwd, + ...(options.environment ? { environment: options.environment } : {}), + ...(options.homePath ? { homePath: options.homePath } : {}), + ...(options.appServerArgs ? { appServerArgs: options.appServerArgs } : {}), + ...(options.launchArgs ? { launchArgs: options.launchArgs } : {}), + }).pipe(Effect.provideService(Scope.Scope, runtimeScope)); const serverNotifications = yield* Queue.unbounded(); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const randomUUIDv4 = (purpose: CodexErrors.CodexAppServerIdentifierPurpose) => diff --git a/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts b/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts new file mode 100644 index 000000000000..4f8ca111ed06 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts @@ -0,0 +1,327 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeAssert from "node:assert/strict"; +import { + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + type OrchestrationCommand, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import { it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import type { ProviderRuntimeBinding } from "../Services/ProviderSessionDirectory.ts"; +import type { ProviderThreadSummary } from "../Services/ProviderAdapter.ts"; +import { + synchronizeDiscoveredProviderThreads, + type ProviderThreadDiscoverySource, +} from "./ProviderThreadDiscovery.ts"; + +const codex = ProviderDriverKind.make("codex"); +const instanceId = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId, model: "gpt-5.4" } as const; + +function discoveredThread( + providerThreadId: string, + overrides: Partial = {}, +): ProviderThreadSummary { + return { + providerThreadId, + cwd: "/workspace/project", + title: undefined, + preview: `Prompt for ${providerThreadId}`, + branch: "main", + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-02T00:00:00.000Z", + ...overrides, + }; +} + +function source(threads: ReadonlyArray): ProviderThreadDiscoverySource { + return { + discoveryKey: "codex:home:/home/test/.codex", + driverKind: codex, + instanceId, + compatibleInstanceIds: [instanceId], + defaultModel: "gpt-5.4", + listThreads: () => Effect.succeed(threads), + }; +} + +function emptyReadModel(): OrchestrationReadModel { + return { + snapshotSequence: 0, + projects: [], + threads: [], + updatedAt: "1970-01-01T00:00:00.000Z", + }; +} + +it.effect( + "imports unlinked Codex threads, groups their project, and persists resume bindings", + () => + Effect.gen(function* () { + const commands: OrchestrationCommand[] = []; + const bindings: ProviderRuntimeBinding[] = []; + const result = yield* synchronizeDiscoveredProviderThreads({ + sources: [ + source([ + discoveredThread("provider-thread-1"), + discoveredThread("provider-thread-2", { + title: "Named in Codex", + createdAt: "2026-01-03T00:00:00.000Z", + updatedAt: "2026-01-04T00:00:00.000Z", + }), + ]), + ], + readModel: emptyReadModel(), + bindings: [], + dispatch: (command) => + Effect.sync(() => { + commands.push(command); + }), + upsertBinding: (binding) => + Effect.sync(() => { + bindings.push(binding); + }), + }); + + NodeAssert.deepEqual(result, { discovered: 2, imported: 2, refreshed: 0 }); + NodeAssert.equal(commands.filter((command) => command.type === "project.create").length, 1); + const createCommands = commands.filter((command) => command.type === "thread.create"); + NodeAssert.equal(createCommands.length, 2); + NodeAssert.equal(createCommands[0]?.title, "Prompt for provider-thread-1"); + NodeAssert.equal(createCommands[1]?.title, "Named in Codex"); + NodeAssert.equal(createCommands[0]?.projectId, createCommands[1]?.projectId); + + const sessionCommands = commands.filter((command) => command.type === "thread.session.set"); + NodeAssert.deepEqual( + sessionCommands.map((command) => ({ + status: command.session.status, + providerName: command.session.providerName, + providerInstanceId: command.session.providerInstanceId, + occurredAt: command.createdAt, + })), + [ + { + status: "stopped", + providerName: "codex", + providerInstanceId: "codex", + occurredAt: "2026-01-02T00:00:00.000Z", + }, + { + status: "stopped", + providerName: "codex", + providerInstanceId: "codex", + occurredAt: "2026-01-04T00:00:00.000Z", + }, + ], + ); + NodeAssert.deepEqual( + bindings.map((binding) => binding.resumeCursor), + [{ threadId: "provider-thread-1" }, { threadId: "provider-thread-2" }], + ); + NodeAssert.deepEqual( + bindings.map( + (binding) => (binding.runtimePayload as { readonly cwd?: string } | undefined)?.cwd, + ), + ["/workspace/project", "/workspace/project"], + ); + }), +); + +it.effect("does not duplicate a Codex thread that already belongs to a T3 thread", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project-existing"); + const threadId = ThreadId.make("t3-thread-existing"); + const readModel: OrchestrationReadModel = { + snapshotSequence: 1, + projects: [ + { + id: projectId, + title: "Project", + workspaceRoot: "/workspace/project", + defaultModelSelection: modelSelection, + scripts: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + deletedAt: null, + }, + ], + threads: [ + { + id: threadId, + projectId, + title: "Created in T3", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }, + ], + updatedAt: "2026-01-01T00:00:00.000Z", + }; + const commands: OrchestrationCommand[] = []; + + const result = yield* synchronizeDiscoveredProviderThreads({ + sources: [source([discoveredThread("provider-thread-existing")])], + readModel, + bindings: [ + { + threadId, + provider: codex, + providerInstanceId: instanceId, + resumeCursor: { threadId: "provider-thread-existing" }, + }, + ], + dispatch: (command) => + Effect.sync(() => { + commands.push(command); + }), + upsertBinding: () => Effect.void, + }); + + NodeAssert.deepEqual(result, { discovered: 1, imported: 0, refreshed: 0 }); + NodeAssert.deepEqual(commands, []); + }), +); + +it.effect("refreshes imported thread metadata from a newer Codex thread/list result", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project-imported"); + const threadId = ThreadId.make("thread-imported"); + const originalTitle = "Original Codex title"; + const readModel: OrchestrationReadModel = { + snapshotSequence: 1, + projects: [ + { + id: projectId, + title: "Project", + workspaceRoot: "/workspace/project", + defaultModelSelection: modelSelection, + scripts: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + deletedAt: null, + }, + ], + threads: [ + { + id: threadId, + projectId, + title: originalTitle, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: "main", + worktreePath: null, + latestTurn: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-02T00:00:00.000Z", + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: { + threadId, + status: "stopped", + providerName: codex, + providerInstanceId: instanceId, + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: "2026-01-02T00:00:00.000Z", + }, + }, + ], + updatedAt: "2026-01-02T00:00:00.000Z", + }; + const commands: OrchestrationCommand[] = []; + const bindings: ProviderRuntimeBinding[] = []; + const updatedAt = "2026-01-05T00:00:00.000Z"; + + const result = yield* synchronizeDiscoveredProviderThreads({ + sources: [ + source([ + discoveredThread("provider-thread-imported", { + title: "Fresh Codex title", + updatedAt, + }), + ]), + ], + readModel, + bindings: [ + { + threadId, + provider: codex, + providerInstanceId: instanceId, + status: "stopped", + runtimeMode: "full-access", + resumeCursor: { threadId: "provider-thread-imported" }, + runtimePayload: { + cwd: "/workspace/project", + modelSelection, + providerThreadDiscovery: { + version: 1, + discoveryKey: "codex:home:/home/test/.codex", + providerThreadId: "provider-thread-imported", + providerUpdatedAt: "2026-01-02T00:00:00.000Z", + providerTitle: originalTitle, + }, + }, + }, + ], + dispatch: (command) => + Effect.sync(() => { + commands.push(command); + }), + upsertBinding: (binding) => + Effect.sync(() => { + bindings.push(binding); + }), + }); + + NodeAssert.deepEqual(result, { discovered: 1, imported: 0, refreshed: 1 }); + NodeAssert.deepEqual( + commands.map((command) => command.type), + ["thread.meta.update", "thread.session.set"], + ); + const sessionCommand = commands.find((command) => command.type === "thread.session.set"); + NodeAssert.equal(sessionCommand?.createdAt, updatedAt); + NodeAssert.equal(readModel.threads[0]?.title, originalTitle); + + const metadata = ( + bindings[0]?.runtimePayload as { + readonly providerThreadDiscovery?: { + readonly providerUpdatedAt?: string; + readonly providerTitle?: string; + }; + } + )?.providerThreadDiscovery; + NodeAssert.deepEqual(metadata, { + version: 1, + discoveryKey: "codex:home:/home/test/.codex", + providerThreadId: "provider-thread-imported", + providerUpdatedAt: updatedAt, + providerTitle: "Fresh Codex title", + }); + }), +); diff --git a/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts b/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts new file mode 100644 index 000000000000..ca1262d62678 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts @@ -0,0 +1,580 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeCrypto from "node:crypto"; +import * as NodePath from "node:path"; +import { + CommandId, + DEFAULT_MODEL, + ProjectId, + type ModelSelection, + type OrchestrationCommand, + type OrchestrationReadModel, + type ProviderDriverKind, + type ProviderInstanceId, + ThreadId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schedule from "effect/Schedule"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; + +import type { OrchestrationDispatchError } from "../../orchestration/Errors.ts"; +import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { + ProviderRuntimeBinding, + ProviderSessionDirectoryWriteError, +} from "../Services/ProviderSessionDirectory.ts"; +import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import { ProviderInstanceRegistry } from "../Services/ProviderInstanceRegistry.ts"; +import type { ProviderThreadSummary } from "../Services/ProviderAdapter.ts"; +import type { ProviderInstance } from "../ProviderDriver.ts"; + +const DISCOVERY_INTERVAL = "30 seconds"; +const DISCOVERY_TIMEOUT = "20 seconds"; +const IMPORT_METADATA_KEY = "providerThreadDiscovery"; +const IMPORT_METADATA_VERSION = 1; + +interface ImportedThreadMetadata { + readonly version: 1; + readonly discoveryKey: string; + readonly providerThreadId: string; + readonly providerUpdatedAt: string; + readonly providerTitle: string; +} + +interface ThreadDiscoveryBinding extends ProviderRuntimeBinding { + readonly threadId: ThreadId; +} + +export interface ProviderThreadDiscoverySource { + readonly discoveryKey: string; + readonly driverKind: ProviderDriverKind; + readonly instanceId: ProviderInstanceId; + readonly compatibleInstanceIds: ReadonlyArray; + readonly defaultModel: string; + readonly listThreads: () => Effect.Effect, never>; +} + +export interface ProviderThreadSynchronizationInput { + readonly sources: ReadonlyArray; + readonly readModel: OrchestrationReadModel; + readonly bindings: ReadonlyArray; + readonly dispatch: ( + command: OrchestrationCommand, + ) => Effect.Effect; + readonly upsertBinding: ( + binding: ProviderRuntimeBinding, + ) => Effect.Effect; +} + +export interface ProviderThreadSynchronizationResult { + readonly discovered: number; + readonly imported: number; + readonly refreshed: number; +} + +export type ProviderThreadSynchronizationError = + | OrchestrationDispatchError + | ProviderSessionDirectoryWriteError; + +type Mutable = { -readonly [Key in keyof T]: T[Key] }; + +type KnownThread = Mutable< + Pick< + OrchestrationReadModel["threads"][number], + | "id" + | "projectId" + | "title" + | "modelSelection" + | "runtimeMode" + | "interactionMode" + | "branch" + | "worktreePath" + | "createdAt" + | "updatedAt" + | "deletedAt" + | "session" + > +>; + +function stableId(prefix: string, ...parts: ReadonlyArray): string { + const digest = NodeCrypto.createHash("sha256") + .update(JSON.stringify(parts), "utf8") + .digest("hex") + .slice(0, 32); + return `${prefix}-${digest}`; +} + +function readProviderThreadId(resumeCursor: unknown | null | undefined): string | undefined { + if (!resumeCursor || typeof resumeCursor !== "object" || Array.isArray(resumeCursor)) { + return undefined; + } + const threadId = "threadId" in resumeCursor ? resumeCursor.threadId : undefined; + return typeof threadId === "string" && threadId.trim().length > 0 ? threadId : undefined; +} + +function readImportMetadata( + runtimePayload: unknown | null | undefined, +): ImportedThreadMetadata | undefined { + if (!runtimePayload || typeof runtimePayload !== "object" || Array.isArray(runtimePayload)) { + return undefined; + } + const raw = + IMPORT_METADATA_KEY in runtimePayload ? runtimePayload[IMPORT_METADATA_KEY] : undefined; + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return undefined; + const metadata = raw as Partial; + return metadata.version === IMPORT_METADATA_VERSION && + typeof metadata.discoveryKey === "string" && + typeof metadata.providerThreadId === "string" && + typeof metadata.providerUpdatedAt === "string" && + typeof metadata.providerTitle === "string" + ? (metadata as ImportedThreadMetadata) + : undefined; +} + +function normalizeTitle(value: string | undefined): string | undefined { + const compact = value?.trim().replace(/\s+/g, " "); + if (!compact) return undefined; + return compact.length <= 120 ? compact : `${compact.slice(0, 117).trimEnd()}...`; +} + +function discoveredThreadTitle(thread: ProviderThreadSummary): string { + return normalizeTitle(thread.title) ?? normalizeTitle(thread.preview) ?? "Codex thread"; +} + +function projectTitle(cwd: string): string { + return normalizeTitle(NodePath.basename(cwd)) ?? normalizeTitle(cwd) ?? "Imported Codex"; +} + +function importedThreadMetadata(input: { + readonly source: ProviderThreadDiscoverySource; + readonly thread: ProviderThreadSummary; + readonly title: string; +}): ImportedThreadMetadata { + return { + version: IMPORT_METADATA_VERSION, + discoveryKey: input.source.discoveryKey, + providerThreadId: input.thread.providerThreadId, + providerUpdatedAt: input.thread.updatedAt, + providerTitle: input.title, + }; +} + +function isNewerTimestamp(next: string, previous: string): boolean { + const nextMs = Date.parse(next); + const previousMs = Date.parse(previous); + return Number.isFinite(nextMs) && Number.isFinite(previousMs) + ? nextMs > previousMs + : next !== previous; +} + +function commandId(kind: string, ...parts: ReadonlyArray): CommandId { + return CommandId.make(stableId(`provider-thread-${kind}`, ...parts)); +} + +function canonicalThreadId( + source: ProviderThreadDiscoverySource, + providerThreadId: string, +): ThreadId { + return ThreadId.make(stableId("codex-thread", source.discoveryKey, providerThreadId)); +} + +function canonicalProjectId( + source: ProviderThreadDiscoverySource, + cwd: string, + salt?: string, +): ProjectId { + return ProjectId.make(stableId("codex-project", source.discoveryKey, cwd, salt ?? "")); +} + +function makeImportedSession(input: { + readonly thread: KnownThread; + readonly source: ProviderThreadDiscoverySource; + readonly instanceId: ProviderInstanceId; + readonly updatedAt: string; +}) { + const existing = input.thread.session; + return { + threadId: input.thread.id, + status: existing?.status ?? ("stopped" as const), + providerName: input.source.driverKind, + providerInstanceId: input.instanceId, + runtimeMode: input.thread.runtimeMode, + activeTurnId: existing?.activeTurnId ?? null, + lastError: existing?.lastError ?? null, + updatedAt: input.updatedAt, + }; +} + +function makeRuntimePayload(input: { + readonly cwd: string; + readonly modelSelection: ModelSelection; + readonly metadata: ImportedThreadMetadata; +}) { + return { + cwd: input.cwd, + modelSelection: input.modelSelection, + [IMPORT_METADATA_KEY]: input.metadata, + }; +} + +/** + * Reconcile provider-native threads into T3's event model and durable resume + * directory. Existing T3-created threads are recognized by resume cursor and + * never duplicated or renamed. + */ +export const synchronizeDiscoveredProviderThreads = Effect.fn( + "synchronizeDiscoveredProviderThreads", +)(function* (input: ProviderThreadSynchronizationInput) { + const projects = [...input.readModel.projects]; + const projectIds = new Set(projects.map((project) => project.id)); + const threads = new Map( + input.readModel.threads.map((thread) => [thread.id, { ...thread }]), + ); + const bindingsBySource = new Map>(); + + for (const source of input.sources) { + const compatibleInstanceIds = new Set(source.compatibleInstanceIds); + const bindings = new Map(); + for (const binding of input.bindings) { + if ( + binding.provider !== source.driverKind || + binding.providerInstanceId === undefined || + !compatibleInstanceIds.has(binding.providerInstanceId) + ) { + continue; + } + const providerThreadId = readProviderThreadId(binding.resumeCursor); + if (providerThreadId !== undefined && !bindings.has(providerThreadId)) { + bindings.set(providerThreadId, binding); + } + } + bindingsBySource.set(source.discoveryKey, bindings); + } + + let discoveredCount = 0; + let importedCount = 0; + let refreshedCount = 0; + + for (const source of input.sources) { + const discoveredThreads = yield* source.listThreads(); + const sourceBindings = bindingsBySource.get(source.discoveryKey)!; + const modelSelection: ModelSelection = { + instanceId: source.instanceId, + model: source.defaultModel, + }; + + for (const discovered of [...discoveredThreads].sort((left, right) => + left.createdAt.localeCompare(right.createdAt), + )) { + discoveredCount += 1; + const title = discoveredThreadTitle(discovered); + const linkedBinding = sourceBindings.get(discovered.providerThreadId); + + if (linkedBinding !== undefined) { + const metadata = readImportMetadata(linkedBinding.runtimePayload); + if (metadata === undefined) { + // A native Codex thread already belongs to a T3-created thread. + continue; + } + const linkedThread = threads.get(linkedBinding.threadId); + if (linkedThread === undefined || linkedThread.deletedAt !== null) continue; + + const needsSessionRepair = linkedThread.session === null; + const providerChanged = isNewerTimestamp(discovered.updatedAt, metadata.providerUpdatedAt); + if (!needsSessionRepair && !providerChanged) continue; + + if ( + providerChanged && + linkedThread.title === metadata.providerTitle && + title !== metadata.providerTitle + ) { + yield* input.dispatch({ + type: "thread.meta.update", + commandId: commandId( + "title", + source.discoveryKey, + discovered.providerThreadId, + discovered.updatedAt, + title, + ), + threadId: linkedThread.id, + title, + }); + linkedThread.title = title; + } + + const targetInstanceId = linkedBinding.providerInstanceId ?? source.instanceId; + const session = makeImportedSession({ + thread: linkedThread, + source, + instanceId: targetInstanceId, + updatedAt: discovered.updatedAt, + }); + yield* input.dispatch({ + type: "thread.session.set", + commandId: commandId( + "refresh", + source.discoveryKey, + discovered.providerThreadId, + discovered.updatedAt, + ), + threadId: linkedThread.id, + session, + createdAt: discovered.updatedAt, + }); + linkedThread.session = session; + linkedThread.updatedAt = discovered.updatedAt; + + const nextMetadata = importedThreadMetadata({ source, thread: discovered, title }); + const nextBinding: ThreadDiscoveryBinding = { + threadId: linkedThread.id, + provider: source.driverKind, + providerInstanceId: targetInstanceId, + status: linkedBinding.status ?? "stopped", + runtimeMode: linkedThread.runtimeMode, + resumeCursor: { threadId: discovered.providerThreadId }, + runtimePayload: makeRuntimePayload({ + cwd: discovered.cwd, + modelSelection: linkedThread.modelSelection, + metadata: nextMetadata, + }), + }; + yield* input.upsertBinding(nextBinding); + sourceBindings.set(discovered.providerThreadId, nextBinding); + refreshedCount += 1; + continue; + } + + const threadId = canonicalThreadId(source, discovered.providerThreadId); + let knownThread = threads.get(threadId); + if (knownThread !== undefined && knownThread.deletedAt !== null) { + // Deleting an imported thread is an explicit local opt-out. Do not + // resurrect it just because it remains in Codex's history. + continue; + } + + if (knownThread === undefined) { + let project = projects.find( + (candidate) => candidate.deletedAt === null && candidate.workspaceRoot === discovered.cwd, + ); + if (project === undefined) { + let projectId = canonicalProjectId(source, discovered.cwd); + if (projectIds.has(projectId)) { + projectId = canonicalProjectId(source, discovered.cwd, discovered.providerThreadId); + } + yield* input.dispatch({ + type: "project.create", + commandId: commandId("project", source.discoveryKey, discovered.cwd, projectId), + projectId, + title: projectTitle(discovered.cwd), + workspaceRoot: discovered.cwd, + defaultModelSelection: modelSelection, + createdAt: discovered.createdAt, + }); + project = { + id: projectId, + title: projectTitle(discovered.cwd), + workspaceRoot: discovered.cwd, + defaultModelSelection: modelSelection, + scripts: [], + createdAt: discovered.createdAt, + updatedAt: discovered.createdAt, + deletedAt: null, + }; + projects.push(project); + projectIds.add(projectId); + } + + yield* input.dispatch({ + type: "thread.create", + commandId: commandId("create", source.discoveryKey, discovered.providerThreadId), + threadId, + projectId: project.id, + title, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: discovered.branch ?? null, + worktreePath: null, + createdAt: discovered.createdAt, + }); + knownThread = { + id: threadId, + projectId: project.id, + title, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: discovered.branch ?? null, + worktreePath: null, + createdAt: discovered.createdAt, + updatedAt: discovered.createdAt, + deletedAt: null, + session: null, + }; + threads.set(threadId, knownThread); + } + + const metadata = importedThreadMetadata({ source, thread: discovered, title }); + const binding: ThreadDiscoveryBinding = { + threadId, + provider: source.driverKind, + providerInstanceId: source.instanceId, + status: "stopped", + runtimeMode: knownThread.runtimeMode, + resumeCursor: { threadId: discovered.providerThreadId }, + runtimePayload: makeRuntimePayload({ + cwd: discovered.cwd, + modelSelection: knownThread.modelSelection, + metadata, + }), + }; + yield* input.upsertBinding(binding); + sourceBindings.set(discovered.providerThreadId, binding); + + const session = makeImportedSession({ + thread: knownThread, + source, + instanceId: source.instanceId, + updatedAt: discovered.updatedAt, + }); + yield* input.dispatch({ + type: "thread.session.set", + commandId: commandId( + "session", + source.discoveryKey, + discovered.providerThreadId, + discovered.updatedAt, + ), + threadId, + session, + createdAt: discovered.updatedAt, + }); + knownThread.session = session; + knownThread.updatedAt = discovered.updatedAt; + importedCount += 1; + } + } + + return { + discovered: discoveredCount, + imported: importedCount, + refreshed: refreshedCount, + } satisfies ProviderThreadSynchronizationResult; +}); + +function buildDiscoverySources( + instances: ReadonlyArray, +): Effect.Effect> { + const grouped = new Map(); + for (const instance of instances) { + if (!instance.enabled || instance.adapter.listThreads === undefined) continue; + const key = `${instance.driverKind}:${instance.continuationIdentity.continuationKey}`; + const current = grouped.get(key) ?? []; + grouped.set(key, [...current, instance]); + } + + return Effect.forEach(Array.from(grouped.entries()), ([discoveryKey, compatibleInstances]) => + Effect.gen(function* () { + const primary = compatibleInstances[0]!; + const snapshot = yield* primary.snapshot.getSnapshot; + const listThreads = primary.adapter.listThreads!; + return { + discoveryKey, + driverKind: primary.driverKind, + instanceId: primary.instanceId, + compatibleInstanceIds: compatibleInstances.map((instance) => instance.instanceId), + defaultModel: snapshot.models[0]?.slug ?? DEFAULT_MODEL, + listThreads: () => + listThreads().pipe( + Effect.timeout(DISCOVERY_TIMEOUT), + Effect.catchCause((cause) => + Effect.logWarning("Could not refresh provider thread discovery", { + discoveryKey, + provider: primary.driverKind, + providerInstanceId: primary.instanceId, + cause: Cause.pretty(cause), + }).pipe(Effect.as([] as ReadonlyArray)), + ), + ), + } satisfies ProviderThreadDiscoverySource; + }), + ); +} + +const makeProviderThreadDiscovery = Effect.gen(function* () { + const registry = yield* ProviderInstanceRegistry; + const directory = yield* ProviderSessionDirectory; + const orchestrationEngine = yield* OrchestrationEngineService; + const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + const synchronizationLock = yield* Semaphore.make(1); + const instanceChanges = yield* registry.subscribeChanges; + + const synchronize = synchronizationLock.withPermits(1)( + Effect.gen(function* () { + const instances = yield* registry.listInstances; + const sources = yield* buildDiscoverySources(instances); + if (sources.length === 0) return; + + const [readModel, persistedBindings] = yield* Effect.all([ + projectionSnapshotQuery.getCommandReadModel(), + directory.listBindings(), + ]); + const activeBindings = yield* Effect.forEach( + instances, + (instance) => + instance.adapter.listSessions().pipe( + Effect.map((sessions) => + sessions.map( + (session): ThreadDiscoveryBinding => ({ + threadId: session.threadId, + provider: instance.driverKind, + providerInstanceId: instance.instanceId, + runtimeMode: session.runtimeMode, + status: session.status === "error" ? "error" : "running", + resumeCursor: session.resumeCursor, + }), + ), + ), + Effect.catchCause(() => Effect.succeed([] as ReadonlyArray)), + ), + { concurrency: "unbounded" }, + ).pipe(Effect.map((groups) => groups.flat())); + + const result = yield* synchronizeDiscoveredProviderThreads({ + sources, + readModel, + bindings: [...persistedBindings, ...activeBindings], + dispatch: (command) => orchestrationEngine.dispatch(command), + upsertBinding: (binding) => directory.upsert(binding), + }); + if (result.imported > 0 || result.refreshed > 0) { + yield* Effect.logInfo("Provider thread discovery synchronized", result); + } else { + yield* Effect.logDebug("Provider thread discovery is current", result); + } + }), + ); + + const synchronizeSafely = synchronize.pipe( + Effect.catchCause((cause) => + Effect.logWarning("Provider thread discovery synchronization failed", { + cause: Cause.pretty(cause), + }), + ), + ); + + yield* synchronizeSafely.pipe( + Effect.repeat(Schedule.spaced(DISCOVERY_INTERVAL)), + Effect.forkScoped, + ); + yield* Stream.fromSubscription(instanceChanges).pipe( + Stream.runForEach(() => synchronizeSafely), + Effect.forkScoped, + ); +}); + +export const ProviderThreadDiscoveryLive = Layer.effectDiscard(makeProviderThreadDiscovery); diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 01eeae7b7bd7..4ce0d1e42ebe 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -42,6 +42,20 @@ export interface ProviderThreadSnapshot { readonly turns: ReadonlyArray; } +/** + * Lightweight provider-native thread metadata used to discover conversations + * that were created outside T3 Code. + */ +export interface ProviderThreadSummary { + readonly providerThreadId: string; + readonly cwd: string; + readonly title: string | undefined; + readonly preview: string | undefined; + readonly branch: string | undefined; + readonly createdAt: string; + readonly updatedAt: string; +} + export interface ProviderAdapterShape { /** * Provider kind implemented by this adapter. @@ -96,6 +110,14 @@ export interface ProviderAdapterShape { */ readonly listSessions: () => Effect.Effect>; + /** + * List persisted provider threads, when the provider supports discovery. + * + * This is optional because not every provider exposes a durable global + * thread catalog. Callers must treat absence as an unsupported capability. + */ + readonly listThreads?: () => Effect.Effect, TError>; + /** * Check whether this adapter owns an active session id. */ diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index ff21c07a861a..581f9eaf63fa 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -30,6 +30,7 @@ import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRe import * as ProviderEventLoggers from "./provider/Layers/ProviderEventLoggers.ts"; import { ProviderServiceLive } from "./provider/Layers/ProviderService.ts"; import { ProviderSessionReaperLive } from "./provider/Layers/ProviderSessionReaper.ts"; +import { ProviderThreadDiscoveryLive } from "./provider/Layers/ProviderThreadDiscovery.ts"; import * as OpenCodeRuntime from "./provider/opencodeRuntime.ts"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as CheckpointStore from "./checkpointing/CheckpointStore.ts"; @@ -362,6 +363,16 @@ const ProviderRuntimeLayerLive = ProviderSessionReaperLive.pipe( Layer.provideMerge(OrchestrationLayerLive), ); +const ProviderThreadDiscoveryLayerLive = ProviderThreadDiscoveryLive.pipe( + Layer.provide( + Layer.mergeAll( + OrchestrationLayerLive, + ProviderSessionDirectoryLayerLive, + ProviderInstanceRegistryHydrationLive, + ), + ), +); + const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( // Core Services Layer.provideMerge(ServerSettingsLayerLive), @@ -380,6 +391,10 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( // `providerInstances` hydration merges `settings.providers.` // with explicit `providerInstances` entries on boot. Layer.provideMerge(ProviderInstanceRegistryHydrationLive), + // Import provider-native threads (currently Codex thread/list) after the + // instance registry and orchestration services are available. The layer + // performs an immediate sync and keeps the shell fresh in the background. + Layer.provideMerge(ProviderThreadDiscoveryLayerLive), // Shared native/canonical NDJSON writers used by both the per-instance // drivers (native stream, written from inside each `Adapter`) and // `ProviderService` (canonical stream, written after event normalization). @@ -392,8 +407,7 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( // no longer transitively provides it. Exposing it at the runtime level // keeps a single Live for all opencode consumers. Layer.provideMerge(OpenCodeRuntime.OpenCodeRuntimeLive), - Layer.provideMerge(WorkspaceLayerLive), - Layer.provideMerge(ProjectFaviconResolverLayerLive), + Layer.provideMerge(Layer.mergeAll(WorkspaceLayerLive, ProjectFaviconResolverLayerLive)), Layer.provideMerge(RepositoryIdentityResolver.layer), Layer.provideMerge(ServerEnvironment.layer), Layer.provideMerge(AuthLayerLive), diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 07f5d3b5fcc1..2f0a400884f4 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -303,6 +303,7 @@ import { revokeUserMessagePreviewUrls, shouldWriteThreadErrorToCurrentServerThread, startNewThreadForProject, + threadHasStarted, waitForStartedServerThread, } from "./ChatView.logic"; import type { ThreadSyncPhase } from "../threadSync"; @@ -4939,7 +4940,11 @@ function ChatViewContent(props: ChatViewProps) { return; } const threadIdForSend = activeThread.id; - const isFirstMessage = !isServerThread || activeThread.messages.length === 0; + // Imported provider threads can have no locally projected messages while + // still carrying a durable stopped session. Treat those as established + // conversations so a follow-up resumes them instead of re-bootstraping or + // replacing their imported title. + const isFirstMessage = !isServerThread || !threadHasStarted(activeThread); const baseBranchForWorktree = isFirstMessage && sendEnvMode === "worktree" && !activeThread.worktreePath ? activeThreadBranch From c72b57d4eb9bbc4c762b0114d7460f137e48b25c Mon Sep 17 00:00:00 2001 From: seeb1337 <63622047+seeb1337@users.noreply.github.com> Date: Mon, 10 Aug 2026 00:13:32 +0100 Subject: [PATCH 2/3] fix(codex): harden imported thread synchronization --- .../src/provider/Layers/CodexAdapter.ts | 2 +- .../Layers/ProviderThreadDiscovery.test.ts | 12 ++- .../Layers/ProviderThreadDiscovery.ts | 90 ++++++++++--------- .../web/src/components/ChatView.logic.test.ts | 19 ++++ apps/web/src/components/ChatView.logic.ts | 7 ++ apps/web/src/components/ChatView.tsx | 14 +-- docs/user/providers-codex.md | 14 +++ 7 files changed, 106 insertions(+), 52 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index f458cc235e75..fa363ad56133 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -1997,7 +1997,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( new ProviderAdapterRequestError({ provider: PROVIDER, method: "thread/list", - detail: cause.message, + detail: "Failed to list persisted Codex threads.", cause, }), ), diff --git a/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts b/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts index 4f8ca111ed06..26a804d8a8d4 100644 --- a/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts +++ b/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts @@ -200,7 +200,7 @@ it.effect("does not duplicate a Codex thread that already belongs to a T3 thread }), ); -it.effect("refreshes imported thread metadata from a newer Codex thread/list result", () => +it.effect("refreshes imported metadata without replacing the projected session state", () => Effect.gen(function* () { const projectId = ProjectId.make("project-imported"); const threadId = ThreadId.make("thread-imported"); @@ -302,10 +302,14 @@ it.effect("refreshes imported thread metadata from a newer Codex thread/list res NodeAssert.deepEqual(result, { discovered: 1, imported: 0, refreshed: 1 }); NodeAssert.deepEqual( commands.map((command) => command.type), - ["thread.meta.update", "thread.session.set"], + ["thread.meta.update"], ); - const sessionCommand = commands.find((command) => command.type === "thread.session.set"); - NodeAssert.equal(sessionCommand?.createdAt, updatedAt); + NodeAssert.equal( + commands[0]?.type === "thread.meta.update" && commands[0].title, + "Fresh Codex title", + ); + NodeAssert.equal(bindings[0]?.status, undefined); + NodeAssert.equal(bindings[0]?.runtimeMode, undefined); NodeAssert.equal(readModel.threads[0]?.title, originalTitle); const metadata = ( diff --git a/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts b/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts index ca1262d62678..c9b821dd910a 100644 --- a/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts +++ b/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts @@ -22,6 +22,7 @@ import * as Stream from "effect/Stream"; import type { OrchestrationDispatchError } from "../../orchestration/Errors.ts"; import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { forkParked } from "../../serverActivation.ts"; import type { ProviderRuntimeBinding, ProviderSessionDirectoryWriteError, @@ -286,55 +287,53 @@ export const synchronizeDiscoveredProviderThreads = Effect.fn( const providerChanged = isNewerTimestamp(discovered.updatedAt, metadata.providerUpdatedAt); if (!needsSessionRepair && !providerChanged) continue; - if ( - providerChanged && - linkedThread.title === metadata.providerTitle && - title !== metadata.providerTitle - ) { + const providerOwnsTitle = linkedThread.title === metadata.providerTitle; + if (providerChanged) { yield* input.dispatch({ type: "thread.meta.update", commandId: commandId( - "title", + "refresh", source.discoveryKey, discovered.providerThreadId, discovered.updatedAt, - title, ), threadId: linkedThread.id, - title, + ...(providerOwnsTitle && title !== metadata.providerTitle ? { title } : {}), }); - linkedThread.title = title; + if (providerOwnsTitle && title !== metadata.providerTitle) { + linkedThread.title = title; + } + linkedThread.updatedAt = discovered.updatedAt; } const targetInstanceId = linkedBinding.providerInstanceId ?? source.instanceId; - const session = makeImportedSession({ - thread: linkedThread, - source, - instanceId: targetInstanceId, - updatedAt: discovered.updatedAt, - }); - yield* input.dispatch({ - type: "thread.session.set", - commandId: commandId( - "refresh", - source.discoveryKey, - discovered.providerThreadId, - discovered.updatedAt, - ), - threadId: linkedThread.id, - session, - createdAt: discovered.updatedAt, - }); - linkedThread.session = session; - linkedThread.updatedAt = discovered.updatedAt; + if (needsSessionRepair) { + const session = makeImportedSession({ + thread: linkedThread, + source, + instanceId: targetInstanceId, + updatedAt: discovered.updatedAt, + }); + yield* input.dispatch({ + type: "thread.session.set", + commandId: commandId( + "session-repair", + source.discoveryKey, + discovered.providerThreadId, + discovered.updatedAt, + ), + threadId: linkedThread.id, + session, + createdAt: discovered.updatedAt, + }); + linkedThread.session = session; + } const nextMetadata = importedThreadMetadata({ source, thread: discovered, title }); const nextBinding: ThreadDiscoveryBinding = { threadId: linkedThread.id, provider: source.driverKind, providerInstanceId: targetInstanceId, - status: linkedBinding.status ?? "stopped", - runtimeMode: linkedThread.runtimeMode, resumeCursor: { threadId: discovered.providerThreadId }, runtimePayload: makeRuntimePayload({ cwd: discovered.cwd, @@ -519,6 +518,18 @@ const makeProviderThreadDiscovery = Effect.gen(function* () { const sources = yield* buildDiscoverySources(instances); if (sources.length === 0) return; + // Provider discovery can require a process spawn and multiple pages. + // Complete it before sampling T3 state so reconciliation works from the + // freshest available projection and active-session view. + const discoveredSources = yield* Effect.forEach(sources, (source) => + source.listThreads().pipe( + Effect.map((threads) => ({ + ...source, + listThreads: () => Effect.succeed(threads), + })), + ), + ); + const [readModel, persistedBindings] = yield* Effect.all([ projectionSnapshotQuery.getCommandReadModel(), directory.listBindings(), @@ -545,9 +556,12 @@ const makeProviderThreadDiscovery = Effect.gen(function* () { ).pipe(Effect.map((groups) => groups.flat())); const result = yield* synchronizeDiscoveredProviderThreads({ - sources, + sources: discoveredSources, readModel, - bindings: [...persistedBindings, ...activeBindings], + // The synchronizer keeps the first binding for a provider thread. + // Prefer the live adapter view so a persisted stopped binding cannot + // mask a follow-up that became active while discovery was running. + bindings: [...activeBindings, ...persistedBindings], dispatch: (command) => orchestrationEngine.dispatch(command), upsertBinding: (binding) => directory.upsert(binding), }); @@ -567,13 +581,9 @@ const makeProviderThreadDiscovery = Effect.gen(function* () { ), ); - yield* synchronizeSafely.pipe( - Effect.repeat(Schedule.spaced(DISCOVERY_INTERVAL)), - Effect.forkScoped, - ); - yield* Stream.fromSubscription(instanceChanges).pipe( - Stream.runForEach(() => synchronizeSafely), - Effect.forkScoped, + yield* forkParked(synchronizeSafely.pipe(Effect.repeat(Schedule.spaced(DISCOVERY_INTERVAL)))); + yield* forkParked( + Stream.fromSubscription(instanceChanges).pipe(Stream.runForEach(() => synchronizeSafely)), ); }); diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 5c026c94a138..514585b67a9f 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -29,6 +29,7 @@ import { resolveThreadMetadataUpdateForNextTurn, resolveSendEnvMode, scheduleEnvironmentReconnectWarning, + shouldPrepareWorktreeForSend, startNewThreadForProject, shouldShowBranchMismatchBanner, shouldWriteThreadErrorToCurrentServerThread, @@ -382,6 +383,24 @@ describe("resolveSendEnvMode", () => { }); }); +describe("shouldPrepareWorktreeForSend", () => { + it("prepares a selected worktree whenever the thread does not have one yet", () => { + expect(shouldPrepareWorktreeForSend({ sendEnvMode: "worktree", worktreePath: null })).toBe( + true, + ); + }); + + it("does not prepare a worktree for local mode or an existing worktree", () => { + expect(shouldPrepareWorktreeForSend({ sendEnvMode: "local", worktreePath: null })).toBe(false); + expect( + shouldPrepareWorktreeForSend({ + sendEnvMode: "worktree", + worktreePath: "/workspace/project/.worktrees/existing", + }), + ).toBe(false); + }); +}); + describe("branchMismatchKey", () => { it("builds a key from thread id and both branches", () => { expect(branchMismatchKey("thread-1", { threadBranch: "feat/a", currentBranch: "feat/b" })).toBe( diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 60df1cd966fd..60db9b17c048 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -257,6 +257,13 @@ export function resolveSendEnvMode(input: { return input.isGitRepo ? input.requestedEnvMode : "local"; } +export function shouldPrepareWorktreeForSend(input: { + sendEnvMode: DraftThreadEnvMode; + worktreePath: string | null; +}): boolean { + return input.sendEnvMode === "worktree" && input.worktreePath === null; +} + export function cloneComposerImageForRetry( image: ComposerImageAttachment, ): ComposerImageAttachment { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 2f0a400884f4..4c6e468f8458 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -303,6 +303,7 @@ import { revokeUserMessagePreviewUrls, shouldWriteThreadErrorToCurrentServerThread, startNewThreadForProject, + shouldPrepareWorktreeForSend, threadHasStarted, waitForStartedServerThread, } from "./ChatView.logic"; @@ -4945,16 +4946,15 @@ function ChatViewContent(props: ChatViewProps) { // conversations so a follow-up resumes them instead of re-bootstraping or // replacing their imported title. const isFirstMessage = !isServerThread || !threadHasStarted(activeThread); - const baseBranchForWorktree = - isFirstMessage && sendEnvMode === "worktree" && !activeThread.worktreePath - ? activeThreadBranch - : null; + const shouldPrepareWorktree = shouldPrepareWorktreeForSend({ + sendEnvMode, + worktreePath: activeThread.worktreePath, + }); + const baseBranchForWorktree = shouldPrepareWorktree ? activeThreadBranch : null; // In worktree mode, require an explicit base branch so we don't silently // fall back to local execution when branch selection is missing. - const shouldCreateWorktree = - isFirstMessage && sendEnvMode === "worktree" && !activeThread.worktreePath; - if (shouldCreateWorktree && !activeThreadBranch) { + if (shouldPrepareWorktree && !activeThreadBranch) { setThreadError(threadIdForSend, "Select a base branch before sending in New worktree mode."); return; } diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index 7c5ea91f043b..db7bf1d09cdc 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -28,6 +28,20 @@ Log in with Codex normally: codex login ``` +## Existing Codex Threads + +T3 Code discovers non-archived conversations from each enabled Codex home and adds them to the +thread list automatically. Discovery runs when the environment becomes active, when provider +settings change, and periodically while T3 Code is running, so recent Codex conversations should +appear within about 30 seconds. + +Opening one of these threads continues the original Codex conversation. T3 Code stores the thread +metadata and continuation link, but it does not copy the existing message history into T3. If you +delete an imported thread from T3 Code, discovery will not add it again. + +Providers that share the same `CODEX_HOME path` also share one discovered thread catalog, so the +same conversation is not added twice for work and personal accounts. + ## I Want Work And Personal Codex Accounts Use one real Codex home and one shadow home. From 9bf0a9fa7b8875058b44d731d8112c3ae775ca8c Mon Sep 17 00:00:00 2001 From: seeb1337 <63622047+seeb1337@users.noreply.github.com> Date: Mon, 10 Aug 2026 00:18:52 +0100 Subject: [PATCH 3/3] fix(codex): preserve concurrent thread renames --- .../decider.titleRegeneration.test.ts | 42 +++++++++++++++++++ apps/server/src/orchestration/decider.ts | 10 ++++- .../Layers/ProviderThreadDiscovery.test.ts | 4 ++ .../Layers/ProviderThreadDiscovery.ts | 4 +- packages/contracts/src/orchestration.ts | 1 + 5 files changed, 58 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration/decider.titleRegeneration.test.ts b/apps/server/src/orchestration/decider.titleRegeneration.test.ts index b29c8ffda676..093d90e2739a 100644 --- a/apps/server/src/orchestration/decider.titleRegeneration.test.ts +++ b/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({ diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index a48bb29e154b..4ca969db8507 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/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 diff --git a/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts b/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts index 26a804d8a8d4..dbcc47b2de16 100644 --- a/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts +++ b/apps/server/src/provider/Layers/ProviderThreadDiscovery.test.ts @@ -308,6 +308,10 @@ it.effect("refreshes imported metadata without replacing the projected session s commands[0]?.type === "thread.meta.update" && commands[0].title, "Fresh Codex title", ); + NodeAssert.equal( + commands[0]?.type === "thread.meta.update" && commands[0].expectedTitle, + originalTitle, + ); NodeAssert.equal(bindings[0]?.status, undefined); NodeAssert.equal(bindings[0]?.runtimeMode, undefined); NodeAssert.equal(readModel.threads[0]?.title, originalTitle); diff --git a/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts b/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts index c9b821dd910a..99e21e342748 100644 --- a/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts +++ b/apps/server/src/provider/Layers/ProviderThreadDiscovery.ts @@ -298,7 +298,9 @@ export const synchronizeDiscoveredProviderThreads = Effect.fn( discovered.updatedAt, ), threadId: linkedThread.id, - ...(providerOwnsTitle && title !== metadata.providerTitle ? { title } : {}), + ...(providerOwnsTitle && title !== metadata.providerTitle + ? { title, expectedTitle: metadata.providerTitle } + : {}), }); if (providerOwnsTitle && title !== metadata.providerTitle) { linkedThread.title = title; diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 35fef721efa7..ed6a6a41ceff 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -753,6 +753,7 @@ const ThreadMetaUpdateCommand = Schema.Struct({ commandId: CommandId, threadId: ThreadId, title: Schema.optional(TrimmedNonEmptyString), + expectedTitle: Schema.optional(TrimmedNonEmptyString), regenerateTitle: Schema.optional(Schema.Literal(true)), modelSelection: Schema.optional(ModelSelection), branch: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)),