diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 6c16ecaf91..f7ea816884 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -267,6 +267,7 @@ export const make = Effect.gen(function* () { peerLinks: true, mcpModeLimitHeader: true, linkFence: true, + threadImport: true, }, }; diff --git a/apps/server/src/mcp/linkOrigin.test.ts b/apps/server/src/mcp/linkOrigin.test.ts index fc5a85e45e..e0bc19ab69 100644 --- a/apps/server/src/mcp/linkOrigin.test.ts +++ b/apps/server/src/mcp/linkOrigin.test.ts @@ -45,6 +45,7 @@ import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as ThreadSearch from "../orchestration-v2/ThreadSearch.ts"; import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../peer/PeerLinks.ts"; +import * as ThreadImportService from "../orchestration-v2/ThreadImportService.ts"; // A linked environment's session drives this one's real orchestrator through // its real T3 tools. What the link starts carries its origin, and the link @@ -159,6 +160,7 @@ const layerTools = Layer.mergeAll( Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadImportService.ThreadImportService)({})), Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), Layer.provide( diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index 2e01b19eda..b7f0ad4df4 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -9,6 +9,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import * as ThreadLaunch from "../../../orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadImportService from "../../../orchestration-v2/ThreadImportService.ts"; import * as ThreadMessageIntake from "../../../orchestration-v2/ThreadMessageIntake.ts"; import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts"; import * as Project from "../../../project/ProjectService.ts"; @@ -203,6 +204,32 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { // The launched thread carries the caller's link, if it has one. "stamped", ), + t3_thread_import: McpToolAccess.startsThreads( + (input) => input, + (input, { runtimeMode, interactionMode, linkOrigin }) => + Effect.gen(function* () { + const imports = yield* ThreadImportService.ThreadImportService; + return yield* imports + .importThread({ ...input, runtimeMode, interactionMode, linkOrigin }) + .pipe( + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ + code: + error._tag === "ThreadImportProjectMismatchError" || + error._tag === "ThreadImportLinkMismatchError" || + error._tag === "ThreadImportValidationError" || + error._tag === "ThreadManagementProviderInstanceDisabledError" + ? "invalid_request" + : "orchestration_error", + message: error.message, + }), + ), + ); + }), + // A moved thread carries the link that moved it, like a launch. + "stamped", + ), t3_project_list: McpToolAccess.reads((input) => Effect.gen(function* () { const projects = yield* access; diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index 4abf469e49..bbae521837 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/apps/server/src/mcp/toolkits/project/tools.ts @@ -17,12 +17,15 @@ import { OrchestratorMcpClientRequestId, OrchestratorMcpEnvironmentTarget, OrchestratorMcpFailure, + OrchestratorMcpThreadImportInput, + OrchestratorMcpThreadImportResult, SourceControlCloneRepositoryInput, SourceControlCloneRepositoryResult, } from "@t3tools/contracts"; import * as FileSystem from "effect/FileSystem"; import * as ServerConfig from "../../../config.ts"; import * as ServerSettings from "../../../serverSettings.ts"; +import * as ThreadImportService from "../../../orchestration-v2/ThreadImportService.ts"; import * as ThreadLaunchService from "../../../orchestration-v2/ThreadLaunchService.ts"; import * as Crypto from "effect/Crypto"; import * as Schema from "effect/Schema"; @@ -174,8 +177,20 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { .annotate(Tool.Destructive, true) .annotate(Tool.OpenWorld, true); +const ThreadImportTool = Tool.make("t3_thread_import", { + ...shared, + description: + "Create a thread here that continues a conversation from another T3 Code environment, as when a thread is handed off to this one. The messages become the thread's history and its first run gets them as context. continuationPrompt starts work only when this call creates the import. Retrying the same source returns the accepted thread and its original continuation run without changing it. It runs within the caller's modes.", + parameters: OrchestratorMcpThreadImportInput, + success: OrchestratorMcpThreadImportResult, + dependencies: [...shared.dependencies, ThreadImportService.ThreadImportService], +}) + .annotate(Tool.Destructive, true) + .annotate(Tool.Idempotent, true); + export const ProjectToolkit = Toolkit.make( ThreadLaunchTool, + ThreadImportTool, ProjectListTool, ProjectReadTool, ProjectCreateTool, diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 59f6869739..f4cc54b173 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -38,6 +38,7 @@ import * as PreviewBrowser from "../../../preview/PreviewBrowser.ts"; import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../../peer/PeerLinks.ts"; +import * as ThreadImportService from "../../../orchestration-v2/ThreadImportService.ts"; const layerStubServices = Layer.mergeAll( Layer.mock(Orchestrator.OrchestratorV2)({}), @@ -50,6 +51,7 @@ const layerStubServices = Layer.mergeAll( Layer.mock(PeerForwarding.PeerForwarding)({}), Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), Layer.mock(PeerLinks.PeerLinks)({}), + Layer.mock(ThreadImportService.ThreadImportService)({}), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ProjectService.ProjectService)({}), diff --git a/apps/server/src/orchestration-v2/ContextHandoffService.ts b/apps/server/src/orchestration-v2/ContextHandoffService.ts index 137f617710..a9471caa25 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffService.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffService.ts @@ -153,7 +153,7 @@ function makeLegacyImportSummary(items: ReadonlyArray): } }); const header = - "Imported conversation history from the previous T3 Code orchestrator. Use it as context; do not repeat it unless the user asks."; + "Imported conversation history from before this thread ran here. Use it as context; do not repeat it unless the user asks."; const maxChars = 32_000; const selected: Array = []; let remaining = maxChars - header.length - 2; diff --git a/apps/server/src/orchestration-v2/ThreadImportService.test.ts b/apps/server/src/orchestration-v2/ThreadImportService.test.ts new file mode 100644 index 0000000000..f8d1900337 --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadImportService.test.ts @@ -0,0 +1,461 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EnvironmentId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + type RunId, +} from "@t3tools/contracts"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; + +import * as ServerConfig from "../config.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as EventSink from "./EventSink.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ThreadImportService from "./ThreadImportService.ts"; +import * as ThreadManagement from "./ThreadManagementService.ts"; +import { type CapturedTurn, makeCapturingCodexAdapter } from "./testkit/CapturingCodexAdapter.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; + +const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }; +const source = { + environmentId: EnvironmentId.make("environment:laptop"), + threadId: ThreadId.make("thread:on-the-laptop"), + handoffId: "handoff-1", +}; +const history = [ + { + role: "user" as const, + text: "Find why the login test flakes.", + createdAt: "2026-10-06T10:00:00.000Z", + }, + { + role: "assistant" as const, + text: "EARLIER_FINDING: the session cookie races the redirect.", + createdAt: "2026-10-06T10:01:00.000Z", + }, +]; +const continuation = "Carry on from where you stopped on the laptop."; +const request = { + source, + projectId: ProjectId.make("project:box"), + title: "Flaky login test", + modelSelection, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + linkOrigin: { sessionId: "session-laptop", label: "T3 Code ยท Laptop" }, + messages: history, +}; +const otherProjectId = ProjectId.make("project:other"); + +const layerFor = ( + workspace: string, + captured: Ref.Ref>, + concurrentProjectId?: ProjectId, + afterImportWrite?: Effect.Effect, +) => { + const layerDatabase = SqlitePersistence.layerMemory; + const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( + { + name: "thread-import", + runtimePolicyOverride: { + cwd: workspace, + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false }, + }, + }, + ProviderAdapterRegistry.layerSingle( + makeCapturingCodexAdapter(captured, { response: "Picked it up.", modelSelection }), + ), + { databaseLayer: layerDatabase }, + ); + const layerThreads = ThreadManagement.layer.pipe( + Layer.provide(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))), + Layer.provide(ServerSettings.layerTest()), + Layer.provideMerge(layerOrchestrator), + ); + const layerImportSink = Layer.effect( + EventSink.EventSinkV2, + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + if (concurrentProjectId === undefined && afterImportWrite === undefined) return sink; + return EventSink.EventSinkV2.of({ + ...sink, + write: (input) => + Effect.gen(function* () { + // Another writer commits after the import looked up its deterministic id. + if (concurrentProjectId !== undefined) { + yield* sink.write({ + ...input, + events: input.events.map((event) => + event.type === "thread.created" + ? { ...event, payload: { ...event.payload, projectId: concurrentProjectId } } + : event, + ), + }); + } + const stored = yield* sink.write(input); + yield* afterImportWrite ?? Effect.void; + return stored; + }), + }); + }), + ).pipe(Layer.provide(layerThreads)); + return ThreadImportService.layer.pipe( + Layer.provide(layerImportSink), + Layer.provideMerge(layerThreads), + Layer.provide(NodeCrypto.layer), + Layer.provideMerge( + GitVcsDriver.layer.pipe( + Layer.provide(ServerConfig.layerTest(workspace, { prefix: "thread-import-" })), + Layer.provide(NodeServices.layer), + ), + ), + Layer.provideMerge(NodeServices.layer), + Layer.provide( + Layer.mock(ProjectService.ProjectService)({ + getById: (id) => + Effect.succeed( + id === request.projectId || id === otherProjectId + ? Option.some({ + id, + title: "Destination", + workspaceRoot: workspace, + repositoryIdentity: null, + faviconPath: null, + defaultModelSelection: modelSelection, + defaultThreadEnvMode: null, + scripts: [], + createdAt: history[0]!.createdAt, + updatedAt: history[0]!.createdAt, + deletedAt: null, + }) + : Option.none(), + ), + }), + ), + ); +}; + +const withImport = ( + body: ( + captured: Ref.Ref>, + workspace: string, + ) => Effect.Effect, + concurrentProjectId?: ProjectId, + afterImportWrite?: Effect.Effect, +) => + Effect.scoped( + Effect.gen(function* () { + const workspace = yield* checkpointWorkspace("thread-import"); + const captured = yield* Ref.make>([]); + return yield* body(captured, workspace).pipe( + Effect.provide(layerFor(workspace, captured, concurrentProjectId, afterImportWrite)), + ); + }), + ); + +const finishRun = Effect.fnUntraced(function* (threadId: ThreadId, runId: RunId) { + const threads = yield* ThreadManagement.ThreadManagementService; + yield* threads.streamStoredEventsFrom({ threadId, eventType: "run.updated" }).pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.id === runId && + event.payload.status === "completed", + ), + Stream.runHead, + ); + return yield* threads.getThreadProjection(threadId); +}); + +it.live("stores runless history and origin once, with a new thread for a different handoff", () => + withImport((captured) => + Effect.gen(function* () { + const imports = yield* ThreadImportService.ThreadImportService; + const threads = yield* ThreadManagement.ThreadManagementService; + const first = yield* imports.importThread(request); + assert.isTrue(first.created); + assert.isNull(first.runId); + const projection = yield* threads.getThreadProjection(first.threadId); + assert.equal( + DateTime.formatIso(projection.thread.updatedAt), + DateTime.formatIso(projection.thread.createdAt), + ); + assert.equal(projection.thread.historyOrigin, "v1_import"); + assert.deepEqual(projection.thread.linkOrigin, request.linkOrigin); + assert.deepEqual( + projection.messages.map(({ role, text, runId, createdAt, attachments }) => ({ + role, + text, + runId, + createdAt: DateTime.formatIso(createdAt), + attachments, + })), + history.map((message) => ({ ...message, runId: null, attachments: [] })), + ); + assert.deepEqual( + projection.turnItems.map((item) => item.runId), + [null, null], + ); + assert.isEmpty(projection.runs); + assert.deepEqual(yield* imports.importThread(request), { ...first, created: false }); + assert.deepEqual(yield* threads.getThreadProjection(first.threadId), projection); + const next = yield* imports.importThread({ + ...request, + source: { ...source, handoffId: "handoff-2" }, + }); + assert.notEqual(next.threadId, first.threadId); + assert.isTrue(next.created); + assert.isNull(next.runId); + assert.isEmpty(yield* Ref.get(captured)); + }), + ), +); + +it.live( + "hands the conversation to the provider and replays its original continuation after later turns", + () => + withImport((captured) => + Effect.gen(function* () { + const imports = yield* ThreadImportService.ThreadImportService; + const threads = yield* ThreadManagement.ThreadManagementService; + const input = { ...request, continuationPrompt: continuation }; + const first = yield* imports.importThread(input); + assert.isNotNull(first.runId); + const initial = yield* finishRun(first.threadId, first.runId!); + assert.lengthOf(initial.contextHandoffs, 1); + const turns = yield* Ref.get(captured); + assert.lengthOf(turns, 1); + assert.include(turns[0]!.text, "Context handoff (manual_context):"); + for (const message of history) assert.include(turns[0]!.text, message.text); + assert.include(turns[0]!.text, `User message:\n${continuation}`); + assert.deepEqual(yield* imports.importThread(input), { ...first, created: false }); + yield* threads.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:later"), + threadId: first.threadId, + messageId: MessageId.make("message:later"), + text: "Now fix the redirect.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const later = yield* threads.getThreadRecords(first.threadId, ["runs"]); + assert.lengthOf(later.runs, 2); + const settled = yield* finishRun(first.threadId, later.runs[1]!.id); + const replayed = yield* imports.importThread(input); + const after = yield* threads.getThreadRecords(first.threadId, ["messages", "runs"]); + assert.deepEqual(after.messages, settled.messages); + assert.deepEqual(after.runs, settled.runs); + assert.lengthOf(yield* Ref.get(captured), 2); + assert.deepEqual(replayed, { ...first, created: false }); + }), + ), +); + +it.live( + "keeps an accepted import unchanged when retries change the prompt, destination, or link", + () => + withImport(() => + Effect.gen(function* () { + const imports = yield* ThreadImportService.ThreadImportService; + const threads = yield* ThreadManagement.ThreadManagementService; + const first = yield* imports.importThread(request); + const before = yield* threads.getThreadProjection(first.threadId); + const differentProject = yield* imports + .importThread({ ...request, projectId: otherProjectId }) + .pipe(Effect.exit); + const promptRetry = yield* imports + .importThread({ ...request, continuationPrompt: continuation }) + .pipe(Effect.exit); + assert.deepEqual(promptRetry, Exit.succeed({ ...first, created: false })); + const afterPromptRetry = yield* threads.getThreadRecords(first.threadId, [ + "messages", + "runs", + ]); + const differentLink = yield* imports + .importThread({ + ...request, + linkOrigin: { sessionId: "another-link", label: "Another environment" }, + runtimeMode: "approval-required", + interactionMode: "plan", + continuationPrompt: "Run as another link.", + }) + .pipe(Effect.exit); + assert.deepEqual( + { + promptRetryRuns: afterPromptRetry.runs, + promptRetryMessages: afterPromptRetry.messages, + differentProjectRejected: Exit.isFailure(differentProject), + differentLinkRejected: Exit.isFailure(differentLink), + projection: yield* threads.getThreadProjection(first.threadId), + }, + { + promptRetryRuns: before.runs, + promptRetryMessages: before.messages, + differentProjectRejected: true, + differentLinkRejected: true, + projection: before, + }, + ); + }), + ), +); + +it.live( + "rejects malformed history and invalid destinations with typed errors and no partial writes", + () => + withImport((_, workspace) => + Effect.gen(function* () { + const imports = yield* ThreadImportService.ThreadImportService; + const sink = yield* EventSink.EventSinkV2; + const fs = yield* FileSystem.FileSystem; + const git = yield* GitVcsDriver.GitVcsDriver; + const worktreePath = `${workspace}/valid-worktree`; + const unavailable = `${workspace}/unavailable-worktree`; + const outside = `${workspace}/not-a-worktree`; + yield* fs.makeDirectory(outside); + for (const args of [ + ["worktree", "add", "--detach", worktreePath], + ["worktree", "add", "--detach", unavailable], + ["worktree", "lock", unavailable], + ]) { + yield* git.execute({ operation: "test.setupWorktrees", cwd: workspace, args }); + } + yield* fs.remove(unavailable, { recursive: true }); + const before = yield* sink.latestSequence(); + const invalid = [ + { ...request, messages: [...history, { ...history[0]!, createdAt: "not a date" }] }, + { ...request, projectId: ProjectId.make("project:missing") }, + { ...request, worktreePath: "/not-a-worktree-of-this-project" }, + { ...request, worktreePath: outside }, + ]; + const outcomes = yield* Effect.forEach(invalid, (input, index) => + imports + .importThread({ ...input, source: { ...source, handoffId: `invalid-${index}` } }) + .pipe( + Effect.exit, + Effect.map((exit) => ({ + failed: Exit.isFailure(exit), + defect: Exit.isFailure(exit) && Cause.hasDies(exit.cause), + })), + ), + ); + assert.deepEqual( + { outcomes, sequence: yield* sink.latestSequence() }, + { + outcomes: invalid.map(() => ({ failed: true, defect: false })), + sequence: before, + }, + ); + const accepted = yield* imports + .importThread({ ...request, worktreePath }) + .pipe(Effect.exit); + assert.isTrue(Exit.isSuccess(accepted)); + if (Exit.isSuccess(accepted)) { + assert.isTrue(accepted.value.created); + const threads = yield* ThreadManagement.ThreadManagementService; + const projection = yield* threads.getThreadProjection(accepted.value.threadId); + assert.equal(projection.thread.worktreePath, worktreePath); + assert.lengthOf(projection.messages, history.length); + } + }), + ), +); + +it.live("recovers a concurrent write without claiming its creation or another project", () => + Effect.gen(function* () { + const recover = (winnerProjectId: ProjectId) => + withImport( + () => + Effect.gen(function* () { + const imports = yield* ThreadImportService.ThreadImportService; + const sink = yield* EventSink.EventSinkV2; + const threads = yield* ThreadManagement.ThreadManagementService; + const result = yield* imports.importThread(request).pipe(Effect.exit); + const snapshot = yield* threads.getShellSnapshot(); + assert.lengthOf(snapshot.threads, 1); + assert.equal(snapshot.threads[0]!.projectId, winnerProjectId); + const projection = yield* threads.getThreadProjection(snapshot.threads[0]!.id); + assert.lengthOf(projection.messages, history.length); + assert.isEmpty(projection.runs); + assert.equal(yield* sink.latestSequence(), 5); + return { + accepted: Exit.isSuccess(result), + created: Exit.isSuccess(result) ? result.value.created : null, + }; + }), + winnerProjectId, + ); + assert.deepEqual( + { + sameProject: yield* recover(request.projectId), + differentProject: yield* recover(otherProjectId), + }, + { + sameProject: { accepted: true, created: false }, + differentProject: { accepted: false, created: null }, + }, + ); + }), +); + +it.live( + "replays committed history without starting work after the first caller is interrupted", + () => + Effect.gen(function* () { + const written = yield* Deferred.make(); + yield* withImport( + (captured) => + Effect.gen(function* () { + const imports = yield* ThreadImportService.ThreadImportService; + const threads = yield* ThreadManagement.ThreadManagementService; + const first = yield* imports + .importThread({ + ...request, + continuationPrompt: continuation, + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(written); + yield* Fiber.interrupt(first); + const replayed = yield* imports.importThread({ + ...request, + continuationPrompt: "A changed request must not start work.", + runtimeMode: "approval-required", + interactionMode: "plan", + }); + assert.isFalse(replayed.created); + assert.isNull(replayed.runId); + const projection = yield* threads.getThreadProjection(replayed.threadId); + assert.lengthOf(projection.messages, history.length); + assert.isEmpty(projection.runs); + assert.equal(projection.thread.runtimeMode, request.runtimeMode); + assert.isEmpty(yield* Ref.get(captured)); + }), + undefined, + Deferred.succeed(written, undefined).pipe(Effect.andThen(Effect.never)), + ); + }), +); diff --git a/apps/server/src/orchestration-v2/ThreadImportService.ts b/apps/server/src/orchestration-v2/ThreadImportService.ts new file mode 100644 index 0000000000..f8ca3001af --- /dev/null +++ b/apps/server/src/orchestration-v2/ThreadImportService.ts @@ -0,0 +1,433 @@ +import { + CommandId, + EventId, + MessageId, + type OrchestrationV2AppThread, + type OrchestrationV2DomainEvent, + type OrchestrationV2ThreadShell, + OrchestrationV2LinkOrigin, + OrchestratorMcpImportedMessage, + OrchestratorMcpThreadImportInput, + type OrchestratorMcpThreadImportResult, + ProviderInteractionMode, + ProjectId, + RuntimeMode, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as KeyedLock from "@t3tools/shared/KeyedLock"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Hex from "effect/encoding/Hex"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +import * as ProjectService from "../project/ProjectService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as EventSink from "./EventSink.ts"; +import * as ThreadManagement from "./ThreadManagementService.ts"; + +const IMPORT_EVENT_PREFIX = "thread-import:v1"; + +const ImportInput = Schema.Struct({ + ...OrchestratorMcpThreadImportInput.fields, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + linkOrigin: Schema.optional(OrchestrationV2LinkOrigin), + messages: Schema.Array( + Schema.Struct({ + ...OrchestratorMcpImportedMessage.fields, + createdAt: Schema.DateTimeUtcFromString, + }), + ).check(Schema.isMaxLength(2_000)), +}); +const decodeImportInput = Schema.decodeUnknownEffect(ImportInput); + +export class ThreadImportValidationError extends Schema.TaggedError()( + "ThreadImportValidationError", + { + reason: Schema.Literals(["input", "project", "worktree"]), + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + switch (this.reason) { + case "input": + return "The conversation import request is invalid."; + case "project": + return "The project was not found."; + case "worktree": + return "worktreePath must be one of the project's git worktrees. t3_worktree_list shows them."; + } + } +} + +export class ThreadImportReadError extends Schema.TaggedError()( + "ThreadImportReadError", + { threadId: ThreadId, cause: Schema.Defect() }, +) { + override get message(): string { + return "The conversation import could not be read."; + } +} + +export class ThreadImportLinkMismatchError extends Schema.TaggedError()( + "ThreadImportLinkMismatchError", + { threadId: ThreadId }, +) { + override get message(): string { + return "This handoff was already imported by another link."; + } +} + +export class ThreadImportWriteError extends Schema.TaggedError()( + "ThreadImportWriteError", + { threadId: ThreadId, cause: Schema.Defect() }, +) { + override get message(): string { + return "The thread could not be imported."; + } +} + +export class ThreadImportProjectMismatchError extends Schema.TaggedError()( + "ThreadImportProjectMismatchError", + { threadId: ThreadId, existingProjectId: ProjectId }, +) { + override get message(): string { + return `This handoff was already imported into another project (${this.existingProjectId}).`; + } +} + +/** + * The thread was imported, but its continuation prompt could not start a run. + * `rejection` is the orchestrator's own reason when it refused the message. + */ +export class ThreadImportContinuationError extends Schema.TaggedError()( + "ThreadImportContinuationError", + { threadId: ThreadId, rejection: Schema.optional(Schema.String), cause: Schema.Defect() }, +) { + override get message(): string { + return this.rejection === undefined + ? "The thread was imported, but its first message could not start." + : `The thread was imported, but its first message could not start: ${this.rejection}`; + } +} + +export type ThreadImportError = + | ThreadImportValidationError + | ThreadImportReadError + | ThreadImportLinkMismatchError + | ThreadManagement.ThreadManagementProviderInstanceDisabledError + | ThreadImportWriteError + | ThreadImportProjectMismatchError + | ThreadImportContinuationError; + +export class ThreadImportService extends Context.Service< + ThreadImportService, + { + readonly importThread: ( + input: OrchestratorMcpThreadImportInput & { + readonly runtimeMode: RuntimeMode; + readonly interactionMode: ProviderInteractionMode; + readonly linkOrigin: OrchestrationV2LinkOrigin | undefined; + }, + ) => Effect.Effect; + } +>()("t3/orchestration-v2/ThreadImportService") {} + +const make = Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const threads = yield* ThreadManagement.ThreadManagementService; + const crypto = yield* Crypto.Crypto; + const projects = yield* ProjectService.ProjectService; + const git = yield* GitVcsDriver.GitVcsDriver; + const fs = yield* FileSystem.FileSystem; + // Import callers wait for the first continuation without holding the + // non-reentrant thread command lock used by dispatch. + const imports = yield* KeyedLock.make(); + + const importedThreadId = (source: OrchestratorMcpThreadImportInput["source"]) => + crypto + .digest( + "SHA-256", + new TextEncoder().encode( + JSON.stringify([source.environmentId, source.threadId, source.handoffId]), + ), + ) + .pipe( + Effect.map((digest) => ThreadId.make(`thread:import:${Hex.encode(digest)}`)), + Effect.orDie, + ); + + const historyEvents = ( + threadId: ThreadId, + messages: typeof ImportInput.Type.messages, + occurredAt: DateTime.Utc, + ): ReadonlyArray => + messages.flatMap((message, index) => { + const suffix = String(index).padStart(6, "0"); + const messageId = MessageId.make(`${threadId}:${suffix}`); + const turnItemId = TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${threadId}:${suffix}`); + const at = message.createdAt; + const common = { + id: turnItemId, + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + // Imported items are numbered before any run here by the position store. + ordinal: index + 1, + status: "completed" as const, + title: null, + startedAt: at, + completedAt: at, + updatedAt: at, + }; + return [ + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:message:${threadId}:${suffix}`), + type: "message.updated", + threadId, + occurredAt, + payload: { + createdBy: message.role === "user" ? "user" : "agent", + creationSource: "server", + id: messageId, + threadId, + runId: null, + nodeId: null, + role: message.role, + text: message.text, + attachments: [], + streaming: false, + createdAt: at, + updatedAt: at, + }, + }, + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${threadId}:${suffix}`), + type: "turn-item.updated", + threadId, + occurredAt, + payload: + message.role === "user" + ? { + ...common, + createdBy: "user", + creationSource: "server", + type: "user_message", + messageId, + inputIntent: "turn_start", + text: message.text, + attachments: [], + } + : { + ...common, + type: "assistant_message", + messageId, + text: message.text, + streaming: false, + }, + }, + ] satisfies ReadonlyArray; + }); + + const readThread = (threadId: ThreadId) => + threads + .getThreadShell(threadId) + .pipe(Effect.mapError((cause) => new ThreadImportReadError({ threadId, cause }))); + + const continuationMessageId = (threadId: ThreadId) => MessageId.make(`${threadId}:continue`); + const continuationCommandId = (threadId: ThreadId) => + CommandId.make(`command:thread-import:${threadId}:continue`); + + const continuationRunId = (threadId: ThreadId) => + threads + .getThreadRecords(threadId, ["messages"], { + messageIds: [continuationMessageId(threadId)], + }) + .pipe( + Effect.map(({ messages }) => messages[0]?.runId ?? null), + Effect.mapError((cause) => new ThreadImportReadError({ threadId, cause })), + ); + + const replay = Effect.fnUntraced(function* ( + existing: OrchestrationV2ThreadShell, + input: typeof ImportInput.Type, + ) { + const threadId = existing.id; + if (existing.projectId !== input.projectId) { + return yield* new ThreadImportProjectMismatchError({ + threadId, + existingProjectId: existing.projectId, + }); + } + if (existing.linkOrigin?.sessionId !== input.linkOrigin?.sessionId) { + return yield* new ThreadImportLinkMismatchError({ threadId }); + } + return { + threadId, + projectId: existing.projectId, + created: false, + runId: yield* continuationRunId(threadId), + }; + }); + + const importNew = Effect.fnUntraced(function* ( + threadId: ThreadId, + input: typeof ImportInput.Type, + ) { + const project = yield* projects + .getById(input.projectId) + .pipe(Effect.mapError((cause) => new ThreadImportReadError({ threadId, cause }))); + if (Option.isNone(project)) { + return yield* new ThreadImportValidationError({ reason: "project" }); + } + if (input.worktreePath !== undefined) { + const paths = yield* git + .listWorktreePaths(project.value.workspaceRoot) + .pipe(Effect.mapError((cause) => new ThreadImportReadError({ threadId, cause }))); + const destination = yield* fs + .realPath(input.worktreePath) + .pipe( + Effect.mapError( + (cause) => new ThreadImportValidationError({ reason: "worktree", cause }), + ), + ); + const realPaths = yield* Effect.forEach(paths, (path) => + fs.realPath(path).pipe(Effect.orElseSucceed(() => null)), + ); + if (!realPaths.includes(destination)) { + return yield* new ThreadImportValidationError({ reason: "worktree" }); + } + } + yield* threads.assertProviderInstanceEnabled({ + projectId: input.projectId, + instanceId: input.modelSelection.instanceId, + commandId: continuationCommandId(threadId), + }); + const now = yield* DateTime.now; + const thread: OrchestrationV2AppThread = { + createdBy: "agent", + creationSource: "mcp", + id: threadId, + projectId: input.projectId, + title: input.title, + providerInstanceId: input.modelSelection.instanceId, + modelSelection: input.modelSelection, + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + branch: input.branch ?? null, + worktreePath: input.worktreePath ?? null, + activeProviderThreadId: null, + historyOrigin: "v1_import", + ...(input.linkOrigin === undefined ? {} : { linkOrigin: input.linkOrigin }), + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + lastVisitedAt: null, + deletedAt: null, + }; + const raced = yield* sink + .write({ + events: [ + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${threadId}:created`), + type: "thread.created", + threadId, + providerInstanceId: input.modelSelection.instanceId, + occurredAt: now, + payload: thread, + }, + ...historyEvents(threadId, input.messages, now), + ], + }) + .pipe( + Effect.as(null), + Effect.catch((cause) => + readThread(threadId).pipe( + Effect.flatMap((raced) => + raced === null + ? Effect.fail(new ThreadImportWriteError({ threadId, cause })) + : Effect.succeed(raced), + ), + ), + ), + ); + if (raced !== null) return yield* replay(raced, input); + + // Only the creator may start work. After failure or interruption, a retry + // returns the stored import and cannot introduce a different prompt or model. + if (input.continuationPrompt !== undefined) { + yield* threads + .dispatch({ + type: "message.dispatch", + commandId: continuationCommandId(threadId), + threadId, + messageId: continuationMessageId(threadId), + text: input.continuationPrompt, + attachments: [], + modelSelection: input.modelSelection, + dispatchMode: { type: "start_immediately" }, + createdBy: "agent", + creationSource: "mcp", + }) + .pipe( + Effect.mapError( + (cause) => + new ThreadImportContinuationError({ + threadId, + ...((cause._tag === "OrchestratorDispatchError" || + cause._tag === "OrchestratorCommandRejectedError") && + typeof cause.cause === "string" && + cause.cause.length > 0 + ? { rejection: Array.from(cause.cause).slice(0, 1000).join("") } + : {}), + cause, + }), + ), + ); + } + return { + threadId, + projectId: input.projectId, + created: true, + runId: yield* continuationRunId(threadId), + }; + }); + + const importThread: ThreadImportService["Service"]["importThread"] = Effect.fn( + "ThreadImportService.importThread", + )(function* (request) { + const input = yield* decodeImportInput(request).pipe( + Effect.mapError((cause) => new ThreadImportValidationError({ reason: "input", cause })), + ); + const threadId = yield* importedThreadId(input.source); + return yield* imports.withLock( + threadId, + Effect.gen(function* () { + const existing = yield* readThread(threadId); + return existing === null + ? yield* importNew(threadId, input) + : yield* replay(existing, input); + }), + ); + }); + + return ThreadImportService.of({ importThread }); +}); + +export const layer = Layer.effect(ThreadImportService, make); diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts index dc29e62330..ef2c4310f2 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts @@ -3,14 +3,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { CommandId, MessageId, - ProviderDriverKind, ProviderInstanceId, - ProviderThreadId, - ProviderTurnId, ThreadId, - TurnItemId, - type OrchestrationV2ProviderSession, - type OrchestrationV2ProviderThread, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; @@ -20,10 +14,8 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Path from "effect/Path"; -import * as PubSub from "effect/PubSub"; import * as References from "effect/References"; import * as Ref from "effect/Ref"; -import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; import { runMigrations } from "../../persistence/Migrations.ts"; @@ -36,7 +28,6 @@ import Migration0046 from "../../persistence/Migrations/046_RepairAutomaticSettl import Migration0047 from "../../persistence/Migrations/047_ProjectionProjectIcon.ts"; import Migration0048 from "../../persistence/Migrations/048_ProjectionThreadBranchPullRequest.ts"; import Migration0049 from "../../persistence/Migrations/049_ProjectionThreadsActiveOrderKey.ts"; -import { CodexProviderCapabilitiesV2 } from "../Adapters/CodexAdapterV2.ts"; import * as EffectWorker from "../EffectWorker.ts"; import * as EventSink from "../EventSink.ts"; import * as EventStore from "../EventStore.ts"; @@ -47,7 +38,7 @@ import * as ProjectionStore from "../ProjectionStore.ts"; import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "../testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; -import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import { type CapturedTurn, makeCapturingCodexAdapter } from "../testkit/CapturingCodexAdapter.ts"; const PROJECT_ID = "project:cutover"; const ACTIVE_THREAD = "thread:cutover:active"; @@ -72,7 +63,6 @@ const LATEST_MARKER = "LATEST_IMPORT_MARKER"; const CONTINUATION_PROMPT = "Continue the migrated thread."; const CONTINUATION_RESPONSE = "codex continuation response"; -const driver = ProviderDriverKind.make("codex"); const instanceId = ProviderInstanceId.make("codex"); const modelSelectionJson = '{"instanceId":"codex","model":"gpt-5.4"}'; const codexModelSelection = { @@ -409,154 +399,6 @@ const seedV1Database = (fixturePath: string, workspace: string) => }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: fixturePath }))), ); -interface CapturedTurn { - readonly threadId: ThreadId; - readonly providerThreadId: ProviderThreadId; - readonly text: string; -} - -const unimplemented = (detail: string) => - Effect.fail(new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail })); - -const makeCodexAdapter = (capturedTurns: Ref.Ref>) => - ({ - instanceId, - driver, - getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), - planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), - openSession: (sessionInput) => - Effect.gen(function* () { - const events = yield* PubSub.unbounded(); - const now = yield* DateTime.now; - const providerSession: OrchestrationV2ProviderSession = { - id: sessionInput.providerSessionId, - driver, - providerInstanceId: instanceId, - status: "ready", - cwd: sessionInput.runtimePolicy.cwd ?? process.cwd(), - model: codexModelSelection.model, - capabilities: CodexProviderCapabilitiesV2, - createdAt: now, - updatedAt: now, - lastError: null, - }; - return { - instanceId, - driver, - providerSessionId: sessionInput.providerSessionId, - providerSession, - events: Stream.fromPubSub(events), - ensureThread: (threadInput) => - Effect.gen(function* () { - const createdAt = yield* DateTime.now; - const nativeThreadId = `${driver}:${threadInput.threadId}`; - return { - id: ProviderThreadId.make(`provider-thread:${nativeThreadId}`), - driver, - providerInstanceId: instanceId, - providerSessionId: sessionInput.providerSessionId, - appThreadId: threadInput.threadId, - ownerNodeId: null, - nativeThreadRef: { - driver, - nativeId: nativeThreadId, - strength: "strong", - }, - nativeConversationHeadRef: null, - status: "idle", - firstRunOrdinal: null, - lastRunOrdinal: null, - handoffIds: [], - forkedFrom: null, - createdAt, - updatedAt: createdAt, - } satisfies OrchestrationV2ProviderThread; - }), - resumeThread: ({ providerThread }) => Effect.succeed(providerThread), - startTurn: (turnInput) => - Effect.gen(function* () { - yield* Ref.update(capturedTurns, (turns) => [ - ...turns, - { - threadId: turnInput.threadId, - providerThreadId: turnInput.providerThread.id, - text: turnInput.message.text, - }, - ]); - const eventTime = yield* DateTime.now; - const providerTurnId = ProviderTurnId.make( - `provider-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, - ); - yield* PubSub.publishAll(events, [ - { - type: "provider_turn.updated", - driver, - providerTurn: { - id: providerTurnId, - providerThreadId: turnInput.providerThread.id, - nodeId: turnInput.rootNodeId, - runAttemptId: turnInput.attemptId, - nativeTurnRef: { - driver, - nativeId: `native-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, - strength: "strong", - }, - ordinal: turnInput.runOrdinal, - status: "completed", - startedAt: eventTime, - completedAt: eventTime, - }, - }, - { - type: "turn_item.updated", - driver, - turnItem: { - id: TurnItemId.make( - `turn-item:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, - ), - threadId: turnInput.threadId, - runId: turnInput.runId, - nodeId: turnInput.rootNodeId, - providerThreadId: turnInput.providerThread.id, - providerTurnId, - nativeItemRef: null, - parentItemId: null, - ordinal: turnInput.runOrdinal * 100 + 1, - status: "completed", - title: null, - startedAt: eventTime, - completedAt: eventTime, - updatedAt: eventTime, - type: "assistant_message", - messageId: MessageId.make( - `message:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, - ), - text: CONTINUATION_RESPONSE, - streaming: false, - }, - }, - { - type: "turn.terminal", - driver, - providerThreadId: turnInput.providerThread.id, - providerTurnId, - runOrdinal: turnInput.runOrdinal, - status: "completed", - failure: null, - threadDisposition: "reusable", - }, - ] satisfies ReadonlyArray); - }), - steerTurn: () => Effect.void, - interruptTurn: () => Effect.void, - respondToRuntimeRequest: () => Effect.void, - readThreadSnapshot: () => unimplemented("readThreadSnapshot unused in cutover test"), - rollbackThread: () => unimplemented("rollbackThread unused in cutover test"), - forkThread: () => unimplemented("forkThread unused in cutover test"), - }; - }), - }) satisfies ProviderAdapter.ProviderAdapterV2["Service"]; - const waitForIdle = Effect.fn("LegacyV1Cutover.waitForIdle")(function* (threadId: ThreadId) { const orchestrator = yield* Orchestrator.OrchestratorV2; for (let attempt = 0; attempt < 1_000; attempt += 1) { @@ -611,7 +453,12 @@ const layerBoot = (input: { }, }, }, - ProviderAdapterRegistry.layerSingle(makeCodexAdapter(input.capturedTurns)), + ProviderAdapterRegistry.layerSingle( + makeCapturingCodexAdapter(input.capturedTurns, { + response: CONTINUATION_RESPONSE, + modelSelection: codexModelSelection, + }), + ), { databaseLayer: layerDatabase }, ); return Layer.mergeAll( diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index b3cfc42701..39963a13cf 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -47,6 +47,7 @@ import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; import * as RuntimeRequestService from "./RuntimeRequestService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; +import * as ThreadImportService from "./ThreadImportService.ts"; import * as ThreadLaunchService from "./ThreadLaunchService.ts"; import * as ThreadLifecycleService from "./ThreadLifecycleService.ts"; import * as ThreadForkService from "./ThreadForkService.ts"; @@ -276,6 +277,11 @@ const layerThreadLaunchProvided = ThreadLaunchService.layer.pipe( ), ), ); +const layerThreadImportProvided = ThreadImportService.layer.pipe( + Layer.provide( + Layer.mergeAll(layerEventSinkProvided, layerThreadManagementProvided, layerProjectService), + ), +); const layerThreadLifecycleProvided = ThreadLifecycleService.layer.pipe( Layer.provide(layerThreadManagementProvided), ); @@ -363,6 +369,7 @@ export const layerProduction = Layer.mergeAll( layerProjectService, layerManagedProjectFoldersProvided, layerThreadLaunchProvided, + layerThreadImportProvided, layerThreadLifecycleProvided, layerScheduledTaskProvided, layerSecretRequestsProvided, diff --git a/apps/server/src/orchestration-v2/testkit/CapturingCodexAdapter.ts b/apps/server/src/orchestration-v2/testkit/CapturingCodexAdapter.ts new file mode 100644 index 0000000000..645c2e7c3f --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/CapturingCodexAdapter.ts @@ -0,0 +1,179 @@ +import { + MessageId, + type ModelSelection, + type OrchestrationV2ProviderSession, + type OrchestrationV2ProviderThread, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + type ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; + +import * as CodexAdapterV2 from "../Adapters/CodexAdapterV2.ts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; + +/** A turn the adapter was asked to run: its thread and the text the provider got. */ +export interface CapturedTurn { + readonly threadId: ThreadId; + readonly providerThreadId: ProviderThreadId; + readonly text: string; +} + +const driver = ProviderDriverKind.make("codex"); +const instanceId = ProviderInstanceId.make("codex"); + +const unimplemented = (detail: string) => + Effect.fail(new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail })); + +/** + * A Codex-shaped adapter that records every turn's input text and completes + * each turn at once with `response`. For tests of what a provider is sent. + */ +export const makeCapturingCodexAdapter = ( + capturedTurns: Ref.Ref>, + options: { readonly response: string; readonly modelSelection: ModelSelection }, +) => + ({ + instanceId, + driver, + getCapabilities: () => Effect.succeed(CodexAdapterV2.CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (sessionInput) => + Effect.gen(function* () { + const events = yield* PubSub.unbounded(); + const now = yield* DateTime.now; + const providerSession: OrchestrationV2ProviderSession = { + id: sessionInput.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd: sessionInput.runtimePolicy.cwd ?? process.cwd(), + model: options.modelSelection.model, + capabilities: CodexAdapterV2.CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }; + return { + instanceId, + driver, + providerSessionId: sessionInput.providerSessionId, + providerSession, + events: Stream.fromPubSub(events), + ensureThread: (threadInput) => + Effect.gen(function* () { + const createdAt = yield* DateTime.now; + const nativeThreadId = `${driver}:${threadInput.threadId}`; + return { + id: ProviderThreadId.make(`provider-thread:${nativeThreadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: sessionInput.providerSessionId, + appThreadId: threadInput.threadId, + ownerNodeId: null, + nativeThreadRef: { + driver, + nativeId: nativeThreadId, + strength: "strong", + }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt, + updatedAt: createdAt, + } satisfies OrchestrationV2ProviderThread; + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turnInput) => + Effect.gen(function* () { + yield* Ref.update(capturedTurns, (turns) => [ + ...turns, + { + threadId: turnInput.threadId, + providerThreadId: turnInput.providerThread.id, + text: turnInput.message.text, + }, + ]); + const eventTime = yield* DateTime.now; + const providerTurnId = ProviderTurnId.make( + `provider-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, + ); + yield* PubSub.publishAll(events, [ + { + type: "provider_turn.updated", + driver, + providerTurn: { + id: providerTurnId, + providerThreadId: turnInput.providerThread.id, + nodeId: turnInput.rootNodeId, + runAttemptId: turnInput.attemptId, + nativeTurnRef: { + driver, + nativeId: `native-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, + strength: "strong", + }, + ordinal: turnInput.runOrdinal, + status: "completed", + startedAt: eventTime, + completedAt: eventTime, + }, + }, + { + type: "turn_item.updated", + driver, + turnItem: { + id: TurnItemId.make( + `turn-item:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, + ), + threadId: turnInput.threadId, + runId: turnInput.runId, + nodeId: turnInput.rootNodeId, + providerThreadId: turnInput.providerThread.id, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: turnInput.runOrdinal * 100 + 1, + status: "completed", + title: null, + startedAt: eventTime, + completedAt: eventTime, + updatedAt: eventTime, + type: "assistant_message", + messageId: MessageId.make( + `message:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, + ), + text: options.response, + streaming: false, + }, + }, + { + type: "turn.terminal", + driver, + providerThreadId: turnInput.providerThread.id, + providerTurnId, + runOrdinal: turnInput.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }, + ] satisfies ReadonlyArray); + }), + steerTurn: () => Effect.void, + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.void, + readThreadSnapshot: () => unimplemented("readThreadSnapshot unused in this test"), + rollbackThread: () => unimplemented("rollbackThread unused in this test"), + forkThread: () => unimplemented("forkThread unused in this test"), + }; + }), + }) satisfies ProviderAdapter.ProviderAdapterV2["Service"]; diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index 296567d8c1..7ee8c356a1 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -38,6 +38,7 @@ import { liveThreadShell } from "../mcp/McpToolAccess.testkit.ts"; import * as PeerForwarding from "./PeerForwarding.ts"; import * as PeerLinks from "./PeerLinks.ts"; import * as PeerLinkRequests from "./PeerLinkRequests.ts"; +import * as ThreadImportService from "../orchestration-v2/ThreadImportService.ts"; import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts"; const laptop = descriptorOf("environment-laptop", "Laptop"); @@ -223,6 +224,7 @@ const serveBox = (seen: Ref.Ref, linkSession: Ref.Ref) => Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadImportService.ThreadImportService)({})), ), ); @@ -290,6 +292,8 @@ const makeLaptop = Effect.gen(function* () { } as never), }), ), + Layer.provide(ServerSettings.layerTest()), + // Services the toolkits declare that these calls never reach. Layer.provide( Layer.mergeAll( Layer.mock(ProviderRegistry.ProviderRegistry)({}), @@ -297,9 +301,9 @@ const makeLaptop = Effect.gen(function* () { Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), Layer.mock(ProjectService.ProjectService)({}), Layer.mock(SecretRequests.SecretRequests)({}), - ServerSettings.layerTest(), Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), + Layer.mock(ThreadImportService.ThreadImportService)({}), ), ), Layer.fresh, diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 709e815083..39e99161ef 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -62,6 +62,7 @@ import * as PeerLinks from "./PeerLinks.ts"; import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts"; import * as RemoteDelegation from "./RemoteDelegation.ts"; import * as PeerLinkRequests from "./PeerLinkRequests.ts"; +import * as ThreadImportService from "../orchestration-v2/ThreadImportService.ts"; // The laptop's agent delegates a task to the box through a link. The box is // its real /mcp behind real OAuth, with one thread the test finishes; the @@ -465,6 +466,7 @@ const boxToolkitLayer = (thread: BoxThread) => { Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadImportService.ThreadImportService)({})), Layer.provide( Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/p" }), ), diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index b51620f4cf..0c1ed45cb2 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -161,6 +161,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ (`linkOrigin`). Another environment links to this one only when it is set, since without it that work could change the user's own threads. */ linkFence: Schema.optionalKey(Schema.Boolean), + /** Accepts `t3_thread_import`, for threads moved here from a linked environment. */ + threadImport: Schema.optionalKey(Schema.Boolean), }); export type ExecutionEnvironmentCapabilities = typeof ExecutionEnvironmentCapabilities.Type; diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 97edccb547..df253eebbe 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -38,6 +38,7 @@ import { ProviderOptionSelection, ProviderOptionSelectionValue, } from "./model.ts"; +import { ModelSelection } from "./modelSelection.ts"; import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; const OrchestratorMcpPrompt = TrimmedNonEmptyString.check(Schema.isMaxLength(120_000)).annotate({ @@ -542,6 +543,44 @@ export const OrchestratorMcpProviderCapability = Schema.Struct({ }); export type OrchestratorMcpProviderCapability = typeof OrchestratorMcpProviderCapability.Type; +/** One message of a conversation moved here from another environment. */ +export const OrchestratorMcpImportedMessage = Schema.Struct({ + role: Schema.Literals(["user", "assistant"]), + text: Schema.String.check(Schema.isMaxLength(200_000)), + createdAt: IsoDateTime, +}); +export type OrchestratorMcpImportedMessage = typeof OrchestratorMcpImportedMessage.Type; + +export const OrchestratorMcpThreadImportInput = Schema.Struct({ + /** Where the conversation comes from; with `handoffId`, this names the import. */ + source: Schema.Struct({ + environmentId: EnvironmentId, + threadId: ThreadId, + handoffId: TrimmedNonEmptyString.check(Schema.isMaxLength(128)), + }), + projectId: ProjectId, + title: TrimmedNonEmptyString.check(Schema.isMaxLength(512)), + modelSelection: ModelSelection, + runtimeMode: Schema.optional(RuntimeMode), + interactionMode: Schema.optional(ProviderInteractionMode), + /** The checkout the thread works in here, already prepared. Omit for the project root. */ + worktreePath: Schema.optional(TrimmedNonEmptyString), + branch: Schema.optional(TrimmedNonEmptyString), + messages: Schema.Array(OrchestratorMcpImportedMessage).check(Schema.isMaxLength(2_000)), + /** Sent as the thread's first message here, so its agent picks up where it left off. */ + continuationPrompt: Schema.optional(OrchestratorMcpPrompt), +}); +export type OrchestratorMcpThreadImportInput = typeof OrchestratorMcpThreadImportInput.Type; + +export const OrchestratorMcpThreadImportResult = Schema.Struct({ + threadId: ThreadId, + projectId: ProjectId, + /** False when this import already happened; the thread is returned as it is. */ + created: Schema.Boolean, + runId: Schema.NullOr(RunId), +}); +export type OrchestratorMcpThreadImportResult = typeof OrchestratorMcpThreadImportResult.Type; + export const OrchestratorMcpCapabilitiesInput = Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget, }); diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 1cbc6820ca..2224f1d20a 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -329,6 +329,10 @@ const T3_MCP_TOOLS: Readonly> = { "environment-update", ), t3_thread_launch: tool(["Launch", "Launching", "Launched", "a project thread"], "thread-create"), + t3_thread_import: tool( + ["Import", "Importing", "Imported", "a thread from another environment"], + "thread-create", + ), t3_project_list: tool(["List", "Listing", "Listed", "projects"], "project-list"), t3_project_read: tool(["Read", "Reading", "Read", "a project"], "project-read"), t3_project_create: tool(["Register", "Registering", "Registered", "a project"], "project-create"),