From 9efa23daab739b797dd42fdec081b10c86067f8e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 16:21:36 -0700 Subject: [PATCH 1/3] feat(mcp): merge back conversation context --- .../ConversationTransferMcpService.test.ts | 53 +++- .../src/mcp/ConversationTransferMcpService.ts | 234 +++++++++++++++++- ...OrchestratorMcpToolkit.integration.test.ts | 233 ++++++++++++++++- .../src/mcp/toolkits/orchestrator/handlers.ts | 6 + .../mcp/toolkits/orchestrator/tools.test.ts | 15 ++ .../src/mcp/toolkits/orchestrator/tools.ts | 15 ++ .../toolkits/worktree/registration.test.ts | 8 + .../src/orchestration-v2/Orchestrator.ts | 50 +++- .../src/orchestration-v2/runtimeLayer.ts | 1 + .../testkit/ProviderReplayHarness.ts | 39 ++- .../orchestrator-mcp-server.md | 10 + docs/user/conversation-forks.md | 4 + .../src/conversationTransferMcp.test.ts | 24 +- .../contracts/src/conversationTransferMcp.ts | 33 +++ packages/contracts/src/orchestrationV2.ts | 8 + packages/shared/src/t3McpToolPresentation.ts | 1 + 16 files changed, 708 insertions(+), 26 deletions(-) diff --git a/apps/server/src/mcp/ConversationTransferMcpService.test.ts b/apps/server/src/mcp/ConversationTransferMcpService.test.ts index ae66c7bc37f8..722a5b4760f5 100644 --- a/apps/server/src/mcp/ConversationTransferMcpService.test.ts +++ b/apps/server/src/mcp/ConversationTransferMcpService.test.ts @@ -9,9 +9,11 @@ import { } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import { OrchestratorProjectionError } from "../orchestration-v2/Orchestrator.ts"; +import { CommandReceiptStoreV2 } from "../orchestration-v2/CommandReceiptStore.ts"; import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; import * as ConversationTransfer from "./ConversationTransferMcpService.ts"; @@ -31,6 +33,11 @@ function projection(input: { projectId, runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: input.threadId, + }, }, runs: [], contextTransfers: [], @@ -52,15 +59,21 @@ function testLayer(input: { readonly source: OrchestrationV2ThreadProjection; readonly dispatch: ThreadManagementService["Service"]["dispatch"]; readonly getThreadProjection?: ThreadManagementService["Service"]["getThreadProjection"]; + readonly getReceipt?: CommandReceiptStoreV2["Service"]["getByCommandId"]; }) { return ConversationTransfer.layer.pipe( Layer.provide( - Layer.mock(ThreadManagementService)({ - getThreadProjection: input.getThreadProjection ?? (() => Effect.succeed(input.parent)), - getProjectThread: ({ threadId }) => - Effect.succeed(threadId === parentThreadId ? input.parent : input.source), - dispatch: input.dispatch, - }), + Layer.mergeAll( + Layer.mock(ThreadManagementService)({ + getThreadProjection: input.getThreadProjection ?? (() => Effect.succeed(input.parent)), + getProjectThread: ({ threadId }) => + Effect.succeed(threadId === parentThreadId ? input.parent : input.source), + dispatch: input.dispatch, + }), + Layer.mock(CommandReceiptStoreV2)({ + getByCommandId: input.getReceipt ?? (() => Effect.succeed(Option.none())), + }), + ), ), ); } @@ -218,4 +231,32 @@ describe("ConversationTransferMcpService", () => { assert.equal(yield* Ref.get(dispatched), 0); }), ); + + it.effect("rejects merge-back from a thread without fork provenance", () => + Effect.gen(function* () { + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const error = yield* Effect.gen(function* () { + const service = yield* ConversationTransfer.ConversationTransferMcpService; + return yield* service + .mergeBack(scope(), { + sourcePoint: { type: "latest_stable" }, + clientRequestId: "not-a-fork", + }) + .pipe(Effect.flip); + }).pipe( + Effect.provide( + testLayer({ + parent, + source: parent, + dispatch: () => Effect.die("dispatch should not run"), + }), + ), + ); + assert.equal(error.code, "invalid_request"); + }), + ); }); diff --git a/apps/server/src/mcp/ConversationTransferMcpService.ts b/apps/server/src/mcp/ConversationTransferMcpService.ts index 0329ad29e446..85270e1ac68f 100644 --- a/apps/server/src/mcp/ConversationTransferMcpService.ts +++ b/apps/server/src/mcp/ConversationTransferMcpService.ts @@ -3,6 +3,8 @@ import { type ConversationForkInput, type ConversationForkNativeEligibility, type ConversationForkResult, + type ConversationMergeBackInput, + type ConversationMergeBackResult, type ConversationTransferListInput, type ConversationTransferListResult, type ConversationTransferReceipt, @@ -23,6 +25,7 @@ import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; +import { CommandReceiptStoreV2 } from "../orchestration-v2/CommandReceiptStore.ts"; import type { OrchestratorV2DispatchResult } from "../orchestration-v2/Orchestrator.ts"; import { ThreadManagementService, @@ -41,6 +44,10 @@ export class ConversationTransferMcpService extends Context.Service< scope: McpInvocationScope, input: ConversationForkInput, ) => Effect.Effect; + readonly mergeBack: ( + scope: McpInvocationScope, + input: ConversationMergeBackInput, + ) => Effect.Effect; } >()("t3/mcp/ConversationTransferMcpService") {} @@ -95,6 +102,12 @@ function forkCommandId(scope: McpInvocationScope, requestKey: string): CommandId ); } +function mergeBackCommandId(scope: McpInvocationScope, requestKey: string): CommandId { + return CommandId.make( + `command:mcp:${stablePart(scope.providerSessionId)}:thread-merge-back:${stablePart(requestKey)}`, + ); +} + function sourceRun( projection: OrchestrationV2ThreadProjection, point: OrchestrationV2ThreadForkSourcePoint, @@ -140,10 +153,14 @@ export function conversationForkNativeEligibility( return "eligible"; } -function receipt(result: OrchestratorV2DispatchResult, commandId: CommandId) { +function receipt( + result: OrchestratorV2DispatchResult, + commandId: CommandId, + commandType: "thread.fork" | "thread.merge_back", +) { return { commandId, - commandType: "thread.fork", + commandType, sequence: result.sequence, eventIds: result.storedEvents.map((stored) => stored.event.id), } satisfies ConversationTransferReceipt; @@ -167,6 +184,7 @@ function forkTransfer( export const make = Effect.gen(function* () { const threads = yield* ThreadManagementService; + const commandReceipts = yield* CommandReceiptStoreV2; const requireCapability = (scope: McpInvocationScope) => scope.capabilities.has("orchestration") @@ -242,17 +260,62 @@ export const make = Effect.gen(function* () { resolutionTiming: "first_target_turn", fallback: "provider_context_capabilities_checked_on_first_target_turn", }, - receipt: receipt(input.dispatched, input.commandId), + receipt: receipt(input.dispatched, input.commandId, "thread.fork"), } satisfies ConversationForkResult; }, ); + const mergeBackResultFromDispatch = Effect.fn( + "ConversationTransferMcpService.mergeBackResultFromDispatch", + )(function* (input: { + readonly request: ConversationMergeBackInput; + readonly commandId: CommandId; + readonly dispatched: OrchestratorV2DispatchResult; + }) { + let transfer: OrchestrationV2ContextTransfer | undefined; + const supersededTransferIds: Array = []; + for (const stored of input.dispatched.storedEvents) { + if ( + stored.event.type === "context-transfer.created" && + stored.event.payload.type === "merge_back" + ) { + transfer = stored.event.payload; + } + if ( + stored.event.type === "context-transfer.updated" && + stored.event.payload.type === "merge_back" && + stored.event.payload.status === "superseded" + ) { + supersededTransferIds.push(stored.event.payload.id); + } + } + if (transfer === undefined || transfer.basePoint === null) { + return yield* failure( + "orchestration_error", + "The merge-back command committed without returning its provenance transfer event.", + ); + } + return { + sourceThreadId: transfer.sourceThreadId, + targetThreadId: transfer.targetThreadId, + scope: "fork_delta_through_source_point", + requestedSourcePoint: input.request.sourcePoint, + canonicalSourcePoint: transfer.sourcePoint, + basePoint: transfer.basePoint, + transfer, + supersededTransferIds, + consumption: "next_target_turn", + receipt: receipt(input.dispatched, input.commandId, "thread.merge_back"), + } satisfies ConversationMergeBackResult; + }); + return ConversationTransferMcpService.of({ list: (scope, input) => Effect.gen(function* () { const { target } = yield* loadScoped(scope, input.threadId); const limit = input.limit ?? 50; const matching = target.contextTransfers + .toReversed() .filter((transfer) => input.type === undefined || transfer.type === input.type) .toSorted( (left, right) => @@ -381,8 +444,169 @@ export const make = Effect.gen(function* () { source, }); }), + mergeBack: (scope, input) => + Effect.gen(function* () { + yield* requireCapability(scope); + const commandId = mergeBackCommandId(scope, input.clientRequestId); + const priorReceipt = yield* commandReceipts.getByCommandId(commandId).pipe( + Effect.map( + Option.filter( + (existing) => + existing.status === "accepted" && existing.commandType === "thread.merge_back", + ), + ), + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to inspect the merge-back retry receipt: ${errorMessage(error)}`, + ), + ), + ); + if (Option.isSome(priorReceipt)) { + const caller = yield* threads + .getThreadProjection(scope.threadId) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to load the calling thread: ${errorMessage(error)}`, + ), + ), + ); + const recordedTarget = yield* threads + .getThreadProjection(priorReceipt.value.threadId) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to load the accepted merge-back target: ${errorMessage(error)}`, + ), + ), + ); + if (recordedTarget.thread.projectId !== caller.thread.projectId) { + return yield* failure( + "thread_not_found", + `The accepted merge-back target is not in the calling project.`, + ); + } + const replayed = yield* threads + .dispatch({ + type: "thread.merge_back", + createdBy: "agent", + creationSource: "mcp", + commandId, + sourceThreadId: input.sourceThreadId ?? scope.threadId, + targetThreadId: priorReceipt.value.threadId, + sourcePoint: input.sourcePoint, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to replay the merge-back receipt: ${errorMessage(error)}`, + ), + ), + ); + return yield* mergeBackResultFromDispatch({ + request: input, + commandId, + dispatched: replayed, + }); + } + + const { parent, target: source } = yield* loadScoped(scope, input.sourceThreadId); + const recordedTargetId = source.thread.lineage.parentThreadId; + if (source.thread.lineage.relationshipToParent !== "fork" || recordedTargetId === null) { + return yield* failure( + "invalid_request", + `Thread ${source.thread.id} is not a conversation fork and cannot merge back.`, + ); + } + const requestedTargetId = input.targetThreadId ?? recordedTargetId; + if (requestedTargetId !== recordedTargetId) { + return yield* failure( + "invalid_request", + `Thread ${source.thread.id} can merge back only to its recorded parent ${recordedTargetId}.`, + ); + } + const { target } = yield* loadScoped(scope, requestedTargetId); + if ( + runtimeModeRank(target.thread.runtimeMode) > runtimeModeRank(parent.thread.runtimeMode) + ) { + return yield* failure( + "runtime_mode_escalation_denied", + `The target thread's ${target.thread.runtimeMode} runtime mode exceeds the calling thread's ${parent.thread.runtimeMode} ceiling.`, + ); + } + if ( + interactionModeRank(target.thread.interactionMode) > + interactionModeRank(parent.thread.interactionMode) + ) { + return yield* failure( + "interaction_mode_escalation_denied", + `The target thread's ${target.thread.interactionMode} interaction mode exceeds the calling thread's ${parent.thread.interactionMode} ceiling.`, + ); + } + const run = sourceRun(source, input.sourcePoint); + if (run === undefined) { + return yield* failure( + "run_not_found", + `No run in thread ${source.thread.id} matches source point ${input.sourcePoint.type}.`, + ); + } + if (run.status !== "completed" && run.status !== "waiting") { + return yield* failure( + "invalid_request", + `Merge-back source run ${run.id} is ${run.status}; only provider-finished runs are supported.`, + ); + } + const forkTransfer = source.contextTransfers.findLast( + (transfer) => + transfer.type === "fork" && + transfer.sourceThreadId === target.thread.id && + transfer.targetThreadId === source.thread.id, + ); + if (forkTransfer === undefined) { + return yield* failure( + "invalid_request", + `No fork transfer connects ${target.thread.id} to ${source.thread.id}.`, + ); + } + + const dispatched = yield* threads + .dispatch({ + type: "thread.merge_back", + createdBy: "agent", + creationSource: "mcp", + commandId, + sourceThreadId: source.thread.id, + targetThreadId: target.thread.id, + sourcePoint: input.sourcePoint, + policyCeiling: { + callerThreadId: parent.thread.id, + runtimeMode: parent.thread.runtimeMode, + interactionMode: parent.thread.interactionMode, + }, + }) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to queue conversation merge-back: ${errorMessage(error)}`, + ), + ), + ); + return yield* mergeBackResultFromDispatch({ + request: input, + commandId, + dispatched, + }); + }), }); }); -export const layer: Layer.Layer = - Layer.effect(ConversationTransferMcpService, make); +export const layer: Layer.Layer< + ConversationTransferMcpService, + never, + CommandReceiptStoreV2 | ThreadManagementService +> = Layer.effect(ConversationTransferMcpService, make); diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 2cf612cf9d6d..3c77112a0507 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "@effect/vitest"; import { CommandId, ConversationForkResult, + ConversationMergeBackResult, ConversationTransferListResult, EnvironmentId, IsoDateTime, @@ -99,6 +100,7 @@ const queuedFollowupResult = "Queued delegated follow-up completed."; const decodeCreateThreadsResult = Schema.decodeUnknownEffect(OrchestratorMcpCreateThreadsResult); const decodeConversationForkResult = Schema.decodeUnknownEffect(ConversationForkResult); +const decodeConversationMergeBackResult = Schema.decodeUnknownEffect(ConversationMergeBackResult); const decodeConversationTransferListResult = Schema.decodeUnknownEffect( ConversationTransferListResult, ); @@ -401,7 +403,39 @@ function makeDeterministicAdapter(input: { readThreadSnapshot: () => unsupported(input.driver, "readThreadSnapshot is unused in this test"), rollbackThread: () => unsupported(input.driver, "rollbackThread is unused in this test"), - forkThread: () => unsupported(input.driver, "forkThread is unused in this test"), + forkThread: (forkInput) => + DateTime.now.pipe( + Effect.map( + (createdAt) => + ({ + ...forkInput.sourceProviderThread, + id: ProviderThreadId.make( + `provider-thread:${input.instanceId}:${forkInput.targetThreadId}:fork`, + ), + providerSessionId: sessionInput.providerSessionId, + appThreadId: forkInput.targetThreadId, + ownerNodeId: forkInput.ownerNodeId ?? null, + nativeThreadRef: { + driver: input.driver, + nativeId: `native-fork:${forkInput.targetThreadId}`, + strength: "strong", + }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: { + providerThreadId: forkInput.sourceProviderThread.id, + ...(forkInput.providerTurnId === undefined + ? {} + : { providerTurnId: forkInput.providerTurnId }), + }, + createdAt, + updatedAt: createdAt, + }) satisfies OrchestrationV2ProviderThread, + ), + ), }; }), }; @@ -2269,6 +2303,203 @@ describe("orchestrator MCP toolkit", () => { transfers: [{ id: forked.transfer.id, type: "fork", status: "pending" }], }); + const forkSendCall = yield* invoke("t3_thread_send", { + threadId: forkedThreadId, + message: "Produce the result that will merge back.", + clientRequestId: "fork-result-for-merge-back", + }); + expect(forkSendCall.isError).toBe(false); + const forkSend = yield* decodeThreadSendResult(forkSendCall.structuredContent).pipe( + Effect.orDie, + ); + const forkWaitCall = yield* invoke("t3_thread_wait", { + threadId: forkedThreadId, + runId: forkSend.runId, + timeoutMs: 10_000, + }); + const forkWait = yield* decodeThreadWaitResult(forkWaitCall.structuredContent).pipe( + Effect.orDie, + ); + expect(forkWait.status).toBe("completed"); + + const mergeBackInput = { + sourceThreadId: forkedThreadId, + targetThreadId: promptedThread.threadId, + sourcePoint: { type: "latest_stable" }, + clientRequestId: "merge-fork-result-back", + } as const; + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-merge:lower-caller-runtime"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-merge:lower-caller-interaction"), + threadId: parentThreadId, + interactionMode: "plan", + }); + const deniedMergeBackCall = yield* invoke("t3_thread_merge_back", { + ...mergeBackInput, + clientRequestId: "merge-fork-result-back-denied-ceiling", + }); + expect(deniedMergeBackCall.structuredContent).toMatchObject({ + _tag: "OrchestratorMcpFailure", + code: "runtime_mode_escalation_denied", + }); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-merge:restore-caller-runtime"), + threadId: parentThreadId, + runtimeMode: "full-access", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-merge:restore-caller-interaction"), + threadId: parentThreadId, + interactionMode: "default", + }); + const deniedMergeAtMutation = yield* orchestrator + .dispatch({ + type: "thread.merge_back", + createdBy: "agent", + creationSource: "mcp", + commandId: CommandId.make("command:mcp-merge:denied-policy-ceiling"), + sourceThreadId: forkedThreadId, + targetThreadId: promptedThread.threadId, + sourcePoint: { type: "latest_stable" }, + policyCeiling: { + callerThreadId: parentThreadId, + runtimeMode: "approval-required", + interactionMode: "plan", + }, + }) + .pipe(Effect.flip); + expect(deniedMergeAtMutation._tag).toBe("OrchestratorDispatchError"); + const mergeBackCall = yield* invoke("t3_thread_merge_back", mergeBackInput); + expect(mergeBackCall.isError).toBe(false); + const mergeBack = yield* decodeConversationMergeBackResult( + mergeBackCall.structuredContent, + ).pipe(Effect.orDie); + expect(mergeBack).toMatchObject({ + sourceThreadId: forkedThreadId, + targetThreadId: promptedThread.threadId, + scope: "fork_delta_through_source_point", + requestedSourcePoint: { type: "latest_stable" }, + transfer: { type: "merge_back", status: "pending" }, + consumption: "next_target_turn", + receipt: { commandType: "thread.merge_back" }, + }); + expect(mergeBack.basePoint.runId).toBe(forked.canonicalSourcePoint.runId); + + const repeatedMergeBackCall = yield* invoke("t3_thread_merge_back", mergeBackInput); + const repeatedMergeBack = yield* decodeConversationMergeBackResult( + repeatedMergeBackCall.structuredContent, + ).pipe(Effect.orDie); + expect(repeatedMergeBack.transfer.id).toBe(mergeBack.transfer.id); + expect(repeatedMergeBack.receipt).toEqual(mergeBack.receipt); + + const replacementMergeCall = yield* invoke("t3_thread_merge_back", { + ...mergeBackInput, + clientRequestId: "merge-fork-result-back-replacement", + }); + const replacementMerge = yield* decodeConversationMergeBackResult( + replacementMergeCall.structuredContent, + ).pipe(Effect.orDie); + expect(replacementMerge.transfer.id).not.toBe(mergeBack.transfer.id); + expect(replacementMerge.supersededTransferIds).toContain(mergeBack.transfer.id); + + const mergeTransfersCall = yield* invoke("t3_thread_transfers", { + threadId: promptedThread.threadId, + type: "merge_back", + limit: 1, + }); + const mergeTransfers = yield* decodeConversationTransferListResult( + mergeTransfersCall.structuredContent, + ).pipe(Effect.orDie); + expect(mergeTransfers.transfers[0]).toMatchObject({ + id: replacementMerge.transfer.id, + sourceThreadId: forkedThreadId, + targetThreadId: promptedThread.threadId, + status: "pending", + }); + + const consumeMergeText = "Consume the selected merge-back delta."; + const consumeMergeCall = yield* invoke("t3_thread_send", { + threadId: promptedThread.threadId, + message: consumeMergeText, + clientRequestId: "consume-selected-merge-back", + }); + const consumeMerge = yield* decodeThreadSendResult( + consumeMergeCall.structuredContent, + ).pipe(Effect.orDie); + const consumeMergeWaitCall = yield* invoke("t3_thread_wait", { + threadId: promptedThread.threadId, + runId: consumeMerge.runId, + timeoutMs: 10_000, + }); + const consumeMergeWait = yield* decodeThreadWaitResult( + consumeMergeWaitCall.structuredContent, + ).pipe(Effect.orDie); + expect(consumeMergeWait.status).toBe("completed"); + const consumedTurn = (yield* Ref.get(capturedTurns)).findLast( + (turn) => + turn.threadId === promptedThread.threadId && turn.text.includes(consumeMergeText), + ); + expect(consumedTurn?.text).toContain( + "Context handoff (merge_back / fork_delta_summary):", + ); + expect(consumedTurn?.text).toContain( + "Claude completed: Produce the result that will merge back.", + ); + const targetAfterMergeConsumption = yield* orchestrator.getThreadProjection( + promptedThread.threadId, + ); + expect( + targetAfterMergeConsumption.contextTransfers.find( + (transfer) => transfer.id === replacementMerge.transfer.id, + )?.status, + ).toBe("consumed"); + expect( + targetAfterMergeConsumption.contextTransfers.find( + (transfer) => transfer.id === mergeBack.transfer.id, + )?.status, + ).toBe("superseded"); + expect( + targetAfterMergeConsumption.contextHandoffs.some( + (handoff) => handoff.transferId === replacementMerge.transfer.id, + ), + ).toBe(true); + expect( + targetAfterMergeConsumption.contextHandoffs.some( + (handoff) => handoff.transferId === mergeBack.transfer.id, + ), + ).toBe(false); + + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make("command:mcp-merge:delete-source-after-acceptance"), + threadId: forkedThreadId, + }); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:mcp-merge:archive-target-after-consumption"), + threadId: promptedThread.threadId, + }); + const lifecycleReplayCall = yield* invoke("t3_thread_merge_back", mergeBackInput); + expect(lifecycleReplayCall.isError).toBe(false); + const lifecycleReplay = yield* decodeConversationMergeBackResult( + lifecycleReplayCall.structuredContent, + ).pipe(Effect.orDie); + expect(lifecycleReplay.transfer.id).toBe(mergeBack.transfer.id); + expect(lifecycleReplay.receipt).toEqual(mergeBack.receipt); + yield* orchestrator.dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("command:mcp-merge:restore-target-after-replay"), + threadId: promptedThread.threadId, + }); + const ordinaryLoopPrompt = "Run an ordinary thread loop iteration."; const sendCall = yield* invoke("t3_thread_send", { threadId: emptyThread.threadId, diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index 658cc9a3bb3d..03e72798c449 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -129,6 +129,12 @@ const handlers = { const service = yield* ConversationTransferMcpService; return yield* service.fork(scope, input); }), + t3_thread_merge_back: (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const service = yield* ConversationTransferMcpService; + return yield* service.mergeBack(scope, input); + }), } satisfies Parameters[0]; export const OrchestratorToolkitHandlersLive = OrchestratorToolkit.toLayer(handlers); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts index 40c28b712947..cd03c6635660 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts @@ -6,6 +6,7 @@ import { DelegateTaskTool, ScheduleTaskTool, ThreadForkTool, + ThreadMergeBackTool, ThreadTransfersTool, ThreadUpdateTool, } from "./tools.ts"; @@ -90,6 +91,11 @@ describe("orchestrator MCP tool guidance", () => { readonly type?: unknown; readonly properties?: Readonly>; }; + const mergeSchema = Tool.getJsonSchema(ThreadMergeBackTool) as { + readonly type?: unknown; + readonly properties?: Readonly>; + readonly required?: ReadonlyArray; + }; assert.equal(forkSchema.type, "object"); assert.hasAllKeys(forkSchema.properties ?? {}, [ @@ -102,5 +108,14 @@ describe("orchestrator MCP tool guidance", () => { assert.include(forkSchema.required ?? [], "clientRequestId"); assert.equal(transfersSchema.type, "object"); assert.hasAllKeys(transfersSchema.properties ?? {}, ["threadId", "type", "limit"]); + assert.equal(mergeSchema.type, "object"); + assert.hasAllKeys(mergeSchema.properties ?? {}, [ + "sourceThreadId", + "targetThreadId", + "sourcePoint", + "clientRequestId", + ]); + assert.include(mergeSchema.required ?? [], "sourcePoint"); + assert.include(mergeSchema.required ?? [], "clientRequestId"); }); }); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 471a06de6d7a..0881dd58f1ab 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -1,6 +1,8 @@ import { ConversationForkInput, ConversationForkResult, + ConversationMergeBackInput, + ConversationMergeBackResult, ConversationTransferListInput, ConversationTransferListResult, OrchestratorMcpCapabilitiesResult, @@ -284,6 +286,18 @@ export const ThreadForkTool = Tool.make("t3_thread_fork", { .annotate(Tool.Title, "Fork a T3 conversation") .annotate(Tool.Destructive, true); +export const ThreadMergeBackTool = Tool.make("t3_thread_merge_back", { + description: + "Queue a fork conversation's provider-finished result for context merge-back into its original parent thread. This changes T3 conversation context, not Git branches or workspaces. Omit sourceThreadId when called from the fork and targetThreadId to use its recorded parent. The durable result reports source, target, fork-base provenance, superseded pending transfers, and next-target-turn consumption. Reuse the required clientRequestId for retries.", + parameters: ConversationMergeBackInput, + success: ConversationMergeBackResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: transferDependencies, +}) + .annotate(Tool.Title, "Merge back conversation context") + .annotate(Tool.Destructive, true); + export const OrchestratorToolkit = Toolkit.make( OrchestratorCapabilitiesTool, DelegateTaskTool, @@ -303,4 +317,5 @@ export const OrchestratorToolkit = Toolkit.make( ThreadInterruptTool, ThreadTransfersTool, ThreadForkTool, + ThreadMergeBackTool, ); diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 4cb2e24bb0d7..b76a1a8c8f93 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -4,11 +4,13 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { HttpBody, HttpClient, HttpRouter } from "effect/unstable/http"; import * as ServerEnvironment from "../../../environment/ServerEnvironment.ts"; import * as GitWorkflowService from "../../../git/GitWorkflowService.ts"; +import { CommandReceiptStoreV2 } from "../../../orchestration-v2/CommandReceiptStore.ts"; import { ThreadManagementService } from "../../../orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "../../../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../../../project/ProjectSetupScriptRunner.ts"; @@ -22,6 +24,9 @@ import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts"; const StubServicesLive = Layer.mergeAll( Layer.mock(ThreadManagementService)({}), + Layer.mock(CommandReceiptStoreV2)({ + getByCommandId: () => Effect.succeed(Option.none()), + }), Layer.mock(ProviderRegistry)({}), Layer.mock(ScheduledTaskService)({}), Layer.mock(ProjectService.ProjectService)({}), @@ -116,6 +121,7 @@ it.effect("production mcp layer lists worktree tools over http", () => expect(toolNames).toContain("delegate_task"); expect(toolNames).toContain("t3_thread_transfers"); expect(toolNames).toContain("t3_thread_fork"); + expect(toolNames).toContain("t3_thread_merge_back"); // The handoff tool mutates thread state, reaches the network (origin // fetch), and runs project setup scripts, so its MCP hints must not @@ -132,6 +138,8 @@ it.effect("production mcp layer lists worktree tools over http", () => expect(transfers?.annotations?.destructiveHint).toBe(false); const fork = tools.find((tool) => tool.name === "t3_thread_fork"); expect(fork?.annotations?.destructiveHint).toBe(true); + const mergeBack = tools.find((tool) => tool.name === "t3_thread_merge_back"); + expect(mergeBack?.annotations?.destructiveHint).toBe(true); // MCP requires every tool input schema to be a top-level object schema. // A non-object schema (e.g. the anyOf produced by an empty diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 93cd00f6cbba..00b43a14c6f9 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2205,6 +2205,48 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); + if (command.policyCeiling !== undefined) { + const ceiling = command.policyCeiling; + const callerProjection = + ceiling.callerThreadId === command.sourceThreadId + ? sourceProjection + : ceiling.callerThreadId === command.targetThreadId + ? targetProjection + : yield* projectionStore.getThreadProjection(ceiling.callerThreadId).pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: ceiling.callerThreadId, + cause, + }), + ), + ); + if ( + runtimeModeRank(targetProjection.thread.runtimeMode) > + runtimeModeRank(ceiling.runtimeMode) || + runtimeModeRank(targetProjection.thread.runtimeMode) > + runtimeModeRank(callerProjection.thread.runtimeMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Merge-back target runtime mode ${targetProjection.thread.runtimeMode} exceeds the caller ceiling.`, + }); + } + if ( + interactionModeRank(targetProjection.thread.interactionMode) > + interactionModeRank(ceiling.interactionMode) || + interactionModeRank(targetProjection.thread.interactionMode) > + interactionModeRank(callerProjection.thread.interactionMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Merge-back target interaction mode ${targetProjection.thread.interactionMode} exceeds the caller ceiling.`, + }); + } + } + if ( sourceProjection.thread.lineage.relationshipToParent !== "fork" || sourceProjection.thread.lineage.parentThreadId !== command.targetThreadId @@ -7270,7 +7312,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.policyCeiling === undefined ? [] : [command.policyCeiling.callerThreadId]), ] : command.type === "thread.merge_back" - ? [command.sourceThreadId, command.targetThreadId] + ? [ + command.sourceThreadId, + command.targetThreadId, + ...(command.policyCeiling === undefined + ? [] + : [command.policyCeiling.callerThreadId]), + ] : [commandThreadId(command)]; return [...new Set(keys)].toSorted((left, right) => (left < right ? -1 : left > right ? 1 : 0)); }; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 637916e7f9bf..698ef42d3c13 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -261,6 +261,7 @@ const providerRuntimeRecoveryProvided = providerRuntimeRecoveryLayer.pipe( ); export const OrchestrationV2LayerLive = Layer.mergeAll( + commandReceiptStoreProvided, orchestratorProvided, threadManagementProvided, effectWorkerProvided, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 94dd1ff78fd6..d1cbd083f48b 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -21,7 +21,10 @@ import { layer as checkpointCaptureServiceLayer } from "../CheckpointCaptureServ import { layer as checkpointServiceLayer } from "../CheckpointService.ts"; import { layer as checkpointRollbackServiceLayer } from "../CheckpointRollbackService.ts"; import { layer as commandPolicyLayer } from "../CommandPolicy.ts"; -import { layer as commandReceiptStoreLayer } from "../CommandReceiptStore.ts"; +import { + CommandReceiptStoreV2, + layer as commandReceiptStoreLayer, +} from "../CommandReceiptStore.ts"; import { layer as contextHandoffServiceLayer } from "../ContextHandoffService.ts"; import { layer as effectOutboxLayer } from "../EffectOutbox.ts"; import { @@ -226,7 +229,10 @@ export function makeOrchestratorV2ProviderReplayLayer< readonly replayGate?: ProviderReplayGate; readonly threadForkServiceLayer?: Layer.Layer; } = {}, -): Layer.Layer { +): Layer.Layer< + CommandReceiptStoreV2 | OrchestratorV2, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const registryLayer = harness.makeProviderAdapterRegistryLayer( scenario.transcript, options.replayGate === undefined ? {} : { replayGate: options.replayGate }, @@ -246,7 +252,10 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( readonly runEffectWorker?: boolean; readonly threadForkServiceLayer?: Layer.Layer; } = {}, -): Layer.Layer { +): Layer.Layer< + CommandReceiptStoreV2 | OrchestratorV2, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const serverConfigLayer = Layer.effect( ServerConfig, makeReplayServerConfig(scenario.name).pipe(Effect.orDie), @@ -411,14 +420,20 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( // orchestrator. Keeping this acquisition in the replay layer makes the // outbox lifecycle explicit and prevents test-only command-side draining. if (options.runEffectWorker === false) { - return orchestratorProvided.pipe(Layer.provide(NodeServices.layer)); + return Layer.merge( + orchestratorProvided.pipe(Layer.provide(NodeServices.layer)), + commandReceiptStoreProvided, + ); } - return Layer.effect( - OrchestratorV2, - Effect.gen(function* () { - const orchestrator = yield* OrchestratorV2; - yield* runEffectWorkerDaemon.pipe(Effect.forkScoped); - return orchestrator; - }), - ).pipe(Layer.provide(replayRuntime)); + return Layer.merge( + Layer.effect( + OrchestratorV2, + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + yield* runEffectWorkerDaemon.pipe(Effect.forkScoped); + return orchestrator; + }), + ).pipe(Layer.provide(replayRuntime)), + commandReceiptStoreProvided, + ); } diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 29b2b4ffb0b4..1ac6a1953a1f 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -345,6 +345,16 @@ turn planner then chooses native fork or checks portable-context capabilities. thread's durable context-transfer records. It is the read path for fork provenance and later resolution or consumption. +`t3_thread_merge_back` validates that the source is a fork of the selected +target and that its explicit source run is provider-finished before dispatching +`thread.merge_back`. The committed transfer records both the selected result +point and the original fork base. It stays pending until the target's next turn; +a newer pending merge-back for the same pair supersedes the older transfer. +The target runtime and interaction modes may not exceed the calling thread's +captured or current ceiling. Stable request keys replay the same accepted +command receipt and transfer before re-evaluating mutable lineage, lifecycle, or +run state. This operation does not invoke Git or change workspace state. + ## Delegated Task Lifecycle The MCP server is a command ingress into V2. It does not call provider adapters diff --git a/docs/user/conversation-forks.md b/docs/user/conversation-forks.md index a46575933613..f85f6f72d26b 100644 --- a/docs/user/conversation-forks.md +++ b/docs/user/conversation-forks.md @@ -9,3 +9,7 @@ The result identifies the source and new conversation, the requested and canonic 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. `clientRequestId` is required for forks. Reusing it returns the original thread and command receipt instead of creating another fork. + +After the fork produces a result, `t3_thread_merge_back` can queue its conversation delta for the original parent. This is a conversation-context operation, not a Git merge, and it never changes branches or workspaces. The source must be the recorded fork, the target must be its original parent, and the source point must identify a provider-finished run. The result includes the fork base, the selected result point, and a durable receipt. + +The merge-back target cannot have broader runtime or interaction permissions than the calling conversation. The parent consumes pending merge-back context on its next turn. Creating a newer merge-back before consumption supersedes the older pending transfer. Retrying an accepted request with the same `clientRequestId` returns the original transfer and does not supersede it, even if the source or target lifecycle has since changed. diff --git a/packages/contracts/src/conversationTransferMcp.test.ts b/packages/contracts/src/conversationTransferMcp.test.ts index 50d240dca2c9..9809b46a280d 100644 --- a/packages/contracts/src/conversationTransferMcp.test.ts +++ b/packages/contracts/src/conversationTransferMcp.test.ts @@ -3,10 +3,15 @@ import * as Effect from "effect/Effect"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; -import { ConversationForkInput, ConversationTransferListInput } from "./conversationTransferMcp.ts"; +import { + ConversationForkInput, + ConversationMergeBackInput, + ConversationTransferListInput, +} from "./conversationTransferMcp.ts"; const decodeFork = Schema.decodeUnknownEffect(ConversationForkInput); const decodeList = Schema.decodeUnknownEffect(ConversationTransferListInput); +const decodeMergeBack = Schema.decodeUnknownEffect(ConversationMergeBackInput); it.effect("requires an explicit stable fork source and a well-formed retry key", () => Effect.gen(function* () { @@ -36,3 +41,20 @@ it.effect("bounds transfer result pages", () => expect(Result.isFailure(tooLarge)).toBe(true); }), ); + +it.effect("keeps merge-back source and target explicit when supplied", () => + Effect.gen(function* () { + const input = yield* decodeMergeBack({ + sourceThreadId: "thread:fork", + targetThreadId: "thread:parent", + sourcePoint: { type: "checkpoint", checkpointId: "checkpoint:fork-result" }, + clientRequestId: "merge-fork-result", + }); + + expect(input).toMatchObject({ + sourceThreadId: "thread:fork", + targetThreadId: "thread:parent", + sourcePoint: { type: "checkpoint", checkpointId: "checkpoint:fork-result" }, + }); + }), +); diff --git a/packages/contracts/src/conversationTransferMcp.ts b/packages/contracts/src/conversationTransferMcp.ts index a9fe2caac695..723d0ad470d9 100644 --- a/packages/contracts/src/conversationTransferMcp.ts +++ b/packages/contracts/src/conversationTransferMcp.ts @@ -110,3 +110,36 @@ export const ConversationForkResult = Schema.Struct({ receipt: ConversationTransferReceipt, }); export type ConversationForkResult = typeof ConversationForkResult.Type; + +export const ConversationMergeBackInput = Schema.Struct({ + sourceThreadId: Schema.optional( + ThreadId.annotate({ + description: "Fork thread containing the result; defaults to this thread.", + }), + ), + targetThreadId: Schema.optional( + ThreadId.annotate({ + description: "Original parent thread; defaults to the fork's recorded parent.", + }), + ), + sourcePoint: OrchestrationV2ThreadForkSourcePoint.annotate({ + description: + "Explicit provider-finished source: the latest completed run, one completed or waiting run id, or its checkpoint.", + }), + clientRequestId: WellFormedRequestKey, +}); +export type ConversationMergeBackInput = typeof ConversationMergeBackInput.Type; + +export const ConversationMergeBackResult = Schema.Struct({ + sourceThreadId: ThreadId, + targetThreadId: ThreadId, + scope: Schema.Literal("fork_delta_through_source_point"), + requestedSourcePoint: OrchestrationV2ThreadForkSourcePoint, + canonicalSourcePoint: OrchestrationV2ContextSourcePoint, + basePoint: OrchestrationV2ContextSourcePoint, + transfer: OrchestrationV2ContextTransfer, + supersededTransferIds: Schema.Array(OrchestrationV2ContextTransfer.fields.id), + consumption: Schema.Literal("next_target_turn"), + receipt: ConversationTransferReceipt, +}); +export type ConversationMergeBackResult = typeof ConversationMergeBackResult.Type; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index fc23a597a195..f0aea1f7ee61 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2300,6 +2300,14 @@ export const OrchestrationV2Command = Schema.Union([ sourceThreadId: ThreadId, targetThreadId: ThreadId, sourcePoint: OrchestrationV2ThreadForkSourcePoint, + /** Optional caller ceiling enforced with the current target and caller projections. */ + policyCeiling: Schema.optional( + Schema.Struct({ + callerThreadId: ThreadId, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + }), + ), createdAt: Schema.optional(Schema.DateTimeUtc), }), Schema.Struct({ diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 5553e29cc0da..93e67e8f99cd 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -51,6 +51,7 @@ const T3_MCP_TOOLS: Record< t3_thread_update: { displayName: "Update T3 thread metadata" }, t3_thread_transfers: { displayName: "List T3 thread context transfers" }, t3_thread_fork: { displayName: "Fork a T3 conversation" }, + t3_thread_merge_back: { displayName: "Merge back T3 conversation context" }, t3_thread_send: { displayName: "Send to a T3 thread", summaryAction: "thread-send" }, t3_thread_wait: { displayName: "Wait for a T3 thread", summaryAction: "thread-wait" }, t3_thread_interrupt: { displayName: "Interrupt a T3 thread", summaryAction: "thread-interrupt" }, From 073dd597e1b5e732980fe8593e7160031d8e9963 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 10:40:30 -0700 Subject: [PATCH 2/3] fix(mcp): enforce merge-back source ceiling --- .../ConversationTransferMcpService.test.ts | 23 ++++++++++- .../src/mcp/ConversationTransferMcpService.ts | 17 +++++++++ ...OrchestratorMcpToolkit.integration.test.ts | 38 +++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 24 ++++++++++++ packages/contracts/src/orchestrationV2.ts | 2 +- 5 files changed, 102 insertions(+), 2 deletions(-) diff --git a/apps/server/src/mcp/ConversationTransferMcpService.test.ts b/apps/server/src/mcp/ConversationTransferMcpService.test.ts index 722a5b4760f5..4346cbbd8ee8 100644 --- a/apps/server/src/mcp/ConversationTransferMcpService.test.ts +++ b/apps/server/src/mcp/ConversationTransferMcpService.test.ts @@ -107,7 +107,7 @@ describe("ConversationTransferMcpService", () => { ); }); - it.effect("rejects missing stable runs and inherited permission escalation before dispatch", () => + it.effect("rejects missing stable runs and inherited transfer escalation before dispatch", () => Effect.gen(function* () { const dispatched = yield* Ref.make(0); const parent = projection({ @@ -141,6 +141,27 @@ describe("ConversationTransferMcpService", () => { ); assert.equal(escalation.code, "runtime_mode_escalation_denied"); + const mergeEscalation = yield* Effect.gen(function* () { + const service = yield* ConversationTransfer.ConversationTransferMcpService; + return yield* service + .mergeBack(scope(), { + sourceThreadId, + sourcePoint: { type: "latest_stable" }, + clientRequestId: "merge-permission-ceiling", + }) + .pipe(Effect.flip); + }).pipe( + Effect.provide( + testLayer({ + parent, + source, + dispatch: () => + Ref.update(dispatched, (count) => count + 1).pipe(Effect.as({} as never)), + }), + ), + ); + assert.equal(mergeEscalation.code, "runtime_mode_escalation_denied"); + const noRun = yield* Effect.gen(function* () { const service = yield* ConversationTransfer.ConversationTransferMcpService; return yield* service diff --git a/apps/server/src/mcp/ConversationTransferMcpService.ts b/apps/server/src/mcp/ConversationTransferMcpService.ts index 85270e1ac68f..ffd3d89cc768 100644 --- a/apps/server/src/mcp/ConversationTransferMcpService.ts +++ b/apps/server/src/mcp/ConversationTransferMcpService.ts @@ -515,6 +515,23 @@ export const make = Effect.gen(function* () { } const { parent, target: source } = yield* loadScoped(scope, input.sourceThreadId); + if ( + runtimeModeRank(source.thread.runtimeMode) > runtimeModeRank(parent.thread.runtimeMode) + ) { + return yield* failure( + "runtime_mode_escalation_denied", + `The source thread's ${source.thread.runtimeMode} runtime mode exceeds the calling thread's ${parent.thread.runtimeMode} ceiling.`, + ); + } + if ( + interactionModeRank(source.thread.interactionMode) > + interactionModeRank(parent.thread.interactionMode) + ) { + return yield* failure( + "interaction_mode_escalation_denied", + `The source thread's ${source.thread.interactionMode} interaction mode exceeds the calling thread's ${parent.thread.interactionMode} ceiling.`, + ); + } const recordedTargetId = source.thread.lineage.parentThreadId; if (source.thread.lineage.relationshipToParent !== "fork" || recordedTargetId === null) { return yield* failure( diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 3c77112a0507..f3edaa0e6b98 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -2340,6 +2340,18 @@ describe("orchestrator MCP toolkit", () => { threadId: parentThreadId, interactionMode: "plan", }); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-merge:lower-target-runtime"), + threadId: promptedThread.threadId, + runtimeMode: "approval-required", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-merge:lower-target-interaction"), + threadId: promptedThread.threadId, + interactionMode: "plan", + }); const deniedMergeBackCall = yield* invoke("t3_thread_merge_back", { ...mergeBackInput, clientRequestId: "merge-fork-result-back-denied-ceiling", @@ -2347,6 +2359,7 @@ describe("orchestrator MCP toolkit", () => { expect(deniedMergeBackCall.structuredContent).toMatchObject({ _tag: "OrchestratorMcpFailure", code: "runtime_mode_escalation_denied", + message: expect.stringContaining("source thread"), }); yield* orchestrator.dispatch({ type: "thread.runtime-mode.set", @@ -2377,6 +2390,19 @@ describe("orchestrator MCP toolkit", () => { }) .pipe(Effect.flip); expect(deniedMergeAtMutation._tag).toBe("OrchestratorDispatchError"); + expect(String(deniedMergeAtMutation.cause)).toContain("Merge-back source runtime mode"); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-merge:restore-target-runtime"), + threadId: promptedThread.threadId, + runtimeMode: "full-access", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-merge:restore-target-interaction"), + threadId: promptedThread.threadId, + interactionMode: "default", + }); const mergeBackCall = yield* invoke("t3_thread_merge_back", mergeBackInput); expect(mergeBackCall.isError).toBe(false); const mergeBack = yield* decodeConversationMergeBackResult( @@ -2393,12 +2419,24 @@ describe("orchestrator MCP toolkit", () => { }); expect(mergeBack.basePoint.runId).toBe(forked.canonicalSourcePoint.runId); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-merge:lower-caller-after-acceptance"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); const repeatedMergeBackCall = yield* invoke("t3_thread_merge_back", mergeBackInput); const repeatedMergeBack = yield* decodeConversationMergeBackResult( repeatedMergeBackCall.structuredContent, ).pipe(Effect.orDie); expect(repeatedMergeBack.transfer.id).toBe(mergeBack.transfer.id); expect(repeatedMergeBack.receipt).toEqual(mergeBack.receipt); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-merge:restore-caller-after-acceptance"), + threadId: parentThreadId, + runtimeMode: "full-access", + }); const replacementMergeCall = yield* invoke("t3_thread_merge_back", { ...mergeBackInput, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 00b43a14c6f9..170a5192aa5d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2221,6 +2221,30 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); + if ( + runtimeModeRank(sourceProjection.thread.runtimeMode) > + runtimeModeRank(ceiling.runtimeMode) || + runtimeModeRank(sourceProjection.thread.runtimeMode) > + runtimeModeRank(callerProjection.thread.runtimeMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Merge-back source runtime mode ${sourceProjection.thread.runtimeMode} exceeds the caller ceiling.`, + }); + } + if ( + interactionModeRank(sourceProjection.thread.interactionMode) > + interactionModeRank(ceiling.interactionMode) || + interactionModeRank(sourceProjection.thread.interactionMode) > + interactionModeRank(callerProjection.thread.interactionMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Merge-back source interaction mode ${sourceProjection.thread.interactionMode} exceeds the caller ceiling.`, + }); + } if ( runtimeModeRank(targetProjection.thread.runtimeMode) > runtimeModeRank(ceiling.runtimeMode) || diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index f0aea1f7ee61..1f54a4fd2927 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2300,7 +2300,7 @@ export const OrchestrationV2Command = Schema.Union([ sourceThreadId: ThreadId, targetThreadId: ThreadId, sourcePoint: OrchestrationV2ThreadForkSourcePoint, - /** Optional caller ceiling enforced with the current target and caller projections. */ + /** Optional caller ceiling enforced with the current source, target, and caller projections. */ policyCeiling: Schema.optional( Schema.Struct({ callerThreadId: ThreadId, From afe96811882382ce92e911edbe21b6251a1ffee3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 10:43:04 -0700 Subject: [PATCH 3/3] test(mcp): type merge-back denial fixture --- apps/server/src/mcp/ConversationTransferMcpService.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/src/mcp/ConversationTransferMcpService.test.ts b/apps/server/src/mcp/ConversationTransferMcpService.test.ts index 4346cbbd8ee8..059d881c325d 100644 --- a/apps/server/src/mcp/ConversationTransferMcpService.test.ts +++ b/apps/server/src/mcp/ConversationTransferMcpService.test.ts @@ -156,7 +156,9 @@ describe("ConversationTransferMcpService", () => { parent, source, dispatch: () => - Ref.update(dispatched, (count) => count + 1).pipe(Effect.as({} as never)), + Ref.update(dispatched, (count) => count + 1).pipe( + Effect.andThen(Effect.die("merge-back dispatch should not run")), + ), }), ), );