diff --git a/packages/app/e2e/browser/agent-message-submission.spec.ts b/packages/app/e2e/browser/agent-message-submission.spec.ts index e7f87ed103..2041a0e001 100644 --- a/packages/app/e2e/browser/agent-message-submission.spec.ts +++ b/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"); diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 691e468363..ee1da0fbf7 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/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(); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 2cea05144d..7ecb65fe16 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/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({