fix(app): keep the next queued message queued when one is sent now

Sending a queued message while the agent runs can replace its turn: the
daemon cancels the running turn and starts a new one for the sent message
(Interrupt, Queue-mode send now, and Steer on providers that cannot steer).
The app read the cancel as the agent stopping and immediately sent the next
queued message, which then replaced the turn just started for the first.

The queue now only drains when the agent is idle by the same rule the UI
uses for "working": no open turn and no sent message still waiting for the
provider to pick it up.

packages/app/e2e/browser/agent-message-submission.spec.ts

@@ -1282,6 +1282,49 @@ test.describe("Agent message submission", () => {
     }
   });
 
+  for (const behavior of ["Steer", "Interrupt"] as const) {
+    test(`keeps the next queued message queued after sending one now with ${behavior}`, async ({
+      page,
+    }, testInfo) => {
+      test.setTimeout(120_000);
+      const agent = await startRunningMockAgent(page, {
+        prefix: `queued-send-now-${behavior.toLowerCase()}-${testInfo.workerIndex}-`,
+        model: "one-minute-stream",
+        prompt: "Keep this turn active while the user sends a queued message now.",
+      });
+      const firstPrompt = "Send the first queued message now.";
+      const secondPrompt = "Keep the second queued message queued.";
+      try {
+        await (behavior === "Steer"
+          ? configureSteerInSettings(page)
+          : configureInterruptInSettings(page));
+        await page.goBack();
+        await expectComposerVisible(page);
+        await expectAgentReadyToInterrupt(page);
+        await queueMessage(page, firstPrompt);
+        await queueMessage(page, secondPrompt);
+        const sendNow = page.getByRole("button", { name: "Send queued message now" });
+        await expect(sendNow).toHaveCount(2);
+
+        await sendNow.first().click();
+
+        await expect(page.getByTestId("user-message").filter({ hasText: firstPrompt })).toHaveCount(
+          1,
+        );
+        await expect(
+          page.getByTestId("user-message").filter({ hasText: firstPrompt }),
+        ).toHaveAttribute("aria-busy", "false");
+        await expectAgentReadyToInterrupt(page);
+        await expect(sendNow).toHaveCount(1);
+        await expect(
+          page.getByTestId("user-message").filter({ hasText: secondPrompt }),
+        ).toHaveCount(0);
+      } finally {
+        await agent.cleanup();
+      }
+    });
+  }
+
   test("replays Claude-shaped steering inside one active turn", async ({ page }, testInfo) => {
     test.setTimeout(120_000);
     await replaySteeredSleepTurnInBrowser(page, testInfo, "claude");

packages/app/src/runtime/host-runtime.test.ts

@@ -17,6 +17,7 @@ import type { AgentPermissionRequest } from "@getpaseo/protocol/agent-types";
 import type { HostConnection, HostProfile } from "@/types/host-connection";
 import { defaultHostAppearance } from "@/hosts/appearance";
 import { useSessionStore, type Agent } from "@/stores/session-store";
+import { createUserMessage } from "@/types/stream";
 import { normalizeAgentSnapshot } from "@/utils/agent-snapshots";
 import { isAgentArchiving, setAgentArchiving } from "@/hooks/use-archive-agent";
 import { queryClient } from "@/data/query-client";
@@ -2799,6 +2800,91 @@ describe("HostRuntimeStore", () => {
     useSessionStore.getState().clearSession(host.serverId);
   });
 
+  it("keeps the next queued message while a sent message replaces the running turn", async () => {
+    const host = makeHost({
+      serverId: "srv_queue_replacement",
+      connections: [{ id: "direct:lan:6767", type: "directTcp", endpoint: "lan:6767" }],
+    });
+    const fakeClient = new FakeDaemonClient();
+    fakeClient.setConnectionState({ status: "connected" });
+    const running = makeFetchAgentsEntry({
+      id: "agent",
+      cwd: "/repo",
+      updatedAt: "2026-07-12T10:00:00.000Z",
+    });
+    fakeClient.fetchAgentsResponses.push(
+      makeFetchAgentsPayload({
+        entries: [{ ...running, agent: { ...running.agent, status: "running" } }],
+      }),
+    );
+    const store = new HostRuntimeStore({
+      deps: {
+        createClient: () => fakeClient as unknown as DaemonClient,
+        connectToDaemon: async () => ({
+          client: fakeClient as unknown as DaemonClient,
+          serverId: host.serverId,
+          hostname: null,
+        }),
+        getClientId: async () => "cid_queue_replacement",
+      },
+    });
+    const sessionStore = useSessionStore.getState();
+    sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
+    sessionStore.updateSessionServerInfo(host.serverId, {
+      serverId: host.serverId,
+      hostname: null,
+      version: null,
+      features: { canonicalSubmittedPrompts: true },
+    });
+    store.syncHosts([host]);
+    await waitForHostOnline(store, host.serverId);
+    await store.refreshAgentDirectory({ serverId: host.serverId, subscribe: {} });
+    store.applyAgentTurnLiveness(host.serverId, "agent", {
+      type: "stream_open",
+      turn: { turnId: "turn-1", startedAt: null },
+    });
+    sessionStore.setQueuedMessages(
+      host.serverId,
+      new Map([["agent", [{ id: "second", text: "keep me queued", attachments: [] }]]]),
+    );
+
+    // "Send now" on the first queued message is in flight; the daemon replaces the turn.
+    const sentNow = createUserMessage({
+      clientMessageId: "sent-now",
+      text: "sent now",
+      timestamp: new Date(),
+    });
+    sessionStore.beginAgentMessageSubmission(host.serverId, "agent", sentNow);
+    store.applyAgentTurnLiveness(host.serverId, "agent", {
+      type: "stream_close",
+      turnId: "turn-1",
+    });
+    store.applyAgentTurnLiveness(host.serverId, "agent", {
+      type: "stream_open",
+      turn: { turnId: "turn-2", startedAt: null },
+    });
+
+    expect(fakeClient.sentAgentMessages).toEqual([]);
+    expect(useSessionStore.getState().sessions[host.serverId]?.queuedMessages.get("agent")).toEqual(
+      [{ id: "second", text: "keep me queued", attachments: [] }],
+    );
+
+    // Once the replacement turn ends, the queue drains as usual.
+    sessionStore.acceptAgentMessageSubmission(host.serverId, "agent", "sent-now");
+    sessionStore.setAgentStreamState(host.serverId, "agent", {
+      acknowledgedClientMessageIds: ["sent-now"],
+    });
+    store.applyAgentTurnLiveness(host.serverId, "agent", {
+      type: "stream_close",
+      turnId: "turn-2",
+    });
+    await fakeClient.waitForSentMessages(1);
+    expect(fakeClient.sentAgentMessages.map(([, text]) => text)).toEqual(["keep me queued"]);
+
+    store.syncHosts([]);
+    useSessionStore.getState().clearSession(host.serverId);
+  });
+
   it("submits an automatically drained message through the submission producer", async () => {
     const host = makeHost({ serverId: "srv_drain_submission" });
     const fakeClient = new FakeDaemonClient();

packages/app/src/runtime/host-runtime.ts

@@ -49,6 +49,7 @@ import { readDesktopManagedLocalCredential } from "@/desktop/daemon/local-creden
 import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities";
 import { BROWSER_AUTOMATION_COMMAND_NAMES } from "@getpaseo/protocol/browser-automation/rpc-schemas";
 import {
+  selectAgentTurnPresentation,
   useSessionStore,
   type Agent,
   type WorkspaceDescriptor,
@@ -2323,6 +2324,9 @@ export class HostRuntimeStore {
     if (!client || !queue?.length || session.initializingAgents.get(agentId) === true) {
       return;
     }
+    // A message sent while the agent runs can replace its turn. The replaced turn closes
+    // before the replacement opens, but the agent is still busy with the sent message.
+    if (selectAgentTurnPresentation(session, agentId).isActive) return;
     this.queuedAgentDrainInFlight.add(drainKey);
     const next = queue[0];
     void sendQueuedComposerMessageNow({