History

feat(mcp): fork existing conversations

Julius Marminge committed 2 months ago

Commit 970ea181 parent 415ed0f18 files · +1214 −7Browse files

Unified · Split

18 files
  1. apps/server/src/mcp/ConversationTransferMcpService.test.ts +174 −0
  2. apps/server/src/mcp/ConversationTransferMcpService.ts +367 −0
  3. apps/server/src/mcp/McpHttpServer.ts +2 −0
  4. apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +286 −4
  5. apps/server/src/mcp/toolkits/orchestrator/handlers.ts +13 −0
  6. apps/server/src/mcp/toolkits/orchestrator/tools.test.ts +26 −0
  7. apps/server/src/mcp/toolkits/orchestrator/tools.ts +37 −0
  8. apps/server/src/mcp/toolkits/worktree/registration.test.ts +7 −0
  9. apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +1 −0
  10. apps/server/src/orchestration-v2/Orchestrator.ts +103 −1
  11. apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +8 −2
  12. docs/orchestration-v2/orchestrator-mcp-server.md +18 −0
  13. docs/user/conversation-forks.md +11 −0
  14. packages/contracts/src/conversationTransferMcp.test.ts +38 −0
  15. packages/contracts/src/conversationTransferMcp.ts +112 −0
  16. packages/contracts/src/index.ts +1 −0
  17. packages/contracts/src/orchestrationV2.ts +8 −0
  18. packages/shared/src/t3McpToolPresentation.ts +2 −0
