diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index fb753b9aa4b..21459a7db43 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -58,6 +58,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.projectsSearchContents]: AuthOrchestrationReadScope, [WS_METHODS.projectsSearchEntries]: AuthOrchestrationReadScope, [WS_METHODS.projectsWriteFile]: AuthOrchestrationOperateScope, + [WS_METHODS.codexListSessions]: AuthOrchestrationReadScope, + [WS_METHODS.codexImportSessions]: AuthOrchestrationOperateScope, [WS_METHODS.shellOpenInEditor]: AuthOrchestrationOperateScope, [WS_METHODS.filesystemBrowse]: AuthOrchestrationReadScope, [WS_METHODS.assetsCreateUrl]: AuthOrchestrationReadScope, diff --git a/apps/server/src/environment/ServerEnvironment.test.ts b/apps/server/src/environment/ServerEnvironment.test.ts index a7aea90f826..153390631c5 100644 --- a/apps/server/src/environment/ServerEnvironment.test.ts +++ b/apps/server/src/environment/ServerEnvironment.test.ts @@ -70,6 +70,7 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => { expect(second.capabilities.repositoryIdentity).toBe(true); expect(second.capabilities.connectionProbe).toBe(true); expect(second.capabilities.threadTitleRegeneration).toBe(true); + expect(second.capabilities.codexSessionImport).toBe(true); }), ); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 2290450e7d3..925dbe65aed 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -143,6 +143,7 @@ export const make = Effect.gen(function* () { threadSettlement: true, threadSnooze: true, threadTitleRegeneration: true, + codexSessionImport: true, ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }), ...(serverSelfUpdate === "boot-service" || serverSelfUpdate === "respawn" ? { serverSelfUpdateProgress: true } diff --git a/apps/server/src/orchestration/decider.historyImport.test.ts b/apps/server/src/orchestration/decider.historyImport.test.ts new file mode 100644 index 00000000000..94fcc3de347 --- /dev/null +++ b/apps/server/src/orchestration/decider.historyImport.test.ts @@ -0,0 +1,115 @@ +import { + CommandId, + EventId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationCommand, + type OrchestrationEvent, +} 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"; +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +const NOW = "2026-01-01T00:00:00.000Z"; +const PROJECT_ID = ProjectId.make("project-history-import"); +const THREAD_ID = ThreadId.make("thread-history-import"); + +const seedReadModel = projectEvent(createEmptyReadModel(NOW), { + sequence: 1, + eventId: EventId.make("evt-project-history-import"), + aggregateKind: "project", + aggregateId: PROJECT_ID, + type: "project.created", + occurredAt: NOW, + commandId: CommandId.make("cmd-project-history-import"), + causationEventId: null, + correlationId: CommandId.make("cmd-project-history-import"), + metadata: {}, + payload: { + projectId: PROJECT_ID, + title: "History Import", + workspaceRoot: "/tmp/history-import", + defaultModelSelection: null, + scripts: [], + createdAt: NOW, + updatedAt: NOW, + }, +}); + +it.layer(NodeServices.layer)("history import decider", (it) => { + it.effect("creates a non-running thread and projects its text snapshot in order", () => + Effect.gen(function* () { + const readModel = yield* seedReadModel; + const command = { + type: "thread.history.import", + commandId: CommandId.make("cmd-history-import"), + threadId: THREAD_ID, + projectId: PROJECT_ID, + title: "Codex conversation", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + interactionMode: "default", + createdAt: NOW, + messages: [ + { + messageId: MessageId.make("message-history-user"), + role: "user", + text: "Please inspect this project.", + turnId: null, + createdAt: "2026-01-01T00:01:00.000Z", + }, + { + messageId: MessageId.make("message-history-assistant"), + role: "assistant", + text: "I found the relevant files.", + turnId: null, + createdAt: "2026-01-01T00:02:00.000Z", + }, + ], + } satisfies OrchestrationCommand; + + const result = yield* decideOrchestrationCommand({ command, readModel }); + const events = Array.isArray(result) ? result : [result]; + + expect(events.map((event) => event.type)).toEqual([ + "thread.created", + "thread.message-sent", + "thread.message-sent", + ]); + const created = events[0]; + expect(created?.type).toBe("thread.created"); + if (created?.type === "thread.created") { + expect(created.payload.branch).toBeNull(); + expect(created.payload.worktreePath).toBeNull(); + } + const messages = events.filter( + (event): event is Extract => + event.type === "thread.message-sent", + ); + expect(messages.map((event) => event.payload.text)).toEqual([ + "Please inspect this project.", + "I found the relevant files.", + ]); + expect(messages.every((event) => event.payload.streaming === false)).toBe(true); + + let projected = readModel; + for (const [index, event] of events.entries()) { + projected = yield* projectEvent(projected, { ...event, sequence: index + 2 }); + } + const thread = projected.threads.find((entry) => entry.id === THREAD_ID); + expect(thread?.session).toBeNull(); + expect(thread?.messages.map((message) => [message.role, message.text])).toEqual([ + ["user", "Please inspect this project."], + ["assistant", "I found the relevant files."], + ]); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index ae9a068864c..63d2a145fc4 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -378,6 +378,65 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.history.import": { + yield* requireProject({ + readModel, + command, + projectId: command.projectId, + }); + yield* requireThreadAbsent({ + readModel, + command, + threadId: command.threadId, + }); + + const created = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.created" as const, + payload: { + threadId: command.threadId, + projectId: command.projectId, + title: command.title, + modelSelection: command.modelSelection, + runtimeMode: command.runtimeMode, + interactionMode: command.interactionMode, + branch: null, + worktreePath: null, + createdAt: command.createdAt, + updatedAt: command.createdAt, + }, + }; + const importedMessages = yield* Effect.forEach(command.messages, (message) => + withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: message.createdAt, + commandId: command.commandId, + }).pipe( + Effect.map((eventBase) => ({ + ...eventBase, + type: "thread.message-sent" as const, + payload: { + threadId: command.threadId, + messageId: message.messageId, + role: message.role, + text: message.text, + turnId: message.turnId, + streaming: false, + createdAt: message.createdAt, + updatedAt: message.createdAt, + }, + })), + ), + ); + return [created, ...importedMessages]; + } + case "thread.delete": { yield* requireThread({ readModel, diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 15d7a1ff021..1bb4f5b87a9 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -37,6 +37,7 @@ import { ServerSettingsService } from "../../serverSettings.ts"; import { ProviderDriverError } from "../Errors.ts"; import { makeCodexAdapter } from "../Layers/CodexAdapter.ts"; import { checkCodexProviderStatus, makePendingCodexProvider } from "../Layers/CodexProvider.ts"; +import { makeCodexThreadHistory } from "../Layers/CodexThreadHistory.ts"; import { ProviderEventLoggers } from "../Layers/ProviderEventLoggers.ts"; import { makeManagedServerProvider } from "../makeManagedServerProvider.ts"; import type { ProviderDriver, ProviderInstance } from "../ProviderDriver.ts"; @@ -160,6 +161,11 @@ export const CodexDriver: ProviderDriver = { environment: processEnv, ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}), }); + const threadHistory = makeCodexThreadHistory({ + config: effectiveConfig, + environment: processEnv, + spawner, + }); const textGeneration = yield* makeCodexTextGeneration(effectiveConfig, processEnv); // Build a managed snapshot whose settings never change — mutations come @@ -207,6 +213,7 @@ export const CodexDriver: ProviderDriver = { enabled, snapshot, adapter, + threadHistory, textGeneration, } satisfies ProviderInstance; }), diff --git a/apps/server/src/provider/Errors.ts b/apps/server/src/provider/Errors.ts index 0cf1522399b..62b392a0476 100644 --- a/apps/server/src/provider/Errors.ts +++ b/apps/server/src/provider/Errors.ts @@ -156,6 +156,24 @@ export class ProviderDriverError extends Schema.TaggedErrorClass()( + "ProviderThreadHistoryError", + { + provider: Schema.String, + operation: Schema.Literals(["list", "read"]), + detail: Schema.String, + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + return `Provider thread history failed (${this.provider}) during ${this.operation}: ${this.detail}`; + } +} + /** * ProviderSessionNotFoundError - Provider-facing session not found. */ diff --git a/apps/server/src/provider/Layers/CodexSessionImport.test.ts b/apps/server/src/provider/Layers/CodexSessionImport.test.ts new file mode 100644 index 00000000000..5879acaca59 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexSessionImport.test.ts @@ -0,0 +1,291 @@ +import { + CommandId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + type OrchestrationCommand, + type OrchestrationProjectShell, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +import * as Stream from "effect/Stream"; + +import * as OrchestrationEngine from "../../orchestration/Services/OrchestrationEngine.ts"; +import * as ProjectionSnapshotQuery from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { ProviderInstance, ProviderThreadHistorySource } from "../ProviderDriver.ts"; +import * as ProviderInstanceRegistry from "../Services/ProviderInstanceRegistry.ts"; +import * as ProviderSessionDirectory from "../Services/ProviderSessionDirectory.ts"; +import { makeCodexSessionImport } from "./CodexSessionImport.ts"; + +const CODEX = ProviderDriverKind.make("codex"); +const PROVIDER_INSTANCE_ID = ProviderInstanceId.make("codex"); +const PROJECT_ID = ProjectId.make("project-codex-import"); +const NOW = "2026-01-01T00:00:00.000Z"; + +const project: OrchestrationProjectShell = { + id: PROJECT_ID, + title: "Codex import project", + workspaceRoot: "/workspace/codex-import", + defaultModelSelection: null, + scripts: [], + createdAt: NOW, + updatedAt: NOW, +}; + +const candidate = (externalThreadId: string) => ({ + externalThreadId, + title: `Session ${externalThreadId}`, + preview: `Preview ${externalThreadId}`, + createdAt: NOW, + updatedAt: "2026-01-02T00:00:00.000Z", + source: "cli", + archived: false, +}); + +function makeInstance(source: ProviderThreadHistorySource): ProviderInstance { + return { + instanceId: PROVIDER_INSTANCE_ID, + driverKind: CODEX, + continuationIdentity: { + driverKind: CODEX, + continuationKey: "codex:home:test", + }, + displayName: undefined, + enabled: true, + snapshot: { + getSnapshot: Effect.succeed({ + models: [{ slug: "gpt-5-codex", isDefault: true }], + }), + } as unknown as ProviderInstance["snapshot"], + adapter: {} as ProviderInstance["adapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + threadHistory: source, + }; +} + +function makeRegistry( + instance: ProviderInstance, +): ProviderInstanceRegistry.ProviderInstanceRegistry["Service"] { + return { + getInstance: (instanceId) => + Effect.succeed(instanceId === instance.instanceId ? instance : undefined), + listInstances: Effect.succeed([instance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: Effect.flatMap(PubSub.unbounded(), (changes) => + PubSub.subscribe(changes), + ), + }; +} + +function makeDirectory(input?: { + readonly bindings?: ReadonlyArray; + readonly upserts?: Array; +}): ProviderSessionDirectory.ProviderSessionDirectory["Service"] { + return { + upsert: (binding) => + Effect.sync(() => { + input?.upserts?.push(binding); + }), + getProvider: () => Effect.die("not used by Codex session import"), + getBinding: () => Effect.succeed(Option.none()), + listThreadIds: () => Effect.succeed([]), + listBindings: () => Effect.succeed(input?.bindings ?? []), + }; +} + +function makeImportLayer(input: { + readonly source: ProviderThreadHistorySource; + readonly dispatches?: Array; + readonly bindings?: ReadonlyArray; + readonly upserts?: Array; +}) { + const instance = makeInstance(input.source); + return Layer.mergeAll( + Layer.succeed( + ProjectionSnapshotQuery.ProjectionSnapshotQuery, + ProjectionSnapshotQuery.ProjectionSnapshotQuery.of({ + getProjectShellById: (projectId: ProjectId) => + Effect.succeed(projectId === PROJECT_ID ? Option.some(project) : Option.none()), + } as unknown as ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"]), + ), + Layer.succeed( + OrchestrationEngine.OrchestrationEngineService, + OrchestrationEngine.OrchestrationEngineService.of({ + dispatch: (command: OrchestrationCommand) => + Effect.sync(() => { + input.dispatches?.push(command); + return { sequence: 1 }; + }), + } as unknown as OrchestrationEngine.OrchestrationEngineService["Service"]), + ), + Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, makeRegistry(instance)), + Layer.succeed( + ProviderSessionDirectory.ProviderSessionDirectory, + makeDirectory({ + ...(input.bindings === undefined ? {} : { bindings: input.bindings }), + ...(input.upserts === undefined ? {} : { upserts: input.upserts }), + }), + ), + ); +} + +it("marks only strict imported bindings as already imported", () => { + const source: ProviderThreadHistorySource = { + listThreads: () => + Effect.succeed({ + threads: [candidate("ordinary-session"), candidate("imported-session")], + truncated: false, + }), + readThreads: () => Effect.succeed([]), + }; + const result = Effect.runSync( + Effect.gen(function* () { + const importer = yield* makeCodexSessionImport; + return yield* importer.list({ + projectId: PROJECT_ID, + providerInstanceId: PROVIDER_INSTANCE_ID, + }); + }).pipe( + Effect.provide( + makeImportLayer({ + source, + bindings: [ + { + threadId: ThreadId.make("ordinary-t3-thread"), + provider: CODEX, + providerInstanceId: PROVIDER_INSTANCE_ID, + resumeCursor: { threadId: "ordinary-session" }, + lastSeenAt: NOW, + }, + { + threadId: ThreadId.make("imported-t3-thread"), + provider: CODEX, + providerInstanceId: PROVIDER_INSTANCE_ID, + resumeCursor: { threadId: "imported-session", requireExistingThread: true }, + lastSeenAt: NOW, + }, + ], + }), + ), + ), + ); + + expect( + result.sessions.map((session) => [session.externalThreadId, session.importedThreadId]), + ).toEqual([ + ["ordinary-session", null], + ["imported-session", ThreadId.make("imported-t3-thread")], + ]); +}); + +it("imports each selected native session once with a strict continuation binding", () => { + const readCalls: Array> = []; + const dispatches: OrchestrationCommand[] = []; + const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; + const source: ProviderThreadHistorySource = { + listThreads: () => Effect.succeed({ threads: [candidate("native-session")], truncated: false }), + readThreads: ({ externalThreadIds }) => { + readCalls.push(externalThreadIds); + return Effect.succeed([ + { + externalThreadId: "native-session", + title: "Imported native session", + createdAt: NOW, + messages: [ + { + externalMessageId: "message-1", + role: "user", + text: "Please continue this work.", + createdAt: NOW, + }, + ], + }, + ]); + }, + }; + + const result = Effect.runSync( + Effect.gen(function* () { + const importer = yield* makeCodexSessionImport; + return yield* importer.import({ + projectId: PROJECT_ID, + providerInstanceId: PROVIDER_INSTANCE_ID, + externalThreadIds: ["native-session", "native-session"], + }); + }).pipe(Effect.provide(makeImportLayer({ source, dispatches, upserts }))), + ); + + expect(readCalls).toEqual([["native-session"]]); + expect(result).toEqual({ + importedThreadIds: [ThreadId.make("codex:codex:native-session")], + alreadyImportedThreadIds: [], + }); + expect(dispatches).toHaveLength(1); + expect(dispatches[0]).toMatchObject({ + type: "thread.history.import", + commandId: CommandId.make("codex-history-import:codex:native-session"), + threadId: ThreadId.make("codex:codex:native-session"), + projectId: PROJECT_ID, + modelSelection: { + instanceId: PROVIDER_INSTANCE_ID, + model: "gpt-5-codex", + }, + messages: [ + expect.objectContaining({ + role: "user", + text: "Please continue this work.", + turnId: null, + }), + ], + }); + expect(upserts).toEqual([ + expect.objectContaining({ + threadId: ThreadId.make("codex:codex:native-session"), + provider: CODEX, + providerInstanceId: PROVIDER_INSTANCE_ID, + status: "stopped", + resumeCursor: { threadId: "native-session", requireExistingThread: true }, + runtimePayload: expect.objectContaining({ + cwd: project.workspaceRoot, + importedFrom: "codex", + }), + }), + ]); +}); + +it("refuses a stale selection before reading or creating any T3 thread", () => { + let readCount = 0; + const dispatches: OrchestrationCommand[] = []; + const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; + const source: ProviderThreadHistorySource = { + listThreads: () => Effect.succeed({ threads: [], truncated: false }), + readThreads: () => { + readCount += 1; + return Effect.succeed([]); + }, + }; + + const error = Effect.runSync( + Effect.gen(function* () { + const importer = yield* makeCodexSessionImport; + return yield* importer + .import({ + projectId: PROJECT_ID, + providerInstanceId: PROVIDER_INSTANCE_ID, + externalThreadIds: ["no-longer-present"], + }) + .pipe(Effect.flip); + }).pipe(Effect.provide(makeImportLayer({ source, dispatches, upserts }))), + ); + + expect(error._tag).toBe("CodexSessionImportError"); + expect(error.operation).toBe("validate-sessions"); + expect(readCount).toBe(0); + expect(dispatches).toEqual([]); + expect(upserts).toEqual([]); +}); diff --git a/apps/server/src/provider/Layers/CodexSessionImport.ts b/apps/server/src/provider/Layers/CodexSessionImport.ts new file mode 100644 index 00000000000..d5bcef51a61 --- /dev/null +++ b/apps/server/src/provider/Layers/CodexSessionImport.ts @@ -0,0 +1,368 @@ +/** + * CodexSessionImport live implementation. + * + * Import is deliberately one-way at the storage layer: Codex remains the + * native session owner, while T3 receives a text-only history projection plus + * a strict continuation binding. This lets either Codex surface keep using + * the original thread without T3 rewriting any of its files. + * + * @module provider/Layers/CodexSessionImport + */ +import { + CodexSessionImportError, + DEFAULT_MODEL, + DEFAULT_PROVIDER_INTERACTION_MODE, + type CodexSessionImportInput, + type CodexSessionListInput, + type ModelSelection, + ProviderDriverKind, + TrimmedNonEmptyString, + ThreadId, + CommandId, + MessageId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; + +import * as OrchestrationEngine from "../../orchestration/Services/OrchestrationEngine.ts"; +import * as ProjectionSnapshotQuery from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { + ProviderInstance, + ProviderThreadHistory, + ProviderThreadHistoryCandidate, + ProviderThreadHistorySource, +} from "../ProviderDriver.ts"; +import { isCodexResumeCursor } from "./CodexSessionRuntime.ts"; +import { ProviderInstanceRegistry } from "../Services/ProviderInstanceRegistry.ts"; +import { CodexSessionImport } from "../Services/CodexSessionImport.ts"; +import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; + +const CODEX_DRIVER_KIND = ProviderDriverKind.make("codex"); + +function importError(operation: string, message: string, cause?: unknown): CodexSessionImportError { + return new CodexSessionImportError({ + operation, + message, + ...(cause === undefined ? {} : { cause }), + }); +} + +function stableSegment(value: string): string { + return encodeURIComponent(value); +} + +function importedThreadId(input: { + readonly providerInstanceId: string; + readonly externalThreadId: string; +}): ThreadId { + return ThreadId.make( + `codex:${stableSegment(input.providerInstanceId)}:${stableSegment(input.externalThreadId)}`, + ); +} + +function importCommandId(input: { + readonly providerInstanceId: string; + readonly externalThreadId: string; +}): CommandId { + return CommandId.make( + `codex-history-import:${stableSegment(input.providerInstanceId)}:${stableSegment( + input.externalThreadId, + )}`, + ); +} + +function importedMessageId(input: { + readonly externalThreadId: string; + readonly externalMessageId: string; +}): MessageId { + return MessageId.make( + `codex:${stableSegment(input.externalThreadId)}:message:${stableSegment(input.externalMessageId)}`, + ); +} + +function uniqueThreadIds(externalThreadIds: ReadonlyArray): ReadonlyArray { + return Array.from(new Set(externalThreadIds)); +} + +function configuredCodexSource( + instance: ProviderInstance, +): ProviderThreadHistorySource | undefined { + return instance.enabled && instance.driverKind === CODEX_DRIVER_KIND + ? instance.threadHistory + : undefined; +} + +const modelSelectionFor = Effect.fn("CodexSessionImport.modelSelectionFor")(function* (input: { + readonly projectDefault: ModelSelection | null; + readonly provider: ProviderInstance; +}): Effect.fn.Return { + if (input.projectDefault?.instanceId === input.provider.instanceId) { + return input.projectDefault; + } + const snapshot = yield* input.provider.snapshot.getSnapshot; + return { + instanceId: input.provider.instanceId, + model: + snapshot.models.find((model) => model.isDefault === true)?.slug ?? + snapshot.models.at(0)?.slug ?? + DEFAULT_MODEL, + }; +}); + +export const makeCodexSessionImport = Effect.gen(function* () { + const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; + const instanceRegistry = yield* ProviderInstanceRegistry; + const sessionDirectory = yield* ProviderSessionDirectory; + + const resolveImportContext = Effect.fn("CodexSessionImport.resolveContext")(function* ( + input: CodexSessionListInput, + ) { + const project = yield* projectionSnapshotQuery + .getProjectShellById(input.projectId) + .pipe( + Effect.mapError((cause) => + importError("resolve-project", "Could not load this T3 project.", cause), + ), + ); + if (Option.isNone(project)) { + return yield* importError("resolve-project", "This T3 project is no longer available."); + } + + const instance = yield* instanceRegistry.getInstance(input.providerInstanceId); + const source = instance ? configuredCodexSource(instance) : undefined; + if (!instance || !source) { + return yield* importError( + "resolve-provider", + "Choose an enabled Codex provider before importing its sessions.", + ); + } + + return { project: project.value, instance, source }; + }); + + const listImportedThreadIds = Effect.fn("CodexSessionImport.listImportedThreadIds")(function* ( + providerInstanceId: string, + ) { + const bindings = yield* sessionDirectory + .listBindings() + .pipe( + Effect.mapError((cause) => + importError( + "list-bindings", + "Could not check which Codex sessions are already imported.", + cause, + ), + ), + ); + const imported = new Map(); + for (const binding of bindings) { + if ( + binding.provider === CODEX_DRIVER_KIND && + binding.providerInstanceId === providerInstanceId && + isCodexResumeCursor(binding.resumeCursor) && + binding.resumeCursor.requireExistingThread === true + ) { + imported.set(binding.resumeCursor.threadId, binding.threadId); + } + } + return imported; + }); + + const list = Effect.fn("CodexSessionImport.list")(function* (input: CodexSessionListInput) { + const context = yield* resolveImportContext(input); + const [sourceResult, imported] = yield* Effect.all([ + context.source + .listThreads({ cwd: context.project.workspaceRoot }) + .pipe( + Effect.mapError((cause) => + importError( + "list-sessions", + "Could not read Codex sessions. Check that Codex is available and try again.", + cause, + ), + ), + ), + listImportedThreadIds(input.providerInstanceId), + ]); + + return { + sessions: sourceResult.threads.map((session) => ({ + ...session, + importedThreadId: imported.get(session.externalThreadId) ?? null, + })), + truncated: sourceResult.truncated, + }; + }); + + const importSessions = Effect.fn("CodexSessionImport.import")(function* ( + input: CodexSessionImportInput, + ) { + const context = yield* resolveImportContext(input); + const externalThreadIds = uniqueThreadIds(input.externalThreadIds); + const [listed, existingImports] = yield* Effect.all([ + context.source + .listThreads({ cwd: context.project.workspaceRoot }) + .pipe( + Effect.mapError((cause) => + importError( + "validate-sessions", + "Could not verify the selected Codex sessions. Refresh the list and try again.", + cause, + ), + ), + ), + listImportedThreadIds(input.providerInstanceId), + ]); + const candidatesById = new Map( + listed.threads.map((candidate) => [candidate.externalThreadId, candidate]), + ); + const unavailableIds = externalThreadIds.filter((threadId) => !candidatesById.has(threadId)); + if (unavailableIds.length > 0) { + return yield* importError( + "validate-sessions", + "One or more selected Codex sessions are no longer available for this project. Refresh the list and try again.", + ); + } + + const alreadyImportedThreadIds = externalThreadIds.flatMap((externalThreadId) => { + const threadId = existingImports.get(externalThreadId); + return threadId ? [threadId] : []; + }); + const idsToImport = externalThreadIds.filter( + (externalThreadId) => !existingImports.has(externalThreadId), + ); + if (idsToImport.length === 0) { + return { importedThreadIds: [], alreadyImportedThreadIds }; + } + + const [histories, modelSelection] = yield* Effect.all([ + context.source + .readThreads({ + cwd: context.project.workspaceRoot, + externalThreadIds: idsToImport, + }) + .pipe( + Effect.mapError((cause) => + importError( + "read-sessions", + "Could not read the selected Codex sessions. They may have changed; refresh and try again.", + cause, + ), + ), + ), + modelSelectionFor({ + projectDefault: context.project.defaultModelSelection, + provider: context.instance, + }).pipe( + Effect.mapError((cause) => + importError( + "resolve-model", + "Could not choose a Codex model for imported sessions.", + cause, + ), + ), + ), + ]); + const historiesById = new Map( + histories.map((history) => [history.externalThreadId, history]), + ); + const missingHistoryIds = idsToImport.filter((threadId) => !historiesById.has(threadId)); + if (missingHistoryIds.length > 0) { + return yield* importError( + "read-sessions", + "Codex did not return every selected session. Refresh the list and try again.", + ); + } + + const importedThreadIds = yield* Effect.forEach( + idsToImport, + (externalThreadId) => { + const history = historiesById.get(externalThreadId); + if (!history) { + return Effect.fail( + importError("read-sessions", "A selected Codex session could not be read."), + ); + } + const threadId = importedThreadId({ + providerInstanceId: input.providerInstanceId, + externalThreadId, + }); + const commandId = importCommandId({ + providerInstanceId: input.providerInstanceId, + externalThreadId, + }); + return orchestrationEngine + .dispatch({ + type: "thread.history.import", + commandId, + threadId, + projectId: input.projectId, + title: TrimmedNonEmptyString.make(history.title), + modelSelection, + runtimeMode: "full-access", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + createdAt: history.createdAt, + messages: history.messages.map((message) => ({ + messageId: importedMessageId({ + externalThreadId, + externalMessageId: message.externalMessageId, + }), + role: message.role, + text: message.text, + // Imported history has no corresponding T3 turn/checkpoint + // events. Keep it unbound so a later T3 revert cannot erase the + // historical snapshot while pruning native T3 turns. + turnId: null, + createdAt: message.createdAt, + })), + }) + .pipe( + Effect.mapError((cause) => + importError( + "create-thread", + "Could not create an imported T3 thread. Some earlier sessions may already be imported; refresh to see the current state.", + cause, + ), + ), + Effect.flatMap(() => + sessionDirectory + .upsert({ + threadId, + provider: CODEX_DRIVER_KIND, + providerInstanceId: input.providerInstanceId, + runtimeMode: "full-access", + status: "stopped", + // Do not fall back to a new native thread if the original was + // deleted after import. A failed resume is safer and clearer. + resumeCursor: { threadId: externalThreadId, requireExistingThread: true }, + runtimePayload: { + cwd: context.project.workspaceRoot, + modelSelection, + importedFrom: "codex", + }, + }) + .pipe( + Effect.mapError((cause) => + importError( + "save-binding", + "The T3 thread was created but its Codex continuation link could not be saved. Retry this import to repair the link.", + cause, + ), + ), + ), + ), + Effect.as(threadId), + ); + }, + { concurrency: 1 }, + ); + + return { importedThreadIds, alreadyImportedThreadIds }; + }); + + return { list, import: importSessions }; +}); + +export const CodexSessionImportLive = Layer.effect(CodexSessionImport, makeCodexSessionImport); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index d7346a0e0db..2b6fd55875b 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -433,6 +433,46 @@ describe("openCodexThread", () => { }), ); + it.effect("does not start a new native thread for a strict imported binding", () => + Effect.gen(function* () { + const calls: Array<"thread/start" | "thread/resume"> = []; + const client = { + request: ( + method: M, + _payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + calls.push(method); + if (method === "thread/resume") { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread not found", + }), + ); + } + return Effect.succeed( + makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M], + ); + }, + }; + + const error = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "deleted-native-thread", + allowResumeFallback: false, + }).pipe(Effect.flip); + + NodeAssert.ok(isCodexAppServerRequestError(error)); + NodeAssert.equal(error.errorMessage, "thread not found"); + NodeAssert.deepStrictEqual(calls, ["thread/resume"]); + }), + ); + it.effect("propagates non-recoverable resume failures", () => Effect.gen(function* () { const client = { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 67108dd4dbb..1869f51988b 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -66,11 +66,17 @@ export function hasConfiguredMcpServer(appServerArgs: ReadonlyArray | un export const CodexResumeCursorSchema = Schema.Struct({ threadId: Schema.String, + /** + * Imported conversations must never degrade into a fresh thread when the + * source session has disappeared. Existing T3-created sessions retain the + * historical recovery fallback for backwards compatibility. + */ + requireExistingThread: Schema.optional(Schema.Boolean), }); const CodexUserInputAnswerObject = Schema.Struct({ answers: Schema.Array(Schema.String), }); -const isCodexResumeCursorSchema = Schema.is(CodexResumeCursorSchema); +export const isCodexResumeCursor = Schema.is(CodexResumeCursorSchema); const isCodexUserInputAnswerObject = Schema.is(CodexUserInputAnswerObject); // TODO: Verify `packages/effect-codex-app-server/scripts/generate.ts` so the generated @@ -258,7 +264,7 @@ function normalizeCodexModelSlug( function readResumeCursorThreadId( resumeCursor: ProviderSession["resumeCursor"], ): string | undefined { - return isCodexResumeCursorSchema(resumeCursor) ? resumeCursor.threadId : undefined; + return isCodexResumeCursor(resumeCursor) ? resumeCursor.threadId : undefined; } function runtimeModeToThreadConfig(input: RuntimeMode): { @@ -462,6 +468,7 @@ export const openCodexThread = (input: { readonly requestedModel: string | undefined; readonly serviceTier: CodexServiceTier | undefined; readonly resumeThreadId: string | undefined; + readonly allowResumeFallback?: boolean; }): Effect.Effect => { const resumeThreadId = input.resumeThreadId; const startParams = buildThreadStartParams({ @@ -475,22 +482,24 @@ export const openCodexThread = (input: { return input.client.request("thread/start", startParams); } - return input.client - .request("thread/resume", { - threadId: resumeThreadId, - ...startParams, - }) - .pipe( - Effect.catchIf(isRecoverableThreadResumeError, (error) => - Effect.logWarning("codex app-server thread resume fell back to fresh start", { - threadId: input.threadId, - requestedRuntimeMode: input.runtimeMode, - resumeThreadId, - recoverable: true, - cause: error, - }).pipe(Effect.andThen(input.client.request("thread/start", startParams))), - ), - ); + const resume = input.client.request("thread/resume", { + threadId: resumeThreadId, + ...startParams, + }); + if (input.allowResumeFallback === false) { + return resume; + } + return resume.pipe( + Effect.catchIf(isRecoverableThreadResumeError, (error) => + Effect.logWarning("codex app-server thread resume fell back to fresh start", { + threadId: input.threadId, + requestedRuntimeMode: input.runtimeMode, + resumeThreadId, + recoverable: true, + cause: error, + }).pipe(Effect.andThen(input.client.request("thread/start", startParams))), + ), + ); }; function readNotificationThreadId(notification: CodexServerNotification): string | undefined { @@ -1227,6 +1236,7 @@ export const makeCodexSessionRuntime = ( requestedModel, serviceTier: options.serviceTier, resumeThreadId: readResumeCursorThreadId(options.resumeCursor), + allowResumeFallback: options.resumeCursor?.requireExistingThread !== true, }); const providerThreadId = opened.thread.id; diff --git a/apps/server/src/provider/Layers/CodexThreadHistory.ts b/apps/server/src/provider/Layers/CodexThreadHistory.ts new file mode 100644 index 00000000000..f8873f279cc --- /dev/null +++ b/apps/server/src/provider/Layers/CodexThreadHistory.ts @@ -0,0 +1,295 @@ +/** + * CodexThreadHistory - read-only discovery of persisted Codex conversations. + * + * This deliberately talks to Codex through its app-server protocol instead of + * reading JSONL rollout files directly. That keeps T3 compatible with Codex's + * own storage changes and, importantly, never modifies a Codex conversation + * file. Discovery may update Codex's own derived state database as part of + * Codex's normal read path, which is why this bridge does not promise a + * filesystem-level "zero write" guarantee. + * + * @module provider/Layers/CodexThreadHistory + */ +import type { CodexSettings } from "@t3tools/contracts"; +import { resolveSpawnCommand } from "@t3tools/shared/shell"; +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 ChildProcess from "effect/unstable/process/ChildProcess"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; +import * as CodexClient from "effect-codex-app-server/client"; +import * as CodexErrors from "effect-codex-app-server/errors"; +import * as CodexSchema from "effect-codex-app-server/schema"; + +import { expandHomePath } from "../../pathExpansion.ts"; +import type { + ProviderThreadHistory, + ProviderThreadHistoryCandidate, + ProviderThreadHistorySource, +} from "../ProviderDriver.ts"; +import { ProviderThreadHistoryError } from "../Errors.ts"; +import { buildCodexInitializeParams } from "./CodexProvider.ts"; +import { codexAppServerArgs, resolveCodexLaunchArgs } from "./codexLaunchArgs.ts"; + +const CODEX_THREAD_DISCOVERY_PAGE_SIZE = 100; +const CODEX_THREAD_DISCOVERY_MAX_RESULTS = 500; +// Read one additional session so an exact 500-result list is not incorrectly +// labelled as truncated. Only the first 500 are returned to the client. +const CODEX_THREAD_DISCOVERY_SCAN_LIMIT = CODEX_THREAD_DISCOVERY_MAX_RESULTS + 1; +const CODEX_THREAD_HISTORY_MAX_MESSAGES = 2_000; +const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const; + +type CodexAppServerClient = CodexClient.CodexAppServerClient["Service"]; +type CodexThreadListEntry = CodexSchema.V2ThreadListResponse["data"][number]; +type CodexThreadReadEntry = CodexSchema.V2ThreadReadResponse["thread"]; + +function toIsoTimestamp(value: number): string { + const timestamp = DateTime.make(Number.isFinite(value) ? value * 1_000 : 0); + return DateTime.formatIso(Option.getOrElse(timestamp, () => DateTime.makeUnsafe(0))); +} + +function trimToNonEmpty(value: string | null | undefined): string | undefined { + const trimmed = value?.trim(); + return trimmed && trimmed.length > 0 ? trimmed : undefined; +} + +function titleFromThread(input: { + readonly name?: string | null; + readonly preview: string; +}): string { + const title = trimToNonEmpty(input.name); + if (title) return title.slice(0, 160); + const preview = input.preview.trim().split(/\r?\n/, 1)[0]?.trim(); + return preview && preview.length > 0 ? preview.slice(0, 160) : "Untitled Codex session"; +} + +function toThreadHistoryError(operation: "list" | "read") { + return (cause: CodexErrors.CodexAppServerError) => + new ProviderThreadHistoryError({ + provider: "codex", + operation, + detail: "Codex app-server could not read persisted thread history.", + cause, + }); +} + +function sourceLabel(source: CodexThreadListEntry["source"]): string { + if (typeof source === "string") return source; + if ("custom" in source) return trimToNonEmpty(source.custom) ?? "custom"; + return "sub-agent"; +} + +function toCandidate( + thread: CodexThreadListEntry, + archived: boolean, +): ProviderThreadHistoryCandidate | undefined { + if ( + thread.ephemeral || + thread.parentThreadId || + thread.source === "exec" || + (typeof thread.source !== "string" && "subAgent" in thread.source) + ) { + return undefined; + } + return { + externalThreadId: thread.id, + title: titleFromThread(thread), + preview: thread.preview, + createdAt: toIsoTimestamp(thread.createdAt), + updatedAt: toIsoTimestamp(thread.updatedAt), + source: sourceLabel(thread.source), + archived, + }; +} + +function textFromUserMessage( + content: ReadonlyArray, +): string { + const parts: string[] = []; + for (const input of content) { + switch (input.type) { + case "text": + if (input.text.trim().length > 0) parts.push(input.text); + break; + case "image": + case "localImage": + parts.push("[Image attachment]"); + break; + case "audio": + case "localAudio": + parts.push("[Audio attachment]"); + break; + case "skill": + parts.push(`[Skill: ${input.name}]`); + break; + case "mention": + parts.push(`@${input.name}`); + break; + } + } + return parts.join("\n").trim(); +} + +function toHistory(thread: CodexThreadReadEntry): ProviderThreadHistory { + const messages: ProviderThreadHistory["messages"][number][] = []; + for (const turn of thread.turns) { + const createdAt = toIsoTimestamp(turn.startedAt ?? thread.createdAt); + for (const item of turn.items) { + if (item.type === "userMessage") { + const text = textFromUserMessage(item.content); + if (text.length > 0) { + messages.push({ + externalMessageId: item.id, + role: "user", + text, + createdAt, + }); + } + continue; + } + if (item.type === "agentMessage" && item.text.trim().length > 0) { + messages.push({ + externalMessageId: item.id, + role: "assistant", + text: item.text, + createdAt, + }); + } + } + } + return { + externalThreadId: thread.id, + title: titleFromThread(thread), + createdAt: toIsoTimestamp(thread.createdAt), + messages: messages.slice(-CODEX_THREAD_HISTORY_MAX_MESSAGES), + }; +} + +function makeAppServerClientRunner(input: { + readonly config: CodexSettings; + readonly environment: NodeJS.ProcessEnv; + readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; +}) { + return ( + cwd: string, + run: (client: CodexAppServerClient) => Effect.Effect, + ): Effect.Effect => + Effect.scoped( + Effect.gen(function* () { + const homePath = input.config.homePath ? expandHomePath(input.config.homePath) : undefined; + const environment = { + ...input.environment, + ...(homePath ? { CODEX_HOME: homePath } : {}), + }; + const spawnCommand = yield* resolveSpawnCommand( + input.config.binaryPath, + codexAppServerArgs(resolveCodexLaunchArgs(input.config.launchArgs, input.environment)), + { env: environment, extendEnv: true }, + ); + const child = yield* input.spawner + .spawn( + ChildProcess.make(spawnCommand.command, spawnCommand.args, { + cwd, + env: environment, + extendEnv: true, + forceKillAfter: CODEX_APP_SERVER_FORCE_KILL_AFTER, + shell: spawnCommand.shell, + }), + ) + .pipe( + Effect.mapError( + (cause) => + new CodexErrors.CodexAppServerSpawnError({ + command: `${input.config.binaryPath} app-server`, + cause, + }), + ), + ); + const clientContext = yield* Layer.build(CodexClient.layerChildProcess(child)); + const client = yield* Effect.service(CodexClient.CodexAppServerClient).pipe( + Effect.provide(clientContext), + ); + yield* client.request("initialize", buildCodexInitializeParams()); + yield* client.notify("initialized", undefined); + return yield* run(client); + }), + ); +} + +/** + * Build a read-only session-history source for one configured Codex instance. + * The driver captures its effective (including shadow-home) configuration, so + * discovery always sees the exact same Codex home as normal T3 continuations. + */ +export function makeCodexThreadHistory(input: { + readonly config: CodexSettings; + readonly environment: NodeJS.ProcessEnv; + readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; +}): ProviderThreadHistorySource { + const withClient = makeAppServerClientRunner(input); + + const listThreads: ProviderThreadHistorySource["listThreads"] = ({ cwd }) => + withClient(cwd, (client) => + Effect.gen(function* () { + const threads: ProviderThreadHistoryCandidate[] = []; + let truncated = false; + for (const archived of [false, true] as const) { + let cursor: string | undefined; + do { + const remaining = CODEX_THREAD_DISCOVERY_SCAN_LIMIT - threads.length; + if (remaining <= 0) { + truncated = true; + break; + } + const response = yield* client.request("thread/list", { + archived, + cwd, + limit: Math.min(CODEX_THREAD_DISCOVERY_PAGE_SIZE, remaining), + sortKey: "recency_at", + sortDirection: "desc", + sourceKinds: ["cli", "vscode", "appServer", "unknown"], + ...(cursor ? { cursor } : {}), + }); + for (const thread of response.data) { + const candidate = toCandidate(thread, archived); + if (candidate) threads.push(candidate); + } + if (threads.length > CODEX_THREAD_DISCOVERY_MAX_RESULTS) { + truncated = true; + break; + } + cursor = response.nextCursor ?? undefined; + } while (cursor); + if (truncated) break; + } + threads.sort((left, right) => right.updatedAt.localeCompare(left.updatedAt)); + return { + threads: threads.slice(0, CODEX_THREAD_DISCOVERY_MAX_RESULTS), + truncated, + }; + }), + ).pipe(Effect.mapError(toThreadHistoryError("list"))); + + const readThreads: ProviderThreadHistorySource["readThreads"] = ({ cwd, externalThreadIds }) => + withClient(cwd, (client) => + Effect.forEach( + externalThreadIds, + (externalThreadId) => + client + .request("thread/read", { + threadId: externalThreadId, + includeTurns: true, + }) + .pipe(Effect.map((response) => toHistory(response.thread))), + { concurrency: 1 }, + ), + ).pipe(Effect.mapError(toThreadHistoryError("read"))); + + return { listThreads, readThreads }; +} + +export const CodexThreadHistoryConstants = { + discoveryMaxResults: CODEX_THREAD_DISCOVERY_MAX_RESULTS, + historyMaxMessages: CODEX_THREAD_HISTORY_MAX_MESSAGES, +} as const; diff --git a/apps/server/src/provider/ProviderDriver.ts b/apps/server/src/provider/ProviderDriver.ts index c738882c23a..86c3b96aee6 100644 --- a/apps/server/src/provider/ProviderDriver.ts +++ b/apps/server/src/provider/ProviderDriver.ts @@ -31,7 +31,11 @@ import type * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; import type * as TextGeneration from "../textGeneration/TextGeneration.ts"; -import type { ProviderAdapterError, ProviderDriverError } from "./Errors.ts"; +import type { + ProviderAdapterError, + ProviderDriverError, + ProviderThreadHistoryError, +} from "./Errors.ts"; import type { ProviderAdapterShape } from "./Services/ProviderAdapter.ts"; import type { ServerProviderShape } from "./Services/ServerProvider.ts"; @@ -52,6 +56,54 @@ export interface ProviderDriverMetadata { readonly supportsMultipleInstances?: boolean; } +/** + * Read-only provider conversation metadata used to adopt an existing native + * session into T3. Providers opt in explicitly; the normal runtime adapter + * remains focused on active session lifecycle and turn delivery. + */ +export interface ProviderThreadHistoryCandidate { + readonly externalThreadId: string; + readonly title: string; + readonly preview: string; + readonly createdAt: string; + readonly updatedAt: string; + readonly source: string; + readonly archived: boolean; +} + +export interface ProviderThreadHistoryMessage { + readonly externalMessageId: string; + readonly role: "user" | "assistant"; + readonly text: string; + readonly createdAt: string; +} + +export interface ProviderThreadHistory { + readonly externalThreadId: string; + readonly title: string; + readonly createdAt: string; + readonly messages: ReadonlyArray; +} + +export interface ProviderThreadHistorySource { + readonly listThreads: (input: { readonly cwd: string }) => Effect.Effect< + { + readonly threads: ReadonlyArray; + readonly truncated: boolean; + }, + ProviderThreadHistoryError + >; + /** + * Read a selection through one provider connection. Import callers always + * batch this work so a multi-select does not spawn one native process per + * source thread. + */ + readonly readThreads: (input: { + readonly cwd: string; + readonly externalThreadIds: ReadonlyArray; + }) => Effect.Effect, ProviderThreadHistoryError>; +} + /** * One materialized provider instance. Held by the registry, looked up by * `instanceId`, torn down by closing the scope it was created in. @@ -71,6 +123,8 @@ export interface ProviderInstance { readonly snapshot: ServerProviderShape; readonly adapter: ProviderAdapterShape; readonly textGeneration: TextGeneration.TextGeneration["Service"]; + /** Optional read-only history bridge for importing native provider sessions. */ + readonly threadHistory?: ProviderThreadHistorySource; } export interface ProviderContinuationIdentity { diff --git a/apps/server/src/provider/Services/CodexSessionImport.ts b/apps/server/src/provider/Services/CodexSessionImport.ts new file mode 100644 index 00000000000..e73bb45e976 --- /dev/null +++ b/apps/server/src/provider/Services/CodexSessionImport.ts @@ -0,0 +1,34 @@ +/** + * CodexSessionImport - adoption of persisted Codex conversations. + * + * The import keeps the native conversation authoritative. T3 stores a + * text-only display snapshot and a strict resume reference; it never copies, + * moves, archives, or deletes the original Codex thread. + * + * @module provider/Services/CodexSessionImport + */ +import type { + CodexSessionImportError, + CodexSessionImportInput, + CodexSessionImportResult, + CodexSessionListInput, + CodexSessionListResult, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; + +export interface CodexSessionImportShape { + /** Discover importable sessions whose native workspace matches a T3 project. */ + readonly list: ( + input: CodexSessionListInput, + ) => Effect.Effect; + /** Adopt selected native sessions into the project's T3 thread list. */ + readonly import: ( + input: CodexSessionImportInput, + ) => Effect.Effect; +} + +export class CodexSessionImport extends Context.Service< + CodexSessionImport, + CodexSessionImportShape +>()("t3/provider/Services/CodexSessionImport") {} diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index fff71dbb4e7..51f751bcf92 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -9,6 +9,8 @@ import { AuthEnvironmentBootstrapTokenType, AuthTokenExchangeGrantType, CommandId, + type CodexSessionImportInput, + type CodexSessionListInput, DEFAULT_SERVER_SETTINGS, EnvironmentId, EventId, @@ -86,6 +88,7 @@ import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSna import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite.ts"; import { PersistenceSqlError } from "./persistence/Errors.ts"; import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts"; +import * as CodexSessionImport from "./provider/Services/CodexSessionImport.ts"; import { makeManualOnlyProviderMaintenanceCapabilities } from "./provider/providerMaintenance.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; @@ -330,6 +333,7 @@ const buildAppUnderTest = (options?: { layers?: { keybindings?: Partial; providerRegistry?: Partial; + codexSessionImport?: Partial; serverSettings?: Partial; externalLauncher?: Partial; vcsDriver?: Partial; @@ -565,18 +569,25 @@ const buildAppUnderTest = (options?: { }), ), Layer.provide( - Layer.mock(ProviderRegistry.ProviderRegistry)({ - getProviders: Effect.succeed([]), - refresh: () => Effect.succeed([]), - refreshInstance: () => Effect.succeed([]), - getProviderMaintenanceCapabilitiesForInstance: (_instanceId, provider) => - Effect.succeed( - makeManualOnlyProviderMaintenanceCapabilities({ provider, packageName: null }), - ), - setProviderMaintenanceActionState: () => Effect.succeed([]), - streamChanges: Stream.empty, - ...options?.layers?.providerRegistry, - }), + Layer.mergeAll( + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([]), + refresh: () => Effect.succeed([]), + refreshInstance: () => Effect.succeed([]), + getProviderMaintenanceCapabilitiesForInstance: (_instanceId, provider) => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ provider, packageName: null }), + ), + setProviderMaintenanceActionState: () => Effect.succeed([]), + streamChanges: Stream.empty, + ...options?.layers?.providerRegistry, + }), + Layer.mock(CodexSessionImport.CodexSessionImport)({ + list: () => Effect.succeed({ sessions: [], truncated: false }), + import: () => Effect.succeed({ importedThreadIds: [], alreadyImportedThreadIds: [] }), + ...options?.layers?.codexSessionImport, + }), + ), ), Layer.provide( Layer.mock(ServerSettings.ServerSettingsService)({ @@ -3895,6 +3906,75 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("routes websocket RPCs for importing Codex sessions", () => + Effect.gen(function* () { + const listInputs: CodexSessionListInput[] = []; + const importInputs: CodexSessionImportInput[] = []; + yield* buildAppUnderTest({ + layers: { + codexSessionImport: { + list: (input) => + Effect.sync(() => { + listInputs.push(input); + return { + sessions: [ + { + externalThreadId: "native-codex-thread", + title: "Existing Codex session", + preview: "Please continue this task.", + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-02T00:00:00.000Z", + source: "cli", + archived: false, + importedThreadId: null, + }, + ], + truncated: false, + }; + }), + import: (input) => + Effect.sync(() => { + importInputs.push(input); + return { + importedThreadIds: [ThreadId.make("codex:codex:native-codex-thread")], + alreadyImportedThreadIds: [], + }; + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const projectId = ProjectId.make("project-codex-session-import-rpc"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + const listed = yield* client[WS_METHODS.codexListSessions]({ + projectId, + providerInstanceId, + }); + const imported = yield* client[WS_METHODS.codexImportSessions]({ + projectId, + providerInstanceId, + externalThreadIds: ["native-codex-thread"], + }); + return { listed, imported }; + }), + ), + ); + + assert.equal(result.listed.sessions[0]?.externalThreadId, "native-codex-thread"); + assert.deepEqual(result.imported.importedThreadIds, [ + ThreadId.make("codex:codex:native-codex-thread"), + ]); + assert.deepEqual(listInputs, [{ projectId, providerInstanceId }]); + assert.deepEqual(importInputs, [ + { projectId, providerInstanceId, externalThreadIds: ["native-codex-thread"] }, + ]); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("does not block server config when editor discovery never resolves", () => Effect.gen(function* () { const discoveryInterrupted = yield* Deferred.make(); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 853bb0b1101..9f362648e34 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -25,6 +25,7 @@ import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/ import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory.ts"; +import { CodexSessionImportLive } from "./provider/Layers/CodexSessionImport.ts"; import * as ProviderSessionRuntime from "./persistence/ProviderSessionRuntime.ts"; import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry.ts"; import * as ProviderEventLoggers from "./provider/Layers/ProviderEventLoggers.ts"; @@ -396,7 +397,7 @@ const RuntimeDependenciesLive = RuntimeCoreDependenciesLive.pipe( Layer.provide(NetService.layer), ); -const RuntimeServicesLive = ServerRuntimeStartup.layer.pipe( +const RuntimeServicesLive = Layer.mergeAll(ServerRuntimeStartup.layer, CodexSessionImportLive).pipe( Layer.provideMerge(RuntimeDependenciesLive), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 06888ef3f70..39923d73f79 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -79,6 +79,7 @@ import { observeRpcStreamEffect as instrumentRpcStreamEffect, } from "./observability/RpcInstrumentation.ts"; import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts"; +import * as CodexSessionImport from "./provider/Services/CodexSessionImport.ts"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import * as ServerSelfUpdate from "./cloud/selfUpdate.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; @@ -359,6 +360,7 @@ const makeWsRpcLayer = ( const previewManager = yield* PreviewManager.PreviewManager; const portDiscovery = yield* PortScanner.PortDiscovery; const providerRegistry = yield* ProviderRegistry.ProviderRegistry; + const codexSessionImport = yield* CodexSessionImport.CodexSessionImport; const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const config = yield* ServerConfig.ServerConfig; @@ -1357,6 +1359,14 @@ const makeWsRpcLayer = ( observeRpcEffect(WS_METHODS.serverProbe, Effect.succeed({}), { "rpc.aggregate": "server", }), + [WS_METHODS.codexListSessions]: (input) => + observeRpcEffect(WS_METHODS.codexListSessions, codexSessionImport.list(input), { + "rpc.aggregate": "codex-session-import", + }), + [WS_METHODS.codexImportSessions]: (input) => + observeRpcEffect(WS_METHODS.codexImportSessions, codexSessionImport.import(input), { + "rpc.aggregate": "codex-session-import", + }), [WS_METHODS.serverGetConfig]: (_input) => observeRpcEffect(WS_METHODS.serverGetConfig, loadServerConfig, { "rpc.aggregate": "server", diff --git a/apps/web/src/components/CodexSessionImportDialog.tsx b/apps/web/src/components/CodexSessionImportDialog.tsx new file mode 100644 index 00000000000..58eb26ba285 --- /dev/null +++ b/apps/web/src/components/CodexSessionImportDialog.tsx @@ -0,0 +1,481 @@ +import { useAtomValue } from "@effect/atom-react"; +import type { ProviderInstanceId, ServerProvider } from "@t3tools/contracts"; +import { + ArchiveIcon, + CheckCircle2Icon, + CircleAlertIcon, + FolderIcon, + Link2Icon, + RefreshCwIcon, +} from "lucide-react"; +import { useEffect, useMemo, useRef, useState } from "react"; + +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "@t3tools/client-runtime/state/runtime"; +import type { SidebarProjectGroupMember } from "../sidebarProjectGrouping"; +import { formatRelativeTimeLabel } from "../timestampFormat"; +import { codexSessionEnvironment } from "../state/codexSessions"; +import { useEnvironmentQuery } from "../state/query"; +import { environmentServerConfigsAtom } from "../state/server"; +import { useAtomCommand } from "../state/use-atom-command"; +import { cn } from "../lib/utils"; +import { Badge } from "./ui/badge"; +import { Button } from "./ui/button"; +import { Checkbox } from "./ui/checkbox"; +import { + Dialog, + DialogDescription, + DialogFooter, + DialogHeader, + DialogPanel, + DialogPopup, + DialogTitle, +} from "./ui/dialog"; +import { Select, SelectItem, SelectPopup, SelectTrigger, SelectValue } from "./ui/select"; +import { toastManager } from "./ui/toast"; + +const INITIAL_SESSION_LIST_COUNT = 100; +const MAX_SESSION_IMPORT_SELECTION = 50; +const CODEX_PROVIDER_DRIVER = "codex"; + +function availableCodexProviders( + providers: ReadonlyArray, +): ReadonlyArray { + return providers.filter( + (provider) => + provider.driver === CODEX_PROVIDER_DRIVER && + provider.enabled && + provider.installed && + provider.status !== "disabled" && + provider.availability !== "unavailable", + ); +} + +function providerLabel(provider: ServerProvider): string { + return provider.displayName ?? "Codex"; +} + +function sessionCountLabel(count: number): string { + return `${count} Codex session${count === 1 ? "" : "s"}`; +} + +interface CodexSessionImportDialogProps { + readonly open: boolean; + readonly onOpenChange: (open: boolean) => void; + readonly project: SidebarProjectGroupMember | null; +} + +/** + * A deliberately explicit import flow: users choose which native Codex + * conversations become T3 threads, while the source data stays in Codex. + */ +export function CodexSessionImportDialog(props: CodexSessionImportDialogProps) { + const serverConfigs = useAtomValue(environmentServerConfigsAtom); + const config = props.project ? serverConfigs.get(props.project.environmentId) : undefined; + const supportsSessionImport = config?.environment.capabilities.codexSessionImport === true; + const providers = useMemo( + () => availableCodexProviders(config?.providers ?? []), + [config?.providers], + ); + const [providerInstanceId, setProviderInstanceId] = useState(null); + const [selectedSessionIds, setSelectedSessionIds] = useState>( + () => new Set(), + ); + const [visibleCount, setVisibleCount] = useState(INITIAL_SESSION_LIST_COUNT); + const initializedProviderRef = useRef(null); + const importSessions = useAtomCommand(codexSessionEnvironment.import, { reportFailure: false }); + const [isImporting, setIsImporting] = useState(false); + + useEffect(() => { + setProviderInstanceId((current) => { + if (current && providers.some((provider) => provider.instanceId === current)) return current; + return providers.at(0)?.instanceId ?? null; + }); + }, [providers]); + + useEffect(() => { + if (props.open) return; + initializedProviderRef.current = null; + setSelectedSessionIds(new Set()); + setVisibleCount(INITIAL_SESSION_LIST_COUNT); + }, [props.open]); + + const queryAtom = useMemo(() => { + if (!props.open || !props.project || !providerInstanceId || !supportsSessionImport) return null; + return codexSessionEnvironment.list({ + environmentId: props.project.environmentId, + input: { + projectId: props.project.id, + providerInstanceId, + }, + }); + }, [props.open, props.project, providerInstanceId, supportsSessionImport]); + const sessions = useEnvironmentQuery(queryAtom); + const selectedProvider = + providers.find((provider) => provider.instanceId === providerInstanceId) ?? null; + const importableSessions = useMemo( + () => sessions.data?.sessions.filter((session) => session.importedThreadId === null) ?? [], + [sessions.data], + ); + const importableIds = useMemo( + () => new Set(importableSessions.map((session) => session.externalThreadId)), + [importableSessions], + ); + const initialSelectionIds = useMemo( + () => + new Set( + importableSessions + .slice(0, MAX_SESSION_IMPORT_SELECTION) + .map((session) => session.externalThreadId), + ), + [importableSessions], + ); + const visibleSessions = useMemo( + () => (sessions.data?.sessions ?? []).slice(0, visibleCount), + [sessions.data, visibleCount], + ); + + useEffect(() => { + if (!sessions.data || !providerInstanceId) return; + if (initializedProviderRef.current !== providerInstanceId) { + initializedProviderRef.current = providerInstanceId; + setSelectedSessionIds(new Set(initialSelectionIds)); + setVisibleCount(INITIAL_SESSION_LIST_COUNT); + return; + } + setSelectedSessionIds((current) => { + const next = new Set([...current].filter((id) => importableIds.has(id))); + return next.size === current.size ? current : next; + }); + }, [importableIds, initialSelectionIds, providerInstanceId, sessions.data]); + + const selectedIds = [...selectedSessionIds].filter((id) => importableIds.has(id)); + const selectedCount = selectedIds.length; + const selectionLimit = Math.min(importableSessions.length, MAX_SESSION_IMPORT_SELECTION); + const selectionLimitReached = selectedCount >= MAX_SESSION_IMPORT_SELECTION; + const allImportableSelected = importableSessions.length > 0 && selectedCount === selectionLimit; + + const toggleSession = (externalThreadId: string) => { + setSelectedSessionIds((current) => { + const next = new Set(current); + if (next.has(externalThreadId)) { + next.delete(externalThreadId); + } else { + const selectedImportableCount = [...next].filter((id) => importableIds.has(id)).length; + if (selectedImportableCount >= MAX_SESSION_IMPORT_SELECTION) return current; + next.add(externalThreadId); + } + return next; + }); + }; + + const handleImport = async () => { + if (!props.project || !providerInstanceId || selectedIds.length === 0 || isImporting) return; + setIsImporting(true); + const result = await importSessions({ + environmentId: props.project.environmentId, + input: { + projectId: props.project.id, + providerInstanceId, + externalThreadIds: selectedIds, + }, + }); + setIsImporting(false); + + if (result._tag === "Failure") { + if (!isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + toastManager.add({ + type: "error", + title: "Could not import Codex sessions", + description: error instanceof Error ? error.message : "Please refresh and try again.", + }); + } + return; + } + + const importedCount = result.value.importedThreadIds.length; + const alreadyImportedCount = result.value.alreadyImportedThreadIds.length; + toastManager.add({ + type: "success", + title: + importedCount > 0 + ? `Imported ${sessionCountLabel(importedCount)}` + : "Those Codex sessions are already in T3", + description: + importedCount > 0 + ? "The original sessions remain unchanged in Codex." + : `${alreadyImportedCount} selected session${alreadyImportedCount === 1 ? " is" : "s are"} already linked.`, + }); + initializedProviderRef.current = null; + setSelectedSessionIds(new Set()); + sessions.refresh(); + }; + + return ( + + + +
+ + + +
+ Import Codex sessions + + Choose native Codex conversations for this project. T3 stores a text snapshot and a + continuation link; the original session stays in Codex. + +
+
+ + + {props.project ? ( +
+ + + {props.project.workspaceRoot} + +
+ ) : null} + + {!supportsSessionImport ? ( +
+

This server needs an update

+

+ Update the T3 server for this environment before importing Codex sessions. +

+
+ ) : providers.length === 0 ? ( +
+

No enabled Codex provider is available

+

+ Add or enable a Codex provider for this environment, then reopen this dialog. +

+
+ ) : ( + <> + {providers.length > 1 && providerInstanceId ? ( + + ) : null} + +
+
+

Available sessions

+

+ New Codex sessions appear when you refresh this list. +

+
+ +
+ + {sessions.error ? ( +
+ +

{sessions.error}

+
+ ) : null} + + {sessions.data?.truncated ? ( +
+ +

+ Showing the 500 most recent matching sessions. Narrow your project history in + Codex if the session you need is not listed. +

+
+ ) : null} + + {sessions.isPending && sessions.data === null ? ( +
+ Reading Codex sessions… +
+ ) : null} + + {!sessions.isPending && sessions.data && sessions.data.sessions.length === 0 ? ( +
+ No Codex sessions match this project path yet. +
+ ) : null} + + {sessions.data && sessions.data.sessions.length > 0 ? ( +
+
+ + {selectedCount} of {importableSessions.length} new sessions selected + {importableSessions.length > MAX_SESSION_IMPORT_SELECTION + ? ` · up to ${MAX_SESSION_IMPORT_SELECTION} at a time` + : ""} + +
+ + +
+
+
+ {visibleSessions.map((session) => { + const imported = session.importedThreadId !== null; + const checked = selectedSessionIds.has(session.externalThreadId); + const selectionDisabled = + imported || isImporting || (!checked && selectionLimitReached); + return ( +
+ toggleSession(session.externalThreadId)} + /> + +
+ ); + })} + {sessions.data.sessions.length > visibleSessions.length ? ( +
+ +
+ ) : null} +
+
+ ) : null} + +

+ Imported history is a snapshot. Continuing a thread from T3 resumes its original + Codex conversation, but messages added only in Codex later are not automatically + mirrored into T3 yet. +

+ + )} + + + + + + + + ); +} diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index ee73b570514..8e53f476b0e 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -138,6 +138,7 @@ import { type SnoozePreset, } from "./Sidebar.snooze"; import { ProjectFavicon } from "./ProjectFavicon"; +import { CodexSessionImportDialog } from "./CodexSessionImportDialog"; import { ProviderInstanceIcon } from "./chat/ProviderInstanceIcon"; import { getTriggerDisplayModelLabel } from "./chat/providerIconUtils"; import { deriveProviderInstanceEntries, type ProviderInstanceEntry } from "../providerInstances"; @@ -1231,6 +1232,9 @@ export default function SidebarV2() { const [projectActionsTarget, setProjectActionsTarget] = useState( null, ); + const [codexImportTarget, setCodexImportTarget] = useState( + null, + ); const [projectScopeMenuOpen, setProjectScopeMenuOpen] = useState(false); const newThreadContext = useHandleNewThread(); const openAddProjectCommandPalette = useCallback( @@ -3137,6 +3141,30 @@ export default function SidebarV2() { + {serverConfigs.get(member.environmentId)?.environment.capabilities + .codexSessionImport === true ? ( +
+
+

Codex sessions

+

+ Import existing conversations from Codex without changing their source + data. +

+
+ +
+ ) : null} {projectActionsTarget.memberProjects.length > 1 ? (