authorgravatar for 130890337+eimexdev@users.noreply.github.comeimexdev <130890337+eimexdev@users.noreply.github.com> 2026-10-04 14:49:56-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-04 18:49:56-03:00
logefecd3cf8bcec3d1891b5f5a27dc2f6d797c6448
treebcb58565e58df2dfc5979eeac2970479ae87e963
parent2188bdd8b5c0251f40daebdb66647dd0150cf076
signature Signed by PGP key B5690EEEBB952194

fix(server): Claude threads no longer get stuck after background commands (#15770)

9 files changed, 291 insertions(+), 16 deletions(-)

apps/server/src/claudeModelOptions.ts+1-1
......@@ -37,7 +37,7 @@ export function compileClaudeModelSelection(
3737 const resolvedEffort = resolveClaudeCatalogEffort(catalog, selection.model, rawEffort);
3838 const effort = normalizeClaudeCatalogEffort(catalog, resolvedEffort, selection.model);
3939 const fastMode = supportsBoolean("fastMode")
40 ? getModelSelectionBooleanOptionValue(selection, "fastMode")
40 ? (getModelSelectionBooleanOptionValue(selection, "fastMode") ?? false)
4141 : undefined;
4242 const thinking = supportsBoolean("thinking")
4343 ? getModelSelectionBooleanOptionValue(selection, "thinking")
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts+76-1
......@@ -126,6 +126,7 @@ function makeClaudeTestTurnInput(input: {
126126 readonly text: string;
127127 readonly attachments: ProviderAdapterV2TurnInput["message"]["attachments"];
128128 readonly providerTurnOrdinal?: number;
129 readonly nativeThreadHasTurns?: boolean;
129130 readonly messageCreatedBy?: ProviderAdapterV2TurnInput["message"]["createdBy"];
130131 readonly messageCreationSource?: ProviderAdapterV2TurnInput["message"]["creationSource"];
131132 readonly modelSelection?: ModelSelection;
......@@ -137,6 +138,9 @@ function makeClaudeTestTurnInput(input: {
137138 runId: RunId.make(`run-${input.attemptId}`),
138139 runOrdinal: 1,
139140 providerTurnOrdinal: input.providerTurnOrdinal ?? 1,
141 ...(input.nativeThreadHasTurns === undefined
142 ? {}
143 : { nativeThreadHasTurns: input.nativeThreadHasTurns }),
140144 attemptId: input.attemptId,
141145 rootNodeId: NodeId.make(`node-${input.attemptId}`),
142146 providerThread: input.providerThread,
......@@ -1783,7 +1787,7 @@ describe("ClaudeAdapterV2 native fork", () => {
17831787});
17841788
17851789describe("ClaudeAdapterV2 native session identity", () => {
1786 const openTurnWithOrdinal = (providerTurnOrdinal: number) =>
1790 const openTurnWithOrdinal = (providerTurnOrdinal: number, nativeThreadHasTurns?: boolean) =>
17871791 Effect.scoped(
17881792 Effect.gen(function* () {
17891793 const fileSystem = yield* FileSystem.FileSystem;
......@@ -1842,6 +1846,7 @@ describe("ClaudeAdapterV2 native session identity", () => {
18421846 text: "Respond with identity ok",
18431847 attachments: [],
18441848 providerTurnOrdinal,
1849 ...(nativeThreadHasTurns === undefined ? {} : { nativeThreadHasTurns }),
18451850 }),
18461851 );
18471852 return openedQueries;
......@@ -1867,6 +1872,14 @@ describe("ClaudeAdapterV2 native session identity", () => {
18671872 assert.equal(openedQueries[0]?.options.sessionId, undefined);
18681873 }),
18691874 );
1875
1876 it.effect("creates a fresh native session despite earlier provider-thread turns", () =>
1877 Effect.gen(function* () {
1878 const openedQueries = yield* openTurnWithOrdinal(4, false);
1879 assert.equal(openedQueries[0]?.options.sessionId, "native-session-identity");
1880 assert.equal(openedQueries[0]?.options.resume, undefined);
1881 }),
1882 );
18701883});
18711884
18721885const encodeJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
......@@ -2182,6 +2195,68 @@ describe("ClaudeAdapterV2 background wake turns", () => {
21822195 });
21832196 const makeWakeHarness = makeWakeHarnessWithOptions();
21842197
2198 it.effect(
2199 "reuses a background shell's query for omitted and explicit Normal, but blocks Fast",
2200 () =>
2201 Effect.scoped(
2202 Effect.gen(function* () {
2203 const harness = yield* makeWakeHarness;
2204 const now = yield* DateTime.now;
2205 const normal = {
2206 instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID,
2207 model: "claude-opus-5-5",
2208 } satisfies ModelSelection;
2209 const turn = (ordinal: number, modelSelection: ModelSelection) =>
2210 makeClaudeTestTurnInput({
2211 threadId: harness.threadId,
2212 providerThread: harness.providerThread,
2213 now,
2214 attemptId: RunAttemptId.make(`attempt-normal-background:${ordinal}`),
2215 text: `Request ${ordinal}`,
2216 attachments: [],
2217 providerTurnOrdinal: ordinal,
2218 modelSelection,
2219 });
2220 yield* harness.runtime.startTurn(turn(1, normal));
2221 const originalOptions = harness.getOpenedOptions();
2222 yield* harness.offerAndWait(wakeTaskStarted);
2223 yield* harness.offerAndWait(turnOneResult);
2224 yield* Queue.take(harness.terminalReceipts);
2225 assert.isTrue(yield* harness.hasPendingBackgroundWork);
2226
2227 yield* harness.runtime.startTurn(
2228 turn(2, {
2229 ...normal,
2230 options: [{ id: "fastMode", value: false }],
2231 }),
2232 );
2233 assert.strictEqual(harness.getOpenedOptions(), originalOptions);
2234 assert.lengthOf(harness.offeredMessages, 2);
2235 yield* harness.offerAndWait(turnOneResult);
2236 yield* Queue.take(harness.terminalReceipts);
2237
2238 const refused = yield* harness.runtime
2239 .startTurn(
2240 turn(3, {
2241 ...normal,
2242 options: [{ id: "fastMode", value: true }],
2243 }),
2244 )
2245 .pipe(Effect.result);
2246 assert.equal(refused._tag, "Failure");
2247 if (refused._tag === "Failure") {
2248 assert.instanceOf(
2249 refused.failure.cause,
2250 ClaudeAdapterV2.ClaudeBackgroundWorkBlocksQueryReplacementError,
2251 );
2252 }
2253 assert.strictEqual(harness.getOpenedOptions(), originalOptions);
2254 assert.lengthOf(harness.offeredMessages, 2);
2255 assert.isTrue(yield* harness.hasPendingBackgroundWork);
2256 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
2257 ),
2258 );
2259
21852260 it.effect.each(["completed", "interrupted"] as const)(
21862261 "projects Claude thinking blocks when %s",
21872262 (status) =>
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+6-5
......@@ -6969,11 +6969,12 @@ export function makeClaudeAdapterV2(
69696969
69706970 const openedWithResume = (yield* Ref.get(openedNativeThreads)).has(nativeThreadId);
69716971 // openedNativeThreads is per session instance and is lost when the
6972 // provider session is idle-released. A prior persisted provider turn
6973 // proves the native session already exists, so the query must resume
6974 // it; reopening with a fixed session id makes the CLI fail fast with
6975 // "Session ID ... is already in use".
6976 const hasPersistedProviderTurn = turnInput.providerTurnOrdinal > 1;
6972 // provider session is idle-released. A prior turn on this native id
6973 // requires resume; sessionId would fail with "already in use".
6974 // A fresh-session fallback keeps provider-thread history but binds
6975 // a new native id, which must be created before it can be resumed.
6976 const hasPersistedProviderTurn =
6977 turnInput.nativeThreadHasTurns ?? turnInput.providerTurnOrdinal > 1;
69776978 const shouldResume =
69786979 resumeSessionAt !== undefined || openedWithResume || hasPersistedProviderTurn;
69796980 const queryOptions = makeClaudeQueryOptions({
apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts+45-4
......@@ -41,7 +41,7 @@ const modelSelection = { instanceId, model: "test-model" };
4141// Codex turns leave commands running, then the thread moves to another
4242// provider thread (a provider switch). Stop on the newer, settled run must
4343// reach both provider threads and end all of the Codex work.
44it.effect("Stop reaches background work an earlier provider thread still runs", () =>
44const stopEarlierBackgroundWork = (failedStart: boolean) =>
4545 Effect.scoped(
4646 Effect.gen(function* () {
4747 const cwd = yield* checkpointWorkspace("background-work-stop");
......@@ -233,7 +233,10 @@ it.effect("Stop reaches background work an earlier provider thread still runs",
233233 const settledRun = (input: {
234234 readonly ordinal: number;
235235 readonly providerThreadId: ProviderThreadId;
236 readonly runningItem?: { readonly id: TurnItemId; readonly kind: "command" | "subagent" };
236 readonly runningItem?: {
237 readonly id: TurnItemId;
238 readonly kind: "command" | "subagent";
239 };
237240 }) => {
238241 const runId = RunId.make(`run:${input.ordinal}`);
239242 const attemptId = RunAttemptId.make(`attempt:${input.ordinal}`);
......@@ -407,11 +410,41 @@ it.effect("Stop reaches background work an earlier provider thread still runs",
407410 ],
408411 });
409412
413 const failedRun = settledRun({ ordinal: 5, providerThreadId: otherProviderThreadId });
414 if (failedStart) {
415 yield* sink.write({
416 events: failedRun.events.flatMap((event): Array<OrchestrationV2DomainEvent> => {
417 switch (event.type) {
418 case "provider-turn.updated":
419 return [];
420 case "run.created":
421 return [{ ...event, payload: { ...event.payload, status: "failed" } }];
422 case "node.updated":
423 return [
424 {
425 ...event,
426 payload: { ...event.payload, status: "failed", providerTurnId: null },
427 },
428 ];
429 case "run-attempt.created":
430 return [
431 {
432 ...event,
433 payload: { ...event.payload, status: "failed", providerTurnId: null },
434 },
435 ];
436 default:
437 return [event];
438 }
439 }),
440 });
441 }
442
410443 yield* orchestrator.dispatch({
411444 type: "run.interrupt",
412445 commandId: CommandId.make("stop-background-work"),
413446 threadId,
414 runId: latestRun.runId,
447 runId: failedStart ? failedRun.runId : latestRun.runId,
415448 });
416449 yield* worker.drain();
417450
......@@ -442,5 +475,13 @@ it.effect("Stop reaches background work an earlier provider thread still runs",
442475 ),
443476 );
444477 }),
445 ),
478 );
479
480it.effect("Stop reaches background work an earlier provider thread still runs", () =>
481 stopEarlierBackgroundWork(false),
482);
483
484it.effect(
485 "Stop reaches earlier background work after the newest run fails before provider start",
486 () => stopEarlierBackgroundWork(true),
446487);
apps/server/src/orchestration-v2/Orchestrator.ts+13-5
......@@ -8040,11 +8040,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
80408040 activeProviderThreadId: projection.thread.activeProviderThreadId,
80418041 runs: projection.runs,
80428042 }).length > 0;
8043 const providerTurn = projection.providerTurns.findLast(
8044 (candidate) =>
8045 candidate.runAttemptId === run?.activeAttemptId &&
8046 (candidate.status === "running" || hasBackgroundWork),
8047 );
8043 // A failed start has no provider turn. Background work still belongs
8044 // to the provider thread, so Stop reaches its latest accepted turn.
8045 const providerTurn =
8046 projection.providerTurns.findLast(
8047 (candidate) =>
8048 candidate.runAttemptId === run?.activeAttemptId &&
8049 (candidate.status === "running" || hasBackgroundWork),
8050 ) ??
8051 (hasBackgroundWork
8052 ? projection.providerTurns.findLast(
8053 (candidate) => candidate.providerThreadId === run?.providerThreadId,
8054 )
8055 : undefined);
80488056 if (run === undefined || rootNode === undefined || providerThread === undefined) {
80498057 return yield* new OrchestratorDispatchError({
80508058 commandId: command.commandId,
apps/server/src/orchestration-v2/ProviderAdapter.ts+2
......@@ -398,6 +398,8 @@ export interface ProviderAdapterV2TurnInput {
398398 readonly threadId: ThreadId;
399399 readonly runId: RunId;
400400 readonly runOrdinal: number;
401 /** Whether the current native session has an accepted turn; omitted when unknown. */
402 readonly nativeThreadHasTurns?: boolean;
401403 readonly providerTurnOrdinal: number;
402404 readonly restartContinuationOfRunId?: RunId;
403405 readonly attemptId: RunAttemptId;
apps/server/src/orchestration-v2/ProviderTurnStartService.ts+7
......@@ -1231,6 +1231,13 @@ export const layer: Layer.Layer<
12311231 .filter((turn) => turn.providerThreadId === providerThread.id)
12321232 .map((turn) => turn.ordinal),
12331233 ) + 1,
1234 // Legacy accepted attempts have no native id. They count only before
1235 // a replacement, while no accepted attempt records a native identity.
1236 nativeThreadHasTurns:
1237 nativeInputRunIds.size > 0 ||
1238 (legacyInputRunIds.size > 0 &&
1239 sameNativeThread &&
1240 !acceptedAttempts.some((source) => source.nativeThreadId !== undefined)),
12341241 shouldStartProviderTurn: runControls.shouldStartProviderTurn,
12351242 shouldFinalizeRun: runControls.shouldFinalizeRun,
12361243 hasUnpairedRunInterruptRequest: runControls.hasUnpairedRunInterruptRequest,
apps/server/src/orchestration-v2/RunExecutionService.ts+4
......@@ -507,6 +507,7 @@ export interface RunExecutionServiceV2StartRootRunInput {
507507 readonly attempt: OrchestrationV2RunAttempt;
508508 readonly attemptId: RunAttemptId;
509509 readonly providerTurnOrdinal: number;
510 readonly nativeThreadHasTurns?: boolean;
510511 readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect<
511512 ReadonlyArray<InheritedBackgroundTurnItemRoute>,
512513 unknown
......@@ -1348,6 +1349,9 @@ export const layer: Layer.Layer<
13481349 runId: input.run.id,
13491350 runOrdinal: input.run.ordinal,
13501351 providerTurnOrdinal: input.providerTurnOrdinal,
1352 ...(input.nativeThreadHasTurns === undefined
1353 ? {}
1354 : { nativeThreadHasTurns: input.nativeThreadHasTurns }),
13511355 ...(input.run.restartContinuationOfRunId === undefined
13521356 ? {}
13531357 : {
apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts+137
......@@ -85,6 +85,9 @@ interface CapturedTurn {
8585 readonly providerThreadId: ProviderThreadId;
8686 readonly text: string;
8787 readonly attachments: ReadonlyArray<ChatAttachment>;
88 readonly nativeThreadId?: string | null;
89 readonly providerTurnOrdinal?: number;
90 readonly nativeThreadHasTurns?: boolean;
8891}
8992
9093function unimplemented(driver: ProviderDriverKind, detail: string) {
......@@ -232,6 +235,11 @@ function makeTestAdapter(input: {
232235 providerThreadId: turnInput.providerThread.id,
233236 text: turnInput.message.text,
234237 attachments: turnInput.message.attachments,
238 nativeThreadId: turnInput.providerThread.nativeThreadRef?.nativeId ?? null,
239 providerTurnOrdinal: turnInput.providerTurnOrdinal,
240 ...(turnInput.nativeThreadHasTurns === undefined
241 ? {}
242 : { nativeThreadHasTurns: turnInput.nativeThreadHasTurns }),
235243 },
236244 ]);
237245 if (
......@@ -1051,6 +1059,135 @@ describe("orchestration v2 provider switching", () => {
10511059 ),
10521060 );
10531061
1062 it.live.each(["native", "legacy"] as const)(
1063 "starts a replacement native thread as new after the resume fallback with %s attempts",
1064 (attempts) =>
1065 Effect.scoped(
1066 Effect.gen(function* () {
1067 const cwd = yield* checkpointWorkspace("resume-fallback-native-turns");
1068 const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]);
1069 const failResumeOnce = yield* Ref.make(false);
1070 const generation = yield* Ref.make(0);
1071 const registry = ProviderAdapterRegistry.makeLayer([
1072 makeTestAdapter({
1073 instanceId: CLAUDE_MODEL_SELECTION.instanceId,
1074 driver: CLAUDE_DRIVER,
1075 capabilities: ClaudeProviderCapabilitiesV2,
1076 modelSelection: CLAUDE_MODEL_SELECTION,
1077 responseByRunOrdinal: {},
1078 capturedTurns,
1079 failResumeOnce,
1080 nativeThreadGeneration: generation,
1081 }),
1082 ]);
1083 yield* Effect.gen(function* () {
1084 const orchestrator = yield* Orchestrator.OrchestratorV2;
1085 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
1086 const runToCompletion = Effect.fn("runToCompletion")(function* (ordinal: number) {
1087 yield* orchestrator.dispatch({
1088 type: "message.dispatch",
1089 commandId: CommandId.make(`resume-fallback:${ordinal}`),
1090 threadId,
1091 messageId: MessageId.make(`resume-fallback:${ordinal}`),
1092 createdBy: "user",
1093 creationSource: "web",
1094 text: `Request ${ordinal}`,
1095 attachments: [],
1096 modelSelection: CLAUDE_MODEL_SELECTION,
1097 dispatchMode: { type: "start_immediately" },
1098 });
1099 yield* orchestrator.streamStoredEvents.pipe(
1100 Stream.filter(
1101 ({ event }) =>
1102 event.type === "run.updated" &&
1103 event.payload.ordinal === ordinal &&
1104 (event.payload.status === "completed" || event.payload.status === "failed"),
1105 ),
1106 Stream.runHead,
1107 );
1108 yield* worker.drain();
1109 assert.equal(
1110 (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)?.status,
1111 "completed",
1112 );
1113 });
1114 yield* orchestrator.dispatch({
1115 type: "thread.create",
1116 commandId: CommandId.make("resume-fallback:create"),
1117 threadId,
1118 projectId,
1119 createdBy: "user",
1120 creationSource: "web",
1121 title: "Resume fallback",
1122 modelSelection: CLAUDE_MODEL_SELECTION,
1123 runtimeMode: "full-access",
1124 interactionMode: "default",
1125 branch: null,
1126 worktreePath: null,
1127 });
1128 yield* runToCompletion(1);
1129 // A detached session resumes on the next turn, as after a restart.
1130 const providerThread = (yield* orchestrator.getThreadProjection(
1131 threadId,
1132 )).providerThreads.at(0);
1133 if (providerThread === undefined || providerThread.providerSessionId === null) {
1134 return yield* Effect.die("Expected a provider session after the first turn.");
1135 }
1136 yield* orchestrator.dispatch({
1137 type: "provider-session.detach",
1138 commandId: CommandId.make("resume-fallback:detach"),
1139 threadId,
1140 providerSessionId: providerThread.providerSessionId,
1141 });
1142 yield* worker.drain();
1143 if (attempts === "legacy") {
1144 // Attempts persisted before native identity existed carry no native id.
1145 const existing = yield* orchestrator.getThreadProjection(threadId);
1146 yield* (yield* EventSink.EventSinkV2).write({
1147 events: existing.attempts.map(
1148 ({ nativeThreadId: _nativeThreadId, ...legacy }, index) => ({
1149 id: EventId.make(`resume-fallback-legacy-attempt:${index}`),
1150 type: "run-attempt.updated" as const,
1151 threadId,
1152 occurredAt: existing.thread.createdAt,
1153 payload: legacy,
1154 }),
1155 ),
1156 });
1157 }
1158 yield* Ref.set(failResumeOnce, true);
1159 yield* runToCompletion(2);
1160 yield* runToCompletion(3);
1161
1162 const turns = yield* Ref.get(capturedTurns);
1163 assert.equal(turns.length, 3);
1164 assert.equal(yield* Ref.get(generation), 2);
1165 // The fallback binds a new native thread to a provider thread that
1166 // already has a turn, so the provider must not resume it.
1167 assert.notEqual(turns[1]?.nativeThreadId, turns[0]?.nativeThreadId);
1168 assert.isAbove(turns[1]?.providerTurnOrdinal ?? 0, 1);
1169 assert.isFalse(turns[1]?.nativeThreadHasTurns);
1170 assert.equal(turns[2]?.nativeThreadId, turns[1]?.nativeThreadId);
1171 assert.isTrue(turns[2]?.nativeThreadHasTurns);
1172 }).pipe(
1173 Effect.provide(
1174 makeOrchestratorV2ReplayLayerWithRegistry(
1175 {
1176 name: "resume-fallback-native-turns",
1177 runtimePolicyOverride: {
1178 cwd,
1179 approvalPolicy: "never",
1180 sandboxPolicy: { type: "readOnly" },
1181 },
1182 },
1183 registry,
1184 ),
1185 ),
1186 );
1187 }),
1188 ),
1189 );
1190
10541191 it.live.each(
10551192 (["failed", "interrupted"] as const).flatMap((status) =>
10561193 [false, true].flatMap((queued) =>