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..2533a2927 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, }); @@ -1148,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", @@ -1164,28 +1180,186 @@ 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("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(); + let releasedLateStart = false; + 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: () => + releasedLateStart + ? Deferred.succeed(lateSessionStopped, undefined).pipe(Effect.asVoid) + : Effect.void, + 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()); + 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(); + }), + ); + + 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(); + // 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") + .map((activity) => + activity.payload && + typeof activity.payload === "object" && + "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(); }), ); @@ -1253,6 +1427,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 e51e83b84..c20c85e0a 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,8 +1729,15 @@ 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; } + let compactedSession: ProviderSession | undefined; const handleCompactionFailure = (cause: Cause.Cause) => { if (Cause.hasInterruptsOnly(cause)) return Effect.void; const detail = formatFailureDetail(cause); @@ -1717,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. - restoreCompaction(event.payload.threadId).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, @@ -1731,18 +1765,44 @@ 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, + 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* 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); } @@ -1751,8 +1811,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 +1949,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 +2433,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 +2650,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 +2688,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 +2710,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 +2745,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..af62ceb74 --- /dev/null +++ b/apps/server/src/orchestration/decider.compaction.test.ts @@ -0,0 +1,266 @@ +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("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", + () => + 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. 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(); + }), + ); + + 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..78133c2a2 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,95 @@ 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.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: + queued.messageId === uncertainMessageId + ? `${detail} Delivery may already have started; check the conversation before resending.` + : 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 +1479,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 +1689,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 +1714,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 +1722,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 +2096,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 +2115,21 @@ 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.", + true, + )), + yield* compactionSessionEvent(command, { + ...thread.session, + compactionQueue: { ...thread.session.compactionQueue, queued: [] }, + updatedAt: command.createdAt, + }), + interruptedEvent, + ]; } case "thread.approval.respond": { @@ -2113,6 +2457,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 +2472,16 @@ 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.", + true, + )), + stopRequestedEvent, + stoppedEvent, + ]; } case "thread.session.apply-lifecycle": { @@ -2171,7 +2525,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 +2630,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 +2658,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 +2670,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..ecb5273d9 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. * @@ -171,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); 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); }), 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 11c840952..c0b4314d0 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,75 @@ 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, + ...(localCheckoutBranchMismatch + ? { branch: localCheckoutBranchMismatch.currentBranch } + : {}), + }); + const result = + settingsResult._tag === "Failure" + ? settingsResult + : await startThreadTurn({ + environmentId, + input: { + threadId, + message: { messageId, role: "user", text: "/compact", attachments: [] }, + modelSelection: context.selectedModelSelection, + runtimeMode, + interactionMode, + sourceEpoch: activeThread.sourceEpoch ?? 0, + 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 { + setUsageLimitsNotice((current) => (current?.threadKey === routeThreadKey ? null : current)); + } + } finally { + sendInFlightRef.current = false; + } + }; + const onSend = async ( e?: { preventDefault: () => void }, submissionIntent: ComposerSubmissionIntent = "foreground", @@ -9947,6 +10015,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) { 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,