authorgravatar for 76637177+SunkenInTime@users.noreply.github.comDara Adedeji <76637177+SunkenInTime@users.noreply.github.com> 2026-10-07 13:14:57-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 10:14:57-07:00
log87c366b1206b13701fbcc38e459d457f0f9da2df
tree94e0bf57338ded95940bf1e4453d389d8653d3c2
parent8d564a3e5c77d283994119eca82881f8c37a4a4d
signature Signed by PGP key B5690EEEBB952194

fix(server): a refused Claude turn no longer throws away its session (#16287)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

3 files changed, 144 insertions(+), 7 deletions(-)

apps/server/src/orchestration-v2/ContextHandoffDelivery.ts+10-5
...@@ -7,7 +7,11 @@ import * as Effect from "effect/Effect";...@@ -7,7 +7,11 @@ import * as Effect from "effect/Effect";
7import * as Schema from "effect/Schema";7import * as Schema from "effect/Schema";
8import { historyCost, renderHistory, selectHistory } from "./ContextHandoffBudget.ts";8import { historyCost, renderHistory, selectHistory } from "./ContextHandoffBudget.ts";
99
10/** Persist before/after injection: an ambiguous pending delivery requires a fresh native thread. */10/**
11 * Persist before/after injection: an ambiguous pending delivery requires a fresh native thread.
12 * Run `unsent` when the provider refused the turn before reading the prompt, so that
13 * refusal does not leave an ambiguous marker behind.
14 */
11export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextHandoffs")(15export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextHandoffs")(
12 function* <InjectError = never, PersistError = never, BudgetError = never>(input: {16 function* <InjectError = never, PersistError = never, BudgetError = never>(input: {
13 readonly handoffs: ReadonlyArray<OrchestrationV2ContextHandoff>;17 readonly handoffs: ReadonlyArray<OrchestrationV2ContextHandoff>;
...@@ -28,7 +32,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH...@@ -28,7 +32,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
28 handoff.delivery.status === "pending",32 handoff.delivery.status === "pending",
29 );33 );
30 if (pending.length === 0 || (input.deferInline && input.inject === undefined))34 if (pending.length === 0 || (input.deferInline && input.inject === undefined))
31 return { context: "", delivered: Effect.void };35 return { context: "", delivered: Effect.void, unsent: Effect.void };
32 const budget = typeof input.budget === "number" ? input.budget : yield* input.budget;36 const budget = typeof input.budget === "number" ? input.budget : yield* input.budget;
33 let coverage = pending37 let coverage = pending
34 .map(38 .map(
...@@ -71,7 +75,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH...@@ -71,7 +75,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
71 budget,75 budget,
72 });76 });
73 if (historyCost(selected.messages, selected.context) > budget) {77 if (historyCost(selected.messages, selected.context) > budget) {
74 if (input.deferInline) return { context: "", delivered: Effect.void };78 if (input.deferInline) return { context: "", delivered: Effect.void, unsent: Effect.void };
75 return yield* new ContextHandoffBudgetError();79 return yield* new ContextHandoffBudgetError();
76 }80 }
77 const omittedItemIds = new Set(selected.omittedItemIds);81 const omittedItemIds = new Set(selected.omittedItemIds);
...@@ -122,7 +126,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH...@@ -122,7 +126,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
122 });126 });
123 if (injected) {127 if (injected) {
124 yield* persist("injected");128 yield* persist("injected");
125 return { context: "", delivered: Effect.void };129 return { context: "", delivered: Effect.void, unsent: Effect.void };
126 }130 }
127 } else {131 } else {
128 // Text-only delivery can also be accepted before a connection drops.132 // Text-only delivery can also be accepted before a connection drops.
...@@ -132,11 +136,12 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH...@@ -132,11 +136,12 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
132 // Compaction APIs cannot accept an inline transcript. Keep it available136 // Compaction APIs cannot accept an inline transcript. Keep it available
133 // for the next ordinary turn when native injection is unsupported.137 // for the next ordinary turn when native injection is unsupported.
134 yield* Effect.forEach(pending, input.persist, { discard: true });138 yield* Effect.forEach(pending, input.persist, { discard: true });
135 return { context: "", delivered: Effect.void };139 return { context: "", delivered: Effect.void, unsent: Effect.void };
136 }140 }
137 return {141 return {
138 context: renderHistory(selected.messages, selected.context),142 context: renderHistory(selected.messages, selected.context),
139 delivered: persist("inline"),143 delivered: persist("inline"),
144 unsent: Effect.forEach(pending, input.persist, { discard: true }),
140 };145 };
141 },146 },
142);147);
apps/server/src/orchestration-v2/ProviderTurnStartService.ts+21-1
...@@ -19,6 +19,7 @@ import * as Effect from "effect/Effect";...@@ -19,6 +19,7 @@ import * as Effect from "effect/Effect";
19import * as FileSystem from "effect/FileSystem";19import * as FileSystem from "effect/FileSystem";
20import * as Layer from "effect/Layer";20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";21import * as Option from "effect/Option";
22import * as Predicate from "effect/Predicate";
22import * as Schema from "effect/Schema";23import * as Schema from "effect/Schema";
2324
24import * as GitWorkflowService from "../git/GitWorkflowService.ts";25import * as GitWorkflowService from "../git/GitWorkflowService.ts";
...@@ -64,6 +65,11 @@ export class ProviderTurnStartError extends Schema.TaggedError<ProviderTurnStart...@@ -64,6 +65,11 @@ export class ProviderTurnStartError extends Schema.TaggedError<ProviderTurnStart
6465
65const isProviderTurnStartError = Schema.is(ProviderTurnStartError);66const isProviderTurnStartError = Schema.is(ProviderTurnStartError);
6667
68/** Claude refuses to replace a process running background work before it reads the prompt. */
69const refusedBeforePrompt = (error: unknown): boolean =>
70 Predicate.isTagged(error, "ClaudeBackgroundWorkBlocksQueryReplacementError") ||
71 (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause));
72
67export interface ProviderTurnStartServiceV2Shape {73export interface ProviderTurnStartServiceV2Shape {
68 /**74 /**
69 * Starts the run's provider turn. When `willRetry` is true, a session open75 * Starts the run's provider turn. When `willRetry` is true, a session open
...@@ -1176,7 +1182,21 @@ export const layer: Layer.Layer<...@@ -1176,7 +1182,21 @@ export const layer: Layer.Layer<
1176 ...turnInput.message,1182 ...turnInput.message,
1177 text: context === "" ? userText : `${context}\n\nUser message:\n${userText}`,1183 text: context === "" ? userText : `${context}\n\nUser message:\n${userText}`,
1178 },1184 },
1179 });1185 }).pipe(
1186 // A pending marker would make the next turn abandon this native
1187 // session, though the refused prompt never reached it.
1188 Effect.tapError((error) =>
1189 refusedBeforePrompt(error)
1190 ? delivery.unsent.pipe(
1191 Effect.catchCause(() =>
1192 Effect.logWarning("Failed to restore unsent context handoffs", {
1193 runId: run.id,
1194 }),
1195 ),
1196 )
1197 : Effect.void,
1198 ),
1199 );
1180 // The provider already accepted the turn. A stale pending marker1200 // The provider already accepted the turn. A stale pending marker
1181 // can force a fresh thread later, but must not stop live ingestion.1201 // can force a fresh thread later, but must not stop live ingestion.
1182 yield* delivery.delivered.pipe(1202 yield* delivery.delivered.pipe(
apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts+113-1
...@@ -34,7 +34,10 @@ import * as SqlClient from "effect/sql/SqlClient";...@@ -34,7 +34,10 @@ import * as SqlClient from "effect/sql/SqlClient";
3434
35import * as SqlitePersistence from "../../persistence/Sqlite.ts";35import * as SqlitePersistence from "../../persistence/Sqlite.ts";
36import { CommandPolicyCapabilityUnsupportedError } from "../CommandPolicy.ts";36import { CommandPolicyCapabilityUnsupportedError } from "../CommandPolicy.ts";
37import { ClaudeProviderCapabilitiesV2 } from "../Adapters/ClaudeAdapterV2.ts";37import {
38 ClaudeBackgroundWorkBlocksQueryReplacementError,
39 ClaudeProviderCapabilitiesV2,
40} from "../Adapters/ClaudeAdapterV2.ts";
38import {41import {
39 CodexProviderCapabilitiesV2,42 CodexProviderCapabilitiesV2,
40 canReuseCodexContextUsage,43 canReuseCodexContextUsage,
...@@ -53,6 +56,7 @@ import {...@@ -53,6 +56,7 @@ import {
53 type ProviderAdapterV2Event,56 type ProviderAdapterV2Event,
54 type ProviderAdapterV2HistoricalContext,57 type ProviderAdapterV2HistoricalContext,
55 ProviderAdapterProtocolError,58 ProviderAdapterProtocolError,
59 ProviderAdapterTurnStartError,
56 type ProviderAdapterV2Shape,60 type ProviderAdapterV2Shape,
57 type ProviderAdapterV2SessionRuntime,61 type ProviderAdapterV2SessionRuntime,
58} from "../ProviderAdapter.ts";62} from "../ProviderAdapter.ts";
...@@ -104,6 +108,8 @@ function makeTestAdapter(input: {...@@ -104,6 +108,8 @@ function makeTestAdapter(input: {
104 readonly capturedTurns: Ref.Ref<ReadonlyArray<CapturedTurn>>;108 readonly capturedTurns: Ref.Ref<ReadonlyArray<CapturedTurn>>;
105 readonly injectedHistory?: Ref.Ref<ReadonlyArray<unknown>>;109 readonly injectedHistory?: Ref.Ref<ReadonlyArray<unknown>>;
106 readonly failStartOnce?: Ref.Ref<boolean>;110 readonly failStartOnce?: Ref.Ref<boolean>;
111 /** Starts left to refuse the way Claude does while background work runs. */
112 readonly refuseStarts?: Ref.Ref<number>;
107 readonly failInjectionOnce?: Ref.Ref<boolean>;113 readonly failInjectionOnce?: Ref.Ref<boolean>;
108 readonly nativeThreadGeneration?: Ref.Ref<number>;114 readonly nativeThreadGeneration?: Ref.Ref<number>;
109 readonly failResume?: boolean;115 readonly failResume?: boolean;
...@@ -226,6 +232,17 @@ function makeTestAdapter(input: {...@@ -226,6 +232,17 @@ function makeTestAdapter(input: {
226 (yield* Ref.getAndSet(input.failStartOnce, false))232 (yield* Ref.getAndSet(input.failStartOnce, false))
227 )233 )
228 return yield* unimplemented(input.driver, "turn start failed after injection");234 return yield* unimplemented(input.driver, "turn start failed after injection");
235 if (
236 input.refuseStarts !== undefined &&
237 (yield* Ref.getAndUpdate(input.refuseStarts, (left) => Math.max(0, left - 1))) > 0
238 )
239 return yield* new ProviderAdapterTurnStartError({
240 driver: input.driver,
241 threadId: turnInput.threadId,
242 providerThreadId: turnInput.providerThread.id,
243 runId: turnInput.runId,
244 cause: new ClaudeBackgroundWorkBlocksQueryReplacementError(),
245 });
229 yield* Effect.yieldNow;246 yield* Effect.yieldNow;
230 yield* Ref.update(input.capturedTurns, (turns) => [247 yield* Ref.update(input.capturedTurns, (turns) => [
231 ...turns,248 ...turns,
...@@ -1188,6 +1205,101 @@ describe("orchestration v2 provider switching", () => {...@@ -1188,6 +1205,101 @@ describe("orchestration v2 provider switching", () => {
1188 ),1205 ),
1189 );1206 );
11901207
1208 it.live("keeps the native session after turns refused before reaching the provider", () =>
1209 Effect.scoped(
1210 Effect.gen(function* () {
1211 const cwd = yield* checkpointWorkspace("refused-start-keeps-native");
1212 const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]);
1213 const refuseStarts = yield* Ref.make(0);
1214 const generation = yield* Ref.make(0);
1215 const registry = ProviderAdapterRegistry.makeLayer([
1216 makeTestAdapter({
1217 instanceId: CLAUDE_MODEL_SELECTION.instanceId,
1218 driver: CLAUDE_DRIVER,
1219 capabilities: ClaudeProviderCapabilitiesV2,
1220 modelSelection: CLAUDE_MODEL_SELECTION,
1221 responseByRunOrdinal: {},
1222 capturedTurns,
1223 refuseStarts,
1224 nativeThreadGeneration: generation,
1225 }),
1226 ]);
1227 yield* Effect.gen(function* () {
1228 const orchestrator = yield* Orchestrator.OrchestratorV2;
1229 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
1230 const run = Effect.fn("run")(function* (ordinal: number) {
1231 yield* orchestrator.dispatch({
1232 type: "message.dispatch",
1233 commandId: CommandId.make(`refused-start:${ordinal}`),
1234 threadId,
1235 messageId: MessageId.make(`refused-start:${ordinal}`),
1236 createdBy: "user",
1237 creationSource: "web",
1238 text: `Request ${ordinal}`,
1239 attachments: [],
1240 modelSelection: CLAUDE_MODEL_SELECTION,
1241 dispatchMode: { type: "start_immediately" },
1242 });
1243 yield* orchestrator.streamStoredEvents.pipe(
1244 Stream.filter(
1245 ({ event }) =>
1246 event.type === "run.updated" &&
1247 event.payload.ordinal === ordinal &&
1248 (event.payload.status === "completed" || event.payload.status === "failed"),
1249 ),
1250 Stream.runHead,
1251 );
1252 yield* worker.drain();
1253 return (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)?.status;
1254 });
1255 yield* orchestrator.dispatch({
1256 type: "thread.create",
1257 commandId: CommandId.make("refused-start:create"),
1258 threadId,
1259 projectId,
1260 createdBy: "user",
1261 creationSource: "web",
1262 title: "Refused start",
1263 modelSelection: CLAUDE_MODEL_SELECTION,
1264 runtimeMode: "full-access",
1265 interactionMode: "default",
1266 branch: null,
1267 worktreePath: null,
1268 });
1269 assert.equal(yield* run(1), "completed");
1270 // Run 3 carries run 2's missed request as a handoff and is refused too.
1271 yield* Ref.set(refuseStarts, 2);
1272 assert.equal(yield* run(2), "failed");
1273 assert.equal(yield* run(3), "failed");
1274 assert.equal(yield* run(4), "completed");
1275
1276 const turns = yield* Ref.get(capturedTurns);
1277 assert.equal(turns.length, 2);
1278 // Nothing reached the provider, so the next turn continues the same
1279 // native session instead of replacing it with a summary.
1280 assert.equal(yield* Ref.get(generation), 1);
1281 assert.equal(turns[1]?.nativeThreadId, turns[0]?.nativeThreadId);
1282 assert.include(turns[1]?.text, "Request 2");
1283 assert.include(turns[1]?.text, "Request 3");
1284 }).pipe(
1285 Effect.provide(
1286 makeOrchestratorV2ReplayLayerWithRegistry(
1287 {
1288 name: "refused-start-keeps-native",
1289 runtimePolicyOverride: {
1290 cwd,
1291 approvalPolicy: "never",
1292 sandboxPolicy: { type: "readOnly" },
1293 },
1294 },
1295 registry,
1296 ),
1297 ),
1298 );
1299 }),
1300 ),
1301 );
1302
1191 it.live.each(1303 it.live.each(
1192 (["failed", "interrupted"] as const).flatMap((status) =>1304 (["failed", "interrupted"] as const).flatMap((status) =>
1193 [false, true].flatMap((queued) =>1305 [false, true].flatMap((queued) =>