diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index fff4fd3e3db76..a38fcf47117ae 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -103,7 +103,7 @@ export interface EventSinkV2Shape { OrchestrationV2ProviderThread, "driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef" >; - readonly guardPendingUserInputCancellations?: boolean; + readonly guardPendingRuntimeRequestCancellations?: boolean; readonly commandId?: CommandId; readonly events: ReadonlyArray; }) => Effect.Effect, EventSinkV2Error>; @@ -117,13 +117,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; @@ -145,7 +145,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; @@ -342,17 +342,13 @@ const layerBase: Layer.Layer< // A user can answer after terminal normalization reads the pending 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) { - if ( - event.type !== "runtime-request.updated" || - event.payload.kind !== "user_input" || - event.payload.status !== "cancelled" - ) + if (event.type !== "runtime-request.updated" || event.payload.status !== "cancelled") continue; const current = yield* projectionStore.getRuntimeRequest( event.threadId, @@ -360,7 +356,6 @@ const layerBase: Layer.Layer< ); if ( current?.status !== "pending" || - current.kind !== "user_input" || current.providerTurnId !== event.payload.providerTurnId || current.responseCapability.type === "message" ) { @@ -376,7 +371,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) ); @@ -497,8 +493,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({ @@ -574,8 +570,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({ @@ -639,8 +635,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/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2d31d3e9bde54..59e6daacc36d3 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -330,7 +330,7 @@ export type ProjectionRuntimeRecoveryState = Pick< | "turnItems" >; -export type ProjectionPendingUserInputs = Pick< +export type ProjectionPendingRuntimeRequests = Pick< OrchestrationV2ThreadProjection, "runtimeRequests" | "nodes" | "turnItems" >; @@ -684,10 +684,11 @@ export interface ProjectionStoreV2Shape { readonly getRuntimeRecoveryProjection: ( threadId: ThreadId, ) => Effect.Effect; - readonly getPendingNativeUserInputs: ( + readonly getPendingNativeRuntimeRequests: ( threadId: ThreadId, providerTurnId: ProviderTurnId, - ) => Effect.Effect; + includeApprovals?: boolean, + ) => Effect.Effect; readonly getPlan: ( threadId: ThreadId, planId: PlanId, @@ -5587,46 +5588,51 @@ export const layer: Layer.Layer = ? cause : new ProjectionStoreReadError({ threadId, cause }); - const getPendingNativeUserInputs: ProjectionStoreV2Shape["getPendingNativeUserInputs"] = ( - threadId, - providerTurnId, - ) => - sql - .withTransaction( - Effect.gen(function* () { - const rows = yield* sql` + const getPendingNativeRuntimeRequests: ProjectionStoreV2Shape["getPendingNativeRuntimeRequests"] = + (threadId, providerTurnId, includeApprovals = false) => + sql + .withTransaction( + Effect.gen(function* () { + const rows = yield* sql` SELECT payload_json FROM orchestration_v2_projection_runtime_requests WHERE thread_id = ${threadId} AND provider_turn_id = ${providerTurnId} - AND kind = 'user_input' AND status = 'pending' + AND status = 'pending' `; - const runtimeRequests = (yield* decodeRows( - decodeRuntimeRequestPayload, - threadId, - )(rows)).filter((request) => request.responseCapability.type !== "message"); - if (runtimeRequests.length === 0) return { runtimeRequests, nodes: [], turnItems: [] }; - const nodeRows = yield* sql` + const runtimeRequests = (yield* decodeRows( + decodeRuntimeRequestPayload, + threadId, + )(rows)).filter( + (request) => + request.responseCapability.type !== "message" && + (includeApprovals || request.kind === "user_input"), + ); + if (runtimeRequests.length === 0) + return { runtimeRequests, nodes: [], turnItems: [] }; + const nodeRows = yield* sql` SELECT payload_json FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId} AND ${sql.in( "node_id", runtimeRequests.map((request) => request.nodeId), )} `; - const itemRows = yield* sql` + const itemRows = yield* sql` SELECT payload_json FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND provider_turn_id = ${providerTurnId} - AND type = 'user_input_request' + AND type IN ('user_input_request', 'approval_request') `; - const requestIds = new Set(runtimeRequests.map((request) => request.id)); - return { - runtimeRequests, - nodes: yield* decodeRows(decodeNodePayload, threadId)(nodeRows), - turnItems: (yield* decodeRows(decodeTurnItemPayload, threadId)(itemRows)).filter( - (item) => item.type === "user_input_request" && requestIds.has(item.requestId), - ), - }; - }), - ) - .pipe(Effect.mapError(controlReadError(threadId))); + const requestIds = new Set(runtimeRequests.map((request) => request.id)); + return { + runtimeRequests, + nodes: yield* decodeRows(decodeNodePayload, threadId)(nodeRows), + turnItems: (yield* decodeRows(decodeTurnItemPayload, threadId)(itemRows)).filter( + (item) => + (item.type === "user_input_request" || item.type === "approval_request") && + requestIds.has(item.requestId), + ), + }; + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); const getPlan: ProjectionStoreV2Shape["getPlan"] = (threadId, planId) => sql @@ -9127,7 +9133,7 @@ export const layer: Layer.Layer = getRun: readOnReader(getRun), canStartQueuedRun: readOnReader(canStartQueuedRun), getHeldQueuedRuns: readOnReader(getHeldQueuedRuns), - getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), + getPendingNativeRuntimeRequests: readOnReader(getPendingNativeRuntimeRequests), hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index a751b0bb462d1..f0b4d86342b68 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -1209,8 +1209,8 @@ layer("ProviderEventIngestorV2", (it) => { }), ); - it.effect.each(["interrupted", "control"] as const)( - "dismisses only native questions when a provider turn ends with %s", + it.effect.each(["interrupted", "failed", "completed", "cancelled", "control"] as const)( + "withdraws approvals only when explicitly abandoned (%s)", (terminal) => Effect.gen(function* () { const now = yield* DateTime.now; @@ -1325,6 +1325,7 @@ layer("ProviderEventIngestorV2", (it) => { providerSessionId, providerInstanceId: modelSelection.instanceId, threadId, + withdrawApprovals: terminal === "control", event: terminal === "control" ? { @@ -1356,7 +1357,10 @@ layer("ProviderEventIngestorV2", (it) => { const stored = yield* ingestor.ingestNormalized(input); const projection = yield* projectionStore.getThreadProjection(threadId); for (const fixture of fixtures) { - const closed = fixture.key === "native" || fixture.key === "unavailable"; + const closed = + fixture.key === "native" || + fixture.key === "unavailable" || + (fixture.key === "approval" && terminal === "control"); const request = projection.runtimeRequests.find( (item) => item.id === fixture.request.id, )!; @@ -1377,10 +1381,10 @@ layer("ProviderEventIngestorV2", (it) => { } assert.equal( stored.filter((entry) => entry.event.type === "runtime-request.updated").length, - 2, + terminal === "control" ? 3 : 2, ); assert.isEmpty( - (yield* projectionStore.getPendingNativeUserInputs(threadId, providerTurnId)) + (yield* projectionStore.getPendingNativeRuntimeRequests(threadId, providerTurnId)) .runtimeRequests, ); const replayed = yield* eventStore.read({ threadId }).pipe(Stream.runCollect); @@ -1390,16 +1394,16 @@ layer("ProviderEventIngestorV2", (it) => { entry.event.type === "runtime-request.updated" && entry.event.payload.status === "cancelled", ).length, - 2, + terminal === "control" ? 3 : 2, ); const repeated = yield* ingestor.ingestNormalized(input); assert.isFalse(repeated.some((entry) => entry.event.type === "runtime-request.updated")); }), ); - it.effect( - "preserves an answer committed after terminal normalization reads a pending question", - () => + it.effect.each(["question", "approval"] as const)( + "preserves an answer committed after terminal normalization reads a pending %s", + (kind) => Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const eventStore = yield* EventStore.EventStoreV2; @@ -1426,7 +1430,7 @@ layer("ProviderEventIngestorV2", (it) => { nodeId, providerTurnId, nativeRequestRef: null, - kind: "user_input", + kind: kind === "approval" ? "command" : "user_input", status: "pending", responseCapability: { type: "live", providerSessionId }, createdAt: now, @@ -1438,7 +1442,7 @@ layer("ProviderEventIngestorV2", (it) => { runId: null, parentNodeId: null, rootNodeId: nodeId, - kind: "user_input_request", + kind: kind === "approval" ? "approval_request" : "user_input_request", status: "waiting", countsForRun: false, providerThreadId, @@ -1464,9 +1468,9 @@ layer("ProviderEventIngestorV2", (it) => { startedAt: now, completedAt: null, updatedAt: now, - type: "user_input_request", - requestId: request.id, - questions: [], + ...(kind === "approval" + ? { type: "approval_request", requestId: request.id, requestKind: "command" } + : { type: "user_input_request", requestId: request.id, questions: [] }), }; const seedEvents: Array = [threadEvent]; for (const payload of [ @@ -1500,6 +1504,7 @@ layer("ProviderEventIngestorV2", (it) => { providerSessionId, providerInstanceId: modelSelection.instanceId, threadId, + withdrawApprovals: kind === "approval", event: { type: "provider_turn.updated", driver: CODEX_DRIVER, @@ -1519,11 +1524,12 @@ layer("ProviderEventIngestorV2", (it) => { .pipe(Effect.forkScoped); yield* Deferred.await(normalized); const answers = { decision: "Use the existing workspace" }; + const response = kind === "approval" ? { decision: "accept" as const } : { answers }; const responseEvents: Array = []; for (const payload of [ { type: "runtime-request.updated" as const, - payload: { ...request, status: "resolved" as const, answers, resolvedAt: now }, + payload: { ...request, status: "resolved" as const, ...response, resolvedAt: now }, }, { type: "node.updated" as const, @@ -1550,7 +1556,8 @@ layer("ProviderEventIngestorV2", (it) => { const projection = yield* projections.getThreadProjection(threadId); const answered = projection.runtimeRequests.find((entry) => entry.id === request.id)!; assert.equal(answered.status, "resolved"); - assert.deepEqual(answered.answers, answers); + if (kind === "approval") assert.equal(answered.decision, "accept"); + else assert.deepEqual(answered.answers, answers); assert.equal(projection.nodes.find((entry) => entry.id === nodeId)!.status, "completed"); assert.equal( projection.turnItems.find((entry) => entry.id === item.id)!.status, diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 7b71e58fb6089..688e442dc1ee2 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -200,6 +200,7 @@ export interface ProviderEventIngestInput { readonly runId?: RunId; readonly nodeId?: NodeId; readonly rawEventId?: RawEventId; + readonly withdrawApprovals?: boolean; readonly event: ProviderAdapter.ProviderAdapterV2Event; readonly analyticsContext?: ProviderTurnAnalyticsContext; } @@ -295,13 +296,17 @@ export const layer: Layer.Layer< ); }); - const dismissNativeUserInputs = Effect.fn("ProviderEventIngestor.dismissNativeUserInputs")( + const dismissNativeRuntimeRequests = Effect.fn("ProviderEventIngestor.dismissNativeUserInputs")( function* ( input: ProviderEventIngestInput, providerTurnId: ProviderTurnId, threadId = input.threadId, ) { - const pending = yield* projections.getPendingNativeUserInputs(threadId, providerTurnId); + const pending = yield* projections.getPendingNativeRuntimeRequests( + threadId, + providerTurnId, + input.withdrawApprovals, + ); const now = yield* DateTime.now; const events: Array = []; for (const request of pending.runtimeRequests) { @@ -525,7 +530,7 @@ export const layer: Layer.Layer< ...(["completed", "interrupted", "failed", "cancelled"].includes( input.event.providerTurn.status, ) - ? yield* dismissNativeUserInputs( + ? yield* dismissNativeRuntimeRequests( input, input.event.providerTurn.id, input.event.threadId, @@ -654,7 +659,10 @@ export const layer: Layer.Layer< // The session manager serves these to clients before ingestion. return []; case "turn.terminal": - const dismissed = yield* dismissNativeUserInputs(input, input.event.providerTurnId); + const dismissed = yield* dismissNativeRuntimeRequests( + input, + input.event.providerTurnId, + ); if (input.event.status !== "failed") { return dismissed; } @@ -748,7 +756,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, @@ -759,7 +767,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, }) @@ -767,7 +775,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, diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index c3d1096cb3c93..befcc7066fa54 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -253,7 +253,8 @@ it.effect( getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"), getThreadProviderContext: () => Effect.die("unused getThreadProviderContext"), getRuntimeResponseContext: () => Effect.die("unused getRuntimeResponseContext"), - getPendingNativeUserInputs: () => Effect.die("unused getPendingNativeUserInputs"), + getPendingNativeRuntimeRequests: () => + Effect.die("unused getPendingNativeRuntimeRequests"), getProviderControlContext: (_threadId, target) => Ref.get(projection).pipe( Effect.map((current) => ({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 7b0759ae99b0b..d0643188a5005 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -1690,6 +1690,7 @@ effectIt.effect("preserves a replacement committed before exhausted start settle IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: () => Effect.sync(() => { harness.setProjection({ ...harness.projection(), providerTurns: [providerTurn] }); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 0e8898cf0623b..957f95f06b520 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -4,6 +4,7 @@ import { ChatAttachmentId, ChatFileAttachment, CommandId, + EventId, MessageId, NodeId, type OrchestrationV2AppThread, @@ -24,6 +25,7 @@ import { ProviderTurnId, RunAttemptId, RunId, + RuntimeRequestId, ServerSettingsError, ThreadId, TurnItemId, @@ -96,6 +98,7 @@ const layerRunExecutionTest = RunExecutionService.layer.pipe( IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: () => Effect.succeed([]), }), ServerSettings.layerTest(), @@ -924,6 +927,7 @@ it.effect("passes the thread's MCP app context to the provider under a safe key" IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: () => Effect.succeed([]), }), ServerSettings.layerTest(), @@ -959,6 +963,7 @@ it.effect("fails the run when its ownership check cannot be read before calling IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: () => Effect.succeed([]), }), ServerSettings.layerTest(), @@ -1172,6 +1177,7 @@ it.effect("starts the provider when checkpoint baseline capture fails", () => IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: () => Effect.succeed([]), }), ServerSettings.layerTest(), @@ -1387,6 +1393,7 @@ it.effect.each(["failure", "interruption", "stale-attempt", "start-guard"] as co IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: () => Effect.succeed([]), }), scenario === "start-guard" @@ -1568,6 +1575,7 @@ it.effect("ingests new member threads of an owned child after the root turn term IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: (input) => Effect.gen(function* () { const eventType = input.event.type; @@ -1824,6 +1832,7 @@ it.effect("ends a nested native subagent and its thread when the root run is int IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: () => Effect.succeed([]), }), ServerSettings.layerTest(), @@ -2271,6 +2280,7 @@ it.effect( IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: (input) => Ref.update(ingested, (current) => [...current, input.event]).pipe(Effect.as([])), }), @@ -2985,28 +2995,177 @@ it.effect("does not refresh pull requests for auxiliary or stale provider termin }), ); -it.effect("refreshes pull requests after a provider stream exits with an error", () => - Effect.gen(function* () { - const { written, observed } = yield* captureRootRunTermination({ - key: "pull-request-refresh:stream-error", - shouldFinalizeRun: () => Effect.succeed(true), - events: () => - Stream.fail( - new ProviderAdapter.ProviderAdapterEventStreamError({ - driver, - providerSessionId: ProviderSessionId.make("session:exited"), - cause: "provider process exited", - }), +it.effect.each(["stream", "terminal", "continuing", "completed", "transferred"] as const)( + "withdraws failed-turn approvals only when children stop (%s)", + (failureMode) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const requestId = RuntimeRequestId.make("request:stream-error"); + const childRequestId = RuntimeRequestId.make("request:stream-error:child"); + const childTurnId = ProviderTurnId.make("provider-turn:stream-error:child"); + const childProviderThreadId = ProviderThreadId.make("provider-thread:stream-error:child"); + const childThreadId = backgroundScenarioIds( + "pull-request-refresh:stream-error", + ).childThreadId; + const observedTurnId = ProviderTurnId.make("provider-turn:stream-error:observed"); + const guarded = yield* Ref.make(false); + const { written, observed, finalEvents } = yield* captureRootRunTermination({ + key: "pull-request-refresh:stream-error", + shouldFinalizeRun: () => Effect.succeed(true), + initialProviderTurnId: null, + normalize: (input) => + Effect.succeed( + input.withdrawApprovals && + input.event.type === "turn.terminal" && + (input.event.providerTurnId === observedTurnId || + (input.event.providerTurnId === childTurnId && input.threadId === childThreadId)) + ? [ + { + id: EventId.make(`event:stream-error:withdraw-request:${input.threadId}`), + type: "runtime-request.updated", + threadId: input.threadId, + occurredAt: now, + payload: { + id: input.threadId === childThreadId ? childRequestId : requestId, + nodeId: NodeId.make("node:stream-error:approval"), + providerTurnId: + input.event.type === "turn.terminal" ? input.event.providerTurnId : null, + nativeRequestRef: null, + kind: "file-change", + status: "cancelled", + responseCapability: { + type: "not_resumable", + reason: "Provider stream ended", + }, + createdAt: now, + resolvedAt: now, + }, + }, + ] + : [], + ), + beforeTerminalWrite: (input) => { + return Ref.set(guarded, input.guardPendingRuntimeRequestCancellations === true).pipe( + Effect.as(true), + ); + }, + events: (ids) => + Stream.concat( + Stream.make( + childThreadCreatedEvent(ids), + { + type: "provider_turn.updated", + threadId: childThreadId, + driver, + providerTurn: { + id: childTurnId, + providerThreadId: childProviderThreadId, + nodeId: NodeId.make("node:stream-error:child-root"), + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + { + type: "runtime_request.updated", + driver, + threadId: childThreadId, + runtimeRequest: { + id: childRequestId, + nodeId: NodeId.make("node:stream-error:child-approval"), + providerTurnId: childTurnId, + nativeRequestRef: null, + kind: "command", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: ProviderSessionId.make("session:stream-error"), + }, + createdAt: now, + resolvedAt: null, + }, + }, + { + type: "provider_turn.updated", + driver, + providerTurn: { + id: observedTurnId, + providerThreadId: ids.providerThreadId, + nodeId: ids.rootNodeId, + runAttemptId: ids.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + ...(failureMode === "transferred" + ? [ + { + type: "subagent.updated" as const, + driver, + subagent: { + ...makeRunOwnedSubagentFixture({ + ids, + providerInstanceId: ProviderInstanceId.make("codex"), + childThreadId, + driver, + status: "running", + }), + providerThreadId: childProviderThreadId, + runId: RunId.make("run:replacement"), + }, + }, + ] + : []), + ), + failureMode === "stream" + ? Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:exited"), + cause: "provider process exited", + }), + ) + : Stream.make({ + ...rootTerminalEvent( + { ...ids, rootProviderTurnId: observedTurnId }, + failureMode === "completed" ? "completed" : "failed", + ), + ...(failureMode === "continuing" ? { childTurnsContinue: true as const } : {}), + }), + ), + }); + assert.isTrue(yield* Ref.get(guarded)); + assert.deepEqual(observed, [ + failureMode === "completed" ? "run:waiting" : "run:failed", + "pull-requests-refreshed", + ]); + assert.deepEqual( + written.map((item) => item.type), + failureMode === "stream" ? ["error"] : [], + ); + const error = written.find((item) => item.type === "error"); + if (failureMode === "stream") + assert.include(error?.failure.message ?? "", "provider event stream closed unexpectedly"); + assert.equal( + finalEvents.some( + (event) => + event.type === "runtime-request.updated" && event.payload.id === childRequestId, ), - }); - assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); - assert.deepEqual( - written.map((item) => item.type), - ["error"], - ); - const error = written.find((item) => item.type === "error"); - assert.include(error?.failure.message ?? "", "provider event stream closed unexpectedly"); - }), + failureMode === "stream" || failureMode === "terminal", + ); + assert.equal( + finalEvents.some( + (event) => event.type === "runtime-request.updated" && event.payload.id === requestId, + ), + failureMode === "stream" || failureMode === "terminal" || failureMode === "transferred", + ); + }), ); it.effect("records a provider's input rejection at turn start in the terminal error", () => @@ -3540,6 +3699,7 @@ function captureRootRunTermination(input: { readonly shouldStartProviderTurn?: RunExecutionService.RunExecutionServiceV2StartRootRunInput["shouldStartProviderTurn"]; readonly appContext?: ReturnType; readonly ingestNormalized?: ProviderEventIngestor.ProviderEventIngestorV2["Service"]["ingestNormalized"]; + readonly normalize?: ProviderEventIngestor.ProviderEventIngestorV2["Service"]["normalize"]; readonly beforeTerminalWrite?: ( request: Parameters[0], ) => Effect.Effect; @@ -3621,6 +3781,7 @@ function captureRootRunTermination(input: { IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: input.normalize ?? (() => Effect.succeed([])), ingestNormalized: input.ingestNormalized ?? (() => Effect.succeed([])), }), ServerSettings.layerTest(), @@ -4160,6 +4321,7 @@ function runBackgroundItemScenario( IdAllocator.layer, ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ + normalize: () => Effect.succeed([]), ingestNormalized: (input) => { const write = Effect.gen(function* () { const event = input.event; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 49cdaf04406b7..7c97f2c17704b 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -21,6 +21,7 @@ import { type ProviderThreadId, type ProviderTurnId, type RunAttemptId, + type RuntimeRequestId, type ThreadId, type TurnItemId, } from "@t3tools/contracts"; @@ -154,6 +155,11 @@ export function selectInheritedBackgroundTurnItems(input: { type SubagentTurnItem = Extract; +type PendingRequestTurn = { + readonly threadId: ThreadId; + readonly providerTurnId: ProviderTurnId; +}; + type OpenRunOwnedSubagentProjection = { readonly subagents: ReadonlyMap; readonly turnItems: ReadonlyMap; @@ -651,6 +657,7 @@ export const layer: Layer.Layer< const writeFinalRunEvents = Effect.fn("RunExecutionService.writeFinalRunEvents")( function* (input: { readonly run: OrchestrationV2Run; + readonly providerSessionId: ProviderSessionId; readonly rootNode: OrchestrationV2ExecutionNode; readonly checkpointScope: OrchestrationV2CheckpointScope; readonly providerThread: OrchestrationV2ProviderThread; @@ -665,6 +672,7 @@ export const layer: Layer.Layer< readonly terminal: ProviderTerminalEvent; readonly completedAt?: DateTime.Utc; readonly failureItemPersisted: boolean; + readonly pendingRequestTurns?: ReadonlyArray; readonly refreshAfterTurn: Effect.Effect; readonly writeIfRunCurrent?: { readonly activeAttemptId: RunAttemptId; @@ -771,8 +779,37 @@ export const layer: Layer.Layer< const checkpointCaptureCommandId = CommandId.make( `command:effect:checkpoint.capture:${input.run.id}`, ); - // Stopped runs capture rollback points for the next message; enqueue before later starts on this effect lane. + // Failed turns without continuing children have no consumer for their pending callbacks. + const cancelledRequestEvents: Array = []; + if (input.terminal.status === "failed" && !childTurnsContinue(input.terminal)) { + const turns = new Map>(); + for (const turn of [ + { threadId: input.run.threadId, providerTurnId: input.terminal.providerTurnId }, + ...(input.pendingRequestTurns ?? []), + ]) { + const seen = turns.get(turn.threadId) ?? new Set(); + if (seen.has(turn.providerTurnId)) continue; + seen.add(turn.providerTurnId); + turns.set(turn.threadId, seen); + const events = yield* providerEventIngestor.normalize({ + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + threadId: turn.threadId, + runId: input.run.id, + nodeId: input.rootNode.id, + event: { ...input.terminal, providerTurnId: turn.providerTurnId }, + withdrawApprovals: true, + }); + cancelledRequestEvents.push( + ...events.filter( + (event) => event.type !== "turn-item.updated" || event.payload.type !== "error", + ), + ); + } + } const finalization = { + guardPendingRuntimeRequestCancellations: true, + // Stopped runs capture rollback points for the next message; enqueue before later starts on this effect lane. effects: input.terminal.status === "completed" || input.terminal.status === "interrupted" || @@ -791,6 +828,7 @@ export const layer: Layer.Layer< ] : [], events: [ + ...cancelledRequestEvents, ...cascadedSubagentEvents, ...(input.abandonedProviderTurn === undefined ? [] @@ -893,6 +931,7 @@ export const layer: Layer.Layer< } satisfies Parameters[0]; if (input.writeIfRunCurrent !== undefined) { const result = yield* eventSink.writeIfRunCurrent({ + ...finalization, threadId: input.run.threadId, runId: input.run.id, activeAttemptId: input.writeIfRunCurrent.activeAttemptId, @@ -911,8 +950,6 @@ export const layer: Layer.Layer< }, } : {}), - events: finalization.events, - effects: finalization.effects, }); if (!result.committed) { return false; @@ -987,11 +1024,12 @@ export const layer: Layer.Layer< const makeFailedTerminalEvent = ( failure: OrchestrationV2ProviderFailure, failureItemOrdinal: number, + providerTurnId?: ProviderTurnId | null, ): ProviderTerminalEvent => ({ type: "turn.terminal", driver: input.providerThread.driver, providerThreadId: input.providerThread.id, - providerTurnId: failedProviderTurnId, + providerTurnId: providerTurnId ?? failedProviderTurnId, runOrdinal: input.run.ordinal, failureItemOrdinal, status: "failed", @@ -1025,6 +1063,7 @@ export const layer: Layer.Layer< cause, }); yield* writeFinalRunEvents({ + providerSessionId: input.providerSessionId, run: input.run, rootNode: input.rootNode, checkpointScope: input.checkpointScope, @@ -1122,6 +1161,13 @@ export const layer: Layer.Layer< : { relatedProviderThreadIds: input.relatedProviderThreadIds }), }), ); + const pendingRequestTurns = new Map(); + const ownedPendingRequestTurns = (routing: ProviderEventRoutingState) => + [...pendingRequestTurns.values()].filter( + (turn) => + routing.ownedProviderTurns.has(turn.providerTurnId) || + (turn.threadId !== input.run.threadId && routing.ownedThreadIds.has(turn.threadId)), + ); const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); const providerThreadOwnerLost = yield* Ref.make(false); @@ -1153,8 +1199,10 @@ export const layer: Layer.Layer< } const providerThread = yield* Ref.get(latestProviderThread); const openSubagents = yield* Ref.get(openRunOwnedSubagents); + const routing = yield* Ref.get(eventRouting); const writeTerminal = (settledTerminal: ProviderTerminalEvent) => writeFinalRunEvents({ + providerSessionId: input.providerSessionId, run: input.run, rootNode: input.rootNode, checkpointScope: input.checkpointScope, @@ -1182,6 +1230,7 @@ export const layer: Layer.Layer< : { loadRunInterruptReason: input.loadRunInterruptReason }), openRunOwnedSubagents: openSubagents, openRunOwnedSubagentsTerminal: terminal, + pendingRequestTurns: ownedPendingRequestTurns(routing), ...(abandonedProviderTurn === undefined ? {} : { abandonedProviderTurn }), terminal: settledTerminal, failureItemPersisted: settledTerminal.status === "failed", @@ -1211,6 +1260,15 @@ export const layer: Layer.Layer< deliverable: boolean, ) { const routing = yield* Ref.get(eventRouting); + if (event.type === "runtime_request.updated") { + const request = event.runtimeRequest; + if (request.status === "pending" && request.providerTurnId !== null) { + pendingRequestTurns.set(request.id, { + threadId: event.threadId ?? input.run.threadId, + providerTurnId: request.providerTurnId, + }); + } else pendingRequestTurns.delete(request.id); + } if (event.type === "provider_turn.updated") { const isRoot = event.providerTurn.runAttemptId === input.attempt.id || @@ -1836,6 +1894,7 @@ export const layer: Layer.Layer< const openSubagents = yield* Ref.get(openRunOwnedSubagents); let nextRetryWarningAt = completedAt; yield* writeFinalRunEvents({ + providerSessionId: input.providerSessionId, run: input.run, rootNode: input.rootNode, checkpointScope: input.checkpointScope, @@ -1846,6 +1905,7 @@ export const layer: Layer.Layer< expectedStatus: "running", }, openRunOwnedSubagents: openSubagents, + pendingRequestTurns: ownedPendingRequestTurns(routing), ...(abandonedProviderTurn === null ? {} : { abandonedProviderTurn }), completedAt, terminal: makeFailedTerminalEvent( @@ -1857,6 +1917,7 @@ export const layer: Layer.Layer< class: "unknown", }), latestItemOrdinal + 1, + routing.rootProviderTurnId, ), failureItemPersisted: false, refreshAfterTurn, @@ -2016,6 +2077,7 @@ export const layer: Layer.Layer< Ref.get(openRunOwnedSubagents).pipe( Effect.flatMap((openSubagents) => writeFinalRunEvents({ + providerSessionId: input.providerSessionId, run: input.run, rootNode: input.rootNode, checkpointScope: input.checkpointScope,