Skip to content

Commit 7ee8a5b

Browse files
t3dotggMarve10sclaude
authored
fix(server): resume fallback keeps the latest turn when an older handoff is pending (#22902)
Co-authored-by: Marve10s <[email protected]> Co-authored-by: Claude Opus 5.5 <[email protected]>
1 parent 2ed9261 commit 7ee8a5b

2 files changed

Lines changed: 139 additions & 1 deletion

File tree

‎apps/server/src/orchestration-v2/ProviderTurnStartService.ts‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -693,6 +693,7 @@ export const layer: Layer.Layer<
693693
return undefined;
694694
});
695695
let effectiveHandoffs = handoffs;
696+
let resumeFellBack = false;
696697
const loadedProviderThread = yield* Effect.gen(function* () {
697698
if (nativeForkTransfer !== undefined) {
698699
const sourceProjection = yield* projectionStore.getThreadRecords(
@@ -818,7 +819,15 @@ export const layer: Layer.Layer<
818819
),
819820
createdAt,
820821
});
821-
effectiveHandoffs = [handoff, ...effectiveHandoffs];
822+
// Delivery ranks later handoffs as newer history. Pending handoffs from
823+
// earlier runs predate this summary; handoffs for this run (provider-switch
824+
// delta, merge-back) are newer and must stay last.
825+
effectiveHandoffs = [
826+
...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id),
827+
handoff,
828+
...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id),
829+
];
830+
resumeFellBack = true;
822831
yield* eventSink.write({
823832
events: [
824833
{
@@ -1145,8 +1154,11 @@ export const layer: Layer.Layer<
11451154
...runningProviderThread,
11461155
contextUsage: handoffUsage,
11471156
};
1157+
// The resume fallback summary already covers every earlier run in order. A
1158+
// retry copy after it would rank older missed requests as the newest history.
11481159
const missedRuns = projection.runs.filter(
11491160
(source) =>
1161+
!resumeFellBack &&
11501162
source.ordinal < run.ordinal &&
11511163
source.providerThreadId === providerThread.id &&
11521164
(source.status === "failed" || source.status === "interrupted") &&

‎apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1182,6 +1182,132 @@ describe("orchestration v2 provider switching", () => {
11821182
),
11831183
);
11841184

1185+
it.live.each(["an older handoff", "a missed request"] as const)(
1186+
"keeps the latest turn when a resume fallback also delivers %s",
1187+
(older) =>
1188+
Effect.scoped(
1189+
Effect.gen(function* () {
1190+
// A run that fails before its provider turn starts is retried as a missed request.
1191+
const missed = older === "a missed request";
1192+
const name = `resume-fallback-handoff-order${missed ? "-missed" : ""}`;
1193+
const cwd = yield* checkpointWorkspace(name);
1194+
const failStartOnce = yield* Ref.make(false);
1195+
const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]);
1196+
const injectedHistory = yield* Ref.make<ReadonlyArray<unknown>>([]);
1197+
const oldResult = `old result ${"o".repeat(2_500)}`;
1198+
const latestResult = `latest result ${"l".repeat(2_500)}`;
1199+
const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([
1200+
makeTestAdapter({
1201+
instanceId: CLAUDE_MODEL_SELECTION.instanceId,
1202+
driver: CLAUDE_DRIVER,
1203+
capabilities: ClaudeProviderCapabilitiesV2,
1204+
modelSelection: CLAUDE_MODEL_SELECTION,
1205+
responseByRunOrdinal: { 1: oldResult, 3: latestResult },
1206+
capturedTurns,
1207+
injectedHistory,
1208+
failResume: true,
1209+
...(missed ? { failStartOnce } : { failedRunOrdinals: new Set([2]) }),
1210+
nativeThreadGeneration: yield* Ref.make(0),
1211+
}),
1212+
]);
1213+
yield* Effect.gen(function* () {
1214+
const orchestrator = yield* Orchestrator.OrchestratorV2;
1215+
const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
1216+
const runAfterDetach = Effect.fn("runAfterDetach")(function* (
1217+
ordinal: number,
1218+
text: string,
1219+
) {
1220+
const providerSessionId = (yield* orchestrator.getThreadProjection(threadId))
1221+
.providerThreads[0]?.providerSessionId;
1222+
if (providerSessionId != null) {
1223+
yield* orchestrator.dispatch({
1224+
type: "provider-session.detach",
1225+
commandId: CommandId.make(`handoff-order:detach:${ordinal}`),
1226+
threadId,
1227+
providerSessionId,
1228+
});
1229+
yield* worker.drain();
1230+
}
1231+
yield* orchestrator.dispatch({
1232+
type: "message.dispatch",
1233+
commandId: CommandId.make(`handoff-order:${ordinal}`),
1234+
threadId,
1235+
messageId: MessageId.make(`handoff-order:${ordinal}`),
1236+
createdBy: "user",
1237+
creationSource: "web",
1238+
text,
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+
});
1254+
yield* orchestrator.dispatch({
1255+
type: "thread.create",
1256+
commandId: CommandId.make("handoff-order:create"),
1257+
threadId,
1258+
projectId,
1259+
createdBy: "user",
1260+
creationSource: "web",
1261+
title: "Resume fallback handoff order",
1262+
modelSelection: CLAUDE_MODEL_SELECTION,
1263+
runtimeMode: "full-access",
1264+
interactionMode: "default",
1265+
branch: null,
1266+
worktreePath: null,
1267+
});
1268+
yield* runAfterDetach(1, `Original brief ${"b".repeat(9_000)}`);
1269+
yield* Ref.set(failStartOnce, missed);
1270+
yield* runAfterDetach(2, missed ? `Request 2 ${"r".repeat(6_000)}` : "Request 2");
1271+
yield* runAfterDetach(3, "Request 3");
1272+
const historyBeforeRun4 = (yield* Ref.get(injectedHistory)).length;
1273+
yield* runAfterDetach(4, "Request 4");
1274+
1275+
const projection = yield* orchestrator.getThreadProjection(threadId);
1276+
assert.deepEqual(
1277+
projection.runs.map((run) => run.status),
1278+
["completed", "failed", "completed", "completed"],
1279+
);
1280+
const run4NativeThreadId = (yield* Ref.get(capturedTurns)).at(-1)?.nativeThreadId;
1281+
assert.equal(
1282+
projection.contextHandoffs.find(
1283+
(handoff) => handoff.targetRunId === projection.runs[1]?.id,
1284+
)?.delivery?.nativeThreadId,
1285+
run4NativeThreadId,
1286+
);
1287+
const run4History = yield* encodeJson(
1288+
(yield* Ref.get(injectedHistory)).slice(historyBeforeRun4),
1289+
);
1290+
assert.include(run4History, "Original brief");
1291+
assert.include(run4History, latestResult);
1292+
}).pipe(
1293+
Effect.provide(
1294+
ProviderReplayHarness.layerWithRegistry(
1295+
{
1296+
name,
1297+
runtimePolicyOverride: {
1298+
cwd,
1299+
approvalPolicy: "never",
1300+
sandboxPolicy: { type: "readOnly" },
1301+
},
1302+
},
1303+
layerRegistry,
1304+
),
1305+
),
1306+
);
1307+
}),
1308+
),
1309+
);
1310+
11851311
it.live.each([
11861312
["background work", new ClaudeBackgroundWorkBlocksQueryReplacementError()],
11871313
["an unreadable append file", new ClaudeAppendSystemPromptFileError({ path: "missing.md" })],

0 commit comments

Comments
 (0)