diff --git a/apps/server/src/claudeModelOptions.ts b/apps/server/src/claudeModelOptions.ts index d018c2f0af32..ef7b92e1a2e9 100644 --- a/apps/server/src/claudeModelOptions.ts +++ b/apps/server/src/claudeModelOptions.ts @@ -37,7 +37,7 @@ export function compileClaudeModelSelection( const resolvedEffort = resolveClaudeCatalogEffort(catalog, selection.model, rawEffort); const effort = normalizeClaudeCatalogEffort(catalog, resolvedEffort, selection.model); const fastMode = supportsBoolean("fastMode") - ? getModelSelectionBooleanOptionValue(selection, "fastMode") + ? (getModelSelectionBooleanOptionValue(selection, "fastMode") ?? false) : undefined; const thinking = supportsBoolean("thinking") ? getModelSelectionBooleanOptionValue(selection, "thinking") diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 2e237bf3bb4c..5099c4911796 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -126,6 +126,7 @@ function makeClaudeTestTurnInput(input: { readonly text: string; readonly attachments: ProviderAdapterV2TurnInput["message"]["attachments"]; readonly providerTurnOrdinal?: number; + readonly nativeThreadHasTurns?: boolean; readonly messageCreatedBy?: ProviderAdapterV2TurnInput["message"]["createdBy"]; readonly messageCreationSource?: ProviderAdapterV2TurnInput["message"]["creationSource"]; readonly modelSelection?: ModelSelection; @@ -137,6 +138,9 @@ function makeClaudeTestTurnInput(input: { runId: RunId.make(`run-${input.attemptId}`), runOrdinal: 1, providerTurnOrdinal: input.providerTurnOrdinal ?? 1, + ...(input.nativeThreadHasTurns === undefined + ? {} + : { nativeThreadHasTurns: input.nativeThreadHasTurns }), attemptId: input.attemptId, rootNodeId: NodeId.make(`node-${input.attemptId}`), providerThread: input.providerThread, @@ -1783,7 +1787,7 @@ describe("ClaudeAdapterV2 native fork", () => { }); describe("ClaudeAdapterV2 native session identity", () => { - const openTurnWithOrdinal = (providerTurnOrdinal: number) => + const openTurnWithOrdinal = (providerTurnOrdinal: number, nativeThreadHasTurns?: boolean) => Effect.scoped( Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1842,6 +1846,7 @@ describe("ClaudeAdapterV2 native session identity", () => { text: "Respond with identity ok", attachments: [], providerTurnOrdinal, + ...(nativeThreadHasTurns === undefined ? {} : { nativeThreadHasTurns }), }), ); return openedQueries; @@ -1867,6 +1872,14 @@ describe("ClaudeAdapterV2 native session identity", () => { assert.equal(openedQueries[0]?.options.sessionId, undefined); }), ); + + it.effect("creates a fresh native session despite earlier provider-thread turns", () => + Effect.gen(function* () { + const openedQueries = yield* openTurnWithOrdinal(4, false); + assert.equal(openedQueries[0]?.options.sessionId, "native-session-identity"); + assert.equal(openedQueries[0]?.options.resume, undefined); + }), + ); }); const encodeJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); @@ -2182,6 +2195,68 @@ describe("ClaudeAdapterV2 background wake turns", () => { }); const makeWakeHarness = makeWakeHarnessWithOptions(); + it.effect( + "reuses a background shell's query for omitted and explicit Normal, but blocks Fast", + () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const normal = { + instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, + model: "claude-opus-5-5", + } satisfies ModelSelection; + const turn = (ordinal: number, modelSelection: ModelSelection) => + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make(`attempt-normal-background:${ordinal}`), + text: `Request ${ordinal}`, + attachments: [], + providerTurnOrdinal: ordinal, + modelSelection, + }); + yield* harness.runtime.startTurn(turn(1, normal)); + const originalOptions = harness.getOpenedOptions(); + yield* harness.offerAndWait(wakeTaskStarted); + yield* harness.offerAndWait(turnOneResult); + yield* Queue.take(harness.terminalReceipts); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + + yield* harness.runtime.startTurn( + turn(2, { + ...normal, + options: [{ id: "fastMode", value: false }], + }), + ); + assert.strictEqual(harness.getOpenedOptions(), originalOptions); + assert.lengthOf(harness.offeredMessages, 2); + yield* harness.offerAndWait(turnOneResult); + yield* Queue.take(harness.terminalReceipts); + + const refused = yield* harness.runtime + .startTurn( + turn(3, { + ...normal, + options: [{ id: "fastMode", value: true }], + }), + ) + .pipe(Effect.result); + assert.equal(refused._tag, "Failure"); + if (refused._tag === "Failure") { + assert.instanceOf( + refused.failure.cause, + ClaudeAdapterV2.ClaudeBackgroundWorkBlocksQueryReplacementError, + ); + } + assert.strictEqual(harness.getOpenedOptions(), originalOptions); + assert.lengthOf(harness.offeredMessages, 2); + assert.isTrue(yield* harness.hasPendingBackgroundWork); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect.each(["completed", "interrupted"] as const)( "projects Claude thinking blocks when %s", (status) => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index ba0cfbd8fbd5..b2beb04c79b3 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6969,11 +6969,12 @@ export function makeClaudeAdapterV2( const openedWithResume = (yield* Ref.get(openedNativeThreads)).has(nativeThreadId); // openedNativeThreads is per session instance and is lost when the - // provider session is idle-released. A prior persisted provider turn - // proves the native session already exists, so the query must resume - // it; reopening with a fixed session id makes the CLI fail fast with - // "Session ID ... is already in use". - const hasPersistedProviderTurn = turnInput.providerTurnOrdinal > 1; + // provider session is idle-released. A prior turn on this native id + // requires resume; sessionId would fail with "already in use". + // A fresh-session fallback keeps provider-thread history but binds + // a new native id, which must be created before it can be resumed. + const hasPersistedProviderTurn = + turnInput.nativeThreadHasTurns ?? turnInput.providerTurnOrdinal > 1; const shouldResume = resumeSessionAt !== undefined || openedWithResume || hasPersistedProviderTurn; const queryOptions = makeClaudeQueryOptions({ diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index 2b1c529ba177..3c183da6268c 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -41,7 +41,7 @@ const modelSelection = { instanceId, model: "test-model" }; // Codex turns leave commands running, then the thread moves to another // provider thread (a provider switch). Stop on the newer, settled run must // reach both provider threads and end all of the Codex work. -it.effect("Stop reaches background work an earlier provider thread still runs", () => +const stopEarlierBackgroundWork = (failedStart: boolean) => Effect.scoped( Effect.gen(function* () { const cwd = yield* checkpointWorkspace("background-work-stop"); @@ -233,7 +233,10 @@ it.effect("Stop reaches background work an earlier provider thread still runs", const settledRun = (input: { readonly ordinal: number; readonly providerThreadId: ProviderThreadId; - readonly runningItem?: { readonly id: TurnItemId; readonly kind: "command" | "subagent" }; + readonly runningItem?: { + readonly id: TurnItemId; + readonly kind: "command" | "subagent"; + }; }) => { const runId = RunId.make(`run:${input.ordinal}`); const attemptId = RunAttemptId.make(`attempt:${input.ordinal}`); @@ -407,11 +410,41 @@ it.effect("Stop reaches background work an earlier provider thread still runs", ], }); + const failedRun = settledRun({ ordinal: 5, providerThreadId: otherProviderThreadId }); + if (failedStart) { + yield* sink.write({ + events: failedRun.events.flatMap((event): Array => { + switch (event.type) { + case "provider-turn.updated": + return []; + case "run.created": + return [{ ...event, payload: { ...event.payload, status: "failed" } }]; + case "node.updated": + return [ + { + ...event, + payload: { ...event.payload, status: "failed", providerTurnId: null }, + }, + ]; + case "run-attempt.created": + return [ + { + ...event, + payload: { ...event.payload, status: "failed", providerTurnId: null }, + }, + ]; + default: + return [event]; + } + }), + }); + } + yield* orchestrator.dispatch({ type: "run.interrupt", commandId: CommandId.make("stop-background-work"), threadId, - runId: latestRun.runId, + runId: failedStart ? failedRun.runId : latestRun.runId, }); yield* worker.drain(); @@ -442,5 +475,13 @@ it.effect("Stop reaches background work an earlier provider thread still runs", ), ); }), - ), + ); + +it.effect("Stop reaches background work an earlier provider thread still runs", () => + stopEarlierBackgroundWork(false), +); + +it.effect( + "Stop reaches earlier background work after the newest run fails before provider start", + () => stopEarlierBackgroundWork(true), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index abdfc29579cd..f6c172c74621 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -8040,11 +8040,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio activeProviderThreadId: projection.thread.activeProviderThreadId, runs: projection.runs, }).length > 0; - const providerTurn = projection.providerTurns.findLast( - (candidate) => - candidate.runAttemptId === run?.activeAttemptId && - (candidate.status === "running" || hasBackgroundWork), - ); + // A failed start has no provider turn. Background work still belongs + // to the provider thread, so Stop reaches its latest accepted turn. + const providerTurn = + projection.providerTurns.findLast( + (candidate) => + candidate.runAttemptId === run?.activeAttemptId && + (candidate.status === "running" || hasBackgroundWork), + ) ?? + (hasBackgroundWork + ? projection.providerTurns.findLast( + (candidate) => candidate.providerThreadId === run?.providerThreadId, + ) + : undefined); if (run === undefined || rootNode === undefined || providerThread === undefined) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, diff --git a/apps/server/src/orchestration-v2/ProviderAdapter.ts b/apps/server/src/orchestration-v2/ProviderAdapter.ts index ac300004322e..0db242085c5e 100644 --- a/apps/server/src/orchestration-v2/ProviderAdapter.ts +++ b/apps/server/src/orchestration-v2/ProviderAdapter.ts @@ -398,6 +398,8 @@ export interface ProviderAdapterV2TurnInput { readonly threadId: ThreadId; readonly runId: RunId; readonly runOrdinal: number; + /** Whether the current native session has an accepted turn; omitted when unknown. */ + readonly nativeThreadHasTurns?: boolean; readonly providerTurnOrdinal: number; readonly restartContinuationOfRunId?: RunId; readonly attemptId: RunAttemptId; diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index a3e3b122ca45..172b64d8a144 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1231,6 +1231,13 @@ export const layer: Layer.Layer< .filter((turn) => turn.providerThreadId === providerThread.id) .map((turn) => turn.ordinal), ) + 1, + // Legacy accepted attempts have no native id. They count only before + // a replacement, while no accepted attempt records a native identity. + nativeThreadHasTurns: + nativeInputRunIds.size > 0 || + (legacyInputRunIds.size > 0 && + sameNativeThread && + !acceptedAttempts.some((source) => source.nativeThreadId !== undefined)), shouldStartProviderTurn: runControls.shouldStartProviderTurn, shouldFinalizeRun: runControls.shouldFinalizeRun, hasUnpairedRunInterruptRequest: runControls.hasUnpairedRunInterruptRequest, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 6aff5da24f1e..7ab2c87cdef4 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -507,6 +507,7 @@ export interface RunExecutionServiceV2StartRootRunInput { readonly attempt: OrchestrationV2RunAttempt; readonly attemptId: RunAttemptId; readonly providerTurnOrdinal: number; + readonly nativeThreadHasTurns?: boolean; readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect< ReadonlyArray, unknown @@ -1348,6 +1349,9 @@ export const layer: Layer.Layer< runId: input.run.id, runOrdinal: input.run.ordinal, providerTurnOrdinal: input.providerTurnOrdinal, + ...(input.nativeThreadHasTurns === undefined + ? {} + : { nativeThreadHasTurns: input.nativeThreadHasTurns }), ...(input.run.restartContinuationOfRunId === undefined ? {} : { diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 9d01f22fe6a0..98ee5d6c8537 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -85,6 +85,9 @@ interface CapturedTurn { readonly providerThreadId: ProviderThreadId; readonly text: string; readonly attachments: ReadonlyArray; + readonly nativeThreadId?: string | null; + readonly providerTurnOrdinal?: number; + readonly nativeThreadHasTurns?: boolean; } function unimplemented(driver: ProviderDriverKind, detail: string) { @@ -232,6 +235,11 @@ function makeTestAdapter(input: { providerThreadId: turnInput.providerThread.id, text: turnInput.message.text, attachments: turnInput.message.attachments, + nativeThreadId: turnInput.providerThread.nativeThreadRef?.nativeId ?? null, + providerTurnOrdinal: turnInput.providerTurnOrdinal, + ...(turnInput.nativeThreadHasTurns === undefined + ? {} + : { nativeThreadHasTurns: turnInput.nativeThreadHasTurns }), }, ]); if ( @@ -1051,6 +1059,135 @@ describe("orchestration v2 provider switching", () => { ), ); + it.live.each(["native", "legacy"] as const)( + "starts a replacement native thread as new after the resume fallback with %s attempts", + (attempts) => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("resume-fallback-native-turns"); + const capturedTurns = yield* Ref.make>([]); + const failResumeOnce = yield* Ref.make(false); + const generation = yield* Ref.make(0); + const registry = ProviderAdapterRegistry.makeLayer([ + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + failResumeOnce, + nativeThreadGeneration: generation, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const runToCompletion = Effect.fn("runToCompletion")(function* (ordinal: number) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`resume-fallback:${ordinal}`), + threadId, + messageId: MessageId.make(`resume-fallback:${ordinal}`), + createdBy: "user", + creationSource: "web", + text: `Request ${ordinal}`, + attachments: [], + modelSelection: CLAUDE_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.ordinal === ordinal && + (event.payload.status === "completed" || event.payload.status === "failed"), + ), + Stream.runHead, + ); + yield* worker.drain(); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)?.status, + "completed", + ); + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("resume-fallback:create"), + threadId, + projectId, + createdBy: "user", + creationSource: "web", + title: "Resume fallback", + modelSelection: CLAUDE_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* runToCompletion(1); + // A detached session resumes on the next turn, as after a restart. + const providerThread = (yield* orchestrator.getThreadProjection( + threadId, + )).providerThreads.at(0); + if (providerThread === undefined || providerThread.providerSessionId === null) { + return yield* Effect.die("Expected a provider session after the first turn."); + } + yield* orchestrator.dispatch({ + type: "provider-session.detach", + commandId: CommandId.make("resume-fallback:detach"), + threadId, + providerSessionId: providerThread.providerSessionId, + }); + yield* worker.drain(); + if (attempts === "legacy") { + // Attempts persisted before native identity existed carry no native id. + const existing = yield* orchestrator.getThreadProjection(threadId); + yield* (yield* EventSink.EventSinkV2).write({ + events: existing.attempts.map( + ({ nativeThreadId: _nativeThreadId, ...legacy }, index) => ({ + id: EventId.make(`resume-fallback-legacy-attempt:${index}`), + type: "run-attempt.updated" as const, + threadId, + occurredAt: existing.thread.createdAt, + payload: legacy, + }), + ), + }); + } + yield* Ref.set(failResumeOnce, true); + yield* runToCompletion(2); + yield* runToCompletion(3); + + const turns = yield* Ref.get(capturedTurns); + assert.equal(turns.length, 3); + assert.equal(yield* Ref.get(generation), 2); + // The fallback binds a new native thread to a provider thread that + // already has a turn, so the provider must not resume it. + assert.notEqual(turns[1]?.nativeThreadId, turns[0]?.nativeThreadId); + assert.isAbove(turns[1]?.providerTurnOrdinal ?? 0, 1); + assert.isFalse(turns[1]?.nativeThreadHasTurns); + assert.equal(turns[2]?.nativeThreadId, turns[1]?.nativeThreadId); + assert.isTrue(turns[2]?.nativeThreadHasTurns); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "resume-fallback-native-turns", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly" }, + }, + }, + registry, + ), + ), + ); + }), + ), + ); + it.live.each( (["failed", "interrupted"] as const).flatMap((status) => [false, true].flatMap((queued) =>