From 0ebd6b2a5b902b5141b04f0b24f162f3e98d40b8 Mon Sep 17 00:00:00 2001 From: Alb11747 Date: Mon, 2 Nov 2026 04:51:23 -0500 Subject: [PATCH 1/2] fix(server): preserve native history in busy Codex forks Let providers advertise native forks while a later source turn is running. Codex supports completed fork boundaries in that state, so retain its native history at dispatch and startup instead of creating a portable handoff. Providers without this capability keep the existing fallback. Co-Authored-By: GPT-6.1 Sol at high effort Co-Authored-By: GPT-6.1 Sol at xhigh effort, review subagent for busy Codex forks --- .../Adapters/CodexAdapterV2.ts | 1 + .../orchestration-v2/CommandPolicy.test.ts | 45 +++++++++++-------- .../src/orchestration-v2/CommandPolicy.ts | 4 +- .../ProviderTurnStartService.test.ts | 33 +++++++++++--- .../ProviderTurnStartService.ts | 8 ++-- packages/contracts/src/orchestrationV2.ts | 1 + 6 files changed, 64 insertions(+), 28 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index f81a3809b35fe..98a7a5cb5e432 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/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, }, diff --git a/apps/server/src/orchestration-v2/CommandPolicy.test.ts b/apps/server/src/orchestration-v2/CommandPolicy.test.ts index 8ed4de3e9dea0..53977d71fae0d 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.test.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.test.ts @@ -267,24 +267,33 @@ 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, busyStrategy: "native_fork" }, + { canForkWhileRunning: false, busyStrategy: "portable_context" }, + { canForkWhileRunning: undefined, busyStrategy: "portable_context" }, + ] as const)( + "forks a busy source with canForkWhileRunning=$canForkWhileRunning", + ({ canForkWhileRunning, 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, + 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); + }), ); }); diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 02f329e536feb..e062b263cdc86 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -211,7 +211,7 @@ export interface CommandPolicyV2Shape { readonly hasStrongNativeSource: 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; @@ -376,7 +376,7 @@ 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.capabilities.threads.canForkThread && (!input.fromSpecificTurn || input.capabilities.threads.canForkFromTurn) && input.capabilities.identity.nativeThreadIds === "strong"; diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index c9311433eecab..b342d77f27c0f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -299,6 +299,7 @@ function makeLocalCommandHarness(input: { readonly sideChat?: { readonly contextRun: "current" | "failed" | "cancelled" | "delivered"; readonly nativeForkFrom?: "idle" | "busy"; + readonly canForkWhileRunning?: boolean; readonly switchedProvider?: boolean; }; }) { @@ -734,7 +735,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,20 +1262,34 @@ effectIt.effect.each([ ); effectIt.effect.each([ - { nativeForkFrom: "idle", forks: true, transferStatus: "consumed", handoffIds: [] }, + { + nativeForkFrom: "idle", + canForkWhileRunning: false, + forks: true, + transferStatus: "consumed", + handoffIds: [], + }, + { + nativeForkFrom: "busy", + canForkWhileRunning: true, + forks: true, + transferStatus: "consumed", + handoffIds: [], + }, { nativeForkFrom: "busy", + canForkWhileRunning: false, 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", + ({ nativeForkFrom, canForkWhileRunning, 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 }, }); yield* harness.start; diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 62a9e8b88e96d..c3ee581e0eb39 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1144,12 +1144,14 @@ 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 + ) { return yield* loadFromProvider( session.forkThread({ sourceProviderThread, diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index abf2b309858be..4b4f93f0accb3 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/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, }); From c38a7d4d9fc1efdf6a1e3f7a9957783aafa0be5a Mon Sep 17 00:00:00 2001 From: Alb11747 Date: Mon, 2 Nov 2026 05:38:20 -0500 Subject: [PATCH 2/2] fix(server): require a native boundary for busy forks Keep the portable fallback when a completed source turn has no native turn reference, including synthetic goal replies. Require that boundary at both dispatch and startup so a running parent cannot add later history to the fork. Co-Authored-By: GPT-6.1 Sol at high effort --- .../orchestration-v2/CommandPolicy.test.ts | 17 +++-- .../src/orchestration-v2/CommandPolicy.ts | 4 +- .../src/orchestration-v2/Orchestrator.ts | 3 + .../ProviderTurnStartService.test.ts | 72 +++++++++++++++++-- .../ProviderTurnStartService.ts | 3 +- 5 files changed, 86 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/CommandPolicy.test.ts b/apps/server/src/orchestration-v2/CommandPolicy.test.ts index 53977d71fae0d..a0902c244aa6b 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.test.ts +++ b/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, @@ -268,12 +269,17 @@ layer("CommandPolicyV2", (it) => { ); it.effect.each([ - { canForkWhileRunning: true, busyStrategy: "native_fork" }, - { canForkWhileRunning: false, busyStrategy: "portable_context" }, - { canForkWhileRunning: undefined, busyStrategy: "portable_context" }, + { 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", - ({ canForkWhileRunning, busyStrategy }) => + "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) => @@ -287,6 +293,7 @@ layer("CommandPolicyV2", (it) => { }, sameProvider: true, hasStrongNativeSource: true, + hasNativeTurnBoundary, sourceRunStatus: "completed", fromSpecificTurn: true, sourceBusy, diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index e062b263cdc86..77624bb32fe61 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -209,6 +209,7 @@ 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. */ @@ -376,7 +377,8 @@ const decideForkExecution: CommandPolicyV2Shape["decideForkExecution"] = (input) (input.sourceRunStatus === "completed" || input.sourceRunStatus === "waiting") && input.sameProvider && input.hasStrongNativeSource && - (!input.sourceBusy || input.capabilities.threads.canForkWhileRunning === true) && + (!input.sourceBusy || + (input.capabilities.threads.canForkWhileRunning === true && input.hasNativeTurnBoundary)) && input.capabilities.threads.canForkThread && (!input.fromSpecificTurn || input.capabilities.threads.canForkFromTurn) && input.capabilities.identity.nativeThreadIds === "strong"; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 75df815127f39..5dfd9834c2f30 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/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), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index b342d77f27c0f..51644fe947aa2 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -300,6 +300,7 @@ function makeLocalCommandHarness(input: { readonly contextRun: "current" | "failed" | "cancelled" | "delivered"; readonly nativeForkFrom?: "idle" | "busy"; readonly canForkWhileRunning?: boolean; + readonly sourceBoundary?: "native" | "synthetic" | "missing"; readonly switchedProvider?: boolean; }; }) { @@ -478,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"), @@ -498,7 +501,7 @@ function makeLocalCommandHarness(input: { threadId: parentThreadId, ordinal: 1, providerThreadId: parentProviderThread.id, - activeAttemptId: null, + activeAttemptId: parentAttemptId, status: "completed" as const, completedAt: now, }, @@ -516,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) { @@ -1265,6 +1301,7 @@ effectIt.effect.each([ { nativeForkFrom: "idle", canForkWhileRunning: false, + sourceBoundary: "missing", forks: true, transferStatus: "consumed", handoffIds: [], @@ -1272,6 +1309,7 @@ effectIt.effect.each([ { nativeForkFrom: "busy", canForkWhileRunning: true, + sourceBoundary: "native", forks: true, transferStatus: "consumed", handoffIds: [], @@ -1279,22 +1317,44 @@ effectIt.effect.each([ { 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 with canForkWhileRunning=$canForkWhileRunning", - ({ nativeForkFrom, canForkWhileRunning, 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, canForkWhileRunning }, + 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 }, ]); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index c3ee581e0eb39..0c924f99ff4ab 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1150,7 +1150,8 @@ export const layer: Layer.Layer< ); if ( !sourceBusy || - session.providerSession.capabilities.threads.canForkWhileRunning === true + (session.providerSession.capabilities.threads.canForkWhileRunning === true && + sourceProviderTurn?.nativeTurnRef != null) ) { return yield* loadFromProvider( session.forkThread({