diff --git a/apps/server/src/orchestration/decider.import.test.ts b/apps/server/src/orchestration/decider.import.test.ts new file mode 100644 index 000000000000..39e37593b6d0 --- /dev/null +++ b/apps/server/src/orchestration/decider.import.test.ts @@ -0,0 +1,114 @@ +import { + CommandId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; + +const NOW = "2026-08-21T03:22:43.000Z"; + +const readModel: OrchestrationReadModel = { + snapshotSequence: 0, + projects: [], + threads: [ + { + id: ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + title: "Imported Codex thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.6-sol", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + pinnedAt: null, + pinOrderKey: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }, + ], + updatedAt: NOW, +}; + +it.layer(NodeServices.layer)("provider history import decider", (it) => { + it.effect("projects an imported assistant message without requesting a provider turn", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.message.import", + commandId: CommandId.make("provider-import:message-1"), + threadId: ThreadId.make("thread-1"), + messageId: MessageId.make("message-1"), + role: "assistant", + text: "Imported response", + turnId: TurnId.make("turn-1"), + createdAt: NOW, + }, + readModel, + }); + + const events = Array.isArray(result) ? result : [result]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("thread.message-sent"); + if (events[0]?.type === "thread.message-sent") { + expect(events[0].payload).toMatchObject({ + role: "assistant", + text: "Imported response", + turnId: "turn-1", + streaming: false, + }); + } + }), + ); + + it.effect("projects imported user history with its provider turn identity", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.message.import", + commandId: CommandId.make("provider-import:user-message-1"), + threadId: ThreadId.make("thread-1"), + messageId: MessageId.make("user-message-1"), + role: "user", + text: "Imported prompt", + turnId: TurnId.make("turn-1"), + createdAt: NOW, + }, + readModel, + }); + + const events = Array.isArray(result) ? result : [result]; + expect(events[0]).toMatchObject({ + type: "thread.message-sent", + payload: { + role: "user", + text: "Imported prompt", + turnId: "turn-1", + streaming: false, + }, + }); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 4f61955fa6aa..7b9f8d280f7d 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1252,6 +1252,33 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.message.import": { + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.message-sent", + payload: { + threadId: command.threadId, + messageId: command.messageId, + role: command.role, + text: command.text, + turnId: command.turnId, + streaming: false, + createdAt: command.createdAt, + updatedAt: command.createdAt, + }, + }; + } + case "thread.message.assistant.complete": { yield* requireThread({ readModel, diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index bc48f94b3866..c81a82bfacac 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -64,6 +64,7 @@ import { } from "./CodexSessionRuntime.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; import { resolveCodexLaunchArgs } from "./codexLaunchArgs.ts"; +import { discoverCodexThreads } from "./CodexThreadDiscovery.ts"; const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError); const isCodexAppServerTransportError = Schema.is(CodexErrors.CodexAppServerTransportError); const isCodexSessionRuntimeThreadIdMissingError = Schema.is( @@ -1866,6 +1867,25 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( })), ); + const discoverPersistedThreads: NonNullable = ( + input, + ) => + Effect.scoped( + discoverCodexThreads(codexConfig, options?.environment, input).pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, childProcessSpawner), + Effect.timeout("2 minutes"), + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/list", + detail: cause.message, + cause, + }), + ), + ), + ); + const rollbackThread: CodexAdapterShape["rollbackThread"] = (threadId, numTurns) => { if (!Number.isInteger(numTurns) || numTurns < 1) { return Effect.fail( @@ -1988,6 +2008,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( sendTurn, interruptTurn, readThread, + discoverPersistedThreads, rollbackThread, uploadFeedback, respondToRequest, diff --git a/apps/server/src/provider/Layers/CodexThreadDiscovery.test.ts b/apps/server/src/provider/Layers/CodexThreadDiscovery.test.ts new file mode 100644 index 000000000000..9ab421a4288c --- /dev/null +++ b/apps/server/src/provider/Layers/CodexThreadDiscovery.test.ts @@ -0,0 +1,200 @@ +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import type * as EffectCodexSchema from "effect-codex-app-server/schema"; + +import { + readCodexThreadSnapshots, + selectCodexThreadsForRead, + toPersistedThread, +} from "./CodexThreadDiscovery.ts"; + +type ReadThread = EffectCodexSchema.V2ThreadReadResponse["thread"]; +type ListedThread = EffectCodexSchema.V2ThreadListResponse["data"][number]; + +function makeReadThread(overrides: Record = {}): ReadThread { + return { + id: "provider-thread-1", + cwd: "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/work/project", + name: null, + preview: "Preview title", + createdAt: 1_777_777_700, + updatedAt: 1_777_777_760, + source: "cli", + status: { type: "idle" }, + threadSource: null, + cliVersion: "1.2.3", + modelProvider: "openai", + gitInfo: null, + turns: [], + ...overrides, + } as unknown as ReadThread; +} + +function makeListedThread(id: string, overrides: Record = {}): ListedThread { + return { + ...makeReadThread({ id, turns: undefined }), + ephemeral: false, + ...overrides, + } as unknown as ListedThread; +} + +it("maps persisted Codex turns to ordered user and assistant history", () => { + const thread = { + id: "provider-thread-1", + cwd: "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/work/project", + name: null, + preview: "Investigate the importer\nMore context", + createdAt: 1_777_777_700, + updatedAt: 1_777_777_760, + source: "cli", + status: { type: "idle" }, + threadSource: null, + cliVersion: "1.2.3", + modelProvider: "openai", + gitInfo: null, + turns: [ + { + id: "turn-1", + startedAt: 1_777_777_710, + completedAt: 1_777_777_750, + status: "completed", + items: [ + { + id: "user-1", + type: "userMessage", + content: [ + { type: "text", text: "Investigate the importer" }, + { type: "localImage", path: "/tmp/screenshot.png" }, + ], + }, + { + id: "tool-1", + type: "commandExecution", + command: "pwd", + commandActions: [], + cwd: "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/work/project", + status: "completed", + }, + { + id: "assistant-1", + type: "agentMessage", + text: "The importer is ready.", + }, + ], + }, + ], + } as unknown as EffectCodexSchema.V2ThreadReadResponse["thread"]; + + expect(toPersistedThread(thread)).toMatchObject({ + providerThreadId: "provider-thread-1", + cwd: "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/work/project", + title: "Investigate the importer", + sourceMetadata: { source: "cli", cliVersion: "1.2.3" }, + messages: [ + { + id: "user-1", + sourceOrdinal: 0, + role: "user", + text: "Investigate the importer", + turnId: "turn-1", + }, + { + id: "assistant-1", + sourceOrdinal: 1, + role: "assistant", + text: "The importer is ready.", + turnId: "turn-1", + }, + ], + }); +}); + +it("prefers a persisted name and falls back when all title text is blank", () => { + expect(toPersistedThread(makeReadThread({ name: " Named thread " })).title).toBe( + "Named thread", + ); + expect(toPersistedThread(makeReadThread({ name: " ", preview: "\n" })).title).toBe( + "Imported Codex thread", + ); +}); + +it("omits partial and empty messages while applying timestamp fallbacks", () => { + const persisted = toPersistedThread( + makeReadThread({ + status: { type: "inProgress" }, + turns: [ + { + id: "turn-active", + startedAt: 1_777_777_710, + completedAt: null, + status: "inProgress", + items: [{ id: "active-agent", type: "agentMessage", text: "Partial" }], + }, + { + id: "turn-complete", + startedAt: null, + completedAt: null, + status: "completed", + items: [ + { + id: "empty-user", + type: "userMessage", + content: [{ type: "text", text: " " }], + }, + { id: "empty-agent", type: "agentMessage", text: "" }, + { id: "agent", type: "agentMessage", text: "Complete" }, + ], + }, + ], + }), + ); + + expect(persisted.messages).toEqual([ + expect.objectContaining({ + id: "agent", + text: "Complete", + createdAt: "2026-05-03T03:09:20.000Z", + }), + ]); + expect(persisted.discoveryCursor).toBe("2026-05-03T03:09:20.000Z:inProgress"); +}); + +it("selects only user-visible threads whose discovery cursor changed", () => { + const unchanged = makeListedThread("unchanged"); + const statusChanged = makeListedThread("status-changed", { status: { type: "idle" } }); + const selected = selectCodexThreadsForRead( + [ + unchanged, + statusChanged, + makeListedThread("new"), + makeListedThread("ephemeral", { ephemeral: true }), + makeListedThread("subagent", { source: { subAgent: { threadId: "parent" } } }), + makeListedThread("memory", { threadSource: "memory_consolidation" }), + makeListedThread("native"), + ], + { + excludeProviderThreadIds: new Set(["native"]), + cursorByProviderThreadId: new Map([ + ["unchanged", "2026-05-03T03:09:20.000Z:idle"], + ["status-changed", "2026-05-03T03:09:20.000Z:active"], + ]), + }, + ); + + expect(selected.map((thread) => thread.id)).toEqual(["status-changed", "new"]); +}); + +it.effect("keeps readable threads when a sibling disappears between list and read", () => + Effect.gen(function* () { + const threads = [makeListedThread("deleted"), makeListedThread("healthy")]; + const discovered = yield* readCodexThreadSnapshots(threads, (threadId) => + threadId === "deleted" + ? Effect.fail({ _tag: "ThreadReadFailed" as const }) + : Effect.succeed({ + thread: makeReadThread({ id: threadId, preview: "Healthy thread" }), + } as EffectCodexSchema.V2ThreadReadResponse), + ); + + expect(discovered.map((thread) => thread.providerThreadId)).toEqual(["healthy"]); + }), +); diff --git a/apps/server/src/provider/Layers/CodexThreadDiscovery.ts b/apps/server/src/provider/Layers/CodexThreadDiscovery.ts new file mode 100644 index 000000000000..5b88ad9a2a0e --- /dev/null +++ b/apps/server/src/provider/Layers/CodexThreadDiscovery.ts @@ -0,0 +1,224 @@ +import { type CodexSettings, TurnId } from "@t3tools/contracts"; +import type { + ProviderPersistedThread, + ProviderPersistedThreadDiscoveryInput, +} from "../Services/ProviderAdapter.ts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +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 * as CodexErrors from "effect-codex-app-server/errors"; +import type * as EffectCodexSchema from "effect-codex-app-server/schema"; + +import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import { expandHomePath } from "../../pathExpansion.ts"; +import { buildCodexInitializeParams } from "./CodexProvider.ts"; +import { codexSessionAppServerArgs, resolveCodexLaunchArgs } from "./codexLaunchArgs.ts"; + +const PAGE_SIZE = 100; +const THREAD_READ_CONCURRENCY = 4; +const FORCE_KILL_AFTER = "2 seconds" as const; + +function diagnosticErrorType(cause: unknown): string { + if (cause instanceof Error) return cause.name; + if (typeof cause === "object" && cause !== null && "_tag" in cause) { + return String(cause._tag); + } + return typeof cause; +} + +function unixSecondsToIso(seconds: number): string { + return DateTime.formatIso(DateTime.makeUnsafe(seconds * 1_000)); +} + +function discoveryCursorForThread( + thread: EffectCodexSchema.V2ThreadListResponse["data"][number], +): string { + return `${unixSecondsToIso(thread.updatedAt)}:${thread.status.type}`; +} + +export function selectCodexThreadsForRead( + threads: ReadonlyArray, + discoveryInput?: ProviderPersistedThreadDiscoveryInput, +): ReadonlyArray { + return threads.filter((thread) => { + if ( + thread.ephemeral || + (typeof thread.source === "object" && "subAgent" in thread.source) || + thread.threadSource === "memory_consolidation" || + discoveryInput?.excludeProviderThreadIds.has(thread.id) === true + ) { + return false; + } + const knownCursor = discoveryInput?.cursorByProviderThreadId.get(thread.id); + return knownCursor !== discoveryCursorForThread(thread); + }); +} + +function titleForThread(thread: EffectCodexSchema.V2ThreadReadResponse["thread"]): string { + const name = thread.name?.trim(); + if (name) return name; + const preview = thread.preview.trim().split("\n", 1)[0]?.trim(); + return preview || "Imported Codex thread"; +} + +function messagesForThread( + thread: EffectCodexSchema.V2ThreadReadResponse["thread"], +): ProviderPersistedThread["messages"] { + const messages: Array = []; + for (const turn of thread.turns) { + if (turn.status === "inProgress") continue; + const startedAt = unixSecondsToIso(turn.startedAt ?? thread.createdAt); + const completedAt = unixSecondsToIso(turn.completedAt ?? turn.startedAt ?? thread.updatedAt); + for (const item of turn.items) { + if (item.type === "userMessage") { + const text = item.content + .filter((content) => content.type === "text") + .map((content) => content.text) + .join("\n") + .trim(); + if (text) { + messages.push({ + id: item.id, + sourceOrdinal: messages.length, + role: "user", + text, + turnId: TurnId.make(turn.id), + createdAt: startedAt, + }); + } + } + if (item.type === "agentMessage" && item.text.length > 0) { + messages.push({ + id: item.id, + sourceOrdinal: messages.length, + role: "assistant", + text: item.text, + turnId: TurnId.make(turn.id), + createdAt: completedAt, + }); + } + } + } + return messages; +} + +export function toPersistedThread( + thread: EffectCodexSchema.V2ThreadReadResponse["thread"], +): ProviderPersistedThread { + return { + providerThreadId: thread.id, + cwd: thread.cwd, + title: titleForThread(thread), + createdAt: unixSecondsToIso(thread.createdAt), + updatedAt: unixSecondsToIso(thread.updatedAt), + discoveryCursor: discoveryCursorForThread(thread), + sourceMetadata: { + source: thread.source, + threadSource: thread.threadSource ?? null, + status: thread.status, + sessionId: thread.sessionId, + forkedFromId: thread.forkedFromId ?? null, + parentThreadId: thread.parentThreadId ?? null, + cliVersion: thread.cliVersion, + modelProvider: thread.modelProvider, + gitInfo: thread.gitInfo ?? null, + }, + messages: messagesForThread(thread), + }; +} + +export function readCodexThreadSnapshots( + threads: ReadonlyArray, + readThread: (threadId: string) => Effect.Effect, +): Effect.Effect> { + return Effect.forEach( + threads, + (thread) => + readThread(thread.id).pipe( + Effect.map((response) => Option.some(toPersistedThread(response.thread))), + Effect.catch((cause) => + Effect.logWarning("skipped unreadable persisted Codex thread", { + providerThreadId: thread.id, + errorType: diagnosticErrorType(cause), + }).pipe(Effect.as(Option.none())), + ), + ), + { concurrency: THREAD_READ_CONCURRENCY }, + ).pipe(Effect.map((threads) => threads.filter(Option.isSome).map((thread) => thread.value))); +} + +export const discoverCodexThreads = Effect.fn("discoverCodexThreads")(function* ( + config: CodexSettings, + environment?: NodeJS.ProcessEnv, + discoveryInput?: ProviderPersistedThreadDiscoveryInput, +) { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const runtimeScope = yield* Scope.Scope; + const resolvedHomePath = config.homePath ? expandHomePath(config.homePath) : undefined; + const env = { + ...environment, + ...(resolvedHomePath ? { CODEX_HOME: resolvedHomePath } : {}), + }; + const extendEnv = environment === undefined; + const launchArgs = resolveCodexLaunchArgs(config.launchArgs, environment); + const appServerArgs = codexSessionAppServerArgs(undefined, launchArgs); + const spawnCommand = yield* resolveSpawnCommand(config.binaryPath, appServerArgs, { + env, + extendEnv, + }); + const child = yield* spawner + .spawn( + ChildProcess.make(spawnCommand.command, spawnCommand.args, { + cwd: process.cwd(), + env, + extendEnv, + forceKillAfter: FORCE_KILL_AFTER, + shell: spawnCommand.shell, + }), + ) + .pipe( + Effect.provideService(Scope.Scope, runtimeScope), + Effect.mapError( + (cause) => + new CodexErrors.CodexAppServerSpawnError({ + command: `${config.binaryPath} app-server`, + cause, + }), + ), + ); + + yield* child.stderr.pipe(Stream.decodeText(), Stream.runDrain, Effect.forkIn(runtimeScope)); + + 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), + ); + yield* client.request("initialize", buildCodexInitializeParams()); + yield* client.notify("initialized", undefined); + + const listed: Array = []; + let cursor: string | null | undefined; + do { + const page = yield* client.request("thread/list", { + ...(cursor !== undefined ? { cursor } : {}), + limit: PAGE_SIZE, + sortKey: "updated_at", + sortDirection: "desc", + }); + listed.push(...page.data); + cursor = page.nextCursor; + } while (cursor); + + return yield* readCodexThreadSnapshots( + selectCodexThreadsForRead(listed, discoveryInput), + (threadId) => client.request("thread/read", { threadId, includeTurns: true }), + ); +}); diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index bd89dc4f8812..8f32b5873c57 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -928,6 +928,42 @@ it.effect( ); routing.layer("ProviderServiceLive routing", (it) => { + it.effect("backfills continuation identity while recovering a legacy binding", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const threadId = asThreadId("thread-legacy-continuation"); + + yield* directory.upsert({ + threadId, + provider: CODEX_DRIVER, + providerInstanceId: codexInstanceId, + resumeCursor: { opaque: "legacy-resume" }, + runtimeMode: "full-access", + runtimePayload: { cwd: "/tmp/legacy-continuation" }, + }); + + yield* provider.sendTurn({ + threadId, + input: "resume legacy thread", + attachments: [], + }); + + const binding = yield* directory.getBinding(threadId); + assert.equal(Option.isSome(binding), true); + if (Option.isSome(binding)) { + const payload = binding.value.runtimePayload; + assert.equal(payload !== null && typeof payload === "object", true); + if (payload !== null && typeof payload === "object" && !Array.isArray(payload)) { + assert.equal( + "continuationKey" in payload ? payload.continuationKey : undefined, + "codex:instance:codex", + ); + } + } + }), + ); + it.effect("routes provider operations and rollback conversation", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; @@ -1491,12 +1527,14 @@ routing.layer("ProviderServiceLive routing", (it) => { const runtimePayload = payload as { cwd: string; model: string | null; + continuationKey: string; activeTurnId: string | null; lastError: string | null; lastRuntimeEvent: string | null; }; assert.equal(runtimePayload.cwd, session.cwd); assert.equal(runtimePayload.model, null); + assert.equal(runtimePayload.continuationKey, "codex:instance:codex"); assert.equal(runtimePayload.activeTurnId, `turn-${String(session.threadId)}`); assert.equal(runtimePayload.lastError, null); assert.equal(runtimePayload.lastRuntimeEvent, "provider.sendTurn"); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index b8cd0df539ac..657cf402bc05 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -136,6 +136,7 @@ function toRuntimePayloadFromSession( session: ProviderSession, extra?: { readonly modelSelection?: unknown; + readonly continuationKey?: string; readonly lastRuntimeEvent?: string; readonly lastRuntimeEventAt?: string; }, @@ -146,6 +147,7 @@ function toRuntimePayloadFromSession( activeTurnId: session.activeTurnId ?? null, lastError: session.lastError ?? null, ...(extra?.modelSelection !== undefined ? { modelSelection: extra.modelSelection } : {}), + ...(extra?.continuationKey !== undefined ? { continuationKey: extra.continuationKey } : {}), ...(extra?.lastRuntimeEvent !== undefined ? { lastRuntimeEvent: extra.lastRuntimeEvent } : {}), ...(extra?.lastRuntimeEventAt !== undefined ? { lastRuntimeEventAt: extra.lastRuntimeEventAt } @@ -317,6 +319,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( threadId: ThreadId, extra?: { readonly modelSelection?: unknown; + readonly continuationKey?: string; readonly lastRuntimeEvent?: string; readonly lastRuntimeEventAt?: string; }, @@ -326,6 +329,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( "ProviderService.upsertSessionBinding", session, ); + const instanceInfo = yield* registry.getInstanceInfo(providerInstanceId); yield* directory.upsert({ threadId, provider: session.provider, @@ -333,7 +337,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( runtimeMode: session.runtimeMode, status: toRuntimeStatus(session), ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}), - runtimePayload: toRuntimePayloadFromSession(session, extra), + runtimePayload: toRuntimePayloadFromSession(session, { + ...extra, + continuationKey: instanceInfo.continuationIdentity.continuationKey, + }), }); }); diff --git a/apps/server/src/provider/Layers/ProviderThreadReconciler.test.ts b/apps/server/src/provider/Layers/ProviderThreadReconciler.test.ts new file mode 100644 index 000000000000..e78b23fe79d0 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderThreadReconciler.test.ts @@ -0,0 +1,556 @@ +import { + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationCommand, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +import type { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; +import type { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { ProviderInstance } from "../ProviderDriver.ts"; +import type { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import type { ProviderPersistedThread } from "../Services/ProviderAdapter.ts"; +import { + groupPersistedThreadDiscoveryCandidates, + providerThreadDiscoveryExclusions, + continuationIdentityDigest, + recoverReconciliationCause, + reconcilePersistedThread, + resolvePersistedContinuationKey, +} from "./ProviderThreadReconciler.ts"; + +const CONTINUATION_KEY = "codex:home:/work/.codex"; +const CONTINUATION_IDENTITY = "8da0416c4575b56e6f63b88a"; +const PROVIDER_THREAD_ID = "0198cb4a-thread"; +const providerIdentity = `${CONTINUATION_KEY.length}:${CONTINUATION_KEY}${PROVIDER_THREAD_ID}`; +const importedThreadId = ThreadId.make(`imported:${CONTINUATION_IDENTITY}:${PROVIDER_THREAD_ID}`); + +const instance = { + instanceId: ProviderInstanceId.make("codex-work"), + driverKind: ProviderDriverKind.make("codex"), + continuationIdentity: { + driverKind: ProviderDriverKind.make("codex"), + continuationKey: CONTINUATION_KEY, + }, +} as ProviderInstance; + +const persistedThread: ProviderPersistedThread = { + providerThreadId: PROVIDER_THREAD_ID, + cwd: "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/work/external", + title: "External work", + createdAt: "2026-08-21T03:22:43.000Z", + updatedAt: "2026-08-21T03:24:43.000Z", + discoveryCursor: "2026-08-21T03:24:43.000Z:idle", + sourceMetadata: { source: "cli" }, + messages: [ + { + id: "user-item", + sourceOrdinal: 0, + role: "user", + text: "Investigate this", + turnId: TurnId.make("turn-1"), + createdAt: "2026-08-21T03:24:43.000Z", + }, + { + id: "assistant-item", + sourceOrdinal: 1, + role: "assistant", + text: "Done", + turnId: TurnId.make("turn-1"), + createdAt: "2026-08-21T03:24:43.000Z", + }, + ], +}; + +it("keeps deterministic fallback candidates for each shared Codex home", () => { + const makeDiscoveryInstance = (instanceId: string, continuationKey: string) => + ({ + ...instance, + instanceId: ProviderInstanceId.make(instanceId), + continuationIdentity: { + driverKind: ProviderDriverKind.make("codex"), + continuationKey, + }, + enabled: true, + adapter: { discoverPersistedThreads: () => Effect.succeed([]) }, + }) as unknown as ProviderInstance; + + const groups = groupPersistedThreadDiscoveryCandidates([ + makeDiscoveryInstance("codex-z", CONTINUATION_KEY), + makeDiscoveryInstance("codex-a", CONTINUATION_KEY), + makeDiscoveryInstance("codex-other", "codex:home:/other/.codex"), + ]); + + expect(groups.map((group) => group.map((candidate) => candidate.instanceId))).toEqual([ + ["codex-other"], + ["codex-a", "codex-z"], + ]); +}); + +it("recovers a continuation key after the owning instance is removed", () => { + expect( + resolvePersistedContinuationKey( + "codex-removed", + { continuationKey: CONTINUATION_KEY }, + new Map([["codex-next", CONTINUATION_KEY]]), + ), + ).toBe(CONTINUATION_KEY); +}); + +it("uses opaque continuation identities and prefers persisted ownership", () => { + expect(continuationIdentityDigest(CONTINUATION_KEY)).toBe(CONTINUATION_IDENTITY); + expect(continuationIdentityDigest(CONTINUATION_KEY)).not.toContain("work"); + expect( + resolvePersistedContinuationKey( + "codex-reconfigured", + { continuationKey: CONTINUATION_KEY }, + new Map([["codex-reconfigured", "codex:home:/different/.codex"]]), + ), + ).toBe(CONTINUATION_KEY); +}); + +it("excludes unresolved legacy native threads from every continuation group", () => { + expect( + Array.from( + providerThreadDiscoveryExclusions( + new Set(["legacy-native-thread"]), + new Set(["current-native-thread"]), + ), + ), + ).toEqual(["legacy-native-thread", "current-native-thread"]); +}); + +it.effect("does not recover a reconciliation interruption", () => + Effect.gen(function* () { + const exit = yield* Effect.exit( + recoverReconciliationCause(Cause.interrupt(), "should not log", {}, undefined), + ); + + expect(exit._tag).toBe("Failure"); + if (exit._tag === "Failure") { + expect(Cause.hasInterrupts(exit.cause)).toBe(true); + } + }), +); + +it.effect("imports an unmatched Codex thread into the unassigned project", () => + Effect.gen(function* () { + const commands: OrchestrationCommand[] = []; + const bindings: Array[0]> = []; + const engine = { + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length }; + }), + readEvents: () => Stream.empty, + streamDomainEvents: Stream.empty, + latestSequence: Effect.succeed(0), + } as OrchestrationEngineService["Service"]; + const directory = { + upsert: (binding: Parameters[0]) => + Effect.sync(() => { + bindings.push(binding); + }), + } as unknown as ProviderSessionDirectory["Service"]; + const snapshots = { + getThreadShellById: () => Effect.succeed(Option.none()), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), + getProjectShellById: () => Effect.succeed(Option.none()), + } as unknown as ProjectionSnapshotQuery["Service"]; + const identities = new Map(); + + yield* reconcilePersistedThread({ + instance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: identities, + directory, + snapshots, + engine, + unassignedWorkspaceRoot: "/t3/provider-imports/codex-work", + }); + + expect(commands.map((command) => command.type)).toEqual([ + "project.create", + "thread.create", + "thread.message.import", + "thread.message.import", + ]); + expect(commands[1]).toMatchObject({ + type: "thread.create", + title: "External work", + worktreePath: "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/work/external", + }); + expect(commands[2]).toMatchObject({ + type: "thread.message.import", + role: "user", + turnId: "turn-1", + }); + expect(commands[3]).toMatchObject({ + type: "thread.message.import", + role: "assistant", + turnId: "turn-1", + }); + const importedMessageIds = commands + .filter((command) => command.type === "thread.message.import") + .map((command) => command.messageId); + expect(importedMessageIds.toSorted()).toEqual(importedMessageIds); + expect(bindings).toHaveLength(1); + expect(bindings[0]).toMatchObject({ + providerInstanceId: "codex-work", + runtimeMode: "full-access", + status: "stopped", + resumeCursor: { threadId: "0198cb4a-thread" }, + runtimePayload: { + imported: true, + providerDiscoveryCursor: "2026-08-21T03:24:43.000Z:idle", + }, + }); + }), +); + +it.effect("reuses the group project when discovery ownership changes", () => + Effect.gen(function* () { + const commands: OrchestrationCommand[] = []; + const handoffInstance = { + ...instance, + instanceId: ProviderInstanceId.make("codex-next"), + } as ProviderInstance; + + yield* reconcilePersistedThread({ + instance: handoffInstance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: new Map(), + directory: { + upsert: () => Effect.void, + } as unknown as ProviderSessionDirectory["Service"], + snapshots: { + getThreadShellById: () => Effect.succeed(Option.none()), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), + getProjectShellById: () => + Effect.succeed( + Option.some({ + defaultModelSelection: { + instanceId: ProviderInstanceId.make("codex-work"), + model: "gpt-5.6-sol", + }, + }), + ), + } as unknown as ProjectionSnapshotQuery["Service"], + engine: { + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length }; + }), + } as unknown as OrchestrationEngineService["Service"], + unassignedWorkspaceRoot: "/t3/provider-imports/shared-home", + }); + + expect(commands.map((command) => command.type)).toEqual([ + "project.meta.update", + "thread.create", + "thread.message.import", + "thread.message.import", + ]); + expect(commands[0]).toMatchObject({ + type: "project.meta.update", + defaultModelSelection: { instanceId: "codex-next", model: "gpt-5.6-sol" }, + }); + }), +); + +it.effect("migrates an existing imported thread when discovery ownership changes", () => + Effect.gen(function* () { + const commands: OrchestrationCommand[] = []; + const bindings: Array[0]> = []; + const handoffInstance = { + ...instance, + instanceId: ProviderInstanceId.make("codex-next"), + } as ProviderInstance; + + yield* reconcilePersistedThread({ + instance: handoffInstance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: new Map([[providerIdentity, importedThreadId]]), + directory: { + upsert: (binding: Parameters[0]) => + Effect.sync(() => { + bindings.push(binding); + }), + } as unknown as ProviderSessionDirectory["Service"], + snapshots: { + getThreadShellById: () => + Effect.succeed( + Option.some({ + id: importedThreadId, + modelSelection: { + instanceId: ProviderInstanceId.make("codex-work"), + model: "gpt-5.6-sol", + }, + }), + ), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), + getProjectShellById: () => + Effect.succeed( + Option.some({ + defaultModelSelection: { + instanceId: ProviderInstanceId.make("codex-work"), + model: "gpt-5.6-sol", + }, + }), + ), + getThreadDetailById: () => + Effect.succeed( + Option.some({ + messages: [ + { id: "existing-user", role: "user", text: "Investigate this" }, + { id: "existing-assistant", role: "assistant", text: "Done" }, + ], + }), + ), + } as unknown as ProjectionSnapshotQuery["Service"], + engine: { + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length }; + }), + } as unknown as OrchestrationEngineService["Service"], + unassignedWorkspaceRoot: "/t3/provider-imports/shared-home", + }); + + expect(commands.map((command) => command.type)).toEqual([ + "project.meta.update", + "thread.meta.update", + ]); + expect(bindings[0]).toMatchObject({ + providerInstanceId: "codex-next", + runtimePayload: { + modelSelection: { instanceId: "codex-next", model: "gpt-5.6-sol" }, + }, + }); + }), +); + +it.effect("imports a new thread directly into its matching project", () => + Effect.gen(function* () { + const commands: OrchestrationCommand[] = []; + const engine = { + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length }; + }), + } as unknown as OrchestrationEngineService["Service"]; + + yield* reconcilePersistedThread({ + instance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: new Map(), + directory: { + upsert: () => Effect.void, + } as unknown as ProviderSessionDirectory["Service"], + snapshots: { + getThreadShellById: () => Effect.succeed(Option.none()), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.some({ id: "project-1" })), + } as unknown as ProjectionSnapshotQuery["Service"], + engine, + unassignedWorkspaceRoot: "/unused", + }); + + expect(commands.map((command) => command.type)).toEqual([ + "thread.create", + "thread.message.import", + "thread.message.import", + ]); + expect(commands[0]).toMatchObject({ + type: "thread.create", + projectId: "project-1", + worktreePath: null, + }); + }), +); + +it.effect("does not advance the discovery watermark after a partial message import", () => + Effect.gen(function* () { + const bindings: Array[0]> = []; + const exit = yield* Effect.exit( + reconcilePersistedThread({ + instance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: new Map(), + directory: { + upsert: (binding: Parameters[0]) => + Effect.sync(() => { + bindings.push(binding); + }), + } as unknown as ProviderSessionDirectory["Service"], + snapshots: { + getThreadShellById: () => Effect.succeed(Option.none()), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.some({ id: "project-1" })), + } as unknown as ProjectionSnapshotQuery["Service"], + engine: { + dispatch: (command: OrchestrationCommand) => + command.type === "thread.message.import" && command.role === "assistant" + ? Effect.die("message import failed") + : Effect.succeed({ sequence: 1 }), + } as unknown as OrchestrationEngineService["Service"], + unassignedWorkspaceRoot: "/unused", + }), + ); + + expect(exit._tag).toBe("Failure"); + expect(bindings).toEqual([]); + }), +); + +it.effect("does not duplicate an identity owned by a compatible instance", () => + Effect.gen(function* () { + const identities = new Map([ + [providerIdentity, ThreadId.make("native-t3-thread")], + ]); + let touched = false; + const failIfTouched = () => + Effect.sync(() => { + touched = true; + return Option.none(); + }); + + yield* reconcilePersistedThread({ + instance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: identities, + directory: { + upsert: () => Effect.sync(() => void (touched = true)), + } as unknown as ProviderSessionDirectory["Service"], + snapshots: { + getThreadShellById: failIfTouched, + getActiveProjectByWorkspaceRoot: failIfTouched, + } as unknown as ProjectionSnapshotQuery["Service"], + engine: { + dispatch: () => Effect.sync(() => void (touched = true)) as never, + } as unknown as OrchestrationEngineService["Service"], + unassignedWorkspaceRoot: "/unused", + }); + + expect(touched).toBe(false); + }), +); + +it.effect("does not re-import turns that T3 already projected under different message ids", () => + Effect.gen(function* () { + const commands: OrchestrationCommand[] = []; + const identities = new Map([[providerIdentity, importedThreadId]]); + + yield* reconcilePersistedThread({ + instance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: identities, + directory: { + upsert: () => Effect.void, + } as unknown as ProviderSessionDirectory["Service"], + snapshots: { + getThreadShellById: () => + Effect.succeed( + Option.some({ + id: importedThreadId, + modelSelection: { + instanceId: instance.instanceId, + model: "gpt-5.6-sol", + }, + }), + ), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.some({ id: "project-1" })), + getThreadDetailById: () => + Effect.succeed( + Option.some({ + messages: [ + { id: "t3-user-id", role: "user", text: "Investigate this" }, + { id: "assistant:user-item", role: "assistant", text: "Done" }, + ], + }), + ), + } as unknown as ProjectionSnapshotQuery["Service"], + engine: { + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length }; + }), + } as unknown as OrchestrationEngineService["Service"], + unassignedWorkspaceRoot: "/unused", + }); + + expect(commands).toEqual([]); + }), +); + +it.effect("appends only messages missing after a deterministic imported id", () => + Effect.gen(function* () { + const commands: OrchestrationCommand[] = []; + + yield* reconcilePersistedThread({ + instance, + thread: persistedThread, + model: "gpt-5.6-sol", + threadByProviderIdentity: new Map([[providerIdentity, importedThreadId]]), + directory: { + upsert: () => Effect.void, + } as unknown as ProviderSessionDirectory["Service"], + snapshots: { + getThreadShellById: () => + Effect.succeed( + Option.some({ + id: importedThreadId, + modelSelection: { + instanceId: instance.instanceId, + model: "gpt-5.6-sol", + }, + }), + ), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.some({ id: "project-1" })), + getThreadDetailById: () => + Effect.succeed( + Option.some({ + messages: [ + { + id: `provider:${CONTINUATION_IDENTITY}:${PROVIDER_THREAD_ID}:message:0000000000:user-item`, + role: "user", + text: "Investigate this", + }, + ], + }), + ), + } as unknown as ProjectionSnapshotQuery["Service"], + engine: { + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + commands.push(command); + return { sequence: commands.length }; + }), + } as unknown as OrchestrationEngineService["Service"], + unassignedWorkspaceRoot: "/unused", + }); + + expect(commands).toHaveLength(1); + expect(commands[0]).toMatchObject({ + type: "thread.message.import", + role: "assistant", + text: "Done", + }); + }), +); diff --git a/apps/server/src/provider/Layers/ProviderThreadReconciler.ts b/apps/server/src/provider/Layers/ProviderThreadReconciler.ts new file mode 100644 index 000000000000..4c275764cc35 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderThreadReconciler.ts @@ -0,0 +1,550 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeCrypto from "node:crypto"; + +import { + CommandId, + DEFAULT_MODEL_BY_PROVIDER, + MessageId, + ProjectId, + ProviderDriverKind, + ThreadId, + TurnId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; + +import { ServerConfig } from "../../config.ts"; +import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { ProviderInstance } from "../ProviderDriver.ts"; +import { ProviderInstanceRegistry } from "../Services/ProviderInstanceRegistry.ts"; +import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import type { ProviderPersistedThread } from "../Services/ProviderAdapter.ts"; + +const CODEX = ProviderDriverKind.make("codex"); +const ACTIVE_RECONCILE_INTERVAL = Duration.seconds(30); +const IDLE_RECONCILE_INTERVAL = Duration.minutes(2); +const ResumeCursor = Schema.Struct({ threadId: Schema.String }); +const isResumeCursor = Schema.is(ResumeCursor); +const ImportedRuntimePayload = Schema.Struct({ + continuationKey: Schema.optional(Schema.String), + providerUpdatedAt: Schema.optional(Schema.String), + providerDiscoveryCursor: Schema.optional(Schema.String), +}); +const isImportedRuntimePayload = Schema.is(ImportedRuntimePayload); + +export function continuationIdentityDigest(continuationKey: string): string { + return NodeCrypto.createHash("sha256").update(continuationKey).digest("hex").slice(0, 24); +} + +function diagnosticErrorType(cause: unknown): string { + if (cause instanceof Error) return cause.name; + if (typeof cause === "object" && cause !== null && "_tag" in cause) { + return String(cause._tag); + } + return typeof cause; +} + +export function recoverReconciliationCause( + cause: Cause.Cause, + message: string, + annotations: Record, + fallback: A, +): Effect.Effect { + return Cause.hasInterrupts(cause) + ? Effect.failCause(cause) + : Effect.logWarning(message, annotations).pipe(Effect.as(fallback)); +} + +export function resolvePersistedContinuationKey( + providerInstanceId: string, + runtimePayload: unknown, + continuationKeyByInstanceId: ReadonlyMap, +): string | undefined { + return ( + (isImportedRuntimePayload(runtimePayload) ? runtimePayload.continuationKey : undefined) ?? + continuationKeyByInstanceId.get(providerInstanceId) + ); +} + +export function providerThreadDiscoveryExclusions( + unresolvedNativeProviderThreadIds: ReadonlySet, + continuationProviderThreadIds: ReadonlySet | undefined, +): ReadonlySet { + return new Set([...unresolvedNativeProviderThreadIds, ...(continuationProviderThreadIds ?? [])]); +} + +function providerIdentityKey(continuationKey: string, providerThreadId: string): string { + return `${continuationKey.length}:${continuationKey}${providerThreadId}`; +} + +function importedThreadIdFor(continuationKey: string, providerThreadId: string): ThreadId { + return ThreadId.make( + `imported:${continuationIdentityDigest(continuationKey)}:${providerThreadId}`, + ); +} + +function importedThreadId(instance: ProviderInstance, providerThreadId: string): ThreadId { + return importedThreadIdFor(instance.continuationIdentity.continuationKey, providerThreadId); +} + +function unassignedProjectId(instance: ProviderInstance): ProjectId { + return ProjectId.make( + `provider-imports:${continuationIdentityDigest(instance.continuationIdentity.continuationKey)}`, + ); +} + +function importCommandId(...parts: ReadonlyArray): CommandId { + return CommandId.make(`provider-import:${parts.join(":")}`); +} + +function importedMessageId( + continuationKey: string, + providerThreadId: string, + providerMessageId: string, + sourceOrdinal: number, +): MessageId { + const orderedOrdinal = String(sourceOrdinal).padStart(10, "0"); + return MessageId.make( + `provider:${continuationIdentityDigest(continuationKey)}:${providerThreadId}:message:${orderedOrdinal}:${providerMessageId}`, + ); +} + +export function groupPersistedThreadDiscoveryCandidates( + instances: ReadonlyArray, +): ReadonlyArray> { + const candidatesByContinuationKey = new Map(); + const discoverable = instances + .filter( + (instance) => + instance.enabled && + instance.driverKind === CODEX && + instance.adapter.discoverPersistedThreads !== undefined, + ) + .sort((left, right) => left.instanceId.localeCompare(right.instanceId)); + for (const instance of discoverable) { + const continuationKey = instance.continuationIdentity.continuationKey; + const candidates = candidatesByContinuationKey.get(continuationKey) ?? []; + candidates.push(instance); + candidatesByContinuationKey.set(continuationKey, candidates); + } + return Array.from(candidatesByContinuationKey.entries()) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([, candidates]) => candidates); +} + +export const reconcilePersistedProviderThreads = Effect.fn("reconcilePersistedProviderThreads")( + function* () { + const registry = yield* ProviderInstanceRegistry; + const directory = yield* ProviderSessionDirectory; + const snapshots = yield* ProjectionSnapshotQuery; + const engine = yield* OrchestrationEngineService; + const serverConfig = yield* ServerConfig; + const path = yield* Path.Path; + const instances = yield* registry.listInstances; + const bindings = yield* directory.listBindings(); + const discoveryCandidateGroups = groupPersistedThreadDiscoveryCandidates(instances); + const continuationKeyByInstanceId = new Map( + instances + .filter((instance) => instance.driverKind === CODEX) + .map( + (instance) => + [instance.instanceId, instance.continuationIdentity.continuationKey] as const, + ), + ); + const threadByProviderIdentity = new Map(); + const excludedThreadIdsByContinuation = new Map>(); + const cursorByThreadIdByContinuation = new Map>(); + const importedOwnerInstanceIdsByContinuation = new Map>(); + const unresolvedNativeProviderThreadIds = new Set(); + + for (const binding of bindings) { + if ( + binding.provider !== CODEX || + !binding.providerInstanceId || + !isResumeCursor(binding.resumeCursor) + ) { + continue; + } + const continuationKey = resolvePersistedContinuationKey( + binding.providerInstanceId, + binding.runtimePayload, + continuationKeyByInstanceId, + ); + if (!continuationKey) { + unresolvedNativeProviderThreadIds.add(binding.resumeCursor.threadId); + continue; + } + threadByProviderIdentity.set( + providerIdentityKey(continuationKey, binding.resumeCursor.threadId), + binding.threadId, + ); + const expectedImportedId = importedThreadIdFor( + continuationKey, + binding.resumeCursor.threadId, + ); + if (binding.threadId !== expectedImportedId) { + const excluded = excludedThreadIdsByContinuation.get(continuationKey) ?? new Set(); + excluded.add(binding.resumeCursor.threadId); + excludedThreadIdsByContinuation.set(continuationKey, excluded); + } else { + const importedOwners = + importedOwnerInstanceIdsByContinuation.get(continuationKey) ?? new Set(); + importedOwners.add(binding.providerInstanceId); + importedOwnerInstanceIdsByContinuation.set(continuationKey, importedOwners); + } + if ( + binding.threadId === expectedImportedId && + isImportedRuntimePayload(binding.runtimePayload) + ) { + const providerCursor = + binding.runtimePayload.providerDiscoveryCursor ?? + binding.runtimePayload.providerUpdatedAt; + if (providerCursor) { + const cursors = cursorByThreadIdByContinuation.get(continuationKey) ?? new Map(); + cursors.set(binding.resumeCursor.threadId, providerCursor); + cursorByThreadIdByContinuation.set(continuationKey, cursors); + } + } + } + + const discoveredCounts = yield* Effect.forEach( + discoveryCandidateGroups, + (candidates) => + Effect.gen(function* () { + let selected: + | { + readonly instance: ProviderInstance; + readonly model: string; + readonly discovered: ReadonlyArray; + } + | undefined; + for (const instance of candidates) { + const attempt = yield* Effect.gen(function* () { + const discover = instance.adapter.discoverPersistedThreads; + if (!discover) return undefined; + const providerSnapshot = yield* instance.snapshot.getSnapshot; + const model = + providerSnapshot.models.find((entry) => entry.isDefault)?.slug ?? + providerSnapshot.models[0]?.slug ?? + DEFAULT_MODEL_BY_PROVIDER[CODEX] ?? + "default"; + const continuationKey = instance.continuationIdentity.continuationKey; + const importedOwners = importedOwnerInstanceIdsByContinuation.get(continuationKey); + const ownerChanged = + importedOwners !== undefined && + Array.from(importedOwners).some( + (ownerInstanceId) => ownerInstanceId !== instance.instanceId, + ); + const discovered = yield* discover({ + excludeProviderThreadIds: providerThreadDiscoveryExclusions( + unresolvedNativeProviderThreadIds, + excludedThreadIdsByContinuation.get(continuationKey), + ), + cursorByProviderThreadId: ownerChanged + ? new Map() + : (cursorByThreadIdByContinuation.get(continuationKey) ?? new Map()), + }); + return { instance, model, discovered } as const; + }).pipe( + Effect.catch((cause) => + Effect.logWarning("persisted provider thread discovery candidate failed", { + provider: instance.driverKind, + providerInstanceId: instance.instanceId, + continuationIdentity: continuationIdentityDigest( + instance.continuationIdentity.continuationKey, + ), + errorType: diagnosticErrorType(cause), + }).pipe(Effect.as(undefined)), + ), + ); + if (attempt) { + selected = attempt; + break; + } + } + if (!selected) return 0; + const { instance, model, discovered } = selected; + + yield* Effect.forEach( + discovered, + (thread) => + reconcilePersistedThread({ + instance, + thread, + model, + threadByProviderIdentity, + directory, + snapshots, + engine, + unassignedWorkspaceRoot: path.join( + serverConfig.stateDir, + "provider-imports", + continuationIdentityDigest(instance.continuationIdentity.continuationKey), + ), + }).pipe( + Effect.catchCause((cause) => + recoverReconciliationCause( + cause, + "skipped persisted provider thread during reconciliation", + { + provider: instance.driverKind, + providerInstanceId: instance.instanceId, + providerThreadId: thread.providerThreadId, + }, + undefined, + ), + ), + ), + { concurrency: 1, discard: true }, + ); + return discovered.length; + }).pipe( + Effect.catchCause((cause) => + recoverReconciliationCause( + cause, + "persisted provider thread discovery failed", + { + provider: CODEX, + continuationIdentity: candidates[0] + ? continuationIdentityDigest(candidates[0].continuationIdentity.continuationKey) + : undefined, + }, + 0, + ), + ), + ), + { concurrency: "unbounded" }, + ); + return discoveredCounts.reduce((total, count) => total + count, 0); + }, +); + +export const reconcilePersistedThread = Effect.fn("reconcilePersistedProviderThread")( + function* (input: { + readonly instance: ProviderInstance; + readonly thread: ProviderPersistedThread; + readonly model: string; + readonly threadByProviderIdentity: Map; + readonly directory: ProviderSessionDirectory["Service"]; + readonly snapshots: ProjectionSnapshotQuery["Service"]; + readonly engine: OrchestrationEngineService["Service"]; + readonly unassignedWorkspaceRoot: string; + }) { + const continuationKey = input.instance.continuationIdentity.continuationKey; + const continuationIdentity = continuationIdentityDigest(continuationKey); + const identity = providerIdentityKey(continuationKey, input.thread.providerThreadId); + const expectedThreadId = importedThreadId(input.instance, input.thread.providerThreadId); + const boundThreadId = input.threadByProviderIdentity.get(identity); + + // A non-imported T3 thread already owns this Codex identity. Its transcript + // is projected from live runtime events and must never be imported again. + if (boundThreadId !== undefined && boundThreadId !== expectedThreadId) return; + + const existingThread = yield* input.snapshots.getThreadShellById(expectedThreadId); + const matchingProject = yield* input.snapshots.getActiveProjectByWorkspaceRoot( + input.thread.cwd, + ); + const projectId = Option.isSome(matchingProject) + ? matchingProject.value.id + : unassignedProjectId(input.instance); + + if (Option.isNone(matchingProject)) { + const existingUnassignedProject = yield* input.snapshots.getProjectShellById(projectId); + if (Option.isNone(existingUnassignedProject)) { + if (Option.isNone(existingThread)) { + yield* input.engine.dispatch({ + type: "project.create", + commandId: importCommandId(continuationIdentity, "unassigned-project"), + projectId, + title: "Unassigned Codex threads", + workspaceRoot: input.unassignedWorkspaceRoot, + defaultModelSelection: { instanceId: input.instance.instanceId, model: input.model }, + createdAt: input.thread.createdAt, + }); + } + } else if ( + existingUnassignedProject.value.defaultModelSelection?.instanceId !== + input.instance.instanceId || + existingUnassignedProject.value.defaultModelSelection.model !== input.model + ) { + yield* input.engine.dispatch({ + type: "project.meta.update", + commandId: importCommandId( + continuationIdentity, + "unassigned-project", + "owner", + input.instance.instanceId, + input.model, + ), + projectId, + defaultModelSelection: { instanceId: input.instance.instanceId, model: input.model }, + }); + } + } + + if (Option.isNone(existingThread)) { + yield* input.engine.dispatch({ + type: "thread.create", + commandId: importCommandId(continuationIdentity, input.thread.providerThreadId, "create"), + threadId: expectedThreadId, + projectId, + title: input.thread.title, + modelSelection: { instanceId: input.instance.instanceId, model: input.model }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: Option.isSome(matchingProject) ? null : input.thread.cwd, + createdAt: input.thread.createdAt, + }); + } else if ( + existingThread.value.modelSelection.instanceId !== input.instance.instanceId || + existingThread.value.modelSelection.model !== input.model + ) { + yield* input.engine.dispatch({ + type: "thread.meta.update", + commandId: importCommandId( + continuationIdentity, + input.thread.providerThreadId, + "owner", + input.instance.instanceId, + input.model, + ), + threadId: expectedThreadId, + modelSelection: { instanceId: input.instance.instanceId, model: input.model }, + }); + } + + const existingDetail = Option.isSome(existingThread) + ? yield* input.snapshots.getThreadDetailById(expectedThreadId) + : Option.none(); + const projectedMessages = Option.isSome(existingDetail) ? existingDetail.value.messages : []; + const projectedIndexById = new Map( + projectedMessages.map((message, index) => [message.id, index]), + ); + const projectedIndexesByContent = new Map>(); + for (const [index, message] of projectedMessages.entries()) { + const indexesByText = projectedIndexesByContent.get(message.role) ?? new Map(); + const indexes = indexesByText.get(message.text) ?? []; + indexes.push(index); + indexesByText.set(message.text, indexes); + projectedIndexesByContent.set(message.role, indexesByText); + } + const nextContentPosition = new Map>(); + let projectedIndex = 0; + const missingMessages = input.thread.messages.filter((message) => { + const deterministicId = importedMessageId( + continuationKey, + input.thread.providerThreadId, + message.id, + message.sourceOrdinal, + ); + const exactIndex = projectedIndexById.get(deterministicId); + if (exactIndex !== undefined && exactIndex >= projectedIndex) { + projectedIndex = exactIndex + 1; + return false; + } + + const indexes = projectedIndexesByContent.get(message.role)?.get(message.text) ?? []; + const positionsByText = nextContentPosition.get(message.role) ?? new Map(); + let contentPosition = positionsByText.get(message.text) ?? 0; + while (contentPosition < indexes.length && indexes[contentPosition]! < projectedIndex) { + contentPosition += 1; + } + positionsByText.set(message.text, contentPosition + 1); + nextContentPosition.set(message.role, positionsByText); + const matchingIndex = indexes[contentPosition]; + if (matchingIndex !== undefined) { + projectedIndex = matchingIndex + 1; + return false; + } + return true; + }); + + yield* Effect.forEach( + missingMessages, + (message) => + input.engine.dispatch({ + type: "thread.message.import", + commandId: importCommandId( + continuationIdentity, + input.thread.providerThreadId, + "message", + message.id, + ), + threadId: expectedThreadId, + messageId: importedMessageId( + continuationKey, + input.thread.providerThreadId, + message.id, + message.sourceOrdinal, + ), + role: message.role, + text: message.text, + turnId: TurnId.make(message.turnId), + createdAt: message.createdAt, + }), + { concurrency: 1, discard: true }, + ); + + // Advance the discovery watermark only after every deterministic message + // command has landed. A crash mid-import then retries safely on the next + // pass instead of permanently hiding the remaining history. + yield* input.directory.upsert({ + threadId: expectedThreadId, + provider: CODEX, + providerInstanceId: input.instance.instanceId, + ...(boundThreadId === undefined + ? { runtimeMode: "full-access" as const, status: "stopped" as const } + : {}), + resumeCursor: { threadId: input.thread.providerThreadId }, + runtimePayload: { + imported: true, + continuationKey, + providerUpdatedAt: input.thread.updatedAt, + providerDiscoveryCursor: input.thread.discoveryCursor, + sourceMetadata: input.thread.sourceMetadata, + modelSelection: { instanceId: input.instance.instanceId, model: input.model }, + }, + }); + input.threadByProviderIdentity.set(identity, expectedThreadId); + }, +); + +export const ProviderThreadReconcilerLive = Layer.effectDiscard( + Effect.gen(function* () { + const registry = yield* ProviderInstanceRegistry; + const changes = yield* registry.subscribeChanges; + const reconcileSemaphore = yield* Semaphore.make(1); + const reconcile = reconcileSemaphore.withPermits(1)( + reconcilePersistedProviderThreads().pipe( + Effect.catchCause((cause) => + recoverReconciliationCause( + cause, + "persisted provider thread reconciliation failed", + {}, + 0, + ), + ), + ), + ); + + yield* Effect.forkScoped( + Effect.forever( + reconcile.pipe( + Effect.flatMap((discoveredCount) => + Effect.sleep(discoveredCount > 0 ? ACTIVE_RECONCILE_INTERVAL : IDLE_RECONCILE_INTERVAL), + ), + ), + ), + ); + yield* Effect.forkScoped(Effect.forever(PubSub.take(changes).pipe(Effect.andThen(reconcile)))); + }), +); diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 634745832b37..bcf4d05a234d 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -44,6 +44,31 @@ export interface ProviderThreadSnapshot { readonly turns: ReadonlyArray; } +export interface ProviderPersistedThreadMessage { + readonly id: string; + readonly sourceOrdinal: number; + readonly role: "user" | "assistant"; + readonly text: string; + readonly turnId: TurnId; + readonly createdAt: string; +} + +export interface ProviderPersistedThread { + readonly providerThreadId: string; + readonly cwd: string; + readonly title: string; + readonly createdAt: string; + readonly updatedAt: string; + readonly discoveryCursor: string; + readonly sourceMetadata: unknown; + readonly messages: ReadonlyArray; +} + +export interface ProviderPersistedThreadDiscoveryInput { + readonly excludeProviderThreadIds: ReadonlySet; + readonly cursorByProviderThreadId: ReadonlyMap; +} + export interface ProviderAdapterShape { /** * Provider kind implemented by this adapter. @@ -108,6 +133,15 @@ export interface ProviderAdapterShape { */ readonly readThread: (threadId: ThreadId) => Effect.Effect; + /** + * Discover durable provider threads that may have been created by another + * compatible client. Adapters omit this capability when their provider has + * no supported persisted-thread listing API. + */ + readonly discoverPersistedThreads?: ( + input?: ProviderPersistedThreadDiscoveryInput, + ) => Effect.Effect, TError>; + /** * Roll back a provider thread by N turns. */ diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 3e41b4390f82..ba5de8f691bf 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -34,6 +34,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 { ProviderThreadReconcilerLive } from "./provider/Layers/ProviderThreadReconciler.ts"; import * as OpenCodeRuntime from "./provider/opencodeRuntime.ts"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as CheckpointStore from "./checkpointing/CheckpointStore.ts"; @@ -241,6 +242,7 @@ const PlatformServicesLive = Layer.unwrap( const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(OrchestrationReactorLive), + Layer.provideMerge(ProviderThreadReconcilerLive), Layer.provideMerge(ProviderRuntimeIngestionLive), Layer.provideMerge(ProviderCommandReactorLive), Layer.provideMerge(CheckpointReactorLive), diff --git a/docs/internals/overview.md b/docs/internals/overview.md index b9454f7b58d0..c6a81f7e3e50 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -82,7 +82,7 @@ reconciles. Command and event names live in [`orchestration.ts`][contracts]. Some commands are client dispatchable (`thread.create`, `thread.turn.start`, `thread.approval.respond`); others are internal and produced only by server-side reactors (`thread.message.assistant.delta`, -`thread.turn.diff.complete`). +`thread.message.import`, `thread.turn.diff.complete`). A turn is complete when its session leaves `running` status, projected by `settledTurnStateForSessionStatus` in [`projector.ts`][projector]. Checkpoint work settling later @@ -100,6 +100,11 @@ Follow-up work runs asynchronously in queue-backed workers built on [`DrainableW count reaches zero, so a test can await "queue empty and current item finished" instead of sleeping. Each of the three services exposes `drain` for exactly this. +[`ProviderThreadReconciler`][reconciler] is a separate background fiber, not a drainable worker. It +discovers durable conversations from adapters that support persisted-thread listing, currently +Codex, and imports missing messages through the same event-sourced command path. It matches threads +to projects by working directory and keeps unmatched conversations in **Unassigned Codex threads**. + Runtime receipts are a test-only mechanism. `RuntimeReceiptBusLive` in [`RuntimeReceiptBus.ts`][receipts] publishes nothing; only the test layer is PubSub-backed. Do not build production behavior on receipts. @@ -147,6 +152,7 @@ already dispatch. [worker]: ../../packages/shared/src/DrainableWorker.ts [ingest]: ../../apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts [cmd]: ../../apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +[reconciler]: ../../apps/server/src/provider/Layers/ProviderThreadReconciler.ts [checkpoint]: ../../apps/server/src/orchestration/Layers/CheckpointReactor.ts [receipts]: ../../apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts [drivers]: ../../apps/server/src/provider/builtInDrivers.ts diff --git a/docs/internals/providers.md b/docs/internals/providers.md index a309d70f03de..277069a01106 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -36,6 +36,10 @@ Two registries separate configuration from live processes: [`ProviderService`][service] sits on top. It combines the adapter registry with the provider session directory to route session and turn operations for a thread, so callers name a thread, not an agent. +Adapters can optionally implement `discoverPersistedThreads` for conversations created outside T3 +Code. Codex implements this capability through [`CodexThreadDiscovery.ts`][codex-discovery], which +lists durable Codex threads and reads completed user and assistant messages. + Adding a driver means writing the driver plus adapter and adding it to `BUILT_IN_DRIVERS`. No orchestration, contract, or client change is required for the common case. @@ -66,6 +70,15 @@ synchronization. 3. [`CheckpointReactor`][checkpoint] captures workspace checkpoints on turn start and completion, and performs reverts. +Persisted conversation discovery is separate from these queue-backed workers. +[`ProviderThreadReconciler`][reconciler] runs when provider instances change and periodically in the +background. It groups compatible instances by continuation identity, asks one healthy adapter in +each group for persisted threads, and imports missing messages through the internal +`thread.message.import` command. Threads are attached to the project matching their working +directory; unmatched threads go into an **Unassigned Codex threads** project. Provider thread IDs, +discovery cursors, and deterministic command and message IDs make retries safe and keep live T3 +threads from being imported twice. + ### Buffered assistant delivery A thread in `buffered` assistant delivery mode accumulates assistant text instead of streaming each @@ -85,6 +98,8 @@ when a request opens (approval) or user input is requested, via [instances]: ../../apps/server/src/provider/Services/ProviderInstanceRegistry.ts [registry]: ../../apps/server/src/provider/Services/ProviderAdapterRegistry.ts [service]: ../../apps/server/src/provider/Layers/ProviderService.ts +[codex-discovery]: ../../apps/server/src/provider/Layers/CodexThreadDiscovery.ts +[reconciler]: ../../apps/server/src/provider/Layers/ProviderThreadReconciler.ts [contracts]: ../../packages/contracts/src/orchestration.ts [worker]: ../../packages/shared/src/DrainableWorker.ts [ingest]: ../../apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index 2396ce4028d2..4334252f0ca5 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -28,6 +28,12 @@ Log in with Codex normally: codex login ``` +T3 Code automatically discovers conversations created by Codex CLI and other Codex clients that +use the same `CODEX_HOME`. Conversations whose working directory matches a T3 Code project appear +in that project. Other conversations remain available under **Unassigned Codex threads**. Opening +an imported conversation shows its user and assistant history, and sending a message continues the +original Codex thread. New external conversations may take up to two minutes to appear. + ## Send feedback to OpenAI In an existing Codex thread, send `/feedback` or `/feedback` followed by a description of the diff --git a/packages/contracts/src/orchestration.test.ts b/packages/contracts/src/orchestration.test.ts index 3e1b9be0bba5..df028ab85ab2 100644 --- a/packages/contracts/src/orchestration.test.ts +++ b/packages/contracts/src/orchestration.test.ts @@ -57,6 +57,26 @@ const decodeOrchestrationEvent = Schema.decodeUnknownEffect(OrchestrationEvent); const decodeThreadMetaUpdatedPayload = Schema.decodeUnknownEffect(ThreadMetaUpdatedPayload); const decodeDispatchCommandError = Schema.decodeUnknownEffect(OrchestrationDispatchCommandError); +it.effect("decodes provider history message imports as internal commands", () => + Effect.gen(function* () { + const command = yield* decodeOrchestrationCommand({ + type: "thread.message.import", + commandId: "provider-import:message-1", + threadId: "thread-1", + messageId: "message-1", + role: "assistant", + text: "Imported response", + turnId: "turn-1", + createdAt: "2026-08-21T03:22:43.000Z", + }); + + assert.strictEqual(command.type, "thread.message.import"); + if (command.type === "thread.message.import") { + assert.strictEqual(command.turnId, "turn-1"); + } + }), +); + it.effect("decodes a dispatch error after its bootstrap thread was deleted", () => Effect.gen(function* () { const error = yield* decodeDispatchCommandError({ diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 1c27e6d3c6b4..defb285196d4 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -992,6 +992,17 @@ const ThreadMessageAssistantCompleteCommand = Schema.Struct({ createdAt: IsoDateTime, }); +const ThreadMessageImportCommand = Schema.Struct({ + type: Schema.Literal("thread.message.import"), + commandId: CommandId, + threadId: ThreadId, + messageId: MessageId, + role: OrchestrationMessageRole, + text: Schema.String, + turnId: TurnId, + createdAt: IsoDateTime, +}); + const ThreadProposedPlanUpsertCommand = Schema.Struct({ type: Schema.Literal("thread.proposed-plan.upsert"), commandId: CommandId, @@ -1040,6 +1051,7 @@ const ThreadTitleRegenerationCompleteCommand = Schema.Struct({ const InternalOrchestrationCommand = Schema.Union([ ThreadSessionSetCommand, + ThreadMessageImportCommand, ThreadMessageAssistantDeltaCommand, ThreadMessageAssistantCompleteCommand, ThreadProposedPlanUpsertCommand,