feat(orchestrator): queue follow-up messages during active turns, with explicit steer

apps/mobile/src/lib/threadActivity.test.ts

@@ -45,6 +45,8 @@ function makeThread(
     archivedAt: null,
     deletedAt: null,
     messages: [],
+    queuedMessages: [],
+    pendingTurnStart: null,
     proposedPlans: [],
     activities: [],
     checkpoints: [],

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: [],

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<ProjectionThreadMessage>,
+  messages: ReadonlyArray<Pick<ProjectionThreadMessage, "attachments">>,
 ): Set<string> {
   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;
@@ -748,6 +752,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":
@@ -908,9 +914,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;
         }
@@ -920,6 +934,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) {
@@ -1507,6 +1582,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,
@@ -1637,6 +1720,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),

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts

@@ -322,6 +322,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
               updatedAt: "2026-02-24T00:00:05.000Z",
             },
           ],
+          queuedMessages: [],
+          pendingTurnStart: null,
           proposedPlans: [
             {
               id: "plan-1",