feat(orchestrator): move a running thread between devices

apps/server/src/auth/RpcAuthorization.ts

@@ -28,6 +28,10 @@ export const RPC_REQUIRED_SCOPES = {
   [ORCHESTRATION_V2_WS_METHODS.getArchivedShellSnapshot]: AuthOrchestrationReadScope,
   [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: AuthOrchestrationReadScope,
   [ORCHESTRATION_V2_WS_METHODS.launchThread]: AuthOrchestrationOperateScope,
+  // Preparing stages bytes and records a hop, so it is an operate even
+  // though nothing the user can see has changed yet.
+  [ORCHESTRATION_V2_WS_METHODS.prepareThreadHandoff]: AuthOrchestrationOperateScope,
+  [ORCHESTRATION_V2_WS_METHODS.receiveThreadHandoff]: AuthOrchestrationOperateScope,
   [ORCHESTRATION_V2_WS_METHODS.subscribeArchivedShell]: AuthOrchestrationReadScope,
   [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: AuthOrchestrationReadScope,
   [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: AuthOrchestrationReadScope,

apps/server/src/config.ts

@@ -33,6 +33,8 @@ export interface ServerDerivedPaths {
   readonly providerStatusCacheDir: string;
   readonly worktreesDir: string;
   readonly attachmentsDir: string;
+  /** Staging area for thread handoff bundle parts, one directory per hop. */
+  readonly handoffsDir: string;
   readonly logsDir: string;
   readonly serverLogPath: string;
   readonly serverTracePath: string;
@@ -119,6 +121,7 @@ export const deriveServerPaths = Effect.fn(function* (
     providerStatusCacheDir,
     worktreesDir: join(baseDir, "worktrees"),
     attachmentsDir,
+    handoffsDir: join(stateDir, "handoffs"),
     logsDir,
     serverLogPath: join(logsDir, "server.log"),
     serverTracePath: join(logsDir, "server.trace.ndjson"),
@@ -144,6 +147,7 @@ export const ensureServerDirectories = Effect.fn(function* (derivedPaths: Server
       fs.makeDirectory(derivedPaths.terminalLogsDir, { recursive: true }),
       fs.makeDirectory(derivedPaths.attachmentsDir, { recursive: true }),
       fs.makeDirectory(derivedPaths.worktreesDir, { recursive: true }),
+      fs.makeDirectory(derivedPaths.handoffsDir, { recursive: true }),
       fs.makeDirectory(path.dirname(derivedPaths.keybindingsConfigPath), { recursive: true }),
       fs.makeDirectory(path.dirname(derivedPaths.settingsPath), { recursive: true }),
       fs.makeDirectory(derivedPaths.providerStatusCacheDir, { recursive: true }),

apps/server/src/environment/ServerEnvironment.ts

@@ -148,6 +148,7 @@ export const make = Effect.gen(function* () {
       threadPinning: true,
       threadTitleRegeneration: true,
       threadVisitedTracking: true,
+      threadHandoff: true,
       ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
       ...(serverSelfUpdate === "boot-service" ? { serverSelfUpdateProgress: true } : {}),
     },

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.testkit.ts

@@ -75,6 +75,7 @@ export function makeReplayServerConfig(
     const terminalLogsDir = path.join(logsDir, "terminals");
     const attachmentsDir = path.join(stateDir, "attachments");
     const worktreesDir = path.join(baseDir, "worktrees");
+    const handoffsDir = path.join(stateDir, "handoffs");
     const providerStatusCacheDir = path.join(baseDir, "caches");
 
     for (const directory of [
@@ -84,6 +85,7 @@ export function makeReplayServerConfig(
       terminalLogsDir,
       attachmentsDir,
       worktreesDir,
+      handoffsDir,
       providerStatusCacheDir,
     ]) {
       yield* fs.makeDirectory(directory, { recursive: true });
@@ -121,6 +123,7 @@ export function makeReplayServerConfig(
       settingsPath: path.join(stateDir, "settings.json"),
       providerStatusCacheDir,
       worktreesDir,
+      handoffsDir,
       attachmentsDir,
       logsDir,
       serverLogPath: path.join(logsDir, "server.log"),

apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.testkit.ts

@@ -538,6 +538,7 @@ function makeReplayServerConfig(
     const terminalLogsDir = path.join(logsDir, "terminals");
     const attachmentsDir = path.join(stateDir, "attachments");
     const worktreesDir = path.join(baseDir, "worktrees");
+    const handoffsDir = path.join(stateDir, "handoffs");
     const providerStatusCacheDir = path.join(baseDir, "caches");
     for (const directory of [
       stateDir,
@@ -546,6 +547,7 @@ function makeReplayServerConfig(
       terminalLogsDir,
       attachmentsDir,
       worktreesDir,
+      handoffsDir,
       providerStatusCacheDir,
     ]) {
       yield* fs.makeDirectory(directory, { recursive: true });
@@ -582,6 +584,7 @@ function makeReplayServerConfig(
       settingsPath: path.join(stateDir, "settings.json"),
       providerStatusCacheDir,
       worktreesDir,
+      handoffsDir,
       attachmentsDir,
       logsDir,
       serverLogPath: path.join(logsDir, "server.log"),

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

@@ -217,6 +217,9 @@ function commandThreadId(command: OrchestrationV2Command): ThreadId {
     case "runtime-request.respond":
     case "checkpoint.rollback":
     case "provider.switch":
+    case "thread.handoff.depart":
+    case "thread.handoff.complete":
+    case "thread.handoff.abort":
       return command.threadId;
     case "delegated_task.request":
     case "delegated_task.wake-policy":
@@ -1353,7 +1356,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           | "thread.runtime-mode.set"
           | "thread.interaction-mode.set"
           | "thread.model-selection.set"
-          | "provider.switch";
+          | "provider.switch"
+          | "thread.handoff.depart"
+          | "thread.handoff.complete"
+          | "thread.handoff.abort";
       }
     >,
     events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
@@ -1430,6 +1436,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       });
     }
 
+    // A thread can only leave once, and only the hop that locked it may report
+    // its landing or release it. Without these the client could lock a thread
+    // twice and lose track of which peer owns it.
+    if (command.type === "thread.handoff.depart" && (thread.handoff ?? null) !== null) {
+      return yield* new OrchestratorDispatchError({
+        commandId: command.commandId,
+        commandType: command.type,
+        cause: `Thread ${command.threadId} is already handed off to ${thread.handoff?.peerEnvironmentId}.`,
+      });
+    }
+    if (command.type === "thread.handoff.complete" || command.type === "thread.handoff.abort") {
+      const link = thread.handoff ?? null;
+      if (link === null || link.handoffId !== command.handoffId) {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: `Thread ${command.threadId} has no handoff ${command.handoffId} in flight.`,
+        });
+      }
+      if (link.presence !== "away") {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: `Thread ${command.threadId} owns handoff ${command.handoffId} and cannot report on it.`,
+        });
+      }
+    }
+
     const providerSwitchPlan =
       command.type === "thread.model-selection.set" || command.type === "provider.switch"
         ? yield* Effect.gen(function* () {
@@ -1638,6 +1672,39 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             modelSelection: command.modelSelection,
             updatedAt: now,
           };
+        case "thread.handoff.depart":
+          return {
+            ...thread,
+            handoff: {
+              handoffId: command.handoffId,
+              presence: "away" as const,
+              peerEnvironmentId: command.peerEnvironmentId,
+              peerThreadId: null,
+              peerLabel: command.peerLabel,
+              previousHandoffId: command.previousHandoffId,
+              hopCount: command.hopCount,
+              updatedAt: now,
+            },
+            updatedAt: now,
+          };
+        case "thread.handoff.complete":
+          return {
+            ...thread,
+            ...(thread.handoff == null
+              ? {}
+              : {
+                  handoff: {
+                    ...thread.handoff,
+                    peerThreadId: command.peerThreadId,
+                    updatedAt: now,
+                  },
+                }),
+            updatedAt: now,
+          };
+        // Releasing the link is what makes this side live again, so an aborted
+        // hop leaves no trace to clean up later.
+        case "thread.handoff.abort":
+          return { ...thread, handoff: null, updatedAt: now };
       }
     })();
     const eventType = (() => {
@@ -1675,6 +1742,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           return "thread.model-selection-updated" as const;
         case "provider.switch":
           return "thread.provider-switched" as const;
+        case "thread.handoff.depart":
+          return "thread.handoff-departed" as const;
+        case "thread.handoff.complete":
+          return "thread.handoff-arrived" as const;
+        case "thread.handoff.abort":
+          return "thread.handoff-failed" as const;
       }
     })();
     yield* emit(
@@ -2778,6 +2851,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   ) =>
     Effect.gen(function* () {
       let projection = yield* getProjectionWithPendingEvents(command.threadId, events);
+      // A departed thread is a read-only record of work that is now running
+      // somewhere else. Refusing the send here — rather than in the client — is
+      // what makes "exactly one side is live" an invariant instead of a
+      // convention, and it is the reason a handoff never has to merge two
+      // divergent conversations.
+      const handoff = projection.thread.handoff ?? null;
+      if (handoff !== null && handoff.presence === "away") {
+        return yield* new OrchestratorDispatchError({
+          commandId: command.commandId,
+          commandType: command.type,
+          cause: `Thread ${command.threadId} is running on ${handoff.peerLabel ?? handoff.peerEnvironmentId}.`,
+        });
+      }
       if (projection.thread.settledOverride !== null) {
         const now = yield* DateTime.now;
         const thread: OrchestrationV2AppThread = {

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

@@ -214,6 +214,10 @@ export function applyToProjection(
     case "thread.interaction-mode-updated":
     case "thread.model-selection-updated":
     case "thread.provider-switched":
+    case "thread.handoff-departed":
+    case "thread.handoff-arrived":
+    case "thread.handoff-returned":
+    case "thread.handoff-failed":
       return {
         ...base,
         thread: event.payload,
@@ -916,6 +920,7 @@ export function threadShellFromProjection(
     pinnedAt: projection.thread.pinnedAt ?? null,
     lastVisitedAt: projection.thread.lastVisitedAt,
     titleRegeneration: projection.thread.titleRegeneration ?? null,
+    handoff: projection.thread.handoff ?? null,
     deletedAt: projection.thread.deletedAt,
   };
 }
@@ -1100,6 +1105,7 @@ function shellFromState(input: {
     pinnedAt: input.state.thread.pinnedAt ?? null,
     lastVisitedAt: input.state.thread.lastVisitedAt,
     titleRegeneration: input.state.thread.titleRegeneration ?? null,
+    handoff: input.state.thread.handoff ?? null,
     deletedAt: input.state.thread.deletedAt,
   };
 }
@@ -1128,7 +1134,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           case "thread.runtime-mode-updated":
           case "thread.interaction-mode-updated":
           case "thread.model-selection-updated":
-          case "thread.provider-switched": {
+          case "thread.provider-switched":
+          case "thread.handoff-departed":
+          case "thread.handoff-arrived":
+          case "thread.handoff-returned":
+          case "thread.handoff-failed": {
             const payloadJson = yield* encodeThreadPayload(event.payload);
             const payload = parseEncodedPayload(payloadJson);
             yield* sql`

apps/server/src/orchestration-v2/ThreadHandoffGit.test.ts

@@ -0,0 +1,82 @@
+import { assert, describe, it } from "@effect/vitest";
+
+import {
+  classifyIncomingTip,
+  handoffPreTagName,
+  handoffRefName,
+  handoffStashLabel,
+  type ClassifyIncomingTipInput,
+} from "./ThreadHandoffGit.ts";
+
+const input = (overrides: Partial<ClassifyIncomingTipInput>): ClassifyIncomingTipInput => ({
+  localTip: "local",
+  incomingTip: "incoming",
+  incomingContainsLocal: false,
+  localContainsIncoming: false,
+  hasCommonAncestor: true,
+  ...overrides,
+});
+
+describe("classifyIncomingTip", () => {
+  it("advances a branch the receiving repository does not have yet", () => {
+    assert.strictEqual(classifyIncomingTip(input({ localTip: null })), "advance");
+  });
+
+  it("absorbs an identical tip instead of moving anything", () => {
+    assert.strictEqual(
+      classifyIncomingTip(input({ localTip: "same", incomingTip: "same" })),
+      "absorb",
+    );
+  });
+
+  it("advances when the incoming commit descends from the local tip", () => {
+    assert.strictEqual(classifyIncomingTip(input({ incomingContainsLocal: true })), "advance");
+  });
+
+  it("absorbs when the receiving side is already ahead", () => {
+    assert.strictEqual(classifyIncomingTip(input({ localContainsIncoming: true })), "absorb");
+  });
+
+  it("refuses when both sides moved, so neither tip is a descendant of the other", () => {
+    assert.strictEqual(classifyIncomingTip(input({})), "diverged");
+  });
+
+  it("refuses unrelated histories rather than treating them as a divergence to rebase", () => {
+    assert.strictEqual(classifyIncomingTip(input({ hasCommonAncestor: false })), "unrelated");
+  });
+
+  it("treats a fast-forward as advance even when common ancestry was not computed", () => {
+    assert.strictEqual(
+      classifyIncomingTip(input({ incomingContainsLocal: true, hasCommonAncestor: false })),
+      "advance",
+    );
+  });
+
+  it("never advances on a tip that only the local side contains", () => {
+    const classification = classifyIncomingTip(
+      input({ localContainsIncoming: true, hasCommonAncestor: true }),
+    );
+
+    assert.notStrictEqual(classification, "advance");
+  });
+});
+
+describe("handoff ref names", () => {
+  it("parks refused commits under an environment-scoped namespace", () => {
+    assert.strictEqual(
+      handoffRefName("environment-mac", "feat/thread-handoff"),
+      "refs/handoff/environment-mac/feat/thread-handoff",
+    );
+  });
+
+  it("names the pre-move tag after the hop that moved the pointer", () => {
+    assert.strictEqual(handoffPreTagName("handoff-1"), "handoff-pre-handoff-1");
+  });
+
+  it("puts the base sha in the stash label so a later pop is legible", () => {
+    assert.strictEqual(
+      handoffStashLabel("handoff-1", "a91f2c4"),
+      "handoff-overwritten-handoff-1-base-a91f2c4",
+    );
+  });
+});