diff --git a/apps/server/src/mcp/threadAccess.ts b/apps/server/src/mcp/threadAccess.ts index ba8dc4c3e5765..72215f6c99548 100644 --- a/apps/server/src/mcp/threadAccess.ts +++ b/apps/server/src/mcp/threadAccess.ts @@ -10,7 +10,9 @@ import { } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import { EventSinkRuntimeRequestNotPendingError } from "../orchestration-v2/EventSink.ts"; import type { OrchestratorV2Error } from "../orchestration-v2/Orchestrator.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import * as OrchestrationMcp from "./OrchestratorMcpService.ts"; @@ -22,20 +24,25 @@ export const unavailable = () => message: "The operation could not be completed.", }); +// Its message is built only from the request id and status, so it is safe to expose. +const isRuntimeRequestNotPending = Schema.is(EventSinkRuntimeRequestNotPendingError); + export const dispatchFailure = ( error: OrchestratorV2Error | ThreadManagement.ThreadManagementProviderInstanceDisabledError, ) => error._tag === "ThreadManagementProviderInstanceDisabledError" ? new OrchestratorMcpFailure({ code: "invalid_request", message: error.message }) - : (error._tag === "OrchestratorDispatchError" || - error._tag === "OrchestratorCommandRejectedError") && - typeof error.cause === "string" && - error.cause.length > 0 - ? new OrchestratorMcpFailure({ - code: "orchestration_error", - message: Array.from(error.cause).slice(0, 1000).join(""), - }) - : unavailable(); + : error._tag === "OrchestratorDispatchError" && isRuntimeRequestNotPending(error.cause) + ? new OrchestratorMcpFailure({ code: "orchestration_error", message: error.cause.message }) + : (error._tag === "OrchestratorDispatchError" || + error._tag === "OrchestratorCommandRejectedError") && + typeof error.cause === "string" && + error.cause.length > 0 + ? new OrchestratorMcpFailure({ + code: "orchestration_error", + message: Array.from(error.cause).slice(0, 1000).join(""), + }) + : unavailable(); export interface CallerLimits { readonly runtimeMode: RuntimeMode; diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 02216cec576de..9b08e0ad31535 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -31,6 +31,7 @@ import { OrchestratorDispatchError, OrchestratorProjectionError, } from "../../orchestration-v2/Orchestrator.ts"; +import { EventSinkRuntimeRequestNotPendingError } from "../../orchestration-v2/EventSink.ts"; import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; import * as ServerConfig from "../../config.ts"; @@ -216,6 +217,17 @@ it("bounds public command rejections and redacts internal dispatch causes", () = new OrchestratorCommandRejectedError({ ...command, cause: "Run is not queued." }), ).message, ).toBe("Run is not queued."); + expect( + dispatchFailure( + new OrchestratorDispatchError({ + ...command, + cause: new EventSinkRuntimeRequestNotPendingError({ + requestId: RuntimeRequestId.make("mcp-core-request"), + status: "cancelled", + }), + }), + ).message, + ).toBe("Runtime request mcp-core-request is cancelled."); for (const cause of [ undefined, "", diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 3d2bc8e4b45a8..16f8b37c5ac4d 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -10,6 +10,7 @@ import { RuntimeRequestId, NodeId, type ProjectId, + ProviderRequestKind, ThreadId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; @@ -63,7 +64,27 @@ export class EventSinkStreamError extends Schema.TaggedError()( + "EventSinkRuntimeRequestNotPendingError", + { + requestId: RuntimeRequestId, + status: Schema.String, + }, +) { + override get message(): string { + return `Runtime request ${this.requestId} is ${this.status}.`; + } +} + +const isRuntimeRequestNotPending = Schema.is(EventSinkRuntimeRequestNotPendingError); + +export const EventSinkV2Error = Schema.Union([ + EventSinkWriteError, + EventSinkStreamError, + EventSinkRuntimeRequestNotPendingError, +]); export type EventSinkV2Error = typeof EventSinkV2Error.Type; export interface EventSinkV2Shape { @@ -291,8 +312,9 @@ const layerBase: Layer.Layer< ); }); - // A user can answer after terminal normalization reads the pending request. + // A user can answer a question or approval after terminal normalization reads it pending. // Recheck inside the write transaction so stale cleanup cannot erase answers. + const isApprovalRequestKind = Schema.is(ProviderRequestKind); const guardUserInputCancellations = /* This helper runs for individual orchestration events; the enclosing operation owns the span. */ Effect.fnUntraced(function* (events: ReadonlyArray) { @@ -301,7 +323,7 @@ const layerBase: Layer.Layer< for (const event of events) { if ( event.type !== "runtime-request.updated" || - event.payload.kind !== "user_input" || + (event.payload.kind !== "user_input" && !isApprovalRequestKind(event.payload.kind)) || event.payload.status !== "cancelled" ) continue; @@ -311,7 +333,7 @@ const layerBase: Layer.Layer< ); if ( current?.status !== "pending" || - current.kind !== "user_input" || + current.kind !== event.payload.kind || current.providerTurnId !== event.payload.providerTurnId || current.responseCapability.type === "message" ) { @@ -327,7 +349,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) ); @@ -655,6 +678,24 @@ const layerBase: Layer.Layer< const existing = yield* existingCommandResult(input.commandId); return { ...existing, committed: false as const, cancelledEffectIds: [] }; } + for (const event of input.events) { + if (event.type !== "runtime-request.updated" || event.payload.status !== "resolved") + continue; + const current = yield* projectionStore.getRuntimeRequest( + event.threadId, + event.payload.id, + ); + if ( + current?.status !== "pending" || + current.kind !== event.payload.kind || + current.providerTurnId !== event.payload.providerTurnId + ) { + return yield* new EventSinkRuntimeRequestNotPendingError({ + requestId: event.payload.id, + status: current?.status ?? "missing", + }); + } + } const normalized = yield* normalizeEvents(input.events); const storedEvents = yield* eventStore.append({ @@ -930,13 +971,14 @@ const layerBase: Layer.Layer< ), commitCommand: (input) => commitCommandEffect(input).pipe( - Effect.mapError( - (cause) => - new EventSinkWriteError({ - commandId: input.commandId, - eventCount: input.events.length, - cause, - }), + Effect.mapError((cause) => + isRuntimeRequestNotPending(cause) + ? cause + : new EventSinkWriteError({ + commandId: input.commandId, + eventCount: input.events.length, + cause, + }), ), ), commitRejectedCommand: (input) => diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index fcadd8036b924..b3a5fa8ceaac4 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -911,6 +911,129 @@ it.effect("SQL projection keeps background thread updates out of updatedAt", () backgroundThreadUpdatesKeepTimestamp.pipe(Effect.provide(layerTest)), ); +const pendingNativeRequestsStayWithTheirTurn = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const suffix = "pending-native-requests"; + const threadId = yield* addRolledBackRecoveryCandidate(suffix); + const providerThreadId = ProviderThreadId.make(`provider-thread:${suffix}`); + const turnId = ProviderTurnId.make(`turn:${suffix}`); + const otherTurnId = ProviderTurnId.make(`turn:${suffix}:other`); + const live = { + type: "live" as const, + providerSessionId: ProviderSessionId.make(`session:${suffix}`), + }; + const requests = [ + ["approval", turnId, "command", "pending", live], + ["other-turn-approval", otherTurnId, "command", "pending", live], + ["session-approval", null, "command", "pending", live], + ["resolved-approval", turnId, "command", "resolved", live], + ["dynamic-tool", turnId, "dynamic_tool_call", "pending", live], + ["question", turnId, "user_input", "pending", live], + ["message-question", turnId, "user_input", "pending", { type: "message" as const }], + ] as const; + for (const [ + ordinal, + [name, providerTurnId, kind, status, responseCapability], + ] of requests.entries()) { + const requestId = RuntimeRequestId.make(`request:${suffix}:${name}`); + const nodeId = NodeId.make(`node:${suffix}:${name}`); + const approval = kind === "command"; + yield* store.apply({ + id: EventId.make(`event:${suffix}:${name}:request`), + type: "runtime-request.updated", + threadId, + nodeId, + occurredAt: now, + payload: { + id: requestId, + nodeId, + providerTurnId, + nativeRequestRef: null, + kind, + status, + responseCapability, + createdAt: now, + resolvedAt: status === "pending" ? null : now, + }, + }); + yield* store.apply({ + id: EventId.make(`event:${suffix}:${name}:node`), + type: "node.updated", + threadId, + nodeId, + occurredAt: now, + payload: { + id: nodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId: nodeId, + kind: approval ? "approval_request" : "user_input_request", + status: "waiting", + countsForRun: false, + providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: requestId, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }); + if (kind === "dynamic_tool_call") continue; + yield* store.apply({ + id: EventId.make(`event:${suffix}:${name}:item`), + type: "turn-item.updated", + threadId, + nodeId, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:${suffix}:${name}`), + threadId, + runId: null, + nodeId, + providerThreadId, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + ...(approval + ? { type: "approval_request" as const, requestId, requestKind: "command" as const } + : { type: "user_input_request" as const, requestId, questions: [] }), + }, + }); + } + + const names = (records: ReadonlyArray<{ readonly id: string }>) => + records.map((record) => record.id.slice(record.id.lastIndexOf(":") + 1)).toSorted(); + const withoutApprovals = yield* store.getPendingNativeRequests(threadId, turnId, { + includeApprovals: false, + }); + assert.deepEqual(names(withoutApprovals.runtimeRequests), ["question"]); + assert.deepEqual(names(withoutApprovals.nodes), ["question"]); + assert.deepEqual(names(withoutApprovals.turnItems), ["question"]); + const withApprovals = yield* store.getPendingNativeRequests(threadId, turnId, { + includeApprovals: true, + }); + assert.deepEqual(names(withApprovals.runtimeRequests), ["approval", "question"]); + assert.deepEqual(names(withApprovals.nodes), ["approval", "question"]); + assert.deepEqual(names(withApprovals.turnItems), ["approval", "question"]); +}); + +it.effect("memory pending native requests: approvals are opt-in and stay with their turn", () => + pendingNativeRequestsStayWithTheirTurn.pipe(Effect.provide(ProjectionStore.layerMemory)), +); + +it.effect("SQL pending native requests: approvals are opt-in and stay with their turn", () => + pendingNativeRequestsStayWithTheirTurn.pipe(Effect.provide(layerTest)), +); + it.effect("memory recovery selection includes unfinished items from missing runs", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 00333529dd1fb..e38b4599be5be 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -22,6 +22,7 @@ import type { OrchestrationV2ExecutionNode, OrchestrationV2PlanArtifact, OrchestrationV2PendingRuntimeRequestSummary, + OrchestrationV2RuntimeRequest, PlanId, OrchestrationV2ConversationMessage, OrchestrationV2DomainEvent, @@ -68,6 +69,7 @@ import { TurnItemId, NodeId, ProviderInstanceId, + ProviderRequestKind, } from "@t3tools/contracts"; import { createOrchestrationV2TurnItemVisibility, @@ -301,11 +303,22 @@ export type ProjectionRuntimeRecoveryState = Pick< | "turnItems" >; -export type ProjectionPendingUserInputs = Pick< +export type ProjectionPendingNativeRequests = Pick< OrchestrationV2ThreadProjection, "runtimeRequests" | "nodes" | "turnItems" >; +const isApprovalRequestKind = Schema.is(ProviderRequestKind); + +function isPendingNativeRequest( + request: OrchestrationV2RuntimeRequest, + includeApprovals: boolean, +): boolean { + return request.kind === "user_input" + ? request.responseCapability.type !== "message" + : includeApprovals && isApprovalRequestKind(request.kind); +} + export type ProjectionThreadProviderContext = Pick< OrchestrationV2ThreadProjection, "thread" | "providerSessions" | "providerThreads" @@ -494,10 +507,11 @@ export interface ProjectionStoreV2Shape { readonly getRuntimeRecoveryProjection: ( threadId: ThreadId, ) => Effect.Effect; - readonly getPendingNativeUserInputs: ( + readonly getPendingNativeRequests: ( threadId: ThreadId, providerTurnId: ProviderTurnId, - ) => Effect.Effect; + options: { readonly includeApprovals: boolean }, + ) => Effect.Effect; readonly getPlan: ( threadId: ThreadId, planId: PlanId, @@ -4656,9 +4670,10 @@ export const layer: Layer.Layer = ? cause : new ProjectionStoreReadError({ threadId, cause }); - const getPendingNativeUserInputs: ProjectionStoreV2Shape["getPendingNativeUserInputs"] = ( + const getPendingNativeRequests: ProjectionStoreV2Shape["getPendingNativeRequests"] = ( threadId, providerTurnId, + { includeApprovals }, ) => sql .withTransaction( @@ -4666,12 +4681,12 @@ export const layer: Layer.Layer = 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"); + )(rows)).filter((request) => isPendingNativeRequest(request, includeApprovals)); if (runtimeRequests.length === 0) return { runtimeRequests, nodes: [], turnItems: [] }; const nodeRows = yield* sql` SELECT payload_json FROM orchestration_v2_projection_nodes @@ -4683,14 +4698,16 @@ export const layer: Layer.Layer = 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), + (item) => + (item.type === "user_input_request" || item.type === "approval_request") && + requestIds.has(item.requestId), ), }; }), @@ -6559,7 +6576,7 @@ export const layer: Layer.Layer = getCheckpointRollbackContext: readOnReader(getCheckpointRollbackContext), getRunMessage: readOnReader(getRunMessage), canStartQueuedRun: readOnReader(canStartQueuedRun), - getPendingNativeUserInputs: readOnReader(getPendingNativeUserInputs), + getPendingNativeRequests: readOnReader(getPendingNativeRequests), hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), @@ -6813,28 +6830,31 @@ export const layerMemory: Layer.Layer = Layer.effect( .map((projection) => projection.thread.id); }), getUnreadableThreadIds: () => Effect.succeed([]), - getPendingNativeUserInputs: Effect.fn("ProjectionStore.getPendingNativeUserInputs")( - function* (threadId, providerTurnId) { - const projection = (yield* Ref.get(replayState)).projections.get(threadId); - if (projection === undefined) return { runtimeRequests: [], nodes: [], turnItems: [] }; - const runtimeRequests = projection.runtimeRequests.filter( - (request) => - request.providerTurnId === providerTurnId && - request.kind === "user_input" && - request.status === "pending" && - request.responseCapability.type !== "message", - ); - const requestIds = new Set(runtimeRequests.map((request) => request.id)); - const nodeIds = new Set(runtimeRequests.map((request) => request.nodeId)); - return { - runtimeRequests, - nodes: projection.nodes.filter((node) => nodeIds.has(node.id)), - turnItems: projection.turnItems.filter( - (item) => item.type === "user_input_request" && requestIds.has(item.requestId), - ), - }; - }, - ), + getPendingNativeRequests: Effect.fn("ProjectionStore.getPendingNativeRequests")(function* ( + threadId, + providerTurnId, + { includeApprovals }, + ) { + const projection = (yield* Ref.get(replayState)).projections.get(threadId); + if (projection === undefined) return { runtimeRequests: [], nodes: [], turnItems: [] }; + const runtimeRequests = projection.runtimeRequests.filter( + (request) => + request.providerTurnId === providerTurnId && + request.status === "pending" && + isPendingNativeRequest(request, includeApprovals), + ); + const requestIds = new Set(runtimeRequests.map((request) => request.id)); + const nodeIds = new Set(runtimeRequests.map((request) => request.nodeId)); + return { + runtimeRequests, + nodes: projection.nodes.filter((node) => nodeIds.has(node.id)), + turnItems: projection.turnItems.filter( + (item) => + (item.type === "user_input_request" || item.type === "approval_request") && + requestIds.has(item.requestId), + ), + }; + }), getPlan: Effect.fn("ProjectionStore.getPlan")(function* (threadId, planId) { const projection = (yield* Ref.get(replayState)).projections.get(threadId); if (projection === undefined) diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 849cedded9292..cb7c0c97612c1 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -4,6 +4,7 @@ import { CheckpointId, CheckpointRef, CheckpointScopeId, + CommandId, MessageId, GitCommandError, type ModelSelection, @@ -24,12 +25,15 @@ import { RuntimeRequestId, TurnItemId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; @@ -37,6 +41,8 @@ import { toolOutputImages } from "@t3tools/shared/toolOutput"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import * as ServerConfig from "../config.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as EffectOutbox from "./EffectOutbox.ts"; import * as EventSink from "./EventSink.ts"; import * as EventStore from "./EventStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -963,8 +969,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", "control", "control-interrupted"] as const)( + "dismisses native questions, and approvals only of a stopped turn, when a provider turn ends with %s", (terminal) => Effect.gen(function* () { const now = yield* DateTime.now; @@ -1001,16 +1007,19 @@ layer("ProviderEventIngestorV2", (it) => { { key: "answered", responseCapability: { type: "live", providerSessionId } }, { key: "other-turn", responseCapability: { type: "live", providerSessionId } }, { key: "approval", responseCapability: { type: "live", providerSessionId } }, + { key: "other-turn-approval", responseCapability: { type: "live", providerSessionId } }, ] as const; + const stopped = terminal === "interrupted" || terminal === "control-interrupted"; const fixtures = specs.map((spec, ordinal) => { const nodeId = NodeId.make(`${threadId}:${spec.key}`); const resolved = spec.key === "answered"; + const approval = spec.key === "approval" || spec.key === "other-turn-approval"; const request: OrchestrationV2RuntimeRequest = { id: RuntimeRequestId.make(`${threadId}:${spec.key}`), nodeId, - providerTurnId: spec.key === "other-turn" ? otherTurnId : providerTurnId, + providerTurnId: spec.key.startsWith("other-turn") ? otherTurnId : providerTurnId, nativeRequestRef: null, - kind: spec.key === "approval" ? "command" : "user_input", + kind: approval ? "command" : "user_input", status: resolved ? "resolved" : "pending", responseCapability: spec.responseCapability, createdAt: now, @@ -1022,7 +1031,7 @@ layer("ProviderEventIngestorV2", (it) => { runId: null, parentNodeId: null, rootNodeId: nodeId, - kind: spec.key === "approval" ? "approval_request" : "user_input_request", + kind: approval ? "approval_request" : "user_input_request", status: resolved ? "completed" : "waiting", countsForRun: false, providerThreadId, @@ -1048,7 +1057,7 @@ layer("ProviderEventIngestorV2", (it) => { startedAt: now, completedAt: resolved ? now : null, updatedAt: now, - ...(spec.key === "approval" + ...(approval ? { type: "approval_request", requestId: request.id, requestKind: "command" } : { type: "user_input_request", @@ -1080,14 +1089,15 @@ layer("ProviderEventIngestorV2", (it) => { providerInstanceId: modelSelection.instanceId, threadId, event: - terminal === "control" + terminal === "control" || terminal === "control-interrupted" ? { type: "turn.terminal" as const, driver: CODEX_DRIVER, providerThreadId, providerTurnId, runOrdinal: 1, - status: "completed" as const, + status: + terminal === "control" ? ("completed" as const) : ("interrupted" as const), failure: null, threadDisposition: "reusable" as const, } @@ -1110,7 +1120,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" && stopped); const request = projection.runtimeRequests.find( (item) => item.id === fixture.request.id, )!; @@ -1123,19 +1136,24 @@ layer("ProviderEventIngestorV2", (it) => { assert.isNotNull(node.completedAt); assert.equal(item.status, "cancelled"); assert.isNotNull(item.completedAt); + if (fixture.key === "approval") { + assert.equal(request.responseCapability.type, "not_resumable"); + } } else { assert.equal(request.status, fixture.request.status); assert.equal(node.status, fixture.node.status); assert.equal(item.status, fixture.item.status); } } + const dismissedCount = stopped ? 3 : 2; assert.equal( stored.filter((entry) => entry.event.type === "runtime-request.updated").length, - 2, + dismissedCount, ); assert.isEmpty( - (yield* projectionStore.getPendingNativeUserInputs(threadId, providerTurnId)) - .runtimeRequests, + (yield* projectionStore.getPendingNativeRequests(threadId, providerTurnId, { + includeApprovals: stopped, + })).runtimeRequests, ); const replayed = yield* eventStore.read({ threadId }).pipe(Stream.runCollect); assert.equal( @@ -1144,7 +1162,7 @@ layer("ProviderEventIngestorV2", (it) => { entry.event.type === "runtime-request.updated" && entry.event.payload.status === "cancelled", ).length, - 2, + dismissedCount, ); const repeated = yield* ingestor.ingestNormalized(input); assert.isFalse(repeated.some((entry) => entry.event.type === "runtime-request.updated")); @@ -1323,6 +1341,373 @@ layer("ProviderEventIngestorV2", (it) => { ), ); + it.effect.each(["interrupted", "cancelled"] as const)( + "preserves an approval resolved after terminal normalization reads it pending (%s)", + (status) => + Effect.scoped( + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const eventStore = yield* EventStore.EventStoreV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const now = yield* DateTime.now; + const threadEvent = yield* threadCreatedEvent(now); + const threadId = threadEvent.threadId; + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThreadId = idAllocator.derive.providerThread({ + driver: CODEX_DRIVER, + nativeThreadId: `${threadId}:approval-race`, + }); + const providerTurnId = idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: `${threadId}:approval-race`, + }); + const nodeId = NodeId.make(`${threadId}:approval`); + const request: OrchestrationV2RuntimeRequest = { + id: RuntimeRequestId.make(`${threadId}:approval`), + nodeId, + providerTurnId, + nativeRequestRef: null, + kind: "command", + status: "pending", + responseCapability: { type: "live", providerSessionId }, + createdAt: now, + resolvedAt: null, + }; + const node: OrchestrationV2ExecutionNode = { + id: nodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId: nodeId, + kind: "approval_request", + status: "waiting", + countsForRun: false, + providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: request.id, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }; + const item: OrchestrationV2TurnItem = { + id: TurnItemId.make(`${threadId}:approval`), + threadId, + runId: null, + nodeId, + providerThreadId, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "approval_request", + requestId: request.id, + requestKind: "command", + }; + const seedEvents: Array = [threadEvent]; + for (const payload of [ + { type: "runtime-request.updated" as const, payload: request }, + { type: "node.updated" as const, payload: node }, + { type: "turn-item.updated" as const, payload: item }, + ]) + seedEvents.push({ + id: yield* idAllocator.allocate.event({ threadId }), + threadId, + occurredAt: now, + ...payload, + }); + yield* eventSink.write({ events: seedEvents }); + const normalized = yield* Deferred.make(); + const releaseTerminalWrite = yield* Deferred.make(); + const gatedSink = EventSink.EventSinkV2.of({ + ...eventSink, + write: (input) => + Deferred.succeed(normalized, undefined).pipe( + Effect.andThen(Deferred.await(releaseTerminalWrite)), + Effect.andThen(eventSink.write(input)), + ), + }); + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2.pipe( + Effect.provide(Layer.fresh(ProviderEventIngestor.layer)), + Effect.provideService(EventSink.EventSinkV2, gatedSink), + ); + const terminal = yield* ingestor + .ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + event: { + type: "provider_turn.updated", + driver: CODEX_DRIVER, + providerTurn: { + id: providerTurnId, + providerThreadId, + nodeId, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status, + startedAt: now, + completedAt: now, + }, + }, + }) + .pipe(Effect.forkScoped); + yield* Deferred.await(normalized); + const responseEvents: Array = []; + for (const payload of [ + { + type: "runtime-request.updated" as const, + payload: { + ...request, + status: "resolved" as const, + decision: "accept" as const, + resolvedAt: now, + }, + }, + { + type: "node.updated" as const, + payload: { ...node, status: "completed" as const, completedAt: now }, + }, + { + type: "turn-item.updated" as const, + payload: { ...item, status: "completed" as const, completedAt: now }, + }, + ]) + responseEvents.push({ + id: yield* idAllocator.allocate.event({ threadId }), + threadId, + occurredAt: now, + ...payload, + }); + yield* eventSink.write({ events: responseEvents }); + yield* Deferred.succeed(releaseTerminalWrite, undefined); + const committedTerminal = yield* Fiber.join(terminal); + const projection = yield* projections.getThreadProjection(threadId); + const approved = projection.runtimeRequests.find((entry) => entry.id === request.id)!; + assert.equal(approved.status, "resolved"); + assert.equal(approved.decision, "accept"); + assert.equal(projection.nodes.find((entry) => entry.id === nodeId)!.status, "completed"); + assert.equal( + projection.turnItems.find((entry) => entry.id === item.id)!.status, + "completed", + ); + assert.deepEqual( + committedTerminal.map((stored) => stored.event.type), + ["provider-turn.updated"], + ); + const history = yield* eventStore.read({ threadId }).pipe(Stream.runCollect); + assert.isFalse( + history.some( + (stored) => + stored.event.type === "runtime-request.updated" && + stored.event.payload.status === "cancelled", + ), + ); + }), + ), + ); + + it.effect.each(["command", "user_input"] as const)( + "rejects a %s response planned before terminal normalization cancelled the request", + (kind) => + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2.pipe(Effect.provide(EffectOutbox.layer)); + const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2.pipe( + Effect.provide(CommandReceiptStore.layer), + ); + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const now = yield* DateTime.now; + const threadEvent = yield* threadCreatedEvent(now); + const threadId = threadEvent.threadId; + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThreadId = idAllocator.derive.providerThread({ + driver: CODEX_DRIVER, + nativeThreadId: `${threadId}:respond-race`, + }); + const providerTurnId = idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: `${threadId}:respond-race`, + }); + const nodeId = NodeId.make(`${threadId}:request`); + const request: OrchestrationV2RuntimeRequest = { + id: RuntimeRequestId.make(`${threadId}:request`), + nodeId, + providerTurnId, + nativeRequestRef: null, + kind, + status: "pending", + responseCapability: { type: "live", providerSessionId }, + createdAt: now, + resolvedAt: null, + }; + const node: OrchestrationV2ExecutionNode = { + id: nodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId: nodeId, + kind: kind === "user_input" ? "user_input_request" : "approval_request", + status: "waiting", + countsForRun: false, + providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: request.id, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }; + const itemBase = { + id: TurnItemId.make(`${threadId}:request`), + threadId, + runId: null, + nodeId, + providerThreadId, + providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "waiting" as const, + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }; + const item: OrchestrationV2TurnItem = + kind === "user_input" + ? { ...itemBase, type: "user_input_request", requestId: request.id, questions: [] } + : { ...itemBase, type: "approval_request", requestId: request.id, requestKind: kind }; + const domainEvents = ( + payloads: ReadonlyArray< + | { + readonly type: "runtime-request.updated"; + readonly payload: OrchestrationV2RuntimeRequest; + } + | { readonly type: "node.updated"; readonly payload: OrchestrationV2ExecutionNode } + | { readonly type: "turn-item.updated"; readonly payload: OrchestrationV2TurnItem } + >, + ) => + Effect.forEach(payloads, (payload) => + Effect.map( + idAllocator.allocate.event({ threadId }), + (id): OrchestrationV2DomainEvent => ({ id, threadId, occurredAt: now, ...payload }), + ), + ); + yield* eventSink.write({ + events: [ + threadEvent, + ...(yield* domainEvents([ + { type: "runtime-request.updated", payload: request }, + { type: "node.updated", payload: node }, + { type: "turn-item.updated", payload: item }, + ])), + ], + }); + // The respond command plans from this pending read before the stop commits. + const planned = yield* projections.getRuntimeResponseContext(threadId, request.id); + assert.equal(planned.request?.status, "pending"); + yield* ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + event: { + type: "provider_turn.updated", + driver: CODEX_DRIVER, + providerTurn: { + id: providerTurnId, + providerThreadId, + nodeId, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "interrupted", + startedAt: now, + completedAt: now, + }, + }, + }); + assert.equal( + (yield* projections.getRuntimeRequest(threadId, request.id))?.status, + "cancelled", + ); + const commandId = CommandId.make(`${threadId}:respond`); + const exit = yield* eventSink + .commitCommand({ + commandId, + threadId, + commandType: "runtime-request.respond", + acceptedAt: now, + events: yield* domainEvents([ + { + type: "runtime-request.updated", + payload: { + ...planned.request!, + status: "resolved", + resolvedAt: now, + ...(kind === "user_input" ? { answers: { q: "yes" } } : { decision: "accept" }), + }, + }, + { + type: "node.updated", + payload: { ...planned.node!, status: "completed", completedAt: now }, + }, + { + type: "turn-item.updated", + payload: { ...planned.item!, status: "completed", completedAt: now }, + }, + ]), + effects: [ + { + id: `effect:${commandId}:runtime-request.respond:${request.id}`, + commandId, + threadId, + request: { + type: "runtime-request.respond", + providerSessionId, + requestId: request.id, + ...(kind === "user_input" ? { answers: { q: "yes" } } : { decision: "accept" }), + }, + }, + ], + }) + .pipe(Effect.exit); + assert.isTrue(Exit.isFailure(exit)); + const error = Exit.isFailure(exit) + ? Cause.findErrorOption(exit.cause).pipe(Option.getOrThrow) + : undefined; + assert.instanceOf(error, EventSink.EventSinkRuntimeRequestNotPendingError); + assert.deepInclude(error, { requestId: request.id, status: "cancelled" }); + assert.isTrue(Option.isNone(yield* receipts.getByCommandId(commandId))); + const projection = yield* projections.getThreadProjection(threadId); + assert.equal( + projection.runtimeRequests.find((entry) => entry.id === request.id)?.status, + "cancelled", + ); + assert.equal(projection.nodes.find((entry) => entry.id === nodeId)?.status, "cancelled"); + assert.equal( + projection.turnItems.find((entry) => entry.id === item.id)?.status, + "cancelled", + ); + assert.deepEqual(yield* outbox.listByCommandId(commandId), []); + }), + ); + it.effect("persists a failed provider terminal as one expected error item", () => 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 8e916ab79319b..f04c1d7774224 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -345,13 +345,18 @@ export const layer: Layer.Layer< ); }); - const dismissNativeUserInputs = Effect.fn("ProviderEventIngestor.dismissNativeUserInputs")( + const dismissNativeRequests = Effect.fn("ProviderEventIngestor.dismissNativeRequests")( function* ( input: ProviderEventIngestInput, providerTurnId: ProviderTurnId, + turnStatus: OrchestrationV2ProviderTurn["status"], threadId = input.threadId, ) { - const pending = yield* projections.getPendingNativeUserInputs(threadId, providerTurnId); + // A stopped turn's approval callbacks are gone with it. A completed or failed root turn + // can leave subagent turns working whose approvals are filed under the root turn. + const pending = yield* projections.getPendingNativeRequests(threadId, providerTurnId, { + includeApprovals: turnStatus === "interrupted" || turnStatus === "cancelled", + }); const now = yield* DateTime.now; const events: Array = []; for (const request of pending.runtimeRequests) { @@ -360,7 +365,19 @@ export const layer: Layer.Layer< type: "runtime-request.updated", threadId, nodeId: request.nodeId, - payload: { ...request, status: "cancelled", resolvedAt: now }, + payload: { + ...request, + status: "cancelled", + resolvedAt: now, + ...(request.kind === "user_input" + ? {} + : { + responseCapability: { + type: "not_resumable" as const, + reason: "The provider turn ended before this request was resolved.", + }, + }), + }, }), ); } @@ -585,9 +602,10 @@ export const layer: Layer.Layer< ...(["completed", "interrupted", "failed", "cancelled"].includes( input.event.providerTurn.status, ) - ? yield* dismissNativeUserInputs( + ? yield* dismissNativeRequests( input, input.event.providerTurn.id, + input.event.providerTurn.status, input.event.threadId, ) : []), @@ -682,7 +700,11 @@ export const layer: Layer.Layer< // Live output is routed to the command output hub before ingestion. return []; case "turn.terminal": - const dismissed = yield* dismissNativeUserInputs(input, input.event.providerTurnId); + const dismissed = yield* dismissNativeRequests( + input, + input.event.providerTurnId, + input.event.status, + ); if (input.event.status !== "failed") { return dismissed; } diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 3ae3a9974b487..1c76ded2a547c 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -237,7 +237,7 @@ it.effect( getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"), getThreadProviderContext: () => Effect.die("unused getThreadProviderContext"), getRuntimeResponseContext: () => Effect.die("unused getRuntimeResponseContext"), - getPendingNativeUserInputs: () => Effect.die("unused getPendingNativeUserInputs"), + getPendingNativeRequests: () => Effect.die("unused getPendingNativeRequests"), getProviderControlContext: (_threadId, target) => Ref.get(projection).pipe( Effect.map((current) => ({