apps/server/src/mcp/ConversationTransferMcpService.test.ts+174−0
@@ -0,0 +1,174 @@
1+import { assert, describe, it } from "@effect/vitest";
2+import {
3+ EnvironmentId,
4+ type OrchestrationV2Run,
5+ ProviderInstanceId,
6+ type OrchestrationV2ThreadProjection,
7+ ProjectId,
8+ ThreadId,
9+} from "@t3tools/contracts";
10+import * as Effect from "effect/Effect";
11+import * as Layer from "effect/Layer";
12+import * as Ref from "effect/Ref";
13+
14+import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts";
15+import type { McpInvocationScope } from "./McpInvocationContext.ts";
16+import * as ConversationTransfer from "./ConversationTransferMcpService.ts";
17+
18+const projectId = ProjectId.make("project:conversation-transfer");
19+const parentThreadId = ThreadId.make("thread:conversation-transfer-parent");
20+const sourceThreadId = ThreadId.make("thread:conversation-transfer-source");
21+
22+function projection(input: {
23+ readonly threadId: ThreadId;
24+ readonly runtimeMode: "approval-required" | "full-access";
25+ readonly interactionMode: "plan" | "default";
26+}): OrchestrationV2ThreadProjection {
27+ return {
28+ thread: {
29+ id: input.threadId,
30+ projectId,
31+ runtimeMode: input.runtimeMode,
32+ interactionMode: input.interactionMode,
33+ },
34+ runs: [],
35+ contextTransfers: [],
36+ } as unknown as OrchestrationV2ThreadProjection;
37+}
38+
39+const scope = (capabilities: ReadonlyArray<"orchestration"> = ["orchestration"]) =>
40+ ({
41+ environmentId: EnvironmentId.make("environment:conversation-transfer"),
42+ threadId: parentThreadId,
43+ providerSessionId: "provider-session:conversation-transfer",
44+ providerInstanceId: ProviderInstanceId.make("codex"),
45+ capabilities: new Set(capabilities),
46+ issuedAt: 1,
47+ }) satisfies McpInvocationScope;
48+
49+function testLayer(input: {
50+ readonly parent: OrchestrationV2ThreadProjection;
51+ readonly source: OrchestrationV2ThreadProjection;
52+ readonly dispatch: ThreadManagementService["Service"]["dispatch"];
53+}) {
54+ return ConversationTransfer.layer.pipe(
55+ Layer.provide(
56+ Layer.mock(ThreadManagementService)({
57+ getThreadProjection: () => Effect.succeed(input.parent),
58+ getProjectThread: ({ threadId }) =>
59+ Effect.succeed(threadId === parentThreadId ? input.parent : input.source),
60+ dispatch: input.dispatch,
61+ }),
62+ ),
63+ );
64+}
65+
66+describe("ConversationTransferMcpService", () => {
67+ it("reports a selected durable source run as a turn-specific native fork without a native turn ref", () => {
68+ const run = {
69+ providerThreadId: "provider-thread:source",
70+ } as unknown as OrchestrationV2Run;
71+ const source = {
72+ providerThreads: [
73+ {
74+ id: "provider-thread:source",
75+ providerSessionId: "provider-session:source",
76+ nativeThreadRef: { strength: "strong" },
77+ },
78+ ],
79+ providerSessions: [
80+ {
81+ id: "provider-session:source",
82+ capabilities: {
83+ threads: { canForkThread: true, canForkFromTurn: false },
84+ identity: { nativeThreadIds: "strong" },
85+ },
86+ },
87+ ],
88+ } as unknown as OrchestrationV2ThreadProjection;
89+ assert.equal(
90+ ConversationTransfer.conversationForkNativeEligibility(source, run),
91+ "provider_does_not_support_turn_fork",
92+ );
93+ });
94+
95+ it.effect("rejects missing stable runs and inherited permission escalation before dispatch", () =>
96+ Effect.gen(function* () {
97+ const dispatched = yield* Ref.make(0);
98+ const parent = projection({
99+ threadId: parentThreadId,
100+ runtimeMode: "approval-required",
101+ interactionMode: "plan",
102+ });
103+ const source = projection({
104+ threadId: sourceThreadId,
105+ runtimeMode: "full-access",
106+ interactionMode: "default",
107+ });
108+ const escalation = yield* Effect.gen(function* () {
109+ const service = yield* ConversationTransfer.ConversationTransferMcpService;
110+ return yield* service
111+ .fork(scope(), {
112+ sourceThreadId,
113+ sourcePoint: { type: "latest_stable" },
114+ clientRequestId: "permission-ceiling",
115+ })
116+ .pipe(Effect.flip);
117+ }).pipe(
118+ Effect.provide(
119+ testLayer({
120+ parent,
121+ source,
122+ dispatch: () =>
123+ Ref.update(dispatched, (count) => count + 1).pipe(Effect.as({} as never)),
124+ }),
125+ ),
126+ );
127+ assert.equal(escalation.code, "runtime_mode_escalation_denied");
128+
129+ const noRun = yield* Effect.gen(function* () {
130+ const service = yield* ConversationTransfer.ConversationTransferMcpService;
131+ return yield* service
132+ .fork(scope(), {
133+ sourcePoint: { type: "run", runId: "run:missing" as never },
134+ clientRequestId: "missing-run",
135+ })
136+ .pipe(Effect.flip);
137+ }).pipe(
138+ Effect.provide(
139+ testLayer({
140+ parent,
141+ source: parent,
142+ dispatch: () =>
143+ Ref.update(dispatched, (count) => count + 1).pipe(Effect.as({} as never)),
144+ }),
145+ ),
146+ );
147+ assert.equal(noRun.code, "run_not_found");
148+ assert.equal(yield* Ref.get(dispatched), 0);
149+ }),
150+ );
151+
152+ it.effect("denies transfer reads without orchestration capability", () =>
153+ Effect.gen(function* () {
154+ const parent = projection({
155+ threadId: parentThreadId,
156+ runtimeMode: "full-access",
157+ interactionMode: "default",
158+ });
159+ const error = yield* Effect.gen(function* () {
160+ const service = yield* ConversationTransfer.ConversationTransferMcpService;
161+ return yield* service.list(scope([]), {}).pipe(Effect.flip);
162+ }).pipe(
163+ Effect.provide(
164+ testLayer({
165+ parent,
166+ source: parent,
167+ dispatch: () => Effect.die("dispatch should not run"),
168+ }),
169+ ),
170+ );
171+ assert.equal(error.code, "capability_denied");
172+ }),
173+ );
174+});
apps/server/src/mcp/ConversationTransferMcpService.ts+367−0
@@ -0,0 +1,367 @@
1+import {
2+ CommandId,
3+ type ConversationForkInput,
4+ type ConversationForkNativeEligibility,
5+ type ConversationForkResult,
6+ type ConversationTransferListInput,
7+ type ConversationTransferListResult,
8+ type ConversationTransferReceipt,
9+ type OrchestrationV2ContextTransfer,
10+ type OrchestrationV2Run,
11+ type OrchestrationV2ThreadForkSourcePoint,
12+ type OrchestrationV2ThreadProjection,
13+ OrchestratorMcpFailure,
14+ type ProviderInteractionMode,
15+ type RuntimeMode,
16+ ThreadId,
17+} from "@t3tools/contracts";
18+import * as Context from "effect/Context";
19+import * as DateTime from "effect/DateTime";
20+import * as Effect from "effect/Effect";
21+import * as Layer from "effect/Layer";
22+import * as Option from "effect/Option";
23+import * as Schema from "effect/Schema";
24+
25+import type { OrchestratorV2DispatchResult } from "../orchestration-v2/Orchestrator.ts";
26+import {
27+ ThreadManagementService,
28+ ThreadManagementThreadNotFoundError,
29+} from "../orchestration-v2/ThreadManagementService.ts";
30+import type { McpInvocationScope } from "./McpInvocationContext.ts";
31+
32+export class ConversationTransferMcpService extends Context.Service<
33+ ConversationTransferMcpService,
34+ {
35+ readonly list: (
36+ scope: McpInvocationScope,
37+ input: ConversationTransferListInput,
38+ ) => Effect.Effect<ConversationTransferListResult, OrchestratorMcpFailure>;
39+ readonly fork: (
40+ scope: McpInvocationScope,
41+ input: ConversationForkInput,
42+ ) => Effect.Effect<ConversationForkResult, OrchestratorMcpFailure>;
43+ }
44+>()("t3/mcp/ConversationTransferMcpService") {}
45+
46+const isThreadNotFound = Schema.is(ThreadManagementThreadNotFoundError);
47+
48+function failure(code: OrchestratorMcpFailure["code"], message: string): OrchestratorMcpFailure {
49+ return new OrchestratorMcpFailure({ code, message });
50+}
51+
52+function errorMessage(error: unknown): string {
53+ return error instanceof Error ? error.message : String(error);
54+}
55+
56+function runtimeModeRank(mode: RuntimeMode): number {
57+ switch (mode) {
58+ case "approval-required":
59+ return 0;
60+ case "auto-accept-edits":
61+ return 1;
62+ case "auto":
63+ return 2;
64+ case "full-access":
65+ return 3;
66+ }
67+}
68+
69+function interactionModeRank(mode: ProviderInteractionMode): number {
70+ return mode === "plan" ? 0 : 1;
71+}
72+
73+function stablePart(value: string): string {
74+ return encodeURIComponent(value);
75+}
76+
77+function targetThreadId(scope: McpInvocationScope, requestKey: string): ThreadId {
78+ return ThreadId.make(
79+ `thread:mcp:${stablePart(scope.providerSessionId)}:conversation-fork:target:${stablePart(requestKey)}`,
80+ );
81+}
82+
83+function forkCommandId(scope: McpInvocationScope, requestKey: string): CommandId {
84+ return CommandId.make(
85+ `command:mcp:${stablePart(scope.providerSessionId)}:thread-fork:${stablePart(requestKey)}`,
86+ );
87+}
88+
89+function sourceRun(
90+ projection: OrchestrationV2ThreadProjection,
91+ point: OrchestrationV2ThreadForkSourcePoint,
92+): OrchestrationV2Run | undefined {
93+ switch (point.type) {
94+ case "latest_stable":
95+ return projection.runs
96+ .filter((run) => run.status === "completed" && run.checkpointId !== null)
97+ .toSorted((left, right) => right.ordinal - left.ordinal)[0];
98+ case "run":
99+ return projection.runs.find((run) => run.id === point.runId);
100+ case "checkpoint":
101+ const checkpoint = projection.checkpoints.find(
102+ (candidate) => candidate.id === point.checkpointId,
103+ );
104+ return checkpoint?.runId === null || checkpoint === undefined
105+ ? undefined
106+ : projection.runs.find((run) => run.id === checkpoint.runId);
107+ }
108+}
109+
110+export function conversationForkNativeEligibility(
111+ projection: OrchestrationV2ThreadProjection,
112+ run: OrchestrationV2Run,
113+): ConversationForkNativeEligibility {
114+ const providerThread = projection.providerThreads.find(
115+ (candidate) => candidate.id === run.providerThreadId,
116+ );
117+ const session = projection.providerSessions.find(
118+ (candidate) => candidate.id === providerThread?.providerSessionId,
119+ );
120+ if (session === undefined) return "source_provider_session_unavailable";
121+ if (!session.capabilities.threads.canForkThread) return "provider_does_not_support_fork";
122+ if (!session.capabilities.threads.canForkFromTurn) {
123+ return "provider_does_not_support_turn_fork";
124+ }
125+ if (
126+ providerThread?.nativeThreadRef?.strength !== "strong" ||
127+ session.capabilities.identity.nativeThreadIds !== "strong"
128+ ) {
129+ return "strong_native_source_unavailable";
130+ }
131+ return "eligible";
132+}
133+
134+function receipt(result: OrchestratorV2DispatchResult, commandId: CommandId) {
135+ return {
136+ commandId,
137+ commandType: "thread.fork",
138+ sequence: result.sequence,
139+ eventIds: result.storedEvents.map((stored) => stored.event.id),
140+ } satisfies ConversationTransferReceipt;
141+}
142+
143+function forkTransfer(
144+ result: OrchestratorV2DispatchResult,
145+ targetThreadId: ThreadId,
146+): OrchestrationV2ContextTransfer | undefined {
147+ for (const stored of result.storedEvents) {
148+ if (
149+ stored.event.type === "context-transfer.created" &&
150+ stored.event.payload.type === "fork" &&
151+ stored.event.payload.targetThreadId === targetThreadId
152+ ) {
153+ return stored.event.payload;
154+ }
155+ }
156+ return undefined;
157+}
158+
159+export const make = Effect.gen(function* () {
160+ const threads = yield* ThreadManagementService;
161+
162+ const requireCapability = (scope: McpInvocationScope) =>
163+ scope.capabilities.has("orchestration")
164+ ? Effect.void
165+ : Effect.fail(failure("capability_denied", "This MCP credential cannot control threads."));
166+
167+ const loadScoped = Effect.fn("ConversationTransferMcpService.loadScoped")(function* (
168+ scope: McpInvocationScope,
169+ requestedThreadId: ThreadId | undefined,
170+ ) {
171+ yield* requireCapability(scope);
172+ const parent = yield* threads
173+ .getThreadProjection(scope.threadId)
174+ .pipe(
175+ Effect.mapError((error) =>
176+ failure(
177+ "orchestration_error",
178+ `Unable to load the calling thread: ${errorMessage(error)}`,
179+ ),
180+ ),
181+ );
182+ const threadId = requestedThreadId ?? scope.threadId;
183+ const target = yield* threads
184+ .getProjectThread({ projectId: parent.thread.projectId, threadId })
185+ .pipe(
186+ Effect.mapError((error) =>
187+ isThreadNotFound(error)
188+ ? failure(
189+ "thread_not_found",
190+ `Thread ${threadId} was not found in the calling project.`,
191+ )
192+ : failure(
193+ "orchestration_error",
194+ `Unable to load thread ${threadId}: ${errorMessage(error)}`,
195+ ),
196+ ),
197+ );
198+ return { parent, target };
199+ });
200+
201+ const resultFromDispatch = Effect.fn("ConversationTransferMcpService.resultFromDispatch")(
202+ function* (input: {
203+ readonly request: ConversationForkInput;
204+ readonly commandId: CommandId;
205+ readonly targetThreadId: ThreadId;
206+ readonly dispatched: OrchestratorV2DispatchResult;
207+ readonly source: OrchestrationV2ThreadProjection | undefined;
208+ }) {
209+ const transfer = forkTransfer(input.dispatched, input.targetThreadId);
210+ if (transfer === undefined) {
211+ return yield* failure(
212+ "orchestration_error",
213+ "The fork command committed without returning its context-transfer event.",
214+ );
215+ }
216+ const run =
217+ input.source === undefined || transfer.sourcePoint.runId === undefined
218+ ? undefined
219+ : input.source.runs.find((candidate) => candidate.id === transfer.sourcePoint.runId);
220+ return {
221+ sourceThreadId: transfer.sourceThreadId,
222+ targetThreadId: input.targetThreadId,
223+ scope: "conversation_through_source_point",
224+ requestedSourcePoint: input.request.sourcePoint,
225+ canonicalSourcePoint: transfer.sourcePoint,
226+ transfer,
227+ providerSupport: {
228+ sourceProviderInstanceId: transfer.sourceProviderInstanceId,
229+ nativeForkEligibility:
230+ input.source === undefined || run === undefined
231+ ? "source_provider_session_unavailable"
232+ : conversationForkNativeEligibility(input.source, run),
233+ resolutionTiming: "first_target_turn",
234+ fallback: "provider_context_capabilities_checked_on_first_target_turn",
235+ },
236+ receipt: receipt(input.dispatched, input.commandId),
237+ } satisfies ConversationForkResult;
238+ },
239+ );
240+
241+ return ConversationTransferMcpService.of({
242+ list: (scope, input) =>
243+ Effect.gen(function* () {
244+ const { target } = yield* loadScoped(scope, input.threadId);
245+ const limit = input.limit ?? 50;
246+ const matching = target.contextTransfers
247+ .filter((transfer) => input.type === undefined || transfer.type === input.type)
248+ .toSorted(
249+ (left, right) =>
250+ DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
251+ );
252+ return {
253+ threadId: target.thread.id,
254+ scope: "current_project",
255+ transfers: matching.slice(0, limit),
256+ hasMore: matching.length > limit,
257+ };
258+ }),
259+ fork: (scope, input) =>
260+ Effect.gen(function* () {
261+ yield* requireCapability(scope);
262+ const commandId = forkCommandId(scope, input.clientRequestId);
263+ const targetId = targetThreadId(scope, input.clientRequestId);
264+ const existingTarget = yield* Effect.option(threads.getThreadProjection(targetId));
265+ const existingTransfer = Option.isSome(existingTarget)
266+ ? existingTarget.value.contextTransfers.find(
267+ (transfer) => transfer.type === "fork" && transfer.targetThreadId === targetId,
268+ )
269+ : undefined;
270+ if (existingTransfer !== undefined) {
271+ const replayed = yield* threads
272+ .dispatch({
273+ type: "thread.fork",
274+ createdBy: "agent",
275+ creationSource: "mcp",
276+ commandId,
277+ sourceThreadId: input.sourceThreadId ?? scope.threadId,
278+ targetThreadId: targetId,
279+ sourcePoint: input.sourcePoint,
280+ ...(input.title === undefined ? {} : { title: input.title }),
281+ })
282+ .pipe(
283+ Effect.mapError((error) =>
284+ failure(
285+ "orchestration_error",
286+ `Unable to replay the fork receipt: ${errorMessage(error)}`,
287+ ),
288+ ),
289+ );
290+ const replaySource = yield* Effect.option(
291+ threads.getThreadProjection(existingTransfer.sourceThreadId),
292+ );
293+ return yield* resultFromDispatch({
294+ request: input,
295+ commandId,
296+ targetThreadId: targetId,
297+ dispatched: replayed,
298+ source: Option.getOrUndefined(replaySource),
299+ });
300+ }
301+
302+ const { parent, target: source } = yield* loadScoped(scope, input.sourceThreadId);
303+ if (
304+ runtimeModeRank(source.thread.runtimeMode) > runtimeModeRank(parent.thread.runtimeMode)
305+ ) {
306+ return yield* failure(
307+ "runtime_mode_escalation_denied",
308+ `The source thread's ${source.thread.runtimeMode} runtime mode exceeds the calling thread's ${parent.thread.runtimeMode} ceiling.`,
309+ );
310+ }
311+ if (
312+ interactionModeRank(source.thread.interactionMode) >
313+ interactionModeRank(parent.thread.interactionMode)
314+ ) {
315+ return yield* failure(
316+ "interaction_mode_escalation_denied",
317+ `The source thread's ${source.thread.interactionMode} interaction mode exceeds the calling thread's ${parent.thread.interactionMode} ceiling.`,
318+ );
319+ }
320+ const run = sourceRun(source, input.sourcePoint);
321+ if (run === undefined) {
322+ return yield* failure(
323+ "run_not_found",
324+ `No run in thread ${source.thread.id} matches source point ${input.sourcePoint.type}.`,
325+ );
326+ }
327+ if (run.status !== "completed") {
328+ return yield* failure(
329+ "invalid_request",
330+ `Fork source run ${run.id} is ${run.status}; only completed runs are stable fork sources.`,
331+ );
332+ }
333+
334+ const dispatched = yield* threads
335+ .dispatch({
336+ type: "thread.fork",
337+ createdBy: "agent",
338+ creationSource: "mcp",
339+ commandId,
340+ sourceThreadId: source.thread.id,
341+ targetThreadId: targetId,
342+ sourcePoint: input.sourcePoint,
343+ ...(input.title === undefined ? {} : { title: input.title }),
344+ policyCeiling: {
345+ callerThreadId: parent.thread.id,
346+ runtimeMode: parent.thread.runtimeMode,
347+ interactionMode: parent.thread.interactionMode,
348+ },
349+ })
350+ .pipe(
351+ Effect.mapError((error) =>
352+ failure("orchestration_error", `Unable to fork the thread: ${errorMessage(error)}`),
353+ ),
354+ );
355+ return yield* resultFromDispatch({
356+ request: input,
357+ commandId,
358+ targetThreadId: targetId,
359+ dispatched,
360+ source,
361+ });
362+ }),
363+ });
364+});
365+
366+export const layer: Layer.Layer<ConversationTransferMcpService, never, ThreadManagementService> =
367+ Layer.effect(ConversationTransferMcpService, make);
apps/server/src/mcp/McpHttpServer.ts+2−0
@@ -11,6 +11,7 @@ import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstab · 10 unmodified lines
1111
1212 import packageJson from "../../package.json" with { type: "json" };
1313 import * as McpInvocationContext from "./McpInvocationContext.ts";
14+import * as ConversationTransferMcpService from "./ConversationTransferMcpService.ts";
1415 import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
1516 import * as ThreadMetadataMcpService from "./ThreadMetadataMcpService.ts";
1617 import * as McpSessionRegistry from "./McpSessionRegistry.ts";
@@ -224,6 +225,7 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll( · 207 unmodified lines
224225
225226 export const OrchestratorToolkitRegistrationLive = McpServer.toolkit(OrchestratorToolkit).pipe(
226227 Layer.provide(OrchestratorToolkitHandlersLive),
228+ Layer.provide(ConversationTransferMcpService.layer),
227229 Layer.provide(OrchestratorMcpService.layer),
228230 Layer.provide(ThreadMetadataMcpService.layer),
229231 );
apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+286−4
@@ -2,6 +2,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; · 1 unmodified line
22 import { describe, expect, it } from "@effect/vitest";
33 import {
44 CommandId,
5+ ConversationForkResult,
6+ ConversationTransferListResult,
57 EnvironmentId,
68 IsoDateTime,
79 MessageId,
@@ -37,6 +39,7 @@ import * as DateTime from "effect/DateTime"; · 29 unmodified lines
3739 import * as Deferred from "effect/Deferred";
3840 import * as Effect from "effect/Effect";
3941 import * as FileSystem from "effect/FileSystem";
42+import * as Fiber from "effect/Fiber";
4043 import * as Layer from "effect/Layer";
4144 import * as Option from "effect/Option";
4245 import * as PubSub from "effect/PubSub";
@@ -50,6 +53,10 @@ import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexA · 7 unmodified lines
5053 import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts";
5154 import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts";
5255 import { layer as threadManagementServiceLayer } from "../orchestration-v2/ThreadManagementService.ts";
56+import {
57+ layer as threadForkServiceLayer,
58+ ThreadForkServiceV2,
59+} from "../orchestration-v2/ThreadForkService.ts";
5360 import {
5461 type ProviderAdapterV2Event,
5562 ProviderAdapterProtocolError,
@@ -91,6 +98,10 @@ const queuedFollowupPrompt = "Complete the queued follow-up and return the final · 35 unmodified lines
9198 const queuedFollowupResult = "Queued delegated follow-up completed.";
9299
93100 const decodeCreateThreadsResult = Schema.decodeUnknownEffect(OrchestratorMcpCreateThreadsResult);
101+const decodeConversationForkResult = Schema.decodeUnknownEffect(ConversationForkResult);
102+const decodeConversationTransferListResult = Schema.decodeUnknownEffect(
103+ ConversationTransferListResult,
104+);
94105 const decodeCreatedThread = Schema.decodeUnknownEffect(OrchestratorMcpCreatedThread);
95106 const decodeDelegateTaskResult = Schema.decodeUnknownEffect(OrchestratorMcpDelegateTaskResult);
96107 const decodeTaskCancelResult = Schema.decodeUnknownEffect(OrchestratorMcpTaskCancelResult);
@@ -517,6 +528,23 @@ describe("orchestrator MCP toolkit", () => { · 420 unmodified lines
517528 Ref.update(continuationOffers, (existing) => [...existing, request]),
518529 take: Effect.never,
519530 });
531+ const forkPlanEntered = yield* Deferred.make<void>();
532+ const releaseForkPlan = yield* Deferred.make<void>();
533+ const gatedThreadForkServiceLayer = Layer.effect(
534+ ThreadForkServiceV2,
535+ Effect.gen(function* () {
536+ const service = yield* ThreadForkServiceV2;
537+ return ThreadForkServiceV2.of({
538+ plan: (input) =>
539+ input.title === "Race-gated fork"
540+ ? Deferred.succeed(forkPlanEntered, undefined).pipe(
541+ Effect.andThen(Deferred.await(releaseForkPlan)),
542+ Effect.andThen(service.plan(input)),
543+ )
544+ : service.plan(input),
545+ });
546+ }),
547+ ).pipe(Layer.provide(threadForkServiceLayer));
520548 // Offers land after the finalize projection writes, so poll briefly
521549 // instead of asserting counts immediately.
522550 const waitForContinuationOffers = (count: number) =>
@@ -553,6 +581,7 @@ describe("orchestrator MCP toolkit", () => { · 30 unmodified lines
553581 },
554582 },
555583 registryLayer,
584+ { threadForkServiceLayer: gatedThreadForkServiceLayer },
556585 ).pipe(Layer.provide(continuationProbeLayer));
557586 const orchestrationLayer = Layer.merge(
558587 orchestratorLayer,
@@ -1952,17 +1981,255 @@ describe("orchestrator MCP toolkit", () => { · 1393 unmodified lines
19521981 `Claude completed: ${createdThreadPrompt}`,
19531982 );
19541983
1955− const forkedThreadId = ThreadId.make("thread:mcp-orchestrator-inherited-read");
1984+ const collisionCreate = yield* decodeCreateThreadsResult(
1985+ (yield* invoke("create_threads", {
1986+ clientRequestId: "fork",
1987+ threads: [{ title: "Create namespace collision sentinel" }],
1988+ })).structuredContent,
1989+ ).pipe(Effect.orDie);
1990+ const collisionFork = yield* decodeConversationForkResult(
1991+ (yield* invoke("t3_thread_fork", {
1992+ sourceThreadId: promptedThread.threadId,
1993+ sourcePoint: { type: "latest_stable" },
1994+ clientRequestId: "0",
1995+ })).structuredContent,
1996+ ).pipe(Effect.orDie);
1997+ expect(collisionCreate.threads[0]?.threadId).not.toBe(collisionFork.targetThreadId);
1998+ expect(
1999+ (yield* orchestrator.getThreadProjection(collisionCreate.threads[0]!.threadId)).thread
2000+ .title,
2001+ ).toBe("Create namespace collision sentinel");
2002+ expect(
2003+ (yield* orchestrator.getThreadProjection(collisionFork.targetThreadId)).thread
2004+ .lineage,
2005+ ).toMatchObject({
2006+ parentThreadId: promptedThread.threadId,
2007+ relationshipToParent: "fork",
2008+ });
2009+
2010+ const occupiedForkTargetId = ThreadId.make(
2011+ "thread:mcp:mcp-provider-session-parent:conversation-fork:target:occupied-target",
2012+ );
19562013 yield* orchestrator.dispatch({
1957− type: "thread.fork",
2014+ type: "thread.create",
19582015 createdBy: "user",
19592016 creationSource: "web",
1960− commandId: CommandId.make("command:mcp-orchestrator-inherited-read"),
2017+ commandId: CommandId.make("command:mcp-fork:occupied-target-create"),
2018+ threadId: occupiedForkTargetId,
2019+ projectId,
2020+ title: "Existing target must survive",
2021+ modelSelection: codexSelection,
2022+ runtimeMode: "full-access",
2023+ interactionMode: "default",
2024+ branch: null,
2025+ worktreePath: cwd,
2026+ });
2027+ const occupiedForkCall = yield* invoke("t3_thread_fork", {
2028+ sourceThreadId: promptedThread.threadId,
2029+ sourcePoint: { type: "latest_stable" },
2030+ clientRequestId: "occupied-target",
2031+ });
2032+ expect(occupiedForkCall.structuredContent).toMatchObject({
2033+ _tag: "OrchestratorMcpFailure",
2034+ code: "orchestration_error",
2035+ });
2036+ const occupiedForkTarget =
2037+ yield* orchestrator.getThreadProjection(occupiedForkTargetId);
2038+ expect(occupiedForkTarget.thread.title).toBe("Existing target must survive");
2039+ expect(occupiedForkTarget.contextTransfers).toEqual([]);
2040+
2041+ const deletedOccupiedForkTargetId = ThreadId.make(
2042+ "thread:mcp:mcp-provider-session-parent:conversation-fork:target:deleted-occupied-target",
2043+ );
2044+ yield* orchestrator.dispatch({
2045+ type: "thread.create",
2046+ createdBy: "user",
2047+ creationSource: "web",
2048+ commandId: CommandId.make("command:mcp-fork:deleted-occupied-target-create"),
2049+ threadId: deletedOccupiedForkTargetId,
2050+ projectId,
2051+ title: "Deleted target must stay deleted",
2052+ modelSelection: codexSelection,
2053+ runtimeMode: "full-access",
2054+ interactionMode: "default",
2055+ branch: null,
2056+ worktreePath: cwd,
2057+ });
2058+ yield* orchestrator.dispatch({
2059+ type: "thread.delete",
2060+ commandId: CommandId.make("command:mcp-fork:deleted-occupied-target-delete"),
2061+ threadId: deletedOccupiedForkTargetId,
2062+ });
2063+ const deletedOccupiedForkCall = yield* invoke("t3_thread_fork", {
2064+ sourceThreadId: promptedThread.threadId,
2065+ sourcePoint: { type: "latest_stable" },
2066+ clientRequestId: "deleted-occupied-target",
2067+ });
2068+ expect(deletedOccupiedForkCall.structuredContent).toMatchObject({
2069+ _tag: "OrchestratorMcpFailure",
2070+ code: "orchestration_error",
2071+ });
2072+ const deletedOccupiedForkTarget = yield* orchestrator.getThreadProjection(
2073+ deletedOccupiedForkTargetId,
2074+ );
2075+ expect(deletedOccupiedForkTarget.thread.title).toBe("Deleted target must stay deleted");
2076+ expect(deletedOccupiedForkTarget.thread.deletedAt).not.toBeNull();
2077+ expect(deletedOccupiedForkTarget.contextTransfers).toEqual([]);
2078+
2079+ const racingForkFiber = yield* Effect.forkChild(
2080+ invoke("t3_thread_fork", {
2081+ sourceThreadId: promptedThread.threadId,
2082+ sourcePoint: { type: "latest_stable" },
2083+ title: "Race-gated fork",
2084+ clientRequestId: "orchestrator-race-gated-fork",
2085+ }),
2086+ );
2087+ yield* Deferred.await(forkPlanEntered);
2088+ const sourceWriterStarted = yield* Deferred.make<void>();
2089+ const callerWriterStarted = yield* Deferred.make<void>();
2090+ const sourceWriterFiber = yield* Effect.forkChild(
2091+ Deferred.succeed(sourceWriterStarted, undefined).pipe(
2092+ Effect.andThen(
2093+ orchestrator.dispatch({
2094+ type: "thread.runtime-mode.set",
2095+ commandId: CommandId.make("command:mcp-fork:race-source-runtime"),
2096+ threadId: promptedThread.threadId,
2097+ runtimeMode: "approval-required",
2098+ }),
2099+ ),
2100+ ),
2101+ );
2102+ const callerWriterFiber = yield* Effect.forkChild(
2103+ Deferred.succeed(callerWriterStarted, undefined).pipe(
2104+ Effect.andThen(
2105+ orchestrator.dispatch({
2106+ type: "thread.interaction-mode.set",
2107+ commandId: CommandId.make("command:mcp-fork:race-caller-interaction"),
2108+ threadId: parentThreadId,
2109+ interactionMode: "plan",
2110+ }),
2111+ ),
2112+ ),
2113+ );
2114+ yield* Deferred.await(sourceWriterStarted);
2115+ yield* Deferred.await(callerWriterStarted);
2116+ yield* Effect.yieldNow;
2117+ yield* Deferred.succeed(releaseForkPlan, undefined);
2118+ const racingForkCall = yield* Fiber.join(racingForkFiber);
2119+ const sourceWriter = yield* Fiber.join(sourceWriterFiber);
2120+ const callerWriter = yield* Fiber.join(callerWriterFiber);
2121+ const racingFork = yield* decodeConversationForkResult(
2122+ racingForkCall.structuredContent,
2123+ ).pipe(Effect.orDie);
2124+ expect(racingFork.receipt.sequence).toBeLessThan(sourceWriter.sequence);
2125+ expect(racingFork.receipt.sequence).toBeLessThan(callerWriter.sequence);
2126+ expect(
2127+ (yield* orchestrator.getThreadProjection(racingFork.targetThreadId)).thread,
2128+ ).toMatchObject({ runtimeMode: "full-access", interactionMode: "default" });
2129+ yield* orchestrator.dispatch({
2130+ type: "thread.runtime-mode.set",
2131+ commandId: CommandId.make("command:mcp-fork:race-restore-source-runtime"),
2132+ threadId: promptedThread.threadId,
2133+ runtimeMode: "full-access",
2134+ });
2135+ yield* orchestrator.dispatch({
2136+ type: "thread.interaction-mode.set",
2137+ commandId: CommandId.make("command:mcp-fork:race-restore-caller-interaction"),
2138+ threadId: parentThreadId,
2139+ interactionMode: "default",
2140+ });
2141+
2142+ const deniedFork = yield* orchestrator
2143+ .dispatch({
2144+ type: "thread.fork",
2145+ createdBy: "agent",
2146+ creationSource: "mcp",
2147+ commandId: CommandId.make("command:mcp-fork:denied-policy-ceiling"),
2148+ sourceThreadId: promptedThread.threadId,
2149+ targetThreadId: ThreadId.make("thread:mcp-fork:denied-policy-ceiling"),
2150+ sourcePoint: { type: "latest_stable" },
2151+ policyCeiling: {
2152+ callerThreadId: parentThreadId,
2153+ runtimeMode: "approval-required",
2154+ interactionMode: "plan",
2155+ },
2156+ })
2157+ .pipe(Effect.flip);
2158+ expect(deniedFork._tag).toBe("OrchestratorDispatchError");
2159+
2160+ const forkCall = yield* invoke("t3_thread_fork", {
2161+ sourceThreadId: promptedThread.threadId,
2162+ sourcePoint: { type: "latest_stable" },
2163+ title: "Inherited read thread",
2164+ clientRequestId: "orchestrator-inherited-read",
2165+ });
2166+ expect(forkCall.isError).toBe(false);
2167+ const forked = yield* decodeConversationForkResult(forkCall.structuredContent).pipe(
2168+ Effect.orDie,
2169+ );
2170+ const forkedThreadId = forked.targetThreadId;
2171+ expect(forked).toMatchObject({
2172+ sourceThreadId: promptedThread.threadId,
2173+ scope: "conversation_through_source_point",
2174+ requestedSourcePoint: { type: "latest_stable" },
2175+ transfer: { type: "fork", status: "pending" },
2176+ providerSupport: { resolutionTiming: "first_target_turn" },
2177+ receipt: { commandType: "thread.fork" },
2178+ });
2179+ expect(forked.canonicalSourcePoint.runId).toBeDefined();
2180+ const forkCheckpointId = forked.canonicalSourcePoint.checkpointId;
2181+ if (forkCheckpointId === undefined) {
2182+ return yield* Effect.die(new Error("Stable fork source checkpoint missing."));
2183+ }
2184+ const checkpointForkCall = yield* invoke("t3_thread_fork", {
2185+ sourceThreadId: promptedThread.threadId,
2186+ sourcePoint: { type: "checkpoint", checkpointId: forkCheckpointId },
2187+ clientRequestId: "orchestrator-checkpoint-fork",
2188+ });
2189+ const checkpointFork = yield* decodeConversationForkResult(
2190+ checkpointForkCall.structuredContent,
2191+ ).pipe(Effect.orDie);
2192+ expect(checkpointFork.canonicalSourcePoint.runId).toBe(
2193+ forked.canonicalSourcePoint.runId,
2194+ );
2195+
2196+ yield* orchestrator.dispatch({
2197+ type: "thread.runtime-mode.set",
2198+ commandId: CommandId.make("command:mcp-fork:lower-caller-runtime"),
2199+ threadId: parentThreadId,
2200+ runtimeMode: "approval-required",
2201+ });
2202+ yield* orchestrator.dispatch({
2203+ type: "thread.interaction-mode.set",
2204+ commandId: CommandId.make("command:mcp-fork:lower-caller-interaction"),
2205+ threadId: parentThreadId,
2206+ interactionMode: "plan",
2207+ });
2208+
2209+ const repeatedForkCall = yield* invoke("t3_thread_fork", {
19612210 sourceThreadId: promptedThread.threadId,
1962− targetThreadId: forkedThreadId,
19632211 sourcePoint: { type: "latest_stable" },
19642212 title: "Inherited read thread",
2213+ clientRequestId: "orchestrator-inherited-read",
2214+ });
2215+ const repeatedFork = yield* decodeConversationForkResult(
2216+ repeatedForkCall.structuredContent,
2217+ ).pipe(Effect.orDie);
2218+ expect(repeatedFork.targetThreadId).toBe(forkedThreadId);
2219+ expect(repeatedFork.receipt).toEqual(forked.receipt);
2220+ yield* orchestrator.dispatch({
2221+ type: "thread.runtime-mode.set",
2222+ commandId: CommandId.make("command:mcp-fork:restore-caller-runtime"),
2223+ threadId: parentThreadId,
2224+ runtimeMode: "full-access",
19652225 });
2226+ yield* orchestrator.dispatch({
2227+ type: "thread.interaction-mode.set",
2228+ commandId: CommandId.make("command:mcp-fork:restore-caller-interaction"),
2229+ threadId: parentThreadId,
2230+ interactionMode: "default",
2231+ });
2232+
19662233 const forkedProjection = yield* orchestrator.getThreadProjection(forkedThreadId);
19672234 expect(forkedProjection.messages).toEqual([]);
19682235 expect(
@@ -1987,6 +2254,21 @@ describe("orchestrator MCP toolkit", () => { · 18 unmodified lines
19872254 creationSource: "mcp",
19882255 });
19892256
2257+ const transfersCall = yield* invoke("t3_thread_transfers", {
2258+ threadId: forkedThreadId,
2259+ type: "fork",
2260+ limit: 1,
2261+ });
2262+ const transfers = yield* decodeConversationTransferListResult(
2263+ transfersCall.structuredContent,
2264+ ).pipe(Effect.orDie);
2265+ expect(transfers).toMatchObject({
2266+ threadId: forkedThreadId,
2267+ scope: "current_project",
2268+ hasMore: false,
2269+ transfers: [{ id: forked.transfer.id, type: "fork", status: "pending" }],
2270+ });
2271+
19902272 const ordinaryLoopPrompt = "Run an ordinary thread loop iteration.";
19912273 const sendCall = yield* invoke("t3_thread_send", {
19922274 threadId: emptyThread.threadId,
apps/server/src/mcp/toolkits/orchestrator/handlers.ts+13−0
@@ -2,6 +2,7 @@ import { OrchestratorToolkit } from "./tools.ts"; · 1 unmodified line
22 import * as Effect from "effect/Effect";
33
44 import { McpInvocationContext } from "../../McpInvocationContext.ts";
5+import { ConversationTransferMcpService } from "../../ConversationTransferMcpService.ts";
56 import { OrchestratorMcpService } from "../../OrchestratorMcpService.ts";
67 import { ThreadMetadataMcpService } from "../../ThreadMetadataMcpService.ts";
78
@@ -116,6 +117,18 @@ const handlers = { · 108 unmodified lines
116117 const service = yield* OrchestratorMcpService;
117118 return yield* service.interruptThread(scope, input);
118119 }),
120+ t3_thread_transfers: (input) =>
121+ Effect.gen(function* () {
122+ const scope = yield* McpInvocationContext;
123+ const service = yield* ConversationTransferMcpService;
124+ return yield* service.list(scope, input);
125+ }),
126+ t3_thread_fork: (input) =>
127+ Effect.gen(function* () {
128+ const scope = yield* McpInvocationContext;
129+ const service = yield* ConversationTransferMcpService;
130+ return yield* service.fork(scope, input);
131+ }),
119132 } satisfies Parameters<typeof OrchestratorToolkit.toLayer>[0];
120133
121134 export const OrchestratorToolkitHandlersLive = OrchestratorToolkit.toLayer(handlers);
apps/server/src/mcp/toolkits/orchestrator/tools.test.ts+26−0
@@ -5,6 +5,8 @@ import { · 4 unmodified lines
55 CreateThreadsTool,
66 DelegateTaskTool,
77 ScheduleTaskTool,
8+ ThreadForkTool,
9+ ThreadTransfersTool,
810 ThreadUpdateTool,
911 } from "./tools.ts";
1012
@@ -77,4 +79,28 @@ describe("orchestrator MCP tool guidance", () => { · 66 unmodified lines
7779 ]);
7880 assert.include(ThreadUpdateTool.description ?? "", "Workspace and branch changes");
7981 });
82+
83+ it("publishes discoverable root-object schemas for conversation transfers", () => {
84+ const forkSchema = Tool.getJsonSchema(ThreadForkTool) as {
85+ readonly type?: unknown;
86+ readonly properties?: Readonly<Record<string, unknown>>;
87+ readonly required?: ReadonlyArray<string>;
88+ };
89+ const transfersSchema = Tool.getJsonSchema(ThreadTransfersTool) as {
90+ readonly type?: unknown;
91+ readonly properties?: Readonly<Record<string, unknown>>;
92+ };
93+
94+ assert.equal(forkSchema.type, "object");
95+ assert.hasAllKeys(forkSchema.properties ?? {}, [
96+ "sourceThreadId",
97+ "sourcePoint",
98+ "title",
99+ "clientRequestId",
100+ ]);
101+ assert.include(forkSchema.required ?? [], "sourcePoint");
102+ assert.include(forkSchema.required ?? [], "clientRequestId");
103+ assert.equal(transfersSchema.type, "object");
104+ assert.hasAllKeys(transfersSchema.properties ?? {}, ["threadId", "type", "limit"]);
105+ });
80106 });
apps/server/src/mcp/toolkits/orchestrator/tools.ts+37−0
@@ -1,4 +1,8 @@
11 import {
2+ ConversationForkInput,
3+ ConversationForkResult,
4+ ConversationTransferListInput,
5+ ConversationTransferListResult,
26 OrchestratorMcpCapabilitiesResult,
37 OrchestratorMcpCreatedThread,
48 OrchestratorMcpCreateThreadsInput,
@@ -32,6 +36,7 @@ import { · 27 unmodified lines
3236 import { Tool, Toolkit } from "effect/unstable/ai";
3337
3438 import * as McpInvocationContext from "../../McpInvocationContext.ts";
39+import { ConversationTransferMcpService } from "../../ConversationTransferMcpService.ts";
3540 import { OrchestratorMcpService } from "../../OrchestratorMcpService.ts";
3641 import { ThreadMetadataMcpService } from "../../ThreadMetadataMcpService.ts";
3742
@@ -40,6 +45,10 @@ const threadMetadataDependencies = [ · 2 unmodified lines
4045 McpInvocationContext.McpInvocationContext,
4146 ThreadMetadataMcpService,
4247 ];
48+const transferDependencies = [
49+ McpInvocationContext.McpInvocationContext,
50+ ConversationTransferMcpService,
51+];
4352
4453 export const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", {
4554 description:
@@ -249,6 +258,32 @@ export const ThreadInterruptTool = Tool.make("t3_thread_interrupt", { · 203 unmodified lines
249258 .annotate(Tool.Title, "Interrupt a T3 thread")
250259 .annotate(Tool.Destructive, true);
251260
261+export const ThreadTransfersTool = Tool.make("t3_thread_transfers", {
262+ description:
263+ "List up to 100 durable context transfers recorded on a T3 thread in the calling project, newest first. Omit threadId for this thread. Filter by transfer type to inspect fork, provider handoff, merge-back, or delegated-task provenance and resolution status.",
264+ parameters: ConversationTransferListInput,
265+ success: ConversationTransferListResult,
266+ failure: OrchestratorMcpFailure,
267+ failureMode: "return",
268+ dependencies: transferDependencies,
269+})
270+ .annotate(Tool.Title, "List thread context transfers")
271+ .annotate(Tool.Readonly, true)
272+ .annotate(Tool.Destructive, false)
273+ .annotate(Tool.Idempotent, true);
274+
275+export const ThreadForkTool = Tool.make("t3_thread_fork", {
276+ description:
277+ "Create a durable T3 conversation fork in the calling project from an explicit completed run, checkpoint, or latest stable run. Omit sourceThreadId for this thread. The new thread inherits the source configuration within the caller's permission ceiling. Native provider fork eligibility is reported now; native-fork or portable-context resolution happens truthfully on the first target turn. Reuse the required clientRequestId for retries.",
278+ parameters: ConversationForkInput,
279+ success: ConversationForkResult,
280+ failure: OrchestratorMcpFailure,
281+ failureMode: "return",
282+ dependencies: transferDependencies,
283+})
284+ .annotate(Tool.Title, "Fork a T3 conversation")
285+ .annotate(Tool.Destructive, true);
286+
252287 export const OrchestratorToolkit = Toolkit.make(
253288 OrchestratorCapabilitiesTool,
254289 DelegateTaskTool,
@@ -266,4 +301,6 @@ export const OrchestratorToolkit = Toolkit.make( · 11 unmodified lines
266301 ThreadSendTool,
267302 ThreadWaitTool,
268303 ThreadInterruptTool,
304+ ThreadTransfersTool,
305+ ThreadForkTool,
269306 );
apps/server/src/mcp/toolkits/worktree/registration.test.ts+7−0
@@ -114,6 +114,8 @@ it.effect("production mcp layer lists worktree tools over http", () => · 113 unmodified lines
114114 // than replacing them.
115115 expect(toolNames).toContain("preview_status");
116116 expect(toolNames).toContain("delegate_task");
117+ expect(toolNames).toContain("t3_thread_transfers");
118+ expect(toolNames).toContain("t3_thread_fork");
117119
118120 // The handoff tool mutates thread state, reaches the network (origin
119121 // fetch), and runs project setup scripts, so its MCP hints must not
@@ -125,6 +127,11 @@ it.effect("production mcp layer lists worktree tools over http", () => · 5 unmodified lines
125127 const status = tools.find((tool) => tool.name === "t3_worktree_status");
126128 expect(status?.annotations?.readOnlyHint).toBe(true);
127129 expect(status?.annotations?.destructiveHint).toBe(false);
130+ const transfers = tools.find((tool) => tool.name === "t3_thread_transfers");
131+ expect(transfers?.annotations?.readOnlyHint).toBe(true);
132+ expect(transfers?.annotations?.destructiveHint).toBe(false);
133+ const fork = tools.find((tool) => tool.name === "t3_thread_fork");
134+ expect(fork?.annotations?.destructiveHint).toBe(true);
128135
129136 // MCP requires every tool input schema to be a top-level object schema.
130137 // A non-object schema (e.g. the anyOf produced by an empty
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+1−0
@@ -807,6 +807,7 @@ export const CLAUDE_READ_ONLY_T3_MCP_ALLOWED_TOOLS: ReadonlyArray<string> = [ · 806 unmodified lines
807807 "mcp__t3-code__orchestrator_capabilities",
808808 "mcp__t3-code__list_scheduled_tasks",
809809 "mcp__t3-code__t3_thread_list",
810+ "mcp__t3-code__t3_thread_transfers",
810811 "mcp__t3-code__t3_thread_wait",
811812 ];
812813
apps/server/src/orchestration-v2/Orchestrator.ts+103−1
@@ -371,6 +371,23 @@ function nextQueuedRun( · 370 unmodified lines
371371 return queuedRunsInDeliveryOrder(projection)[0];
372372 }
373373
374+function runtimeModeRank(mode: OrchestrationV2AppThread["runtimeMode"]): number {
375+ switch (mode) {
376+ case "approval-required":
377+ return 0;
378+ case "auto-accept-edits":
379+ return 1;
380+ case "auto":
381+ return 2;
382+ case "full-access":
383+ return 3;
384+ }
385+}
386+
387+function interactionModeRank(mode: OrchestrationV2AppThread["interactionMode"]): number {
388+ return mode === "plan" ? 0 : 1;
389+}
390+
374391 function latestStableRun(projection: OrchestrationV2ThreadProjection): OrchestrationV2Run | null {
375392 return (
376393 projection.runs
@@ -2034,6 +2051,67 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 1657 unmodified lines
20342051 }),
20352052 ),
20362053 );
2054+ const targetIdentityExists = yield* projectionStore
2055+ .getThreadProjection(command.targetThreadId)
2056+ .pipe(
2057+ Effect.as(true),
2058+ Effect.catchTags({
2059+ ProjectionStoreThreadNotFoundError: () => Effect.succeed(false),
2060+ }),
2061+ Effect.mapError(
2062+ (cause) =>
2063+ new OrchestratorProjectionError({
2064+ threadId: command.targetThreadId,
2065+ cause,
2066+ }),
2067+ ),
2068+ );
2069+ if (targetIdentityExists) {
2070+ return yield* new OrchestratorDispatchError({
2071+ commandId: command.commandId,
2072+ commandType: command.type,
2073+ cause: `Fork target thread ${command.targetThreadId} already exists.`,
2074+ });
2075+ }
2076+
2077+ if (command.policyCeiling !== undefined) {
2078+ const callerProjection =
2079+ command.policyCeiling.callerThreadId === command.sourceThreadId
2080+ ? sourceProjection
2081+ : yield* projectionStore.getThreadProjection(command.policyCeiling.callerThreadId).pipe(
2082+ Effect.mapError(
2083+ (cause) =>
2084+ new OrchestratorProjectionError({
2085+ threadId: command.policyCeiling!.callerThreadId,
2086+ cause,
2087+ }),
2088+ ),
2089+ );
2090+ if (
2091+ runtimeModeRank(sourceProjection.thread.runtimeMode) >
2092+ runtimeModeRank(command.policyCeiling.runtimeMode) ||
2093+ runtimeModeRank(sourceProjection.thread.runtimeMode) >
2094+ runtimeModeRank(callerProjection.thread.runtimeMode)
2095+ ) {
2096+ return yield* new OrchestratorDispatchError({
2097+ commandId: command.commandId,
2098+ commandType: command.type,
2099+ cause: `Fork source runtime mode ${sourceProjection.thread.runtimeMode} exceeds the caller ceiling.`,
2100+ });
2101+ }
2102+ if (
2103+ interactionModeRank(sourceProjection.thread.interactionMode) >
2104+ interactionModeRank(command.policyCeiling.interactionMode) ||
2105+ interactionModeRank(sourceProjection.thread.interactionMode) >
2106+ interactionModeRank(callerProjection.thread.interactionMode)
2107+ ) {
2108+ return yield* new OrchestratorDispatchError({
2109+ commandId: command.commandId,
2110+ commandType: command.type,
2111+ cause: `Fork source interaction mode ${sourceProjection.thread.interactionMode} exceeds the caller ceiling.`,
2112+ });
2113+ }
2114+ }
20372115
20382116 const sourceRun = runForSourcePoint(sourceProjection, command.sourcePoint);
20392117
@@ -7183,8 +7261,32 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 5143 unmodified lines
71837261 } satisfies OrchestratorV2DispatchResult;
71847262 });
71857263
7264+ const dispatchLockKeys = (command: OrchestrationV2Command): ReadonlyArray<ThreadId> => {
7265+ const keys =
7266+ command.type === "thread.fork"
7267+ ? [
7268+ command.sourceThreadId,
7269+ command.targetThreadId,
7270+ ...(command.policyCeiling === undefined ? [] : [command.policyCeiling.callerThreadId]),
7271+ ]
7272+ : command.type === "thread.merge_back"
7273+ ? [command.sourceThreadId, command.targetThreadId]
7274+ : [commandThreadId(command)];
7275+ return [...new Set(keys)].toSorted((left, right) => (left < right ? -1 : left > right ? 1 : 0));
7276+ };
7277+
7278+ const withDispatchLocks = <A, E, R>(
7279+ keys: ReadonlyArray<ThreadId>,
7280+ effect: Effect.Effect<A, E, R>,
7281+ ): Effect.Effect<A, E, R> => {
7282+ const [key, ...remaining] = keys;
7283+ return key === undefined
7284+ ? effect
7285+ : threadDispatch.withLock(key, withDispatchLocks(remaining, effect));
7286+ };
7287+
71867288 const dispatchWithReceipt = (command: OrchestrationV2Command) =>
7187− threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command));
7289+ withDispatchLocks(dispatchLockKeys(command), dispatchWithReceiptEffect(command));
71887290
71897291 const handleTerminalRun = (stored: OrchestrationV2StoredEvent) =>
71907292 Effect.gen(function* () {
apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts+8−2
@@ -53,7 +53,10 @@ import { · 52 unmodified lines
5353 } from "../RuntimePolicy.ts";
5454 import { layer as turnItemPositionStoreLayer } from "../TurnItemPositionStore.ts";
5555 import { layer as runtimeRequestServiceLayer } from "../RuntimeRequestService.ts";
56−import { layer as threadForkServiceLayer } from "../ThreadForkService.ts";
56+import {
57+ layer as defaultThreadForkServiceLayer,
58+ ThreadForkServiceV2,
59+} from "../ThreadForkService.ts";
5760 import {
5861 runOrchestratorV2Scenario,
5962 type OrchestratorV2ScenarioStepError,
@@ -180,6 +183,7 @@ export function runOrchestratorV2ProviderReplayScenario< · 120 unmodified lines
180183 >;
181184 readonly enableLegacyTokenStreaming?: boolean;
182185 readonly runEffectWorker?: boolean;
186+ readonly threadForkServiceLayer?: Layer.Layer<ThreadForkServiceV2>;
183187 } = {},
184188 ): Effect.Effect<
185189 OrchestratorV2ScenarioResult,
@@ -220,6 +224,7 @@ export function makeOrchestratorV2ProviderReplayLayer< · 34 unmodified lines
220224 readonly enableLegacyTokenStreaming?: boolean;
221225 readonly runEffectWorker?: boolean;
222226 readonly replayGate?: ProviderReplayGate;
227+ readonly threadForkServiceLayer?: Layer.Layer<ThreadForkServiceV2>;
223228 } = {},
224229 ): Layer.Layer<OrchestratorV2, Error | MigrationError | PlatformError.PlatformError | SqlError> {
225230 const registryLayer = harness.makeProviderAdapterRegistryLayer(
@@ -239,6 +244,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry<Error>( · 13 unmodified lines
239244 >;
240245 readonly enableLegacyTokenStreaming?: boolean;
241246 readonly runEffectWorker?: boolean;
247+ readonly threadForkServiceLayer?: Layer.Layer<ThreadForkServiceV2>;
242248 } = {},
243249 ): Layer.Layer<OrchestratorV2, Error | MigrationError | PlatformError.PlatformError | SqlError> {
244250 const serverConfigLayer = Layer.effect(
@@ -392,7 +398,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry<Error>( · 147 unmodified lines
392398 providerSessionManagerProvided,
393399 providerSwitchServiceProvided,
394400 runExecutionServiceProvided,
395− threadForkServiceLayer,
401+ options.threadForkServiceLayer ?? defaultThreadForkServiceLayer,
396402 ),
397403 ),
398404 );
docs/orchestration-v2/orchestrator-mcp-server.md+18−0
@@ -327,6 +327,24 @@ Without `runId`, it selects the newest interruptible run. A terminal run is · 326 unmodified lines
327327 returned unchanged, and a thread with no active provider turn returns
328328 `no_active_run`.
329329
330+### Conversation forks and transfer discovery
331+
332+`t3_thread_fork` dispatches the existing `thread.fork` command through
333+`ThreadManagementService`, including its project admission and deleted-source
334+checks. The request names a stable source point and supplies an idempotency key;
335+the target thread ID and command ID are stable for that key. The response is
336+derived from the committed `thread.created` and `context-transfer.created`
337+events rather than an optional final projection read.
338+
339+Fork creation records lineage and a pending context transfer. Native eligibility
340+is reported from the source provider session and strong provider refs, but
341+resolution remains deferred to the first target turn. The existing provider
342+turn planner then chooses native fork or checks portable-context capabilities.
343+
344+`t3_thread_transfers` exposes a bounded, newest-first view of one current-project
345+thread's durable context-transfer records. It is the read path for fork
346+provenance and later resolution or consumption.
347+
330348 ## Delegated Task Lifecycle
331349
332350 The MCP server is a command ingress into V2. It does not call provider adapters
docs/user/conversation-forks.md+11−0
@@ -0,0 +1,11 @@
1+# Fork conversations with agents
2+
3+Agents connected through T3 Code's built-in MCP server can create a durable conversation fork without changing Git branches or workspaces.
4+
5+`t3_thread_fork` requires an explicit stable source point: the latest completed run, a specific completed run, or a checkpoint belonging to a completed run. Omitting `sourceThreadId` uses the current conversation. The new conversation stays in the same project and inherits the source conversation's provider selection and modes, limited by the calling agent's permission ceiling.
6+
7+The result identifies the source and new conversation, the requested and canonical source points, and the durable fork transfer. It also says whether a native provider fork is currently eligible. The transfer remains pending until the new conversation's first turn, when T3 Code chooses native fork or portable context using the active provider's capabilities. A pending result does not mean provider context has already moved.
8+
9+Use `t3_thread_transfers` to inspect fork and other context-transfer records on a conversation. Results are newest first and can be filtered by transfer type.
10+
11+`clientRequestId` is required for forks. Reusing it returns the original thread and command receipt instead of creating another fork.
packages/contracts/src/conversationTransferMcp.test.ts+38−0
@@ -0,0 +1,38 @@
1+import { expect, it } from "@effect/vitest";
2+import * as Effect from "effect/Effect";
3+import * as Result from "effect/Result";
4+import * as Schema from "effect/Schema";
5+
6+import { ConversationForkInput, ConversationTransferListInput } from "./conversationTransferMcp.ts";
7+
8+const decodeFork = Schema.decodeUnknownEffect(ConversationForkInput);
9+const decodeList = Schema.decodeUnknownEffect(ConversationTransferListInput);
10+
11+it.effect("requires an explicit stable fork source and a well-formed retry key", () =>
12+ Effect.gen(function* () {
13+ const valid = yield* decodeFork({
14+ sourcePoint: { type: "run", runId: "run:completed" },
15+ clientRequestId: "fork-completed-run",
16+ });
17+ const missingPoint = yield* Effect.result(
18+ decodeFork({ clientRequestId: "fork-without-point" }),
19+ );
20+ const malformedKey = yield* Effect.result(
21+ decodeFork({ sourcePoint: { type: "latest_stable" }, clientRequestId: "fork-\ud800" }),
22+ );
23+
24+ expect(valid.sourcePoint).toEqual({ type: "run", runId: "run:completed" });
25+ expect(Result.isFailure(missingPoint)).toBe(true);
26+ expect(Result.isFailure(malformedKey)).toBe(true);
27+ }),
28+);
29+
30+it.effect("bounds transfer result pages", () =>
31+ Effect.gen(function* () {
32+ const valid = yield* decodeList({ limit: 100 });
33+ const tooLarge = yield* Effect.result(decodeList({ limit: 101 }));
34+
35+ expect(valid.limit).toBe(100);
36+ expect(Result.isFailure(tooLarge)).toBe(true);
37+ }),
38+);
packages/contracts/src/conversationTransferMcp.ts+112−0
@@ -0,0 +1,112 @@
1+import * as Schema from "effect/Schema";
2+
3+import { CommandId, NonNegativeInt, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts";
4+import {
5+ OrchestrationV2ContextSourcePoint,
6+ OrchestrationV2ContextTransfer,
7+ OrchestrationV2ContextTransferType,
8+ OrchestrationV2ThreadForkSourcePoint,
9+} from "./orchestrationV2.ts";
10+import { ProviderInstanceId } from "./providerInstance.ts";
11+
12+function hasWellFormedUnicode(value: string): boolean {
13+ for (let index = 0; index < value.length; index += 1) {
14+ const code = value.charCodeAt(index);
15+ if (code >= 0xd800 && code <= 0xdbff) {
16+ if (index + 1 >= value.length) return false;
17+ const next = value.charCodeAt(index + 1);
18+ if (next < 0xdc00 || next > 0xdfff) return false;
19+ index += 1;
20+ } else if (code >= 0xdc00 && code <= 0xdfff) {
21+ return false;
22+ }
23+ }
24+ return true;
25+}
26+
27+const WellFormedRequestKey = Schema.String.check(
28+ Schema.makeFilter(
29+ (value: string) =>
30+ (value.length > 0 && value === value.trim()) ||
31+ "The request key must be non-empty and must not have surrounding whitespace.",
32+ ),
33+ Schema.isMaxLength(256),
34+ Schema.makeFilter(
35+ (value: string) =>
36+ hasWellFormedUnicode(value) ||
37+ "The request key must not contain unpaired Unicode surrogates.",
38+ ),
39+).annotate({ description: "Stable idempotency key to reuse when retrying this mutation." });
40+
41+const TransferPageLimit = Schema.Int.check(Schema.isBetween({ minimum: 1, maximum: 100 }));
42+
43+export const ConversationTransferListInput = Schema.Struct({
44+ threadId: Schema.optional(
45+ ThreadId.annotate({
46+ description: "Thread in the calling project; defaults to this thread.",
47+ }),
48+ ),
49+ type: Schema.optional(OrchestrationV2ContextTransferType),
50+ limit: Schema.optional(TransferPageLimit),
51+});
52+export type ConversationTransferListInput = typeof ConversationTransferListInput.Type;
53+
54+export const ConversationTransferListResult = Schema.Struct({
55+ threadId: ThreadId,
56+ scope: Schema.Literal("current_project"),
57+ transfers: Schema.Array(OrchestrationV2ContextTransfer),
58+ hasMore: Schema.Boolean,
59+});
60+export type ConversationTransferListResult = typeof ConversationTransferListResult.Type;
61+
62+export const ConversationForkInput = Schema.Struct({
63+ sourceThreadId: Schema.optional(
64+ ThreadId.annotate({
65+ description: "Source thread in the calling project; defaults to this thread.",
66+ }),
67+ ),
68+ sourcePoint: OrchestrationV2ThreadForkSourcePoint.annotate({
69+ description:
70+ "Explicit stable source: the latest completed run, one completed run id, or a checkpoint on a completed run.",
71+ }),
72+ title: Schema.optional(TrimmedNonEmptyString),
73+ clientRequestId: WellFormedRequestKey,
74+});
75+export type ConversationForkInput = typeof ConversationForkInput.Type;
76+
77+export const ConversationForkNativeEligibility = Schema.Literals([
78+ "eligible",
79+ "provider_does_not_support_fork",
80+ "provider_does_not_support_turn_fork",
81+ "strong_native_source_unavailable",
82+ "source_provider_session_unavailable",
83+]);
84+export type ConversationForkNativeEligibility = typeof ConversationForkNativeEligibility.Type;
85+
86+export const ConversationForkProviderSupport = Schema.Struct({
87+ sourceProviderInstanceId: Schema.NullOr(ProviderInstanceId),
88+ nativeForkEligibility: ConversationForkNativeEligibility,
89+ resolutionTiming: Schema.Literal("first_target_turn"),
90+ fallback: Schema.Literal("provider_context_capabilities_checked_on_first_target_turn"),
91+});
92+export type ConversationForkProviderSupport = typeof ConversationForkProviderSupport.Type;
93+
94+export const ConversationTransferReceipt = Schema.Struct({
95+ commandId: CommandId,
96+ commandType: Schema.Literals(["thread.fork", "thread.merge_back"]),
97+ sequence: NonNegativeInt,
98+ eventIds: Schema.Array(Schema.String),
99+});
100+export type ConversationTransferReceipt = typeof ConversationTransferReceipt.Type;
101+
102+export const ConversationForkResult = Schema.Struct({
103+ sourceThreadId: ThreadId,
104+ targetThreadId: ThreadId,
105+ scope: Schema.Literal("conversation_through_source_point"),
106+ requestedSourcePoint: OrchestrationV2ThreadForkSourcePoint,
107+ canonicalSourcePoint: OrchestrationV2ContextSourcePoint,
108+ transfer: OrchestrationV2ContextTransfer,
109+ providerSupport: ConversationForkProviderSupport,
110+ receipt: ConversationTransferReceipt,
111+});
112+export type ConversationForkResult = typeof ConversationForkResult.Type;
packages/contracts/src/index.ts+1−0
@@ -33,6 +33,7 @@ export * from "./orchestrationProject.ts"; · 32 unmodified lines
3333 export * from "./orchestrationV2.ts";
3434 export * from "./applicationEvent.ts";
3535 export * from "./orchestratorMcp.ts";
36+export * from "./conversationTransferMcp.ts";
3637 export * from "./threadMetadataMcp.ts";
3738 export * from "./orchestration.ts";
3839 export * from "./t3ProjectFile.ts";
packages/contracts/src/orchestrationV2.ts+8−0
@@ -2283,6 +2283,14 @@ export const OrchestrationV2Command = Schema.Union([ · 2282 unmodified lines
22832283 targetThreadId: ThreadId,
22842284 sourcePoint: OrchestrationV2ThreadForkSourcePoint,
22852285 title: Schema.optional(TrimmedNonEmptyString),
2286+ /** Optional caller ceiling enforced with the current source and caller projections. */
2287+ policyCeiling: Schema.optional(
2288+ Schema.Struct({
2289+ callerThreadId: ThreadId,
2290+ runtimeMode: RuntimeMode,
2291+ interactionMode: ProviderInteractionMode,
2292+ }),
2293+ ),
22862294 createdAt: Schema.optional(Schema.DateTimeUtc),
22872295 }),
22882296 Schema.Struct({
packages/shared/src/t3McpToolPresentation.ts+2−0
@@ -49,6 +49,8 @@ const T3_MCP_TOOLS: Record< · 48 unmodified lines
4949 t3_thread_list: { displayName: "List T3 threads", summaryAction: "thread-list" },
5050 t3_thread_read: { displayName: "Read a T3 thread", summaryAction: "thread-read" },
5151 t3_thread_update: { displayName: "Update T3 thread metadata" },
52+ t3_thread_transfers: { displayName: "List T3 thread context transfers" },
53+ t3_thread_fork: { displayName: "Fork a T3 conversation" },
5254 t3_thread_send: { displayName: "Send to a T3 thread", summaryAction: "thread-send" },
5355 t3_thread_wait: { displayName: "Wait for a T3 thread", summaryAction: "thread-wait" },
5456 t3_thread_interrupt: { displayName: "Interrupt a T3 thread", summaryAction: "thread-interrupt" },