Skip to content

Commit 32bf48f

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
authored
fix(server): preserve settled runs during delegated acknowledgements (#25672)
Co-authored-by: Claude Opus 5.5 <[email protected]>
1 parent fe867b8 commit 32bf48f

3 files changed

Lines changed: 141 additions & 23 deletions

File tree

‎apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -974,6 +974,104 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => {
974974
);
975975
}),
976976
);
977+
978+
it.effect.each([
979+
{ before: "running", after: "failed", capture: false },
980+
{ before: "waiting", after: "completed", capture: true },
981+
] as const)(
982+
"keeps a $after run that settled between task_status's acknowledge read and commit",
983+
({ before, after, capture }) =>
984+
Effect.gen(function* () {
985+
const orchestrator = yield* Orchestrator.OrchestratorV2;
986+
const sink = yield* EventSink.EventSinkV2;
987+
const now = yield* DateTime.now;
988+
const threadId = ThreadId.make(`thread:ack-settle-race-${after}`);
989+
const runId = RunId.make(`run:ack-settle-race-${after}`);
990+
const taskId = NodeId.make(`node:ack-settle-race-${after}-task`);
991+
const checkpointId = CheckpointId.make(`checkpoint:ack-settle-race-${after}`);
992+
yield* seedParentWithTerminalTask({
993+
threadId,
994+
projectId: ProjectId.make(`project:ack-settle-race-${after}`),
995+
runId,
996+
rootNodeId: NodeId.make(`node:ack-settle-race-${after}-root`),
997+
taskId,
998+
deliveryState: "claimed",
999+
deliveryTaskIds: [taskId],
1000+
now,
1001+
});
1002+
const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
1003+
yield* sink.write({
1004+
events: [
1005+
{
1006+
id: EventId.make(`event:ack-settle-race-${after}-before`),
1007+
type: "run.updated",
1008+
threadId,
1009+
runId,
1010+
occurredAt: now,
1011+
payload: { ...parent, status: before },
1012+
},
1013+
],
1014+
});
1015+
const { delegatedCompletion: _cohort, ...settledRun } = parent;
1016+
const commit = sink.commitCommand;
1017+
let settled = false;
1018+
yield* Effect.acquireUseRelease(
1019+
Effect.sync(() =>
1020+
vi.spyOn(sink, "commitCommand").mockImplementation((input) =>
1021+
Effect.gen(function* () {
1022+
if (
1023+
!settled &&
1024+
input.commandType === "delegated_task.completion-delivery.acknowledge"
1025+
) {
1026+
settled = true;
1027+
// The provider turn ends (or its checkpoint lands) after acknowledge read the run.
1028+
yield* commit({
1029+
commandId: CommandId.make(`command:ack-settle-race-${after}`),
1030+
threadId,
1031+
commandType: capture ? "checkpoint.capture" : "provider.event",
1032+
acceptedAt: now,
1033+
effects: [],
1034+
events: [
1035+
{
1036+
id: EventId.make(`event:ack-settle-race-${after}-settled`),
1037+
type: "run.updated",
1038+
threadId,
1039+
runId,
1040+
occurredAt: now,
1041+
payload: {
1042+
...settledRun,
1043+
status: after,
1044+
completedAt: now,
1045+
checkpointId: capture ? checkpointId : null,
1046+
},
1047+
},
1048+
],
1049+
});
1050+
}
1051+
return yield* commit(input);
1052+
}),
1053+
),
1054+
),
1055+
() =>
1056+
orchestrator.dispatch({
1057+
type: "delegated_task.completion-delivery.acknowledge",
1058+
commandId: CommandId.make(`command:ack-settle-race-${after}:task-status`),
1059+
parentThreadId: threadId,
1060+
taskId,
1061+
observedByRunId: runId,
1062+
}),
1063+
(spy) => Effect.sync(() => spy.mockRestore()),
1064+
);
1065+
assert.isTrue(settled);
1066+
const run = (yield* orchestrator.getThreadProjection(threadId)).runs.find(
1067+
(candidate) => candidate.id === runId,
1068+
)!;
1069+
assert.deepEqual(
1070+
{ status: run.status, completedAt: run.completedAt, checkpointId: run.checkpointId },
1071+
{ status: after, completedAt: now, checkpointId: capture ? checkpointId : null },
1072+
);
1073+
}),
1074+
);
9771075
});
9781076

