fix(server): allow provider handoffs during background work

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

@@ -7723,7 +7723,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       pendingForkTransfer === undefined && (!isProviderSwitch || canResumeAcrossInstances)
         ? unseenPublicationRuns(projection, activeProviderThread?.id, contextBoundary)
         : [];
-    const sessionsToDetach = uniqueSessions([
+    const releasableSessions = uniqueSessions([
       ...projection.providerSessions.filter(
         (session) =>
           providerSwitchPlan?.releaseProviderSessionIds.includes(session.id) &&
@@ -7732,8 +7732,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       ),
       ...(freshContext ? sessionsReleasedByFreshContext(projection, activeProviderThread) : []),
     ]);
-    for (const detaching of sessionsToDetach) {
+    const busySessionIds = new Set<ProviderSessionId>();
+    for (const detaching of releasableSessions) {
       if (yield* sessionHasBackgroundWork(detaching, projection.providerThreads)) {
+        // Independent handoffs can keep the old session alive, as queued sends do.
+        if (
+          !freshContext &&
+          !continuationChanged &&
+          providerSwitchPlan?.transition.type === "create_with_handoff"
+        ) {
+          busySessionIds.add(detaching.id);
+          continue;
+        }
         return yield* new OrchestratorDispatchError({
           commandId: command.commandId,
           commandType: command.type,
@@ -7743,6 +7753,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         });
       }
     }
+    const sessionsToDetach = releasableSessions.filter(
+      (session) => !busySessionIds.has(session.id),
+    );
 
     if (
       pendingForkTransfer === undefined &&

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

@@ -6,6 +6,7 @@ import {
   EventId,
   CommandId,
   CheckpointId,
+  ContextHandoffId,
   ContextTransferId,
   CheckpointRef,
   CheckpointScopeId,
@@ -549,6 +550,86 @@ it.effect("record reads restrict provider histories to delivery identities", ()
   }).pipe(Effect.provide(layerTest)),
 );
 
+it.effect("keeps the latest provider binding when outgoing background work updates", () =>
+  Effect.gen(function* () {
+    const store = yield* ProjectionStore.ProjectionStoreV2;
+    const threadId = yield* addRolledBackRecoveryCandidate("background-provider-binding");
+    let projection = yield* store.getThreadProjection(threadId);
+    const at = projection.updatedAt;
+    const firstId = ProviderThreadId.make("provider-thread:background-first");
+    const secondId = ProviderThreadId.make("provider-thread:background-second");
+    const first: OrchestrationV2ProviderThread = {
+      id: firstId,
+      appThreadId: threadId,
+      driver,
+      providerInstanceId,
+      providerSessionId: ProviderSessionId.make("session:background-first"),
+      ownerNodeId: null,
+      nativeThreadRef: { driver, nativeId: "background-first", strength: "strong" },
+      nativeConversationHeadRef: null,
+      status: "idle",
+      firstRunOrdinal: 1,
+      lastRunOrdinal: 1,
+      handoffIds: [],
+      forkedFrom: null,
+      createdAt: at,
+      updatedAt: at,
+    };
+    const second = {
+      ...first,
+      id: secondId,
+      providerSessionId: ProviderSessionId.make("session:background-second"),
+      nativeThreadRef: { driver, nativeId: "background-second", strength: "strong" as const },
+      firstRunOrdinal: 2,
+      lastRunOrdinal: 2,
+    };
+    const backgroundUpdate = {
+      ...first,
+      pendingBackgroundTasks: [{ taskId: "command:background-first", kind: "command" as const }],
+      updatedAt: DateTime.add(at, { seconds: 1 }),
+    };
+    for (const [name, payload, activeId] of [
+      ["first", first, firstId],
+      ["handoff", second, secondId],
+      ["background", backgroundUpdate, secondId],
+      ["rollback", { ...second, lastRunOrdinal: null }, secondId],
+      ["background-after-rollback", backgroundUpdate, secondId],
+      ["return", { ...first, lastRunOrdinal: 3 }, firstId],
+      ["late-second", second, firstId],
+      ["retry", { ...second, lastRunOrdinal: 3 }, secondId],
+      ["late-same-run", { ...first, lastRunOrdinal: 3 }, secondId],
+      [
+        "retry-return",
+        {
+          ...first,
+          lastRunOrdinal: 3,
+          handoffIds: [ContextHandoffId.make("handoff:retry-return")],
+        },
+        firstId,
+      ],
+      ["late-retry-target", { ...second, lastRunOrdinal: 3 }, firstId],
+    ] as const) {
+      const event = {
+        id: EventId.make(`event:background-provider-binding:${name}`),
+        type: "provider-thread.updated" as const,
+        threadId,
+        occurredAt: payload.updatedAt,
+        payload,
+      };
+      yield* store.apply(event);
+      projection = ProjectionStore.applyToProjection(projection, event);
+      const stored = yield* store.getThreadProjection(threadId);
+      assert.equal(stored.thread.activeProviderThreadId, activeId, name);
+      assert.equal(projection.thread.activeProviderThreadId, activeId, name);
+      assert.isTrue(DateTime.Equivalence(stored.thread.updatedAt, projection.thread.updatedAt));
+      assert.deepEqual(
+        stored.providerThreads.find((thread) => thread.id === payload.id)?.pendingBackgroundTasks,
+        payload.pendingBackgroundTasks ?? [],
+      );
+    }
+  }).pipe(Effect.provide(layerTest)),
+);
+
 it.effect("finds the thread behind a provider's native thread id", () =>
   Effect.gen(function* () {
     const store = yield* ProjectionStore.ProjectionStoreV2;

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

@@ -859,6 +859,24 @@ function isQueuedProviderThreadPlaceholder(providerThread: OrchestrationV2Provid
   );
 }
 
+function canActivateProviderThread(
+  providerThread: OrchestrationV2ProviderThread,
+  activeProviderThreadId: ProviderThreadId | null,
+  activeLastRunOrdinal: number | null | undefined,
+  previous: OrchestrationV2ProviderThread | undefined,
+): boolean {
+  return (
+    !isQueuedProviderThreadPlaceholder(providerThread) &&
+    (activeProviderThreadId === null ||
+      activeLastRunOrdinal === undefined ||
+      providerThread.id === activeProviderThreadId ||
+      ((providerThread.lastRunOrdinal ?? 0) >= (activeLastRunOrdinal ?? 0) &&
+        (previous === undefined ||
+          (providerThread.lastRunOrdinal ?? 0) > (previous.lastRunOrdinal ?? 0) ||
+          providerThread.handoffIds.some((id) => !previous.handoffIds.includes(id)))))
+  );
+}
+
 export function applyToProjection(
   projection: OrchestrationV2ThreadProjection,
   event: OrchestrationV2DomainEvent,
@@ -980,7 +998,13 @@ export function applyToProjection(
         ...base,
         thread:
           event.payload.appThreadId === base.thread.id &&
-          !isQueuedProviderThreadPlaceholder(event.payload)
+          canActivateProviderThread(
+            event.payload,
+            base.thread.activeProviderThreadId,
+            base.providerThreads.find((thread) => thread.id === base.thread.activeProviderThreadId)
+              ?.lastRunOrdinal,
+            base.providerThreads.find((thread) => thread.id === event.payload.id),
+          )
             ? {
                 ...base.thread,
                 activeProviderThreadId: event.payload.id,
@@ -2458,6 +2482,28 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
             case "provider-thread.updated": {
               const payloadJson = yield* encodeProviderThreadPayload(event.payload);
               const payload = parseEncodedPayload(payloadJson);
+              const threadRows =
+                event.payload.appThreadId === null ||
+                isQueuedProviderThreadPlaceholder(event.payload)
+                  ? []
+                  : yield* sql<
+                      PayloadRow & {
+                        active_provider_thread_id: string | null;
+                        active_last_run_ordinal: number | null;
+                        previous_provider_thread_json: string | null;
+                      }
+                    >`
+                    SELECT threads.payload_json, active.provider_thread_id AS active_provider_thread_id,
+                      active.last_run_ordinal AS active_last_run_ordinal,
+                      previous.payload_json AS previous_provider_thread_json
+                    FROM orchestration_v2_projection_threads AS threads
+                    LEFT JOIN orchestration_v2_projection_provider_threads AS active
+                      ON active.provider_thread_id = threads.active_provider_thread_id
+                    LEFT JOIN orchestration_v2_projection_provider_threads AS previous
+                      ON previous.provider_thread_id = ${event.payload.id}
+                    WHERE threads.thread_id = ${event.payload.appThreadId}
+                    LIMIT 1
+                  `;
               yield* sql`
               INSERT INTO orchestration_v2_projection_provider_threads (
                 provider_thread_id,
@@ -2501,34 +2547,37 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 updated_at = excluded.updated_at,
                 payload_json = excluded.payload_json
             `;
-              if (
-                event.payload.appThreadId !== null &&
-                !isQueuedProviderThreadPlaceholder(event.payload)
-              ) {
-                const threadRows = yield* sql<PayloadRow>`
-                SELECT payload_json
-                FROM orchestration_v2_projection_threads
-                WHERE thread_id = ${event.payload.appThreadId}
-                LIMIT 1
-              `;
-                const threadRow = threadRows[0];
-                if (threadRow !== undefined) {
-                  const thread = yield* decodeThreadPayload(threadRow.payload_json);
-                  const updatedThread = {
-                    ...thread,
-                    activeProviderThreadId: event.payload.id,
-                    updatedAt: event.payload.updatedAt,
-                  };
-                  const updatedThreadPayloadJson = yield* encodeThreadPayload(updatedThread);
-                  yield* sql`
+              const threadRow = threadRows[0];
+              if (threadRow !== undefined) {
+                const thread = yield* decodeThreadPayload(threadRow.payload_json);
+                const previous =
+                  thread.activeProviderThreadId === event.payload.id ||
+                  threadRow.previous_provider_thread_json === null
+                    ? undefined
+                    : yield* decodeProviderThreadPayload(threadRow.previous_provider_thread_json);
+                const updatedThread = {
+                  ...thread,
+                  activeProviderThreadId: canActivateProviderThread(
+                    event.payload,
+                    thread.activeProviderThreadId,
+                    threadRow.active_provider_thread_id === null
+                      ? undefined
+                      : threadRow.active_last_run_ordinal,
+                    previous,
+                  )
+                    ? event.payload.id
+                    : thread.activeProviderThreadId,
+                  updatedAt: event.payload.updatedAt,
+                };
+                const updatedThreadPayloadJson = yield* encodeThreadPayload(updatedThread);
+                yield* sql`
                   UPDATE orchestration_v2_projection_threads
                   SET
-                    active_provider_thread_id = ${event.payload.id},
+                    active_provider_thread_id = ${updatedThread.activeProviderThreadId},
                     updated_at = ${stringField(parseEncodedPayload(updatedThreadPayloadJson), "updatedAt")},
                     payload_json = ${updatedThreadPayloadJson}
                   WHERE thread_id = ${event.payload.appThreadId}
                 `;
-                }
               }
               break;
             }

apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts

@@ -3046,9 +3046,13 @@ describe("orchestration v2 provider switching", () => {
       { background: false, queued: false, sharedHistory: false },
     ],
     [
-      "refuses the hand-off while the previous instance still runs background work",
+      "keeps a busy previous session running when an immediate hand-off starts",
       { background: true, queued: false, sharedHistory: false },
     ],
+    [
+      "refuses an immediate account switch while the replaced session still runs background work",
+      { background: true, queued: false, sharedHistory: true },
+    ],
     [
       "keeps a busy previous session running when a queued hand-off starts",
       { background: true, queued: true, sharedHistory: false },
@@ -3219,7 +3223,7 @@ describe("orchestration v2 provider switching", () => {
           );
           assert.equal(firstSession?.status, "ready");
 
-          if (background) {
+          if (background && sharedHistory) {
             // Native subagents live only in the adapter's roster, so the projection's
             // pendingBackgroundTasks stays empty while the probe still reports work.
             assert.deepEqual(afterFirst.providerThreads[0]?.pendingBackgroundTasks ?? [], []);
@@ -3249,7 +3253,22 @@ describe("orchestration v2 provider switching", () => {
               [secondSelection.instanceId, "completed"],
             ],
           );
-          assert.lengthOf(afterSecond.contextHandoffs, 1);
+          assert.lengthOf(afterSecond.contextHandoffs, sharedHistory ? 0 : 1);
+          if (background && !sharedHistory) {
+            assert.deepEqual(yield* Ref.get(closedSessions), []);
+            assert.equal(
+              projection.providerSessions.find((session) => session.id === firstSession!.id)
+                ?.status,
+              "ready",
+            );
+            assert.equal(projection.thread.providerInstanceId, secondSelection.instanceId);
+            assert.notEqual(
+              projection.runs[0]?.providerThreadId,
+              projection.runs[1]?.providerThreadId,
+            );
+            assert.include((yield* Ref.get(capturedTurns))[1]!.text, "First account complete");
+            return;
+          }
           assert.deepEqual(
             projection.providerSessions.map((session) => session.providerInstanceId),
             [secondSelection.instanceId],