mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-23 09:42:53 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4e1eb513b6 |
@@ -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");
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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({
|
||||
|
||||
Reference in New Issue
Block a user