diff --git a/.github/upstream-candidates.json b/.github/upstream-candidates.json index e7c15c52c70..1ad3b0a1c1a 100644 --- a/.github/upstream-candidates.json +++ b/.github/upstream-candidates.json @@ -24,6 +24,12 @@ "sourceSha": "f7eaa00b99e67a9c1caf09e2fe6ff1f536f66b91", "status": "active", "purpose": "Show answered provider questions in the web thread timeline" + }, + { + "upstreamPr": 4245, + "sourceSha": "6be48eb238cf310a60cc9571240601e774c7d381", + "status": "active", + "purpose": "Queue follow-up messages server-side during active turns with explicit steer controls" } ] } diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index b1bdb61b961..80720340796 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -45,6 +45,8 @@ function makeThread( archivedAt: null, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index bee749e80bb..cd74c3fbe75 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -150,6 +150,8 @@ describe("OrchestrationEngine", () => { settledAt: null, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], @@ -162,6 +164,8 @@ describe("OrchestrationEngine", () => { threads: projectionSnapshot.threads.map((thread) => ({ ...thread, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 1f24a4a0200..2ec1ca1a83d 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -28,6 +28,7 @@ import { type ProjectionThreadProposedPlan, ProjectionThreadProposedPlanRepository, } from "../../persistence/Services/ProjectionThreadProposedPlans.ts"; +import { ProjectionQueuedMessageRepository } from "../../persistence/Services/ProjectionQueuedMessages.ts"; import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts"; import { type ProjectionTurn, @@ -40,6 +41,7 @@ import { ProjectionStateRepositoryLive } from "../../persistence/Layers/Projecti import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts"; import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts"; import { ProjectionThreadProposedPlanRepositoryLive } from "../../persistence/Layers/ProjectionThreadProposedPlans.ts"; +import { ProjectionQueuedMessageRepositoryLive } from "../../persistence/Layers/ProjectionQueuedMessages.ts"; import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts"; import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts"; @@ -59,6 +61,7 @@ export const ORCHESTRATION_PROJECTOR_NAMES = { projects: "projection.projects", threads: "projection.threads", threadMessages: "projection.thread-messages", + queuedMessages: "projection.queued-messages", threadProposedPlans: "projection.thread-proposed-plans", threadActivities: "projection.thread-activities", threadSessions: "projection.thread-sessions", @@ -329,7 +332,7 @@ function retainProjectionProposedPlansAfterRevert( function collectThreadAttachmentRelativePaths( threadId: string, - messages: ReadonlyArray, + messages: ReadonlyArray>, ): Set { const threadSegment = toSafeThreadAttachmentSegment(threadId); if (!threadSegment) { @@ -476,6 +479,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const projectionThreadRepository = yield* ProjectionThreadRepository; const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository; const projectionThreadProposedPlanRepository = yield* ProjectionThreadProposedPlanRepository; + const projectionQueuedMessageRepository = yield* ProjectionQueuedMessageRepository; const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository; const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository; const projectionTurnRepository = yield* ProjectionTurnRepository; @@ -782,6 +786,8 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } case "thread.message-sent": + case "thread.message-queued": + case "thread.queued-message-removed": case "thread.proposed-plan-upserted": case "thread.activity-appended": case "thread.approval-response-requested": @@ -942,9 +948,17 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti yield* Effect.forEach(keptRows, projectionThreadMessageRepository.upsert, { concurrency: 1, }).pipe(Effect.asVoid); + // Queued messages survive a revert (they are not part of the + // timeline), so their attachment files must survive pruning too. + const queuedRows = yield* projectionQueuedMessageRepository.listByThreadId({ + threadId: event.payload.threadId, + }); attachmentSideEffects.prunedThreadRelativePaths.set( event.payload.threadId, - collectThreadAttachmentRelativePaths(event.payload.threadId, keptRows), + collectThreadAttachmentRelativePaths(event.payload.threadId, [ + ...keptRows, + ...queuedRows, + ]), ); return; } @@ -954,6 +968,67 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } }); + const applyQueuedMessagesProjection: ProjectorDefinition["apply"] = Effect.fn( + "applyQueuedMessagesProjection", + )(function* (event, attachmentSideEffects) { + switch (event.type) { + case "thread.message-queued": + yield* projectionQueuedMessageRepository.upsert({ + messageId: event.payload.messageId, + threadId: event.payload.threadId, + text: event.payload.text, + attachments: event.payload.attachments, + modelSelection: event.payload.modelSelection ?? null, + sourceProposedPlanThreadId: event.payload.sourceProposedPlan?.threadId ?? null, + sourceProposedPlanId: event.payload.sourceProposedPlan?.planId ?? null, + queuedAt: event.payload.queuedAt, + }); + return; + + case "thread.queued-message-removed": { + // A user removal orphans the removed message's attachment files — + // prune to what the timeline and remaining queue still reference. + // Dispatch removals keep everything: the same attachments re-enter + // the timeline via the paired thread.message-sent. + const removedQueuedMessage = + event.payload.reason === "user" + ? (yield* projectionQueuedMessageRepository.listByThreadId({ + threadId: event.payload.threadId, + })).find((entry) => entry.messageId === event.payload.messageId) + : undefined; + yield* projectionQueuedMessageRepository.deleteByMessageId({ + threadId: event.payload.threadId, + messageId: event.payload.messageId, + }); + if (removedQueuedMessage && (removedQueuedMessage.attachments?.length ?? 0) > 0) { + const retainedMessageRows = yield* projectionThreadMessageRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + const retainedQueuedRows = yield* projectionQueuedMessageRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + attachmentSideEffects.prunedThreadRelativePaths.set( + event.payload.threadId, + collectThreadAttachmentRelativePaths(event.payload.threadId, [ + ...retainedMessageRows, + ...retainedQueuedRows, + ]), + ); + } + return; + } + + case "thread.deleted": + yield* projectionQueuedMessageRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + return; + + default: + return; + } + }); + const applyThreadProposedPlansProjection: ProjectorDefinition["apply"] = Effect.fn( "applyThreadProposedPlansProjection", )(function* (event, _attachmentSideEffects) { @@ -1093,19 +1168,22 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti case "thread.session-set": { const turnId = event.payload.session.activeTurnId; if (turnId === null || event.payload.session.status !== "running") { - if ( - event.payload.session.status === "error" || - event.payload.session.status === "stopped" || - event.payload.session.status === "interrupted" - ) { - yield* projectionTurnRepository.deletePendingTurnStartByThreadId({ - threadId: event.payload.threadId, - }); - } // Leaving the "running" session status is the turn-end signal: // settle still-running turns so their duration reflects the whole // turn rather than the last assistant message. const settledTurnState = settledTurnStateForSessionStatus(event.payload.session.status); + // Any settled status abandons an unadopted pending turn start — + // including "ready": a mid-turn steer re-arms the pending row + // without a fresh adoption, and a stale row would block queue + // drains (and re-arm the read model's pendingTurnStart flag on + // restart hydration). Ready-with-genuinely-pending starts never + // reach this projection: ingestion maps that shape to + // "starting" before dispatching the session set. + if (settledTurnState !== null) { + yield* projectionTurnRepository.deletePendingTurnStartByThreadId({ + threadId: event.payload.threadId, + }); + } if (settledTurnState === null) { return; } @@ -1541,6 +1619,14 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti name: ORCHESTRATION_PROJECTOR_NAMES.projects, apply: applyProjectsProjection, }, + // queuedMessages must bootstrap before threadMessages: the revert + // handler in threadMessages reads the queued-message projection to + // retain queued attachments, so on replay that table has to be + // populated first or the prune deletes files still referenced. + { + name: ORCHESTRATION_PROJECTOR_NAMES.queuedMessages, + apply: applyQueuedMessagesProjection, + }, { name: ORCHESTRATION_PROJECTOR_NAMES.threadMessages, apply: applyThreadMessagesProjection, @@ -1671,6 +1757,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect( Layer.provideMerge(ProjectionThreadRepositoryLive), Layer.provideMerge(ProjectionThreadMessageRepositoryLive), Layer.provideMerge(ProjectionThreadProposedPlanRepositoryLive), + Layer.provideMerge(ProjectionQueuedMessageRepositoryLive), Layer.provideMerge(ProjectionThreadActivityRepositoryLive), Layer.provideMerge(ProjectionThreadSessionRepositoryLive), Layer.provideMerge(ProjectionTurnRepositoryLive), diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 843ebb3f626..c6bc03b795f 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -325,6 +325,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { updatedAt: "2026-02-24T00:00:05.000Z", }, ], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [ { id: "plan-1", diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 01e8edefa01..8024d7e060c 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -18,6 +18,7 @@ import { type OrchestrationMessage, type OrchestrationProjectShell, type OrchestrationProposedPlan, + type OrchestrationQueuedMessage, type OrchestrationProject, type OrchestrationSession, type OrchestrationThreadActivity, @@ -48,6 +49,7 @@ import { ProjectionState } from "../../persistence/Services/ProjectionState.ts"; import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts"; import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionThreadMessages.ts"; import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts"; +import { ProjectionQueuedMessage } from "../../persistence/Services/ProjectionQueuedMessages.ts"; import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts"; import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts"; import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts"; @@ -77,6 +79,12 @@ const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields( }), ); const ProjectionThreadProposedPlanDbRowSchema = ProjectionThreadProposedPlan; +const ProjectionQueuedMessageDbRowSchema = ProjectionQueuedMessage.mapFields( + Struct.assign({ + attachments: Schema.fromJsonString(Schema.Array(ChatAttachment)), + modelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)), + }), +); const ProjectionThreadDbRowSchema = ProjectionThread.mapFields( Struct.assign({ modelSelection: Schema.fromJsonString(ModelSelection), @@ -298,6 +306,26 @@ function mapThreadActivityRow( return row.sequence !== null ? Object.assign(activity, { sequence: row.sequence }) : activity; } +function mapQueuedMessageRow( + row: Schema.Schema.Type, +): OrchestrationQueuedMessage { + return { + messageId: row.messageId, + text: row.text, + attachments: row.attachments, + ...(row.modelSelection !== null ? { modelSelection: row.modelSelection } : {}), + ...(row.sourceProposedPlanThreadId !== null && row.sourceProposedPlanId !== null + ? { + sourceProposedPlan: { + threadId: row.sourceProposedPlanThreadId, + planId: row.sourceProposedPlanId, + }, + } + : {}), + queuedAt: row.queuedAt, + }; +} + function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { return (cause: unknown): ProjectionRepositoryError => Schema.isSchemaError(cause) @@ -499,6 +527,45 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const listQueuedMessageRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionQueuedMessageDbRowSchema, + execute: () => + sql` + SELECT + message_id AS "messageId", + thread_id AS "threadId", + text, + attachments_json AS "attachments", + model_selection_json AS "modelSelection", + source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", + source_proposed_plan_id AS "sourceProposedPlanId", + queued_at AS "queuedAt" + FROM projection_queued_messages + ORDER BY thread_id ASC, rowid ASC + `, + }); + + const listQueuedMessageRowsByThread = SqlSchema.findAll({ + Request: ThreadIdLookupInput, + Result: ProjectionQueuedMessageDbRowSchema, + execute: ({ threadId }) => + sql` + SELECT + message_id AS "messageId", + thread_id AS "threadId", + text, + attachments_json AS "attachments", + model_selection_json AS "modelSelection", + source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", + source_proposed_plan_id AS "sourceProposedPlanId", + queued_at AS "queuedAt" + FROM projection_queued_messages + WHERE thread_id = ${threadId} + ORDER BY rowid ASC + `, + }); + const listThreadActivityRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadActivityDbRowSchema, @@ -638,6 +705,50 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const PendingTurnStartDbRowSchema = Schema.Struct({ + threadId: ThreadId, + messageId: MessageId, + requestedAt: IsoDateTime, + }); + + const getPendingTurnStartRowByThread = SqlSchema.findOneOption({ + Request: ThreadIdLookupInput, + Result: PendingTurnStartDbRowSchema, + execute: ({ threadId }) => + sql` + SELECT + thread_id AS "threadId", + pending_message_id AS "messageId", + requested_at AS "requestedAt" + FROM projection_turns + WHERE thread_id = ${threadId} + AND turn_id IS NULL + AND state = 'pending' + AND pending_message_id IS NOT NULL + AND checkpoint_turn_count IS NULL + ORDER BY requested_at DESC + LIMIT 1 + `, + }); + + const listPendingTurnStartRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: PendingTurnStartDbRowSchema, + execute: () => + sql` + SELECT + thread_id AS "threadId", + pending_message_id AS "messageId", + requested_at AS "requestedAt" + FROM projection_turns + WHERE turn_id IS NULL + AND state = 'pending' + AND pending_message_id IS NOT NULL + AND checkpoint_turn_count IS NULL + ORDER BY thread_id ASC, requested_at DESC + `, + }); + const listActiveLatestTurnRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionLatestTurnDbRowSchema, @@ -1147,6 +1258,22 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), + listQueuedMessageRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getSnapshot:listQueuedMessages:query", + "ProjectionSnapshotQuery.getSnapshot:listQueuedMessages:decodeRows", + ), + ), + ), + listPendingTurnStartRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getSnapshot:listPendingTurnStarts:query", + "ProjectionSnapshotQuery.getSnapshot:listPendingTurnStarts:decodeRows", + ), + ), + ), listThreadActivityRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -1196,6 +1323,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { threadRows, messageRows, proposedPlanRows, + queuedMessageRows, + pendingTurnStartRows, activityRows, sessionRows, checkpointRows, @@ -1204,6 +1333,20 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ]) => Effect.gen(function* () { const messagesByThread = new Map>(); + const queuedMessagesByThread = new Map>(); + // Rows are ordered requested_at DESC per thread; first wins. + const pendingTurnStartByThread = new Map< + string, + { messageId: MessageId; requestedAt: string } + >(); + for (const row of pendingTurnStartRows) { + if (!pendingTurnStartByThread.has(row.threadId)) { + pendingTurnStartByThread.set(row.threadId, { + messageId: row.messageId, + requestedAt: row.requestedAt, + }); + } + } const proposedPlansByThread = new Map>(); const activitiesByThread = new Map>(); const checkpointsByThread = new Map>(); @@ -1253,6 +1396,13 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { proposedPlansByThread.set(row.threadId, threadProposedPlans); } + for (const row of queuedMessageRows) { + updatedAt = maxIso(updatedAt, row.queuedAt); + const threadQueuedMessages = queuedMessagesByThread.get(row.threadId) ?? []; + threadQueuedMessages.push(mapQueuedMessageRow(row)); + queuedMessagesByThread.set(row.threadId, threadQueuedMessages); + } + for (const row of activityRows) { updatedAt = maxIso(updatedAt, row.createdAt); const threadActivities = activitiesByThread.get(row.threadId) ?? []; @@ -1363,6 +1513,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedAt: row.snoozedAt, deletedAt: row.deletedAt, messages: messagesByThread.get(row.threadId) ?? [], + queuedMessages: queuedMessagesByThread.get(row.threadId) ?? [], + pendingTurnStart: pendingTurnStartByThread.get(row.threadId) ?? null, proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], activities: activitiesByThread.get(row.threadId) ?? [], // The full snapshot is unwindowed, so there is never more to load. @@ -1421,6 +1573,22 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), + listQueuedMessageRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandReadModel:listQueuedMessages:query", + "ProjectionSnapshotQuery.getCommandReadModel:listQueuedMessages:decodeRows", + ), + ), + ), + listPendingTurnStartRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandReadModel:listPendingTurnStarts:query", + "ProjectionSnapshotQuery.getCommandReadModel:listPendingTurnStarts:decodeRows", + ), + ), + ), listThreadSessionRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -1449,7 +1617,16 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ) .pipe( Effect.flatMap( - ([projectRows, threadRows, proposedPlanRows, sessionRows, latestTurnRows, stateRows]) => + ([ + projectRows, + threadRows, + proposedPlanRows, + queuedMessageRows, + pendingTurnStartRows, + sessionRows, + latestTurnRows, + stateRows, + ]) => Effect.sync(() => { let updatedAt: string | null = null; const projects: OrchestrationProject[] = []; @@ -1523,8 +1700,33 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latestTurnByThread.set(row.threadId, mapLatestTurn(row)); } const proposedPlansByThread = new Map>(); + const queuedMessagesByThread = new Map>(); + // Rows are ordered requested_at DESC per thread; first wins. + const pendingTurnStartByThread = new Map< + string, + { messageId: MessageId; requestedAt: string } + >(); + for (const row of pendingTurnStartRows) { + if (!pendingTurnStartByThread.has(row.threadId)) { + pendingTurnStartByThread.set(row.threadId, { + messageId: row.messageId, + requestedAt: row.requestedAt, + }); + } + } const sessionByThread = new Map(); + for (let index = 0; index < queuedMessageRows.length; index += 1) { + const row = queuedMessageRows[index]; + if (!row) { + continue; + } + updatedAt = maxIso(updatedAt, row.queuedAt); + const threadQueuedMessages = queuedMessagesByThread.get(row.threadId) ?? []; + threadQueuedMessages.push(mapQueuedMessageRow(row)); + queuedMessagesByThread.set(row.threadId, threadQueuedMessages); + } + for (let index = 0; index < sessionRows.length; index += 1) { const row = sessionRows[index]; if (!row) { @@ -1567,6 +1769,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedAt: row.snoozedAt, deletedAt: row.deletedAt, messages: [], + queuedMessages: queuedMessagesByThread.get(row.threadId) ?? [], + pendingTurnStart: pendingTurnStartByThread.get(row.threadId) ?? null, proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], activities: [], checkpoints: [], @@ -2125,6 +2329,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { threadRow, messageRows, proposedPlanRows, + queuedMessageRows, + pendingTurnStartRow, activityRows, checkpointRows, latestTurnRow, @@ -2154,6 +2360,22 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), + listQueuedMessageRowsByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadDetailById:listQueuedMessages:query", + "ProjectionSnapshotQuery.getThreadDetailById:listQueuedMessages:decodeRows", + ), + ), + ), + getPendingTurnStartRowByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadDetailById:getPendingTurnStart:query", + "ProjectionSnapshotQuery.getThreadDetailById:getPendingTurnStart:decodeRow", + ), + ), + ), listThreadActivityRowsByThread({ threadId }).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -2225,6 +2447,13 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { } return message; }), + queuedMessages: queuedMessageRows.map(mapQueuedMessageRow), + pendingTurnStart: Option.isSome(pendingTurnStartRow) + ? { + messageId: pendingTurnStartRow.value.messageId, + requestedAt: pendingTurnStartRow.value.requestedAt, + } + : null, proposedPlans: proposedPlanRows.map(mapProposedPlanRow), // The query fetches WINDOW+1 ascending rows; if it returned the extra // one, older activities exist beyond the window — drop that oldest row diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 516df2d85fc..33d6973398e 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -483,11 +483,61 @@ describe("ProviderCommandReactor", () => { ); } + // Queue-by-default holds turn.start commands while the session is + // starting/running. The mock provider never emits runtime events, so + // tests that send a follow-up on an idle thread must settle the session + // back to ready first — in production, turn completion does this. The + // projected session is preserved apart from status/activeTurnId so + // provider identity fields survive the settle. + const harnessRuntime = runtime; + let settleIndex = 0; + const settleSession = async (threadId: ThreadId = ThreadId.make("thread-1")) => { + settleIndex += 1; + const readModel = await harnessRuntime.runPromise(snapshotQuery.getSnapshot()); + const session = readModel.threads.find((entry) => entry.id === threadId)?.session; + if (!session) { + return; + } + // Mirror the real lifecycle: report the turn running (stamping + // latestTurn so the decider's pending-turn-start guard clears), then + // settle back to ready. A lone ready-set would leave latestTurn null + // and the just-sent message would read as a still-pending start. + await harnessRuntime.runPromise( + engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make(`cmd-session-run-${settleIndex}`), + threadId, + session: { + ...session, + status: "running", + activeTurnId: asTurnId(`turn-settle-${settleIndex}`), + updatedAt: now, + }, + createdAt: now, + }), + ); + await harnessRuntime.runPromise( + engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make(`cmd-session-settle-${settleIndex}`), + threadId, + session: { + ...session, + status: "ready", + activeTurnId: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + }; + return { engine, readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()), readTurns: (threadId: ThreadId) => Effect.runPromise(turnRepository.listByThreadId({ threadId })), + settleSession, startSession, sendTurn, interruptTurn, @@ -1633,6 +1683,7 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.sendTurn.mock.calls.length === 1); + await harness.settleSession(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", @@ -1687,6 +1738,7 @@ describe("ProviderCommandReactor", () => { yield* Effect.promise(() => waitFor(() => harness.sendTurn.mock.calls.length === 1)); + yield* Effect.promise(() => harness.settleSession()); yield* harness.engine.dispatch({ type: "thread.turn.start", commandId: CommandId.make("cmd-turn-start-restricted-2"), @@ -1807,6 +1859,7 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); + await harness.settleSession(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", @@ -1856,6 +1909,7 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.sendTurn.mock.calls.length === 1); + await harness.settleSession(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", @@ -1932,6 +1986,7 @@ describe("ProviderCommandReactor", () => { }), ); + await harness.settleSession(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", @@ -1998,6 +2053,7 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); + await harness.settleSession(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", @@ -2082,6 +2138,7 @@ describe("ProviderCommandReactor", () => { return thread?.runtimeMode === "approval-required"; }); await waitFor(() => harness.startSession.mock.calls.length === 2); + await harness.settleSession(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", @@ -2254,6 +2311,7 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); + await harness.settleSession(); await Effect.runPromise( harness.engine.dispatch({ type: "thread.turn.start", diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 74ece50cd31..b057f5afddb 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1523,22 +1523,35 @@ describe("ProviderRuntimeIngestion", () => { threadId, ); - // The steer: a user-requested turn start while the old turn still runs. + // The steer: a follow-up sent mid-turn queues by default, then the + // explicit queue.steer command dispatches it into the running turn. await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.turn.start", - commandId: CommandId.make("cmd-turn-start-steer"), - threadId, - message: { - messageId: asMessageId("msg-steer"), - role: "user", - text: "actually, do 15 instead", - attachments: [], - }, - interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, - runtimeMode: "approval-required", - createdAt, - }), + harness.engine + .dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-turn-start-steer"), + threadId, + message: { + messageId: asMessageId("msg-steer"), + role: "user", + text: "actually, do 15 instead", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + }) + .pipe( + Effect.andThen( + harness.engine.dispatch({ + type: "thread.queue.steer", + commandId: CommandId.make("cmd-queue-steer"), + threadId, + messageId: asMessageId("msg-steer"), + createdAt, + }), + ), + ), ); // The provider session tracks the new turn before emitting turn.started diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index a8a51b30260..d901340e3ac 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -30,6 +30,8 @@ import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts"; import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; +import { ProjectionQueuedMessageRepository } from "../../persistence/Services/ProjectionQueuedMessages.ts"; +import { ProjectionQueuedMessageRepositoryLive } from "../../persistence/Layers/ProjectionQueuedMessages.ts"; import { isGitRepository } from "../../git/Utils.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; @@ -692,6 +694,7 @@ const make = Effect.gen(function* () { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; const providerService = yield* ProviderService; const projectionTurnRepository = yield* ProjectionTurnRepository; + const projectionQueuedMessageRepository = yield* ProjectionQueuedMessageRepository; const serverSettingsService = yield* ServerSettingsService; const providerCommandId = (event: ProviderRuntimeEvent, tag: string) => crypto.randomUUIDv4.pipe( @@ -746,6 +749,29 @@ const make = Effect.gen(function* () { ), ); + const dispatchQueueDrain = Effect.fn("dispatchQueueDrain")(function* ( + threadId: ThreadId, + event: ProviderRuntimeEvent, + now: string, + ) { + const queuedMessages = yield* projectionQueuedMessageRepository.listByThreadId({ threadId }); + if (queuedMessages.length === 0) { + return; + } + yield* orchestrationEngine + .dispatch({ + type: "thread.queue.drain", + commandId: yield* providerCommandId(event, "queue-drain"), + threadId, + createdAt: now, + }) + .pipe( + // A drain losing the race to a fresh user turn (or an already-empty + // queue) is expected; the next natural completion re-attempts. + Effect.catchTag("OrchestrationCommandInvariantError", () => Effect.void), + ); + }); + const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (threadId: ThreadId) { return yield* projectionSnapshotQuery .getThreadDetailById(threadId) @@ -1313,6 +1339,7 @@ const make = Effect.gen(function* () { }); const hasPendingTurnStart = Option.isSome(pendingTurnStart) && thread.session?.status === "starting"; + let drainQueueAfterTurnFinalize = false; const conflictsWithActiveTurn = activeTurnId !== null && eventTurnId !== undefined && !sameId(activeTurnId, eventTurnId); @@ -1450,6 +1477,15 @@ const make = Effect.gen(function* () { }, createdAt: now, }); + + // Auto-drain queued follow-ups only on natural completion — + // never after an interrupt (the user stopped the agent to take + // control) or a failure. The dispatch happens after the + // assistant-finalization block below so the drained user message + // sequences after the assistant text it follows up on. + drainQueueAfterTurnFinalize = + event.type === "turn.completed" && + normalizeRuntimeTurnState(event.payload.state) === "completed"; } } @@ -1668,6 +1704,10 @@ const make = Effect.gen(function* () { updatedAt: now, }); } + + if (drainQueueAfterTurnFinalize) { + yield* dispatchQueueDrain(thread.id, event, now); + } } if (event.type === "session.exited") { @@ -1828,4 +1868,7 @@ const make = Effect.gen(function* () { export const ProviderRuntimeIngestionLive = Layer.effect( ProviderRuntimeIngestionService, make, -).pipe(Layer.provide(ProjectionTurnRepositoryLive)); +).pipe( + Layer.provide(ProjectionTurnRepositoryLive), + Layer.provide(ProjectionQueuedMessageRepositoryLive), +); diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 3b558d24739..a3b80fc23c5 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -14,6 +14,8 @@ import { ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema, + ThreadMessageQueuedPayload as ContractsThreadMessageQueuedPayloadSchema, + ThreadQueuedMessageRemovedPayload as ContractsThreadQueuedMessageRemovedPayloadSchema, ThreadProposedPlanUpsertedPayload as ContractsThreadProposedPlanUpsertedPayloadSchema, ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema, ThreadTurnDiffCompletedPayload as ContractsThreadTurnDiffCompletedPayloadSchema, @@ -44,6 +46,8 @@ export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema; +export const ThreadMessageQueuedPayload = ContractsThreadMessageQueuedPayloadSchema; +export const ThreadQueuedMessageRemovedPayload = ContractsThreadQueuedMessageRemovedPayloadSchema; export const ThreadProposedPlanUpsertedPayload = ContractsThreadProposedPlanUpsertedPayloadSchema; export const ThreadSessionSetPayload = ContractsThreadSessionSetPayloadSchema; export const ThreadTurnDiffCompletedPayload = ContractsThreadTurnDiffCompletedPayloadSchema; diff --git a/apps/server/src/orchestration/commandInvariants.test.ts b/apps/server/src/orchestration/commandInvariants.test.ts index 05f6e1694f0..de461a10d65 100644 --- a/apps/server/src/orchestration/commandInvariants.test.ts +++ b/apps/server/src/orchestration/commandInvariants.test.ts @@ -73,6 +73,8 @@ const wireReadModel: OrchestrationReadModel = { settledAt: null, latestTurn: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, session: null, activities: [], proposedPlans: [], @@ -98,6 +100,8 @@ const wireReadModel: OrchestrationReadModel = { settledAt: null, latestTurn: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, session: null, activities: [], proposedPlans: [], diff --git a/apps/server/src/orchestration/commandReadModel.test.ts b/apps/server/src/orchestration/commandReadModel.test.ts index 6cc7aad55ee..575266b51a3 100644 --- a/apps/server/src/orchestration/commandReadModel.test.ts +++ b/apps/server/src/orchestration/commandReadModel.test.ts @@ -47,6 +47,8 @@ function makeThread( hasMoreActivities: false, latestTurn: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, session: null, activities: [], proposedPlans: [], diff --git a/apps/server/src/orchestration/decider.queue.test.ts b/apps/server/src/orchestration/decider.queue.test.ts new file mode 100644 index 00000000000..be110be8c42 --- /dev/null +++ b/apps/server/src/orchestration/decider.queue.test.ts @@ -0,0 +1,487 @@ +import { + CommandId, + DEFAULT_PROVIDER_INTERACTION_MODE, + EventId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationEvent, + type OrchestrationSessionStatus, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; + +import { decideOrchestrationCommand } from "./decider.ts"; +import { findThreadById, type CommandReadModel } from "./commandReadModel.ts"; +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +const NOW = "2026-01-01T00:00:00.000Z"; +const asCommandId = (value: string): CommandId => CommandId.make(value); +const asEventId = (value: string): EventId => EventId.make(value); +const asMessageId = (value: string): MessageId => MessageId.make(value); +const asProjectId = (value: string): ProjectId => ProjectId.make(value); +const asThreadId = (value: string): ThreadId => ThreadId.make(value); + +const THREAD_ID = asThreadId("thread-queue"); + +const seedReadModel = Effect.gen(function* () { + const initial = createEmptyReadModel(NOW); + const withProject = yield* projectEvent(initial, { + sequence: 1, + eventId: asEventId("evt-project-create"), + aggregateKind: "project", + aggregateId: asProjectId("project-queue"), + type: "project.created", + occurredAt: NOW, + commandId: asCommandId("cmd-project-create"), + causationEventId: null, + correlationId: asCommandId("cmd-project-create"), + metadata: {}, + payload: { + projectId: asProjectId("project-queue"), + title: "Project Queue", + workspaceRoot: "/tmp/project-queue", + defaultModelSelection: null, + scripts: [], + createdAt: NOW, + updatedAt: NOW, + }, + }); + + return yield* projectEvent(withProject, { + sequence: 2, + eventId: asEventId("evt-thread-create"), + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.created", + occurredAt: NOW, + commandId: asCommandId("cmd-thread-create"), + causationEventId: null, + correlationId: asCommandId("cmd-thread-create"), + metadata: {}, + payload: { + threadId: THREAD_ID, + projectId: asProjectId("project-queue"), + title: "Thread Queue", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: NOW, + updatedAt: NOW, + }, + }); +}); + +const withSessionStatus = ( + readModel: CommandReadModel, + status: OrchestrationSessionStatus, + sequence: number, +) => + projectEvent(readModel, { + sequence, + eventId: asEventId(`evt-session-${status}-${sequence}`), + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.session-set", + occurredAt: NOW, + commandId: asCommandId(`cmd-session-${status}-${sequence}`), + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { + threadId: THREAD_ID, + session: { + threadId: THREAD_ID, + status, + providerName: "codex", + runtimeMode: "full-access", + activeTurnId: status === "running" ? TurnId.make("turn-active") : null, + lastError: null, + updatedAt: NOW, + }, + }, + }); + +const turnStartCommand = (suffix: string) => + ({ + type: "thread.turn.start", + commandId: asCommandId(`cmd-turn-start-${suffix}`), + threadId: THREAD_ID, + message: { + messageId: asMessageId(`message-${suffix}`), + role: "user", + text: `Follow up ${suffix}`, + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + createdAt: NOW, + }) as const; + +const applyPlanned = ( + readModel: CommandReadModel, + planned: + | Omit + | ReadonlyArray>, +) => + Effect.gen(function* () { + let nextReadModel = readModel; + let nextSequence = readModel.snapshotSequence; + for (const event of Array.isArray(planned) ? planned : [planned]) { + nextSequence += 1; + nextReadModel = yield* projectEvent(nextReadModel, { ...event, sequence: nextSequence }); + } + return nextReadModel; + }); + +it.layer(NodeServices.layer)("decider queue flows", (it) => { + it.effect("starts a turn immediately when the thread is idle", () => + Effect.gen(function* () { + const readModel = yield* seedReadModel; + const planned = yield* decideOrchestrationCommand({ + command: turnStartCommand("idle"), + readModel, + }); + const events = Array.isArray(planned) ? planned : [planned]; + expect(events.map((event) => event.type)).toEqual([ + "thread.message-sent", + "thread.turn-start-requested", + ]); + }), + ); + + it.effect("queues a follow-up while a turn is running", () => + Effect.gen(function* () { + const readModel = yield* withSessionStatus(yield* seedReadModel, "running", 3); + const planned = yield* decideOrchestrationCommand({ + command: turnStartCommand("busy"), + readModel, + }); + const events = Array.isArray(planned) ? planned : [planned]; + expect(events.map((event) => event.type)).toEqual(["thread.message-queued"]); + + const projected = yield* applyPlanned(readModel, planned); + const thread = findThreadById(projected, THREAD_ID); + expect(thread?.queuedMessages.map((entry) => entry.messageId)).toEqual([ + asMessageId("message-busy"), + ]); + }), + ); + + it.effect("queues a follow-up racing in behind a just-dispatched turn start", () => + Effect.gen(function* () { + // First send on an idle thread: dispatches message-sent + + // turn-start-requested, but the provider has not reported any + // session status yet — the pending-start window. + let readModel = yield* seedReadModel; + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("pending-1"), readModel }), + ); + + // Second send in that window must queue, not open a second turn. + const planned = yield* decideOrchestrationCommand({ + command: turnStartCommand("pending-2"), + readModel, + }); + const events = Array.isArray(planned) ? planned : [planned]; + expect(events.map((event) => event.type)).toEqual(["thread.message-queued"]); + + // A drain in the same window is rejected and keeps the queue intact. + readModel = yield* applyPlanned(readModel, planned); + const drainError = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type: "thread.queue.drain", + commandId: asCommandId("cmd-drain-pending"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel, + }), + ); + expect(drainError.message).toContain("busy"); + }), + ); + + it.effect("drains after a mid-turn steer once the turn settles to ready", () => + Effect.gen(function* () { + // Two messages queue while running; the first is steered mid-turn + // (re-arming pendingTurnStart), then the turn completes → ready. + let readModel = yield* withSessionStatus(yield* seedReadModel, "running", 3); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("settle-1"), readModel }), + ); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("settle-2"), readModel }), + ); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ + command: { + type: "thread.queue.steer", + commandId: asCommandId("cmd-steer-settle"), + threadId: THREAD_ID, + messageId: asMessageId("message-settle-1"), + createdAt: NOW, + }, + readModel, + }), + ); + // The steer left pendingTurnStart armed with the session still running. + let thread = findThreadById(readModel, THREAD_ID); + expect(thread?.pendingTurnStart?.messageId).toEqual(asMessageId("message-settle-1")); + + // Turn end: session settles to ready, which must release the pending + // flag so the queued follow-up can drain. + readModel = yield* withSessionStatus(readModel, "ready", readModel.snapshotSequence + 1); + thread = findThreadById(readModel, THREAD_ID); + expect(thread?.pendingTurnStart).toBeNull(); + + const planned = yield* decideOrchestrationCommand({ + command: { + type: "thread.queue.drain", + commandId: asCommandId("cmd-drain-settle"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel, + }); + const events = Array.isArray(planned) ? planned : [planned]; + expect(events.map((event) => event.type)).toEqual([ + "thread.queued-message-removed", + "thread.message-sent", + "thread.turn-start-requested", + ]); + }), + ); + + it.effect("rejects queueing past the per-thread cap", () => + Effect.gen(function* () { + let readModel = yield* withSessionStatus(yield* seedReadModel, "running", 3); + for (let index = 0; index < 50; index += 1) { + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ + command: turnStartCommand(`cap-${index}`), + readModel, + }), + ); + } + + const error = yield* Effect.flip( + decideOrchestrationCommand({ command: turnStartCommand("cap-overflow"), readModel }), + ); + expect(error.message).toContain("queued messages"); + + const thread = findThreadById(readModel, THREAD_ID); + expect(thread?.queuedMessages).toHaveLength(50); + }), + ); + + it.effect("steer dispatches a queued message even while running", () => + Effect.gen(function* () { + let readModel = yield* withSessionStatus(yield* seedReadModel, "running", 3); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("steer"), readModel }), + ); + + const planned = yield* decideOrchestrationCommand({ + command: { + type: "thread.queue.steer", + commandId: asCommandId("cmd-steer"), + threadId: THREAD_ID, + messageId: asMessageId("message-steer"), + createdAt: NOW, + }, + readModel, + }); + const events = Array.isArray(planned) ? planned : [planned]; + expect(events.map((event) => event.type)).toEqual([ + "thread.queued-message-removed", + "thread.message-sent", + "thread.turn-start-requested", + ]); + + const projected = yield* applyPlanned(readModel, planned); + const thread = findThreadById(projected, THREAD_ID); + expect(thread?.queuedMessages).toEqual([]); + expect(thread?.messages.map((entry) => entry.id)).toContain(asMessageId("message-steer")); + }), + ); + + it.effect("steer rejects while the session is still starting", () => + Effect.gen(function* () { + let readModel = yield* withSessionStatus(yield* seedReadModel, "starting", 3); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ + command: turnStartCommand("steer-starting"), + readModel, + }), + ); + + const error = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type: "thread.queue.steer", + commandId: asCommandId("cmd-steer-starting"), + threadId: THREAD_ID, + messageId: asMessageId("message-steer-starting"), + createdAt: NOW, + }, + readModel, + }), + ); + expect(error.message).toContain("starting"); + + // The queued message survives the rejected steer. + const thread = findThreadById(readModel, THREAD_ID); + expect(thread?.queuedMessages.map((entry) => entry.messageId)).toEqual([ + asMessageId("message-steer-starting"), + ]); + }), + ); + + it.effect("remove deletes a queued message without dispatching", () => + Effect.gen(function* () { + let readModel = yield* withSessionStatus(yield* seedReadModel, "running", 3); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("remove"), readModel }), + ); + + const planned = yield* decideOrchestrationCommand({ + command: { + type: "thread.queue.remove", + commandId: asCommandId("cmd-remove"), + threadId: THREAD_ID, + messageId: asMessageId("message-remove"), + createdAt: NOW, + }, + readModel, + }); + const events = Array.isArray(planned) ? planned : [planned]; + expect(events.map((event) => event.type)).toEqual(["thread.queued-message-removed"]); + + const projected = yield* applyPlanned(readModel, planned); + const thread = findThreadById(projected, THREAD_ID); + expect(thread?.queuedMessages).toEqual([]); + expect(thread?.messages.map((entry) => entry.id)).not.toContain( + asMessageId("message-remove"), + ); + }), + ); + + it.effect("steer and remove reject unknown queued messages", () => + Effect.gen(function* () { + const readModel = yield* seedReadModel; + for (const type of ["thread.queue.steer", "thread.queue.remove"] as const) { + const error = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type, + commandId: asCommandId(`cmd-${type}-missing`), + threadId: THREAD_ID, + messageId: asMessageId("message-missing"), + createdAt: NOW, + }, + readModel, + }), + ); + expect(error.message).toContain("does not exist"); + } + }), + ); + + it.effect("drain dispatches the queue head once the thread is idle", () => + Effect.gen(function* () { + let readModel = yield* withSessionStatus(yield* seedReadModel, "running", 3); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("drain-1"), readModel }), + ); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("drain-2"), readModel }), + ); + readModel = yield* withSessionStatus(readModel, "ready", readModel.snapshotSequence + 1); + + const planned = yield* decideOrchestrationCommand({ + command: { + type: "thread.queue.drain", + commandId: asCommandId("cmd-drain"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel, + }); + const events = Array.isArray(planned) ? planned : [planned]; + expect(events.map((event) => event.type)).toEqual([ + "thread.queued-message-removed", + "thread.message-sent", + "thread.turn-start-requested", + ]); + + const projected = yield* applyPlanned(readModel, planned); + const thread = findThreadById(projected, THREAD_ID); + // FIFO: the first queued message dispatches, the second stays queued. + expect(thread?.messages.map((entry) => entry.id)).toContain(asMessageId("message-drain-1")); + expect(thread?.queuedMessages.map((entry) => entry.messageId)).toEqual([ + asMessageId("message-drain-2"), + ]); + }), + ); + + it.effect("drain rejects while the thread is busy and keeps the queue intact", () => + Effect.gen(function* () { + let readModel = yield* withSessionStatus(yield* seedReadModel, "running", 3); + readModel = yield* applyPlanned( + readModel, + yield* decideOrchestrationCommand({ command: turnStartCommand("drain-busy"), readModel }), + ); + + const error = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type: "thread.queue.drain", + commandId: asCommandId("cmd-drain-busy"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel, + }), + ); + expect(error.message).toContain("busy"); + }), + ); + + it.effect("drain rejects when the queue is empty", () => + Effect.gen(function* () { + const readModel = yield* seedReadModel; + const error = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type: "thread.queue.drain", + commandId: asCommandId("cmd-drain-empty"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel, + }), + ); + expect(error.message).toContain("no queued messages"); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.settled.test.ts b/apps/server/src/orchestration/decider.settled.test.ts index 8debbf5d05c..471730a1e34 100644 --- a/apps/server/src/orchestration/decider.settled.test.ts +++ b/apps/server/src/orchestration/decider.settled.test.ts @@ -46,6 +46,8 @@ function makeReadModel( settledAt: settledOverride === "settled" ? SETTLED_AT : null, deletedAt: null, messages, + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities, checkpoints: [], diff --git a/apps/server/src/orchestration/decider.snoozed.test.ts b/apps/server/src/orchestration/decider.snoozed.test.ts index 4ca22995e00..bf682d69f2d 100644 --- a/apps/server/src/orchestration/decider.snoozed.test.ts +++ b/apps/server/src/orchestration/decider.snoozed.test.ts @@ -51,6 +51,8 @@ function makeReadModel(input: { snoozedAt: input.snoozedAt ?? (input.snoozedUntil != null ? SNOOZED_AT : null), deletedAt: null, messages: input.messages ?? [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: input.activities ?? [], checkpoints: [], diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 38cc6d63808..ca7422f786e 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1,10 +1,16 @@ -import { EventId, type OrchestrationCommand, type OrchestrationEvent } from "@t3tools/contracts"; +import { + EventId, + type OrchestrationCommand, + type OrchestrationEvent, + type OrchestrationQueuedMessage, + type OrchestrationThread, +} from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import type * as PlatformError from "effect/PlatformError"; -import type { CommandReadModel } from "./commandReadModel.ts"; +import { findThreadById, type CommandReadModel } from "./commandReadModel.ts"; import { OrchestrationCommandInvariantError } from "./Errors.ts"; import { listThreadsByProjectId, @@ -174,6 +180,232 @@ type DecideOrchestrationCommandResult = | PlannedOrchestrationEvent | ReadonlyArray; +/** + * A turn is considered active while its session is starting or running. + * Follow-up `thread.turn.start` commands arriving in that window are queued + * instead of dispatched; explicit `thread.queue.steer` bypasses the queue. + */ +function isThreadTurnActive(thread: OrchestrationThread): boolean { + const status = thread.session?.status; + return status === "running" || status === "starting"; +} + +/** + * Settle-guard heuristic: the latest user message is newer than any turn + * activity, i.e. a turn start that no session has picked up yet. Message + * timestamps are client-supplied, so queue/dispatch decisions must NOT + * hang off this (see `pendingTurnStart` on the thread for the event-driven + * signal); settle tolerates the skew — it merely delays settling within + * the grace window. + */ +function hasRecentUnadoptedUserMessage(thread: OrchestrationThread, occurredAt: string): boolean { + const latestUserMessageAtMs = thread.messages.reduce( + (latest, message) => + message.role === "user" ? Math.max(latest, Date.parse(message.createdAt)) : latest, + Number.NEGATIVE_INFINITY, + ); + const latestTurnAtMs = + thread.latestTurn === null + ? Number.NEGATIVE_INFINITY + : Math.max( + ...[ + thread.latestTurn.requestedAt, + thread.latestTurn.startedAt, + thread.latestTurn.completedAt, + ].map((candidate) => + candidate == null ? Number.NEGATIVE_INFINITY : Date.parse(candidate), + ), + ); + const queuedAgeMs = Date.parse(occurredAt) - latestUserMessageAtMs; + return ( + thread.session?.status !== "error" && + Number.isFinite(latestUserMessageAtMs) && + latestUserMessageAtMs > latestTurnAtMs && + Math.abs(queuedAgeMs) <= QUEUED_TURN_START_GRACE_MS + ); +} + +/** + * Enforced here in the decider so a `thread.message-queued` event past the + * cap is never emitted — projections and persistence stay in lockstep with + * the event stream instead of each clamping independently. + */ +const MAX_THREAD_QUEUED_MESSAGES = 50; + +interface TurnStartMessageInput { + readonly messageId: OrchestrationQueuedMessage["messageId"]; + readonly text: string; + readonly attachments: OrchestrationQueuedMessage["attachments"]; + readonly modelSelection?: OrchestrationQueuedMessage["modelSelection"]; + readonly titleSeed?: string; + readonly sourceProposedPlan?: OrchestrationQueuedMessage["sourceProposedPlan"]; +} + +/** + * Plan the `thread.message-sent` + `thread.turn-start-requested` event pair + * shared by immediate turn starts, queued-message steers, and queue drains. + */ +const planTurnStartEvents = Effect.fn("planTurnStartEvents")(function* ({ + commandId, + thread, + message, + occurredAt, +}: { + readonly commandId: OrchestrationCommand["commandId"]; + readonly thread: OrchestrationThread; + readonly message: TurnStartMessageInput; + readonly occurredAt: string; +}): Effect.fn.Return< + ReadonlyArray, + PlatformError.PlatformError, + Crypto.Crypto +> { + const userMessageEvent: PlannedOrchestrationEvent = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: thread.id, + occurredAt, + commandId, + })), + type: "thread.message-sent", + payload: { + threadId: thread.id, + messageId: message.messageId, + role: "user", + text: message.text, + attachments: message.attachments, + turnId: null, + streaming: false, + createdAt: occurredAt, + updatedAt: occurredAt, + }, + }; + const turnStartRequestedEvent: PlannedOrchestrationEvent = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: thread.id, + occurredAt, + commandId, + })), + causationEventId: userMessageEvent.eventId, + type: "thread.turn-start-requested", + payload: { + threadId: thread.id, + messageId: message.messageId, + ...(message.modelSelection !== undefined ? { modelSelection: message.modelSelection } : {}), + ...(message.titleSeed !== undefined ? { titleSeed: message.titleSeed } : {}), + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + ...(message.sourceProposedPlan !== undefined + ? { sourceProposedPlan: message.sourceProposedPlan } + : {}), + createdAt: occurredAt, + }, + }; + // Real activity resets lifecycle overrides. It wakes an explicitly + // settled thread, clears a keep-active pin, and spends a snooze return + // ticket when the user re-engages. + const lifecycleResetEvents: Array = []; + if (thread.settledOverride !== null) { + lifecycleResetEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: thread.id, + occurredAt, + commandId, + })), + type: "thread.unsettled", + payload: { + threadId: thread.id, + reason: "activity", + updatedAt: occurredAt, + }, + }); + } + if (thread.snoozedUntil !== null) { + lifecycleResetEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: thread.id, + occurredAt, + commandId, + })), + type: "thread.unsnoozed", + payload: { + threadId: thread.id, + reason: "activity", + updatedAt: occurredAt, + }, + }); + } + return [...lifecycleResetEvents, userMessageEvent, turnStartRequestedEvent]; +}); + +/** + * Plan the dispatch of one queued message: remove it from the queue and + * start (or steer into) a turn with it. `sourceProposedPlan` is dropped + * rather than failed when the referenced plan no longer exists — the + * message itself must still send. + */ +const planQueuedMessageDispatch = Effect.fn("planQueuedMessageDispatch")(function* ({ + commandId, + readModel, + thread, + queuedMessage, + occurredAt, +}: { + readonly commandId: OrchestrationCommand["commandId"]; + readonly readModel: CommandReadModel; + readonly thread: OrchestrationThread; + readonly queuedMessage: OrchestrationQueuedMessage; + readonly occurredAt: string; +}): Effect.fn.Return< + ReadonlyArray, + PlatformError.PlatformError, + Crypto.Crypto +> { + const sourceProposedPlan = queuedMessage.sourceProposedPlan; + const sourceThread = sourceProposedPlan + ? findThreadById(readModel, sourceProposedPlan.threadId) + : undefined; + const sourcePlanStillValid = + sourceProposedPlan !== undefined && + sourceThread !== undefined && + sourceThread.projectId === thread.projectId && + sourceThread.proposedPlans.some((entry) => entry.id === sourceProposedPlan.planId); + + const removedEvent: PlannedOrchestrationEvent = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: thread.id, + occurredAt, + commandId, + })), + type: "thread.queued-message-removed", + payload: { + threadId: thread.id, + messageId: queuedMessage.messageId, + reason: "dispatched", + removedAt: occurredAt, + }, + }; + const turnStartEvents = yield* planTurnStartEvents({ + commandId, + thread, + message: { + messageId: queuedMessage.messageId, + text: queuedMessage.text, + attachments: queuedMessage.attachments, + ...(queuedMessage.modelSelection !== undefined + ? { modelSelection: queuedMessage.modelSelection } + : {}), + ...(sourcePlanStillValid ? { sourceProposedPlan } : {}), + }, + occurredAt, + }); + return [removedEvent, ...turnStartEvents]; +}); + const decideCommandSequence = Effect.fn("decideCommandSequence")(function* ({ commands, readModel, @@ -736,87 +968,169 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" detail: `Proposed plan '${sourceProposedPlan?.planId}' belongs to thread '${sourceThread.id}' in a different project.`, }); } - const userMessageEvent: Omit = { - ...(yield* withEventBase({ - aggregateKind: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - })), - type: "thread.message-sent", - payload: { - threadId: command.threadId, + // Queue-by-default: a follow-up arriving while a turn is active is + // held server-side and drained on natural turn completion. Explicit + // mid-turn injection goes through `thread.queue.steer`. Bootstrap + // starts are exempt — their thread was created in the same dispatch. + // `pendingTurnStart` covers the window where a turn start was just + // dispatched (by a send or a queue drain) but the provider has not + // reported the session status yet — a send in that gap must queue, + // not open a second concurrent turn ahead of already-queued chips. + if ( + command.bootstrap === undefined && + (isThreadTurnActive(targetThread) || targetThread.pendingTurnStart !== null) + ) { + if (targetThread.queuedMessages.length >= MAX_THREAD_QUEUED_MESSAGES) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${command.threadId}' already has ${MAX_THREAD_QUEUED_MESSAGES} queued messages.`, + }); + } + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.message-queued", + payload: { + threadId: command.threadId, + messageId: command.message.messageId, + text: command.message.text, + attachments: command.message.attachments, + ...(command.modelSelection !== undefined + ? { modelSelection: command.modelSelection } + : {}), + ...(sourceProposedPlan !== undefined ? { sourceProposedPlan } : {}), + queuedAt: command.createdAt, + }, + }; + } + + return yield* planTurnStartEvents({ + commandId: command.commandId, + thread: targetThread, + message: { messageId: command.message.messageId, - role: "user", text: command.message.text, attachments: command.message.attachments, - turnId: null, - streaming: false, - createdAt: command.createdAt, - updatedAt: command.createdAt, + ...(command.modelSelection !== undefined + ? { modelSelection: command.modelSelection } + : {}), + ...(command.titleSeed !== undefined ? { titleSeed: command.titleSeed } : {}), + ...(sourceProposedPlan !== undefined ? { sourceProposedPlan } : {}), }, - }; - const turnStartRequestedEvent: Omit = { + occurredAt: command.createdAt, + }); + } + + case "thread.queue.steer": { + const thread = yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + const queuedMessage = thread.queuedMessages.find( + (entry) => entry.messageId === command.messageId, + ); + if (!queuedMessage) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Queued message '${command.messageId}' does not exist on thread '${command.threadId}'.`, + }); + } + // While a turn start is still pending there is no provider turn to + // steer into — dispatching now would race the pending turn/start with + // a second one. The message stays queued; steer again once running. + // A session that already tracks an active turn (running, or a + // provider reconnect mid-turn) is steerable. Rejected shapes: starting + // with no active turn, and any session without an active turn while a + // just-dispatched turn start is still awaiting adoption — including + // "running" with a null activeTurnId (provider waiting). + const steerRacesPendingStart = + (thread.session?.status === "starting" && thread.session.activeTurnId === null) || + (thread.pendingTurnStart !== null && (thread.session?.activeTurnId ?? null) === null); + if (steerRacesPendingStart) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${command.threadId}' is still starting a turn; steer once it is running.`, + }); + } + // Otherwise dispatch unconditionally: with a running turn the provider + // adapters treat the resulting sendTurn as a steer; on an idle thread + // this degrades to a normal turn start. + return yield* planQueuedMessageDispatch({ + commandId: command.commandId, + readModel, + thread, + queuedMessage, + occurredAt: command.createdAt, + }); + } + + case "thread.queue.remove": { + const thread = yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + const queuedMessage = thread.queuedMessages.find( + (entry) => entry.messageId === command.messageId, + ); + if (!queuedMessage) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Queued message '${command.messageId}' does not exist on thread '${command.threadId}'.`, + }); + } + return { ...(yield* withEventBase({ aggregateKind: "thread", aggregateId: command.threadId, occurredAt: command.createdAt, commandId: command.commandId, })), - causationEventId: userMessageEvent.eventId, - type: "thread.turn-start-requested", + type: "thread.queued-message-removed", payload: { threadId: command.threadId, - messageId: command.message.messageId, - ...(command.modelSelection !== undefined - ? { modelSelection: command.modelSelection } - : {}), - ...(command.titleSeed !== undefined ? { titleSeed: command.titleSeed } : {}), - runtimeMode: targetThread.runtimeMode, - interactionMode: targetThread.interactionMode, - ...(sourceProposedPlan !== undefined ? { sourceProposedPlan } : {}), - createdAt: command.createdAt, + messageId: command.messageId, + reason: "user", + removedAt: command.createdAt, }, }; - // 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, - }, + } + + case "thread.queue.drain": { + const thread = yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + const queuedMessage = thread.queuedMessages.at(0); + if (!queuedMessage) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${command.threadId}' has no queued messages to drain.`, }); } - 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, - }, + // The user may have raced a new message in between turn completion and + // this drain; keep the queue intact and let the next completion retry. + // The pending-start check also stops a duplicate drain from double- + // dispatching while the first drained turn's session is still unset. + if (isThreadTurnActive(thread) || thread.pendingTurnStart !== null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${command.threadId}' is busy; queued messages drain on turn completion.`, }); } - return [...lifecycleResetEvents, userMessageEvent, turnStartRequestedEvent]; + return yield* planQueuedMessageDispatch({ + commandId: command.commandId, + readModel, + thread, + queuedMessage, + occurredAt: command.createdAt, + }); } case "thread.turn.interrupt": { diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 81fe043f54f..11c8e21b376 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -108,6 +108,8 @@ describe("orchestration projector", () => { snoozedAt: null, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index d1dfe8ea070..a5a39ee0081 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -19,6 +19,8 @@ import { import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts"; import { MessageSentPayloadSchema, + ThreadMessageQueuedPayload, + ThreadQueuedMessageRemovedPayload, ProjectCreatedPayload, ProjectDeletedPayload, ProjectMetaUpdatedPayload, @@ -38,6 +40,7 @@ import { ThreadRevertedPayload, ThreadSessionSetPayload, ThreadTurnDiffCompletedPayload, + ThreadTurnStartRequestedPayload, } from "./Schemas.ts"; type ThreadPatch = Partial>; @@ -312,6 +315,8 @@ export function projectEvent( snoozedAt: null, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, activities: [], checkpoints: [], session: null, @@ -517,6 +522,93 @@ export function projectEvent( }; }); + case "thread.message-queued": + return decodeForEvent(ThreadMessageQueuedPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => { + const thread = findThreadById(nextBase, payload.threadId); + if (!thread) { + return nextBase; + } + + // No cap here: the decider refuses to emit `thread.message-queued` + // past its queue limit, so replaying the event stream stays in + // lockstep with the persisted projection. + const queuedMessages = [ + ...thread.queuedMessages.filter((entry) => entry.messageId !== payload.messageId), + { + messageId: payload.messageId, + text: payload.text, + attachments: payload.attachments, + ...(payload.modelSelection !== undefined + ? { modelSelection: payload.modelSelection } + : {}), + ...(payload.sourceProposedPlan !== undefined + ? { sourceProposedPlan: payload.sourceProposedPlan } + : {}), + queuedAt: payload.queuedAt, + }, + ]; + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + queuedMessages, + updatedAt: event.occurredAt, + }), + }; + }), + ); + + case "thread.queued-message-removed": + return decodeForEvent( + ThreadQueuedMessageRemovedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = findThreadById(nextBase, payload.threadId); + if (!thread) { + return nextBase; + } + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + queuedMessages: thread.queuedMessages.filter( + (entry) => entry.messageId !== payload.messageId, + ), + updatedAt: event.occurredAt, + }), + }; + }), + ); + + case "thread.turn-start-requested": + return decodeForEvent( + ThreadTurnStartRequestedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = findThreadById(nextBase, payload.threadId); + if (!thread) { + return nextBase; + } + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + pendingTurnStart: { + messageId: payload.messageId, + requestedAt: payload.createdAt, + }, + updatedAt: event.occurredAt, + }), + }; + }), + ); + case "thread.session-set": return Effect.gen(function* () { const payload = yield* decodeForEvent( @@ -540,10 +632,23 @@ export function projectEvent( // Leaving the "running" session status is the turn-end signal: settle // a still-running latest turn so its duration reflects the whole turn. const settledTurnState = settledTurnStateForSessionStatus(session.status); + // Mirrors the SQL pipeline's pending-turn-start clearing: a running + // session with an active turn adopts the pending start; every settled + // or terminal status (ready/idle/error/stopped/interrupted) clears it + // — a mid-turn steer re-arms the flag without any adopting session + // transition, so the turn-end ready must release it or drains would + // be rejected forever. Only "starting" (and running while waiting on + // a turn id) keeps it pending. + const pendingTurnStart = + (session.status === "running" && session.activeTurnId !== null) || + settledTurnStateForSessionStatus(session.status) !== null + ? null + : thread.pendingTurnStart; return { ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { session, + pendingTurnStart, latestTurn: session.status === "running" && session.activeTurnId !== null ? { diff --git a/apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts b/apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts new file mode 100644 index 00000000000..0e3d20e948d --- /dev/null +++ b/apps/server/src/persistence/Layers/ProjectionQueuedMessages.ts @@ -0,0 +1,131 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Struct from "effect/Struct"; +import { ChatAttachment, ModelSelection } from "@t3tools/contracts"; + +import { toPersistenceSqlError } from "../Errors.ts"; +import { + DeleteProjectionQueuedMessageInput, + DeleteProjectionQueuedMessagesInput, + ListProjectionQueuedMessagesInput, + ProjectionQueuedMessage, + ProjectionQueuedMessageRepository, + type ProjectionQueuedMessageRepositoryShape, +} from "../Services/ProjectionQueuedMessages.ts"; + +const ProjectionQueuedMessageDbRowSchema = ProjectionQueuedMessage.mapFields( + Struct.assign({ + attachments: Schema.fromJsonString(Schema.Array(ChatAttachment)), + modelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)), + }), +); + +const makeProjectionQueuedMessageRepository = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // Replace-then-insert (not ON CONFLICT UPDATE) so a re-queued messageId + // gets a fresh rowid and moves to the queue tail — the same semantics as + // the in-memory projector. Drain order is rowid order: a monotonic + // insertion sequence that never ties, unlike queued_at timestamps. + const upsertProjectionQueuedMessageRow = SqlSchema.void({ + Request: ProjectionQueuedMessage, + execute: (row) => sql` + INSERT OR REPLACE INTO projection_queued_messages ( + message_id, + thread_id, + text, + attachments_json, + model_selection_json, + source_proposed_plan_thread_id, + source_proposed_plan_id, + queued_at + ) + VALUES ( + ${row.messageId}, + ${row.threadId}, + ${row.text}, + ${JSON.stringify(row.attachments)}, + ${row.modelSelection !== null ? JSON.stringify(row.modelSelection) : null}, + ${row.sourceProposedPlanThreadId}, + ${row.sourceProposedPlanId}, + ${row.queuedAt} + ) + `, + }); + + const listProjectionQueuedMessageRows = SqlSchema.findAll({ + Request: ListProjectionQueuedMessagesInput, + Result: ProjectionQueuedMessageDbRowSchema, + execute: ({ threadId }) => sql` + SELECT + message_id AS "messageId", + thread_id AS "threadId", + text, + attachments_json AS "attachments", + model_selection_json AS "modelSelection", + source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", + source_proposed_plan_id AS "sourceProposedPlanId", + queued_at AS "queuedAt" + FROM projection_queued_messages + WHERE thread_id = ${threadId} + ORDER BY rowid ASC + `, + }); + + const deleteProjectionQueuedMessageRow = SqlSchema.void({ + Request: DeleteProjectionQueuedMessageInput, + execute: ({ threadId, messageId }) => sql` + DELETE FROM projection_queued_messages + WHERE thread_id = ${threadId} AND message_id = ${messageId} + `, + }); + + const deleteProjectionQueuedMessageRows = SqlSchema.void({ + Request: DeleteProjectionQueuedMessagesInput, + execute: ({ threadId }) => sql` + DELETE FROM projection_queued_messages + WHERE thread_id = ${threadId} + `, + }); + + const upsert: ProjectionQueuedMessageRepositoryShape["upsert"] = (row) => + upsertProjectionQueuedMessageRow(row).pipe( + Effect.mapError(toPersistenceSqlError("ProjectionQueuedMessageRepository.upsert:query")), + ); + + const listByThreadId: ProjectionQueuedMessageRepositoryShape["listByThreadId"] = (input) => + listProjectionQueuedMessageRows(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionQueuedMessageRepository.listByThreadId:query"), + ), + ); + + const deleteByMessageId: ProjectionQueuedMessageRepositoryShape["deleteByMessageId"] = (input) => + deleteProjectionQueuedMessageRow(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionQueuedMessageRepository.deleteByMessageId:query"), + ), + ); + + const deleteByThreadId: ProjectionQueuedMessageRepositoryShape["deleteByThreadId"] = (input) => + deleteProjectionQueuedMessageRows(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionQueuedMessageRepository.deleteByThreadId:query"), + ), + ); + + return { + upsert, + listByThreadId, + deleteByMessageId, + deleteByThreadId, + } satisfies ProjectionQueuedMessageRepositoryShape; +}); + +export const ProjectionQueuedMessageRepositoryLive = Layer.effect( + ProjectionQueuedMessageRepository, + makeProjectionQueuedMessageRepository, +); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index d25895671a9..7fb29b54ea1 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -47,6 +47,7 @@ import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; import Migration0033 from "./Migrations/033_ProjectionThreadsSettled.ts"; import Migration0034 from "./Migrations/034_ProjectionThreadsSnoozed.ts"; +import Migration0035 from "./Migrations/035_ProjectionQueuedMessages.ts"; /** * Migration loader with all migrations defined inline. @@ -93,6 +94,7 @@ export const migrationEntries = [ [32, "AuthPairingProofKeyThumbprint", Migration0032], [33, "ProjectionThreadsSettled", Migration0033], [34, "ProjectionThreadsSnoozed", Migration0034], + [35, "ProjectionQueuedMessages", Migration0035], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/035_ProjectionQueuedMessages.ts b/apps/server/src/persistence/Migrations/035_ProjectionQueuedMessages.ts new file mode 100644 index 00000000000..33135eab60e --- /dev/null +++ b/apps/server/src/persistence/Migrations/035_ProjectionQueuedMessages.ts @@ -0,0 +1,26 @@ +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` + CREATE TABLE IF NOT EXISTS projection_queued_messages ( + message_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + text TEXT NOT NULL, + attachments_json TEXT NOT NULL, + model_selection_json TEXT, + source_proposed_plan_thread_id TEXT, + source_proposed_plan_id TEXT, + queued_at TEXT NOT NULL + ) + `; + + // Drain order is rowid (insertion) order; the index only serves the + // per-thread filter. + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_queued_messages_thread + ON projection_queued_messages(thread_id) + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionQueuedMessages.ts b/apps/server/src/persistence/Services/ProjectionQueuedMessages.ts new file mode 100644 index 00000000000..6a884dd7b57 --- /dev/null +++ b/apps/server/src/persistence/Services/ProjectionQueuedMessages.ts @@ -0,0 +1,61 @@ +import { + ChatAttachment, + IsoDateTime, + MessageId, + ModelSelection, + OrchestrationProposedPlanId, + ThreadId, +} from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; + +import type { ProjectionRepositoryError } from "../Errors.ts"; + +export const ProjectionQueuedMessage = Schema.Struct({ + messageId: MessageId, + threadId: ThreadId, + text: Schema.String, + attachments: Schema.Array(ChatAttachment), + modelSelection: Schema.NullOr(ModelSelection), + sourceProposedPlanThreadId: Schema.NullOr(ThreadId), + sourceProposedPlanId: Schema.NullOr(OrchestrationProposedPlanId), + queuedAt: IsoDateTime, +}); +export type ProjectionQueuedMessage = typeof ProjectionQueuedMessage.Type; + +export const ListProjectionQueuedMessagesInput = Schema.Struct({ + threadId: ThreadId, +}); +export type ListProjectionQueuedMessagesInput = typeof ListProjectionQueuedMessagesInput.Type; + +export const DeleteProjectionQueuedMessageInput = Schema.Struct({ + threadId: ThreadId, + messageId: MessageId, +}); +export type DeleteProjectionQueuedMessageInput = typeof DeleteProjectionQueuedMessageInput.Type; + +export const DeleteProjectionQueuedMessagesInput = Schema.Struct({ + threadId: ThreadId, +}); +export type DeleteProjectionQueuedMessagesInput = typeof DeleteProjectionQueuedMessagesInput.Type; + +export interface ProjectionQueuedMessageRepositoryShape { + readonly upsert: ( + queuedMessage: ProjectionQueuedMessage, + ) => Effect.Effect; + readonly listByThreadId: ( + input: ListProjectionQueuedMessagesInput, + ) => Effect.Effect, ProjectionRepositoryError>; + readonly deleteByMessageId: ( + input: DeleteProjectionQueuedMessageInput, + ) => Effect.Effect; + readonly deleteByThreadId: ( + input: DeleteProjectionQueuedMessagesInput, + ) => Effect.Effect; +} + +export class ProjectionQueuedMessageRepository extends Context.Service< + ProjectionQueuedMessageRepository, + ProjectionQueuedMessageRepositoryShape +>()("t3/persistence/Services/ProjectionQueuedMessages/ProjectionQueuedMessageRepository") {} diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 67108dd4dbb..fa0403b758d 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -1302,6 +1302,52 @@ export const makeCodexSessionRuntime = ( ...(input.effort ? { effort: input.effort } : {}), ...(input.interactionMode ? { interactionMode: input.interactionMode } : {}), }); + + // Steer the active turn via the protocol-native `turn/steer` — + // appending input mid-turn instead of opening a second provider + // turn while the first is still settling. The `expectedTurnId` + // precondition fails when the turn just ended or is not steerable + // (e.g. review/compact); only a server rejection of the request + // itself (operation "receive-response" — the steer was NOT + // applied) falls back to a fresh `turn/start`. Decode failures on + // an accepted steer's response, and transport/protocol failures, + // propagate: the input may already be appended, and retrying via + // turn/start would duplicate it. + const activeSession = yield* Ref.get(sessionRef); + if (activeSession.status === "running" && activeSession.activeTurnId !== undefined) { + const steeredTurnId = yield* client + .request("turn/steer", { + threadId: providerThreadId, + expectedTurnId: activeSession.activeTurnId, + input: params.input, + }) + .pipe( + Effect.map((steerResponse) => TurnId.make(steerResponse.turnId)), + Effect.catchIf( + (cause): cause is CodexErrors.CodexAppServerRequestError => + cause._tag === "CodexAppServerRequestError" && + cause.operation === "receive-response", + (cause) => + Effect.as( + Effect.logDebug("Codex turn/steer rejected; falling back to turn/start.", { + cause, + }), + null, + ), + ), + ); + if (steeredTurnId !== null) { + const steeredProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); + return { + threadId: options.threadId, + turnId: steeredTurnId, + ...(steeredProviderThreadId + ? { resumeCursor: { threadId: steeredProviderThreadId } } + : {}), + } satisfies ProviderTurnStartResult; + } + } + const rawResponse = yield* client.raw.request("turn/start", params); const response = yield* decodeV2TurnStartResponse(rawResponse).pipe( Effect.mapError((error) => diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index e59521df49d..13198541d0f 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -108,6 +108,7 @@ function makeReadModel( hasActionableProposedPlan: false, latestTurn: null, messages: [], + queuedMessages: [], session: thread.session, activities: [], proposedPlans: [], diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 68a8cb3d683..64fc03661c2 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -170,6 +170,8 @@ const makeDefaultOrchestrationReadModel = () => { settledAt: null, latestTurn: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, session: null, activities: [], proposedPlans: [], @@ -5528,6 +5530,8 @@ it.layer(NodeServices.layer)("server router seam", (it) => { settledAt: null, latestTurn: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, session: null, activities: [], proposedPlans: [], diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 241340af35c..a5b5c04fd12 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -279,6 +279,8 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< { type: | "thread.message-sent" + | "thread.message-queued" + | "thread.queued-message-removed" | "thread.proposed-plan-upserted" | "thread.activity-appended" | "thread.turn-diff-completed" @@ -288,6 +290,8 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< > { return ( event.type === "thread.message-sent" || + event.type === "thread.message-queued" || + event.type === "thread.queued-message-removed" || event.type === "thread.proposed-plan-upserted" || event.type === "thread.activity-appended" || event.type === "thread.turn-diff-completed" || diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 57c12959ffb..acbf93fa613 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -48,6 +48,8 @@ function makeThread(overrides: Partial = {}): Thread { interactionMode: "default", session: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], @@ -457,6 +459,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "ready", latestTurn: completedTurn, latestUserMessageId: localDispatch.latestUserMessageId, + projectedMessageIds: new Set(), session: readySession, hasPendingApproval: false, hasPendingUserInput: false, @@ -483,6 +486,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "ready", latestTurn: newerTurn, latestUserMessageId: localDispatch.latestUserMessageId, + projectedMessageIds: new Set(), session: { ...readySession, updatedAt: newerTurn.completedAt }, hasPendingApproval: false, hasPendingUserInput: false, @@ -510,6 +514,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "running", latestTurn: runningTurn, latestUserMessageId: localDispatch.latestUserMessageId, + projectedMessageIds: new Set(), session: { ...readySession, status: "running", @@ -526,6 +531,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "running", latestTurn: runningTurn, latestUserMessageId: localDispatch.latestUserMessageId, + projectedMessageIds: new Set(), session: { ...readySession, status: "running", @@ -567,18 +573,63 @@ describe("hasServerAcknowledgedLocalDispatch", () => { }), ); + // Dispatch without a known messageId keeps the legacy heuristic: a new + // latest user message acknowledges it. expect( hasServerAcknowledgedLocalDispatch({ localDispatch, phase: "running", latestTurn: runningTurn, latestUserMessageId: MessageId.make("message-steer"), + projectedMessageIds: new Set(), session: runningSession, hasPendingApproval: false, hasPendingUserInput: false, threadError: null, }), ).toBe(true); + + const correlatedDispatch = createLocalDispatchSnapshot( + makeThread({ latestTurn: runningTurn, session: runningSession }), + { messageId: MessageId.make("message-mine") }, + ); + + // A correlated dispatch acknowledges when its exact message appears in + // the timeline or the queue... + for (const projectedMessageIds of [ + new Set(["message-mine"]), + new Set(["message-other", "message-mine"]), + ]) { + expect( + hasServerAcknowledgedLocalDispatch({ + localDispatch: correlatedDispatch, + phase: "running", + latestTurn: runningTurn, + latestUserMessageId: correlatedDispatch.latestUserMessageId, + projectedMessageIds, + session: runningSession, + hasPendingApproval: false, + hasPendingUserInput: false, + threadError: null, + }), + ).toBe(true); + } + + // ...but ignores unrelated queue/timeline changes (another client's + // queued message, or a different queued chip being steered/removed). + expect( + hasServerAcknowledgedLocalDispatch({ + localDispatch: correlatedDispatch, + phase: "running", + latestTurn: runningTurn, + latestUserMessageId: MessageId.make("message-someone-else"), + projectedMessageIds: new Set(["message-someone-else"]), + session: runningSession, + hasPendingApproval: false, + hasPendingUserInput: false, + threadError: null, + }), + ).toBe(false); }); it("acknowledges pending user interaction and errors immediately", () => { @@ -588,6 +639,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "ready" as const, latestTurn: null, latestUserMessageId: localDispatch.latestUserMessageId, + projectedMessageIds: new Set(), session: null, hasPendingApproval: false, hasPendingUserInput: false, diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 466c9b24c87..b46c505ef67 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -69,6 +69,8 @@ export function buildLocalDraftThread( interactionMode: draftThread.interactionMode, session: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, createdAt: draftThread.createdAt, updatedAt: draftThread.createdAt, archivedAt: null, @@ -456,6 +458,13 @@ export async function waitForStartedServerThread( export interface LocalDispatchSnapshot { startedAt: string; preparingWorktree: boolean; + /** + * The messageId of the send this dispatch is waiting on, when the send + * path knows it. Acknowledgment then correlates with this exact message + * being projected (timeline or queue) instead of global tail heuristics + * that other clients' activity could satisfy. + */ + expectedMessageId: ChatMessage["id"] | null; latestUserMessageId: ChatMessage["id"] | null; latestTurnTurnId: TurnId | null; latestTurnRequestedAt: string | null; @@ -467,7 +476,7 @@ export interface LocalDispatchSnapshot { export function createLocalDispatchSnapshot( activeThread: Thread | undefined, - options?: { preparingWorktree?: boolean }, + options?: { preparingWorktree?: boolean; messageId?: ChatMessage["id"] }, ): LocalDispatchSnapshot { const latestTurn = activeThread?.latestTurn ?? null; const session = activeThread?.session ?? null; @@ -475,6 +484,7 @@ export function createLocalDispatchSnapshot( return { startedAt: new Date().toISOString(), preparingWorktree: Boolean(options?.preparingWorktree), + expectedMessageId: options?.messageId ?? null, latestUserMessageId: latestUserMessage?.id ?? null, latestTurnTurnId: latestTurn?.turnId ?? null, latestTurnRequestedAt: latestTurn?.requestedAt ?? null, @@ -490,6 +500,7 @@ export function hasServerAcknowledgedLocalDispatch(input: { phase: SessionPhase; latestTurn: Thread["latestTurn"] | null; latestUserMessageId: ChatMessage["id"] | null; + projectedMessageIds: ReadonlySet; session: Thread["session"] | null; hasPendingApproval: boolean; hasPendingUserInput: boolean; @@ -504,6 +515,7 @@ export function hasServerAcknowledgedLocalDispatch(input: { const latestTurn = input.latestTurn ?? null; const session = input.session ?? null; + const expectedMessageId = input.localDispatch.expectedMessageId; const latestUserMessageChanged = input.localDispatch.latestUserMessageId !== input.latestUserMessageId; const latestTurnChanged = @@ -512,12 +524,18 @@ export function hasServerAcknowledgedLocalDispatch(input: { input.localDispatch.latestTurnStartedAt !== (latestTurn?.startedAt ?? null) || input.localDispatch.latestTurnCompletedAt !== (latestTurn?.completedAt ?? null); + // The dispatched message being projected (timeline or queue) is the + // strongest acknowledgment in every phase: the server also queues sends + // while a session is starting ("connecting" phase), where neither turn + // nor session fields necessarily change. + if (expectedMessageId !== null && input.projectedMessageIds.has(expectedMessageId)) { + return true; + } + if (input.phase === "running") { - // Steering adds a user message to the current running turn without - // necessarily changing any of the turn timestamps. Treat that projected - // message as the server acknowledgment so the composer does not remain - // stuck in its local "Sending" state until the turn settles. - if (latestUserMessageChanged) { + // Dispatches without a known messageId (e.g. plan-implementation flows) + // keep the legacy latest-user-message heuristic. + if (expectedMessageId === null && latestUserMessageChanged) { return true; } if (!latestTurnChanged) { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 57d6b3edd1f..ba1d926e573 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -238,6 +238,7 @@ import { import { ThreadErrorBanner } from "./chat/ThreadErrorBanner"; import { resolveThreadPr } from "./ThreadStatusIndicators"; import { ComposerBannerStack, type ComposerBannerStackItem } from "./chat/ComposerBannerStack"; +import { QueuedMessageChips } from "./chat/QueuedMessageChips"; import { DRAFT_HERO_TRANSITION_ANIMATION_ID, DRAFT_HERO_TRANSITION_DURATION_MS, @@ -496,6 +497,20 @@ function useLocalDispatchState(input: { const [localDispatch, setLocalDispatch] = useState(null); const latestUserMessageId = input.activeThread?.messages.findLast((message) => message.role === "user")?.id ?? null; + const activeThreadMessages = input.activeThread?.messages; + const activeThreadQueuedMessages = input.activeThread?.queuedMessages; + // Every server-projected id for this thread — timeline messages plus the + // held queue — so acknowledgment can match the exact dispatched message. + const projectedMessageIds = useMemo(() => { + const ids = new Set(); + for (const message of activeThreadMessages ?? []) { + ids.add(message.id); + } + for (const queuedMessage of activeThreadQueuedMessages ?? []) { + ids.add(queuedMessage.messageId); + } + return ids; + }, [activeThreadMessages, activeThreadQueuedMessages]); const resetLocalDispatch = useCallback(() => { setLocalDispatch(null); @@ -508,6 +523,7 @@ function useLocalDispatchState(input: { phase: input.phase, latestTurn: input.activeLatestTurn, latestUserMessageId, + projectedMessageIds, session: input.activeThread?.session ?? null, hasPendingApproval: input.activePendingApproval !== null, hasPendingUserInput: input.activePendingUserInput !== null, @@ -521,12 +537,13 @@ function useLocalDispatchState(input: { input.phase, input.threadError, latestUserMessageId, + projectedMessageIds, localDispatch, ], ); const activeLocalDispatch = serverAcknowledgedLocalDispatch ? null : localDispatch; const beginLocalDispatch = useCallback( - (options?: { preparingWorktree?: boolean }) => { + (options?: { preparingWorktree?: boolean; messageId?: MessageId }) => { const preparingWorktree = Boolean(options?.preparingWorktree); setLocalDispatch((current) => { const active = serverAcknowledgedLocalDispatch ? null : current; @@ -1159,6 +1176,12 @@ function ChatViewContent(props: ChatViewProps) { const interruptThreadTurn = useAtomCommand(threadEnvironment.interruptTurn, { reportFailure: false, }); + const steerQueuedThreadMessage = useAtomCommand(threadEnvironment.steerQueuedMessage, { + reportFailure: false, + }); + const removeQueuedThreadMessage = useAtomCommand(threadEnvironment.removeQueuedMessage, { + reportFailure: false, + }); const respondToThreadApproval = useAtomCommand(threadEnvironment.respondToApproval, { reportFailure: false, }); @@ -3926,10 +3949,16 @@ function ChatViewContent(props: ChatViewProps) { useEffect(() => { if (!activeThread?.id) return; - if (activeThread.messages.length === 0) { + if (activeThread.messages.length === 0 && activeThread.queuedMessages.length === 0) { return; } - const serverIds = new Set(activeThread.messages.map((message) => message.id)); + // A queued message is server-acknowledged too — it renders as a chip + // above the composer, so its optimistic timeline copy must go. + const persistedMessageIds = new Set(activeThread.messages.map((message) => message.id)); + const serverIds = new Set([ + ...persistedMessageIds, + ...activeThread.queuedMessages.map((message) => message.messageId), + ]); const removedMessages = optimisticUserMessages.filter((message) => serverIds.has(message.id)); if (removedMessages.length === 0) { return; @@ -3941,7 +3970,10 @@ function ChatViewContent(props: ChatViewProps) { }, 0); for (const removedMessage of removedMessages) { const previewUrls = collectUserMessageBlobPreviewUrls(removedMessage); - if (previewUrls.length > 0) { + // Handoff keeps blob previews alive only for messages entering the + // timeline; a queued-only acknowledgment renders as a text chip, so + // its previews would never be promoted — revoke them instead. + if (previewUrls.length > 0 && persistedMessageIds.has(removedMessage.id)) { handoffAttachmentPreviews(removedMessage.id, previewUrls); continue; } @@ -3950,7 +3982,13 @@ function ChatViewContent(props: ChatViewProps) { return () => { window.clearTimeout(timer); }; - }, [activeThread?.id, activeThread?.messages, handoffAttachmentPreviews, optimisticUserMessages]); + }, [ + activeThread?.id, + activeThread?.messages, + activeThread?.queuedMessages, + handoffAttachmentPreviews, + optimisticUserMessages, + ]); useEffect(() => { setOptimisticUserMessages((existing) => { @@ -4766,7 +4804,11 @@ function ChatViewContent(props: ChatViewProps) { void dockTransition.catch(() => resolveDockStarted?.()); await dockStarted; } - beginLocalDispatch({ preparingWorktree: Boolean(baseBranchForWorktree) }); + const messageIdForSend = newMessageId(); + beginLocalDispatch({ + preparingWorktree: Boolean(baseBranchForWorktree), + messageId: messageIdForSend, + }); const composerImagesSnapshot = [...composerImages]; const composerTerminalContextsSnapshot = [...sendableComposerTerminalContexts]; @@ -4785,7 +4827,6 @@ function ChatViewContent(props: ChatViewProps) { messageTextWithPreviewAnnotations, composerReviewCommentsSnapshot, ); - const messageIdForSend = newMessageId(); const messageCreatedAt = new Date().toISOString(); const outgoingMessageText = formatOutgoingPrompt({ provider: ctxSelectedProvider, @@ -4954,7 +4995,7 @@ function ChatViewContent(props: ChatViewProps) { : {}), } : undefined; - beginLocalDispatch({ preparingWorktree: false }); + beginLocalDispatch({ preparingWorktree: false, messageId: messageIdForSend }); const startResult = await startThreadTurn({ environmentId, input: { @@ -5048,6 +5089,36 @@ function ChatViewContent(props: ChatViewProps) { } }; + const onSteerQueuedMessage = async (messageId: MessageId) => { + if (!activeThread) return; + const result = await steerQueuedThreadMessage({ + environmentId, + input: { threadId: activeThread.id, messageId }, + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + setThreadError( + activeThread.id, + error instanceof Error ? error.message : "Failed to steer the queued message.", + ); + } + }; + + const onRemoveQueuedMessage = async (messageId: MessageId) => { + if (!activeThread) return; + const result = await removeQueuedThreadMessage({ + environmentId, + input: { threadId: activeThread.id, messageId }, + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + setThreadError( + activeThread.id, + error instanceof Error ? error.message : "Failed to remove the queued message.", + ); + } + }; + const onRespondToApproval = useCallback( async (requestId: ApprovalRequestId, decision: ProviderApprovalDecision) => { if (!activeThreadId) return; @@ -5257,7 +5328,7 @@ function ChatViewContent(props: ChatViewProps) { }); sendInFlightRef.current = true; - beginLocalDispatch({ preparingWorktree: false }); + beginLocalDispatch({ preparingWorktree: false, messageId: messageIdForSend }); setThreadError(threadIdForSend, null); // Position this sent row once LegendList has measured the anchored tail. @@ -5991,7 +6062,17 @@ function ChatViewContent(props: ChatViewProps) { ) : ( - + <> + {isServerThread && activeThread ? ( + void onSteerQueuedMessage(messageId)} + onRemove={(messageId) => void onRemoveQueuedMessage(messageId)} + /> + ) : null} + + )}
= {}): Thread { interactionMode: "default", session: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], createdAt: "2026-03-01T00:00:00.000Z", archivedAt: null, diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index e256dba36b7..0195453e76b 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -1177,6 +1177,8 @@ function makeThread(overrides: Partial = {}): Thread { interactionMode: DEFAULT_INTERACTION_MODE, session: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], createdAt: "2026-03-09T10:00:00.000Z", archivedAt: null, @@ -1202,24 +1204,28 @@ describe("getFallbackThreadIdAfterDelete", () => { projectId: ProjectId.make("project-1"), createdAt: "2026-03-09T10:00:00.000Z", messages: [], + queuedMessages: [], }), makeThread({ id: ThreadId.make("thread-active"), projectId: ProjectId.make("project-1"), createdAt: "2026-03-09T10:05:00.000Z", messages: [], + queuedMessages: [], }), makeThread({ id: ThreadId.make("thread-newest"), projectId: ProjectId.make("project-1"), createdAt: "2026-03-09T10:10:00.000Z", messages: [], + queuedMessages: [], }), makeThread({ id: ThreadId.make("thread-other-project"), projectId: ProjectId.make("project-2"), createdAt: "2026-03-09T10:20:00.000Z", messages: [], + queuedMessages: [], }), ], deletedThreadId: ThreadId.make("thread-active"), @@ -1237,18 +1243,21 @@ describe("getFallbackThreadIdAfterDelete", () => { projectId: ProjectId.make("project-1"), createdAt: "2026-03-09T10:05:00.000Z", messages: [], + queuedMessages: [], }), makeThread({ id: ThreadId.make("thread-newest"), projectId: ProjectId.make("project-1"), createdAt: "2026-03-09T10:10:00.000Z", messages: [], + queuedMessages: [], }), makeThread({ id: ThreadId.make("thread-next"), projectId: ProjectId.make("project-1"), createdAt: "2026-03-09T10:07:00.000Z", messages: [], + queuedMessages: [], }), ], deletedThreadId: ThreadId.make("thread-active"), diff --git a/apps/web/src/components/chat/QueuedMessageChips.tsx b/apps/web/src/components/chat/QueuedMessageChips.tsx new file mode 100644 index 00000000000..e158ab5e147 --- /dev/null +++ b/apps/web/src/components/chat/QueuedMessageChips.tsx @@ -0,0 +1,68 @@ +import { memo } from "react"; +import { CornerDownRightIcon, ListEndIcon, Trash2Icon } from "lucide-react"; +import type { MessageId, OrchestrationQueuedMessage } from "@t3tools/contracts"; + +import { Button } from "../ui/button"; + +/** + * Queued follow-up messages held server-side while a turn runs. Each chip + * offers Steer (send now, injecting into the active turn) and delete; the + * queue otherwise auto-drains in order when the turn completes naturally. + */ +export const QueuedMessageChips = memo(function QueuedMessageChips({ + queuedMessages, + disabled, + onSteer, + onRemove, +}: { + readonly queuedMessages: ReadonlyArray; + readonly disabled?: boolean; + readonly onSteer: (messageId: MessageId) => void; + readonly onRemove: (messageId: MessageId) => void; +}) { + if (queuedMessages.length === 0) { + return null; + } + + return ( +
+ {queuedMessages.map((queuedMessage) => ( +
+
+ ))} +
+ ); +}); diff --git a/apps/web/src/lib/threadSort.test.ts b/apps/web/src/lib/threadSort.test.ts index ca9a5986c66..c7f6312c64c 100644 --- a/apps/web/src/lib/threadSort.test.ts +++ b/apps/web/src/lib/threadSort.test.ts @@ -23,6 +23,8 @@ function makeThread(overrides: Partial = {}): Thread { interactionMode: "default", session: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], createdAt: "2026-03-09T10:00:00.000Z", archivedAt: null, @@ -107,6 +109,7 @@ describe("sortThreads", () => { createdAt: "2026-03-09T10:05:00.000Z", updatedAt: "2026-03-09T10:05:00.000Z", messages: [], + queuedMessages: [], }), ], "updated_at", @@ -126,12 +129,14 @@ describe("sortThreads", () => { createdAt: "2026-03-09T10:00:00.000Z", updatedAt: "invalid-date" as never, messages: [], + queuedMessages: [], }), makeThread({ id: ThreadId.make("thread-2"), createdAt: "2026-03-09T09:00:00.000Z", updatedAt: "2026-03-09T09:30:00.000Z", messages: [], + queuedMessages: [], }), ], "updated_at", @@ -151,12 +156,14 @@ describe("sortThreads", () => { createdAt: "invalid-created-at" as never, updatedAt: "invalid-updated-at" as never, messages: [], + queuedMessages: [], }), makeThread({ id: ThreadId.make("thread-2"), createdAt: "invalid-created-at" as never, updatedAt: "invalid-updated-at" as never, messages: [], + queuedMessages: [], }), ], "updated_at", diff --git a/apps/web/src/worktreeCleanup.test.ts b/apps/web/src/worktreeCleanup.test.ts index 89734357889..16b01270da4 100644 --- a/apps/web/src/worktreeCleanup.test.ts +++ b/apps/web/src/worktreeCleanup.test.ts @@ -20,6 +20,8 @@ function makeThread(overrides: Partial = {}): Thread { interactionMode: DEFAULT_INTERACTION_MODE, session: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, checkpoints: [], activities: [], proposedPlans: [], diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index ad25d6544dc..45386bcafb8 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -44,6 +44,8 @@ export type SetThreadRuntimeModeInput = CommandInput<"thread.runtime-mode.set">; export type SetThreadInteractionModeInput = CommandInput<"thread.interaction-mode.set">; export type StartThreadTurnInput = CommandInput<"thread.turn.start">; export type InterruptThreadTurnInput = CommandInput<"thread.turn.interrupt">; +export type SteerQueuedMessageInput = CommandInput<"thread.queue.steer">; +export type RemoveQueuedMessageInput = CommandInput<"thread.queue.remove">; export type RespondToThreadApprovalInput = CommandInput<"thread.approval.respond">; export type RespondToThreadUserInputInput = CommandInput<"thread.user-input.respond">; export type RevertThreadCheckpointInput = CommandInput<"thread.checkpoint.revert">; @@ -254,6 +256,30 @@ export const interruptThreadTurn: (input: InterruptThreadTurnInput) => CommandEf }); }); +export const steerQueuedMessage: (input: SteerQueuedMessageInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.steerQueuedMessage", +)(function* (input) { + const metadata = yield* timestampedCommandMetadata(input); + return yield* dispatch({ + ...input, + type: "thread.queue.steer", + commandId: metadata.commandId, + createdAt: metadata.createdAt, + }); +}); + +export const removeQueuedMessage: (input: RemoveQueuedMessageInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.removeQueuedMessage", +)(function* (input) { + const metadata = yield* timestampedCommandMetadata(input); + return yield* dispatch({ + ...input, + type: "thread.queue.remove", + commandId: metadata.commandId, + createdAt: metadata.createdAt, + }); +}); + export const respondToThreadApproval: (input: RespondToThreadApprovalInput) => CommandEffect = Effect.fn("EnvironmentCommands.respondToThreadApproval")(function* (input) { const metadata = yield* timestampedCommandMetadata(input); diff --git a/packages/client-runtime/src/state/entities.test.ts b/packages/client-runtime/src/state/entities.test.ts index e08fd9e552f..e0179865455 100644 --- a/packages/client-runtime/src/state/entities.test.ts +++ b/packages/client-runtime/src/state/entities.test.ts @@ -207,6 +207,8 @@ describe("environment entity projections", () => { worktreePath: "/repo/stale-worktree", deletedAt: null, messages, + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], @@ -322,6 +324,8 @@ describe("environment entity projections", () => { ...THREAD_SHELL, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 6c128eb01ab..d1444705ba6 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -7,6 +7,7 @@ import { type CreateThreadInput, type DeleteThreadInput, type InterruptThreadTurnInput, + type RemoveQueuedMessageInput, type RespondToThreadApprovalInput, type RespondToThreadUserInputInput, type RevertThreadCheckpointInput, @@ -15,6 +16,7 @@ import { type SettleThreadInput, type SnoozeThreadInput, type StartThreadTurnInput, + type SteerQueuedMessageInput, type StopThreadSessionInput, type UnarchiveThreadInput, type UnsettleThreadInput, @@ -24,6 +26,7 @@ import { createThread, deleteThread, interruptThreadTurn, + removeQueuedMessage, respondToThreadApproval, respondToThreadUserInput, revertThreadCheckpoint, @@ -32,6 +35,7 @@ import { settleThread, snoozeThread, startThreadTurn, + steerQueuedMessage, stopThreadSession, unarchiveThread, unsettleThread, @@ -45,6 +49,7 @@ export type { CreateThreadInput, DeleteThreadInput, InterruptThreadTurnInput, + RemoveQueuedMessageInput, RespondToThreadApprovalInput, RespondToThreadUserInputInput, RevertThreadCheckpointInput, @@ -53,6 +58,7 @@ export type { SettleThreadInput, SnoozeThreadInput, StartThreadTurnInput, + SteerQueuedMessageInput, StopThreadSessionInput, UnarchiveThreadInput, UnsettleThreadInput, @@ -148,6 +154,18 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + steerQueuedMessage: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:steer-queued-message", + execute: (input: SteerQueuedMessageInput) => steerQueuedMessage(input), + scheduler, + concurrency, + }), + removeQueuedMessage: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:remove-queued-message", + execute: (input: RemoveQueuedMessageInput) => removeQueuedMessage(input), + scheduler, + concurrency, + }), respondToApproval: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:respond-to-approval", execute: (input: RespondToThreadApprovalInput) => respondToThreadApproval(input), diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 2df8efd4405..84a7760e56b 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -54,6 +54,8 @@ const baseThread: OrchestrationThread = { settledAt: null, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], @@ -412,6 +414,79 @@ describe("applyThreadDetailEvent", () => { }); }); + describe("thread.message-queued / thread.queued-message-removed", () => { + const queuedPayload = { + threadId: ThreadId.make("thread-1"), + messageId: MessageId.make("queued-1"), + text: "Queued follow-up", + attachments: [], + queuedAt: "2026-04-01T06:00:00.000Z", + }; + + it("appends a queued message and replaces an existing messageId", () => { + const queued = applyThreadDetailEvent(baseThread, { + ...baseEventFields, + sequence: 6, + occurredAt: "2026-04-01T06:00:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.message-queued", + payload: queuedPayload, + }); + + expect(queued.kind).toBe("updated"); + if (queued.kind !== "updated") return; + expect(queued.thread.queuedMessages).toHaveLength(1); + expect(queued.thread.queuedMessages[0]?.text).toBe("Queued follow-up"); + + const replaced = applyThreadDetailEvent(queued.thread, { + ...baseEventFields, + sequence: 7, + occurredAt: "2026-04-01T06:01:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.message-queued", + payload: { ...queuedPayload, text: "Edited follow-up" }, + }); + + expect(replaced.kind).toBe("updated"); + if (replaced.kind !== "updated") return; + expect(replaced.thread.queuedMessages).toHaveLength(1); + expect(replaced.thread.queuedMessages[0]?.text).toBe("Edited follow-up"); + }); + + it("removes a queued message and leaves others intact", () => { + const threadWithQueue: OrchestrationThread = { + ...baseThread, + queuedMessages: [ + { ...queuedPayload, messageId: MessageId.make("queued-1") }, + { ...queuedPayload, messageId: MessageId.make("queued-2") }, + ], + }; + + const result = applyThreadDetailEvent(threadWithQueue, { + ...baseEventFields, + sequence: 8, + occurredAt: "2026-04-01T06:02:00.000Z", + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-1"), + type: "thread.queued-message-removed", + payload: { + threadId: ThreadId.make("thread-1"), + messageId: MessageId.make("queued-1"), + reason: "dispatched", + removedAt: "2026-04-01T06:02:00.000Z", + }, + }); + + expect(result.kind).toBe("updated"); + if (result.kind !== "updated") return; + expect(result.thread.queuedMessages.map((entry) => entry.messageId)).toEqual([ + MessageId.make("queued-2"), + ]); + }); + }); + describe("thread.session-set", () => { it("settles a running latestTurn when the session leaves the running status", () => { const threadWithRunningTurn: OrchestrationThread = { diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index ee6c580418f..d6a543acffb 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -117,6 +117,8 @@ export function applyThreadDetailEvent( snoozedAt: null, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], @@ -236,6 +238,10 @@ export function applyThreadDetailEvent( : {}), runtimeMode: event.payload.runtimeMode, interactionMode: event.payload.interactionMode, + pendingTurnStart: { + messageId: event.payload.messageId, + requestedAt: event.payload.createdAt, + }, updatedAt: event.occurredAt, }, }; @@ -361,6 +367,45 @@ export function applyThreadDetailEvent( }; } + // ── Queued messages ───────────────────────────────────────────── + case "thread.message-queued": { + const queuedMessage = { + messageId: event.payload.messageId, + text: event.payload.text, + attachments: event.payload.attachments, + ...(event.payload.modelSelection !== undefined + ? { modelSelection: event.payload.modelSelection } + : {}), + ...(event.payload.sourceProposedPlan !== undefined + ? { sourceProposedPlan: event.payload.sourceProposedPlan } + : {}), + queuedAt: event.payload.queuedAt, + }; + return { + kind: "updated", + thread: { + ...thread, + queuedMessages: [ + ...thread.queuedMessages.filter((entry) => entry.messageId !== queuedMessage.messageId), + queuedMessage, + ], + updatedAt: event.occurredAt, + }, + }; + } + + case "thread.queued-message-removed": + return { + kind: "updated", + thread: { + ...thread, + queuedMessages: thread.queuedMessages.filter( + (entry) => entry.messageId !== event.payload.messageId, + ), + updatedAt: event.occurredAt, + }, + }; + // ── Session ───────────────────────────────────────────────────── case "thread.session-set": { // Leaving the "running" session status is the turn-end signal: settle a @@ -398,12 +443,25 @@ export function applyThreadDetailEvent( } : thread.latestTurn; + // Mirrors the server projections' pending-turn-start clearing: a + // running session with an active turn adopts it; terminal statuses + // abandon it. + const sessionStatus = event.payload.session.status; + const pendingTurnStart = + (sessionStatus === "running" && event.payload.session.activeTurnId !== null) || + sessionStatus === "error" || + sessionStatus === "stopped" || + sessionStatus === "interrupted" + ? null + : thread.pendingTurnStart; + return { kind: "updated", thread: { ...thread, session: event.payload.session, latestTurn, + pendingTurnStart, updatedAt: event.occurredAt, }, }; diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index de027fc57d2..66faddc9829 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -74,6 +74,8 @@ const BASE_THREAD: OrchestrationThread = { settledAt: null, deletedAt: null, messages: [], + queuedMessages: [], + pendingTurnStart: null, proposedPlans: [], activities: [], checkpoints: [], diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 31c9ae937eb..33407fb1347 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -343,6 +343,22 @@ export const OrchestrationLatestTurn = Schema.Struct({ }); export type OrchestrationLatestTurn = typeof OrchestrationLatestTurn.Type; +/** + * A follow-up message held server-side while a turn is running. Queued + * messages are not part of the thread timeline; they enter it via the + * normal `thread.message-sent` event when dispatched (auto-drain on turn + * completion, or an explicit `thread.queue.steer`). + */ +export const OrchestrationQueuedMessage = Schema.Struct({ + messageId: MessageId, + text: Schema.String, + attachments: Schema.Array(ChatAttachment), + modelSelection: Schema.optional(ModelSelection), + sourceProposedPlan: Schema.optional(SourceProposedPlanReference), + queuedAt: IsoDateTime, +}); +export type OrchestrationQueuedMessage = typeof OrchestrationQueuedMessage.Type; + export const OrchestrationThread = Schema.Struct({ id: ThreadId, projectId: ProjectId, @@ -370,6 +386,22 @@ export const OrchestrationThread = Schema.Struct({ snoozedAt: Schema.optional(Schema.NullOr(IsoDateTime)), deletedAt: Schema.NullOr(IsoDateTime), messages: Schema.Array(OrchestrationMessage), + queuedMessages: Schema.Array(OrchestrationQueuedMessage).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + ), + /** + * The turn start that was requested but not yet adopted by a provider + * session. Non-null between `thread.turn-start-requested` and the session + * reporting running (or a terminal status / start failure). While set, + * follow-up sends queue and drains hold — the session status alone cannot + * see this window. Read-model twin of the SQL pending-turn-start row. + */ + pendingTurnStart: Schema.NullOr( + Schema.Struct({ + messageId: MessageId, + requestedAt: IsoDateTime, + }), + ).pipe(Schema.withDecodingDefault(Effect.succeed(null))), proposedPlans: Schema.Array(OrchestrationProposedPlan).pipe( Schema.withDecodingDefault(Effect.succeed([])), ), @@ -721,6 +753,26 @@ const ThreadTurnInterruptCommand = Schema.Struct({ createdAt: IsoDateTime, }); +/** + * Send a queued message immediately, steering the active turn. Degrades to + * a normal turn start when the thread is idle by the time it is processed. + */ +const ThreadQueueSteerCommand = Schema.Struct({ + type: Schema.Literal("thread.queue.steer"), + commandId: CommandId, + threadId: ThreadId, + messageId: MessageId, + createdAt: IsoDateTime, +}); + +const ThreadQueueRemoveCommand = Schema.Struct({ + type: Schema.Literal("thread.queue.remove"), + commandId: CommandId, + threadId: ThreadId, + messageId: MessageId, + createdAt: IsoDateTime, +}); + const ThreadApprovalRespondCommand = Schema.Struct({ type: Schema.Literal("thread.approval.respond"), commandId: CommandId, @@ -771,6 +823,8 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ThreadInteractionModeSetCommand, ThreadTurnStartCommand, ThreadTurnInterruptCommand, + ThreadQueueSteerCommand, + ThreadQueueRemoveCommand, ThreadApprovalRespondCommand, ThreadUserInputRespondCommand, ThreadCheckpointRevertCommand, @@ -796,6 +850,8 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadInteractionModeSetCommand, ClientThreadTurnStartCommand, ThreadTurnInterruptCommand, + ThreadQueueSteerCommand, + ThreadQueueRemoveCommand, ThreadApprovalRespondCommand, ThreadUserInputRespondCommand, ThreadCheckpointRevertCommand, @@ -860,6 +916,18 @@ const ThreadActivityAppendCommand = Schema.Struct({ createdAt: IsoDateTime, }); +/** + * Server-internal: dispatch the queued-message head as a turn after a + * natural (non-interrupted) turn completion. Rejected when the queue is + * empty or the thread is busy again; the dispatcher ignores the rejection. + */ +const ThreadQueueDrainCommand = Schema.Struct({ + type: Schema.Literal("thread.queue.drain"), + commandId: CommandId, + threadId: ThreadId, + createdAt: IsoDateTime, +}); + const ThreadRevertCompleteCommand = Schema.Struct({ type: Schema.Literal("thread.revert.complete"), commandId: CommandId, @@ -875,6 +943,7 @@ const InternalOrchestrationCommand = Schema.Union([ ThreadProposedPlanUpsertCommand, ThreadTurnDiffCompleteCommand, ThreadActivityAppendCommand, + ThreadQueueDrainCommand, ThreadRevertCompleteCommand, ]); export type InternalOrchestrationCommand = typeof InternalOrchestrationCommand.Type; @@ -901,6 +970,8 @@ export const OrchestrationEventType = Schema.Literals([ "thread.runtime-mode-set", "thread.interaction-mode-set", "thread.message-sent", + "thread.message-queued", + "thread.queued-message-removed", "thread.turn-start-requested", "thread.turn-interrupt-requested", "thread.approval-response-requested", @@ -1040,6 +1111,26 @@ export const ThreadMessageSentPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadMessageQueuedPayload = Schema.Struct({ + threadId: ThreadId, + messageId: MessageId, + text: Schema.String, + attachments: Schema.Array(ChatAttachment), + modelSelection: Schema.optional(ModelSelection), + sourceProposedPlan: Schema.optional(SourceProposedPlanReference), + queuedAt: IsoDateTime, +}); + +export const ThreadQueuedMessageRemovedReason = Schema.Literals(["user", "dispatched"]); +export type ThreadQueuedMessageRemovedReason = typeof ThreadQueuedMessageRemovedReason.Type; + +export const ThreadQueuedMessageRemovedPayload = Schema.Struct({ + threadId: ThreadId, + messageId: MessageId, + reason: ThreadQueuedMessageRemovedReason, + removedAt: IsoDateTime, +}); + export const ThreadTurnStartRequestedPayload = Schema.Struct({ threadId: ThreadId, messageId: MessageId, @@ -1212,6 +1303,16 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.message-sent"), payload: ThreadMessageSentPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.message-queued"), + payload: ThreadMessageQueuedPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.queued-message-removed"), + payload: ThreadQueuedMessageRemovedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.turn-start-requested"),