fix(orchestration): resume continuations after planned worktree handoffs

apps/server/src/mcp/WorktreeMcpService.test.ts

@@ -4,14 +4,18 @@ import * as NodeServices from "@effect/platform-node/NodeServices";
 import {
   CommandId,
   EnvironmentId,
+  EventId,
+  RunId,
   type OrchestrationV2ThreadProjection,
+  type OrchestrationV2ServerCommand,
   type Project,
   ProjectId,
   ProviderInstanceId,
   ThreadId,
   WorktreeMcpHandoffInput,
 } from "@t3tools/contracts";
 import * as Cause from "effect/Cause";
+import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
@@ -28,6 +32,7 @@ import * as ServerConfig from "../config.ts";
 import {
   OrchestratorDispatchError,
   OrchestratorProjectionError,
+  type OrchestratorV2DispatchResult,
 } from "../orchestration-v2/Orchestrator.ts";
 import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
 import * as ProjectService from "../project/ProjectService.ts";
@@ -106,7 +111,7 @@ interface HarnessOptions {
   readonly threadAttachedOnRecheck?: boolean;
   readonly threadArchivedOnRecheck?: boolean;
   readonly threadReadFailsOnRecheck?: boolean;
-  readonly continuation?: "queued" | "fails" | "dies";
+  readonly continuation?: "queued" | "started";
   readonly projectMissing?: boolean;
   readonly projectReadFails?: boolean;
   readonly existingBranchWorktreePath?: string | null;
@@ -122,7 +127,7 @@ interface HarnessOptions {
 const makeHarness = (options: HarnessOptions = {}) => {
   const thread = options.thread === undefined ? {} : options.thread;
   const scope = makeScope(options.capabilities ?? new Set(["preview", "worktree"]));
-  const dispatch = vi.fn((_: unknown) =>
+  const dispatch = vi.fn((command: OrchestrationV2ServerCommand) =>
     (options.dispatchGate ?? Effect.void).pipe(
       Effect.andThen(
         options.dispatchInterrupts
@@ -131,7 +136,45 @@ const makeHarness = (options: HarnessOptions = {}) => {
             ? Effect.die(new Error("dispatch defect"))
             : options.dispatchFails
               ? (Effect.fail("simulated dispatch failure") as never)
-              : Effect.succeed({ sequence: 1, storedEvents: [] }),
+              : Effect.succeed({
+                  sequence: 1,
+                  storedEvents:
+                    command.type !== "thread.metadata.update" ||
+                    command.worktreeContinuation === undefined
+                      ? []
+                      : [
+                          {
+                            sequence: 1,
+                            commandId: command.commandId,
+                            event: {
+                              id: EventId.make("event-worktree-continuation"),
+                              type: "run.created",
+                              threadId,
+                              occurredAt: DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"),
+                              payload: {
+                                id: RunId.make("run-worktree-continuation"),
+                                threadId,
+                                ordinal: 2,
+                                providerInstanceId: ProviderInstanceId.make("provider-test"),
+                                modelSelection: {
+                                  instanceId: ProviderInstanceId.make("provider-test"),
+                                  model: "test-model",
+                                },
+                                providerThreadId: null,
+                                userMessageId: command.worktreeContinuation.messageId,
+                                rootNodeId: null,
+                                activeAttemptId: null,
+                                status: options.continuation === "started" ? "starting" : "queued",
+                                requestedAt: DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"),
+                                startedAt: null,
+                                completedAt: null,
+                                checkpointId: null,
+                                contextHandoffId: null,
+                              },
+                            },
+                          },
+                        ],
+                } satisfies OrchestratorV2DispatchResult),
       ),
     ),
   );
@@ -179,22 +222,9 @@ const makeHarness = (options: HarnessOptions = {}) => {
       ? Effect.succeed(makeProjection(thread))
       : Effect.fail(new OrchestratorProjectionError({ threadId: id }));
   });
-  const sendToThread = vi.fn((_: unknown) => {
-    switch (options.continuation ?? "queued") {
-      case "fails":
-        return Effect.fail(
-          new ThreadManagementService.ThreadManagementThreadArchivedError({
-            threadId,
-          }),
-        );
-      case "dies":
-        return Effect.die(new Error("send defect"));
-      default:
-        return Effect.succeed({
-          delivery: "queued",
-        } as ThreadManagementService.ThreadManagementSendResult);
-    }
-  });
+  const sendToThread = vi.fn(() =>
+    Effect.die(new Error("Standalone continuation dispatch is unsafe")),
+  );
   const getById = vi.fn((id: ProjectId) =>
     options.projectReadFails
       ? (Effect.fail("simulated project read failure") as never)
@@ -398,16 +428,44 @@ const runStatus = (harness: ReturnType<typeof makeHarness>) =>
 const decodeHandoffInput = Schema.decodeUnknownEffect(WorktreeMcpHandoffInput);
 
 describe("t3_worktree_handoff", () => {
-  it.effect("reports a continuation failure without failing the handoff", () => {
-    const harness = makeHarness({ continuation: "fails" });
+  it.effect("queues the continuation prompt as the thread's next message", () => {
+    const harness = makeHarness();
     return Effect.gen(function* () {
       const result = yield* runHandoff(harness, {
-        branch: "feature/continue-fails",
+        branch: "feature/continue",
+        continuationPrompt: "Keep fixing the login bug in the new worktree.",
+      });
+
+      expect(result.continuation).toEqual({ status: "scheduled", delivery: "queued" });
+      expect(harness.dispatch).toHaveBeenCalledExactlyOnceWith(
+        expect.objectContaining({
+          type: "thread.metadata.update",
+          threadId,
+          worktreePath: result.worktreePath,
+          expectedWorktreePath: null,
+          worktreeContinuation: {
+            messageId: expect.stringContaining("worktree-continuation"),
+            text: "Keep fixing the login bug in the new worktree.",
+          },
+        }),
+      );
+      expect(harness.sendToThread).not.toHaveBeenCalled();
+      expect(harness.dispatch.mock.invocationCallOrder[0]).toBeLessThan(
+        harness.runForThread.mock.invocationCallOrder[0]!,
+      );
+    });
+  });
+
+  it.effect("reports a continuation started when no run was blocking the handoff", () => {
+    const harness = makeHarness({ continuation: "started" });
+    return Effect.gen(function* () {
+      const result = yield* runHandoff(harness, {
+        branch: "feature/continue-idle",
         continuationPrompt: "Keep going.",
       });
-      expect(result.continuation).toMatchObject({ status: "failed" });
-      expect(result.worktreePath).toBe("/worktrees/project/feature/continue-fails");
-      expect(harness.dispatch).toHaveBeenCalled();
+      expect(result.continuation).toEqual({ status: "scheduled", delivery: "started" });
+      expect(harness.dispatch).toHaveBeenCalledTimes(1);
+      expect(harness.sendToThread).not.toHaveBeenCalled();
     });
   });
 
@@ -637,7 +695,12 @@ describe("t3_worktree_handoff", () => {
       yield* Fiber.join(interruption);
 
       expect(harness.dispatch).toHaveBeenCalledTimes(1);
-      expect(harness.sendToThread).toHaveBeenCalledTimes(1);
+      expect(harness.dispatch).toHaveBeenCalledWith(
+        expect.objectContaining({
+          worktreeContinuation: expect.objectContaining({ text: "Keep going in the worktree." }),
+        }),
+      );
+      expect(harness.sendToThread).not.toHaveBeenCalled();
       expect(harness.runForThread).toHaveBeenCalledTimes(1);
     }),
   );

apps/server/src/mcp/WorktreeMcpService.ts

@@ -147,7 +147,6 @@ const make = Effect.gen(function* () {
           );
         return {
           commandId: CommandId.make(part("command", "worktree-handoff")),
-          continuationCommandId: CommandId.make(part("command", "worktree-continuation")),
           continuationMessageId: MessageId.make(part("message", "worktree-continuation")),
         };
       }),
@@ -293,14 +292,22 @@ const make = Effect.gen(function* () {
               `Thread '${scope.thread.threadId}' was archived while the worktree was being created; the handoff was rolled back.`,
             );
           }
-          yield* threadManagement
+          return yield* threadManagement
             .dispatch({
               type: "thread.metadata.update",
               commandId: ids.commandId,
               threadId: scope.thread.threadId,
               branch: worktree.worktree.refName,
               worktreePath,
               expectedWorktreePath: null,
+              ...(input.continuationPrompt === undefined
+                ? {}
+                : {
+                    worktreeContinuation: {
+                      messageId: ids.continuationMessageId,
+                      text: input.continuationPrompt,
+                    },
+                  }),
             })
             .pipe(
               Effect.catchCause((cause) =>
@@ -321,33 +328,14 @@ const make = Effect.gen(function* () {
           ),
         );
 
-        // Queue continuation immediately after binding; detachment terminates the caller, so relaunch needs a durable message.
-        const queueContinuation: Effect.Effect<WorktreeMcpContinuationStatus, WorktreeMcpFailure> =
-          Effect.suspend(() =>
-            input.continuationPrompt === undefined
-              ? Effect.succeed<WorktreeMcpContinuationStatus>({ status: "skipped" })
-              : threadManagement
-                  .sendToThread({
-                    projectId: projection.thread.projectId,
-                    commandId: ids.continuationCommandId,
-                    threadId: scope.thread.threadId,
-                    messageId: ids.continuationMessageId,
-                    text: input.continuationPrompt,
-                    attachments: [],
-                    mode: "queue",
-                    createdBy: "agent",
-                    creationSource: "mcp",
-                  })
-                  .pipe(
-                    Effect.map((sendResult): WorktreeMcpContinuationStatus => ({
-                      status: "scheduled",
-                      delivery: sendResult.delivery,
-                    })),
-                    reportFailed("worktree handoff continuation failed to queue"),
-                  ),
-          );
-
-        const continuation = yield* recheckAndBind.pipe(Effect.andThen(queueContinuation));
+        const bound = yield* recheckAndBind;
+        const queued = bound.storedEvents.some(
+          ({ event }) => event.type === "run.created" && event.payload.status === "queued",
+        );
+        const continuation: WorktreeMcpContinuationStatus =
+          input.continuationPrompt === undefined
+            ? { status: "skipped" }
+            : { status: "scheduled", delivery: queued ? "queued" : "started" };
 
         yield* vcsStatusBroadcaster
           .refreshStatus(worktreePath)

apps/server/src/orchestration-v2/EffectOutbox.ts

@@ -38,6 +38,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
     type: Schema.Literal("provider-session.detach"),
     providerSessionId: ProviderSessionId,
     detail: Schema.optional(Schema.String),
+    reason: Schema.optional(Schema.Literal("workspace_changed")),
     revokeMcpCredential: Schema.optional(Schema.Boolean),
   }),
   Schema.Struct({

apps/server/src/orchestration-v2/EffectWorker.ts

@@ -124,6 +124,7 @@ export const layerExecutor: Layer.Layer<
                 providerSessionId: effect.request.providerSessionId,
                 threadId: effect.threadId,
                 ...(effect.request.detail === undefined ? {} : { detail: effect.request.detail }),
+                ...(effect.request.reason === undefined ? {} : { reason: effect.request.reason }),
                 ...(effect.request.revokeMcpCredential === undefined
                   ? {}
                   : { revokeMcpCredential: effect.request.revokeMcpCredential }),

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -3572,6 +3572,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         cause: `Thread ${command.threadId} branch changed before the metadata update could be applied.`,
       });
     }
+    if (command.type === "thread.metadata.update" && command.worktreeContinuation !== undefined) {
+      if (isProviderNativeSubagentThread(thread)) {
+        return yield* new OrchestratorSubagentThreadReadOnlyError({
+          commandId: command.commandId,
+          threadId: command.threadId,
+        });
+      }
+      if (
+        command.worktreePath == null ||
+        command.worktreePath === thread.worktreePath ||
+        thread.archivedAt !== null
+      ) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause:
+            "A worktree continuation requires an active thread and a changed worktree binding.",
+        });
+      }
+    }
     if (command.type === "thread.metadata.update" && command.expectedEmpty === true) {
       const records = yield* projectionStore
         .getThreadRecords(command.threadId, ["runs"])
@@ -4621,6 +4641,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
                           ? "Runtime mode changed."
                           : "Provider or model selection changed.",
                 // Terminal detaches revoke MCP credentials; reattachments need other detach reasons to preserve them.
+                ...(command.type === "thread.metadata.update"
+                  ? { reason: "workspace_changed" as const }
+                  : {}),
                 ...(command.type === "thread.archive" ? { revokeMcpCredential: true } : {}),
               },
             } satisfies PendingOrchestrationEffectV2;
@@ -5850,6 +5873,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     command: Extract<OrchestrationV2Command, { readonly type: "message.dispatch" }>,
     events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
     effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
+    worktreeContinuationOfRunId?: RunId,
   ) {
     let projection = yield* getProjectionWithPendingEvents(command.threadId, events);
     if (command.manualContinuationOfRunId !== undefined) {
@@ -5921,6 +5945,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       if (
         !source ||
         !(
+          (source.id === worktreeContinuationOfRunId && isActiveRun(source)) ||
           isRestartContinuationSource(source, projection.providerTurns) ||
           continuationFailedBeforePrompt(source, projection.providerTurns)
         ) ||
@@ -5936,6 +5961,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         ) ||
         (yield* stopReachedRun(command, command.threadId, source.id))
       ) {
+        if (worktreeContinuationOfRunId !== undefined) {
+          return yield* new OrchestratorDispatchError({
+            commandId: command.commandId,
+            commandType: command.type,
+            cause: "The active run changed before the workspace continuation could be queued.",
+          });
+        }
         // Preserve the current row so stale automatic deliveries receive an
         // accepted receipt without changing work or repeatedly retrying.
         yield* emit(
@@ -13630,7 +13662,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       case "thread.active.reorder":
       case "thread.tag.set":
       case "thread.mark-unread":
-      case "thread.metadata.update":
       case "thread.pull-request.link":
       case "thread.pull-request.unlink":
       case "thread.pull-request-link.sync":
@@ -13643,6 +13674,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       case "provider.switch":
         yield* dispatchThreadMutation(command, events, effects);
         break;
+      case "thread.metadata.update": {
+        const sourceRun =
+          command.worktreeContinuation === undefined
+            ? undefined
+            : (yield* getProjectionWithPendingEvents(command.threadId, events)).runs.find(
+                isActiveRun,
+              );
+        yield* dispatchThreadMutation(command, events, effects);
+        if (command.worktreeContinuation !== undefined) {
+          yield* dispatchMessage(
+            {
+              type: "message.dispatch",
+              commandId: command.commandId,
+              threadId: command.threadId,
+              ...command.worktreeContinuation,
+              ...(sourceRun ? { restartContinuationOfRunId: sourceRun.id } : {}),
+              attachments: [],
+              createdBy: "agent",
+              creationSource: "mcp",
+              dispatchMode: { type: "queue_after_active" },
+            },
+            events,
+            effects,
+            sourceRun?.id,
+          );
+        }
+        break;
+      }
       case "thread.pull-request-watch.sync":
         yield* dispatchPullRequestWatchSync(command, events, effects);
         break;