From dd363c430c63c38630f3c55cd506d79179000e65 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Thu, 24 Sep 2026 02:11:29 -0600 Subject: [PATCH] fix(server): preserve pending turn admission during provider bind A reused provider can still report its previous turn as running when a new exact turn is reserved. Bind that reservation as starting so matching turn.started can be accepted, reject malformed and terminal binds, and retain strict incarnation fences. --- .../Layers/ProviderCommandReactor.test.ts | 94 +++++++++++++++++++ .../Layers/ProviderCommandReactor.ts | 15 ++- .../Layers/ProviderRuntimeIngestion.test.ts | 21 +++++ .../decider.sessionLifecycle.test.ts | 6 ++ apps/server/src/orchestration/decider.ts | 1 + 5 files changed, 136 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index a3be323be0..8ca84fd7c1 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -2005,6 +2005,100 @@ describe("ProviderCommandReactor", () => { }), ); + effectIt.effect( + "keeps exact admission pending while a provider reports a running prior turn", + () => + Effect.gen(function* () { + const sendEntered = yield* Deferred.make(); + const releaseSend = yield* Deferred.make(); + const requestId = CommandId.make("cmd-running-provider-exact-admission"); + const messageId = asMessageId("message-running-provider-exact-admission"); + const harness = yield* Effect.promise(() => + createHarness({ + startSessionEffect: (session) => Effect.succeed({ ...session, status: "running" }), + sendTurnEffect: () => + Deferred.succeed(sendEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseSend)), + Effect.as({ threadId: ThreadId.make("thread-1"), turnId: asTurnId("new-turn") }), + ), + }), + ); + + yield* harness.engine.dispatch({ + type: "thread.turn.start", + commandId: requestId, + threadId: ThreadId.make("thread-1"), + message: { messageId, role: "user", text: "follow up", attachments: [] }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: isoAt(0), + }); + yield* Deferred.await(sendEntered); + let thread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + expect(thread?.session?.status).toBe("starting"); + expect(thread?.session?.pendingTurnRequestId).toBe(requestId); + expect(thread?.session?.pendingTurnMessageId).toBe(messageId); + + yield* Deferred.succeed(releaseSend, undefined); + yield* Effect.promise(() => + waitFor(async () => { + const thread = (await harness.readModel()).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + return thread?.session?.activeTurnId === asTurnId("new-turn"); + }), + ); + thread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + expect(thread?.session?.status).toBe("running"); + expect(thread?.session?.activeTurnRequestId).toBe(requestId); + expect(thread?.session?.pendingTurnRequestId).toBeUndefined(); + }), + ); + + for (const status of ["error", "closed"] as const) { + effectIt.effect(`rejects exact admission when the provider session is ${status}`, () => + Effect.gen(function* () { + const requestId = CommandId.make(`cmd-terminal-provider-${status}`); + const harness = yield* Effect.promise(() => + createHarness({ + startSessionEffect: (session) => Effect.succeed({ ...session, status }), + }), + ); + yield* harness.engine.dispatch({ + type: "thread.turn.start", + commandId: requestId, + threadId: ThreadId.make("thread-1"), + message: { + messageId: asMessageId(`message-terminal-provider-${status}`), + role: "user", + text: "follow up", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: isoAt(0), + }); + yield* Effect.promise(() => + waitFor(async () => { + const thread = (await harness.readModel()).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + return thread?.session?.status === "error"; + }), + ); + const thread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === ThreadId.make("thread-1"), + ); + expect(thread?.session?.failedTurnRequestId).toBe(requestId); + expect(harness.sendTurn).not.toHaveBeenCalled(); + }), + ); + } + effectIt.effect("does not disarm when raw turn start ingestion never accepts the CAS", () => Effect.gen(function* () { const testClock = yield* TestClock.make(); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index e5fcf3d49c..fd9d22bb73 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -1048,10 +1048,23 @@ const make = Effect.gen(function* () { detail: `Provider session '${session.threadId}' started without a provider instance or session incarnation id.`, }); } + if ( + options?.pendingTurnStart === true && + (session.status === "error" || session.status === "closed") + ) { + return yield* new ProviderAdapterRequestError({ + provider: providerErrorLabel(session.provider), + method: "thread.turn.start", + detail: `Provider session '${session.threadId}' is ${session.status}; it cannot accept a new turn.`, + }); + } const sessionBinding: OrchestrationSession = { threadId, + // The provider can still report the previous turn as running while + // a new exact admission is reserved. The reservation owns the + // projected lifecycle until its matching turn.started is accepted. status: - options?.pendingTurnStart === true && session.status === "ready" + options?.pendingTurnStart === true ? "starting" : mapProviderSessionStatusToOrchestrationStatus(session.status), providerName: session.provider, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 9bb1d25e74..91cdbb9373 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1663,6 +1663,27 @@ describe("ProviderRuntimeIngestion", () => { expect(thread?.session?.pendingTurnRequestId).toBe(requestId); expect(thread?.messages.some((message) => message.role === "assistant")).toBe(false); + // The prior turn can finish on the same provider session after the new + // admission is reserved. Its terminal event must not clear that pending + // admission before the matching turn.started arrives. + await harness.emitAndDrain([ + { + type: "turn.completed", + eventId: asEventId("evt-runtime-session-b-prior-turn-completed"), + provider: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + threadId, + turnId: asTurnId("turn-runtime-session-b-prior"), + admissionRequestId: CommandId.make("cmd-runtime-session-b-prior"), + sessionIncarnationId: sessionB, + createdAt: "2026-01-01T00:00:02.500Z", + payload: { state: "completed" }, + }, + ]); + thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); + expect(thread?.session?.status).toBe("starting"); + expect(thread?.session?.pendingTurnRequestId).toBe(requestId); + const currentTurnId = asTurnId("turn-runtime-session-b"); harness.emit({ type: "turn.started", diff --git a/apps/server/src/orchestration/decider.sessionLifecycle.test.ts b/apps/server/src/orchestration/decider.sessionLifecycle.test.ts index f60a02abe8..79b2158d40 100644 --- a/apps/server/src/orchestration/decider.sessionLifecycle.test.ts +++ b/apps/server/src/orchestration/decider.sessionLifecycle.test.ts @@ -199,6 +199,12 @@ it.layer(NodeServices.layer)("session lifecycle CAS decider", (it) => { readModel: makeReadModel(current), }); expect(stale).toEqual([]); + + const runningWithPendingRequest = yield* decideOrchestrationCommand({ + command: { ...command, session: { ...acceptedSession, status: "running" } }, + readModel: makeReadModel(current), + }); + expect(runningWithPendingRequest).toEqual([]); }), ); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 8c9a180a60..5afe8eb9a2 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -2629,6 +2629,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" thread.session.activeTurnId !== null || command.session.pendingTurnRequestId !== command.requestId || command.session.pendingTurnMessageId !== command.messageId || + command.session.status !== "starting" || command.session.providerInstanceId !== command.modelSelection.instanceId || command.session.runtimeMode !== command.runtimeMode || command.session.sessionIncarnationId === undefined ||