From 2518cd61e08fc49c6ac3b80286478cc440152a6d Mon Sep 17 00:00:00 2001 From: maria Date: Fri, 11 Sep 2026 01:46:33 -0300 Subject: [PATCH 1/7] fix(web): preserve drafts when compacting context (#11103) (cherry picked from commit ef6fa118749572da1905eca5d0543d1e98f0b1b2) --- apps/web/src/components/ChatView.tsx | 88 +++++++++++++++++-- apps/web/src/components/chat/ChatComposer.tsx | 36 ++------ 2 files changed, 85 insertions(+), 39 deletions(-) diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 11c840952..01f9bbe3c 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -6393,6 +6393,7 @@ export default function ChatView(props: ChatViewProps) { (message) => message.role === "user" && !isCompactCommandMessage(message), ) ?? false; const compactThreadUnavailable = + rollbackActive || !activeThread || !activeThreadHasCompactableConversation || !activeProject || @@ -6406,15 +6407,13 @@ export default function ChatView(props: ChatViewProps) { pendingApprovals.length > 0 || pendingUserInputs.length > 0 || showPlanFollowUpPrompt; - const compactDisabled = compactThreadUnavailable || composerHasUnsentContent; + const compactDisabled = compactThreadUnavailable; const compactDisabledReason = compactDisabled - ? composerHasUnsentContent - ? "Send or clear your draft before compacting" - : !activeProject - ? "Choose a project before compacting" - : !manualCompactionProviderAvailable - ? "Compaction is unavailable for this provider" - : "Compacting is unavailable right now" + ? !activeProject + ? "Choose a project before compacting" + : !manualCompactionProviderAvailable + ? "Compaction is unavailable for this provider" + : "Compacting is unavailable right now" : null; const resumeCompactionBannerItem = useMemo(() => { if ( @@ -7028,6 +7027,78 @@ export default function ChatView(props: ChatViewProps) { [activeThread, environmentId, recoverThreadRollback, rollbackRecoveryPending, setThreadError], ); + const onCompactContext = async () => { + if (compactDisabled || !activeThread || !clientSettingsHydrated || sendInFlightRef.current) { + return; + } + const context = composerRef.current?.getSendContext(); + if (!context?.providerAvailable || context.selectedModelSelection === null) return; + + // Compaction is a standalone command; the draft and its attachments stay local. + const threadId = activeThread.id; + const messageId = newMessageId(); + const createdAt = new Date().toISOString(); + sendInFlightRef.current = true; + beginLocalDispatch(); + setThreadError(threadId, null); + setOptimisticUserMessages((messages) => [ + ...messages, + { + id: messageId, + role: "user", + text: "/compact", + turnId: null, + createdAt, + updatedAt: createdAt, + streaming: false, + }, + ]); + scrollToEnd(); + try { + const settingsResult = await persistThreadSettingsForNextTurn({ + threadId, + createdAt, + modelSelection: context.selectedModelSelection, + ...(localCheckoutBranchMismatch + ? { branch: localCheckoutBranchMismatch.currentBranch } + : {}), + runtimeMode, + interactionMode: context.interactionMode, + }); + const result = + settingsResult._tag === "Failure" + ? settingsResult + : await startThreadTurn({ + environmentId, + input: { + threadId, + message: { messageId, role: "user", text: "/compact", attachments: [] }, + modelSelection: context.selectedModelSelection, + runtimeMode, + interactionMode: context.interactionMode, + createdAt, + }, + }); + if (result._tag === "Failure") { + setOptimisticUserMessages((messages) => + messages.filter((message) => message.id !== messageId), + ); + resetLocalDispatch(); + if (!isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + setThreadError( + threadId, + error instanceof Error ? error.message : "Failed to compact context.", + ); + } + } else { + clearUsageLimitsFor(routeThreadKey); + } + } finally { + sendInFlightRef.current = false; + } + }; + const onSend = async ( e?: { preventDefault: () => void }, submissionIntent: ComposerSubmissionIntent = "foreground", @@ -9947,6 +10018,7 @@ export default function ChatView(props: ChatViewProps) { onPageScrollKeyDown={onComposerPageScrollKeyDown} onPageScrollKeyUp={onComposerPageScrollKeyUp} onPageScrollRelease={onComposerPageScrollRelease} + onCompactContext={onCompactContext} onSend={onSend} onQueueFollowUp={() => onSend(undefined, "foreground", undefined, "follow-up") diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 32db7c561..b3eb707f2 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -1501,6 +1501,7 @@ export interface ChatComposerProps { isStartingProviderConflictThread: boolean; // Callbacks + onCompactContext: () => void; onSend: (e?: { preventDefault: () => void }, intent?: ComposerSubmissionIntent) => void; onQueueFollowUp: () => void; onClearSessionInputQueue: () => Promise; @@ -1638,6 +1639,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) onPageScrollKeyDown, onPageScrollKeyUp, onPageScrollRelease, + onCompactContext, onSend, onQueueFollowUp, onClearSessionInputQueue, @@ -2841,7 +2843,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) /** * Count of pasted images still being compressed, per thread. Reserved * against the attachment limit so concurrent pastes can't overshoot it, - * and checked before sending or compacting so an image cannot move into + * and checked before sending so an image cannot move into * the next draft. */ const pendingImageCompressionsRef = useRef>(new Map()); @@ -3911,7 +3913,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) // session's, so re-check the selected entry rather than trusting the // thread-level gate alone. !compactCommandAvailable || - composerSendState.hasSendableContent || activePendingApproval !== null || pendingUserInputs.length > 0 || phase === "running" || @@ -3921,46 +3922,19 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ) { return; } - // The compact buttons cannot see the compression counter (it lives in - // a ref), so they render enabled during a paste; toast instead of - // silently ignoring the click. - if ((pendingImageCompressionsRef.current.get(attachmentTargetKey) ?? 0) > 0) { - toastManager.add({ - type: "info", - title: "Still compressing a pasted image.", - description: "Compact again once its thumbnail appears.", - }); - return; - } - - promptRef.current = "/compact"; - setComposerDraftPrompt(composerDraftTarget, "/compact"); - submitComposer(); - // A blocked dispatch (busy send ref, provider preflight rejection) - // would leave the injected "/compact" behind as if the user typed it. - // Clearing here is safe even when the send did dispatch: the send - // snapshots its prompt synchronously and clears the draft itself. - if (promptRef.current === "/compact") { - promptRef.current = ""; - setComposerDraftPrompt(composerDraftTarget, ""); - } + onCompactContext(); }, [ activePendingApproval, activeThreadId, - attachmentTargetKey, compactCommandAvailable, compactDisabled, - composerDraftTarget, - composerSendState.hasSendableContent, isConnecting, isSendBusy, noProviderAvailable, + onCompactContext, providerTurnUnavailable, pendingUserInputs.length, phase, - promptRef, - setComposerDraftPrompt, - submitComposer, ]); const expandMobileComposer = useCallback(() => { if (composerBlurFrameRef.current !== null) { From 6e5c7c20f3a227e063a25a7e8bf68018a2e5d2af Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 12 Sep 2026 00:47:34 -0600 Subject: [PATCH 2/7] fix(server): queue compaction messages through durable turn admission Adapt upstream compaction queueing to the pure decider and exact provider admission. Persist FIFO and in-flight ownership, settle compaction through correlated commands, and expose explicit cancellation receipts on stop, failure, and restart. Preserve composer drafts and acknowledge queued messages without waiting for provider startup. Upstream-Commit: 57aee3e19f1910f3323f06384bb2a1d02b79e369 Completes Pylon API adaptation of ef6fa118749572da1905eca5d0543d1e98f0b1b2. --- .../Layers/OrchestrationEngine.ts | 6 + .../Layers/ProjectionPipeline.ts | 1 + .../Layers/ProjectionSnapshotQuery.ts | 9 + .../Layers/ProviderCommandReactor.test.ts | 116 ++++- .../Layers/ProviderCommandReactor.ts | 245 +++++++--- .../orchestration/decider.compaction.test.ts | 235 ++++++++++ apps/server/src/orchestration/decider.ts | 443 ++++++++++++++++-- .../Layers/ProjectionThreadSessions.ts | 6 + apps/server/src/persistence/Migrations.ts | 2 + .../060_ProjectionThreadCompactionQueue.ts | 7 + .../Services/ProjectionThreadSessions.ts | 2 + .../web/src/components/ChatView.logic.test.ts | 25 + apps/web/src/components/ChatView.logic.ts | 12 + apps/web/src/components/ChatView.tsx | 9 +- docs/user/context-compaction.md | 7 + packages/contracts/src/orchestration.ts | 57 +++ 16 files changed, 1046 insertions(+), 136 deletions(-) create mode 100644 apps/server/src/orchestration/decider.compaction.test.ts create mode 100644 apps/server/src/persistence/Migrations/060_ProjectionThreadCompactionQueue.ts create mode 100644 docs/user/context-compaction.md diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index c29200f0d..45c2b8b1c 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -154,6 +154,8 @@ const makeOrchestrationEngine = Effect.gen(function* () { const safeForOwnedThread = command.type === "thread.rollback.status.set" || command.type === "thread.revert.complete" || + command.type === "thread.compaction.complete" || + command.type === "thread.compaction.queue.sent" || (command.type === "thread.session.set" && ["idle", "ready", "interrupted", "stopped", "error"].includes(command.session.status)); const owned = active.find((record) => record.threadId === command.threadId); @@ -165,6 +167,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { } if ( command.type !== "thread.turn.start" && + command.type !== "thread.compaction.queue.resume" && command.type !== "thread.input-queue.follow-up" && command.type !== "thread.approval.respond" && command.type !== "thread.user-input.respond" && @@ -485,6 +488,9 @@ const makeOrchestrationEngine = Effect.gen(function* () { envelope.command.type === "thread.meta.update" || envelope.command.type === "thread.runtime-mode.set" || envelope.command.type === "thread.interaction-mode.set" || + envelope.command.type === "thread.compaction.complete" || + envelope.command.type === "thread.compaction.queue.resume" || + envelope.command.type === "thread.compaction.queue.sent" || envelope.command.type === "thread.turn.admission.accept" || envelope.command.type === "thread.turn.admission.fail" || envelope.command.type === "thread.session.bind-pending" || diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index b9b5c7b2f..001b25add 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1424,6 +1424,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti startedAt: incoming.startedAt ?? null, sessionIncarnationId: incoming.sessionIncarnationId ?? null, harnessRefinementStatus: incoming.harnessRefinementStatus ?? null, + compactionQueue: incoming.compactionQueue ?? null, pendingTurnRequestId: preserveHistoricalPending ? preserved.pendingTurnRequestId : (incoming.pendingTurnRequestId ?? null), diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 5e1028e40..e892414ee 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -1,4 +1,5 @@ import { + OrchestrationCompactionQueue, AgentSessionImportSource, ApprovalRequestId, ChatAttachment, @@ -148,6 +149,7 @@ const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( ); const ProjectionThreadSessionDbRowSchema = Schema.Struct({ ...ProjectionThreadSession.fields, + compactionQueue: Schema.NullOr(Schema.fromJsonString(OrchestrationCompactionQueue)), restored: Schema.Number, pendingTurnRequestAmbiguous: Schema.Number, }); @@ -404,6 +406,7 @@ function mapSessionRow( ): OrchestrationSession { return { threadId: row.threadId, + ...(row.compactionQueue ? { compactionQueue: row.compactionQueue } : {}), status: row.status, providerName: row.providerName, ...(row.providerInstanceId !== null ? { providerInstanceId: row.providerInstanceId } : {}), @@ -962,6 +965,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { started_at AS "startedAt", session_incarnation_id AS "sessionIncarnationId", harness_refinement_status AS "harnessRefinementStatus", + compaction_queue_json AS "compactionQueue", pending_turn_request_id AS "pendingTurnRequestId", pending_turn_request_ambiguous AS "pendingTurnRequestAmbiguous", pending_turn_message_id AS "pendingTurnMessageId", @@ -1000,6 +1004,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { sessions.started_at AS "startedAt", sessions.session_incarnation_id AS "sessionIncarnationId", sessions.harness_refinement_status AS "harnessRefinementStatus", + sessions.compaction_queue_json AS "compactionQueue", sessions.pending_turn_request_id AS "pendingTurnRequestId", sessions.pending_turn_request_ambiguous AS "pendingTurnRequestAmbiguous", sessions.pending_turn_message_id AS "pendingTurnMessageId", @@ -1042,6 +1047,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { sessions.started_at AS "startedAt", sessions.session_incarnation_id AS "sessionIncarnationId", sessions.harness_refinement_status AS "harnessRefinementStatus", + sessions.compaction_queue_json AS "compactionQueue", sessions.pending_turn_request_id AS "pendingTurnRequestId", sessions.pending_turn_request_ambiguous AS "pendingTurnRequestAmbiguous", sessions.pending_turn_message_id AS "pendingTurnMessageId", @@ -1503,6 +1509,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { sessions.started_at AS "startedAt", sessions.session_incarnation_id AS "sessionIncarnationId", sessions.harness_refinement_status AS "harnessRefinementStatus", + sessions.compaction_queue_json AS "compactionQueue", sessions.pending_turn_request_id AS "pendingTurnRequestId", sessions.pending_turn_request_ambiguous AS "pendingTurnRequestAmbiguous", sessions.pending_turn_message_id AS "pendingTurnMessageId", @@ -1862,6 +1869,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { started_at AS "startedAt", session_incarnation_id AS "sessionIncarnationId", harness_refinement_status AS "harnessRefinementStatus", + compaction_queue_json AS "compactionQueue", pending_turn_request_id AS "pendingTurnRequestId", pending_turn_request_ambiguous AS "pendingTurnRequestAmbiguous", pending_turn_message_id AS "pendingTurnMessageId", @@ -2576,6 +2584,7 @@ pending_approval_requests AS ( sessionsByThread.set(row.threadId, { threadId: row.threadId, status: row.status, + ...(row.compactionQueue ? { compactionQueue: row.compactionQueue } : {}), providerName: row.providerName, ...(row.providerInstanceId !== null ? { providerInstanceId: row.providerInstanceId } diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 8e36a053e..6cd37660e 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -575,7 +575,8 @@ describe("ProviderCommandReactor", () => { } } return ( - command.type === "thread.session.set" && command.session.status === "ready" + (command.type === "thread.session.set" && command.session.status === "ready") || + (command.type === "thread.compaction.complete" && command.success) ? (input?.beforeReadySessionDispatch?.() ?? Effect.void) : Effect.void ).pipe(Effect.andThen(engine.dispatch(command))); @@ -1108,13 +1109,20 @@ describe("ProviderCommandReactor", () => { }), ); - effectIt.effect("keeps turns blocked until compaction restores the session", () => + effectIt.effect("queues turns until compaction restores the exact session reservation", () => Effect.gen(function* () { + const queuedSent = yield* Deferred.make(); + let sendCount = 0; const readyDispatchStarted = yield* Deferred.make(); const releaseReadyDispatch = yield* Deferred.make(); let blockReadyDispatch = false; const harness = yield* Effect.promise(() => createHarness({ + sendTurnEffect: () => + Effect.gen(function* () { + if (++sendCount === 3) yield* Deferred.succeed(queuedSent, undefined); + return { threadId: ThreadId.make("thread-1"), turnId: asTurnId("turn-1") }; + }), beforeReadySessionDispatch: () => blockReadyDispatch ? Deferred.succeed(readyDispatchStarted, undefined).pipe( @@ -1125,7 +1133,12 @@ describe("ProviderCommandReactor", () => { ); const threadId = ThreadId.make("thread-1"); const now = "2026-01-01T00:00:00.000Z"; - const dispatchTurn = (id: string, text: string, createdAt: string) => + const dispatchTurn = ( + id: string, + text: string, + createdAt: string, + runtimeMode: "approval-required" | "full-access" = "approval-required", + ) => harness.engine.dispatch({ type: "thread.turn.start", commandId: CommandId.make(`cmd-${id}`), @@ -1137,7 +1150,7 @@ describe("ProviderCommandReactor", () => { attachments: [], }, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, - runtimeMode: "approval-required", + runtimeMode, createdAt, }); @@ -1164,28 +1177,99 @@ describe("ProviderCommandReactor", () => { yield* dispatchTurn("blocked-compact", "/compact", "2026-01-01T00:00:01.000Z"); yield* Deferred.await(readyDispatchStarted); - // Upstream blocks this in the reactor and records a failure activity. - // Pylon's decider owns turn exclusivity: the compaction still holds the - // thread's pending admission, so the command is refused outright and - // never reaches the reactor's own `compactingThreadIds` guard. - const blockedTurn = yield* dispatchTurn( + yield* dispatchTurn( "during-compact-recovery", - "too soon", + "queued after compact", "2026-01-01T00:00:02.000Z", - ).pipe(Effect.result); - expect(blockedTurn._tag).toBe("Failure"); + ); + yield* dispatchTurn( + "during-compact-second", + "second queued message", + "2026-01-01T00:00:03.000Z", + "full-access", + ); + const queuedThread = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === threadId, + ); + expect(queuedThread?.session?.pendingTurnRequestId).toBe("cmd-blocked-compact"); + expect(queuedThread?.session?.compactionQueue?.queued.map((entry) => entry.text)).toEqual([ + "queued after compact", + "second queued message", + ]); expect(harness.sendTurn).toHaveBeenCalledTimes(1); expect(yield* Effect.promise(() => harness.readPendingTurnStarts())).toEqual([ { threadId: "thread-1" }, ]); yield* Deferred.succeed(releaseReadyDispatch, undefined); - yield* Effect.promise(() => - waitFor(async () => { - const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); - return thread?.session?.status === "ready"; + yield* Deferred.await(queuedSent); + yield* Effect.promise(() => harness.drain()); + expect(harness.sendTurn).toHaveBeenCalledTimes(3); + expect(harness.sendTurn).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ input: "queued after compact" }), + ); + expect(harness.sendTurn).toHaveBeenLastCalledWith( + expect.objectContaining({ input: "second queued message" }), + ); + const resumed = (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === threadId, + ); + expect(resumed?.runtimeMode).toBe("full-access"); + expect( + resumed?.messages.filter( + (message) => message.id === "user-message-during-compact-recovery", + ), + ).toHaveLength(1); + }), + ); + + effectIt.effect("reconciles persisted compaction messages explicitly after restart", () => + Effect.gen(function* () { + const harness = yield* Effect.promise(() => + createHarness({ + beforeReactorStart: Effect.gen(function* () { + const engine = yield* OrchestrationEngineService; + for (const [id, text] of [ + ["restarted-compact", "/compact"], + ["restarted-one", "first"], + ["restarted-two", "second"], + ]) { + yield* engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(id!), + threadId: ThreadId.make("thread-1"), + message: { + messageId: MessageId.make(`message-${id}`), + role: "user", + text: text!, + attachments: [], + }, + runtimeMode: "approval-required", + interactionMode: "default", + createdAt: "2026-01-01T00:00:00.000Z", + }); + } + }), }), ); + yield* Effect.promise(() => harness.drain()); + const thread = (yield* Effect.promise(() => harness.readModel())).threads[0]; + expect(thread?.session?.compactionQueue).toBeUndefined(); + expect(thread?.session?.pendingTurnRequestId).toBeUndefined(); + expect( + thread?.activities + .filter((activity) => activity.kind === "provider.turn.start.failed") + .map((activity) => + activity.payload && + typeof activity.payload === "object" && + "requestId" in activity.payload + ? activity.payload.requestId + : undefined, + ), + ).toEqual(["message-restarted-one", "message-restarted-two"]); + expect(harness.compactThread).not.toHaveBeenCalled(); + expect(harness.sendTurn).not.toHaveBeenCalled(); }), ); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index e51e83b84..dd591e71a 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -79,6 +79,7 @@ type ProviderIntentEvent = Extract< type: | "thread.meta-updated" | "thread.runtime-mode-set" + | "thread.session-set" | "thread.turn-start-requested" | "thread.input-queue-follow-up-requested" | "thread.turn-interrupt-requested" @@ -642,65 +643,66 @@ const make = Effect.gen(function* () { .pipe(Effect.map(Option.getOrUndefined)); }); - const setThreadSession = (input: { - readonly threadId: ThreadId; - readonly session: OrchestrationSession; - readonly createdAt: string; - }) => - serverCommandId("provider-session-set").pipe( - Effect.flatMap((commandId) => - orchestrationEngine.dispatch({ - type: "thread.session.set", - commandId, - threadId: input.threadId, - session: input.session, - createdAt: input.createdAt, - }), - ), - Effect.asVoid, - ); + const completeCompaction = Effect.fnUntraced(function* ( + threadId: ThreadId, + requestId: CommandId, + success: boolean, + detail?: string, + expectedSession?: ProviderSession, + reconcileInFlight = false, + ) { + yield* orchestrationEngine.dispatch({ + type: "thread.compaction.complete", + commandId: yield* serverCommandId("compaction-complete"), + threadId, + requestId, + success, + ...(detail ? { detail } : {}), + reconcileInFlight, + ...(expectedSession + ? { + expectedSessionIncarnationId: expectedSession.sessionIncarnationId ?? null, + expectedProviderInstanceId: expectedSession.providerInstanceId ?? null, + } + : {}), + createdAt: DateTime.formatIso(yield* DateTime.now), + }); + }); - const restoreCompaction = Effect.fnUntraced(function* (threadId: ThreadId, fromRunning = false) { - if (stoppingThreadIds.has(threadId)) { - compactingThreadIds.delete(threadId); - return; - } - const thread = yield* resolveThreadShell(threadId); - if (!thread?.session) return; - if ( - thread.session.status !== "starting" && - thread.session.status !== "ready" && - (!fromRunning || thread.session.status !== "running") - ) - return; - const completedAt = DateTime.formatIso(yield* DateTime.now); - if (stoppingThreadIds.has(threadId)) { - compactingThreadIds.delete(threadId); - return; - } - yield* setThreadSession({ + const markCompactionQueueSent = Effect.fnUntraced(function* ( + threadId: ThreadId, + requestId: CommandId, + sentRequestId: CommandId, + ) { + yield* orchestrationEngine.dispatch({ + type: "thread.compaction.queue.sent", + commandId: yield* serverCommandId("compaction-queue-sent"), threadId, - session: { - ...thread.session, - status: "ready", - activeTurnId: null, - // Pylon's decider reserves a turn admission for every `thread.turn.start`, - // including the `/compact` one, but compaction never becomes a provider - // turn that could accept it. Retire it here or the spread above carries - // it forward and the decider rejects every later turn on this thread. - pendingTurnRequestId: undefined, - pendingTurnMessageId: undefined, - pendingTurnRequestedAt: undefined, - pendingTurnDeadlineAt: undefined, - pendingTurnSessionId: undefined, - activeTurnRequestId: undefined, - lastError: null, - updatedAt: completedAt, - }, - createdAt: completedAt, + requestId, + sentRequestId, + createdAt: DateTime.formatIso(yield* DateTime.now), }); }); + const drainCompactionQueue = Effect.fnUntraced(function* (threadId: ThreadId) { + const session = (yield* resolveThreadShell(threadId))?.session; + const queue = session?.compactionQueue; + if (!queue || queue.phase !== "draining" || queue.inFlightRequestId !== undefined) return; + yield* orchestrationEngine + .dispatch({ + type: "thread.compaction.queue.resume", + commandId: yield* serverCommandId("compaction-queue-resume"), + threadId, + requestId: queue.requestId, + createdAt: DateTime.formatIso(yield* DateTime.now), + }) + .pipe( + Effect.catchCause((cause) => + completeCompaction(threadId, queue.requestId, false, formatFailureDetail(cause)), + ), + ); + }); + /** * Marks a stop in flight so a compaction finishing underneath it cannot flip * the session back to ready. Pylon's decider records `stopped` in the same @@ -1680,6 +1682,13 @@ const make = Effect.gen(function* () { if ((yield* Clock.currentTimeMillis) >= admissionDeadlineMs) { yield* failAdmission(PROVIDER_TURN_ADMISSION_TIMEOUT_DETAIL); + if (isCompactCommandMessage(message)) + yield* completeCompaction( + event.payload.threadId, + requestId, + false, + PROVIDER_TURN_ADMISSION_TIMEOUT_DETAIL, + ); return; } @@ -1688,12 +1697,24 @@ const make = Effect.gen(function* () { // compaction or sends the adapter's own slash command. if (isCompactCommandMessage(message)) { if (!hasOtherUserMessages) { - return yield* appendTurnStartFailure( + yield* appendTurnStartFailure( "Context compaction failed", "Context compaction requires an existing conversation.", ); + yield* completeCompaction( + event.payload.threadId, + requestId, + false, + "Context compaction requires an existing conversation.", + ); + return; } const latestSession = (yield* resolveThreadShell(event.payload.threadId))?.session; + if ( + latestSession?.compactionQueue?.requestId !== requestId || + latestSession.pendingTurnRequestId !== requestId + ) + return; // Pylon admits the turn before the reactor observes its intent event, so // this thread already reads as "starting" for the compaction's own // request. Upstream's status check would reject every compaction here; @@ -1708,6 +1729,12 @@ const make = Effect.gen(function* () { "Context compaction failed", "Context compaction is unavailable while a provider turn is running.", ); + yield* completeCompaction( + event.payload.threadId, + requestId, + false, + "Context compaction is unavailable while a provider turn is running.", + ); return; } const handleCompactionFailure = (cause: Cause.Cause) => { @@ -1717,7 +1744,7 @@ const make = Effect.gen(function* () { Effect.ensuring( // A no-op unless the session actually reached a restorable state, // so this covers both a failed ensure and a failed compaction. - restoreCompaction(event.payload.threadId).pipe( + completeCompaction(event.payload.threadId, requestId, false, detail).pipe( Effect.catchCause((restoreCause) => Effect.logWarning("failed to restore provider session after compaction failure", { threadId: event.payload.threadId, @@ -1731,18 +1758,27 @@ const make = Effect.gen(function* () { }; compactingThreadIds.add(event.payload.threadId); yield* Effect.gen(function* () { - // Deliberately no pending turn admission: `restoreCompaction` spreads - // the existing session forward, so an admission recorded here would - // survive it and the decider would reject every later turn on the - // thread. `compactingThreadIds` is what serialises compaction instead. - yield* ensureSessionForThread(event.payload.threadId, event.payload.createdAt, { - ...(event.payload.modelSelection !== undefined - ? { modelSelection: event.payload.modelSelection } - : {}), - runtimeMode: event.payload.runtimeMode, - interactionMode: event.payload.interactionMode, - pendingTurnStart: true, - }); + // Compaction retains its original reservation until the exact completion + // command retires it; provider lifecycle snapshots cannot erase the FIFO. + const compactionSession = yield* ensureSessionForThread( + event.payload.threadId, + event.payload.createdAt, + { + ...(event.payload.modelSelection !== undefined + ? { modelSelection: event.payload.modelSelection } + : {}), + runtimeMode: event.payload.runtimeMode, + interactionMode: event.payload.interactionMode, + pendingTurnStart: true, + }, + ); + if (!compactionSession) return yield* Effect.interrupt; + const current = (yield* resolveThreadShell(event.payload.threadId))?.session; + if ( + current?.compactionQueue?.requestId !== requestId || + current.sessionIncarnationId !== compactionSession.sessionIncarnationId + ) + return yield* Effect.interrupt; if (event.payload.modelSelection !== undefined) { threadModelSelections.set(event.payload.threadId, event.payload.modelSelection); } @@ -1751,8 +1787,14 @@ const make = Effect.gen(function* () { event.payload.modelSelection, event.payload.messageId, ); + return compactionSession; }).pipe( - Effect.andThen(restoreCompaction(event.payload.threadId, true)), + Effect.tap(() => + Effect.sync(() => void compactingThreadIds.delete(event.payload.threadId)), + ), + Effect.flatMap((session) => + completeCompaction(event.payload.threadId, requestId, true, undefined, session), + ), Effect.catchCause((cause) => handleCompactionFailure(cause).pipe( Effect.catchCause((recoveryCause) => @@ -1883,6 +1925,21 @@ const make = Effect.gen(function* () { admissionFibers.set(requestId, admissionFiber); admissionFiberThreads.set(requestId, event.payload.threadId); yield* Fiber.await(admissionFiber).pipe( + Effect.tap(() => + thread.session?.compactionQueue?.inFlightRequestId === requestId + ? markCompactionQueueSent( + event.payload.threadId, + thread.session.compactionQueue.requestId, + requestId, + ).pipe( + Effect.catchCause((cause) => + Effect.logWarning("failed to settle compaction queue admission", { + cause: Cause.pretty(cause), + }), + ), + ) + : Effect.void, + ), Effect.ensuring( Effect.sync(() => { if (admissionFibers.get(requestId) === admissionFiber) { @@ -2352,6 +2409,9 @@ const make = Effect.gen(function* () { eventType: event.type, }); switch (event.type) { + case "thread.session-set": + yield* drainCompactionQueue(event.payload.threadId); + return; case "thread.meta-updated": yield* threadTitleRegenerationWorker.enqueue(event); return; @@ -2566,6 +2626,30 @@ const make = Effect.gen(function* () { const processDomainEventSafely = (event: ProviderIntentEvent) => processDomainEvent(event).pipe( + Effect.ensuring( + Effect.gen(function* () { + if ( + event.type !== "thread.turn-start-requested" || + !event.commandId || + admissionFibers.has(event.commandId) + ) + return; + const queue = (yield* resolveThreadShell(event.payload.threadId))?.session + ?.compactionQueue; + if (queue?.inFlightRequestId === event.commandId) + yield* markCompactionQueueSent( + event.payload.threadId, + queue.requestId, + event.commandId, + ); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("failed to settle unstarted compaction queue entry", { + cause: Cause.pretty(cause), + }), + ), + ), + ), Effect.catchCause((cause) => { if (Cause.hasInterruptsOnly(cause)) { return Effect.interrupt; @@ -2580,6 +2664,13 @@ const make = Effect.gen(function* () { const worker = yield* makeDrainableWorker(processDomainEventSafely); const start: ProviderCommandReactorShape["start"] = Effect.fn("start")(function* () { + const interruptedCompactions = (yield* projectionSnapshotQuery + .getCommandReadModel() + .pipe(Effect.orDie)).threads.flatMap((thread) => + thread.session?.compactionQueue + ? [{ threadId: thread.id, requestId: thread.session.compactionQueue.requestId }] + : [], + ); const interruptedTitleRegenerations = yield* findInterruptedThreadTitleRegenerations().pipe( Effect.catchCause((cause) => { if (Cause.hasInterruptsOnly(cause)) { @@ -2595,6 +2686,7 @@ const make = Effect.gen(function* () { if ( (event.type === "thread.meta-updated" && event.payload.regenerateTitle === true) || event.type === "thread.runtime-mode-set" || + event.type === "thread.session-set" || event.type === "thread.turn-start-requested" || event.type === "thread.input-queue-follow-up-requested" || event.type === "thread.turn-interrupt-requested" || @@ -2629,7 +2721,20 @@ const make = Effect.gen(function* () { ); }), ); - const reconcileStopsThenAdmissions = reconcilePendingSessionStops().pipe( + const reconcileStopsThenAdmissions = Effect.forEach( + interruptedCompactions, + ({ threadId, requestId }) => + completeCompaction( + threadId, + requestId, + false, + "The server restarted before context compaction and its queue completed. Delivery of the in-flight message may have started; check the transcript before resending.", + undefined, + true, + ), + { discard: true }, + ).pipe( + Effect.andThen(reconcilePendingSessionStops()), Effect.catchCause((cause) => { if (Cause.hasInterruptsOnly(cause)) return Effect.interrupt; return Effect.logWarning( diff --git a/apps/server/src/orchestration/decider.compaction.test.ts b/apps/server/src/orchestration/decider.compaction.test.ts new file mode 100644 index 000000000..81c792be6 --- /dev/null +++ b/apps/server/src/orchestration/decider.compaction.test.ts @@ -0,0 +1,235 @@ +import { + CommandId, + MessageId, + ProjectId, + ProviderInstanceId, + RuntimeSessionId, + ThreadId, + type OrchestrationCommand, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import { decideOrchestrationCommand } from "./decider.ts"; +import { projectEvent } from "./projector.ts"; + +const NOW = "2026-09-12T00:00:00.000Z"; +const threadId = ThreadId.make("compaction"); +const rootId = CommandId.make("compact"); +const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }; +const run = Effect.fnUntraced(function* ( + readModel: OrchestrationReadModel, + command: OrchestrationCommand, +) { + const result = yield* decideOrchestrationCommand({ readModel, command }); + const events = Array.isArray(result) ? result : [result]; + let next = readModel; + for (const event of events) + next = yield* projectEvent(next, { ...event, sequence: next.snapshotSequence + 1 }).pipe( + Effect.orDie, + ); + return { readModel: next, events }; +}); +const turn = ( + id: string, + text = id, + runtimeMode: "full-access" | "approval-required" = "full-access", +) => ({ + type: "thread.turn.start" as const, + commandId: CommandId.make(id), + threadId, + message: { + messageId: MessageId.make(`message-${id}`), + role: "user" as const, + text, + attachments: [], + }, + modelSelection, + runtimeMode, + interactionMode: "default" as const, + sourceEpoch: 0, + createdAt: NOW, +}); +const init = Effect.fnUntraced(function* () { + let state: OrchestrationReadModel = { + snapshotSequence: 0, + projects: [], + threads: [], + updatedAt: NOW, + }; + state = (yield* run(state, { + type: "project.create", + commandId: CommandId.make("project"), + projectId: ProjectId.make("project"), + title: "Project", + workspaceRoot: "/tmp/compaction-project", + defaultModelSelection: modelSelection, + createdAt: NOW, + })).readModel; + state = (yield* run(state, { + type: "thread.create", + commandId: CommandId.make("thread"), + threadId, + projectId: ProjectId.make("project"), + title: "Thread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: NOW, + })).readModel; + return (yield* run(state, turn("compact", "/compact"))).readModel; +}); +const complete = (success: boolean) => ({ + type: "thread.compaction.complete" as const, + commandId: CommandId.make("complete"), + threadId, + requestId: rootId, + success, + createdAt: NOW, +}); +const resume = (id: string) => ({ + type: "thread.compaction.queue.resume" as const, + commandId: CommandId.make(id), + threadId, + requestId: rootId, + createdAt: NOW, +}); + +it.layer(NodeServices.layer)("compaction admission FIFO", (it) => { + it.effect( + "accepts queued messages without replacing compact admission and preserves FIFO settings", + () => + Effect.gen(function* () { + let state = yield* init(); + const first = yield* run(state, turn("one", "first", "approval-required")); + expect(first.events.map((event) => event.type)).toEqual([ + "thread.message-sent", + "thread.session-set", + ]); + state = (yield* run(first.readModel, turn("two", "second"))).readModel; + expect(state.threads[0]?.session?.pendingTurnRequestId).toBe(rootId); + state = (yield* run(state, complete(true))).readModel; + const admitted = yield* run(state, resume("resume-one")); + expect( + admitted.events.find((event) => event.type === "thread.turn-start-requested"), + ).toMatchObject({ + payload: { messageId: "message-one", runtimeMode: "approval-required" }, + }); + state = admitted.readModel; + expect(state.threads[0]?.session?.compactionQueue).toMatchObject({ + inFlightRequestId: "resume-one", + inFlightMessageId: "message-one", + queued: [{ messageId: "message-two" }], + }); + expect( + state.threads[0]?.messages.filter((message) => message.id === "message-one"), + ).toHaveLength(1); + expect((yield* run(state, resume("duplicate-resume"))).events).toEqual([]); + state = (yield* run(state, { + type: "thread.compaction.queue.sent", + commandId: CommandId.make("sent"), + threadId, + requestId: rootId, + sentRequestId: CommandId.make("resume-one"), + createdAt: NOW, + })).readModel; + // Send settlement alone cannot bypass the exact pending admission. + expect((yield* run(state, resume("still-pending"))).events).toEqual([]); + }), + ); + + it.effect("cancels every queued bubble on stop and ignores stale completion", () => + Effect.gen(function* () { + let state = yield* init(); + state = (yield* run(state, turn("one"))).readModel; + state = (yield* run(state, turn("two"))).readModel; + const stopped = yield* run(state, { + type: "thread.session.stop", + commandId: CommandId.make("stop"), + threadId, + createdAt: NOW, + }); + expect( + stopped.events.filter((event) => event.type === "thread.activity-appended"), + ).toHaveLength(2); + expect(stopped.readModel.threads[0]?.session).toMatchObject({ status: "stopped" }); + expect(stopped.readModel.threads[0]?.session?.compactionQueue).toBeUndefined(); + expect((yield* run(stopped.readModel, complete(true))).events).toEqual([]); + }), + ); + + it.effect( + "does not let an old provider snapshot erase the queue or completion drain a replacement incarnation", + () => + Effect.gen(function* () { + let state = yield* init(); + const old = state.threads[0]!.session!; + state = (yield* run(state, turn("one"))).readModel; + state = (yield* run(state, { + type: "thread.session.set", + commandId: CommandId.make("snapshot"), + threadId, + session: { ...old, sessionIncarnationId: RuntimeSessionId.make("replacement") }, + createdAt: NOW, + })).readModel; + expect(state.threads[0]?.session?.compactionQueue?.queued).toHaveLength(1); + const canceled = yield* run(state, { + ...complete(true), + expectedSessionIncarnationId: RuntimeSessionId.make("original"), + }); + expect( + canceled.events.filter((event) => event.type === "thread.activity-appended"), + ).toHaveLength(1); + expect(canceled.readModel.threads[0]?.session?.compactionQueue).toBeUndefined(); + expect(canceled.readModel.threads[0]?.session?.sessionIncarnationId).toBe("replacement"); + }), + ); + + it.effect( + "records uncertain in-flight delivery and cancels remaining bubbles during restart reconciliation", + () => + Effect.gen(function* () { + let state = yield* init(); + state = (yield* run(state, turn("one"))).readModel; + state = (yield* run(state, turn("two"))).readModel; + state = (yield* run(state, complete(true))).readModel; + state = (yield* run(state, resume("resume-one"))).readModel; + const canceled = yield* run(state, { + ...complete(false), + reconcileInFlight: true, + detail: "Server restarted; delivery may have started.", + }); + expect( + canceled.events + .filter((event) => event.type === "thread.activity-appended") + .map((event) => event.payload.activity.payload), + ).toEqual([ + { requestId: "message-one", detail: "Server restarted; delivery may have started." }, + { requestId: "message-two", detail: "Server restarted; delivery may have started." }, + ]); + expect(canceled.readModel.threads[0]?.session?.compactionQueue).toBeUndefined(); + }), + ); + + it.effect( + "rejects stale source epochs before queueing and retains ordinary admission exclusion", + () => + Effect.gen(function* () { + const state = yield* init(); + expect( + (yield* run(state, { ...turn("stale"), sourceEpoch: 1 }).pipe(Effect.result))._tag, + ).toBe("Failure"); + const ordinary = { + ...state, + threads: state.threads.map((thread) => ({ + ...thread, + session: { ...thread.session!, compactionQueue: undefined }, + })), + }; + expect((yield* run(ordinary, turn("blocked")).pipe(Effect.result))._tag).toBe("Failure"); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 24b7e845c..f476f96c3 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -7,6 +7,7 @@ import { UserInputRequestedPayload, isImportedAgentSessionMessageId, type OrchestrationCommand, + type OrchestrationSession, type OrchestrationEvent, type OrchestrationReadModel, type OrchestrationThread, @@ -218,6 +219,86 @@ const decideCommandSequence = Effect.fn("decideCommandSequence")(function* ({ return plannedEvents; }); +const compactionSessionEvent = Effect.fnUntraced(function* ( + command: { + readonly commandId: OrchestrationCommand["commandId"]; + readonly threadId: OrchestrationThread["id"]; + readonly createdAt: string; + }, + session: OrchestrationSession, +): Effect.fn.Return { + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.session-set", + payload: { threadId: command.threadId, session }, + }; +}); + +const canceledCompactionMessages = Effect.fnUntraced(function* ( + command: { + readonly commandId: OrchestrationCommand["commandId"]; + readonly threadId: OrchestrationThread["id"]; + readonly createdAt: string; + }, + session: OrchestrationSession | null, + detail: string, + reconcileInFlight = false, +): Effect.fn.Return< + ReadonlyArray, + PlatformError.PlatformError, + Crypto.Crypto +> { + const queue = session?.compactionQueue; + const messages = [...(queue?.queued ?? [])]; + const uncertainMessageId = + reconcileInFlight && queue?.inFlightRequestId !== session?.activeTurnRequestId + ? queue?.inFlightMessageId + : undefined; + return yield* Effect.forEach( + [...(uncertainMessageId ? [{ messageId: uncertainMessageId }] : []), ...messages], + (queued) => + Effect.gen(function* () { + const base = yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + }); + return { + ...base, + type: "thread.activity-appended" as const, + payload: { + threadId: command.threadId, + activity: { + id: base.eventId, + kind: "provider.turn.start.failed", + summary: "Queued message was not sent", + tone: "error" as const, + turnId: null, + createdAt: command.createdAt, + payload: { requestId: queued.messageId, detail }, + }, + }, + }; + }), + ); +}); + +// Provider snapshots do not own the server's FIFO. Preserve the newest queue +// across lifecycle writes, and never resurrect a queue from a stale snapshot. +const preserveCompactionQueue = ( + current: OrchestrationSession | null, + incoming: OrchestrationSession, +): OrchestrationSession => ({ + ...incoming, + compactionQueue: current?.compactionQueue, +}); + export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand")(function* ({ command, readModel, @@ -1389,6 +1470,108 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" detail: `Proposed plan '${sourceProposedPlan?.planId}' belongs to thread '${sourceThread.id}' in a different project.`, }); } + // Real activity resets ANY override: it wakes an explicitly settled + // thread, and it clears a keep-active pin back to neutral so the + // thread can auto-settle again after this burst of work goes stale. + // A snooze clears the same way — sending a message to a snoozed + // thread is the user re-engaging, so the return ticket is spent. + const lifecycleResetEvents: Array> = []; + if (targetThread.settledOverride !== null) { + lifecycleResetEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.unsettled", + payload: { + threadId: command.threadId, + reason: "activity", + updatedAt: command.createdAt, + }, + }); + } + if (targetThread.snoozedUntil != null) { + lifecycleResetEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.unsnoozed", + payload: { + threadId: command.threadId, + reason: "activity", + updatedAt: command.createdAt, + }, + }); + } + const isCompaction = + command.message.attachments.length === 0 && + command.message.text.trim().toLowerCase() === "/compact"; + if (isCompaction && targetThread.session?.status === "running") { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "Context compaction is unavailable while a provider turn is running.", + }); + } + const compaction = targetThread.session?.compactionQueue; + if (compaction && targetThread.session) { + if (isCompaction) + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: + "Wait for context compaction and its queued messages to finish before compacting again.", + }); + return [ + ...lifecycleResetEvents, + { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.message-sent", + payload: { + threadId: command.threadId, + messageId: command.message.messageId, + role: "user", + text: command.message.text, + attachments: command.message.attachments, + turnId: null, + streaming: false, + createdAt: command.createdAt, + updatedAt: command.createdAt, + }, + }, + yield* compactionSessionEvent(command, { + ...targetThread.session, + compactionQueue: { + ...compaction, + queued: [ + ...compaction.queued, + { + requestId: command.commandId, + messageId: command.message.messageId, + text: command.message.text, + attachments: command.message.attachments, + modelSelection: effectiveModelSelection, + runtimeMode: command.runtimeMode, + interactionMode: command.interactionMode, + sourceEpoch: actualSourceEpoch, + ...(sourceProposedPlan ? { sourceProposedPlan } : {}), + ...(command.titleSeed ? { titleSeed: command.titleSeed } : {}), + createdAt: command.createdAt, + }, + ], + }, + updatedAt: command.createdAt, + }), + ]; + } if ( targetThread.session?.status === "starting" && targetThread.session.pendingTurnRequestId !== undefined && @@ -1497,6 +1680,15 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" ? { providerInstanceId: targetThread.session.providerInstanceId } : {}), runtimeMode: targetThread.session?.runtimeMode ?? targetThread.runtimeMode, + ...(isCompaction + ? { + compactionQueue: { + requestId: command.commandId, + phase: "running" as const, + queued: [], + }, + } + : {}), pendingTurnRequestId: command.commandId, pendingTurnMessageId: command.message.messageId, pendingTurnRequestedAt: admissionRequestedAt, @@ -1513,44 +1705,6 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }); } - // Real activity resets ANY override: it wakes an explicitly settled - // thread, and it clears a keep-active pin back to neutral so the - // thread can auto-settle again after this burst of work goes stale. - // A snooze clears the same way — sending a message to a snoozed - // thread is the user re-engaging, so the return ticket is spent. - const lifecycleResetEvents: Array> = []; - if (targetThread.settledOverride !== null) { - lifecycleResetEvents.push({ - ...(yield* withEventBase({ - aggregateKind: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - })), - type: "thread.unsettled", - payload: { - threadId: command.threadId, - reason: "activity", - updatedAt: command.createdAt, - }, - }); - } - if (targetThread.snoozedUntil != null) { - lifecycleResetEvents.push({ - ...(yield* withEventBase({ - aggregateKind: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - })), - type: "thread.unsnoozed", - payload: { - threadId: command.threadId, - reason: "activity", - updatedAt: command.createdAt, - }, - }); - } return [ ...lifecycleResetEvents, userMessageEvent, @@ -1559,6 +1713,172 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" ]; } + case "thread.compaction.complete": { + const thread = yield* requireThread({ readModel, command, threadId: command.threadId }); + const session = thread.session; + const compaction = session?.compactionQueue; + if (!session || !compaction || compaction.requestId !== command.requestId) return []; + const sameIncarnation = + (command.expectedSessionIncarnationId === undefined || + command.expectedSessionIncarnationId === (session.sessionIncarnationId ?? null)) && + (command.expectedProviderInstanceId === undefined || + command.expectedProviderInstanceId === (session.providerInstanceId ?? null)); + const ownsAdmission = session.pendingTurnRequestId === command.requestId && sameIncarnation; + const failed = + !command.success || + !sameIncarnation || + session.status === "stopped" || + session.status === "error"; + return [ + ...(failed + ? yield* canceledCompactionMessages( + command, + session, + command.detail ?? + "Context compaction did not complete on the original session. Send this message again to continue.", + command.reconcileInFlight, + ) + : []), + yield* compactionSessionEvent(command, { + ...session, + ...(ownsAdmission + ? { + status: session.status === "stopped" ? ("stopped" as const) : ("ready" as const), + pendingTurnRequestId: undefined, + pendingTurnMessageId: undefined, + pendingTurnRequestedAt: undefined, + pendingTurnDeadlineAt: undefined, + pendingTurnSessionId: undefined, + activeTurnRequestId: undefined, + activeTurnId: null, + } + : {}), + compactionQueue: failed + ? undefined + : { + ...compaction, + phase: "draining", + expectedSessionIncarnationId: command.expectedSessionIncarnationId, + expectedProviderInstanceId: command.expectedProviderInstanceId, + }, + updatedAt: command.createdAt, + }), + ]; + } + case "thread.compaction.queue.sent": { + const thread = yield* requireThread({ readModel, command, threadId: command.threadId }); + const session = thread.session; + const compaction = session?.compactionQueue; + if ( + !session || + !compaction || + compaction.requestId !== command.requestId || + compaction.inFlightRequestId !== command.sentRequestId + ) + return []; + return yield* compactionSessionEvent(command, { + ...session, + compactionQueue: { + ...compaction, + inFlightRequestId: undefined, + inFlightMessageId: undefined, + }, + updatedAt: command.createdAt, + }); + } + case "thread.compaction.queue.resume": { + const thread = yield* requireThread({ readModel, command, threadId: command.threadId }); + const session = thread.session; + const compaction = session?.compactionQueue; + if ( + !session || + !compaction || + compaction.requestId !== command.requestId || + compaction.phase !== "draining" || + compaction.inFlightRequestId !== undefined + ) + return []; + const sameIncarnation = + (compaction.expectedSessionIncarnationId === undefined || + compaction.expectedSessionIncarnationId === (session.sessionIncarnationId ?? null)) && + (compaction.expectedProviderInstanceId === undefined || + compaction.expectedProviderInstanceId === (session.providerInstanceId ?? null)); + if (!sameIncarnation || session.status === "stopped" || session.status === "error") + return [ + ...(yield* canceledCompactionMessages( + command, + session, + "The session stopped before its queued messages could be sent. Send this message again to continue.", + )), + yield* compactionSessionEvent(command, { + ...session, + compactionQueue: undefined, + updatedAt: command.createdAt, + }), + ]; + if (session.pendingTurnRequestId !== undefined) return []; + const queued = compaction.queued[0]; + if (!queued) + return yield* compactionSessionEvent(command, { + ...session, + compactionQueue: undefined, + updatedAt: command.createdAt, + }); + // Decide the normal admission against a temporary queue-free view, then + // persist the new reservation and dequeue atomically in the same transaction. + const admissionReadModel = { + ...readModel, + threads: readModel.threads.map((entry) => + entry.id === thread.id + ? { ...entry, session: { ...session, compactionQueue: undefined } } + : entry, + ), + }; + const result = yield* decideOrchestrationCommand({ + readModel: admissionReadModel, + command: { + type: "thread.turn.start", + commandId: command.commandId, + threadId: command.threadId, + message: { + messageId: queued.messageId, + role: "user", + text: queued.text, + attachments: queued.attachments, + }, + modelSelection: queued.modelSelection, + runtimeMode: queued.runtimeMode, + interactionMode: queued.interactionMode, + sourceEpoch: queued.sourceEpoch, + sourceProposedPlan: queued.sourceProposedPlan, + titleSeed: queued.titleSeed, + createdAt: queued.createdAt, + }, + }); + const events = Array.isArray(result) ? result : [result]; + let projected: OrchestrationReadModel = admissionReadModel; + for (const event of events) + projected = yield* projectEvent(projected, { + ...event, + sequence: projected.snapshotSequence + 1, + }).pipe(Effect.orDie); + const admittedSession = + projected.threads.find((entry) => entry.id === thread.id)?.session ?? session; + return [ + ...events, + yield* compactionSessionEvent(command, { + ...admittedSession, + compactionQueue: { + ...compaction, + queued: compaction.queued.slice(1), + inFlightRequestId: command.commandId, + inFlightMessageId: queued.messageId, + }, + updatedAt: command.createdAt, + }), + ]; + } + case "thread.turn.admission.accept": { const thread = yield* requireThread({ readModel, @@ -1767,12 +2087,12 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.turn.interrupt": { - yield* requireThread({ + const thread = yield* requireThread({ readModel, command, threadId: command.threadId, }); - return { + const interruptedEvent: PlannedOrchestrationEvent = { ...(yield* withEventBase({ aggregateKind: "thread", aggregateId: command.threadId, @@ -1786,6 +2106,20 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" createdAt: command.createdAt, }, }; + if (!thread.session?.compactionQueue) return interruptedEvent; + return [ + ...(yield* canceledCompactionMessages( + command, + thread.session, + "Context compaction was interrupted. Send this message again to continue.", + )), + yield* compactionSessionEvent(command, { + ...thread.session, + compactionQueue: { ...thread.session.compactionQueue, queued: [] }, + updatedAt: command.createdAt, + }), + interruptedEvent, + ]; } case "thread.approval.respond": { @@ -2113,6 +2447,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" lastError: null, }), status: "stopped", + compactionQueue: undefined, pendingStopRequestId: command.commandId, pendingStopProviderInstanceId: targetSession?.providerInstanceId ?? null, pendingStopSessionIncarnationId: targetSession?.sessionIncarnationId ?? null, @@ -2127,7 +2462,15 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }, }, }; - return [stopRequestedEvent, stoppedEvent]; + return [ + ...(yield* canceledCompactionMessages( + command, + targetSession, + "The session was stopped during context compaction. Send this message again to continue.", + )), + stopRequestedEvent, + stoppedEvent, + ]; } case "thread.session.apply-lifecycle": { @@ -2171,7 +2514,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" type: "thread.session-set", payload: { threadId: command.threadId, - session: command.session, + session: preserveCompactionQueue(thread.session, command.session), }, }; const isSessionActivity = @@ -2276,7 +2619,18 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" type: "thread.session-set", payload: { threadId: command.threadId, - session: command.session, + session: { + ...preserveCompactionQueue(thread.session, command.session), + ...(thread.session?.compactionQueue?.inFlightRequestId === command.requestId + ? { + compactionQueue: { + ...thread.session.compactionQueue, + expectedProviderInstanceId: command.session.providerInstanceId ?? null, + expectedSessionIncarnationId: command.session.sessionIncarnationId ?? null, + }, + } + : {}), + }, }, }); return acceptedEvents; @@ -2293,7 +2647,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" thread.session?.pendingTurnRequestId !== undefined && (command.session.pendingTurnRequestId === undefined || command.session.pendingTurnRequestId === thread.session.pendingTurnRequestId); - const session = preservesPendingAdmission + const providerSession = preservesPendingAdmission ? { ...thread.session, ...command.session, @@ -2305,6 +2659,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" command.session.pendingTurnSessionId ?? thread.session.pendingTurnSessionId, } : command.session; + const session = preserveCompactionQueue(thread.session, providerSession); const sessionSetEvent: Omit = { ...(yield* withEventBase({ aggregateKind: "thread", diff --git a/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts b/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts index 0a84b05f6..8ba0117c5 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts @@ -1,3 +1,4 @@ +import { OrchestrationCompactionQueue } from "@t3tools/contracts"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqlSchema from "effect/unstable/sql/SqlSchema"; import * as Effect from "effect/Effect"; @@ -17,6 +18,7 @@ import { const ProjectionThreadSessionDbRow = Schema.Struct({ ...ProjectionThreadSession.fields, + compactionQueue: Schema.NullOr(Schema.fromJsonString(OrchestrationCompactionQueue)), restored: Schema.Number, pendingTurnRequestAmbiguous: Schema.Number, }); @@ -46,6 +48,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { started_at, session_incarnation_id, harness_refinement_status, + compaction_queue_json, pending_turn_request_id, pending_turn_request_ambiguous, pending_turn_message_id, @@ -73,6 +76,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { ${row.startedAt}, ${row.sessionIncarnationId}, ${row.harnessRefinementStatus}, + ${row.compactionQueue ? JSON.stringify(row.compactionQueue) : null}, ${row.pendingTurnRequestId}, ${row.pendingTurnRequestAmbiguous ? 1 : 0}, ${row.pendingTurnMessageId}, @@ -100,6 +104,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { started_at = excluded.started_at, session_incarnation_id = excluded.session_incarnation_id, harness_refinement_status = excluded.harness_refinement_status, + compaction_queue_json = excluded.compaction_queue_json, pending_turn_request_id = excluded.pending_turn_request_id, pending_turn_request_ambiguous = excluded.pending_turn_request_ambiguous, pending_turn_message_id = excluded.pending_turn_message_id, @@ -134,6 +139,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () { started_at AS "startedAt", session_incarnation_id AS "sessionIncarnationId", harness_refinement_status AS "harnessRefinementStatus", + compaction_queue_json AS "compactionQueue", pending_turn_request_id AS "pendingTurnRequestId", pending_turn_request_ambiguous AS "pendingTurnRequestAmbiguous", pending_turn_message_id AS "pendingTurnMessageId", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 41fabfbef..2e3f5acec 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -70,6 +70,7 @@ import Migration0057 from "./Migrations/057_ProjectionThreadsActiveOrderKey.ts"; import Migration0055 from "./Migrations/055_ProjectionThreadBranchPullRequest.ts"; import Migration0058 from "./Migrations/058_ProjectionProjectsAutoPull.ts"; import Migration0059 from "./Migrations/059_ProjectionThreadPullRequests.ts"; +import Migration0060 from "./Migrations/060_ProjectionThreadCompactionQueue.ts"; /** * Migration loader with all migrations defined inline. * @@ -81,6 +82,7 @@ import Migration0059 from "./Migrations/059_ProjectionThreadPullRequests.ts"; * returns migrations sorted by ID. */ const migrationEntries = [ + [60, "ProjectionThreadCompactionQueue", Migration0060], [1, "OrchestrationEvents", Migration0001], [2, "OrchestrationCommandReceipts", Migration0002], [3, "CheckpointDiffBlobs", Migration0003], diff --git a/apps/server/src/persistence/Migrations/060_ProjectionThreadCompactionQueue.ts b/apps/server/src/persistence/Migrations/060_ProjectionThreadCompactionQueue.ts new file mode 100644 index 000000000..d1736dcc7 --- /dev/null +++ b/apps/server/src/persistence/Migrations/060_ProjectionThreadCompactionQueue.ts @@ -0,0 +1,7 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`ALTER TABLE projection_thread_sessions ADD COLUMN compaction_queue_json TEXT`; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreadSessions.ts b/apps/server/src/persistence/Services/ProjectionThreadSessions.ts index ac2e7c940..f20bc759f 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadSessions.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadSessions.ts @@ -12,6 +12,7 @@ import { IsoDateTime, MessageId, OrchestrationSessionStatus, + OrchestrationCompactionQueue, SessionHarnessRefinementStatus, ProviderInstanceId, RuntimeSessionId, @@ -35,6 +36,7 @@ export const ProjectionThreadSession = Schema.Struct({ startedAt: Schema.NullOr(IsoDateTime), sessionIncarnationId: Schema.NullOr(RuntimeSessionId), harnessRefinementStatus: Schema.NullOr(SessionHarnessRefinementStatus), + compactionQueue: Schema.optional(Schema.NullOr(OrchestrationCompactionQueue)), pendingTurnRequestId: Schema.NullOr(CommandId), /** Internal replay state: duplicate legacy events cannot identify one exact request. */ pendingTurnRequestAmbiguous: Schema.Boolean, diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index a7627f983..c401cb2d4 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1,6 +1,7 @@ import { scopeThreadRef } from "@t3tools/client-runtime/environment"; import { ANTIGRAVITY_DEFAULT_MODEL, + CommandId, EnvironmentId, MessageId, ProjectId, @@ -1956,6 +1957,30 @@ describe("hasServerAcknowledgedLocalDispatch", () => { ).toBe(false); }); + it("acknowledges compaction reservations while the provider is starting", () => { + const localDispatch = createLocalDispatchSnapshot( + makeThread({ latestTurn: completedTurn, session: readySession }), + ); + const messageId = MessageId.make("message-compact"); + expect( + hasServerAcknowledgedLocalDispatch({ + localDispatch, + phase: "connecting", + latestTurn: completedTurn, + latestUserMessageId: messageId, + session: { + ...readySession, + status: "starting", + pendingTurnMessageId: messageId, + compactionQueue: { requestId: CommandId.make("compact"), phase: "running", queued: [] }, + }, + hasPendingApproval: false, + hasPendingUserInput: false, + threadError: null, + }), + ).toBe(true); + }); + it("keeps a follow-up active while its provider session is starting", () => { const localDispatch = createLocalDispatchSnapshot( makeThread({ latestTurn: completedTurn, session: readySession }), diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index a5e2ed05f..528995cca 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1201,6 +1201,18 @@ export function hasServerAcknowledgedLocalDispatch(input: { ) { return true; } + const compactionQueue = input.session?.compactionQueue; + if ( + compactionQueue && + input.latestUserMessageId !== input.localDispatch.latestUserMessageId && + (input.session?.pendingTurnMessageId === input.latestUserMessageId || + compactionQueue.inFlightMessageId === input.latestUserMessageId || + compactionQueue.queued.some((message) => message.messageId === input.latestUserMessageId)) + ) { + // The persisted compaction FIFO is an acknowledgment even while the + // provider is starting. Keep the composer available for another message. + return true; + } if (input.phase === "connecting") { return false; } diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 01f9bbe3c..c0b4314d0 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -7057,13 +7057,9 @@ export default function ChatView(props: ChatViewProps) { try { const settingsResult = await persistThreadSettingsForNextTurn({ threadId, - createdAt, - modelSelection: context.selectedModelSelection, ...(localCheckoutBranchMismatch ? { branch: localCheckoutBranchMismatch.currentBranch } : {}), - runtimeMode, - interactionMode: context.interactionMode, }); const result = settingsResult._tag === "Failure" @@ -7075,7 +7071,8 @@ export default function ChatView(props: ChatViewProps) { message: { messageId, role: "user", text: "/compact", attachments: [] }, modelSelection: context.selectedModelSelection, runtimeMode, - interactionMode: context.interactionMode, + interactionMode, + sourceEpoch: activeThread.sourceEpoch ?? 0, createdAt, }, }); @@ -7092,7 +7089,7 @@ export default function ChatView(props: ChatViewProps) { ); } } else { - clearUsageLimitsFor(routeThreadKey); + setUsageLimitsNotice((current) => (current?.threadKey === routeThreadKey ? null : current)); } } finally { sendInFlightRef.current = false; diff --git a/docs/user/context-compaction.md b/docs/user/context-compaction.md new file mode 100644 index 000000000..82004eccf --- /dev/null +++ b/docs/user/context-compaction.md @@ -0,0 +1,7 @@ +# Compacting context + +Use **Compact context** to shorten an existing supported conversation without clearing the composer draft or its attachments. Pylon submits `/compact` separately. Messages sent while that operation runs are saved in order and sent after compaction completes, with the model and permissions selected for each message. + +Stopping or interrupting cancels the unsent queue and leaves a failure message on each affected message. If compaction fails or the server restarts, unsent messages remain visible with an explanation. An interrupted send may already have reached the provider; check the conversation before sending it again. + +Prime Agent's native **Compact now** and **Abort compaction** controls retain their own provider status and admission rules. See [Prime Agent](providers-prime-agent.md) for those controls. diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index ce77ea7cb..7684608b1 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -541,6 +541,31 @@ export const OrchestrationSessionStatus = Schema.Literals([ ]); export type OrchestrationSessionStatus = typeof OrchestrationSessionStatus.Type; +/** Durable FIFO while the exact slash-command compaction owns turn admission. */ +export const OrchestrationCompactionQueue = Schema.Struct({ + requestId: CommandId, + phase: Schema.Literals(["running", "draining"]), + inFlightRequestId: Schema.optional(CommandId), + inFlightMessageId: Schema.optional(MessageId), + expectedSessionIncarnationId: Schema.optional(Schema.NullOr(RuntimeSessionId)), + expectedProviderInstanceId: Schema.optional(Schema.NullOr(ProviderInstanceId)), + queued: Schema.Array( + Schema.Struct({ + requestId: CommandId, + messageId: MessageId, + text: Schema.String, + attachments: Schema.Array(ChatAttachment), + modelSelection: ModelSelection, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + sourceEpoch: NonNegativeInt, + sourceProposedPlan: Schema.optional(SourceProposedPlanReference), + titleSeed: Schema.optional(TrimmedNonEmptyString), + createdAt: IsoDateTime, + }), + ), +}); + export const OrchestrationSession = Schema.Struct({ threadId: ThreadId, status: OrchestrationSessionStatus, @@ -554,6 +579,7 @@ export const OrchestrationSession = Schema.Struct({ /** Immutable identity of the current provider-session incarnation. */ sessionIncarnationId: Schema.optional(RuntimeSessionId), harnessRefinementStatus: Schema.optional(SessionHarnessRefinementStatus), + compactionQueue: Schema.optional(OrchestrationCompactionQueue), /** Correlates the user message currently waiting for provider turn admission. */ pendingTurnRequestId: Schema.optional(CommandId), /** Exact user message owned by the pending admission. */ @@ -1837,7 +1863,38 @@ const ThreadPullRequestLinkSyncCommand = Schema.Struct({ stack: Schema.NullOr(ThreadPullRequestStack), }); +const ThreadCompactionCompleteCommand = Schema.Struct({ + type: Schema.Literal("thread.compaction.complete"), + commandId: CommandId, + threadId: ThreadId, + requestId: CommandId, + success: Schema.Boolean, + expectedSessionIncarnationId: Schema.optional(Schema.NullOr(RuntimeSessionId)), + expectedProviderInstanceId: Schema.optional(Schema.NullOr(ProviderInstanceId)), + reconcileInFlight: Schema.optional(Schema.Boolean), + detail: Schema.optional(TrimmedNonEmptyString), + createdAt: IsoDateTime, +}); +const ThreadCompactionQueueResumeCommand = Schema.Struct({ + type: Schema.Literal("thread.compaction.queue.resume"), + commandId: CommandId, + threadId: ThreadId, + requestId: CommandId, + createdAt: IsoDateTime, +}); +const ThreadCompactionQueueSentCommand = Schema.Struct({ + type: Schema.Literal("thread.compaction.queue.sent"), + commandId: CommandId, + threadId: ThreadId, + requestId: CommandId, + sentRequestId: CommandId, + createdAt: IsoDateTime, +}); + const InternalOrchestrationCommand = Schema.Union([ + ThreadCompactionCompleteCommand, + ThreadCompactionQueueResumeCommand, + ThreadCompactionQueueSentCommand, ThreadAutoSettleCommand, ThreadPullRequestSyncCommand, ThreadPullRequestLinkSyncCommand, From 62c1598f6f578caa1670ece666392116a5685103 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 12 Sep 2026 00:55:12 -0600 Subject: [PATCH 3/7] fix(server): preserve exact compaction ownership across Stop Use the ordinary pending-session bind CAS for compaction startup, retain explicit null expectations, and emit uncertain-delivery receipts for dequeued messages interrupted before admission. Add deterministic slow-start Stop and in-flight cancellation regression coverage. --- .../Layers/ProviderCommandReactor.test.ts | 79 +++++++++++++++++++ .../Layers/ProviderCommandReactor.ts | 28 ++++++- .../orchestration/decider.compaction.test.ts | 33 +++++++- apps/server/src/orchestration/decider.ts | 15 +++- 4 files changed, 150 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 6cd37660e..3bbeb6fc7 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -1161,6 +1161,9 @@ describe("ProviderCommandReactor", () => { commandId: CommandId.make("cmd-session-ready-before-blocked-compact"), threadId, session: { + sessionIncarnationId: (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === threadId, + )?.session?.sessionIncarnationId, threadId, status: "ready", providerName: "codex", @@ -1224,6 +1227,79 @@ describe("ProviderCommandReactor", () => { }), ); + effectIt.effect("keeps Stop authoritative when compaction session startup finishes late", () => + Effect.gen(function* () { + const started = yield* Deferred.make(); + const releaseStart = yield* Deferred.make(); + const lateSessionStopped = yield* Deferred.make(); + const threadId = ThreadId.make("thread-1"); + const harness = yield* Effect.promise(() => + createHarness({ + startSessionEffect: (session) => + Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(releaseStart)), + Effect.as(session), + ), + stopSessionEffect: () => + Deferred.succeed(lateSessionStopped, undefined).pipe(Effect.asVoid), + beforeReactorStart: Effect.gen(function* () { + const engine = yield* OrchestrationEngineService; + yield* engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("seed-context"), + threadId, + message: { + messageId: MessageId.make("seed-context"), + role: "user", + text: "existing conversation", + attachments: [], + }, + runtimeMode: "approval-required", + interactionMode: "default", + createdAt: "2026-01-01T00:00:00.000Z", + }); + yield* engine.dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("stop-seed"), + threadId, + createdAt: "2026-01-01T00:00:00.000Z", + }); + }), + }), + ); + yield* harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("slow-compact"), + threadId, + message: { + messageId: MessageId.make("slow-compact"), + role: "user", + text: "/compact", + attachments: [], + }, + runtimeMode: "approval-required", + interactionMode: "default", + createdAt: "2026-01-01T00:00:01.000Z", + }); + yield* Deferred.await(started); + yield* harness.engine.dispatch({ + type: "thread.session.stop", + commandId: CommandId.make("stop-slow-compact"), + threadId, + createdAt: "2026-01-01T00:00:02.000Z", + }); + yield* Effect.promise(() => harness.drain()); + yield* Deferred.succeed(releaseStart, undefined); + yield* Deferred.await(lateSessionStopped); + yield* Effect.promise(() => harness.drain()); + expect((yield* Effect.promise(() => harness.readModel())).threads[0]?.session).toMatchObject({ + status: "stopped", + }); + expect(harness.compactThread).not.toHaveBeenCalled(); + expect(harness.sendTurn).not.toHaveBeenCalled(); + }), + ); + effectIt.effect("reconciles persisted compaction messages explicitly after restart", () => Effect.gen(function* () { const harness = yield* Effect.promise(() => @@ -1337,6 +1413,9 @@ describe("ProviderCommandReactor", () => { commandId: CommandId.make("cmd-session-ready-before-compact"), threadId, session: { + sessionIncarnationId: (yield* Effect.promise(() => harness.readModel())).threads.find( + (entry) => entry.id === threadId, + )?.session?.sessionIncarnationId, threadId, status: "ready", providerName: "codex", diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index dd591e71a..c20c85e0a 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -1737,6 +1737,7 @@ const make = Effect.gen(function* () { ); return; } + let compactedSession: ProviderSession | undefined; const handleCompactionFailure = (cause: Cause.Cause) => { if (Cause.hasInterruptsOnly(cause)) return Effect.void; const detail = formatFailureDetail(cause); @@ -1744,7 +1745,13 @@ const make = Effect.gen(function* () { Effect.ensuring( // A no-op unless the session actually reached a restorable state, // so this covers both a failed ensure and a failed compaction. - completeCompaction(event.payload.threadId, requestId, false, detail).pipe( + completeCompaction( + event.payload.threadId, + requestId, + false, + detail, + compactedSession, + ).pipe( Effect.catchCause((restoreCause) => Effect.logWarning("failed to restore provider session after compaction failure", { threadId: event.payload.threadId, @@ -1770,15 +1777,32 @@ const make = Effect.gen(function* () { runtimeMode: event.payload.runtimeMode, interactionMode: event.payload.interactionMode, pendingTurnStart: true, + pendingTurnRequestId: requestId, + pendingTurnMessageId: event.payload.messageId, + pendingTurnRequestedAt: admissionRequestedAt, + pendingTurnDeadlineAt: admissionDeadlineAt, + expectedProviderInstanceId: + event.payload.admissionIntent !== undefined + ? event.payload.admissionIntent.expectedProviderInstanceId + : (thread.session?.providerInstanceId ?? null), + expectedSessionIncarnationId: + event.payload.admissionIntent !== undefined + ? event.payload.admissionIntent.expectedSessionIncarnationId + : (thread.session?.sessionIncarnationId ?? null), }, ); if (!compactionSession) return yield* Effect.interrupt; + compactedSession = compactionSession; const current = (yield* resolveThreadShell(event.payload.threadId))?.session; if ( current?.compactionQueue?.requestId !== requestId || current.sessionIncarnationId !== compactionSession.sessionIncarnationId ) - return yield* Effect.interrupt; + return yield* new ProviderAdapterRequestError({ + provider: compactionSession.provider, + method: "thread.compact", + detail: "The provider session changed before compaction could start.", + }); if (event.payload.modelSelection !== undefined) { threadModelSelections.set(event.payload.threadId, event.payload.modelSelection); } diff --git a/apps/server/src/orchestration/decider.compaction.test.ts b/apps/server/src/orchestration/decider.compaction.test.ts index 81c792be6..af62ceb74 100644 --- a/apps/server/src/orchestration/decider.compaction.test.ts +++ b/apps/server/src/orchestration/decider.compaction.test.ts @@ -161,6 +161,33 @@ it.layer(NodeServices.layer)("compaction admission FIFO", (it) => { }), ); + it.effect("accounts for an in-flight head when Stop cancels its pending admission", () => + Effect.gen(function* () { + let state = yield* init(); + state = (yield* run(state, turn("one"))).readModel; + state = (yield* run(state, turn("two"))).readModel; + state = (yield* run(state, complete(true))).readModel; + state = (yield* run(state, resume("resume-one"))).readModel; + const stopped = yield* run(state, { + type: "thread.session.stop", + commandId: CommandId.make("stop-pending"), + threadId, + createdAt: NOW, + }); + const receipts = stopped.events.filter((event) => event.type === "thread.activity-appended"); + expect(receipts).toHaveLength(2); + expect(receipts[0]?.payload.activity).toMatchObject({ + summary: "Queued message delivery was interrupted", + payload: { requestId: "message-one" }, + }); + expect(receipts[1]?.payload.activity).toMatchObject({ + summary: "Queued message was not sent", + payload: { requestId: "message-two" }, + }); + expect(stopped.readModel.threads[0]?.session?.compactionQueue).toBeUndefined(); + }), + ); + it.effect( "does not let an old provider snapshot erase the queue or completion drain a replacement incarnation", () => @@ -207,7 +234,11 @@ it.layer(NodeServices.layer)("compaction admission FIFO", (it) => { .filter((event) => event.type === "thread.activity-appended") .map((event) => event.payload.activity.payload), ).toEqual([ - { requestId: "message-one", detail: "Server restarted; delivery may have started." }, + { + requestId: "message-one", + detail: + "Server restarted; delivery may have started. Delivery may already have started; check the conversation before resending.", + }, { requestId: "message-two", detail: "Server restarted; delivery may have started." }, ]); expect(canceled.readModel.threads[0]?.session?.compactionQueue).toBeUndefined(); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index f476f96c3..78133c2a2 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -277,11 +277,20 @@ const canceledCompactionMessages = Effect.fnUntraced(function* ( activity: { id: base.eventId, kind: "provider.turn.start.failed", - summary: "Queued message was not sent", + summary: + queued.messageId === uncertainMessageId + ? "Queued message delivery was interrupted" + : "Queued message was not sent", tone: "error" as const, turnId: null, createdAt: command.createdAt, - payload: { requestId: queued.messageId, detail }, + payload: { + requestId: queued.messageId, + detail: + queued.messageId === uncertainMessageId + ? `${detail} Delivery may already have started; check the conversation before resending.` + : detail, + }, }, }, }; @@ -2112,6 +2121,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" command, thread.session, "Context compaction was interrupted. Send this message again to continue.", + true, )), yield* compactionSessionEvent(command, { ...thread.session, @@ -2467,6 +2477,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" command, targetSession, "The session was stopped during context compaction. Send this message again to continue.", + true, )), stopRequestedEvent, stoppedEvent, From 9e4efeaac7ca250c8aca55ccc6d3d60fd4d9e514 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 12 Sep 2026 00:56:47 -0600 Subject: [PATCH 4/7] test(server): wait for late compaction session cleanup --- .../Layers/ProviderCommandReactor.test.ts | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 3bbeb6fc7..0acf65eb8 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -1232,6 +1232,7 @@ describe("ProviderCommandReactor", () => { const started = yield* Deferred.make(); const releaseStart = yield* Deferred.make(); const lateSessionStopped = yield* Deferred.make(); + let releasedLateStart = false; const threadId = ThreadId.make("thread-1"); const harness = yield* Effect.promise(() => createHarness({ @@ -1241,7 +1242,9 @@ describe("ProviderCommandReactor", () => { Effect.as(session), ), stopSessionEffect: () => - Deferred.succeed(lateSessionStopped, undefined).pipe(Effect.asVoid), + releasedLateStart + ? Deferred.succeed(lateSessionStopped, undefined).pipe(Effect.asVoid) + : Effect.void, beforeReactorStart: Effect.gen(function* () { const engine = yield* OrchestrationEngineService; yield* engine.dispatch({ @@ -1289,12 +1292,20 @@ describe("ProviderCommandReactor", () => { createdAt: "2026-01-01T00:00:02.000Z", }); yield* Effect.promise(() => harness.drain()); + releasedLateStart = true; yield* Deferred.succeed(releaseStart, undefined); yield* Deferred.await(lateSessionStopped); yield* Effect.promise(() => harness.drain()); expect((yield* Effect.promise(() => harness.readModel())).threads[0]?.session).toMatchObject({ status: "stopped", }); + expect(harness.stopSession).toHaveBeenCalledWith( + expect.objectContaining({ + expectedSessionIncarnationId: "session-1", + expectedAdmissionRequestId: "slow-compact", + invalidateStartReservation: false, + }), + ); expect(harness.compactThread).not.toHaveBeenCalled(); expect(harness.sendTurn).not.toHaveBeenCalled(); }), From e4c500bbdef91d3f1fe7610f74e537875975d82f Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 12 Sep 2026 01:04:11 -0600 Subject: [PATCH 5/7] test(server): compare restart receipts independently of activity order --- .../src/orchestration/Layers/ProviderCommandReactor.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 0acf65eb8..2533a2927 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -1344,6 +1344,8 @@ describe("ProviderCommandReactor", () => { const thread = (yield* Effect.promise(() => harness.readModel())).threads[0]; expect(thread?.session?.compactionQueue).toBeUndefined(); expect(thread?.session?.pendingTurnRequestId).toBeUndefined(); + // Cancellation receipts share a timestamp and have generated activity IDs; + // their presentation order is independent of the separately tested send FIFO. expect( thread?.activities .filter((activity) => activity.kind === "provider.turn.start.failed") @@ -1353,7 +1355,8 @@ describe("ProviderCommandReactor", () => { "requestId" in activity.payload ? activity.payload.requestId : undefined, - ), + ) + .sort(), ).toEqual(["message-restarted-one", "message-restarted-two"]); expect(harness.compactThread).not.toHaveBeenCalled(); expect(harness.sendTurn).not.toHaveBeenCalled(); From 2ded1a62402a0f5ce6367e0ae98915a7f3c851f8 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 12 Sep 2026 01:06:48 -0600 Subject: [PATCH 6/7] fix(server): register compaction migration after pull-request persistence --- apps/server/src/persistence/Migrations.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 2e3f5acec..ecb5273d9 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -82,7 +82,6 @@ import Migration0060 from "./Migrations/060_ProjectionThreadCompactionQueue.ts"; * returns migrations sorted by ID. */ const migrationEntries = [ - [60, "ProjectionThreadCompactionQueue", Migration0060], [1, "OrchestrationEvents", Migration0001], [2, "OrchestrationCommandReceipts", Migration0002], [3, "CheckpointDiffBlobs", Migration0003], @@ -173,6 +172,7 @@ const migrationEntries = [ // connections and its lineage runs through 57. [58, "ProjectionProjectsAutoPull", Migration0058], [59, "ProjectionThreadPullRequests", Migration0059], + [60, "ProjectionThreadCompactionQueue", Migration0060], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); From c9c901df5ac115af1dcdb194698575488ae19743 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 12 Sep 2026 01:20:01 -0600 Subject: [PATCH 7/7] test(server): migrate replay fixtures before current projection bootstrap Keep migration 48 assertions on its historical schema, then apply current migrations before comparing a modern projector rebuild. This includes the compaction queue column required by the current session repository and matches production startup ordering. --- .../048_ProjectionThreadSessionPendingTurnRequest.test.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/apps/server/src/persistence/Migrations/048_ProjectionThreadSessionPendingTurnRequest.test.ts b/apps/server/src/persistence/Migrations/048_ProjectionThreadSessionPendingTurnRequest.test.ts index b6aae941a..5354c02c1 100644 --- a/apps/server/src/persistence/Migrations/048_ProjectionThreadSessionPendingTurnRequest.test.ts +++ b/apps/server/src/persistence/Migrations/048_ProjectionThreadSessionPendingTurnRequest.test.ts @@ -229,9 +229,10 @@ layer("048_ProjectionThreadSessionPendingTurnRequest", (it) => { DELETE FROM projection_state WHERE projector IN ('projection.thread-sessions', 'projection.thread-turns') `; - // The current projector repository reads the additive pending-stop - // columns introduced immediately after this historical migration. - yield* runMigrations({ toMigrationInclusive: 49 }); + // The historical assertions above stop at migration 48. Rebuilding with + // today's projector requires all current repository columns, just like + // production startup, including the compaction queue added in migration 60. + yield* runMigrations(); yield* projectionPipeline.bootstrap; assert.deepStrictEqual(yield* readPendingSessionRows, migrated); }),