# Files changed · PR \#16751 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/16751/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [feat\(server\): hand a thread off to a linked environment](https://git.cafe/pingdotgg/t3code/pull/16751.md) Version: 10 Head commit: be89710151b212a24dc62721e0b3c46e67441475 Base commit: 75b2c90834b04ac7254bb95a713fd1c39a047384 Comparison base: 75b2c90834b04ac7254bb95a713fd1c39a047384 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/auth/RpcAuthorization.ts Status: modified; +4; −0 ```diff @@ -84,6 +84,10 @@ [WS_METHODS.serverGetDatabaseStorage]: AuthOrchestrationReadScope, [WS_METHODS.serverGetSettings]: AuthOrchestrationReadScope, [WS_METHODS.peerLinksList]: AuthAccessReadScope, + // Moving a thread drives the linked environment through this one's link. + [WS_METHODS.threadHandoffOptions]: AuthOrchestrationReadScope, + [WS_METHODS.threadHandoffStart]: AuthOrchestrationOperateScope, + [WS_METHODS.threadHandoffCancel]: AuthOrchestrationOperateScope, [WS_METHODS.serverUpdateSettings]: AuthSettingsWriteScope, [WS_METHODS.serverSearchAcpRegistry]: AuthOrchestrationReadScope, [WS_METHODS.serverPrepareAcpRegistryAgent]: AuthProvidersManageScope, ``` ## apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts Status: modified; +3; −0 ```diff @@ -89,6 +89,7 @@ import * as PeerForwarding from "../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../peer/PeerLinks.ts"; +import * as ThreadHandoff from "../peer/handoff/ThreadHandoff.ts"; import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; @@ -742,6 +743,7 @@ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), Layer.provide( Layer.mock(ProjectService.ProjectService)({ @@ -4438,6 +4440,7 @@ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provideMerge( SecretRequests.layer.pipe( ``` ## apps/server/src/mcp/linkOrigin.test.ts Status: modified; +16; −7 ```diff @@ -46,6 +46,8 @@ import * as PeerLinkRequests from "../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../peer/PeerLinks.ts"; import * as ThreadImportService from "../orchestration-v2/ThreadImportService.ts"; +import * as ThreadHandoff from "../peer/handoff/ThreadHandoff.ts"; +import * as HandoffImport from "../peer/handoff/HandoffImport.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 @@ -156,13 +158,20 @@ }), }), ), + // Services the toolkits declare that these calls never reach. + Layer.provide( + Layer.mergeAll( + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(PeerForwarding.PeerForwarding)({}), + Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), + Layer.mock(PeerLinks.PeerLinks)({}), + Layer.mock(ThreadImportService.ThreadImportService)({}), + Layer.mock(ThreadHandoff.ThreadHandoff)({}), + Layer.mock(HandoffImport.HandoffImport)({}), - Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), + Layer.mock(ThreadSearch.ThreadSearch)({}), - Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + ), + ), - 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( Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/p" }), ), ``` ## apps/server/src/mcp/toolkits/core.test.ts Status: modified; +6; −0 ```diff @@ -64,6 +64,7 @@ import * as PeerForwarding from "../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../peer/PeerLinks.ts"; +import * as ThreadHandoff from "../../peer/handoff/ThreadHandoff.ts"; import { ThreadToolkit } from "./thread/tools.ts"; import { WorktreeToolkit } from "./worktree/tools.ts"; import { DeviceToolkit } from "./device/tools.ts"; @@ -97,6 +98,7 @@ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), ); it("publishes unique tool names with reference-free object-root inputs", () => { @@ -728,6 +730,7 @@ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), @@ -787,6 +790,7 @@ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), ), ), @@ -876,6 +880,7 @@ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), ), ), ), @@ -969,6 +974,7 @@ Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), Layer.provide(Layer.mock(ProjectService.ProjectService)({})), Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), ``` ## apps/server/src/mcp/toolkits/orchestrator/handlers.ts Status: modified; +45; −0 ```diff @@ -5,6 +5,7 @@ import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../../peer/PeerLinks.ts"; +import * as ThreadHandoff from "../../../peer/handoff/ThreadHandoff.ts"; import { readCaller } from "../../threadAccess.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as McpToolAccess from "../../McpToolAccess.ts"; @@ -145,6 +146,50 @@ return yield* service.waitForThread(scope, input); }), ), + t3_thread_handoff: McpToolAccess.writesThreads( + (input) => [input.threadId], + (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + const handoff = yield* ThreadHandoff.ThreadHandoff; + const state = yield* handoff + .start({ + threadId: input.threadId, + callerThreadId: scope.thread?.threadId, + environmentId: input.environmentId, + projectId: input.projectId, + continuationPrompt: input.continuationPrompt, + }) + .pipe( + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ + code: + error.reason === "not_found" + ? "thread_not_found" + : error.reason === "refused" + ? "invalid_request" + : "orchestration_error", + message: error.message, + }), + ), + ); + return { + state: state.state, + environmentId: state.environmentId, + label: state.label, + threadId: state.state === "departed" ? state.threadId : null, + message: + state.state === "pending" + ? `This thread moves to ${state.label} when this turn ends. End the turn now; a message to this thread before then cancels the move.` + : state.state === "departed" + ? `The thread moved to ${state.label}.` + : state.state === "failed" + ? `The move failed: ${state.lastError}` + : `The thread is moving to ${state.label}.`, + }; + }), + ), t3_environment_links: McpToolAccess.reads(() => Effect.gen(function* () { const scope = yield* McpInvocationContext.McpInvocationContext; ``` ## apps/server/src/mcp/toolkits/orchestrator/tools.ts Status: modified; +17; −0 ```diff @@ -7,6 +7,8 @@ OrchestratorMcpEnvironmentLinksResult, OrchestratorMcpEnvironmentUnlinkInput, OrchestratorMcpEnvironmentUnlinkResult, + OrchestratorMcpThreadHandoffInput, + OrchestratorMcpThreadHandoffResult, OrchestratorMcpCreateThreadsInput, OrchestratorMcpCreateThreadsResult, OrchestratorMcpDelegateTaskInput, @@ -45,6 +47,7 @@ import * as PeerForwarding from "../../../peer/PeerForwarding.ts"; import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../../peer/PeerLinks.ts"; +import * as ThreadHandoff from "../../../peer/handoff/ThreadHandoff.ts"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import * as OrchestratorMcpService from "../../OrchestratorMcpService.ts"; import * as ThreadMetadataMcpService from "../../ThreadMetadataMcpService.ts"; @@ -296,6 +299,19 @@ .annotate(Tool.Title, "Interrupt a T3 thread") .annotate(Tool.Destructive, true); +const ThreadHandoffTool = Tool.make("t3_thread_handoff", { + description: + "Move a thread to a linked environment (t3_environment_links), such as the user's VPS, with its conversation and its git work: unpushed commits, uncommitted edits and new files. It continues there in a new worktree of the project with the same repository, and the copy here becomes read-only. Omit threadId to move the calling thread: the move then happens as soon as this turn ends, and continuationPrompt becomes its first message there, so end the turn after saying what you will do there. A message to the thread before then cancels the move. Another thread must be idle to move.", + parameters: OrchestratorMcpThreadHandoffInput, + success: OrchestratorMcpThreadHandoffResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: [...dependencies, ThreadHandoff.ThreadHandoff], +}) + .annotate(Tool.Title, "Move a thread to a linked environment") + .annotate(Tool.Destructive, true) + .annotate(Tool.OpenWorld, true); + const EnvironmentLinksTool = Tool.make("t3_environment_links", { description: "List the other T3 Code environments this one is linked to, such as the user's other machines, and whether each answers now. A machine the user names that is not listed is not linked yet: call t3_environment_link for it rather than looking elsewhere. Pass one's environmentId to any tool that takes environmentId, such as t3_thread_launch, t3_thread_send, t3_pending_request_respond, t3_queue_list or t3_project_list, to act there; delegate_task takes it as target.environmentId. A linked environment lets this one's agents do at most its access there, and never more than the calling agent may do here. An expired link has to be linked again from a new pairing code.", @@ -356,4 +372,5 @@ EnvironmentLinksTool, EnvironmentLinkTool, EnvironmentUnlinkTool, + ThreadHandoffTool, ); ``` ## apps/server/src/mcp/toolkits/project/handlers.ts Status: modified; +33; −10 ```diff @@ -10,6 +10,7 @@ import * as Option from "effect/Option"; import * as ThreadLaunch from "../../../orchestration-v2/ThreadLaunchService.ts"; import * as ThreadImportService from "../../../orchestration-v2/ThreadImportService.ts"; +import * as HandoffImport from "../../../peer/handoff/HandoffImport.ts"; import * as ThreadMessageIntake from "../../../orchestration-v2/ThreadMessageIntake.ts"; import * as Claims from "../../../orchestration-v2/AttachmentClaims.ts"; import * as Project from "../../../project/ProjectService.ts"; @@ -217,21 +218,43 @@ code: "invalid_request", message: "The project was not found.", }); + if (input.worktreePath !== undefined && input.bundle !== undefined) + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "A bundle lands in a new worktree; omit worktreePath.", + }); if (input.worktreePath !== undefined) yield* assertProjectWorktree(project.workspaceRoot, input.worktreePath); const imports = yield* ThreadImportService.ThreadImportService; + const handoffs = yield* HandoffImport.HandoffImport; + const bundle = input.bundle; + const workspace = + bundle === undefined + ? undefined + : handoffs.applyBundle({ + repoRoot: project.workspaceRoot, + handoffId: input.source.handoffId, + bundle, + }); return yield* imports + .importThread({ + ...input, + runtimeMode, + interactionMode, + linkOrigin, + ...(workspace === undefined ? {} : { workspace }), - .importThread({ ...input, runtimeMode, interactionMode, linkOrigin }) + }) .pipe( + Effect.mapError((error) => + error._tag === "OrchestratorMcpFailure" + ? error + : new OrchestratorMcpFailure({ + code: + error._tag === "ThreadImportProjectMismatchError" + ? "invalid_request" + : "orchestration_error", + message: error.message, - Effect.mapError( - (error) => - new OrchestratorMcpFailure({ - code: - error._tag === "ThreadImportProjectMismatchError" - ? "invalid_request" + }), - : "orchestration_error", - message: error.message, - }), ), ); }), ``` ## apps/server/src/mcp/toolkits/project/tools.ts Status: modified; +2; −0 ```diff @@ -26,6 +26,7 @@ import * as ServerConfig from "../../../config.ts"; import * as ServerSettings from "../../../serverSettings.ts"; import * as ThreadImportService from "../../../orchestration-v2/ThreadImportService.ts"; +import * as HandoffImport from "../../../peer/handoff/HandoffImport.ts"; import * as ThreadLaunchService from "../../../orchestration-v2/ThreadLaunchService.ts"; import * as Crypto from "effect/Crypto"; import * as Schema from "effect/Schema"; @@ -186,6 +187,7 @@ dependencies: [ ...shared.dependencies, ThreadImportService.ThreadImportService, + HandoffImport.HandoffImport, GitVcsDriver.GitVcsDriver, FileSystem.FileSystem, ], ``` ## apps/server/src/mcp/toolkits/worktree/registration.test.ts Status: modified; +4; −0 ```diff @@ -40,6 +40,8 @@ import * as PeerLinkRequests from "../../../peer/PeerLinkRequests.ts"; import * as PeerLinks from "../../../peer/PeerLinks.ts"; import * as ThreadImportService from "../../../orchestration-v2/ThreadImportService.ts"; +import * as ThreadHandoff from "../../../peer/handoff/ThreadHandoff.ts"; +import * as HandoffImport from "../../../peer/handoff/HandoffImport.ts"; const layerStubServices = Layer.mergeAll( Layer.mock(Orchestrator.OrchestratorV2)({}), @@ -53,6 +55,8 @@ Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), Layer.mock(PeerLinks.PeerLinks)({}), Layer.mock(ThreadImportService.ThreadImportService)({}), + Layer.mock(ThreadHandoff.ThreadHandoff)({}), + Layer.mock(HandoffImport.HandoffImport)({}), Layer.mock(SecretRequests.SecretRequests)({}), Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(ProjectService.ProjectService)({}), ``` ## apps/server/src/observability/RpcInstrumentation.ts Status: modified; +3; −0 ```diff @@ -63,6 +63,9 @@ [WS_METHODS.peerLinksLink]: "peerLinks", [WS_METHODS.peerLinksUnlink]: "peerLinks", [WS_METHODS.peerLinksAnswerRequest]: "peerLinks", + [WS_METHODS.threadHandoffOptions]: "threadHandoff", + [WS_METHODS.threadHandoffStart]: "threadHandoff", + [WS_METHODS.threadHandoffCancel]: "threadHandoff", [WS_METHODS.serverUpdateSettings]: "server", [WS_METHODS.serverSearchAcpRegistry]: "server", [WS_METHODS.serverPrepareAcpRegistryAgent]: "server", ``` ## apps/server/src/orchestration-v2/Orchestrator.ts Status: modified; +88; −0 ```diff @@ -223,6 +223,23 @@ } } +/** The thread is moving, or has moved, to a linked environment and takes no new turns here. */ +export class OrchestratorThreadMovedError extends Schema.TaggedError()( + "OrchestratorThreadMovedError", + { + commandId: CommandId, + threadId: ThreadId, + label: Schema.String, + state: Schema.Literals(["departing", "departed"]), + }, +) { + override get message(): string { + return this.state === "departed" + ? `This thread moved to ${this.label}. Continue it there.` + : `This thread is moving to ${this.label}.`; + } +} + export class OrchestratorThreadAboveModeLimitError extends Schema.TaggedError()( "OrchestratorThreadAboveModeLimitError", { @@ -288,6 +305,7 @@ OrchestratorCommandPreviouslyRejectedError, OrchestratorCommandIdConflictError, OrchestratorSubagentThreadReadOnlyError, + OrchestratorThreadMovedError, OrchestratorThreadAboveModeLimitError, ]); export type OrchestratorV2Error = typeof OrchestratorV2Error.Type; @@ -488,6 +506,7 @@ case "checkpoint.rollback.fail": case "thread.background-work.settle": case "thread.stop": + case "thread.handoff.update": case "provider.switch": return command.threadId; case "delegated_task.request": @@ -1382,6 +1401,8 @@ if ( projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || + // A thread waiting to move, moving or moved starts nothing more here. + (projection.thread.handoff !== undefined && projection.thread.handoff.state !== "failed") || projection.runs.some(isBlockingRun) || projection.runs.some((run) => run.status === "queued" && run.queueHeld === true) ) { @@ -7311,6 +7332,61 @@ }); }); + /** Moves a thread through a handoff to a linked environment; see the command. */ + const dispatchThreadHandoffUpdate = Effect.fn("orchestrationV2.dispatch.threadHandoffUpdate")( + function* ( + command: Extract, + events: Ref.Ref>, + ) { + const projection = yield* projectionStore + .getThreadRecords(command.threadId, ["runs"]) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), + ), + ); + const thread = projection.thread; + const reject = (cause: string) => + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause, + }); + if (thread.deletedAt !== null) return yield* reject(`Thread ${thread.id} is deleted.`); + const current = thread.handoff?.state ?? null; + if (current !== command.expected) { + return yield* reject( + current === null + ? "This thread is not being moved." + : `This thread's move is ${current}, not ${command.expected ?? "absent"}.`, + ); + } + const next = command.handoff?.state ?? null; + if ( + next === "departing" && + projection.runs.some((run) => isBlockingRun(run) || run.status === "queued") + ) { + return yield* reject("A thread moves only while no turn runs or waits on it."); + } + const now = yield* DateTime.now; + const { handoff: _previous, ...rest } = thread; + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + ...rest, + ...(command.handoff === null ? {} : { handoff: command.handoff }), + updatedAt: now, + }, + }); + }, + ); + /** * Records a task that runs in a linked environment on its parent: the * task, its node and its turn item, as a local delegation does, but no child @@ -11840,6 +11916,15 @@ threadId: command.threadId, }); } + // A thread moving away, or gone, takes no new turns here. + if (thread.handoff?.state === "departing" || thread.handoff?.state === "departed") { + return yield* new OrchestratorThreadMovedError({ + commandId: command.commandId, + threadId: command.threadId, + label: thread.handoff.label, + state: thread.handoff.state, + }); + } yield* dispatchMessage(command, events, effects); yield* recordSubagentFollowup(command, events).pipe(mapDispatchError(command)); break; @@ -11977,6 +12062,9 @@ case "delegated_task.remote.request": yield* dispatchRemoteDelegatedTaskRequest(command, events); break; + case "thread.handoff.update": + yield* dispatchThreadHandoffUpdate(command, events); + break; case "delegated_task.remote.complete": yield* dispatchRemoteDelegatedTaskComplete(command, events); break; ``` ## apps/server/src/orchestration-v2/ProjectionStore.ts Status: modified; +2; −0 ```diff @@ -1613,6 +1613,7 @@ ...(projection.thread.delegatedFrom === undefined ? {} : { delegatedFrom: projection.thread.delegatedFrom }), + ...(projection.thread.handoff === undefined ? {} : { handoff: projection.thread.handoff }), latestRunId: latestRun?.id ?? null, latestRunRequestedAt: latestRun?.requestedAt ?? null, latestRunStartedAt: latestRun?.startedAt ?? null, @@ -1871,6 +1872,7 @@ ...(input.state.thread.delegatedFrom === undefined ? {} : { delegatedFrom: input.state.thread.delegatedFrom }), + ...(input.state.thread.handoff === undefined ? {} : { handoff: input.state.thread.handoff }), latestRunId: input.state.latestRunId, latestRunRequestedAt: input.state.latestRunRequestedAt, latestRunStartedAt: input.state.latestRunStartedAt, ``` ## apps/server/src/orchestration-v2/ThreadImportService.ts Status: modified; +19; −4 ```diff @@ -68,6 +68,13 @@ | ThreadImportProjectMismatchError | ThreadImportContinuationError; +/** A checkout an import prepared for its thread. */ +export interface ImportedWorkspace { + readonly worktreePath: string; + readonly branch: string | null; + readonly undo: Effect.Effect; +} + /** * Creates a thread here seeded with a conversation from another environment, * for a thread moving here. The history is runless and marked as imported, @@ -78,13 +85,18 @@ export class ThreadImportService extends Context.Service< ThreadImportService, { + readonly importThread: ( - readonly importThread: ( input: OrchestratorMcpThreadImportInput & { readonly runtimeMode: RuntimeMode; readonly interactionMode: ProviderInteractionMode; readonly linkOrigin: OrchestrationV2LinkOrigin | undefined; + /** + * Prepares the checkout the thread works in, run only when this import + * creates the thread; `undo` runs if the thread then cannot be written. + */ + readonly workspace?: Effect.Effect; }, - ) => Effect.Effect; + ) => Effect.Effect; } >()("t3/orchestration-v2/ThreadImportService") {} @@ -190,6 +202,7 @@ .pipe(Effect.orElseSucceed(() => null)); let created = false; if (existing === null) { + const workspace = input.workspace === undefined ? undefined : yield* input.workspace; const now = yield* DateTime.now; const thread: OrchestrationV2AppThread = { createdBy: "agent", @@ -201,8 +214,9 @@ modelSelection: input.modelSelection, runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, + branch: workspace === undefined ? (input.branch ?? null) : workspace.branch, + worktreePath: + workspace === undefined ? (input.worktreePath ?? null) : workspace.worktreePath, - branch: input.branch ?? null, - worktreePath: input.worktreePath ?? null, activeProviderThreadId: null, historyOrigin: "v1_import", ...(input.linkOrigin === undefined ? {} : { linkOrigin: input.linkOrigin }), @@ -240,6 +254,7 @@ Effect.flatMap((raced) => (raced === null ? Effect.failCause(cause) : Effect.void)), ), ), + Effect.tapError(() => workspace?.undo ?? Effect.void), Effect.mapError((cause) => new ThreadImportWriteError({ threadId, cause })), ); created = true; ``` ## apps/server/src/orchestration-v2/ThreadMessageIntake.ts Status: modified; +1; −0 ```diff @@ -24,6 +24,7 @@ case "OrchestratorCommandPreviouslyRejectedError": case "OrchestratorCommandIdConflictError": case "OrchestratorSubagentThreadReadOnlyError": + case "OrchestratorThreadMovedError": case "OrchestratorThreadAboveModeLimitError": case "ThreadManagementProviderInstanceDisabledError": return true; ``` ## apps/server/src/orchestration-v2/testkit/CapturingCodexAdapter.ts Status: modified; +91; −64 ```diff @@ -38,7 +38,12 @@ */ export const makeCapturingCodexAdapter = ( capturedTurns: Ref.Ref>, + options: { + readonly response: string; + readonly modelSelection: ModelSelection; - options: { readonly response: string; readonly modelSelection: ModelSelection }, + /** Holds each turn open until this completes, so a test can act mid-turn. */ + readonly holdTurn?: Effect.Effect; + }, ) => ({ instanceId, @@ -61,6 +66,78 @@ updatedAt: now, lastError: null, }; + const providerTurnUpdate = ( + turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, + status: "running" | "completed", + startedAt: DateTime.Utc, + completedAt: DateTime.Utc | null, + ): ProviderAdapter.ProviderAdapterV2Event => ({ + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turnInput.threadId}:${turnInput.runOrdinal}`), + 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, + startedAt, + completedAt, + }, + }); + const finishTurn = (turnInput: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + const eventTime = yield* DateTime.now; + const providerTurnId = ProviderTurnId.make( + `provider-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, + ); + yield* PubSub.publishAll(events, [ + providerTurnUpdate(turnInput, "completed", eventTime, 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); + }); return { instanceId, driver, @@ -104,69 +181,19 @@ text: turnInput.message.text, }, ]); - const eventTime = yield* DateTime.now; - const providerTurnId = ProviderTurnId.make( - `provider-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, - ); + if (options.holdTurn !== undefined) { + // Ends the turn later, as a live provider does; startTurn returns now. + // Until then the turn is running, so Stop and steering can target it. - yield* PubSub.publishAll(events, [ - { + yield* PubSub.publish( + events, + providerTurnUpdate(turnInput, "running", yield* DateTime.now, null), + ); + yield* Effect.forkDetach( + options.holdTurn.pipe(Effect.andThen(finishTurn(turnInput))), + ); + return; + } + yield* finishTurn(turnInput); - 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, ``` ## apps/server/src/peer/PeerForwarding.test.ts Status: modified; +20; −9 ```diff @@ -39,6 +39,8 @@ import * as PeerLinks from "./PeerLinks.ts"; import * as PeerLinkRequests from "./PeerLinkRequests.ts"; import * as ThreadImportService from "../orchestration-v2/ThreadImportService.ts"; +import * as ThreadHandoff from "./handoff/ThreadHandoff.ts"; +import * as HandoffImport from "./handoff/HandoffImport.ts"; import { descriptorOf, layerLinkingEnvironment, linkTo, servePeer } from "./PeerLinks.testkit.ts"; const laptop = descriptorOf("environment-laptop", "Laptop"); @@ -215,16 +217,23 @@ ), ), Layer.provide(NodeServices.layer), - Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})), - Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})), - Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), - Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), Layer.provide(ServerSettings.layerTest()), + // Services the toolkits declare that these calls never reach. + Layer.provide( + Layer.mergeAll( + Layer.mock(ProviderRegistry.ProviderRegistry)({}), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(PeerForwarding.PeerForwarding)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), - Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), - Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), + Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), + Layer.mock(PeerLinks.PeerLinks)({}), - Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), - Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), - Layer.provide(Layer.mock(ThreadImportService.ThreadImportService)({})), + Layer.mock(ThreadImportService.ThreadImportService)({}), + Layer.mock(ThreadHandoff.ThreadHandoff)({}), + Layer.mock(HandoffImport.HandoffImport)({}), + ), + ), ), ); @@ -304,6 +313,8 @@ Layer.mock(RemoteDelegation.RemoteDelegation)({}), Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), Layer.mock(ThreadImportService.ThreadImportService)({}), + Layer.mock(ThreadHandoff.ThreadHandoff)({}), + Layer.mock(HandoffImport.HandoffImport)({}), ), ), Layer.fresh, ``` ## apps/server/src/peer/PeerLinkRequests.test.ts Status: modified; +2; −0 ```diff @@ -65,6 +65,7 @@ type ServedPeer, } from "./PeerLinks.testkit.ts"; import * as RemoteDelegation from "./RemoteDelegation.ts"; +import * as ThreadHandoff from "./handoff/ThreadHandoff.ts"; // The laptop's agent asks the user to link the box. The laptop is its real // orchestrator and toolkit; the box is its real descriptor and MCP OAuth on a @@ -156,6 +157,7 @@ Layer.provideMerge(PeerLinkRequests.layer), Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), Layer.provide(Layer.succeedContext(linking)), Layer.provide(NodeCrypto.layer), Layer.provideMerge(layerThreads), ``` ## apps/server/src/peer/PeerLinks.testkit.ts Status: modified; +18; −3 ```diff @@ -57,9 +57,16 @@ * socket. The descriptor can be swapped, as when its address comes to answer * as another environment, and every bearer it receives is recorded. */ -export const servePeer = ( +export const servePeer = ( initial: ExecutionEnvironmentDescriptor, tools: Layer.Layer, + /** More routes the peer serves, such as attachment uploads. */ + routes: Layer.Layer = Layer.empty as Layer.Layer, + /** + * The peer's own config and secret store, when its tools need the same ones + * its auth uses, as signed uploads do. A fresh pair otherwise. + */ + base?: Context.Context, ) => Effect.gen(function* () { const descriptor = yield* Ref.make(initial); @@ -69,11 +76,18 @@ getEnvironmentId: Ref.get(descriptor).pipe(Effect.map((current) => current.environmentId)), getDescriptor: Ref.get(descriptor), }); + const layerBase = + base === undefined + ? ServerSecretStore.layer.pipe( + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-peer-" }), + ), + ) + : Layer.succeedContext(base); const authContext = yield* EnvironmentAuth.layer.pipe( Layer.provide(Sqlite.layerMemory), - Layer.provideMerge(ServerSecretStore.layer), Layer.provideMerge(ServerEnvironment.layerIdentity), + Layer.provideMerge(layerBase), - Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-peer-" })), Layer.provideMerge(NodeServices.layer), Layer.fresh, Layer.build, @@ -108,6 +122,7 @@ ), Layer.provide(McpOAuth.layerMcpClientAuthenticator), ), + routes, ).pipe( Layer.provide(layerRecordBearers), Layer.provide(layerEnvironment), ``` ## apps/server/src/peer/RemoteDelegation.test.ts Status: modified; +15; −5 ```diff @@ -63,6 +63,8 @@ import * as RemoteDelegation from "./RemoteDelegation.ts"; import * as PeerLinkRequests from "./PeerLinkRequests.ts"; import * as ThreadImportService from "../orchestration-v2/ThreadImportService.ts"; +import * as ThreadHandoff from "./handoff/ThreadHandoff.ts"; +import * as HandoffImport from "./handoff/HandoffImport.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 @@ -462,12 +464,19 @@ Layer.provide(NodeCrypto.layer), Layer.provide(layerBoxThreads), Layer.provide(layerBoxLaunches), - Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})), - Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})), - Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), - Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})), + // Services the toolkits declare that these calls never reach. - Layer.provide(Layer.mock(ThreadImportService.ThreadImportService)({})), Layer.provide( + Layer.mergeAll( + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(PeerForwarding.PeerForwarding)({}), + Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), + Layer.mock(PeerLinks.PeerLinks)({}), + Layer.mock(ThreadImportService.ThreadImportService)({}), + Layer.mock(ThreadHandoff.ThreadHandoff)({}), + Layer.mock(HandoffImport.HandoffImport)({}), + ), + ), + Layer.provide( Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/p" }), ), Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), @@ -581,6 +590,7 @@ ), ), Layer.provide(Layer.succeedContext(linking)), + Layer.provide(Layer.mock(ThreadHandoff.ThreadHandoff)({})), Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})), Layer.provide(NodeCrypto.layer), Layer.provideMerge(layerThreads), ``` ## apps/server/src/peer/handoff/HandoffImport.ts Status: added; +131; −0 ```diff @@ -1,0 +1,131 @@ +import { OrchestratorMcpFailure } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import * as HostProcess from "@t3tools/shared/HostProcess"; + +import { deletePendingAttachment } from "../../assets/AttachmentUpload.ts"; +import { + parseThreadSegmentFromAttachmentId, + resolveAttachmentPathById, +} from "../../attachmentStore.ts"; +import * as ServerConfig from "../../config.ts"; +import type * as ThreadImportService from "../../orchestration-v2/ThreadImportService.ts"; +import * as ServerSettings from "../../serverSettings.ts"; +import * as VcsProcess from "../../vcs/VcsProcess.ts"; +import { resolveWorktreesDirectory } from "../../worktreesDirectory.ts"; +import * as HandoffGit from "./HandoffGit.ts"; + +/** + * The checkout a handed-off thread works in here: its bundle, uploaded as a + * pending attachment, applied in a new worktree of the project. The upload is + * spent either way; the worktree and branch are undone if the import fails. + */ +export class HandoffImport extends Context.Service< + HandoffImport, + { + readonly applyBundle: (input: { + readonly repoRoot: string; + readonly handoffId: string; + readonly bundle: { + readonly attachmentId: string; + readonly branch: string | null; + readonly tip: string; + readonly snapshot: string; + }; + }) => Effect.Effect; + } +>()("t3/peer/handoff/HandoffImport") {} + +const make = Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const settings = yield* ServerSettings.ServerSettingsService; + const path = yield* Path.Path; + const fileSystem = yield* FileSystem.FileSystem; + const processes = yield* VcsProcess.VcsProcess; + const git = yield* HandoffGit.HandoffGit; + + const applyBundle: HandoffImport["Service"]["applyBundle"] = (input) => + Effect.gen(function* () { + if (parseThreadSegmentFromAttachmentId(input.bundle.attachmentId) !== "pending") { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "The bundle must be a pending upload from t3_attachment_prepare_upload.", + }); + } + const bundlePath = resolveAttachmentPathById({ + attachmentsDir: config.attachmentsDir, + attachmentId: input.bundle.attachmentId, + }); + if (bundlePath === null) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: "The bundle upload was not found. Upload it again.", + }); + } + const worktreesSetting = yield* settings.getSettings.pipe( + Effect.map((current) => current.worktreesDirectory), + Effect.orElseSucceed(() => ""), + ); + const parent = resolveWorktreesDirectory( + worktreesSetting, + config.worktreesDir, + path, + yield* HostProcess.HomeDirectory, + ); + if (parent === null) { + return yield* new OrchestratorMcpFailure({ + code: "invalid_request", + message: + "This environment's worktree location is not usable. Fix it in Settings → Storage.", + }); + } + const name = (input.bundle.branch ?? `handoff-${input.handoffId}`).replace(/\//g, "-"); + const worktreePath = path.join(parent, path.basename(input.repoRoot), name); + const applied = yield* git + .apply({ + repoRoot: input.repoRoot, + worktreePath, + handoffId: input.handoffId, + bundlePath, + branch: input.bundle.branch, + tip: input.bundle.tip, + snapshot: input.bundle.snapshot, + }) + .pipe( + Effect.mapError( + (error) => + new OrchestratorMcpFailure({ + code: + error._tag === "HandoffGitStepError" ? "orchestration_error" : "invalid_request", + message: error.message, + }), + ), + Effect.ensuring( + deletePendingAttachment(input.bundle.attachmentId).pipe( + Effect.provideService(ServerConfig.ServerConfig, config), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.ignore, + ), + ), + ); + return { + worktreePath, + branch: applied.branch, + undo: processes + .run({ + operation: "HandoffImport.undo", + command: "git", + cwd: input.repoRoot, + args: ["worktree", "remove", "--force", worktreePath], + }) + .pipe(Effect.ignore), + } satisfies ThreadImportService.ImportedWorkspace; + }); + + return HandoffImport.of({ applyBundle }); +}); + +export const layer = Layer.effect(HandoffImport, make); ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/16751/diffs?after=ImFwcHMvc2VydmVyL3NyYy9wZWVyL2hhbmRvZmYvSGFuZG9mZkltcG9ydC50cyI&expectedVersion=10&format=markdown)