Skip to content

Commit 8625d7a

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
committed
fix(server): a withdrawn approval no longer erases a committed answer (#24690)
An approval answer commits before its outbox delivery reaches the provider. If Claude withdraws the request in that gap, the adapter's cancellation events overwrote the resolved request, losing the decision and feedback and flipping the node and item outcome. The EventSink cancellation guard now covers approvals too, and checks node and item cancellations against their request, since the adapter writes each in its own batch. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 5b99f0b commit 8625d7a

3 files changed

Lines changed: 189 additions & 23 deletions

File tree

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

Lines changed: 45 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ export interface EventSinkV2Shape {
7373
OrchestrationV2ProviderThread,
7474
"driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
7575
>;
76-
readonly guardPendingUserInputCancellations?: boolean;
76+
readonly guardPendingRuntimeRequestCancellations?: boolean;
7777
readonly commandId?: CommandId;
7878
readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
7979
}) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
@@ -87,13 +87,13 @@ export interface EventSinkV2Shape {
8787
OrchestrationV2ProviderThread,
8888
"driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
8989
>;
90-
readonly guardPendingUserInputCancellations?: boolean;
90+
readonly guardPendingRuntimeRequestCancellations?: boolean;
9191
readonly commandId?: CommandId;
9292
readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
9393
readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
9494
}) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
9595
readonly writeIfRunCurrent: (input: {
96-
readonly guardPendingUserInputCancellations?: boolean;
96+
readonly guardPendingRuntimeRequestCancellations?: boolean;
9797
readonly commandId?: CommandId;
9898
readonly threadId: ThreadId;
9999
readonly runId: RunId;
@@ -110,7 +110,7 @@ export interface EventSinkV2Shape {
110110
>;
111111
/** Commits only if the expected attempt and ordinal still own the provider thread, protecting newer attempts. */
112112
readonly writeIfProviderThreadOwner: (input: {
113-
readonly guardPendingUserInputCancellations?: boolean;
113+
readonly guardPendingRuntimeRequestCancellations?: boolean;
114114
readonly commandId?: CommandId;
115115
readonly providerThreadId: ProviderThreadId;
116116
readonly runId: RunId;
@@ -278,29 +278,53 @@ const layerBase: Layer.Layer<
278278
);
279279
});
280280

281-
// A user can answer after terminal normalization reads the pending request.
281+
// A user can answer after terminal normalization reads the pending request, and an approval
282+
// answer commits before its outbox delivery reaches a provider that may withdraw the request.
282283
// Recheck inside the write transaction so stale cleanup cannot erase answers.
283-
const guardUserInputCancellations =
284+
const guardRuntimeRequestCancellations =
284285
/* This helper runs for individual orchestration events; the enclosing operation owns the span. */
285286
Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2DomainEvent>) {
286287
const staleRequests = new Set<RuntimeRequestId>();
287288
const staleNodes = new Set<NodeId>();
288289
for (const event of events) {
290+
// Providers can withdraw the node and item in writes of their own, so check each against its request.
289291
if (
290-
event.type !== "runtime-request.updated" ||
291-
event.payload.kind !== "user_input" ||
292-
event.payload.status !== "cancelled"
293-
)
292+
event.type === "node.updated" &&
293+
event.payload.status === "cancelled" &&
294+
event.payload.runtimeRequestId !== null
295+
) {
296+
const current = yield* projectionStore.getRuntimeRequest(
297+
event.threadId,
298+
event.payload.runtimeRequestId,
299+
);
300+
if (current?.status === "resolved") staleNodes.add(event.payload.id);
301+
continue;
302+
}
303+
if (
304+
event.type === "turn-item.updated" &&
305+
(event.payload.type === "user_input_request" ||
306+
event.payload.type === "approval_request") &&
307+
event.payload.status === "cancelled"
308+
) {
309+
const current = yield* projectionStore.getRuntimeRequest(
310+
event.threadId,
311+
event.payload.requestId,
312+
);
313+
if (current?.status === "resolved") staleRequests.add(event.payload.requestId);
314+
continue;
315+
}
316+
if (event.type !== "runtime-request.updated" || event.payload.status !== "cancelled")
294317
continue;
295318
const current = yield* projectionStore.getRuntimeRequest(
296319
event.threadId,
297320
event.payload.id,
298321
);
299322
if (
300323
current?.status !== "pending" ||
301-
current.kind !== "user_input" ||
302-
current.providerTurnId !== event.payload.providerTurnId ||
303-
current.responseCapability.type === "message"
324+
(event.payload.kind === "user_input" &&
325+
(current.kind !== "user_input" ||
326+
current.providerTurnId !== event.payload.providerTurnId ||
327+
current.responseCapability.type === "message"))
304328
) {
305329
staleRequests.add(event.payload.id);
306330
staleNodes.add(event.payload.nodeId);
@@ -314,7 +338,8 @@ const layerBase: Layer.Layer<
314338
return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
315339
case "turn-item.updated":
316340
return (
317-
event.payload.type !== "user_input_request" ||
341+
(event.payload.type !== "user_input_request" &&
342+
event.payload.type !== "approval_request") ||
318343
event.payload.status !== "cancelled" ||
319344
!staleRequests.has(event.payload.requestId)
320345
);
@@ -430,8 +455,8 @@ const layerBase: Layer.Layer<
430455
)
431456
: input.events;
432457
const normalized = yield* normalizeEvents(
433-
input.guardPendingUserInputCancellations === true
434-
? yield* guardUserInputCancellations(events)
458+
input.guardPendingRuntimeRequestCancellations === true
459+
? yield* guardRuntimeRequestCancellations(events)
435460
: events,
436461
);
437462
const committed = yield* eventStore.append({
@@ -489,8 +514,8 @@ const layerBase: Layer.Layer<
489514
}
490515

491516
const normalized = yield* normalizeEvents(
492-
input.guardPendingUserInputCancellations === true
493-
? yield* guardUserInputCancellations(input.events)
517+
input.guardPendingRuntimeRequestCancellations === true
518+
? yield* guardRuntimeRequestCancellations(input.events)
494519
: input.events,
495520
);
496521
const storedEvents = yield* eventStore.append({
@@ -554,8 +579,8 @@ const layerBase: Layer.Layer<
554579
}
555580

556581
const normalized = yield* normalizeEvents(
557-
input.guardPendingUserInputCancellations === true
558-
? yield* guardUserInputCancellations(input.events)
582+
input.guardPendingRuntimeRequestCancellations === true
583+
? yield* guardRuntimeRequestCancellations(input.events)
559584
: input.events,
560585
);
561586
const storedEvents = yield* eventStore.append({

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

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -468,6 +468,147 @@ layer("ProviderEventIngestorV2", (it) => {
468468
}),
469469
);
470470

471+
// regression: #24690 an answer commits before its outbox delivery reaches the provider
472+
it.effect.each([
473+
{ decision: "decline", feedback: "Use a safer command." },
474+
{ decision: "accept" },
475+
] as const)("keeps a committed $decision when the provider withdraws the approval", (answer) =>
476+
Effect.gen(function* () {
477+
const now = yield* DateTime.now;
478+
const eventSink = yield* EventSink.EventSinkV2;
479+
const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
480+
const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
481+
const idAllocator = yield* IdAllocator.IdAllocatorV2;
482+
const threadEvent = yield* threadCreatedEvent(now);
483+
const threadId = threadEvent.threadId;
484+
yield* eventSink.write({ events: [threadEvent] });
485+
const providerSessionId = yield* idAllocator.allocate.providerSession({
486+
providerInstanceId: modelSelection.instanceId,
487+
threadId,
488+
});
489+
const nodeId = NodeId.make(`${threadId}:approval`);
490+
const request: OrchestrationV2RuntimeRequest = {
491+
id: RuntimeRequestId.make(`${threadId}:approval`),
492+
nodeId,
493+
providerTurnId: null,
494+
nativeRequestRef: null,
495+
kind: "command",
496+
status: "pending",
497+
responseCapability: { type: "live", providerSessionId, supportsFeedback: true },
498+
createdAt: now,
499+
resolvedAt: null,
500+
};
501+
const node: OrchestrationV2ExecutionNode = {
502+
id: nodeId,
503+
threadId,
504+
runId: null,
505+
parentNodeId: null,
506+
rootNodeId: nodeId,
507+
kind: "approval_request",
508+
status: "waiting",
509+
countsForRun: false,
510+
providerThreadId: null,
511+
providerTurnId: null,
512+
nativeItemRef: null,
513+
runtimeRequestId: request.id,
514+
checkpointScopeId: null,
515+
startedAt: now,
516+
completedAt: null,
517+
};
518+
const item: OrchestrationV2TurnItem = {
519+
id: TurnItemId.make(`${threadId}:approval`),
520+
threadId,
521+
runId: null,
522+
nodeId,
523+
providerThreadId: null,
524+
providerTurnId: null,
525+
nativeItemRef: null,
526+
parentItemId: null,
527+
ordinal: 0,
528+
status: "waiting",
529+
title: null,
530+
startedAt: now,
531+
completedAt: null,
532+
updatedAt: now,
533+
type: "approval_request",
534+
requestId: request.id,
535+
requestKind: "command",
536+
};
537+
const ingest = (status: "waiting" | "cancelled") =>
538+
Effect.forEach(
539+
[
540+
{
541+
type: "node.updated" as const,
542+
driver: CODEX_DRIVER,
543+
node: status === "waiting" ? node : { ...node, status, completedAt: now },
544+
},
545+
{
546+
type: "runtime_request.updated" as const,
547+
driver: CODEX_DRIVER,
548+
runtimeRequest:
549+
status === "waiting" ? request : { ...request, status, resolvedAt: now },
550+
},
551+
{
552+
type: "turn_item.updated" as const,
553+
driver: CODEX_DRIVER,
554+
turnItem: status === "waiting" ? item : { ...item, status, completedAt: now },
555+
},
556+
],
557+
(event) =>
558+
ingestor.ingestNormalized({
559+
providerSessionId,
560+
providerInstanceId: modelSelection.instanceId,
561+
threadId,
562+
event,
563+
}),
564+
{ concurrency: 1 },
565+
);
566+
yield* ingest("waiting");
567+
568+
// The orchestrator's response transaction; its outbox delivery never runs here.
569+
const outcome = answer.decision === "accept" ? "completed" : "cancelled";
570+
const resolved = { ...request, status: "resolved" as const, resolvedAt: now, ...answer };
571+
const answeredNode: OrchestrationV2ExecutionNode = {
572+
...node,
573+
status: outcome,
574+
completedAt: now,
575+
};
576+
const answeredItem: OrchestrationV2TurnItem = { ...item, status: outcome, completedAt: now };
577+
const base = { threadId, nodeId, occurredAt: now };
578+
yield* eventSink.write({
579+
events: [
580+
{
581+
...base,
582+
id: yield* idAllocator.allocate.event({ threadId }),
583+
type: "runtime-request.updated",
584+
payload: resolved,
585+
},
586+
{
587+
...base,
588+
id: yield* idAllocator.allocate.event({ threadId }),
589+
type: "node.updated",
590+
payload: answeredNode,
591+
},
592+
{
593+
...base,
594+
id: yield* idAllocator.allocate.event({ threadId }),
595+
type: "turn-item.updated",
596+
payload: answeredItem,
597+
},
598+
],
599+
});
600+
601+
yield* ingest("cancelled");
602+
const projection = yield* projectionStore.getThreadProjection(threadId);
603+
const stored = projection.runtimeRequests.find((entry) => entry.id === request.id);
604+
assert.equal(stored?.status, "resolved");
605+
assert.equal(stored?.decision, answer.decision);
606+
assert.equal(stored?.feedback, "feedback" in answer ? answer.feedback : undefined);
607+
assert.equal(projection.nodes.find((entry) => entry.id === nodeId)?.status, outcome);
608+
assert.equal(projection.turnItems.find((entry) => entry.id === item.id)?.status, outcome);
609+
}),
610+
);
611+
471612
it.effect("rolls back runs a provider rewound off its active branch", () =>
472613
Effect.gen(function* () {
473614
const now = yield* DateTime.now;

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -761,7 +761,7 @@ export const layer: Layer.Layer<
761761
if (input.writeIfProviderThreadOwner !== undefined) {
762762
const ownerResult = yield* eventSink
763763
.writeIfProviderThreadOwner({
764-
guardPendingUserInputCancellations: true,
764+
guardPendingRuntimeRequestCancellations: true,
765765
...(input.commandId === undefined ? {} : { commandId: input.commandId }),
766766
...input.writeIfProviderThreadOwner,
767767
events,
@@ -772,15 +772,15 @@ export const layer: Layer.Layer<
772772
if (input.writeIfRunCurrent === undefined) {
773773
return yield* eventSink
774774
.write({
775-
guardPendingUserInputCancellations: true,
775+
guardPendingRuntimeRequestCancellations: true,
776776
...(input.commandId === undefined ? {} : { commandId: input.commandId }),
777777
events,
778778
})
779779
.pipe(retryLockTimeout, Effect.mapError(mapWriteError));
780780
}
781781
const result = yield* eventSink
782782
.writeIfRunCurrent({
783-
guardPendingUserInputCancellations: true,
783+
guardPendingRuntimeRequestCancellations: true,
784784
...(input.commandId === undefined ? {} : { commandId: input.commandId }),
785785
threadId: input.threadId,
786786
...input.writeIfRunCurrent,

0 commit comments

Comments
 (0)