fix: replay queued cloud messages via resume on terminal status

When a follow-up was queued during the final turn of a cloud run, the
terminal-status branch in handleCloudTaskUpdate silently dropped it —
the auto-flush above had also raced ahead with sendCloudPrompt against
a now-finished run, so the user_message command never reached the cloud.

Skip the auto-flush when this update brings the run terminal, and have
the terminal-status branch dequeue any pending messages and replay them
via resumeCloudRun (which spins up a fresh task run carrying the prompt
as pending_user_message).

Generated-By: PostHog Code
Task-Id: 8aeaf6f9-8b18-426a-9453-c668ca17d227

Author: VojtechBartos

apps/code/src/renderer/features/sessions/service/service.test.ts

@@ -70,6 +70,7 @@ const mockSessionStoreSetters = vi.hoisted(() => ({
   appendOptimisticItem: vi.fn(),
   clearOptimisticItems: vi.fn(),
   replaceOptimisticWithEvent: vi.fn(),
+  updateCloudStatus: vi.fn(),
 }));
 
 const mockGetConfigOptionByCategory = vi.hoisted(() =>
@@ -915,6 +916,68 @@ describe("SessionService", () => {
     });
   });
 
+  describe("handleCloudTaskUpdate terminal-status routing", () => {
+    it("dequeues queued messages instead of clearing them when the run reaches a terminal status", async () => {
+      const service = getSessionService();
+      let capturedOnData:
+        | ((payload: Record<string, unknown>) => void)
+        | undefined;
+      mockTrpcCloudTask.onUpdate.subscribe.mockImplementation(
+        (
+          _input: unknown,
+          opts: { onData: (p: Record<string, unknown>) => void },
+        ) => {
+          capturedOnData = opts.onData;
+          return { unsubscribe: vi.fn() };
+        },
+      );
+
+      const queuedMessage = {
+        id: "queue-1",
+        content: "gimme a joke",
+        queuedAt: Date.now(),
+      };
+      const session = createMockSession({
+        isCloud: true,
+        cloudStatus: "in_progress",
+        cloudBranch: "main",
+        messageQueue: [queuedMessage],
+      });
+      mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(session);
+      mockSessionStoreSetters.getSessions.mockReturnValue({
+        "run-123": session,
+      });
+      mockSessionStoreSetters.dequeueMessages.mockReturnValue([
+        queuedMessage,
+      ] as never);
+
+      service.watchCloudTask(
+        "task-123",
+        "run-123",
+        "https://api.anthropic.com",
+        123,
+      );
+
+      expect(capturedOnData).toBeDefined();
+      capturedOnData?.({
+        kind: "status",
+        taskId: "task-123",
+        status: "completed",
+        stage: undefined,
+        output: undefined,
+        errorMessage: undefined,
+        branch: "main",
+      });
+
+      // Queue is drained via dequeueMessages (so resumeCloudRun can replay it),
+      // not silently dropped via clearMessageQueue.
+      expect(mockSessionStoreSetters.dequeueMessages).toHaveBeenCalledWith(
+        "task-123",
+      );
+      expect(mockSessionStoreSetters.clearMessageQueue).not.toHaveBeenCalled();
+    });
+  });
+
   describe("reset", () => {
     it("clears connecting tasks", () => {
       const service = getSessionService();

apps/code/src/renderer/features/sessions/service/service.ts

@@ -2924,9 +2924,17 @@ export class SessionService {
       }
     }
 
-    // Flush queued messages when a cloud turn completes (detected via live log updates)
+    const isTerminalUpdate =
+      (update.kind === "status" || update.kind === "snapshot") &&
+      isTerminalStatus(update.status);
+
+    // Flush queued messages when a cloud turn completes mid-run. Skip when
+    // the same update brings the run to a terminal status: queued commands
+    // cannot be delivered to a finished run, so the terminal-status block
+    // below replays them via resumeCloudRun instead.
     const sessionAfterLogs = sessionStoreSetters.getSessions()[taskRunId];
     if (
+      !isTerminalUpdate &&
       sessionAfterLogs &&
       !sessionAfterLogs.isPromptPending &&
       sessionAfterLogs.messageQueue.length > 0
@@ -2968,13 +2976,28 @@ export class SessionService {
       }
 
       if (isTerminalStatus(update.status)) {
-        // Clean up any pending resume messages that couldn't be sent
         const session = sessionStoreSetters.getSessions()[taskRunId];
-        if (
-          session &&
-          (session.messageQueue.length > 0 || session.isPromptPending)
-        ) {
-          sessionStoreSetters.clearMessageQueue(session.taskId);
+        // A user message queued during the final turn cannot be delivered to
+        // a finished run. Replay it through resumeCloudRun, which spins up a
+        // fresh task run carrying the prompt as `pending_user_message`.
+        if (session && session.messageQueue.length > 0) {
+          const queued = sessionStoreSetters.dequeueMessages(session.taskId);
+          const combinedPrompt = combineQueuedCloudPrompts(queued);
+          sessionStoreSetters.updateSession(taskRunId, {
+            isPromptPending: false,
+          });
+          if (combinedPrompt) {
+            this.resumeCloudRun(session, combinedPrompt).catch((err) => {
+              log.error("Failed to resume cloud run with queued messages", {
+                taskId: session.taskId,
+                error: err,
+              });
+              toast.error(
+                "Failed to send follow-up message. Please try again.",
+              );
+            });
+          }
+        } else if (session?.isPromptPending) {
           sessionStoreSetters.updateSession(taskRunId, {
             isPromptPending: false,
           });