fix(server): preserve native history in busy Codex forks

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

@@ -250,6 +250,7 @@ export const CodexProviderCapabilitiesV2 = {
     canRollbackThread: true,
     canForkThread: true,
     canForkFromTurn: true,
+    canForkWhileRunning: true,
     canForkFromSubagentThread: true,
     exposesNativeThreadId: true,
   },

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

@@ -258,6 +258,7 @@ layer("CommandPolicyV2", (it) => {
         capabilities: CursorProviderCapabilitiesV2,
         sameProvider: true,
         hasStrongNativeSource: true,
+        hasNativeTurnBoundary: false,
         sourceRunStatus: "completed",
         fromSpecificTurn: true,
         sourceBusy: false,
@@ -267,24 +268,39 @@ layer("CommandPolicyV2", (it) => {
     }),
   );
 
-  it.effect("forks a busy source through portable context", () =>
-    Effect.gen(function* () {
-      const policy = yield* CommandPolicy.CommandPolicyV2;
-      const decide = (sourceBusy: boolean) =>
-        policy.decideForkExecution({
-          commandId,
-          threadId,
-          providerInstanceId: ProviderInstanceId.make("codex"),
-          capabilities: CodexProviderCapabilitiesV2,
-          sameProvider: true,
-          hasStrongNativeSource: true,
-          sourceRunStatus: "completed",
-          fromSpecificTurn: true,
-          sourceBusy,
-        });
+  it.effect.each([
+    { canForkWhileRunning: true, hasNativeTurnBoundary: true, busyStrategy: "native_fork" },
+    { canForkWhileRunning: false, hasNativeTurnBoundary: true, busyStrategy: "portable_context" },
+    {
+      canForkWhileRunning: undefined,
+      hasNativeTurnBoundary: true,
+      busyStrategy: "portable_context",
+    },
+    { canForkWhileRunning: true, hasNativeTurnBoundary: false, busyStrategy: "portable_context" },
+  ] as const)(
+    "forks a busy source with canForkWhileRunning=$canForkWhileRunning and hasNativeTurnBoundary=$hasNativeTurnBoundary",
+    ({ canForkWhileRunning, hasNativeTurnBoundary, busyStrategy }) =>
+      Effect.gen(function* () {
+        const policy = yield* CommandPolicy.CommandPolicyV2;
+        const decide = (sourceBusy: boolean) =>
+          policy.decideForkExecution({
+            commandId,
+            threadId,
+            providerInstanceId: ProviderInstanceId.make("codex"),
+            capabilities: {
+              ...CodexProviderCapabilitiesV2,
+              threads: { ...CodexProviderCapabilitiesV2.threads, canForkWhileRunning },
+            },
+            sameProvider: true,
+            hasStrongNativeSource: true,
+            hasNativeTurnBoundary,
+            sourceRunStatus: "completed",
+            fromSpecificTurn: true,
+            sourceBusy,
+          });
 
-      assert.equal(yield* decide(false), "native_fork");
-      assert.equal(yield* decide(true), "portable_context");
-    }),
+        assert.equal(yield* decide(false), "native_fork");
+        assert.equal(yield* decide(true), busyStrategy);
+      }),
   );
 });

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

@@ -209,9 +209,10 @@ export interface CommandPolicyV2Shape {
     input: CapabilityCheckInput & {
       readonly sameProvider: boolean;
       readonly hasStrongNativeSource: boolean;
+      readonly hasNativeTurnBoundary: boolean;
       readonly sourceRunStatus: OrchestrationV2Run["status"];
       readonly fromSpecificTurn: boolean;
-      /** Later source runs are still active; providers cannot fork a thread mid-turn. */
+      /** Later source runs are still active. */
       readonly sourceBusy: boolean;
     },
   ) => Effect.Effect<ForkExecutionPolicyV2, CommandPolicyV2Error>;
@@ -376,7 +377,8 @@ const decideForkExecution: CommandPolicyV2Shape["decideForkExecution"] = (input)
     (input.sourceRunStatus === "completed" || input.sourceRunStatus === "waiting") &&
     input.sameProvider &&
     input.hasStrongNativeSource &&
