diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 707c87c43c9..0618da489b2 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -111,6 +111,7 @@ function createProviderServiceHarness( stopSession: () => unsupported(), listSessions, getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), + readThread: () => unsupported(), getInstanceInfo: (instanceId) => Effect.succeed({ instanceId, diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 926182a3ef0..08f66fe172b 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1432,6 +1432,113 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("projects an imported transcript into thread messages", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const now = "2026-01-01T00:00:00.000Z"; + const threadId = ThreadId.make("thread-imported-transcript"); + + yield* eventStore.append({ + type: "thread.created", + eventId: EventId.make("evt-it1"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-it1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-it1"), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-imported-transcript"), + title: "Imported transcript", + modelSelection: { + instanceId: ProviderInstanceId.make("claudeAgent"), + model: "sonnet", + }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.messages-imported", + eventId: EventId.make("evt-it2"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-01-01T00:00:01.000Z", + commandId: CommandId.make("cmd-it2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-it2"), + metadata: {}, + payload: { + threadId, + messages: [ + { + messageId: MessageId.make( + "imported:claudeAgent:thread-imported-transcript:000000:uuid-1", + ), + role: "user", + text: "Fix the flaky test", + createdAt: "2026-01-01T00:00:01.000Z", + updatedAt: "2026-01-01T00:00:01.000Z", + }, + { + messageId: MessageId.make( + "imported:claudeAgent:thread-imported-transcript:000001:uuid-2", + ), + role: "assistant", + text: "Looking at it", + createdAt: "2026-01-01T00:00:01.000Z", + updatedAt: "2026-01-01T00:00:01.000Z", + }, + ], + }, + }); + + yield* projectionPipeline.bootstrap; + + const messageRows = yield* sql<{ + readonly messageId: string; + readonly role: string; + readonly text: string; + readonly turnId: string | null; + readonly isStreaming: number; + }>` + SELECT + message_id AS "messageId", + role, + text, + turn_id AS "turnId", + is_streaming AS "isStreaming" + FROM projection_thread_messages + WHERE thread_id = ${threadId} + ORDER BY message_id ASC + `; + assert.deepEqual(messageRows, [ + { + messageId: "imported:claudeAgent:thread-imported-transcript:000000:uuid-1", + role: "user", + text: "Fix the flaky test", + turnId: null, + isStreaming: 0, + }, + { + messageId: "imported:claudeAgent:thread-imported-transcript:000001:uuid-2", + role: "assistant", + text: "Looking at it", + turnId: null, + isStreaming: 0, + }, + ]); + }), + ); + it.effect("settles a superseded running turn when a new turn becomes active", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 1f24a4a0200..a776f02251d 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -782,6 +782,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } case "thread.message-sent": + case "thread.messages-imported": case "thread.proposed-plan-upserted": case "thread.activity-appended": case "thread.approval-response-requested": @@ -916,6 +917,25 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.messages-imported": { + yield* Effect.forEach( + event.payload.messages, + (message) => + projectionThreadMessageRepository.upsert({ + messageId: message.messageId, + threadId: event.payload.threadId, + turnId: null, + role: message.role, + text: message.text, + isStreaming: false, + createdAt: message.createdAt, + updatedAt: message.updatedAt, + }), + { concurrency: 1 }, + ).pipe(Effect.asVoid); + return; + } + case "thread.reverted": { const existingRows = yield* projectionThreadMessageRepository.listByThreadId({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index c49646b7a4b..c0f1916f41a 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -335,6 +335,7 @@ describe("ProviderCommandReactor", () => { }, }); }, + readThread: () => unsupported(), rollbackConversation: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index b6bff8c766a..135f0410a08 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -62,7 +62,7 @@ function toNonEmptyProviderInput(value: string | undefined): string | undefined return normalized && normalized.length > 0 ? normalized : undefined; } -function mapProviderSessionStatusToOrchestrationStatus( +export function mapProviderSessionStatusToOrchestrationStatus( status: "connecting" | "ready" | "running" | "error" | "closed", ): OrchestrationSession["status"] { switch (status) { diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 74ece50cd31..30f0eabcc67 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -120,6 +120,7 @@ function createProviderServiceHarness() { }, }); }, + readThread: () => unsupported(), rollbackConversation: () => unsupported(), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 3b558d24739..c8c11973f32 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -14,6 +14,7 @@ import { ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema, + ThreadMessagesImportedPayload as ContractsThreadMessagesImportedPayloadSchema, ThreadProposedPlanUpsertedPayload as ContractsThreadProposedPlanUpsertedPayloadSchema, ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema, ThreadTurnDiffCompletedPayload as ContractsThreadTurnDiffCompletedPayloadSchema, @@ -44,6 +45,7 @@ export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema; +export const MessagesImportedPayloadSchema = ContractsThreadMessagesImportedPayloadSchema; export const ThreadProposedPlanUpsertedPayload = ContractsThreadProposedPlanUpsertedPayloadSchema; export const ThreadSessionSetPayload = ContractsThreadSessionSetPayloadSchema; export const ThreadTurnDiffCompletedPayload = ContractsThreadTurnDiffCompletedPayloadSchema; diff --git a/apps/server/src/orchestration/decider.projectScripts.test.ts b/apps/server/src/orchestration/decider.projectScripts.test.ts index a0c06840733..fa3e2b12aed 100644 --- a/apps/server/src/orchestration/decider.projectScripts.test.ts +++ b/apps/server/src/orchestration/decider.projectScripts.test.ts @@ -457,4 +457,99 @@ it.layer(NodeServices.layer)("decider project scripts", (it) => { }); }), ); + + it.effect("emits thread.messages-imported from thread.messages.import", () => + Effect.gen(function* () { + const now = "2026-01-01T00:00:00.000Z"; + const initial = createEmptyReadModel(now); + const withProject = yield* projectEvent(initial, { + sequence: 1, + eventId: asEventId("evt-project-create"), + aggregateKind: "project", + aggregateId: asProjectId("project-1"), + type: "project.created", + occurredAt: now, + commandId: CommandId.make("cmd-project-create"), + causationEventId: null, + correlationId: CommandId.make("cmd-project-create"), + metadata: {}, + payload: { + projectId: asProjectId("project-1"), + title: "Project", + workspaceRoot: "/tmp/project", + defaultModelSelection: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + const readModel = yield* projectEvent(withProject, { + sequence: 2, + eventId: asEventId("evt-thread-create"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.created", + occurredAt: now, + commandId: CommandId.make("cmd-thread-create"), + causationEventId: null, + correlationId: CommandId.make("cmd-thread-create"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-1"), + projectId: asProjectId("project-1"), + title: "Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("claudeAgent"), + model: "sonnet", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + const messages = [ + { + messageId: asMessageId("imported:claudeAgent:thread-1:000000:uuid-1"), + role: "user" as const, + text: "hello", + createdAt: now, + updatedAt: now, + }, + { + messageId: asMessageId("imported:claudeAgent:thread-1:000001:uuid-2"), + role: "assistant" as const, + text: "hi", + createdAt: now, + updatedAt: now, + }, + ]; + + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.messages.import", + commandId: CommandId.make("cmd-messages-import"), + threadId: ThreadId.make("thread-1"), + messages, + createdAt: now, + }, + readModel, + }); + + const singleResult = Array.isArray(result) ? null : result; + if (singleResult === null) { + throw new Error("Expected a single messages-imported event."); + } + expect(singleResult).toMatchObject({ + type: "thread.messages-imported", + payload: { + threadId: ThreadId.make("thread-1"), + messages, + }, + }); + }), + ); }); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 100369ae6e3..81269f67b0b 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -940,6 +940,27 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.messages.import": { + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.messages-imported", + payload: { + threadId: command.threadId, + messages: command.messages, + }, + }; + } + case "thread.session.set": { const thread = yield* requireThread({ readModel, diff --git a/apps/server/src/orchestration/importThread.test.ts b/apps/server/src/orchestration/importThread.test.ts new file mode 100644 index 00000000000..33afa8c03d9 --- /dev/null +++ b/apps/server/src/orchestration/importThread.test.ts @@ -0,0 +1,271 @@ +import { + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationCommand, + type OrchestrationProject, + type ProviderSession, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; + +import { makeImportThread } from "./importThread.ts"; + +const projectId = ProjectId.make("project-1"); +const workspaceRoot = "/home/dev/app"; +const codexInstanceId = ProviderInstanceId.make("codex"); +const cursorInstanceId = ProviderInstanceId.make("cursor"); +const codexModelSelection = { instanceId: codexInstanceId, model: "gpt-5-codex" }; +const lastActivityAtMs = 1767225600000; + +const project: OrchestrationProject = { + id: projectId, + title: "App", + workspaceRoot, + defaultModelSelection: null, + scripts: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + deletedAt: null, +}; + +const testCrypto = Crypto.make({ + randomBytes: (size) => new Uint8Array(size), + digest: (_algorithm, data) => Effect.succeed(data), +}); + +const codexSession: ProviderSession = { + threadId: ThreadId.make("unused"), + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + status: "ready", + runtimeMode: "full-access", + cwd: workspaceRoot, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", +}; + +const codexTranscript = [ + { type: "userMessage", id: "item-1", content: [{ type: "text", text: "Ship it" }] }, + { type: "agentMessage", id: "item-2", text: "Shipped" }, +]; + +function makeHarness(options?: { + readonly projectMissing?: boolean; + readonly instanceId?: ProviderInstanceId; + readonly driverKind?: string; + readonly items?: ReadonlyArray; + readonly snapshotWorkspaceRoot?: string; + readonly startSessionNeverSettles?: boolean; +}) { + const dispatched: OrchestrationCommand[] = []; + const instanceId = options?.instanceId ?? codexInstanceId; + const driverKind = ProviderDriverKind.make(options?.driverKind ?? "codex"); + const importSession = makeImportThread({ + crypto: testCrypto, + orchestrationEngine: { + dispatch: (command) => { + dispatched.push(command); + return Effect.succeed({ sequence: dispatched.length }); + }, + }, + projectionSnapshotQuery: { + getProjectShellById: () => + Effect.succeed(options?.projectMissing === true ? Option.none() : Option.some(project)), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.some(project)), + }, + providerService: { + getInstanceInfo: () => + Effect.succeed({ + instanceId, + driverKind, + displayName: undefined, + enabled: true, + continuationIdentity: { + driverKind, + continuationKey: `${driverKind}:instance:${instanceId}`, + }, + }), + startSession: (threadId) => + options?.startSessionNeverSettles === true + ? Effect.never + : Effect.succeed({ ...codexSession, threadId }), + readThread: (threadId) => + Effect.succeed({ + threadId, + workspaceRoot: options?.snapshotWorkspaceRoot ?? workspaceRoot, + lastActivityAtMs, + turns: [{ id: TurnId.make("turn-1"), items: options?.items ?? [] }], + }), + }, + }); + return { dispatched, importSession }; +} + +const dispatchedTypes = (dispatched: ReadonlyArray) => + dispatched.map((command) => command.type); + +describe("importThread", () => { + it.effect("creates a thread, imports the transcript, then binds the resumed session", () => + Effect.gen(function* () { + const harness = makeHarness({ items: codexTranscript }); + + const result = yield* harness.importSession.importThread({ + projectId, + modelSelection: codexModelSelection, + externalId: "codex-thread-1", + }); + + expect(dispatchedTypes(harness.dispatched)).toEqual([ + "thread.create", + "thread.messages.import", + "thread.session.set", + ]); + const created = harness.dispatched[0]; + expect(created?.type === "thread.create" && created.projectId).toBe(projectId); + expect(created?.type === "thread.create" && created.threadId).toBe(result.threadId); + const imported = harness.dispatched[1]; + expect( + imported?.type === "thread.messages.import" && + imported.messages.map((message) => [message.role, message.text]), + ).toEqual([ + ["user", "Ship it"], + ["assistant", "Shipped"], + ]); + }), + ); + + it.effect("stamps imported messages with the provider's own activity time", () => + Effect.gen(function* () { + const harness = makeHarness({ items: codexTranscript }); + + yield* harness.importSession.importThread({ + projectId, + modelSelection: codexModelSelection, + externalId: "codex-thread-1", + }); + + const imported = harness.dispatched[1]; + const expectedCreatedAt = DateTime.formatIso(DateTime.makeUnsafe(lastActivityAtMs)); + expect( + imported?.type === "thread.messages.import" && + imported.messages.every((message) => message.createdAt === expectedCreatedAt), + ).toBe(true); + }), + ); + + it.effect("refuses a codex thread that ran in another workspace", () => + Effect.gen(function* () { + const harness = makeHarness({ + items: codexTranscript, + snapshotWorkspaceRoot: "/home/dev/other", + }); + + const failure = yield* Effect.flip( + harness.importSession.importThread({ + projectId, + modelSelection: codexModelSelection, + externalId: "codex-thread-elsewhere", + }), + ); + + expect(failure.message).toContain("ran in /home/dev/other"); + expect(dispatchedTypes(harness.dispatched)).toEqual(["thread.create", "thread.delete"]); + }), + ); + + it.effect("deletes the thread it created when the session has nothing to import", () => + Effect.gen(function* () { + const harness = makeHarness({ items: [] }); + + const failure = yield* Effect.flip( + harness.importSession.importThread({ + projectId, + modelSelection: codexModelSelection, + externalId: "codex-thread-empty", + }), + ); + + expect(failure.message).toContain("has no conversation to import"); + expect(dispatchedTypes(harness.dispatched)).toEqual(["thread.create", "thread.delete"]); + }), + ); + + it.effect("deletes the thread it created when the import is interrupted", () => + Effect.gen(function* () { + const harness = makeHarness({ items: codexTranscript, startSessionNeverSettles: true }); + + const fiber = yield* Effect.forkChild( + harness.importSession.importThread({ + projectId, + modelSelection: codexModelSelection, + externalId: "codex-thread-interrupted", + }), + ); + yield* Effect.yieldNow; + yield* Fiber.interrupt(fiber); + + expect(dispatchedTypes(harness.dispatched)).toEqual(["thread.create", "thread.delete"]); + }), + ); + + it.effect("refuses to import into a project that no longer exists", () => + Effect.gen(function* () { + const harness = makeHarness({ projectMissing: true }); + + const failure = yield* Effect.flip( + harness.importSession.importThread({ + projectId, + modelSelection: codexModelSelection, + externalId: "codex-thread-1", + }), + ); + + expect(failure.message).toContain("no longer exists"); + expect(harness.dispatched).toEqual([]); + }), + ); + + it.effect("refuses drivers other than Claude Code and Codex", () => + Effect.gen(function* () { + const harness = makeHarness({ instanceId: cursorInstanceId, driverKind: "cursor" }); + + const failure = yield* Effect.flip( + harness.importSession.importThread({ + projectId, + modelSelection: { instanceId: cursorInstanceId, model: "auto" }, + externalId: "cursor-thread-1", + }), + ); + + expect(failure.message).toContain("only supported for Claude Code and Codex"); + expect(harness.dispatched).toEqual([]); + }), + ); +}); + +describe("resolveImportSession", () => { + it.effect("reports no workspace for providers that cannot be located on disk", () => + Effect.gen(function* () { + const harness = makeHarness(); + + const resolved = yield* harness.importSession.resolveImportSession({ + instanceId: codexInstanceId, + externalId: "codex-thread-1", + }); + + expect(resolved).toEqual({ + externalId: "codex-thread-1", + workspaceRoot: null, + projectId: null, + title: null, + }); + }), + ); +}); diff --git a/apps/server/src/orchestration/importThread.ts b/apps/server/src/orchestration/importThread.ts new file mode 100644 index 00000000000..28f27874959 --- /dev/null +++ b/apps/server/src/orchestration/importThread.ts @@ -0,0 +1,379 @@ +import { getSessionInfo, getSessionMessages } from "@anthropic-ai/claude-agent-sdk"; +import { + CommandId, + DEFAULT_PROVIDER_INTERACTION_MODE, + DEFAULT_RUNTIME_MODE, + OrchestrationImportThreadError, + ThreadId, + type OrchestrationImportThreadInput, + type OrchestrationImportThreadResult, + type OrchestrationResolveImportSessionInput, + type OrchestrationResolveImportSessionResult, + ProviderDriverKind, + type ProviderInstanceId, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; + +import type { ProviderServiceShape } from "../provider/Services/ProviderService.ts"; +import { mapClaudeSessionMessages, mapCodexThreadSnapshot } from "./importedMessages.ts"; +import { mapProviderSessionStatusToOrchestrationStatus } from "./Layers/ProviderCommandReactor.ts"; +import type { OrchestrationEngineShape } from "./Services/OrchestrationEngine.ts"; +import type { ProjectionSnapshotQueryShape } from "./Services/ProjectionSnapshotQuery.ts"; + +const IMPORTED_THREAD_TITLE_MAX_CHARS = 120; + +type ImportableDriver = "claudeAgent" | "codex"; + +function isImportableDriver(driver: string): driver is ImportableDriver { + return driver === "claudeAgent" || driver === "codex"; +} + +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +const readClaudeSessionInfo = (externalId: string) => + Effect.tryPromise({ + try: () => getSessionInfo(externalId, {}), + catch: (cause) => + new OrchestrationImportThreadError({ + message: `Could not read Claude Code session '${externalId}'.`, + cause, + }), + }); + +const claudeSessionNotFound = (externalId: string) => + new OrchestrationImportThreadError({ + message: `No Claude Code session '${externalId}' exists on this machine. Run /status inside the session to copy its id, or pick it from "claude --resume".`, + }); + +function transcriptTimestamp(epochMs: number | undefined): string | undefined { + return epochMs === undefined || !Number.isFinite(epochMs) + ? undefined + : DateTime.formatIso(DateTime.makeUnsafe(epochMs)); +} + +function importedThreadTitle(input: { + readonly externalId: string; + readonly summary: string | undefined; +}): string { + const summary = input.summary?.trim() ?? ""; + return summary.length > 0 + ? summary.slice(0, IMPORTED_THREAD_TITLE_MAX_CHARS) + : `Imported session ${input.externalId.slice(0, 8)}`; +} + +export const makeImportThread = (dependencies: { + readonly crypto: Crypto.Crypto; + readonly orchestrationEngine: Pick; + readonly projectionSnapshotQuery: Pick< + ProjectionSnapshotQueryShape, + "getProjectShellById" | "getActiveProjectByWorkspaceRoot" + >; + readonly providerService: Pick< + ProviderServiceShape, + "getInstanceInfo" | "startSession" | "readThread" + >; +}) => { + const { crypto, orchestrationEngine, projectionSnapshotQuery, providerService } = dependencies; + + const commandId = (tag: string) => + crypto.randomUUIDv4.pipe( + Effect.orDie, + Effect.map((uuid) => CommandId.make(`import-thread:${tag}:${uuid}`)), + ); + + const requireImportableDriver = Effect.fnUntraced(function* ( + instanceId: ProviderInstanceId, + ): Effect.fn.Return { + const instanceInfo = yield* providerService.getInstanceInfo(instanceId).pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: `Provider instance '${instanceId}' is not available.`, + cause, + }), + ), + ); + const driver = instanceInfo.driverKind; + if (isImportableDriver(driver)) { + return driver; + } + return yield* new OrchestrationImportThreadError({ + message: `Importing an existing session is only supported for Claude Code and Codex, not '${driver}'.`, + }); + }); + + const readImportedMessages = Effect.fnUntraced(function* (input: { + readonly driver: ImportableDriver; + readonly threadId: ThreadId; + readonly externalId: string; + readonly workspaceRoot: string; + readonly transcriptAt: string | undefined; + readonly importedAt: string; + }) { + if (input.driver === "claudeAgent") { + const sessionMessages = yield* Effect.tryPromise({ + try: () => getSessionMessages(input.externalId, {}), + catch: (cause) => + new OrchestrationImportThreadError({ + message: `Could not read the transcript for Claude Code session '${input.externalId}'.`, + cause, + }), + }); + return mapClaudeSessionMessages({ + threadId: input.threadId, + importedAt: input.transcriptAt ?? input.importedAt, + messages: sessionMessages, + }); + } + + const snapshot = yield* providerService.readThread(input.threadId).pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: `Could not read the transcript for Codex thread '${input.externalId}'.`, + cause, + }), + ), + ); + const snapshotWorkspaceRoot = snapshot.workspaceRoot?.trim() ?? ""; + if (snapshotWorkspaceRoot.length > 0 && snapshotWorkspaceRoot !== input.workspaceRoot) { + return yield* new OrchestrationImportThreadError({ + message: `Codex thread '${input.externalId}' ran in ${snapshotWorkspaceRoot}, not in ${input.workspaceRoot}. Import it from a thread in that project.`, + }); + } + return mapCodexThreadSnapshot({ + threadId: input.threadId, + importedAt: transcriptTimestamp(snapshot.lastActivityAtMs) ?? input.importedAt, + snapshot, + }); + }); + + const deletingCreatedThreadUnlessImported = ( + threadId: ThreadId, + effect: Effect.Effect, + ) => + effect.pipe( + Effect.onExit((exit) => + Exit.isSuccess(exit) + ? Effect.void + : commandId("delete").pipe( + Effect.flatMap((deleteCommandId) => + orchestrationEngine.dispatch({ + type: "thread.delete", + commandId: deleteCommandId, + threadId, + }), + ), + Effect.catchCause((cause) => + Effect.logWarning("failed to delete thread after an abandoned import", { + threadId, + cause, + }), + ), + ), + ), + ); + + const resolveImportSession = Effect.fnUntraced(function* ( + input: OrchestrationResolveImportSessionInput, + ): Effect.fn.Return { + const driver = yield* requireImportableDriver(input.instanceId); + if (driver !== "claudeAgent") { + return { externalId: input.externalId, workspaceRoot: null, projectId: null, title: null }; + } + + const sessionInfo = yield* readClaudeSessionInfo(input.externalId); + if (sessionInfo === undefined) { + return yield* claudeSessionNotFound(input.externalId); + } + const title = sessionInfo.summary.trim(); + const workspaceRoot = sessionInfo.cwd?.trim() ?? ""; + if (workspaceRoot.length === 0) { + return { + externalId: input.externalId, + workspaceRoot: null, + projectId: null, + title: title.length > 0 ? title : null, + }; + } + + const project = yield* projectionSnapshotQuery + .getActiveProjectByWorkspaceRoot(workspaceRoot) + .pipe( + Effect.map(Option.getOrUndefined), + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: `Could not look up a project for '${workspaceRoot}'.`, + cause, + }), + ), + ); + + return { + externalId: input.externalId, + workspaceRoot, + projectId: project?.id ?? null, + title: title.length > 0 ? title : null, + }; + }); + + const importThread = Effect.fnUntraced(function* ( + input: OrchestrationImportThreadInput, + ): Effect.fn.Return { + const project = yield* projectionSnapshotQuery.getProjectShellById(input.projectId).pipe( + Effect.map(Option.getOrUndefined), + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: `Could not read project '${input.projectId}'.`, + cause, + }), + ), + ); + if (project === undefined) { + return yield* new OrchestrationImportThreadError({ + message: `Project '${input.projectId}' no longer exists.`, + }); + } + + const driver = yield* requireImportableDriver(input.modelSelection.instanceId); + const sessionInfo = + driver === "claudeAgent" ? yield* readClaudeSessionInfo(input.externalId) : undefined; + if (driver === "claudeAgent") { + if (sessionInfo === undefined) { + return yield* claudeSessionNotFound(input.externalId); + } + const sessionWorkspaceRoot = sessionInfo.cwd?.trim() ?? ""; + if (sessionWorkspaceRoot.length > 0 && sessionWorkspaceRoot !== project.workspaceRoot) { + return yield* new OrchestrationImportThreadError({ + message: `Claude Code session '${input.externalId}' ran in ${sessionWorkspaceRoot}, not in ${project.workspaceRoot}. Import it from a thread in that project.`, + }); + } + } + + const createdAt = yield* nowIso; + const threadId = yield* crypto.randomUUIDv4.pipe(Effect.orDie, Effect.map(ThreadId.make)); + yield* orchestrationEngine + .dispatch({ + type: "thread.create", + commandId: yield* commandId("create"), + threadId, + projectId: input.projectId, + title: importedThreadTitle({ + externalId: input.externalId, + summary: sessionInfo?.summary, + }), + modelSelection: input.modelSelection, + runtimeMode: DEFAULT_RUNTIME_MODE, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + branch: null, + worktreePath: null, + createdAt, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: "Could not create a thread for the imported session.", + cause, + }), + ), + ); + + yield* deletingCreatedThreadUnlessImported( + threadId, + Effect.gen(function* () { + const session = yield* providerService + .startSession(threadId, { + threadId, + provider: ProviderDriverKind.make(driver), + providerInstanceId: input.modelSelection.instanceId, + cwd: project.workspaceRoot, + modelSelection: input.modelSelection, + resumeCursor: + driver === "claudeAgent" + ? { resume: input.externalId } + : { threadId: input.externalId }, + runtimeMode: DEFAULT_RUNTIME_MODE, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: `Could not resume session '${input.externalId}' for this thread.`, + cause, + }), + ), + ); + + const importedAt = yield* nowIso; + const messages = yield* readImportedMessages({ + driver, + threadId, + externalId: input.externalId, + workspaceRoot: project.workspaceRoot, + transcriptAt: transcriptTimestamp(sessionInfo?.lastModified), + importedAt, + }); + if (messages.length === 0) { + return yield* new OrchestrationImportThreadError({ + message: `Session '${input.externalId}' has no conversation to import.`, + }); + } + yield* orchestrationEngine + .dispatch({ + type: "thread.messages.import", + commandId: yield* commandId("messages"), + threadId, + messages, + createdAt: importedAt, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: "Could not store the imported transcript.", + cause, + }), + ), + ); + + yield* orchestrationEngine + .dispatch({ + type: "thread.session.set", + commandId: yield* commandId("session"), + threadId, + session: { + threadId, + status: mapProviderSessionStatusToOrchestrationStatus(session.status), + providerName: session.provider, + providerInstanceId: input.modelSelection.instanceId, + runtimeMode: DEFAULT_RUNTIME_MODE, + activeTurnId: null, + lastError: session.lastError ?? null, + updatedAt: session.updatedAt, + }, + createdAt: yield* nowIso, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: "Could not bind the resumed session to this thread.", + cause, + }), + ), + ); + }), + ); + + return { threadId }; + }); + + return { resolveImportSession, importThread }; +}; diff --git a/apps/server/src/orchestration/importedMessages.test.ts b/apps/server/src/orchestration/importedMessages.test.ts new file mode 100644 index 00000000000..04a4fe59f4d --- /dev/null +++ b/apps/server/src/orchestration/importedMessages.test.ts @@ -0,0 +1,162 @@ +import type { SessionMessage as ClaudeSessionMessage } from "@anthropic-ai/claude-agent-sdk"; +import { ThreadId, TurnId } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { mapClaudeSessionMessages, mapCodexThreadSnapshot } from "./importedMessages.ts"; + +const threadId = ThreadId.make("thread-import"); +const importedAt = "2026-01-01T00:00:00.000Z"; + +const claudeMessage = ( + input: Pick & { readonly message: unknown }, +): ClaudeSessionMessage => ({ + type: input.type, + uuid: input.uuid, + session_id: "session-1", + message: input.message, + parent_tool_use_id: null, +}); + +describe("mapClaudeSessionMessages", () => { + it("keeps user and assistant text and skips everything else", () => { + const messages = mapClaudeSessionMessages({ + threadId, + importedAt, + messages: [ + claudeMessage({ + type: "user", + uuid: "uuid-1", + message: { role: "user", content: "Fix the flaky test" }, + }), + claudeMessage({ + type: "system", + uuid: "uuid-2", + message: { role: "system", content: "compact boundary" }, + }), + claudeMessage({ + type: "assistant", + uuid: "uuid-3", + message: { + role: "assistant", + content: [ + { type: "text", text: "Looking at it" }, + { type: "tool_use", id: "tool-1", name: "Bash", input: {} }, + { type: "text", text: "Done" }, + ], + }, + }), + claudeMessage({ + type: "user", + uuid: "uuid-4", + message: { + role: "user", + content: [{ type: "tool_result", tool_use_id: "tool-1", content: "ok" }], + }, + }), + ], + }); + + expect(messages).toEqual([ + { + messageId: "imported:claudeAgent:thread-import:000000:uuid-1", + role: "user", + text: "Fix the flaky test", + createdAt: importedAt, + updatedAt: importedAt, + }, + { + messageId: "imported:claudeAgent:thread-import:000002:uuid-3", + role: "assistant", + text: "Looking at it\n\nDone", + createdAt: importedAt, + updatedAt: importedAt, + }, + ]); + }); + + it("orders message ids so a transcript past ten messages still sorts chronologically", () => { + const messages = mapClaudeSessionMessages({ + threadId, + importedAt, + messages: Array.from({ length: 12 }, (_, index) => + claudeMessage({ + type: "user", + uuid: `uuid-${index}`, + message: { role: "user", content: `message ${index}` }, + }), + ), + }); + + const sortedByMessageId = [...messages].sort((left, right) => + left.messageId < right.messageId ? -1 : left.messageId > right.messageId ? 1 : 0, + ); + expect(sortedByMessageId.map((message) => message.text)).toEqual( + messages.map((message) => message.text), + ); + }); + + it("derives the same message ids when the same transcript is imported again", () => { + const messages = [ + claudeMessage({ + type: "user", + uuid: "uuid-1", + message: { role: "user", content: "First" }, + }), + claudeMessage({ + type: "assistant", + uuid: "uuid-2", + message: { role: "assistant", content: "Second" }, + }), + ]; + + expect(mapClaudeSessionMessages({ threadId, importedAt, messages })).toEqual( + mapClaudeSessionMessages({ threadId, importedAt, messages }), + ); + }); +}); + +describe("mapCodexThreadSnapshot", () => { + it("keeps user and agent messages and skips empty text", () => { + const messages = mapCodexThreadSnapshot({ + threadId, + importedAt, + snapshot: { + threadId, + turns: [ + { + id: TurnId.make("turn-1"), + items: [ + { type: "userMessage", id: "item-1", content: [{ type: "text", text: "Ship it" }] }, + { type: "reasoning", id: "item-2", summary: ["thinking"] }, + { type: "agentMessage", id: "item-3", text: "Shipped" }, + ], + }, + { + id: TurnId.make("turn-2"), + items: [ + { type: "userMessage", id: "item-4", content: [{ type: "image", url: "x" }] }, + { type: "agentMessage", id: "item-5", text: " " }, + ], + }, + ], + }, + }); + + expect(messages).toEqual([ + { + messageId: "imported:codex:thread-import:000000:item-1", + role: "user", + text: "Ship it", + createdAt: importedAt, + updatedAt: importedAt, + }, + { + messageId: "imported:codex:thread-import:000002:item-3", + role: "assistant", + text: "Shipped", + createdAt: importedAt, + updatedAt: importedAt, + }, + ]); + }); +}); diff --git a/apps/server/src/orchestration/importedMessages.ts b/apps/server/src/orchestration/importedMessages.ts new file mode 100644 index 00000000000..4bc12b8399c --- /dev/null +++ b/apps/server/src/orchestration/importedMessages.ts @@ -0,0 +1,121 @@ +import type { SessionMessage as ClaudeSessionMessage } from "@anthropic-ai/claude-agent-sdk"; +import { MessageId, type ThreadId, type ThreadImportedMessage } from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; + +import type { ProviderThreadSnapshot } from "../provider/Services/ProviderAdapter.ts"; + +const TextPart = Schema.Struct({ + type: Schema.Literal("text"), + text: Schema.String, +}); +const isTextPart = Schema.is(TextPart); + +const ClaudeMessageBody = Schema.Struct({ + content: Schema.Union([Schema.String, Schema.Array(Schema.Unknown)]), +}); +const isClaudeMessageBody = Schema.is(ClaudeMessageBody); + +const CodexUserMessageItem = Schema.Struct({ + type: Schema.Literal("userMessage"), + id: Schema.String, + content: Schema.Array(Schema.Unknown), +}); +const isCodexUserMessageItem = Schema.is(CodexUserMessageItem); + +const CodexAgentMessageItem = Schema.Struct({ + type: Schema.Literal("agentMessage"), + id: Schema.String, + text: Schema.String, +}); +const isCodexAgentMessageItem = Schema.is(CodexAgentMessageItem); + +const IMPORTED_MESSAGE_INDEX_DIGITS = 6; + +interface ImportedTranscriptEntry { + readonly role: ThreadImportedMessage["role"]; + readonly text: string; + readonly sourceId: string; +} + +function joinTextParts(parts: ReadonlyArray): string { + return parts + .filter(isTextPart) + .map((part) => part.text) + .join("\n\n") + .trim(); +} + +function toImportedMessages(input: { + readonly threadId: ThreadId; + readonly provider: string; + readonly importedAt: string; + readonly entries: ReadonlyArray; +}): ReadonlyArray { + return input.entries.flatMap((entry, index) => { + if (entry === null || entry.text.length === 0) { + return []; + } + return [ + { + messageId: MessageId.make( + `imported:${input.provider}:${input.threadId}:${String(index).padStart(IMPORTED_MESSAGE_INDEX_DIGITS, "0")}:${entry.sourceId}`, + ), + role: entry.role, + text: entry.text, + createdAt: input.importedAt, + updatedAt: input.importedAt, + }, + ]; + }); +} + +function readClaudeEntry(message: ClaudeSessionMessage): ImportedTranscriptEntry | null { + if (message.type !== "user" && message.type !== "assistant") { + return null; + } + const body = message.message; + if (!isClaudeMessageBody(body)) { + return null; + } + return { + role: message.type, + text: typeof body.content === "string" ? body.content.trim() : joinTextParts(body.content), + sourceId: message.uuid, + }; +} + +function readCodexEntry(item: unknown): ImportedTranscriptEntry | null { + if (isCodexUserMessageItem(item)) { + return { role: "user", text: joinTextParts(item.content), sourceId: item.id }; + } + if (isCodexAgentMessageItem(item)) { + return { role: "assistant", text: item.text.trim(), sourceId: item.id }; + } + return null; +} + +export function mapClaudeSessionMessages(input: { + readonly threadId: ThreadId; + readonly importedAt: string; + readonly messages: ReadonlyArray; +}): ReadonlyArray { + return toImportedMessages({ + threadId: input.threadId, + provider: "claudeAgent", + importedAt: input.importedAt, + entries: input.messages.map(readClaudeEntry), + }); +} + +export function mapCodexThreadSnapshot(input: { + readonly threadId: ThreadId; + readonly importedAt: string; + readonly snapshot: ProviderThreadSnapshot; +}): ReadonlyArray { + return toImportedMessages({ + threadId: input.threadId, + provider: "codex", + importedAt: input.importedAt, + entries: input.snapshot.turns.flatMap((turn) => turn.items).map(readCodexEntry), + }); +} diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 9c07a312023..aba2542bd70 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -209,6 +209,77 @@ describe("orchestration projector", () => { expect(unarchived.threads[0]?.archivedAt).toBeNull(); }); + it("applies thread.messages-imported into the command read model", async () => { + const now = "2026-01-01T00:00:00.000Z"; + const importedAt = "2025-12-31T09:00:00.000Z"; + const withThread = await Effect.runPromise( + projectEvent( + createEmptyReadModel(now), + makeEvent({ + sequence: 1, + type: "thread.created", + aggregateKind: "thread", + aggregateId: "thread-import", + occurredAt: now, + commandId: "cmd-thread-create", + payload: { + threadId: "thread-import", + projectId: "project-1", + title: "Imported", + modelSelection: { + provider: ProviderDriverKind.make("claudeAgent"), + model: "sonnet", + }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }), + ), + ); + + const importEvent = makeEvent({ + sequence: 2, + type: "thread.messages-imported", + aggregateKind: "thread", + aggregateId: "thread-import", + occurredAt: now, + commandId: "cmd-messages-import", + payload: { + threadId: "thread-import", + messages: [ + { + messageId: "imported:claudeAgent:thread-import:000000:uuid-1", + role: "user", + text: "Fix the flaky test", + createdAt: importedAt, + updatedAt: importedAt, + }, + { + messageId: "imported:claudeAgent:thread-import:000001:uuid-2", + role: "assistant", + text: "Looking at it", + createdAt: importedAt, + updatedAt: importedAt, + }, + ], + }, + }); + + const next = await Effect.runPromise(projectEvent(withThread, importEvent)); + expect(next.threads[0]?.messages.map((message) => [message.role, message.text])).toEqual([ + ["user", "Fix the flaky test"], + ["assistant", "Looking at it"], + ]); + expect(next.threads[0]?.messages.every((message) => message.turnId === null)).toBe(true); + expect(next.threads[0]?.messages[0]?.createdAt).toBe(importedAt); + + const replayed = await Effect.runPromise(projectEvent(next, importEvent)); + expect(replayed.threads[0]?.messages).toHaveLength(2); + }); + it("keeps projector forward-compatible for unhandled event types", async () => { const now = "2026-01-01T00:00:00.000Z"; const model = createEmptyReadModel(now); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 0504cb36f9a..f299c68e117 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -11,6 +11,7 @@ import * as Schema from "effect/Schema"; import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts"; import { MessageSentPayloadSchema, + MessagesImportedPayloadSchema, ProjectCreatedPayload, ProjectDeletedPayload, ProjectMetaUpdatedPayload, @@ -497,6 +498,44 @@ export function projectEvent( }; }); + case "thread.messages-imported": + return Effect.gen(function* () { + const payload = yield* decodeForEvent( + MessagesImportedPayloadSchema, + event.payload, + event.type, + "payload", + ); + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + + const knownMessageIds = new Set(thread.messages.map((entry) => entry.id)); + const importedMessages: OrchestrationMessage[] = payload.messages + .filter((message) => !knownMessageIds.has(message.messageId)) + .map((message) => ({ + id: message.messageId, + role: message.role, + text: message.text, + turnId: null, + streaming: false, + createdAt: message.createdAt, + updatedAt: message.updatedAt, + })); + if (importedMessages.length === 0) { + return nextBase; + } + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: [...thread.messages, ...importedMessages].slice(-MAX_THREAD_MESSAGES), + updatedAt: event.occurredAt, + }), + }; + }); + case "thread.session-set": return Effect.gen(function* () { const payload = yield* decodeForEvent( diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 4ae654a5187..e2585932517 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -91,6 +91,8 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { (): Promise => Promise.resolve({ threadId: "provider-thread-1", + workspaceRoot: "/tmp/codex-thread-1", + lastActivityAtMs: 0, turns: [], }), ); @@ -99,6 +101,8 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape { (_numTurns: number): Promise => Promise.resolve({ threadId: "provider-thread-1", + workspaceRoot: "/tmp/codex-thread-1", + lastActivityAtMs: 0, turns: [], }), ); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 38a5887cdc3..efa9570b106 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -1593,6 +1593,8 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ), Effect.map((snapshot) => ({ threadId, + workspaceRoot: snapshot.workspaceRoot, + lastActivityAtMs: snapshot.lastActivityAtMs, turns: snapshot.turns, })), ); @@ -1617,6 +1619,8 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ), Effect.map((snapshot) => ({ threadId, + workspaceRoot: snapshot.workspaceRoot, + lastActivityAtMs: snapshot.lastActivityAtMs, turns: snapshot.turns, })), ); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index d7346a0e0db..0b131119de9 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -19,6 +19,7 @@ import { hasConfiguredMcpServer, isRecoverableThreadResumeError, openCodexThread, + parseThreadSnapshot, } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); @@ -469,3 +470,28 @@ describe("openCodexThread", () => { }), ); }); + +describe("parseThreadSnapshot", () => { + it("converts the codex thread's unix seconds into epoch milliseconds", () => { + const snapshot = parseThreadSnapshot({ + thread: { + id: "thread-1", + cwd: "/repo/app", + createdAt: 1767225600, + recencyAt: 1767312000, + turns: [], + }, + }); + + NodeAssert.equal(snapshot.workspaceRoot, "/repo/app"); + NodeAssert.equal(snapshot.lastActivityAtMs, 1767312000000); + }); + + it("falls back to the creation time when the thread has no recency stamp", () => { + const snapshot = parseThreadSnapshot({ + thread: { id: "thread-2", cwd: "/repo/app", createdAt: 1767225600, turns: [] }, + }); + + NodeAssert.equal(snapshot.lastActivityAtMs, 1767225600000); + }); +}); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 67108dd4dbb..c8c3f2fc072 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -128,6 +128,8 @@ export interface CodexThreadTurnSnapshot { export interface CodexThreadSnapshot { readonly threadId: string; readonly turns: ReadonlyArray; + readonly workspaceRoot: string; + readonly lastActivityAtMs: number; } export interface CodexSessionRuntimeShape { @@ -695,11 +697,25 @@ function updateSession( }); } -function parseThreadSnapshot( - response: EffectCodexSchema.V2ThreadReadResponse | EffectCodexSchema.V2ThreadRollbackResponse, -): CodexThreadSnapshot { +const MILLISECONDS_PER_SECOND = 1_000; + +export function parseThreadSnapshot(response: { + readonly thread: { + readonly id: string; + readonly cwd: string; + readonly createdAt: number; + readonly recencyAt?: number | null; + readonly turns: ReadonlyArray<{ + readonly id: string; + readonly items: ReadonlyArray; + }>; + }; +}): CodexThreadSnapshot { return { threadId: response.thread.id, + workspaceRoot: response.thread.cwd, + lastActivityAtMs: + (response.thread.recencyAt ?? response.thread.createdAt) * MILLISECONDS_PER_SECOND, turns: response.thread.turns.map((turn) => ({ id: TurnId.make(turn.id), items: turn.items, diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 2eaaeb8ce3c..73ccb00f367 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -967,6 +967,22 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const getInstanceInfo: ProviderServiceMethod<"getInstanceInfo"> = (instanceId) => registry.getInstanceInfo(instanceId); + const readThread: ProviderServiceMethod<"readThread"> = Effect.fn("readThread")( + function* (threadId) { + const routed = yield* resolveRoutableSession({ + threadId, + operation: "ProviderService.readThread", + allowRecovery: true, + }); + yield* Effect.annotateCurrentSpan({ + "provider.operation": "read-thread", + "provider.kind": routed.adapter.provider, + "provider.thread_id": threadId, + }); + return yield* routed.adapter.readThread(routed.threadId); + }, + ); + const rollbackConversation: ProviderServiceMethod<"rollbackConversation"> = Effect.fn( "rollbackConversation", )(function* (rawInput) { @@ -1078,6 +1094,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( listSessions, getCapabilities, getInstanceInfo, + readThread, rollbackConversation, // Each access creates a fresh PubSub subscription so that multiple // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 3843c8acbcd..88106af3475 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -173,6 +173,7 @@ describe("ProviderSessionReaper", () => { }, }); }, + readThread: () => unsupported(), rollbackConversation: () => unsupported(), streamEvents: Stream.empty, }; diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 01eeae7b7bd..3e2e6b7da3f 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -40,6 +40,8 @@ export interface ProviderThreadTurnSnapshot { export interface ProviderThreadSnapshot { readonly threadId: ThreadId; readonly turns: ReadonlyArray; + readonly workspaceRoot?: string; + readonly lastActivityAtMs?: number; } export interface ProviderAdapterShape { diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 4d4cb4fa01a..eacc8ea8674 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -29,7 +29,7 @@ import type * as Effect from "effect/Effect"; import type * as Stream from "effect/Stream"; import type { ProviderServiceError } from "../Errors.ts"; -import type { ProviderAdapterCapabilities } from "./ProviderAdapter.ts"; +import type { ProviderAdapterCapabilities, ProviderThreadSnapshot } from "./ProviderAdapter.ts"; import type { ProviderInstanceRoutingInfo } from "./ProviderAdapterRegistry.ts"; /** @@ -97,6 +97,10 @@ export interface ProviderServiceShape { instanceId: ProviderInstanceId, ) => Effect.Effect; + readonly readThread: ( + threadId: ThreadId, + ) => Effect.Effect; + /** * Roll back provider conversation state by a number of turns. */ diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 74a4de594a1..861804859da 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -170,6 +170,16 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { }, } as unknown as OrchestrationEvent), ).toBe(false); + expect( + AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({ + ...base, + type: "thread.messages-imported", + payload: { + threadId: "thread-1" as ThreadId, + messages: [], + }, + } as unknown as OrchestrationEvent), + ).toBe(false); expect( AgentAwarenessRelay.shouldPublishAgentAwarenessEvent({ ...base, diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 58de98f1ca1..a0de3dc46ef 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -74,6 +74,7 @@ export function shouldPublishAgentAwarenessEvent(event: OrchestrationEvent): boo // before the real running state arrives. Provider lifecycle events publish // the authoritative starting/running state instead. return false; + case "thread.messages-imported": case "thread.proposed-plan-upserted": case "thread.runtime-mode-set": case "thread.interaction-mode-set": diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 871b79eca90..dc96e354d9a 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -83,6 +83,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 ProviderService from "./provider/Services/ProviderService.ts"; import { makeManualOnlyProviderMaintenanceCapabilities } from "./provider/providerMaintenance.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; @@ -340,6 +341,7 @@ const buildAppUnderTest = (options?: { terminalManager?: Partial; orchestrationEngine?: Partial; projectionSnapshotQuery?: Partial; + providerService?: Partial; checkpointDiffQuery?: Partial; browserTraceCollector?: Partial; serverLifecycleEvents?: Partial; @@ -544,18 +546,23 @@ 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(ProviderService.ProviderService)({ + ...options?.layers?.providerService, + }), + ), ), Layer.provide( Layer.mock(ServerSettings.ServerSettingsService)({ diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index f6f46d1e76e..ce44475b073 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -68,6 +68,7 @@ import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as ServerConfig from "./config.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; +import { makeImportThread } from "./orchestration/importThread.ts"; import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; @@ -77,6 +78,7 @@ import { observeRpcStreamEffect as instrumentRpcStreamEffect, } from "./observability/RpcInstrumentation.ts"; import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts"; +import * as ProviderService from "./provider/Services/ProviderService.ts"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import * as ServerSelfUpdate from "./cloud/selfUpdate.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; @@ -259,6 +261,7 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< { type: | "thread.message-sent" + | "thread.messages-imported" | "thread.proposed-plan-upserted" | "thread.activity-appended" | "thread.turn-diff-completed" @@ -268,6 +271,7 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< > { return ( event.type === "thread.message-sent" || + event.type === "thread.messages-imported" || event.type === "thread.proposed-plan-upserted" || event.type === "thread.activity-appended" || event.type === "thread.turn-diff-completed" || @@ -287,6 +291,8 @@ const SHELL_RESUME_MAX_GAP = 1_000; const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], + [ORCHESTRATION_WS_METHODS.resolveImportSession, AuthOrchestrationReadScope], + [ORCHESTRATION_WS_METHODS.importThread, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.getFullThreadDiff, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.replayEvents, AuthOrchestrationReadScope], @@ -409,6 +415,13 @@ const makeWsRpcLayer = ( const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; const checkpointDiffQuery = yield* CheckpointDiffQuery.CheckpointDiffQuery; + const providerService = yield* ProviderService.ProviderService; + const importSession = makeImportThread({ + crypto, + orchestrationEngine, + projectionSnapshotQuery, + providerService, + }); const keybindings = yield* Keybindings.Keybindings; const externalLauncher = yield* ExternalLauncher.ExternalLauncher; const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; @@ -1172,6 +1185,18 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "orchestration" }, ), + [ORCHESTRATION_WS_METHODS.resolveImportSession]: (input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.resolveImportSession, + importSession.resolveImportSession(input), + { "rpc.aggregate": "orchestration" }, + ), + [ORCHESTRATION_WS_METHODS.importThread]: (input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.importThread, + importSession.importThread(input), + { "rpc.aggregate": "orchestration" }, + ), [ORCHESTRATION_WS_METHODS.getTurnDiff]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.getTurnDiff, diff --git a/apps/web/src/components/CommandPalette.logic.test.ts b/apps/web/src/components/CommandPalette.logic.test.ts index 902b7e87773..70ce518a31d 100644 --- a/apps/web/src/components/CommandPalette.logic.test.ts +++ b/apps/web/src/components/CommandPalette.logic.test.ts @@ -1,8 +1,17 @@ import { describe, expect, it, vi } from "vite-plus/test"; -import { EnvironmentId, ProjectId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import { + EnvironmentId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + type ServerProvider, + ThreadId, +} from "@t3tools/contracts"; import type { Thread } from "../types"; import { + buildImportSessionProviderOptions, buildThreadActionItems, + describeImportSessionStep, enumerateCommandPaletteItems, filterCommandPaletteGroups, type CommandPaletteGroup, @@ -193,3 +202,120 @@ describe("buildThreadActionItems", () => { expect(items.map((item) => item.value)).toEqual(["thread:thread-active"]); }); }); + +function provider(input: { + driver: string; + instanceId: string; + status?: ServerProvider["status"]; + models?: ServerProvider["models"]; +}): ServerProvider { + return { + instanceId: ProviderInstanceId.make(input.instanceId), + driver: ProviderDriverKind.make(input.driver), + enabled: true, + installed: true, + version: null, + status: input.status ?? "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-01-01T00:00:00.000Z", + models: input.models ?? [], + slashCommands: [], + skills: [], + }; +} + +const providerModel = (slug: string) => ({ slug, name: slug, isCustom: false, capabilities: {} }); + +describe("buildImportSessionProviderOptions", () => { + it("offers Claude and Codex with their own field labels and default models", () => { + const options = buildImportSessionProviderOptions([ + provider({ + driver: "claudeAgent", + instanceId: "claudeAgent", + models: [providerModel("sonnet")], + }), + provider({ driver: "codex", instanceId: "codex", models: [providerModel("gpt-5-codex")] }), + ]); + + expect(options).toEqual([ + { + driverKind: "claudeAgent", + label: "Claude Code", + fieldLabel: "Session ID", + placeholder: "0de413a1-1796-43c7-9abf-967c5aedd890", + hint: "Run /status inside a Claude Code session to copy its id, or pick one from claude --resume.", + modelSelection: { instanceId: ProviderInstanceId.make("claudeAgent"), model: "sonnet" }, + }, + { + driverKind: "codex", + label: "Codex", + fieldLabel: "Thread ID", + placeholder: "019a2f8c-4e1b-7c3d-9f10-2b6a5d8e4c77", + hint: "Run codex resume to list thread ids.", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex" }, + }, + ]); + }); + + it("drops providers that are not ready and providers this feature does not support", () => { + const options = buildImportSessionProviderOptions([ + provider({ + driver: "claudeAgent", + instanceId: "claudeAgent", + status: "error", + models: [providerModel("sonnet")], + }), + provider({ driver: "cursor", instanceId: "cursor", models: [providerModel("auto")] }), + provider({ driver: "codex", instanceId: "codex", models: [providerModel("gpt-5-codex")] }), + ]); + + expect(options.map((option) => option.driverKind)).toEqual(["codex"]); + }); +}); + +describe("describeImportSessionStep", () => { + it("imports straight away before the session is resolved", () => { + expect(describeImportSessionStep(null)).toEqual({ + notice: null, + confirmLabel: "Import", + missingProjectWorkspaceRoot: null, + }); + }); + + it("imports straight away once the session maps to an existing project", () => { + expect( + describeImportSessionStep({ + externalId: "session-1", + workspaceRoot: "/home/dev/app", + projectId: PROJECT_ID, + title: "Fix the flaky test", + }), + ).toEqual({ notice: null, confirmLabel: "Import", missingProjectWorkspaceRoot: null }); + }); + + it("asks to add the workspace when no project matches it", () => { + expect( + describeImportSessionStep({ + externalId: "session-1", + workspaceRoot: "/home/dev/app", + projectId: null, + title: null, + }), + ).toEqual({ + notice: "This session ran in /home/dev/app, which is not a project yet.", + confirmLabel: "Add project & import", + missingProjectWorkspaceRoot: "/home/dev/app", + }); + }); + + it("imports into the current project when the provider reports no workspace", () => { + expect( + describeImportSessionStep({ + externalId: "thread-1", + workspaceRoot: null, + projectId: null, + title: null, + }), + ).toEqual({ notice: null, confirmLabel: "Import", missingProjectWorkspaceRoot: null }); + }); +}); diff --git a/apps/web/src/components/CommandPalette.logic.ts b/apps/web/src/components/CommandPalette.logic.ts index f69c38e1a0f..7294bd0bc60 100644 --- a/apps/web/src/components/CommandPalette.logic.ts +++ b/apps/web/src/components/CommandPalette.logic.ts @@ -1,6 +1,9 @@ import { type KeybindingCommand, type FilesystemBrowseEntry, + type ModelSelection, + type OrchestrationResolveImportSessionResult, + type ServerProvider, THREAD_JUMP_KEYBINDING_COMMANDS, } from "@t3tools/contracts"; import type { SidebarThreadSortOrder } from "@t3tools/contracts/settings"; @@ -8,6 +11,11 @@ import * as Arr from "effect/Array"; import * as Result from "effect/Result"; import { type ReactNode } from "react"; import { sortThreads } from "../lib/threadSort"; +import { + deriveProviderInstanceEntries, + getDefaultProviderInstanceModel, + isProviderInstancePickerReady, +} from "../providerInstances"; import { formatRelativeTimeLabel } from "../timestampFormat"; import { type Project, type SidebarThreadSummary, type Thread } from "../types"; @@ -366,6 +374,71 @@ export function buildRootGroups(input: { return groups; } +export interface ImportSessionProviderOption { + readonly driverKind: "claudeAgent" | "codex"; + readonly label: string; + readonly fieldLabel: string; + readonly placeholder: string; + readonly hint: string; + readonly modelSelection: ModelSelection; +} + +const IMPORT_SESSION_PROVIDERS = [ + { + driverKind: "claudeAgent", + label: "Claude Code", + fieldLabel: "Session ID", + placeholder: "0de413a1-1796-43c7-9abf-967c5aedd890", + hint: "Run /status inside a Claude Code session to copy its id, or pick one from claude --resume.", + }, + { + driverKind: "codex", + label: "Codex", + fieldLabel: "Thread ID", + placeholder: "019a2f8c-4e1b-7c3d-9f10-2b6a5d8e4c77", + hint: "Run codex resume to list thread ids.", + }, +] as const satisfies ReadonlyArray>; + +export function buildImportSessionProviderOptions( + providers: ReadonlyArray, +): ImportSessionProviderOption[] { + const entries = deriveProviderInstanceEntries(providers); + return IMPORT_SESSION_PROVIDERS.flatMap((candidate) => { + const readyEntries = entries.filter( + (item) => item.driverKind === candidate.driverKind && isProviderInstancePickerReady(item), + ); + const entry = readyEntries.find((item) => item.isDefault) ?? readyEntries[0]; + if (!entry) { + return []; + } + const model = getDefaultProviderInstanceModel(providers, entry.instanceId); + if (!model) { + return []; + } + return [{ ...candidate, modelSelection: { instanceId: entry.instanceId, model } }]; + }); +} + +export interface ImportSessionStep { + readonly notice: string | null; + readonly confirmLabel: string; + readonly missingProjectWorkspaceRoot: string | null; +} + +export function describeImportSessionStep( + resolved: OrchestrationResolveImportSessionResult | null, +): ImportSessionStep { + if (resolved === null || resolved.projectId !== null || resolved.workspaceRoot === null) { + return { notice: null, confirmLabel: "Import", missingProjectWorkspaceRoot: null }; + } + return { + notice: `This session ran in ${resolved.workspaceRoot}, which is not a project yet.`, + confirmLabel: "Add project & import", + missingProjectWorkspaceRoot: resolved.workspaceRoot, + }; +} + export function getCommandPaletteInputPlaceholder(mode: CommandPaletteMode): string { switch (mode) { case "root": diff --git a/apps/web/src/components/CommandPalette.tsx b/apps/web/src/components/CommandPalette.tsx index aa7547c8ba6..02c9ab13604 100644 --- a/apps/web/src/components/CommandPalette.tsx +++ b/apps/web/src/components/CommandPalette.tsx @@ -11,6 +11,7 @@ import { type EnvironmentId, type FilesystemBrowseResult, type ProjectId, + type ScopedProjectRef, type SourceControlDiscoveryResult, type SourceControlProviderKind, type SourceControlRepositoryInfo, @@ -25,6 +26,7 @@ import { CornerLeftUpIcon, FolderIcon, FolderPlusIcon, + ImportIcon, LinkIcon, MessageSquareIcon, SettingsIcon, @@ -89,6 +91,7 @@ import { import { ADDON_ICON_CLASS, buildBrowseGroups, + buildImportSessionProviderOptions, buildProjectActionItems, buildRootGroups, buildThreadActionItems, @@ -96,6 +99,7 @@ import { type CommandPaletteActionItem, type CommandPaletteSubmenuItem, type CommandPaletteView, + type ImportSessionProviderOption, filterBrowseEntries, filterCommandPaletteGroups, getCommandPaletteInputPlaceholder, @@ -106,6 +110,7 @@ import { import { orderItemsByPreferredIds, sortLogicalProjectsForSidebar } from "./Sidebar.logic"; import { resolveEnvironmentOptionLabel } from "./BranchToolbar.logic"; import { CommandPaletteResults } from "./CommandPaletteResults"; +import { ImportSessionDialog } from "./ImportSessionDialog"; import { AzureDevOpsIcon, BitbucketIcon, GitHubIcon, GitLabIcon } from "./Icons"; import { ProjectFavicon } from "./ProjectFavicon"; import { ThreadRowLeadingStatus, ThreadRowTrailingStatus } from "./ThreadStatusIndicators"; @@ -376,6 +381,11 @@ function reduceCommandPaletteUiState( } } +interface ImportSessionRequest { + readonly projectRef: ScopedProjectRef; + readonly providerOptions: ReadonlyArray; +} + export function CommandPalette({ children }: { children: ReactNode }) { const [state, dispatch] = useReducer(reduceCommandPaletteUiState, { open: false, @@ -386,6 +396,14 @@ export function CommandPalette({ children }: { children: ReactNode }) { const openAddProject = useCallback(() => dispatch({ _tag: "OpenAddProject" }), []); const openNewThreadIn = useCallback(() => dispatch({ _tag: "OpenNewThreadIn" }), []); const clearOpenIntent = useCallback(() => dispatch({ _tag: "ClearOpenIntent" }), []); + const [importSession, setImportSession] = useState(null); + const openImportSession = useCallback( + (request: ImportSessionRequest) => { + setOpen(false); + setImportSession(request); + }, + [setOpen], + ); const keybindings = useAtomValue(primaryServerKeybindingsAtom); const composerHandleRef = useRef(null); const routeTarget = useParams({ @@ -408,7 +426,7 @@ export function CommandPalette({ children }: { children: ReactNode }) { terminalOpen, }, }); - if (command !== "commandPalette.toggle") { + if (command !== "commandPalette.toggle" || importSession !== null) { return; } event.preventDefault(); @@ -417,7 +435,7 @@ export function CommandPalette({ children }: { children: ReactNode }) { }; window.addEventListener("keydown", onKeyDown); return () => window.removeEventListener("keydown", onKeyDown); - }, [keybindings, terminalOpen, toggleOpen]); + }, [importSession, keybindings, terminalOpen, toggleOpen]); useEffect( () => @@ -442,8 +460,21 @@ export function CommandPalette({ children }: { children: ReactNode }) { openIntent={state.openIntent} setOpen={setOpen} clearOpenIntent={clearOpenIntent} + openImportSession={openImportSession} /> + {importSession ? ( + { + if (!nextOpen) { + setImportSession(null); + } + }} + /> + ) : null} ); } @@ -453,6 +484,7 @@ function CommandPaletteDialog(props: { readonly openIntent: CommandPaletteOpenIntent | null; readonly setOpen: (open: boolean) => void; readonly clearOpenIntent: () => void; + readonly openImportSession: (request: ImportSessionRequest) => void; }) { if (!props.open) { return null; @@ -463,6 +495,7 @@ function CommandPaletteDialog(props: { openIntent={props.openIntent} setOpen={props.setOpen} clearOpenIntent={props.clearOpenIntent} + openImportSession={props.openImportSession} /> ); } @@ -471,9 +504,10 @@ function OpenCommandPaletteDialog(props: { readonly openIntent: CommandPaletteOpenIntent | null; readonly setOpen: (open: boolean) => void; readonly clearOpenIntent: () => void; + readonly openImportSession: (request: ImportSessionRequest) => void; }) { const navigate = useNavigate(); - const { clearOpenIntent, openIntent, setOpen } = props; + const { clearOpenIntent, openImportSession, openIntent, setOpen } = props; const composerHandleRef = useComposerHandleContext(); const [query, setQuery] = useState(""); const deferredQuery = useDeferredValue(query); @@ -1145,6 +1179,15 @@ function OpenCommandPaletteDialog(props: { projectThreadItems, ]); + const importSessionProviderOptions = useMemo( + () => buildImportSessionProviderOptions(providers), + [providers], + ); + const importSessionProjectRef = + currentProjectEnvironmentId && currentProjectId + ? scopeProjectRef(currentProjectEnvironmentId, currentProjectId) + : defaultProjectRef; + const actionItems: Array = []; if (projects.length > 0) { @@ -1186,6 +1229,23 @@ function OpenCommandPaletteDialog(props: { }); } + if (importSessionProviderOptions.length > 0 && importSessionProjectRef) { + actionItems.push({ + kind: "action", + value: "action:import-session", + searchTerms: ["import session", "resume", "continue", "claude", "codex", "session id"], + title: "Import session...", + icon: , + keepOpen: true, + run: async () => { + openImportSession({ + projectRef: importSessionProjectRef, + providerOptions: importSessionProviderOptions, + }); + }, + }); + } + actionItems.push({ kind: "action", value: "action:add-project", diff --git a/apps/web/src/components/ImportSessionDialog.tsx b/apps/web/src/components/ImportSessionDialog.tsx new file mode 100644 index 00000000000..51463808f6f --- /dev/null +++ b/apps/web/src/components/ImportSessionDialog.tsx @@ -0,0 +1,280 @@ +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { inferProjectTitleFromPath } from "@t3tools/client-runtime/state/projects"; +import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; +import type { OrchestrationResolveImportSessionResult, ScopedProjectRef } from "@t3tools/contracts"; +import { useNavigate } from "@tanstack/react-router"; +import * as Cause from "effect/Cause"; +import { useCallback, useEffect, useRef, useState } from "react"; + +import { newProjectId } from "~/lib/utils"; +import { projectEnvironment } from "~/state/projects"; +import { threadEnvironment } from "~/state/threads"; +import { useAtomCommand } from "~/state/use-atom-command"; +import { buildThreadRouteParams } from "~/threadRoutes"; +import { + describeImportSessionStep, + type ImportSessionProviderOption, +} from "./CommandPalette.logic"; +import { Button } from "./ui/button"; +import { + Dialog, + DialogDescription, + DialogFooter, + DialogHeader, + DialogPanel, + DialogPopup, + DialogTitle, +} from "./ui/dialog"; +import { Input } from "./ui/input"; + +interface ImportSessionDialogProps { + open: boolean; + projectRef: ScopedProjectRef; + providerOptions: ReadonlyArray; + onOpenChange: (open: boolean) => void; +} + +function failureMessage( + result: { readonly cause: Cause.Cause }, + fallback: string, +): string { + const error = squashAtomCommandFailure(result); + return error instanceof Error && error.message.trim().length > 0 ? error.message : fallback; +} + +export function ImportSessionDialog({ + open, + projectRef, + providerOptions, + onOpenChange, +}: ImportSessionDialogProps) { + const navigate = useNavigate(); + const externalIdInputRef = useRef(null); + const [driverKind, setDriverKind] = useState(providerOptions[0]?.driverKind); + const [externalId, setExternalId] = useState(""); + const [externalIdDirty, setExternalIdDirty] = useState(false); + const [isImporting, setIsImporting] = useState(false); + const [errorMessage, setErrorMessage] = useState(null); + const [resolved, setResolved] = useState(null); + const importInFlightRef = useRef(false); + const externalIdRef = useRef(""); + const createProject = useAtomCommand(projectEnvironment.create, { reportFailure: false }); + const resolveImportSession = useAtomCommand(threadEnvironment.resolveImportSession, { + reportFailure: false, + }); + const importThread = useAtomCommand(threadEnvironment.importThread, { reportFailure: false }); + + const selectedOption = + providerOptions.find((option) => option.driverKind === driverKind) ?? providerOptions[0]; + const step = describeImportSessionStep(resolved); + + const stopImporting = useCallback((message: string | null) => { + importInFlightRef.current = false; + setIsImporting(false); + setErrorMessage(message); + }, []); + + useEffect(() => { + if (!open) return; + const frame = window.requestAnimationFrame(() => { + externalIdInputRef.current?.focus(); + }); + return () => { + window.cancelAnimationFrame(frame); + }; + }, [open]); + + const handleImport = useCallback(async () => { + if (importInFlightRef.current) { + return; + } + const trimmedExternalId = externalId.trim(); + externalIdRef.current = trimmedExternalId; + if (!selectedOption || trimmedExternalId.length === 0) { + setExternalIdDirty(true); + return; + } + importInFlightRef.current = true; + setErrorMessage(null); + setIsImporting(true); + + let session = resolved?.externalId === trimmedExternalId ? resolved : null; + if (session === null) { + const resolveResult = await resolveImportSession({ + environmentId: projectRef.environmentId, + input: { + instanceId: selectedOption.modelSelection.instanceId, + externalId: trimmedExternalId, + }, + }); + if (resolveResult._tag === "Failure") { + return stopImporting(failureMessage(resolveResult, "Could not find that session.")); + } + if (externalIdRef.current !== trimmedExternalId) { + return stopImporting(null); + } + session = resolveResult.value; + setResolved(session); + if (session.projectId === null && session.workspaceRoot !== null) { + return stopImporting(null); + } + } + + let projectId = session.projectId ?? projectRef.projectId; + if (session.projectId === null && session.workspaceRoot !== null) { + const createdProjectId = newProjectId(); + const createResult = await createProject({ + environmentId: projectRef.environmentId, + input: { + projectId: createdProjectId, + title: inferProjectTitleFromPath(session.workspaceRoot), + workspaceRoot: session.workspaceRoot, + defaultModelSelection: selectedOption.modelSelection, + }, + }); + if (createResult._tag === "Failure") { + return stopImporting( + failureMessage(createResult, `Could not add ${session.workspaceRoot} as a project.`), + ); + } + projectId = createdProjectId; + session = { ...session, projectId: createdProjectId }; + setResolved(session); + } + + const importResult = await importThread({ + environmentId: projectRef.environmentId, + input: { + projectId, + modelSelection: selectedOption.modelSelection, + externalId: trimmedExternalId, + }, + }); + if (importResult._tag === "Failure") { + return stopImporting(failureMessage(importResult, "Could not import that session.")); + } + + stopImporting(null); + onOpenChange(false); + await navigate({ + to: "/$environmentId/$threadId", + params: buildThreadRouteParams( + scopeThreadRef(projectRef.environmentId, importResult.value.threadId), + ), + }); + }, [ + createProject, + externalId, + importThread, + navigate, + onOpenChange, + projectRef, + resolveImportSession, + resolved, + selectedOption, + stopImporting, + ]); + + const validationMessage = + externalIdDirty && externalId.trim().length === 0 + ? `Paste the ${selectedOption?.fieldLabel.toLowerCase() ?? "session id"} you want to continue.` + : null; + + return ( + { + if (!isImporting) { + onOpenChange(nextOpen); + } + }} + > + + + Import session + + Continue a Claude Code or Codex session that already exists on this machine. Its + conversation is copied into a new thread in the project it ran in, and the next turn + resumes it. + + + +
+ {providerOptions.map((option) => ( + + ))} +
+ + + + {step.notice ?

{step.notice}

: null} + + {(validationMessage ?? errorMessage) ? ( +

{validationMessage ?? errorMessage}

+ ) : null} + + + + + + + + ); +} diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index ad25d6544dc..2fc328aa4e3 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -2,6 +2,8 @@ import { CommandId, ORCHESTRATION_WS_METHODS, type ClientOrchestrationCommand, + type OrchestrationImportThreadInput, + type OrchestrationResolveImportSessionInput, } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; @@ -48,6 +50,8 @@ export type RespondToThreadApprovalInput = CommandInput<"thread.approval.respond export type RespondToThreadUserInputInput = CommandInput<"thread.user-input.respond">; export type RevertThreadCheckpointInput = CommandInput<"thread.checkpoint.revert">; export type StopThreadSessionInput = CommandInput<"thread.session.stop">; +export type ImportThreadInput = OrchestrationImportThreadInput; +export type ResolveImportSessionInput = OrchestrationResolveImportSessionInput; type DispatchTag = typeof ORCHESTRATION_WS_METHODS.dispatchCommand; type CommandEffect = Effect.Effect< @@ -56,6 +60,20 @@ type CommandEffect = Effect.Effect< Crypto.Crypto | EnvironmentSupervisor >; +type ImportThreadTag = typeof ORCHESTRATION_WS_METHODS.importThread; +type ImportThreadEffect = Effect.Effect< + EnvironmentRpcSuccess, + EnvironmentRpcFailure | EnvironmentRpcUnavailableError, + EnvironmentSupervisor +>; + +type ResolveImportSessionTag = typeof ORCHESTRATION_WS_METHODS.resolveImportSession; +type ResolveImportSessionEffect = Effect.Effect< + EnvironmentRpcSuccess, + EnvironmentRpcFailure | EnvironmentRpcUnavailableError, + EnvironmentSupervisor +>; + function commandId(input: { readonly commandId?: CommandId }) { return Effect.gen(function* () { if (input.commandId !== undefined) { @@ -298,3 +316,17 @@ export const stopThreadSession: (input: StopThreadSessionInput) => CommandEffect createdAt: metadata.createdAt, }); }); + +export const resolveImportSession: ( + input: ResolveImportSessionInput, +) => ResolveImportSessionEffect = Effect.fn("EnvironmentCommands.resolveImportSession")( + function* (input) { + return yield* request(ORCHESTRATION_WS_METHODS.resolveImportSession, input); + }, +); + +export const importThread: (input: ImportThreadInput) => ImportThreadEffect = Effect.fn( + "EnvironmentCommands.importThread", +)(function* (input) { + return yield* request(ORCHESTRATION_WS_METHODS.importThread, input); +}); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 6c128eb01ab..0e6f5b2c6b9 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -6,6 +6,8 @@ import { type ArchiveThreadInput, type CreateThreadInput, type DeleteThreadInput, + type ImportThreadInput, + type ResolveImportSessionInput, type InterruptThreadTurnInput, type RespondToThreadApprovalInput, type RespondToThreadUserInputInput, @@ -23,7 +25,9 @@ import { archiveThread, createThread, deleteThread, + importThread, interruptThreadTurn, + resolveImportSession, respondToThreadApproval, respondToThreadUserInput, revertThreadCheckpoint, @@ -44,7 +48,9 @@ export type { ArchiveThreadInput, CreateThreadInput, DeleteThreadInput, + ImportThreadInput, InterruptThreadTurnInput, + ResolveImportSessionInput, RespondToThreadApprovalInput, RespondToThreadUserInputInput, RevertThreadCheckpointInput, @@ -69,6 +75,11 @@ export function createThreadEnvironmentAtoms( key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => JSON.stringify([environmentId, input.threadId]), }; + const importConcurrency = { + mode: "serial" as const, + key: ({ environmentId, input }: { environmentId: string; input: { externalId: string } }) => + JSON.stringify([environmentId, input.externalId]), + }; return { create: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:create", @@ -172,5 +183,17 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + resolveImportSession: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:resolve-import-session", + execute: (input: ResolveImportSessionInput) => resolveImportSession(input), + scheduler, + concurrency: importConcurrency, + }), + importThread: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:import-thread", + execute: (input: ImportThreadInput) => importThread(input), + scheduler, + concurrency: importConcurrency, + }), }; } diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 211f8748f4e..285fda517d4 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -396,6 +396,57 @@ describe("applyThreadDetailEvent", () => { }); }); + describe("thread.messages-imported", () => { + const importedEvent = { + ...baseEventFields, + sequence: 8, + occurredAt: "2026-04-01T07:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.messages-imported", + payload: { + threadId: ThreadId.make("thread-1"), + messages: [ + { + messageId: MessageId.make("imported:claudeAgent:thread-1:000000:uuid-1"), + role: "user", + text: "Fix the flaky test", + createdAt: "2026-04-01T07:00:00.000Z", + updatedAt: "2026-04-01T07:00:00.000Z", + }, + { + messageId: MessageId.make("imported:claudeAgent:thread-1:000001:uuid-2"), + role: "assistant", + text: "Looking at it", + createdAt: "2026-04-01T07:00:00.000Z", + updatedAt: "2026-04-01T07:00:00.000Z", + }, + ], + }, + } as const; + + it("appends the imported transcript", () => { + const result = applyThreadDetailEvent(baseThread, importedEvent); + + expect(result.kind).toBe("updated"); + if (result.kind === "updated") { + expect(result.thread.messages.map((message) => [message.role, message.text])).toEqual([ + ["user", "Fix the flaky test"], + ["assistant", "Looking at it"], + ]); + } + }); + + it("returns unchanged when the snapshot already carries the imported messages", () => { + const imported = applyThreadDetailEvent(baseThread, importedEvent); + if (imported.kind !== "updated") { + throw new Error("Expected the first import to update the thread."); + } + + expect(applyThreadDetailEvent(imported.thread, importedEvent).kind).toBe("unchanged"); + }); + }); + describe("thread.session-set", () => { it("settles a running latestTurn when the session leaves the running status", () => { const threadWithRunningTurn: OrchestrationThread = { diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index ce0dca52f5a..5ca82ab8889 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -322,6 +322,33 @@ export function applyThreadDetailEvent( }; } + case "thread.messages-imported": { + const knownMessageIds = new Set(thread.messages.map((entry) => entry.id)); + const importedMessages: OrchestrationMessage[] = Arr.filter( + event.payload.messages, + (message) => !knownMessageIds.has(message.messageId), + ).map((message) => ({ + id: message.messageId, + role: message.role, + text: message.text, + turnId: null, + streaming: false, + createdAt: message.createdAt, + updatedAt: message.updatedAt, + })); + if (importedMessages.length === 0) { + return { kind: "unchanged" }; + } + return { + kind: "updated", + thread: { + ...thread, + messages: Arr.appendAll(thread.messages, importedMessages), + updatedAt: event.occurredAt, + }, + }; + } + // ── Session ───────────────────────────────────────────────────── case "thread.session-set": { // Leaving the "running" session status is the turn-end signal: settle a diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 84b7a8fa07f..a5ccb8f77e6 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -24,6 +24,8 @@ import { ProviderInstanceId } from "./providerInstance.ts"; export const ORCHESTRATION_WS_METHODS = { dispatchCommand: "orchestration.dispatchCommand", + resolveImportSession: "orchestration.resolveImportSession", + importThread: "orchestration.importThread", getTurnDiff: "orchestration.getTurnDiff", getFullThreadDiff: "orchestration.getFullThreadDiff", replayEvents: "orchestration.replayEvents", @@ -746,6 +748,23 @@ const ThreadSessionStopCommand = Schema.Struct({ createdAt: IsoDateTime, }); +export const ThreadImportedMessage = Schema.Struct({ + messageId: MessageId, + role: Schema.Literals(["user", "assistant"]), + text: Schema.String, + createdAt: IsoDateTime, + updatedAt: IsoDateTime, +}); +export type ThreadImportedMessage = typeof ThreadImportedMessage.Type; + +const ThreadMessagesImportCommand = Schema.Struct({ + type: Schema.Literal("thread.messages.import"), + commandId: CommandId, + threadId: ThreadId, + messages: Schema.Array(ThreadImportedMessage), + createdAt: IsoDateTime, +}); + const DispatchableClientOrchestrationCommand = Schema.Union([ ProjectCreateCommand, ProjectMetaUpdateCommand, @@ -862,6 +881,7 @@ const ThreadRevertCompleteCommand = Schema.Struct({ const InternalOrchestrationCommand = Schema.Union([ ThreadSessionSetCommand, + ThreadMessagesImportCommand, ThreadMessageAssistantDeltaCommand, ThreadMessageAssistantCompleteCommand, ThreadProposedPlanUpsertCommand, @@ -893,6 +913,7 @@ export const OrchestrationEventType = Schema.Literals([ "thread.runtime-mode-set", "thread.interaction-mode-set", "thread.message-sent", + "thread.messages-imported", "thread.turn-start-requested", "thread.turn-interrupt-requested", "thread.approval-response-requested", @@ -1032,6 +1053,11 @@ export const ThreadMessageSentPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadMessagesImportedPayload = Schema.Struct({ + threadId: ThreadId, + messages: Schema.Array(ThreadImportedMessage), +}); + export const ThreadTurnStartRequestedPayload = Schema.Struct({ threadId: ThreadId, messageId: MessageId, @@ -1204,6 +1230,11 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.message-sent"), payload: ThreadMessageSentPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.messages-imported"), + payload: ThreadMessagesImportedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.turn-start-requested"), @@ -1341,6 +1372,34 @@ export const DispatchResult = Schema.Struct({ }); export type DispatchResult = typeof DispatchResult.Type; +export const OrchestrationResolveImportSessionInput = Schema.Struct({ + instanceId: ProviderInstanceId, + externalId: TrimmedNonEmptyString, +}); +export type OrchestrationResolveImportSessionInput = + typeof OrchestrationResolveImportSessionInput.Type; + +export const OrchestrationResolveImportSessionResult = Schema.Struct({ + externalId: TrimmedNonEmptyString, + workspaceRoot: Schema.NullOr(TrimmedNonEmptyString), + projectId: Schema.NullOr(ProjectId), + title: Schema.NullOr(TrimmedNonEmptyString), +}); +export type OrchestrationResolveImportSessionResult = + typeof OrchestrationResolveImportSessionResult.Type; + +export const OrchestrationImportThreadInput = Schema.Struct({ + projectId: ProjectId, + modelSelection: ModelSelection, + externalId: TrimmedNonEmptyString, +}); +export type OrchestrationImportThreadInput = typeof OrchestrationImportThreadInput.Type; + +export const OrchestrationImportThreadResult = Schema.Struct({ + threadId: ThreadId, +}); +export type OrchestrationImportThreadResult = typeof OrchestrationImportThreadResult.Type; + export const OrchestrationGetTurnDiffInput = TurnCountRange.mapFields( Struct.assign({ threadId: ThreadId, @@ -1376,6 +1435,14 @@ export const OrchestrationRpcSchemas = { input: ClientOrchestrationCommand, output: DispatchResult, }, + resolveImportSession: { + input: OrchestrationResolveImportSessionInput, + output: OrchestrationResolveImportSessionResult, + }, + importThread: { + input: OrchestrationImportThreadInput, + output: OrchestrationImportThreadResult, + }, getTurnDiff: { input: OrchestrationGetTurnDiffInput, output: OrchestrationGetTurnDiffResult, @@ -1418,6 +1485,14 @@ export class OrchestrationDispatchCommandError extends Schema.TaggedErrorClass()( + "OrchestrationImportThreadError", + { + message: TrimmedNonEmptyString, + cause: Schema.optional(Schema.Defect()), + }, +) {} + export class OrchestrationGetTurnDiffError extends Schema.TaggedErrorClass()( "OrchestrationGetTurnDiffError", { diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index fa2d23b8ef2..a8dfc0e67f2 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -54,6 +54,9 @@ import { OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, OrchestrationGetTurnDiffInput, + OrchestrationImportThreadError, + OrchestrationImportThreadInput, + OrchestrationResolveImportSessionInput, OrchestrationReplayEventsError, OrchestrationReplayEventsInput, OrchestrationRpcSchemas, @@ -616,6 +619,21 @@ export const WsOrchestrationDispatchCommandRpc = Rpc.make( }, ); +export const WsOrchestrationResolveImportSessionRpc = Rpc.make( + ORCHESTRATION_WS_METHODS.resolveImportSession, + { + payload: OrchestrationResolveImportSessionInput, + success: OrchestrationRpcSchemas.resolveImportSession.output, + error: Schema.Union([OrchestrationImportThreadError, EnvironmentAuthorizationError]), + }, +); + +export const WsOrchestrationImportThreadRpc = Rpc.make(ORCHESTRATION_WS_METHODS.importThread, { + payload: OrchestrationImportThreadInput, + success: OrchestrationRpcSchemas.importThread.output, + error: Schema.Union([OrchestrationImportThreadError, EnvironmentAuthorizationError]), +}); + export const WsOrchestrationGetTurnDiffRpc = Rpc.make(ORCHESTRATION_WS_METHODS.getTurnDiff, { payload: OrchestrationGetTurnDiffInput, success: OrchestrationRpcSchemas.getTurnDiff.output, @@ -763,6 +781,8 @@ export const WsRpcGroup = RpcGroup.make( WsSubscribeServerLifecycleRpc, WsSubscribeAuthAccessRpc, WsOrchestrationDispatchCommandRpc, + WsOrchestrationResolveImportSessionRpc, + WsOrchestrationImportThreadRpc, WsOrchestrationGetTurnDiffRpc, WsOrchestrationGetFullThreadDiffRpc, WsOrchestrationReplayEventsRpc,