9791077
const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`);

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

Lines changed: 31 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,8 @@ export interface EventSinkV2Shape {
146146
readonly effectTypes: ReadonlyArray<EffectOutbox.OrchestrationEffectRequestV2["type"]>;
147147
readonly reason: string;
148148
};
149+
/** Like writeWithEffects' updateDelegatedCompletionOnly, for this run's run.updated only. */
150+
readonly updateDelegatedCompletionOnlyRunId?: RunId;
149151
}) => Effect.Effect<
150152
{
151153
readonly receipt: CommandReceiptStore.CommandReceiptV2;
@@ -383,6 +385,28 @@ const layerBase: Layer.Layer<
383385
return yield* runOutcomes.prepare(storedEvents);
384386
});
385387

388+
// The enclosing write or commit span already traces this per-event projection merge.
389+
const updateDelegatedCompletionOnly = Effect.fnUntraced(function* (
390+
events: ReadonlyArray<OrchestrationV2DomainEvent>,
391+
includesRun: (runId: RunId) => boolean,
392+
) {
393+
return yield* Effect.forEach(events, (event) =>
394+
Effect.gen(function* () {
395+
if (event.type !== "run.updated" || !includesRun(event.payload.id)) return event;
396+
const { runs } = yield* projectionStore.getThreadRecords(event.threadId, ["runs"], {
397+
runIds: [event.payload.id],
398+
});
399+
const current = runs[0];
400+
return current === undefined
401+
? event
402+
: {
403+
...event,
404+
payload: { ...current, delegatedCompletion: event.payload.delegatedCompletion },
405+
};
406+
}),
407+
);
408+
});
409+
386410
const writeEffect = Effect.fn("orchestrationV2.EventSink.write")(function* (
387411
input: Parameters<EventSinkV2Shape["writeWithEffects"]>[0],
388412
) {
@@ -411,26 +435,7 @@ const layerBase: Layer.Layer<
411435
// Checkpoint capture can finish after a cohort writer reads its parent run.
412436
// Merge in this transaction so the event log and projection keep that completion.
413437
const events = input.updateDelegatedCompletionOnly
414-
? yield* Effect.forEach(input.events, (event) =>
415-
Effect.gen(function* () {
416-
if (event.type !== "run.updated") return event;
417-
const { runs } = yield* projectionStore.getThreadRecords(
418-
event.threadId,
419-
["runs"],
420-
{ runIds: [event.payload.id] },
421-
);
422-
const current = runs[0];
423-
return current === undefined
424-
? event
425-
: {
426-
...event,
427-
payload: {
428-
...current,
429-
delegatedCompletion: event.payload.delegatedCompletion,
430-
},
431-
};
432-
}),
433-
)
438+
? yield* updateDelegatedCompletionOnly(input.events, () => true)
434439
: input.events;
435440
const normalized = yield* normalizeEvents(
436441
input.guardPendingUserInputCancellations === true
@@ -646,7 +651,12 @@ const layerBase: Layer.Layer<
646651
return { ...existing, committed: false as const, cancelledEffectIds: [] };
647652
}
648653

649-
const normalized = yield* normalizeEvents(input.events);
654+
const runId = input.updateDelegatedCompletionOnlyRunId;
655+
const normalized = yield* normalizeEvents(
656+
runId === undefined
657+
? input.events
658+
: yield* updateDelegatedCompletionOnly(input.events, (id) => id === runId),
659+
);
650660
const storedEvents = yield* eventStore.append({
651661
commandId: input.commandId,
652662
events: normalized,

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

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2358,7 +2358,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
23582358
run: deliveryRun,
23592359
now,
23602360
});
2361-
return;
2361+
return parentRun.id;
23622362
}
23632363
const message = completionDeliveryMessage(projection, delivery);
23642364
if (message !== undefined) {
@@ -2382,6 +2382,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
23822382
});
23832383
}
23842384
}
2385+
return parentRun.id;
23852386
});
23862387

23872388
const disposeDelegatedCompletionCohort = Effect.fn(
@@ -11752,6 +11753,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1175211753
readonly effectTypes: ReadonlyArray<OrchestrationEffectRequestV2["type"]>;
1175311754
readonly reason: string;
1175411755
};
11756+
readonly updateDelegatedCompletionOnlyRunId?: RunId;
1175511757
},
1175611758
OrchestratorV2Error
1175711759
> {
@@ -11769,6 +11771,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1176911771
readonly reason: string;
1177011772
}
1177111773
| undefined;
11774+
let updateDelegatedCompletionOnlyRunId: RunId | undefined;
1177211775
switch (command.type) {
1177311776
case "thread.create":
1177411777
yield* dispatchThreadCreate(command, events);
@@ -12080,7 +12083,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1208012083
break;
1208112084
case "delegated_task.completion-delivery.acknowledge":
1208212085
case "delegated_task.completion-delivery.dispose":
12083-
yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events);
12086+
updateDelegatedCompletionOnlyRunId =
12087+
yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events);
1208412088
break;
1208512089
case "thread.created.record":
1208612090
yield* dispatchCreatedThreadRecord(command, events);
@@ -12095,6 +12099,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1209512099
events: yield* Ref.get(events),
1209612100
effects: yield* Ref.get(effects),
1209712101
...(cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects }),
12102+
...(updateDelegatedCompletionOnlyRunId === undefined
12103+
? {}
12104+
: { updateDelegatedCompletionOnlyRunId }),
1209812105
};
1209912106
});
1210012107

@@ -12271,6 +12278,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1227112278
...(plan.cancelUnsettledEffects === undefined
1227212279
? {}
1227312280
: { cancelUnsettledEffects: plan.cancelUnsettledEffects }),
12281+
...(plan.updateDelegatedCompletionOnlyRunId === undefined
12282+
? {}
12283+
: { updateDelegatedCompletionOnlyRunId: plan.updateDelegatedCompletionOnlyRunId }),
1227412284
})
1227512285
.pipe(
1227612286
Effect.mapError(

0 commit comments

Comments
 (0)