From 6014e2082e9c133167c0295950ec238c8646f37e Mon Sep 17 00:00:00 2001 From: flamboh Date: Fri, 4 Sep 2026 00:51:34 -0700 Subject: [PATCH] fix(server): queue messages sent during context compaction --- .../Layers/ProjectionPipeline.test.ts | 139 ++++++++++++ .../Layers/ProjectionPipeline.ts | 66 +++--- .../Layers/ProviderCommandReactor.test.ts | 187 ++++++++++++++- .../Layers/ProviderCommandReactor.ts | 212 ++++++++++++------ docs/user/composer.md | 2 +- 5 files changed, 505 insertions(+), 101 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 1b8a451175f3..72b2d3ea9ea4 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -3259,6 +3259,145 @@ it.layer(makeProjectionPipelinePrefixedTestLayer("t3-pending-turn-terminal-test- assert.deepEqual(pendingRows, [{ messageId: "new-message" }]); }), ); + + it.effect("promotes a turn start deferred behind compaction when the session restores", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-deferred-compaction-turn"); + const compactMessageId = MessageId.make("message-deferred-compact"); + const queuedMessageId = MessageId.make("message-deferred-queued"); + const compactAt = "2026-02-26T16:00:00.000Z"; + const queuedAt = "2026-02-26T16:00:01.000Z"; + + for (const [index, message] of [ + { id: compactMessageId, text: "/compact", createdAt: compactAt }, + { id: queuedMessageId, text: "send after compact", createdAt: queuedAt }, + ].entries()) { + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.make(`evt-deferred-message-${index}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: message.createdAt, + commandId: CommandId.make(`cmd-deferred-message-${index}`), + causationEventId: null, + correlationId: CorrelationId.make(`cmd-deferred-message-${index}`), + metadata: {}, + payload: { + threadId, + messageId: message.id, + role: "user", + text: message.text, + attachments: [], + turnId: null, + streaming: false, + createdAt: message.createdAt, + updatedAt: message.createdAt, + }, + }); + yield* eventStore.append({ + type: "thread.turn-start-requested", + eventId: EventId.make(`evt-deferred-turn-${index}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: message.createdAt, + commandId: CommandId.make(`cmd-deferred-turn-${index}`), + causationEventId: null, + correlationId: CorrelationId.make(`cmd-deferred-turn-${index}`), + metadata: {}, + payload: { + threadId, + messageId: message.id, + ...(index === 1 + ? { + sourceProposedPlan: { + threadId: ThreadId.make("thread-deferred-plan-source"), + planId: "plan-deferred", + }, + } + : {}), + runtimeMode: "full-access", + createdAt: message.createdAt, + }, + }); + } + yield* eventStore.append({ + type: "thread.activity-appended", + eventId: EventId.make("evt-deferred-compaction-complete"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-02-26T16:00:02.000Z", + commandId: CommandId.make("cmd-deferred-compaction-complete"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-deferred-compaction-complete"), + metadata: {}, + payload: { + threadId, + activity: { + id: EventId.make("activity-deferred-compaction-complete"), + tone: "info", + kind: "context-compaction", + summary: "Context compacted", + payload: { requestId: compactMessageId }, + turnId: null, + createdAt: "2026-02-26T16:00:02.000Z", + }, + }, + }); + yield* eventStore.append({ + type: "thread.session-set", + eventId: EventId.make("evt-deferred-session-ready"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-02-26T16:00:03.000Z", + commandId: CommandId.make("server:provider-session-set:deferred"), + causationEventId: null, + correlationId: CorrelationId.make("server:provider-session-set:deferred"), + metadata: {}, + payload: { + threadId, + session: { + threadId, + status: "ready", + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: "2026-02-26T16:00:03.000Z", + }, + }, + }); + + yield* projectionPipeline.bootstrap; + + const pendingRows = yield* sql<{ + readonly messageId: string; + readonly sourceThreadId: string | null; + readonly sourcePlanId: string | null; + readonly requestedAt: string; + }>` + SELECT + pending_message_id AS "messageId", + source_proposed_plan_thread_id AS "sourceThreadId", + source_proposed_plan_id AS "sourcePlanId", + requested_at AS "requestedAt" + FROM projection_turns + WHERE thread_id = ${threadId} + AND turn_id IS NULL + AND state = 'pending' + `; + assert.deepEqual(pendingRows, [ + { + messageId: queuedMessageId, + sourceThreadId: "thread-deferred-plan-source", + sourcePlanId: "plan-deferred", + requestedAt: queuedAt, + }, + ]); + }), + ); }, ); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 1ba1b6afa4f3..cd812edaab88 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1,6 +1,7 @@ import { ApprovalRequestId, type ChatAttachment, + type MessageId, type OrchestrationEvent, type OrchestrationSessionStatus, ThreadId, @@ -503,6 +504,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const serverConfig = yield* ServerConfig; + const compactRequestIds = new Map(); const applyProjectsProjection: ProjectorDefinition["apply"] = Effect.fn( "applyProjectsProjection", @@ -1263,35 +1265,35 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti )(function* (event, _attachmentSideEffects) { switch (event.type) { case "thread.created": + compactRequestIds.delete(event.payload.threadId); yield* projectionTurnRepository.deleteByThreadId({ threadId: event.payload.threadId, }); return; case "thread.turn-start-requested": { - const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({ - threadId: event.payload.threadId, - }); - if (Option.isSome(pendingTurnStart)) { - const pendingMessage = yield* projectionThreadMessageRepository.getByMessageId({ - messageId: pendingTurnStart.value.messageId, - }); - if ( - Option.isSome(pendingMessage) && - pendingMessage.value.role === "user" && - (pendingMessage.value.attachments?.length ?? 0) === 0 && - pendingMessage.value.text.trim().toLowerCase() === "/compact" - ) { - return; - } - } - yield* projectionTurnRepository.replacePendingTurnStart({ + const nextPendingTurnStart = { threadId: event.payload.threadId, messageId: event.payload.messageId, sourceProposedPlanThreadId: event.payload.sourceProposedPlan?.threadId ?? null, sourceProposedPlanId: event.payload.sourceProposedPlan?.planId ?? null, requestedAt: event.payload.createdAt, + }; + const requestedMessage = yield* projectionThreadMessageRepository.getByMessageId({ + messageId: event.payload.messageId, }); + const isCompactRequest = + Option.isSome(requestedMessage) && + requestedMessage.value.role === "user" && + (requestedMessage.value.attachments?.length ?? 0) === 0 && + requestedMessage.value.text.trim().toLowerCase() === "/compact"; + if (isCompactRequest) { + if (compactRequestIds.has(event.payload.threadId)) { + return; + } + compactRequestIds.set(event.payload.threadId, event.payload.messageId); + } + yield* projectionTurnRepository.replacePendingTurnStart(nextPendingTurnStart); return; } @@ -1328,16 +1330,29 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti case "thread.session-set": { const turnId = event.payload.session.activeTurnId; if (turnId === null || event.payload.session.status !== "running") { - if ( - (event.payload.session.status === "ready" && - event.commandId?.startsWith("server:provider-session-set:") === true) || + const restoredReadySession = + event.payload.session.status === "ready" && + event.commandId?.startsWith("server:provider-session-set:") === true; + const terminalSession = event.payload.session.status === "error" || event.payload.session.status === "stopped" || - event.payload.session.status === "interrupted" - ) { - yield* projectionTurnRepository.deletePendingTurnStartByThreadId({ - threadId: event.payload.threadId, - }); + event.payload.session.status === "interrupted"; + if (restoredReadySession || terminalSession) { + const pendingTurnStart = + yield* projectionTurnRepository.getPendingTurnStartByThreadId({ + threadId: event.payload.threadId, + }); + const compactRequestId = compactRequestIds.get(event.payload.threadId); + const pendingTurnBelongsToCompaction = + compactRequestId === undefined || + Option.isNone(pendingTurnStart) || + pendingTurnStart.value.messageId === compactRequestId; + if (terminalSession || pendingTurnBelongsToCompaction) { + yield* projectionTurnRepository.deletePendingTurnStartByThreadId({ + threadId: event.payload.threadId, + }); + } + compactRequestIds.delete(event.payload.threadId); } // Leaving the "running" session status is the turn-end signal: // settle still-running turns so their duration reflects the whole @@ -1461,6 +1476,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti yield* projectionTurnRepository.deletePendingTurnStartByThreadId({ threadId: event.payload.threadId, }); + compactRequestIds.delete(event.payload.threadId); return; } diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 1c5e834ec2d7..f9e878d7000c 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -260,7 +260,7 @@ describe("ProviderCommandReactor", () => { ), ); }); - const sendTurn = vi.fn((_: unknown) => + const sendTurn = vi.fn((_) => Effect.succeed({ threadId: ThreadId.make("thread-1"), turnId: asTurnId("turn-1"), @@ -351,7 +351,7 @@ describe("ProviderCommandReactor", () => { const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; const service: ProviderServiceShape = { startSession: startSession as ProviderServiceShape["startSession"], - sendTurn: sendTurn as ProviderServiceShape["sendTurn"], + sendTurn, compactThread, interruptTurn: interruptTurn as ProviderServiceShape["interruptTurn"], respondToRequest: respondToRequest as ProviderServiceShape["respondToRequest"], @@ -910,13 +910,21 @@ describe("ProviderCommandReactor", () => { }), ); - effectIt.effect("keeps turns blocked until compaction restores the session", () => + effectIt.effect("queues turns until failed compaction restores the session", () => Effect.gen(function* () { const readyDispatchStarted = yield* Deferred.make(); const releaseReadyDispatch = yield* Deferred.make(); let blockReadyDispatch = false; const harness = yield* Effect.promise(() => createHarness({ + compactThreadEffect: () => + Effect.fail( + new ProviderAdapterRequestError({ + provider: "codex", + method: "thread.compact", + detail: "compaction failed", + }), + ), beforeReadySessionDispatch: () => blockReadyDispatch ? Deferred.succeed(readyDispatchStarted, undefined).pipe( @@ -966,29 +974,188 @@ describe("ProviderCommandReactor", () => { yield* dispatchTurn("blocked-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"); + const queuedMessageId = asMessageId("user-message-during-compact-recovery"); + yield* dispatchTurn( + "during-compact-recovery", + "send after compact", + "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 - ); + return thread?.messages.some((message) => message.id === queuedMessageId) === true; }), ); expect(harness.sendTurn).toHaveBeenCalledTimes(1); + const compactingThread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === threadId, + ); + expect( + compactingThread?.activities.some( + (activity) => + activity.kind === "provider.turn.start.failed" && + (activity.payload as { readonly requestId?: unknown } | null)?.requestId === + queuedMessageId, + ), + ).toBe(false); expect(yield* Effect.promise(() => harness.readPendingTurnStarts())).toEqual([ { threadId: "thread-1" }, ]); yield* Deferred.succeed(releaseReadyDispatch, undefined); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 2)); + expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({ + threadId, + input: "send after compact", + }); + }), + ); + + effectIt.effect("sends turns queued during compaction in FIFO order", () => + Effect.gen(function* () { + const releaseCompaction = yield* Deferred.make(); + const harness = yield* Effect.promise(() => + createHarness({ compactThreadEffect: () => Deferred.await(releaseCompaction) }), + ); + 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-fifo-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-fifo-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("fifo-compact", "/compact", "2026-01-01T00:00:01.000Z"); + yield* Effect.promise(() => waitFor(() => harness.compactThread.mock.calls.length === 1)); + yield* dispatchTurn("fifo-first", "first queued", "2026-01-01T00:00:02.000Z"); + yield* dispatchTurn("fifo-second", "second queued", "2026-01-01T00:00:03.000Z"); yield* Effect.promise(() => waitFor(async () => { const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); - return thread?.session?.status === "ready"; + return thread?.messages.some((message) => message.text === "second queued") === true; }), ); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + + yield* Deferred.succeed(releaseCompaction, undefined); + yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 3)); + expect(harness.sendTurn.mock.calls.slice(1).map(([request]) => request.input)).toEqual([ + "first queued", + "second queued", + ]); + 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 a queued turn when the thread stops during compaction", () => + Effect.gen(function* () { + const releaseCompaction = yield* Deferred.make(); + const harness = yield* Effect.promise(() => + createHarness({ compactThreadEffect: () => Deferred.await(releaseCompaction) }), + ); + 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)); + const queuedMessageId = asMessageId("user-message-stopped-queued"); + yield* dispatchTurn("stopped-queued", "never send", "2026-01-01T00:00:02.000Z"); + yield* harness.engine.dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("cmd-stop-with-queued-turn"), + threadId, + createdAt: "2026-01-01T00:00:03.000Z", + }); + yield* Effect.promise(() => + waitFor(async () => { + const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); + return thread?.session?.status === "stopped"; + }), + ); + + yield* Deferred.succeed(releaseCompaction, undefined); + 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" && + (activity.payload as { readonly requestId?: unknown } | null)?.requestId === + queuedMessageId, + ) === true + ); + }), + ); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); }), ); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 1beab8ed22bb..86481f151fdd 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -345,6 +345,16 @@ const make = Effect.gen(function* () { const threadModelSelections = new Map(); const compactingThreadIds = new Set(); const stoppingThreadIds = new Set(); + type TurnStartRequestedEvent = Extract< + ProviderIntentEvent, + { type: "thread.turn-start-requested" } + >; + type QueuedTurnStart = { + readonly event: TurnStartRequestedEvent; + readonly message: ThreadTitleMessage; + }; + // Compaction fibers replay these events in arrival order after restoring the session. + const queuedTurnStarts = new Map>(); const appendProviderFailureActivity = (input: { readonly threadId: ThreadId; @@ -1169,14 +1179,87 @@ const make = Effect.gen(function* () { processThreadTitleRegenerationSafely, ); - const processTurnStartRequested = Effect.fn("processTurnStartRequested")(function* ( - event: Extract, + const appendTurnStartFailure = ( + event: TurnStartRequestedEvent, + summary: string, + detail: string, + ) => + appendProviderFailureActivity({ + threadId: event.payload.threadId, + kind: "provider.turn.start.failed", + summary, + detail, + turnId: null, + createdAt: event.payload.createdAt, + requestId: event.payload.messageId, + }); + + const handleTurnStartFailure = (event: TurnStartRequestedEvent, cause: Cause.Cause) => { + if (Cause.hasInterruptsOnly(cause)) { + return Effect.void; + } + const detail = formatFailureDetail(cause); + return setThreadSessionErrorOnTurnStartFailure({ + threadId: event.payload.threadId, + detail, + createdAt: event.payload.createdAt, + }).pipe( + Effect.flatMap(() => appendTurnStartFailure(event, "Provider turn start failed", detail)), + Effect.asVoid, + ); + }; + + const recoverTurnStartFailure = (event: TurnStartRequestedEvent, cause: Cause.Cause) => + handleTurnStartFailure(event, cause).pipe( + Effect.catchCause((recoveryCause) => + Effect.logWarning("provider command reactor failed to recover turn start failure", { + eventType: event.type, + threadId: event.payload.threadId, + cause: Cause.pretty(recoveryCause), + originalCause: Cause.pretty(cause), + }), + ), + ); + + const sendTurnForEvent = Effect.fn("sendTurnForEvent")(function* ( + event: TurnStartRequestedEvent, + message: ThreadTitleMessage, + forkSend = false, ) { - const key = turnStartKeyForEvent(event); - if (yield* hasHandledTurnStartRecently(key)) { + 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(event, cause).pipe(Effect.as(Option.none())), + ), + ); + + if (Option.isNone(sendTurnRequest)) { return; } + const sendTurn = providerService.sendTurn(sendTurnRequest.value).pipe( + Effect.asVoid, + Effect.catchCause((cause) => recoverTurnStartFailure(event, cause)), + ); + if (forkSend) { + yield* sendTurn.pipe(Effect.forkScoped); + return; + } + yield* sendTurn; + }); + + const processTurnStartRequestedBody = Effect.fn("processTurnStartRequestedBody")(function* ( + event: TurnStartRequestedEvent, + ) { const thread = yield* resolveThread(event.payload.threadId); if (!thread) { return; @@ -1194,44 +1277,6 @@ const make = Effect.gen(function* () { }); return; } - const appendTurnStartFailure = (summary: string, detail: string) => - appendProviderFailureActivity({ - threadId: event.payload.threadId, - kind: "provider.turn.start.failed", - summary, - detail, - turnId: null, - createdAt: event.payload.createdAt, - requestId: event.payload.messageId, - }); - - const handleTurnStartFailure = (cause: Cause.Cause) => { - if (Cause.hasInterruptsOnly(cause)) { - return Effect.void; - } - const detail = formatFailureDetail(cause); - return setThreadSessionErrorOnTurnStartFailure({ - threadId: event.payload.threadId, - detail, - createdAt: event.payload.createdAt, - }).pipe( - Effect.flatMap(() => appendTurnStartFailure("Provider turn start failed", detail)), - Effect.asVoid, - ); - }; - - const recoverTurnStartFailure = (cause: Cause.Cause) => - handleTurnStartFailure(cause).pipe( - Effect.catchCause((recoveryCause) => - Effect.logWarning("provider command reactor failed to recover turn start failure", { - eventType: event.type, - threadId: event.payload.threadId, - cause: Cause.pretty(recoveryCause), - originalCause: Cause.pretty(cause), - }), - ), - ); - const authCommandHandled = yield* Effect.gen(function* () { // Native account commands belong to the thread's existing provider session. const instanceId = @@ -1278,7 +1323,9 @@ const make = Effect.gen(function* () { createdAt: event.payload.createdAt, }); return true; - }).pipe(Effect.catchCause((cause) => recoverTurnStartFailure(cause).pipe(Effect.as(true)))); + }).pipe( + Effect.catchCause((cause) => recoverTurnStartFailure(event, cause).pipe(Effect.as(true))), + ); if (authCommandHandled) { return; } @@ -1330,11 +1377,11 @@ const make = Effect.gen(function* () { detail, createdAt: event.payload.createdAt, }).pipe( - Effect.flatMap(() => appendTurnStartFailure("Context compaction failed", detail)), + Effect.flatMap(() => appendTurnStartFailure(event, "Context compaction failed", detail)), Effect.asVoid, ); } - return appendTurnStartFailure("Context compaction failed", detail).pipe( + return appendTurnStartFailure(event, "Context compaction failed", detail).pipe( Effect.ensuring( restoreCompaction(event.payload.threadId).pipe( Effect.catchCause((restoreCause) => @@ -1362,6 +1409,7 @@ const make = Effect.gen(function* () { if (isCompactCommand) { if (nonCompactUserMessageCount === 0) { return yield* appendTurnStartFailure( + event, "Context compaction failed", "Context compaction requires an existing conversation.", ); @@ -1373,6 +1421,7 @@ const make = Effect.gen(function* () { latestThread?.session?.status === "running" ) { yield* appendTurnStartFailure( + event, "Context compaction failed", "Context compaction is unavailable while a provider turn is running.", ); @@ -1400,37 +1449,70 @@ const make = Effect.gen(function* () { Effect.andThen(restoreCompaction(event.payload.threadId, true)), Effect.catchCause(recoverCompactionFailure), Effect.ensuring(Effect.sync(() => void compactingThreadIds.delete(event.payload.threadId))), + Effect.andThen(drainQueuedTurnStarts(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 queued = queuedTurnStarts.get(event.payload.threadId); + if (queued) { + queued.push({ event, message }); + } else { + queuedTurnStarts.set(event.payload.threadId, [{ event, message }]); + } + return; } - 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()))), + yield* sendTurnForEvent(event, message, true); + }); + + const failQueuedTurnStarts = Effect.fnUntraced(function* ( + events: ReadonlyArray, + detail: string, + ) { + yield* Effect.forEach( + events, + ({ event }) => appendTurnStartFailure(event, "Provider turn start failed", detail), + { discard: true }, ); + }); - if (Option.isNone(sendTurnRequest)) { + const drainQueuedTurnStarts = Effect.fnUntraced(function* (threadId: ThreadId) { + const queued = queuedTurnStarts.get(threadId); + if (!queued || queued.length === 0) { return; } + queuedTurnStarts.delete(threadId); - yield* providerService - .sendTurn(sendTurnRequest.value) - .pipe(Effect.asVoid, Effect.catchCause(recoverTurnStartFailure), Effect.forkScoped); + const thread = yield* resolveThread(threadId); + if (stoppingThreadIds.has(threadId) || thread?.session?.status === "stopped") { + return yield* failQueuedTurnStarts( + queued, + "The provider session stopped before context compaction finished.", + ); + } + if (!thread?.session || thread.session.status === "error") { + return yield* failQueuedTurnStarts( + queued, + thread?.session?.lastError ?? + "The provider session could not be restored after context compaction.", + ); + } + + yield* Effect.forEach(queued, ({ event, message }) => sendTurnForEvent(event, message), { + concurrency: 1, + discard: true, + }); + }); + + const processTurnStartRequested = Effect.fn("processTurnStartRequested")(function* ( + event: TurnStartRequestedEvent, + ) { + const key = turnStartKeyForEvent(event); + if (yield* hasHandledTurnStartRecently(key)) { + return; + } + yield* processTurnStartRequestedBody(event); }); const processTurnInterruptRequested = Effect.fn("processTurnInterruptRequested")(function* ( diff --git a/docs/user/composer.md b/docs/user/composer.md index fce133cd27d8..ddf5fb4dc77a 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -213,7 +213,7 @@ such as System, Personal, Project, or App. On mobile, these menus are available on the **New task** screen before you start a thread. They use the skills and commands from the selected environment and provider. -In a thread with prior conversation context, send `/compact` to reduce context usage. Web and desktop also offer this action from the context meter, and the work log records token counts when the provider reports them. +In a thread with prior conversation context, send `/compact` to reduce context usage. Messages sent while compaction runs are held and sent when it finishes. Web and desktop also offer this action from the context meter, and the work log records token counts when the provider reports them. By default, the `/` menu includes skills. To keep this menu command-only, turn off **Show skills in slash menu** in **Settings → General**. Skill results use the `/skill:Skill Name` label and add the