# Files changed · PR \#8697 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/8697/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [feat\(mcp\): merge back conversation context](https://git.cafe/pingdotgg/t3code/pull/8697.md) Version: 3 Head commit: afe96811882382ce92e911edbe21b6251a1ffee3 Base commit: 1ad23c2032b0aa1ce9a55e4d50ce73cd5f6fc20a Comparison base: 1ad23c2032b0aa1ce9a55e4d50ce73cd5f6fc20a Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/mcp/ConversationTransferMcpService.test.ts Status: modified; +71; −7 ```diff @@ -9,9 +9,11 @@ } 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 @@ projectId, runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: input.threadId, + }, }, runs: [], contextTransfers: [], @@ -52,15 +59,21 @@ 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)({ + 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())), + }), + ), - getThreadProjection: input.getThreadProjection ?? (() => Effect.succeed(input.parent)), - getProjectThread: ({ threadId }) => - Effect.succeed(threadId === parentThreadId ? input.parent : input.source), - dispatch: input.dispatch, - }), ), ); } @@ -94,7 +107,7 @@ ); }); - 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({ @@ -128,6 +141,29 @@ ); 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.andThen(Effect.die("merge-back dispatch should not run")), + ), + }), + ), + ); + assert.equal(mergeEscalation.code, "runtime_mode_escalation_denied"); + const noRun = yield* Effect.gen(function* () { const service = yield* ConversationTransfer.ConversationTransferMcpService; return yield* service @@ -218,4 +254,32 @@ 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"); + }), + ); }); ``` ## apps/server/src/mcp/ConversationTransferMcpService.ts Status: modified; +246; −5 ```diff @@ -3,6 +3,8 @@ type ConversationForkInput, type ConversationForkNativeEligibility, type ConversationForkResult, + type ConversationMergeBackInput, + type ConversationMergeBackResult, type ConversationTransferListInput, type ConversationTransferListResult, type ConversationTransferReceipt, @@ -23,6 +25,7 @@ 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 @@ scope: McpInvocationScope, input: ConversationForkInput, ) => Effect.Effect; + readonly mergeBack: ( + scope: McpInvocationScope, + input: ConversationMergeBackInput, + ) => Effect.Effect; } >()("t3/mcp/ConversationTransferMcpService") {} @@ -95,6 +102,12 @@ ); } +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 @@ 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 @@ 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 @@ 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,186 @@ 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); + 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( + "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< + ConversationTransferMcpService, + never, + CommandReceiptStoreV2 | ThreadManagementService -export const layer: Layer.Layer = +> = Layer.effect(ConversationTransferMcpService, make); - Layer.effect(ConversationTransferMcpService, make); ``` ## apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts Status: modified; +270; −1 ```diff @@ -3,6 +3,7 @@ import { CommandId, ConversationForkResult, + ConversationMergeBackResult, ConversationTransferListResult, EnvironmentId, IsoDateTime, @@ -99,6 +100,7 @@ const decodeCreateThreadsResult = Schema.decodeUnknownEffect(OrchestratorMcpCreateThreadsResult); const decodeConversationForkResult = Schema.decodeUnknownEffect(ConversationForkResult); +const decodeConversationMergeBackResult = Schema.decodeUnknownEffect(ConversationMergeBackResult); const decodeConversationTransferListResult = Schema.decodeUnknownEffect( ConversationTransferListResult, ); @@ -401,7 +403,39 @@ readThreadSnapshot: () => unsupported(input.driver, "readThreadSnapshot is unused in this test"), rollbackThread: () => unsupported(input.driver, "rollbackThread is unused in this test"), + forkThread: (forkInput) => + DateTime.now.pipe( + Effect.map( + (createdAt) => + ({ + ...forkInput.sourceProviderThread, + id: ProviderThreadId.make( - forkThread: () => unsupported(input.driver, "forkThread is unused in this test"), + `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,241 @@ 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", + }); + 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", + }); + expect(deniedMergeBackCall.structuredContent).toMatchObject({ + _tag: "OrchestratorMcpFailure", + code: "runtime_mode_escalation_denied", + message: expect.stringContaining("source thread"), + }); + 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"); + 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( + 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); + + 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, + 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, ``` ## apps/server/src/mcp/toolkits/orchestrator/handlers.ts Status: modified; +6; −0 ```diff @@ -129,6 +129,12 @@ 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); ``` ## apps/server/src/mcp/toolkits/orchestrator/tools.test.ts Status: modified; +15; −0 ```diff @@ -6,6 +6,7 @@ DelegateTaskTool, ScheduleTaskTool, ThreadForkTool, + ThreadMergeBackTool, ThreadTransfersTool, ThreadUpdateTool, } from "./tools.ts"; @@ -90,6 +91,11 @@ 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 @@ 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"); }); }); ``` ## apps/server/src/mcp/toolkits/orchestrator/tools.ts Status: modified; +15; −0 ```diff @@ -1,6 +1,8 @@ import { ConversationForkInput, ConversationForkResult, + ConversationMergeBackInput, + ConversationMergeBackResult, ConversationTransferListInput, ConversationTransferListResult, OrchestratorMcpCapabilitiesResult, @@ -284,6 +286,18 @@ .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 @@ ThreadInterruptTool, ThreadTransfersTool, ThreadForkTool, + ThreadMergeBackTool, ); ``` ## apps/server/src/mcp/toolkits/worktree/registration.test.ts Status: modified; +8; −0 ```diff @@ -4,11 +4,13 @@ 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 @@ 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 @@ 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 @@ 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 ``` ## apps/server/src/orchestration-v2/Orchestrator.ts Status: modified; +73; −1 ```diff @@ -2205,6 +2205,72 @@ ), ); + 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(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) || + 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 +7336,13 @@ ...(command.policyCeiling === undefined ? [] : [command.policyCeiling.callerThreadId]), ] : command.type === "thread.merge_back" + ? [ + command.sourceThreadId, + command.targetThreadId, + ...(command.policyCeiling === undefined - ? [command.sourceThreadId, command.targetThreadId] + ? [] + : [command.policyCeiling.callerThreadId]), + ] : [commandThreadId(command)]; return [...new Set(keys)].toSorted((left, right) => (left < right ? -1 : left > right ? 1 : 0)); }; ``` ## apps/server/src/orchestration-v2/runtimeLayer.ts Status: modified; +1; −0 ```diff @@ -261,6 +261,7 @@ ); export const OrchestrationV2LayerLive = Layer.mergeAll( + commandReceiptStoreProvided, orchestratorProvided, threadManagementProvided, effectWorkerProvided, ``` ## apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts Status: modified; +27; −12 ```diff @@ -21,7 +21,10 @@ import { layer as checkpointServiceLayer } from "../CheckpointService.ts"; import { layer as checkpointRollbackServiceLayer } from "../CheckpointRollbackService.ts"; import { layer as commandPolicyLayer } from "../CommandPolicy.ts"; +import { + CommandReceiptStoreV2, + layer as commandReceiptStoreLayer, -import { layer as commandReceiptStoreLayer } from "../CommandReceiptStore.ts"; +} from "../CommandReceiptStore.ts"; import { layer as contextHandoffServiceLayer } from "../ContextHandoffService.ts"; import { layer as effectOutboxLayer } from "../EffectOutbox.ts"; import { @@ -226,7 +229,10 @@ readonly replayGate?: ProviderReplayGate; readonly threadForkServiceLayer?: Layer.Layer; } = {}, +): Layer.Layer< + CommandReceiptStoreV2 | OrchestratorV2, + Error | MigrationError | PlatformError.PlatformError | SqlError -): Layer.Layer { +> { const registryLayer = harness.makeProviderAdapterRegistryLayer( scenario.transcript, options.replayGate === undefined ? {} : { replayGate: options.replayGate }, @@ -246,7 +252,10 @@ readonly runEffectWorker?: boolean; readonly threadForkServiceLayer?: Layer.Layer; } = {}, +): Layer.Layer< + CommandReceiptStoreV2 | OrchestratorV2, + Error | MigrationError | PlatformError.PlatformError | SqlError -): Layer.Layer { +> { const serverConfigLayer = Layer.effect( ServerConfig, makeReplayServerConfig(scenario.name).pipe(Effect.orDie), @@ -411,14 +420,20 @@ // 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 Layer.merge( + orchestratorProvided.pipe(Layer.provide(NodeServices.layer)), + commandReceiptStoreProvided, - return orchestratorProvided.pipe(Layer.provide(NodeServices.layer)); + ); } + 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, - return Layer.effect( + ); - OrchestratorV2, - Effect.gen(function* () { - const orchestrator = yield* OrchestratorV2; - yield* runEffectWorkerDaemon.pipe(Effect.forkScoped); - return orchestrator; - }), - ).pipe(Layer.provide(replayRuntime)); } ``` ## docs/orchestration-v2/orchestrator-mcp-server.md Status: modified; +10; −0 ```diff @@ -345,6 +345,16 @@ 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 ``` ## docs/user/conversation-forks.md Status: modified; +4; −0 ```diff @@ -9,3 +9,7 @@ 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. ``` ## packages/contracts/src/conversationTransferMcp.test.ts Status: modified; +23; −1 ```diff @@ -3,10 +3,15 @@ import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; +import { + ConversationForkInput, -import { ConversationForkInput, ConversationTransferListInput } from "./conversationTransferMcp.ts"; + 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 @@ 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" }, + }); + }), +); ``` ## packages/contracts/src/conversationTransferMcp.ts Status: modified; +33; −0 ```diff @@ -110,3 +110,36 @@ 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; ``` ## packages/contracts/src/orchestrationV2.ts Status: modified; +8; −0 ```diff @@ -2300,6 +2300,14 @@ sourceThreadId: ThreadId, targetThreadId: ThreadId, sourcePoint: OrchestrationV2ThreadForkSourcePoint, + /** Optional caller ceiling enforced with the current source, target, and caller projections. */ + policyCeiling: Schema.optional( + Schema.Struct({ + callerThreadId: ThreadId, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + }), + ), createdAt: Schema.optional(Schema.DateTimeUtc), }), Schema.Struct({ ``` ## packages/shared/src/t3McpToolPresentation.ts Status: modified; +1; −0 ```diff @@ -51,6 +51,7 @@ 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" }, ```