From 520e9fe29523810a996b4cb0ee05508f2e2e76df Mon Sep 17 00:00:00 2001 From: F4llen Date: Sat, 5 Sep 2026 03:10:38 -0400 Subject: [PATCH 1/5] test(server): expect turns requested during compaction to be sent --- .../Layers/ProviderCommandReactor.test.ts | 112 +++++++++++++++--- 1 file changed, 96 insertions(+), 16 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index a8b26fe52cbd..0386eb0ba3ad 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -931,7 +931,7 @@ describe("ProviderCommandReactor", () => { }), ); - effectIt.effect("keeps turns blocked until compaction restores the session", () => + effectIt.effect("sends turns requested during compaction once it settles", () => Effect.gen(function* () { const readyDispatchStarted = yield* Deferred.make(); const releaseReadyDispatch = yield* Deferred.make(); @@ -964,11 +964,11 @@ describe("ProviderCommandReactor", () => { createdAt, }); - yield* dispatchTurn("before-blocked-compact", "hello", now); + yield* dispatchTurn("before-queued-compact", "hello", now); yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 1)); yield* harness.engine.dispatch({ type: "thread.session.set", - commandId: CommandId.make("cmd-session-ready-before-blocked-compact"), + commandId: CommandId.make("cmd-session-ready-before-queued-compact"), threadId, session: { threadId, @@ -984,32 +984,112 @@ describe("ProviderCommandReactor", () => { }); blockReadyDispatch = true; - yield* dispatchTurn("blocked-compact", "/compact", "2026-01-01T00:00:01.000Z"); + yield* dispatchTurn("queued-compact", "/compact", "2026-01-01T00:00:01.000Z"); yield* Deferred.await(readyDispatchStarted); - yield* dispatchTurn("during-compact-recovery", "too soon", "2026-01-01T00:00:02.000Z"); - yield* Effect.promise(() => - waitFor(async () => { - const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); - return ( - thread?.activities.some( - (activity) => activity.kind === "provider.turn.start.failed", - ) === true - ); - }), - ); + yield* dispatchTurn("queued-first", "first", "2026-01-01T00:00:02.000Z"); + yield* dispatchTurn("queued-second", "second", "2026-01-01T00:00:03.000Z"); + yield* Effect.promise(() => harness.drain()); expect(harness.sendTurn).toHaveBeenCalledTimes(1); expect(yield* Effect.promise(() => harness.readPendingTurnStarts())).toEqual([ { threadId: "thread-1" }, ]); yield* Deferred.succeed(releaseReadyDispatch, undefined); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 3)); + expect( + harness.sendTurn.mock.calls.map(([input]) => (input as { input: string }).input), + ).toEqual(["hello", "first", "second"]); + const thread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === threadId, + ); + expect( + thread?.activities.filter((activity) => activity.kind === "provider.turn.start.failed"), + ).toEqual([]); + }), + ); + + effectIt.effect("fails turns queued during compaction when the thread stops", () => + Effect.gen(function* () { + const releaseCompaction = yield* Deferred.make(); + const harness = yield* Effect.promise(() => + createHarness({ + compactThreadEffect: () => + Deferred.await(releaseCompaction).pipe( + Effect.andThen(Effect.die("Compaction stopped")), + ), + }), + ); + const threadId = ThreadId.make("thread-1"); + const now = "2026-01-01T00:00:00.000Z"; + const dispatchTurn = (id: string, text: string, createdAt: string) => + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`cmd-${id}`), + threadId, + message: { + messageId: asMessageId(`user-message-${id}`), + role: "user", + text, + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + }); + + yield* dispatchTurn("before-stopped-compact", "hello", now); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 1)); + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-ready-before-stopped-compact"), + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + yield* dispatchTurn("stopped-compact", "/compact", "2026-01-01T00:00:01.000Z"); + yield* Effect.promise(() => waitFor(() => harness.compactThread.mock.calls.length === 1)); + yield* dispatchTurn("queued-then-stopped", "after stop", "2026-01-01T00:00:02.000Z"); + yield* harness.engine.dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-queued-compact"), + threadId, + createdAt: "2026-01-01T00:00:03.000Z", + }); + yield* Effect.promise(() => waitFor(() => harness.stopSession.mock.calls.length === 1)); + yield* Deferred.succeed(releaseCompaction, undefined); yield* Effect.promise(() => waitFor(async () => { const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); - return thread?.session?.status === "ready"; + return ( + thread?.activities.filter((activity) => activity.kind === "provider.turn.start.failed") + .length === 2 + ); }), ); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + const thread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === threadId, + ); + expect( + thread?.activities.find( + (activity) => + activity.kind === "provider.turn.start.failed" && + (activity.payload as { requestId?: string }).requestId === + "user-message-queued-then-stopped", + ), + ).toMatchObject({ + payload: { detail: "The thread stopped before this message could be sent." }, + }); }), ); From cc9b80c83686cde0f85091a1798fdf3c5f86ac36 Mon Sep 17 00:00:00 2001 From: F4llen Date: Sat, 5 Sep 2026 03:12:41 -0400 Subject: [PATCH 2/5] fix(server): send messages queued during context compaction Since #9293 turned /compact into a server-side operation, a message sent while a thread compacts was rejected with a turn-start failure and the client dropped it. Messages sent during a running turn steer into that turn, but compaction runs outside any turn, so there was nothing to steer into. The reactor now keeps a per-thread latch for the in-flight compaction instead of a bare flag. A turn start that arrives while the latch is held waits for it, then sends through the normal path once the session is restored. If the thread stopped or errored meanwhile, the message gets a turn-start failure so the client restores the draft. The message was already persisted before the reactor saw it, so nothing new needs to survive a restart. --- .../Layers/ProviderCommandReactor.ts | 90 ++++++++++++------- 1 file changed, 58 insertions(+), 32 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index fc963bcc9cb2..ec116d148b88 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -18,6 +18,7 @@ import * as Cache from "effect/Cache"; import * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; @@ -345,7 +346,18 @@ const make = Effect.gen(function* () { ); const threadModelSelections = new Map(); - const compactingThreadIds = new Set(); + // A thread compacts outside any provider turn, so a message sent meanwhile + // has nothing to steer into. The latch lets that send wait for the session + // to be restored instead of failing; the message itself is already + // persisted, so nothing here needs to survive a restart. + const compactionLatches = new Map>(); + const settleCompactionLatch = (threadId: ThreadId) => + Effect.suspend(() => { + const latch = compactionLatches.get(threadId); + if (latch === undefined) return Effect.void; + compactionLatches.delete(threadId); + return Deferred.succeed(latch, undefined).pipe(Effect.asVoid); + }); const stoppingThreadIds = new Set(); const appendProviderFailureActivity = (input: { @@ -449,7 +461,7 @@ const make = Effect.gen(function* () { const restoreCompaction = Effect.fnUntraced(function* (threadId: ThreadId, fromRunning = false) { if (stoppingThreadIds.has(threadId)) { - compactingThreadIds.delete(threadId); + yield* settleCompactionLatch(threadId); return; } const thread = yield* resolveThreadShell(threadId); @@ -462,7 +474,7 @@ const make = Effect.gen(function* () { return; const completedAt = DateTime.formatIso(yield* DateTime.now); if (stoppingThreadIds.has(threadId)) { - compactingThreadIds.delete(threadId); + yield* settleCompactionLatch(threadId); return; } yield* setThreadSession({ @@ -1379,7 +1391,7 @@ const make = Effect.gen(function* () { } const latestThread = yield* resolveThreadShell(event.payload.threadId); if ( - compactingThreadIds.has(event.payload.threadId) || + compactionLatches.has(event.payload.threadId) || latestThread?.session?.status === "starting" || latestThread?.session?.status === "running" ) { @@ -1389,7 +1401,7 @@ const make = Effect.gen(function* () { ); return; } - compactingThreadIds.add(event.payload.threadId); + compactionLatches.set(event.payload.threadId, yield* Deferred.make()); yield* Effect.gen(function* () { yield* ensureSessionForThread( event.payload.threadId, @@ -1410,38 +1422,52 @@ const make = Effect.gen(function* () { }).pipe( Effect.andThen(restoreCompaction(event.payload.threadId, true)), Effect.catchCause(recoverCompactionFailure), - Effect.ensuring(Effect.sync(() => void compactingThreadIds.delete(event.payload.threadId))), + Effect.ensuring(settleCompactionLatch(event.payload.threadId)), Effect.forkScoped, ); return; } - if (compactingThreadIds.has(event.payload.threadId)) { - return yield* appendTurnStartFailure( - "Provider turn start failed", - "Wait for context compaction to finish before sending another message.", + + const startTurn = Effect.gen(function* () { + const sendTurnRequest = yield* buildSendTurnRequestForThread({ + threadId: event.payload.threadId, + messageText: message.text, + ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), + ...(event.payload.modelSelection !== undefined + ? { modelSelection: event.payload.modelSelection } + : {}), + interactionMode: event.payload.interactionMode, + createdAt: event.payload.createdAt, + }).pipe( + Effect.map(Option.some), + Effect.catchCause((cause) => handleTurnStartFailure(cause).pipe(Effect.as(Option.none()))), ); - } - const sendTurnRequest = yield* buildSendTurnRequestForThread({ - threadId: event.payload.threadId, - messageText: message.text, - ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), - ...(event.payload.modelSelection !== undefined - ? { modelSelection: event.payload.modelSelection } - : {}), - interactionMode: event.payload.interactionMode, - createdAt: event.payload.createdAt, - }).pipe( - Effect.map(Option.some), - Effect.catchCause((cause) => handleTurnStartFailure(cause).pipe(Effect.as(Option.none()))), - ); - if (Option.isNone(sendTurnRequest)) { - return; - } + if (Option.isNone(sendTurnRequest)) { + return; + } - yield* providerService - .sendTurn(sendTurnRequest.value) - .pipe(Effect.asVoid, Effect.catchCause(recoverTurnStartFailure), Effect.forkScoped); + yield* providerService + .sendTurn(sendTurnRequest.value) + .pipe(Effect.asVoid, Effect.catchCause(recoverTurnStartFailure), Effect.forkScoped); + }); + + const compactionLatch = compactionLatches.get(event.payload.threadId); + if (compactionLatch === undefined) { + return yield* startTurn; + } + yield* Deferred.await(compactionLatch).pipe( + Effect.andThen(resolveThreadShell(event.payload.threadId)), + Effect.flatMap((latestThread) => + latestThread?.session?.status === "stopped" || latestThread?.session?.status === "error" + ? appendTurnStartFailure( + "Provider turn start failed", + "The thread stopped before this message could be sent.", + ) + : startTurn, + ), + Effect.forkScoped, + ); }); const processTurnInterruptRequested = Effect.fn("processTurnInterruptRequested")(function* ( @@ -1636,7 +1662,7 @@ const make = Effect.gen(function* () { } const now = event.payload.createdAt; - const wasCompacting = compactingThreadIds.has(thread.id); + const wasCompacting = compactionLatches.has(thread.id); stoppingThreadIds.add(thread.id); const clearStopping = Effect.sync(() => void stoppingThreadIds.delete(thread.id)); yield* ( @@ -1652,7 +1678,7 @@ const make = Effect.gen(function* () { const detail = formatFailureDetail(cause); return Effect.sync(() => { stoppingThreadIds.delete(thread.id); - return wasCompacting && !compactingThreadIds.has(thread.id); + return wasCompacting && !compactionLatches.has(thread.id); }).pipe( Effect.flatMap((compactionSettled) => compactionSettled ? restoreCompaction(thread.id) : Effect.void, From 2378ed76c31e04c213c82371bffe0ffa4a761baa Mon Sep 17 00:00:00 2001 From: F4llen Date: Sat, 5 Sep 2026 03:18:37 -0400 Subject: [PATCH 3/5] test(server): expect queued compaction sends to run one at a time --- .../Layers/ProviderCommandReactor.test.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 0386eb0ba3ad..a50b1d5d50cd 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -995,7 +995,19 @@ describe("ProviderCommandReactor", () => { { threadId: "thread-1" }, ]); + const releaseFirstSend = yield* Deferred.make(); + harness.sendTurn.mockImplementation((input: unknown) => + ((input as { input: string }).input === "first" + ? Deferred.await(releaseFirstSend) + : Effect.void + ).pipe(Effect.as({ threadId, turnId: asTurnId("turn-1") })), + ); yield* Deferred.succeed(releaseReadyDispatch, undefined); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length >= 2)); + yield* Effect.promise(() => harness.drain()); + expect(harness.sendTurn).toHaveBeenCalledTimes(2); + + yield* Deferred.succeed(releaseFirstSend, undefined); yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 3)); expect( harness.sendTurn.mock.calls.map(([input]) => (input as { input: string }).input), From 2f8b6dce02add88e094e57df6a1cf0dcce966a44 Mon Sep 17 00:00:00 2001 From: F4llen Date: Sat, 5 Sep 2026 03:19:55 -0400 Subject: [PATCH 4/5] fix(server): drain compaction-queued sends one at a time Waiters woken by the same latch ran as independent fibers, so two messages queued during compaction could reach the provider out of order. Each queued send now waits for the one queued before it and awaits its own provider call before releasing the next, so the thread sees them in the order they were sent. --- .../Layers/ProviderCommandReactor.ts | 79 ++++++++++++------- 1 file changed, 52 insertions(+), 27 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index ec116d148b88..e89c1f968dbc 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -351,6 +351,9 @@ const make = Effect.gen(function* () { // to be restored instead of failing; the message itself is already // persisted, so nothing here needs to survive a restart. const compactionLatches = new Map>(); + // Sends waiting on a compaction leave one at a time, each behind the send + // queued before it, so they reach the provider in the order they were sent. + const queuedTurnStartTails = new Map>(); const settleCompactionLatch = (threadId: ThreadId) => Effect.suspend(() => { const latch = compactionLatches.get(threadId); @@ -1428,43 +1431,65 @@ const make = Effect.gen(function* () { return; } - const startTurn = Effect.gen(function* () { - const sendTurnRequest = yield* buildSendTurnRequestForThread({ - threadId: event.payload.threadId, - messageText: message.text, - ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), - ...(event.payload.modelSelection !== undefined - ? { modelSelection: event.payload.modelSelection } - : {}), - interactionMode: event.payload.interactionMode, - createdAt: event.payload.createdAt, - }).pipe( - Effect.map(Option.some), - Effect.catchCause((cause) => handleTurnStartFailure(cause).pipe(Effect.as(Option.none()))), - ); + const startTurn = (queued: boolean) => + Effect.gen(function* () { + const sendTurnRequest = yield* buildSendTurnRequestForThread({ + threadId: event.payload.threadId, + messageText: message.text, + ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), + ...(event.payload.modelSelection !== undefined + ? { modelSelection: event.payload.modelSelection } + : {}), + interactionMode: event.payload.interactionMode, + createdAt: event.payload.createdAt, + }).pipe( + Effect.map(Option.some), + Effect.catchCause((cause) => + handleTurnStartFailure(cause).pipe(Effect.as(Option.none())), + ), + ); - if (Option.isNone(sendTurnRequest)) { - return; - } + if (Option.isNone(sendTurnRequest)) { + return; + } - yield* providerService - .sendTurn(sendTurnRequest.value) - .pipe(Effect.asVoid, Effect.catchCause(recoverTurnStartFailure), Effect.forkScoped); - }); + const send = providerService + .sendTurn(sendTurnRequest.value) + .pipe(Effect.asVoid, Effect.catchCause(recoverTurnStartFailure)); + yield* queued ? send : Effect.forkScoped(send); + }); - const compactionLatch = compactionLatches.get(event.payload.threadId); - if (compactionLatch === undefined) { - return yield* startTurn; + if (!compactionLatches.has(event.payload.threadId)) { + return yield* startTurn(false); } - yield* Deferred.await(compactionLatch).pipe( - Effect.andThen(resolveThreadShell(event.payload.threadId)), + const threadId = event.payload.threadId; + const previousTail = queuedTurnStartTails.get(threadId); + const tail = yield* Deferred.make(); + queuedTurnStartTails.set(threadId, tail); + const awaitCompactionSettled: Effect.Effect = Effect.suspend(() => { + const latch = compactionLatches.get(threadId); + return latch === undefined + ? Effect.void + : Deferred.await(latch).pipe(Effect.andThen(awaitCompactionSettled)); + }); + yield* (previousTail === undefined ? Effect.void : Deferred.await(previousTail)).pipe( + Effect.andThen(awaitCompactionSettled), + Effect.andThen(resolveThreadShell(threadId)), Effect.flatMap((latestThread) => latestThread?.session?.status === "stopped" || latestThread?.session?.status === "error" ? appendTurnStartFailure( "Provider turn start failed", "The thread stopped before this message could be sent.", ) - : startTurn, + : startTurn(true), + ), + Effect.ensuring( + Effect.suspend(() => { + if (queuedTurnStartTails.get(threadId) === tail) { + queuedTurnStartTails.delete(threadId); + } + return Deferred.succeed(tail, undefined); + }), ), Effect.forkScoped, ); From 4505e074c605c99e1ded61170cef311a06e02fb8 Mon Sep 17 00:00:00 2001 From: F4llen Date: Sat, 5 Sep 2026 07:45:31 -0400 Subject: [PATCH 5/5] fix(server): keep post-compaction sends behind a draining send A turn requested after the compaction latch settles found no latch to wait on, so it took the fast path and could overtake a queued send that was still draining. --- .../Layers/ProviderCommandReactor.test.ts | 84 +++++++++++++++++++ .../Layers/ProviderCommandReactor.ts | 9 +- 2 files changed, 90 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index a50b1d5d50cd..4ee8b9a8c4da 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -1021,6 +1021,90 @@ describe("ProviderCommandReactor", () => { }), ); + effectIt.effect("queues turns requested while a compaction-queued send is draining", () => + Effect.gen(function* () { + const readyDispatchStarted = yield* Deferred.make(); + const releaseReadyDispatch = yield* Deferred.make(); + let blockReadyDispatch = false; + const harness = yield* Effect.promise(() => + createHarness({ + beforeReadySessionDispatch: () => + blockReadyDispatch + ? Deferred.succeed(readyDispatchStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseReadyDispatch)), + ) + : Effect.void, + }), + ); + const threadId = ThreadId.make("thread-1"); + const now = "2026-01-01T00:00:00.000Z"; + const dispatchTurn = (id: string, text: string, createdAt: string) => + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(`cmd-${id}`), + threadId, + message: { + messageId: asMessageId(`user-message-${id}`), + role: "user", + text, + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + }); + + yield* dispatchTurn("before-draining-compact", "hello", now); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 1)); + yield* harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-ready-before-draining-compact"), + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + + blockReadyDispatch = true; + yield* dispatchTurn("draining-compact", "/compact", "2026-01-01T00:00:01.000Z"); + yield* Deferred.await(readyDispatchStarted); + yield* dispatchTurn("draining-first", "first", "2026-01-01T00:00:02.000Z"); + + const firstSendStarted = yield* Deferred.make(); + const releaseFirstSend = yield* Deferred.make(); + harness.sendTurn.mockImplementation((input: unknown) => + ((input as { input: string }).input === "first" + ? Deferred.succeed(firstSendStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseFirstSend)), + ) + : Effect.void + ).pipe(Effect.as({ threadId, turnId: asTurnId("turn-1") })), + ); + yield* Deferred.succeed(releaseReadyDispatch, undefined); + yield* Deferred.await(firstSendStarted); + + // The latch is gone by now, so this send only stays in order if it waits + // on the send still draining ahead of it. + yield* dispatchTurn("draining-late", "late", "2026-01-01T00:00:04.000Z"); + yield* Effect.promise(() => harness.drain()); + expect(harness.sendTurn).toHaveBeenCalledTimes(2); + + yield* Deferred.succeed(releaseFirstSend, undefined); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 3)); + expect( + harness.sendTurn.mock.calls.map(([input]) => (input as { input: string }).input), + ).toEqual(["hello", "first", "late"]); + }), + ); + effectIt.effect("fails turns queued during compaction when the thread stops", () => Effect.gen(function* () { const releaseCompaction = yield* Deferred.make(); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index e89c1f968dbc..18053a1030bb 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -1459,11 +1459,14 @@ const make = Effect.gen(function* () { yield* queued ? send : Effect.forkScoped(send); }); - if (!compactionLatches.has(event.payload.threadId)) { - return yield* startTurn(false); - } const threadId = event.payload.threadId; const previousTail = queuedTurnStartTails.get(threadId); + // A send that is still draining behind a settled compaction has no latch + // left to wait on, so a turn arriving now has to queue behind its tail to + // stay in submission order. + if (!compactionLatches.has(threadId) && previousTail === undefined) { + return yield* startTurn(false); + } const tail = yield* Deferred.make(); queuedTurnStartTails.set(threadId, tail); const awaitCompactionSettled: Effect.Effect = Effect.suspend(() => {