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..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, @@ -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); + }), ); }); diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 02f329e536feb..77624bb32fe61 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/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; @@ -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"; 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 c9311433eecab..51644fe947aa2 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/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 }, ]); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 62a9e8b88e96d..0c924f99ff4ab 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/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, 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, });