From 109ce2a3068f1e04bb38badfd425c20bd746df17 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:29:49 +0000 Subject: [PATCH 01/13] feat(provider): expose readThread on the ProviderService facade Transports resolve provider adapters through ProviderService; the adapter registry is not visible past ProviderServiceLive. Reading a provider thread snapshot from an RPC handler needs a facade method, mirroring how rollbackConversation routes. --- .../Layers/CheckpointReactor.test.ts | 1 + .../Layers/ProviderCommandReactor.test.ts | 1 + .../Layers/ProviderRuntimeIngestion.test.ts | 1 + .../src/provider/Layers/ProviderService.ts | 17 +++++++++++++++++ .../Layers/ProviderSessionReaper.test.ts | 1 + .../src/provider/Services/ProviderService.ts | 6 +++++- 6 files changed, 26 insertions(+), 1 deletion(-) 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/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/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/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/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. */ From 50f22d49a5eb452b8ce0f37448eaddc298f9e37a Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:29:50 +0000 Subject: [PATCH 02/13] feat(provider): carry codex workspace root and last activity on thread snapshots thread/read already returns the thread's cwd and its recency stamp; the adapter dropped both. Callers that need to know where a codex thread ran, or when it was last active, had nothing to read. Codex reports those stamps in Unix seconds, so parseThreadSnapshot converts them and the field name carries the unit. --- .../src/provider/Layers/CodexAdapter.test.ts | 4 +++ .../src/provider/Layers/CodexAdapter.ts | 4 +++ .../Layers/CodexSessionRuntime.test.ts | 26 +++++++++++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 22 +++++++++++++--- .../src/provider/Services/ProviderAdapter.ts | 2 ++ 5 files changed, 55 insertions(+), 3 deletions(-) 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/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 { From 5532f290a09c2e19d7846e899f104b86fd027065 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:29:51 +0000 Subject: [PATCH 03/13] feat(orchestration): add the thread.messages.import command and its event Adds the command, the thread.messages-imported event, and the payloads for orchestration.resolveImportSession and orchestration.importThread, plus the decider case that turns one into the other. thread.messages.import stays in the internal command union: only the server dispatches it, and a client must not be able to write a transcript. --- .../decider.projectScripts.test.ts | 95 +++++++++++++++++++ apps/server/src/orchestration/decider.ts | 21 ++++ packages/contracts/src/orchestration.ts | 75 +++++++++++++++ 3 files changed, 191 insertions(+) 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/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", { From 1eecf6b4ea70bee6b0747dbac40336a131b540c3 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:30:12 +0000 Subject: [PATCH 04/13] feat(orchestration): map provider transcripts to imported thread messages Pure mappers from a Claude session transcript or a codex thread snapshot to ThreadImportedMessage. Non-user/assistant entries and empty text are skipped. Message ids are derived from the thread id, provider, transcript index and source message id, so a repeated import cannot duplicate rows. The index is zero-padded because the message projection breaks ties on message id, and an unpadded index would order a ten-message transcript 1, 10, 2. --- .../orchestration/importedMessages.test.ts | 162 ++++++++++++++++++ .../src/orchestration/importedMessages.ts | 121 +++++++++++++ 2 files changed, 283 insertions(+) create mode 100644 apps/server/src/orchestration/importedMessages.test.ts create mode 100644 apps/server/src/orchestration/importedMessages.ts 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), + }); +} From c9113f3d38adb521caaa5cb43ebafa73db289220 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:30:14 +0000 Subject: [PATCH 05/13] feat(orchestration): project imported thread messages Applies thread.messages-imported through the existing message projection. No new table, no migration. The command read model folds the same event so the decider's message-based invariants see an imported transcript, matching how every other event that changes thread messages is registered. --- .../Layers/ProjectionPipeline.test.ts | 107 ++++++++++++++++++ .../Layers/ProjectionPipeline.ts | 20 ++++ apps/server/src/orchestration/Schemas.ts | 2 + .../src/orchestration/projector.test.ts | 71 ++++++++++++ apps/server/src/orchestration/projector.ts | 39 +++++++ 5 files changed, 239 insertions(+) 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/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/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( From f35b330caa236f26ef7b8fa540f732866cfc3601 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:30:15 +0000 Subject: [PATCH 06/13] fix(relay): do not publish agent awareness for an imported transcript Copying a transcript into a thread is not agent activity. Every sibling event that adds message content is already excluded; thread.messages-imported fell through to the default and queued a spurious alert. --- apps/server/src/relay/AgentAwarenessRelay.test.ts | 10 ++++++++++ apps/server/src/relay/AgentAwarenessRelay.ts | 1 + 2 files changed, 11 insertions(+) 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": From 09569acf977087c519b5d3b39bcf539088438562 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:30:16 +0000 Subject: [PATCH 07/13] feat(orchestration): resume an existing provider session into a new thread resolveImportSession reads where a Claude session ran and whether a project already covers that workspace, so a caller can offer to add the missing project before anything is created. importThread creates the thread in that project, starts the provider session with a resume cursor, copies the transcript in and binds the session. Both drivers refuse a session that ran in another workspace. A failure after the thread exists deletes it, and ThreadDeletionReactor stops the provider session behind that. resolveImportSession takes the orchestration read scope; importThread takes the operate scope. --- .../Layers/ProviderCommandReactor.ts | 2 +- .../src/orchestration/importThread.test.ts | 248 +++++++++++++ apps/server/src/orchestration/importThread.ts | 326 ++++++++++++++++++ apps/server/src/server.test.ts | 31 +- apps/server/src/ws.ts | 25 ++ packages/contracts/src/rpc.ts | 20 ++ 6 files changed, 639 insertions(+), 13 deletions(-) create mode 100644 apps/server/src/orchestration/importThread.test.ts create mode 100644 apps/server/src/orchestration/importThread.ts 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/importThread.test.ts b/apps/server/src/orchestration/importThread.test.ts new file mode 100644 index 00000000000..3f9c5d21483 --- /dev/null +++ b/apps/server/src/orchestration/importThread.test.ts @@ -0,0 +1,248 @@ +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 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; +}) { + 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) => 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("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..5689cf6ccfa --- /dev/null +++ b/apps/server/src/orchestration/importThread.ts @@ -0,0 +1,326 @@ +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 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 importFailure = (message: string) => (cause: unknown) => + new OrchestrationImportThreadError({ message, cause }); + +const readClaudeSessionInfo = (externalId: string) => + Effect.tryPromise({ + try: () => getSessionInfo(externalId, {}), + catch: importFailure(`Could not read Claude Code session '${externalId}'.`), + }); + +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(importFailure(`Provider instance '${instanceId}' is not available.`))); + 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: importFailure( + `Could not read the transcript for Claude Code session '${input.externalId}'.`, + ), + }); + return mapClaudeSessionMessages({ + threadId: input.threadId, + importedAt: input.transcriptAt ?? input.importedAt, + messages: sessionMessages, + }); + } + + const snapshot = yield* providerService + .readThread(input.threadId) + .pipe( + Effect.mapError( + importFailure(`Could not read the transcript for Codex thread '${input.externalId}'.`), + ), + ); + 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 deletingCreatedThreadOnFailure = ( + threadId: ThreadId, + effect: Effect.Effect, + ) => + effect.pipe( + Effect.tapError(() => + commandId("delete").pipe( + Effect.flatMap((deleteCommandId) => + orchestrationEngine.dispatch({ + type: "thread.delete", + commandId: deleteCommandId, + threadId, + }), + ), + Effect.catchCause((cause) => + Effect.logWarning("failed to delete thread after a failed 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(importFailure(`Could not look up a project for '${workspaceRoot}'.`)), + ); + + 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(importFailure(`Could not read project '${input.projectId}'.`)), + ); + 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(importFailure("Could not create a thread for the imported session."))); + + yield* deletingCreatedThreadOnFailure( + 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( + importFailure(`Could not resume session '${input.externalId}' for this thread.`), + ), + ); + + 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(importFailure("Could not store the imported transcript."))); + + 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(importFailure("Could not bind the resumed session to this thread.")), + ); + }), + ); + + return { threadId }; + }); + + return { resolveImportSession, importThread }; +}; 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/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, From cf680e0bafbe1a082242b4055d6dd43cd31829e3 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:30:18 +0000 Subject: [PATCH 08/13] feat(client-runtime): add the session-import operations resolveImportSession and importThread operations plus their atom commands, serialised per session id so the same import cannot run twice concurrently. The thread reducer applies thread.messages-imported so an open thread shows the transcript without waiting for a fresh snapshot. --- .../client-runtime/src/operations/commands.ts | 32 ++++++++++++ .../src/state/threadCommands.ts | 23 +++++++++ .../src/state/threadReducer.test.ts | 51 +++++++++++++++++++ .../client-runtime/src/state/threadReducer.ts | 27 ++++++++++ 4 files changed, 133 insertions(+) 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 From e48610f268970cb7559d3ffde33889d52c408c56 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 02:30:19 +0000 Subject: [PATCH 09/13] feat(web): add the Import session command palette action Opens a dialog that takes one session id, offers Claude Code and Codex, and labels the field Session ID or Thread ID to match. On submit it resolves where the session ran. If a project already covers that workspace the thread lands there; if none does, the dialog says so and the button becomes Add project & import, so nothing is created without a second press. --- .../components/CommandPalette.logic.test.ts | 128 +++++++- .../src/components/CommandPalette.logic.ts | 72 +++++ apps/web/src/components/CommandPalette.tsx | 62 +++- .../src/components/ImportSessionDialog.tsx | 278 ++++++++++++++++++ 4 files changed, 538 insertions(+), 2 deletions(-) create mode 100644 apps/web/src/components/ImportSessionDialog.tsx 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..9fe6399a688 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,70 @@ 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 entry = entries.find( + (item) => item.driverKind === candidate.driverKind && isProviderInstancePickerReady(item), + ); + 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..c32715f2756 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({ @@ -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..7146d2a4361 --- /dev/null +++ b/apps/web/src/components/ImportSessionDialog.tsx @@ -0,0 +1,278 @@ +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; + } + + 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} + + + + + + + + ); +} From 581459f8957e0c269daf07f721617cb85e378b9f Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 08:43:14 +0000 Subject: [PATCH 10/13] fix(web): bind the import provider option to the default instance deriveProviderInstanceEntries preserves server ordering, where a configured custom instance can precede the synthesised default. Taking the first ready entry meant the generic Claude Code / Codex option silently resolved through a custom instance and its model configuration. --- apps/web/src/components/CommandPalette.logic.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/web/src/components/CommandPalette.logic.ts b/apps/web/src/components/CommandPalette.logic.ts index 9fe6399a688..7294bd0bc60 100644 --- a/apps/web/src/components/CommandPalette.logic.ts +++ b/apps/web/src/components/CommandPalette.logic.ts @@ -405,9 +405,10 @@ export function buildImportSessionProviderOptions( ): ImportSessionProviderOption[] { const entries = deriveProviderInstanceEntries(providers); return IMPORT_SESSION_PROVIDERS.flatMap((candidate) => { - const entry = entries.find( + const readyEntries = entries.filter( (item) => item.driverKind === candidate.driverKind && isProviderInstancePickerReady(item), ); + const entry = readyEntries.find((item) => item.isDefault) ?? readyEntries[0]; if (!entry) { return []; } From fcb9ee59c63251921e3e1b055c211cf10155f0f4 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 08:43:15 +0000 Subject: [PATCH 11/13] fix(web): ignore the palette shortcut while the import dialog is open The global keydown listener toggled the command palette regardless of the import dialog, stacking two modals with competing focus handling. --- apps/web/src/components/CommandPalette.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/web/src/components/CommandPalette.tsx b/apps/web/src/components/CommandPalette.tsx index c32715f2756..02c9ab13604 100644 --- a/apps/web/src/components/CommandPalette.tsx +++ b/apps/web/src/components/CommandPalette.tsx @@ -426,7 +426,7 @@ export function CommandPalette({ children }: { children: ReactNode }) { terminalOpen, }, }); - if (command !== "commandPalette.toggle") { + if (command !== "commandPalette.toggle" || importSession !== null) { return; } event.preventDefault(); @@ -435,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( () => From b5250c8f91d935d43095f41b8332900adcb0c0b5 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 08:43:17 +0000 Subject: [PATCH 12/13] fix(web): reuse the created project when a failed import is retried A project created for the session stayed invisible to the retry path, so pressing Import again built a second project for the same workspace and hit the duplicate-workspace invariant. --- apps/web/src/components/ImportSessionDialog.tsx | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/web/src/components/ImportSessionDialog.tsx b/apps/web/src/components/ImportSessionDialog.tsx index 7146d2a4361..51463808f6f 100644 --- a/apps/web/src/components/ImportSessionDialog.tsx +++ b/apps/web/src/components/ImportSessionDialog.tsx @@ -138,6 +138,8 @@ export function ImportSessionDialog({ ); } projectId = createdProjectId; + session = { ...session, projectId: createdProjectId }; + setResolved(session); } const importResult = await importThread({ From db5946cee1a3d199db594ab18c30046f9b66cc49 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Mon, 27 Jul 2026 08:43:18 +0000 Subject: [PATCH 13/13] fix(orchestration): delete the created thread when an import is interrupted tapError never runs on fiber interruption, so a client disconnecting mid import left the thread in the sidebar and its provider session running. onExit compensates on any non-success exit. Error construction moves to each failure boundary, per the service conventions, so the message and cause stay visible where they originate. --- .../src/orchestration/importThread.test.ts | 25 +++- apps/server/src/orchestration/importThread.ts | 139 ++++++++++++------ 2 files changed, 120 insertions(+), 44 deletions(-) diff --git a/apps/server/src/orchestration/importThread.test.ts b/apps/server/src/orchestration/importThread.test.ts index 3f9c5d21483..33afa8c03d9 100644 --- a/apps/server/src/orchestration/importThread.test.ts +++ b/apps/server/src/orchestration/importThread.test.ts @@ -12,6 +12,7 @@ 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"; @@ -61,6 +62,7 @@ function makeHarness(options?: { readonly driverKind?: string; readonly items?: ReadonlyArray; readonly snapshotWorkspaceRoot?: string; + readonly startSessionNeverSettles?: boolean; }) { const dispatched: OrchestrationCommand[] = []; const instanceId = options?.instanceId ?? codexInstanceId; @@ -90,7 +92,10 @@ function makeHarness(options?: { continuationKey: `${driverKind}:instance:${instanceId}`, }, }), - startSession: (threadId) => Effect.succeed({ ...codexSession, threadId }), + startSession: (threadId) => + options?.startSessionNeverSettles === true + ? Effect.never + : Effect.succeed({ ...codexSession, threadId }), readThread: (threadId) => Effect.succeed({ threadId, @@ -192,6 +197,24 @@ describe("importThread", () => { }), ); + 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 }); diff --git a/apps/server/src/orchestration/importThread.ts b/apps/server/src/orchestration/importThread.ts index 5689cf6ccfa..28f27874959 100644 --- a/apps/server/src/orchestration/importThread.ts +++ b/apps/server/src/orchestration/importThread.ts @@ -15,6 +15,7 @@ import { 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"; @@ -33,13 +34,14 @@ function isImportableDriver(driver: string): driver is ImportableDriver { const nowIso = Effect.map(DateTime.now, DateTime.formatIso); -const importFailure = (message: string) => (cause: unknown) => - new OrchestrationImportThreadError({ message, cause }); - const readClaudeSessionInfo = (externalId: string) => Effect.tryPromise({ try: () => getSessionInfo(externalId, {}), - catch: importFailure(`Could not read Claude Code session '${externalId}'.`), + catch: (cause) => + new OrchestrationImportThreadError({ + message: `Could not read Claude Code session '${externalId}'.`, + cause, + }), }); const claudeSessionNotFound = (externalId: string) => @@ -86,9 +88,15 @@ export const makeImportThread = (dependencies: { const requireImportableDriver = Effect.fnUntraced(function* ( instanceId: ProviderInstanceId, ): Effect.fn.Return { - const instanceInfo = yield* providerService - .getInstanceInfo(instanceId) - .pipe(Effect.mapError(importFailure(`Provider instance '${instanceId}' is not available.`))); + 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; @@ -109,9 +117,11 @@ export const makeImportThread = (dependencies: { if (input.driver === "claudeAgent") { const sessionMessages = yield* Effect.tryPromise({ try: () => getSessionMessages(input.externalId, {}), - catch: importFailure( - `Could not read the transcript for Claude Code session '${input.externalId}'.`, - ), + catch: (cause) => + new OrchestrationImportThreadError({ + message: `Could not read the transcript for Claude Code session '${input.externalId}'.`, + cause, + }), }); return mapClaudeSessionMessages({ threadId: input.threadId, @@ -120,13 +130,15 @@ export const makeImportThread = (dependencies: { }); } - const snapshot = yield* providerService - .readThread(input.threadId) - .pipe( - Effect.mapError( - importFailure(`Could not read the transcript for Codex thread '${input.externalId}'.`), - ), - ); + 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({ @@ -140,24 +152,29 @@ export const makeImportThread = (dependencies: { }); }); - const deletingCreatedThreadOnFailure = ( + const deletingCreatedThreadUnlessImported = ( threadId: ThreadId, effect: Effect.Effect, ) => effect.pipe( - Effect.tapError(() => - commandId("delete").pipe( - Effect.flatMap((deleteCommandId) => - orchestrationEngine.dispatch({ - type: "thread.delete", - commandId: deleteCommandId, - threadId, - }), - ), - Effect.catchCause((cause) => - Effect.logWarning("failed to delete thread after a failed import", { threadId, cause }), - ), - ), + 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, + }), + ), + ), ), ); @@ -188,7 +205,13 @@ export const makeImportThread = (dependencies: { .getActiveProjectByWorkspaceRoot(workspaceRoot) .pipe( Effect.map(Option.getOrUndefined), - Effect.mapError(importFailure(`Could not look up a project for '${workspaceRoot}'.`)), + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: `Could not look up a project for '${workspaceRoot}'.`, + cause, + }), + ), ); return { @@ -202,12 +225,16 @@ export const makeImportThread = (dependencies: { const importThread = Effect.fnUntraced(function* ( input: OrchestrationImportThreadInput, ): Effect.fn.Return { - const project = yield* projectionSnapshotQuery - .getProjectShellById(input.projectId) - .pipe( - Effect.map(Option.getOrUndefined), - Effect.mapError(importFailure(`Could not read project '${input.projectId}'.`)), - ); + 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.`, @@ -248,9 +275,17 @@ export const makeImportThread = (dependencies: { worktreePath: null, createdAt, }) - .pipe(Effect.mapError(importFailure("Could not create a thread for the imported session."))); + .pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: "Could not create a thread for the imported session.", + cause, + }), + ), + ); - yield* deletingCreatedThreadOnFailure( + yield* deletingCreatedThreadUnlessImported( threadId, Effect.gen(function* () { const session = yield* providerService @@ -268,7 +303,11 @@ export const makeImportThread = (dependencies: { }) .pipe( Effect.mapError( - importFailure(`Could not resume session '${input.externalId}' for this thread.`), + (cause) => + new OrchestrationImportThreadError({ + message: `Could not resume session '${input.externalId}' for this thread.`, + cause, + }), ), ); @@ -294,7 +333,15 @@ export const makeImportThread = (dependencies: { messages, createdAt: importedAt, }) - .pipe(Effect.mapError(importFailure("Could not store the imported transcript."))); + .pipe( + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: "Could not store the imported transcript.", + cause, + }), + ), + ); yield* orchestrationEngine .dispatch({ @@ -314,7 +361,13 @@ export const makeImportThread = (dependencies: { createdAt: yield* nowIso, }) .pipe( - Effect.mapError(importFailure("Could not bind the resumed session to this thread.")), + Effect.mapError( + (cause) => + new OrchestrationImportThreadError({ + message: "Could not bind the resumed session to this thread.", + cause, + }), + ), ); }), );