Skip to content

Commit e69ab40

Browse files
Marve10st3dotgg
authored andcommitted
fix(server): resume fallback keeps the latest turn when an older handoff is pending
1 parent e822ba7 commit e69ab40

2 files changed

Lines changed: 127 additions & 1 deletion

File tree

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

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -747,7 +747,14 @@ export const layer: Layer.Layer<
747747
),
748748
createdAt,
749749
});
750-
effectiveHandoffs = [handoff, ...effectiveHandoffs];
750+
// Delivery ranks later handoffs as newer history. Pending handoffs from
751+
// earlier runs predate this summary; handoffs for this run (provider-switch
752+
// delta, merge-back) are newer and must stay last.
753+
effectiveHandoffs = [
754+
...effectiveHandoffs.filter((candidate) => candidate.targetRunId !== run.id),
755+
handoff,
756+
...effectiveHandoffs.filter((candidate) => candidate.targetRunId === run.id),
757+
];
751758
yield* eventSink.write({
752759
events: [
753760
{

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

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

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

0 commit comments

Comments
 (0)