diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index edae3077912ec..581cda4d25b97 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -102,7 +102,7 @@ export interface EventSinkV2Shape { OrchestrationV2ProviderThread, "driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef" >; - readonly guardPendingUserInputCancellations?: boolean; + readonly guardPendingRuntimeRequestCancellations?: boolean; readonly commandId?: CommandId; readonly events: ReadonlyArray; }) => Effect.Effect, EventSinkV2Error>; @@ -116,13 +116,13 @@ export interface EventSinkV2Shape { OrchestrationV2ProviderThread, "driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef" >; - readonly guardPendingUserInputCancellations?: boolean; + readonly guardPendingRuntimeRequestCancellations?: boolean; readonly commandId?: CommandId; readonly events: ReadonlyArray; readonly effects: ReadonlyArray; }) => Effect.Effect, EventSinkV2Error>; readonly writeIfRunCurrent: (input: { - readonly guardPendingUserInputCancellations?: boolean; + readonly guardPendingRuntimeRequestCancellations?: boolean; readonly commandId?: CommandId; readonly threadId: ThreadId; readonly runId: RunId; @@ -144,7 +144,7 @@ export interface EventSinkV2Shape { >; /** Commits only if the expected attempt and ordinal still own the provider thread, protecting newer attempts. */ readonly writeIfProviderThreadOwner: (input: { - readonly guardPendingUserInputCancellations?: boolean; + readonly guardPendingRuntimeRequestCancellations?: boolean; readonly commandId?: CommandId; readonly providerThreadId: ProviderThreadId; readonly runId: RunId; @@ -328,30 +328,55 @@ const layerBase: Layer.Layer< ); }); - // A user can answer after terminal normalization reads the pending request. + // A user can answer after terminal normalization reads the pending request, and an approval + // answer commits before its outbox delivery reaches a provider that may withdraw the request. // Recheck inside the write transaction so stale cleanup cannot erase answers. - const guardUserInputCancellations = + const guardRuntimeRequestCancellations = /* This helper runs for individual orchestration events; the enclosing operation owns the span. */ Effect.fnUntraced(function* (events: ReadonlyArray) { const staleRequests = new Set(); const staleNodes = new Set(); for (const event of events) { + // Providers can withdraw the node and item in writes of their own, so check each against its request. if ( - event.type !== "runtime-request.updated" || - event.payload.kind !== "user_input" || - event.payload.status !== "cancelled" - ) + event.type === "node.updated" && + event.payload.status === "cancelled" && + event.payload.runtimeRequestId !== null + ) { + const current = yield* projectionStore.getRuntimeRequest( + event.threadId, + event.payload.runtimeRequestId, + ); + if (current?.status === "resolved") staleNodes.add(event.payload.id); + continue; + } + if ( + event.type === "turn-item.updated" && + (event.payload.type === "user_input_request" || + event.payload.type === "approval_request") && + event.payload.status === "cancelled" + ) { + const current = yield* projectionStore.getRuntimeRequest( + event.threadId, + event.payload.requestId, + ); + if (current?.status === "resolved") staleRequests.add(event.payload.requestId); + continue; + } + if (event.type !== "runtime-request.updated" || event.payload.status !== "cancelled") continue; const current = yield* projectionStore.getRuntimeRequest( event.threadId, event.payload.id, ); - if ( - current?.status !== "pending" || - current.kind !== "user_input" || - current.providerTurnId !== event.payload.providerTurnId || - current.responseCapability.type === "message" - ) { + const stale = + event.payload.kind === "user_input" + ? current?.status !== "pending" || + current.kind !== "user_input" || + current.providerTurnId !== event.payload.providerTurnId || + current.responseCapability.type === "message" + : current?.status === "resolved"; + if (stale) { staleRequests.add(event.payload.id); staleNodes.add(event.payload.nodeId); } @@ -364,7 +389,8 @@ const layerBase: Layer.Layer< return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id); case "turn-item.updated": return ( - event.payload.type !== "user_input_request" || + (event.payload.type !== "user_input_request" && + event.payload.type !== "approval_request") || event.payload.status !== "cancelled" || !staleRequests.has(event.payload.requestId) ); @@ -485,8 +511,8 @@ const layerBase: Layer.Layer< ? yield* updateDelegatedCompletionOnly(input.events, () => true) : input.events; const normalized = yield* normalizeEvents( - input.guardPendingUserInputCancellations === true - ? yield* guardUserInputCancellations(events) + input.guardPendingRuntimeRequestCancellations === true + ? yield* guardRuntimeRequestCancellations(events) : events, ); const committed = yield* eventStore.append({ @@ -562,8 +588,8 @@ const layerBase: Layer.Layer< } const normalized = yield* normalizeEvents( - input.guardPendingUserInputCancellations === true - ? yield* guardUserInputCancellations(input.events) + input.guardPendingRuntimeRequestCancellations === true + ? yield* guardRuntimeRequestCancellations(input.events) : input.events, ); const storedEvents = yield* eventStore.append({ @@ -627,8 +653,8 @@ const layerBase: Layer.Layer< } const normalized = yield* normalizeEvents( - input.guardPendingUserInputCancellations === true - ? yield* guardUserInputCancellations(input.events) + input.guardPendingRuntimeRequestCancellations === true + ? yield* guardRuntimeRequestCancellations(input.events) : input.events, ); const storedEvents = yield* eventStore.append({ diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index f69444d235712..b4881c55ce25f 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -567,6 +567,145 @@ layer("ProviderEventIngestorV2", (it) => { }), ); + // regression: #24690 an answer commits before its outbox delivery reaches the provider + it.effect.each([ + { name: "keeps a committed decline", answer: { decision: "decline", feedback: "No." } }, + { name: "keeps a committed accept", answer: { decision: "accept" } }, + { name: "cancels an approval withdrawn before it was projected", answer: null }, + ] as const)("$name when the provider withdraws the approval", ({ answer }) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const eventSink = yield* EventSink.EventSinkV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const threadEvent = yield* threadCreatedEvent(now); + const threadId = threadEvent.threadId; + yield* eventSink.write({ events: [threadEvent] }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const nodeId = NodeId.make(`${threadId}:approval`); + const request: OrchestrationV2RuntimeRequest = { + id: RuntimeRequestId.make(`${threadId}:approval`), + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "command", + status: "pending", + responseCapability: { type: "live", providerSessionId, supportsFeedback: true }, + createdAt: now, + resolvedAt: null, + }; + const node: OrchestrationV2ExecutionNode = { + id: nodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId: nodeId, + kind: "approval_request", + status: "waiting", + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: request.id, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }; + const item: OrchestrationV2TurnItem = { + id: TurnItemId.make(`${threadId}:approval`), + threadId, + runId: null, + nodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 0, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "approval_request", + requestId: request.id, + requestKind: "command", + }; + const ingest = (status: "waiting" | "cancelled") => + Effect.forEach( + [ + { + type: "node.updated" as const, + driver: CODEX_DRIVER, + node: status === "waiting" ? node : { ...node, status, completedAt: now }, + }, + { + type: "runtime_request.updated" as const, + driver: CODEX_DRIVER, + runtimeRequest: + status === "waiting" ? request : { ...request, status, resolvedAt: now }, + }, + { + type: "turn_item.updated" as const, + driver: CODEX_DRIVER, + turnItem: status === "waiting" ? item : { ...item, status, completedAt: now }, + }, + ], + (event) => + ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + event, + }), + { concurrency: 1 }, + ); + const outcome = answer?.decision === "accept" ? "completed" : "cancelled"; + if (answer !== null) { + yield* ingest("waiting"); + // The orchestrator's response transaction; its outbox delivery never runs here. + const base = { threadId, nodeId, occurredAt: now }; + yield* eventSink.write({ + events: [ + { + ...base, + id: yield* idAllocator.allocate.event({ threadId }), + type: "runtime-request.updated", + payload: { ...request, status: "resolved", resolvedAt: now, ...answer }, + }, + { + ...base, + id: yield* idAllocator.allocate.event({ threadId }), + type: "node.updated", + payload: { ...node, status: outcome, completedAt: now }, + }, + { + ...base, + id: yield* idAllocator.allocate.event({ threadId }), + type: "turn-item.updated", + payload: { ...item, status: outcome, completedAt: now }, + }, + ], + }); + } + + yield* ingest("cancelled"); + const projection = yield* projectionStore.getThreadProjection(threadId); + const stored = projection.runtimeRequests.find((entry) => entry.id === request.id); + assert.equal(stored?.status, answer === null ? "cancelled" : "resolved"); + assert.equal(stored?.decision, answer?.decision); + assert.equal( + stored?.feedback, + answer !== null && "feedback" in answer ? answer.feedback : undefined, + ); + assert.equal(projection.nodes.find((entry) => entry.id === nodeId)?.status, outcome); + assert.equal(projection.turnItems.find((entry) => entry.id === item.id)?.status, outcome); + }), + ); + it.effect("rolls back runs a provider rewound off its active branch", () => Effect.gen(function* () { const now = yield* DateTime.now; diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 6fd6e7bc3129c..c4701dceb4e19 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -715,7 +715,7 @@ export const layer: Layer.Layer< if (input.writeIfProviderThreadOwner !== undefined) { const ownerResult = yield* eventSink .writeIfProviderThreadOwner({ - guardPendingUserInputCancellations: true, + guardPendingRuntimeRequestCancellations: true, ...(input.commandId === undefined ? {} : { commandId: input.commandId }), ...input.writeIfProviderThreadOwner, events, @@ -726,7 +726,7 @@ export const layer: Layer.Layer< if (input.writeIfRunCurrent === undefined) { return yield* eventSink .write({ - guardPendingUserInputCancellations: true, + guardPendingRuntimeRequestCancellations: true, ...(input.commandId === undefined ? {} : { commandId: input.commandId }), events, }) @@ -734,7 +734,7 @@ export const layer: Layer.Layer< } const result = yield* eventSink .writeIfRunCurrent({ - guardPendingUserInputCancellations: true, + guardPendingRuntimeRequestCancellations: true, ...(input.commandId === undefined ? {} : { commandId: input.commandId }), threadId: input.threadId, ...input.writeIfRunCurrent,