-    !input.sourceBusy &&
+    (!input.sourceBusy ||
+      (input.capabilities.threads.canForkWhileRunning === true && input.hasNativeTurnBoundary)) &&
     input.capabilities.threads.canForkThread &&
     (!input.fromSpecificTurn || input.capabilities.threads.canForkFromTurn) &&
     input.capabilities.identity.nativeThreadIds === "strong";

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

@@ -6702,6 +6702,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
               sameProvider:
                 pendingForkTransfer.sourceProviderInstanceId === modelSelection.instanceId,
               hasStrongNativeSource: sourceProviderThread?.nativeThreadRef?.strength === "strong",
+              hasNativeTurnBoundary:
+                sourceProjection?.providerTurns.find((turn) => turn.id === sourceProviderTurnId)
+                  ?.nativeTurnRef != null,
               sourceRunStatus: sourceRun.status,
               fromSpecificTurn: sourceRun !== null,
               sourceBusy: sourceRunsAfterFork.some(isActiveRun),

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

@@ -299,6 +299,8 @@ function makeLocalCommandHarness(input: {
   readonly sideChat?: {
     readonly contextRun: "current" | "failed" | "cancelled" | "delivered";
     readonly nativeForkFrom?: "idle" | "busy";
+    readonly canForkWhileRunning?: boolean;
+    readonly sourceBoundary?: "native" | "synthetic" | "missing";
     readonly switchedProvider?: boolean;
   };
 }) {
@@ -477,6 +479,8 @@ function makeLocalCommandHarness(input: {
   };
   const parentThreadId = ThreadId.make("side-chat-parent");
   const parentRunId = RunId.make("side-chat-parent-run");
+  const parentAttemptId = RunAttemptId.make("side-chat-parent-attempt");
+  const parentTurnId = ProviderTurnId.make("side-chat-parent-turn");
   const parentProviderThread = {
     ...providerThread,
     id: ProviderThreadId.make("side-chat-parent-provider-thread"),
@@ -497,7 +501,7 @@ function makeLocalCommandHarness(input: {
         threadId: parentThreadId,
         ordinal: 1,
         providerThreadId: parentProviderThread.id,
-        activeAttemptId: null,
+        activeAttemptId: parentAttemptId,
         status: "completed" as const,
         completedAt: now,
       },
@@ -515,8 +519,41 @@ function makeLocalCommandHarness(input: {
         : []),
     ],
     providerThreads: [parentProviderThread],
-    attempts: [],
-    providerTurns: [],
+    attempts: [
+      {
+        ...projection.attempts[0]!,
+        id: parentAttemptId,
+        runId: parentRunId,
+        providerThreadId: parentProviderThread.id,
+        providerTurnId: input.sideChat?.sourceBoundary === "missing" ? null : parentTurnId,
+        status: "completed" as const,
+        startedAt: now,
+        completedAt: now,
+      },
+    ],
+    providerTurns:
+      input.sideChat?.sourceBoundary === "missing"
+        ? []
+        : [
+            {
+              id: parentTurnId,
+              providerThreadId: parentProviderThread.id,
+              nodeId: rootNodeId,
+              runAttemptId: parentAttemptId,
+              ordinal: 1,
+              nativeTurnRef:
+                input.sideChat?.sourceBoundary === "synthetic"
+                  ? null
+                  : {
+                      driver: providerThread.driver,
+                      nativeId: "parent-native-turn",
+                      strength: "strong" as const,
+                    },
+              status: "completed" as const,
+              startedAt: now,
+              completedAt: now,
+            },
+          ],
     turnItems: [],
   };
   if (input.sideChat !== undefined) {
@@ -734,7 +771,15 @@ function makeLocalCommandHarness(input: {
     status: "ready",
     cwd: "/tmp/native-account-command",
     model: null,
-    capabilities: CodexProviderCapabilitiesV2,
+    capabilities: {
+      ...CodexProviderCapabilitiesV2,
+      threads: {
+        ...CodexProviderCapabilitiesV2.threads,
+        canForkWhileRunning:
+          input.sideChat?.canForkWhileRunning ??
+          CodexProviderCapabilitiesV2.threads.canForkWhileRunning,
+      },
+    },
     createdAt: now,
     updatedAt: now,
     lastError: null,
@@ -1253,25 +1298,63 @@ effectIt.effect.each([
 );
 
 effectIt.effect.each([
-  { nativeForkFrom: "idle", forks: true, transferStatus: "consumed", handoffIds: [] },
+  {
+    nativeForkFrom: "idle",
+    canForkWhileRunning: false,
+    sourceBoundary: "missing",
+    forks: true,
+    transferStatus: "consumed",
+    handoffIds: [],
+  },
   {
     nativeForkFrom: "busy",
+    canForkWhileRunning: true,
+    sourceBoundary: "native",
+    forks: true,
+    transferStatus: "consumed",
+    handoffIds: [],
+  },
+  {
+    nativeForkFrom: "busy",
+    canForkWhileRunning: false,
+    sourceBoundary: "native",
+    forks: false,
+    transferStatus: "resolved_portable",
+    handoffIds: ["side-chat-fork-context"],
+  },
+  {
+    nativeForkFrom: "busy",
+    canForkWhileRunning: true,
+    sourceBoundary: "synthetic",
+    forks: false,
+    transferStatus: "resolved_portable",
+    handoffIds: ["side-chat-fork-context"],
+  },
+  {
+    nativeForkFrom: "busy",
+    canForkWhileRunning: true,
+    sourceBoundary: "missing",
     forks: false,
     transferStatus: "resolved_portable",
     handoffIds: ["side-chat-fork-context"],
   },
 ] as const)(
-  "runs a planned native side chat fork when the parent is $nativeForkFrom",
-  ({ nativeForkFrom, forks, transferStatus, handoffIds }) =>
+  "runs a planned native side chat fork when the parent is $nativeForkFrom with canForkWhileRunning=$canForkWhileRunning and sourceBoundary=$sourceBoundary",
+  ({ nativeForkFrom, canForkWhileRunning, sourceBoundary, forks, transferStatus, handoffIds }) =>
     Effect.gen(function* () {
       const harness = makeLocalCommandHarness({
         text: "What is the main thread doing?",
-        sideChat: { contextRun: "current", nativeForkFrom },
+        sideChat: { contextRun: "current", nativeForkFrom, canForkWhileRunning, sourceBoundary },
       });
 
       yield* harness.start;
 
       expect(harness.forkThread).toHaveBeenCalledTimes(forks ? 1 : 0);
+      if (forks && sourceBoundary === "native") {
+        expect(harness.forkThread).toHaveBeenCalledWith(
+          expect.objectContaining({ providerTurnId: "side-chat-parent-turn" }),
+        );
+      }
       expect(harness.projection().contextTransfers).toMatchObject([
         { id: "side-chat-fork", status: transferStatus },
       ]);

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

@@ -1144,12 +1144,15 @@ export const layer: Layer.Layer<
               cause: `Native fork transfer ${nativeForkTransfer.id} has no source provider execution.`,
             });
           }
-          // Providers refuse to fork a busy thread, so a source that started a turn after
-          // dispatch planned this fork gets the portable context dispatch gives a busy source.
+          // The source may have started a turn after dispatch planned this fork.
           const sourceBusy = sourceProjection.runs.some(
             (candidate) => candidate.ordinal > sourceRun.ordinal && isActiveRun(candidate),
           );
-          if (!sourceBusy) {
+          if (
+            !sourceBusy ||
+            (session.providerSession.capabilities.threads.canForkWhileRunning === true &&
+              sourceProviderTurn?.nativeTurnRef != null)
+          ) {
             return yield* loadFromProvider(
               session.forkThread({
                 sourceProviderThread,

packages/contracts/src/orchestrationV2.ts

@@ -272,6 +272,7 @@ export const OrchestrationV2ThreadCapabilities = Schema.Struct({
   canRollbackThread: Schema.Boolean,
   canForkThread: Schema.Boolean,
   canForkFromTurn: Schema.Boolean,
+  canForkWhileRunning: Schema.optional(Schema.Boolean),
   canForkFromSubagentThread: Schema.Boolean,
   exposesNativeThreadId: Schema.Boolean,
 });