From d68fafe93b585257d4c919577f4d4ad43b730ab6 Mon Sep 17 00:00:00 2001 From: MikoyChinese Date: Fri, 31 Jul 2026 17:47:43 +0800 Subject: [PATCH] fix(app): automatically send queued messages --- .../app/e2e/agent-message-submission.spec.ts | 50 ++++ packages/app/e2e/helpers/composer.ts | 2 + .../app/e2e/helpers/daemon-websocket-gate.ts | 60 +++- .../directory-sync/agent-replica.test.ts | 98 +++++- .../runtime/directory-sync/agent-replica.ts | 59 ++-- packages/app/src/runtime/host-runtime.test.ts | 283 +++++++++++++++++- .../utils/agent-directory-reconciliation.ts | 4 +- .../src/utils/agent-directory-sync.test.ts | 4 +- .../app/src/utils/agent-directory-sync.ts | 6 +- 9 files changed, 519 insertions(+), 47 deletions(-) create mode 100644 packages/app/e2e/agent-message-submission.spec.ts diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts new file mode 100644 index 0000000000..12afc3f2b2 --- /dev/null +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -0,0 +1,50 @@ +import { expect, test } from "./fixtures"; +import { + expectQueuedMessageButton, + fillComposerDraft, + sendDraftToQueue, + startRunningMockAgent, +} from "./helpers/composer"; +import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate"; + +test.describe("Agent message submission", () => { + test("automatically sends the next queued message after an authoritative turn completion", async ({ + page, + }) => { + test.setTimeout(120_000); + const gate = await installDaemonWebSocketGate(page); + const agent = await startRunningMockAgent(page, { + prefix: "queued-auto-drain-", + model: "ten-second-stream", + prompt: "Run the first turn before the queued follow-up.", + }); + const queuedPrompt = "Automatically send this queued follow-up."; + + try { + gate.holdNextStoppedAgentUpdate(agent.agentId); + await fillComposerDraft(page, queuedPrompt); + await sendDraftToQueue(page); + await expectQueuedMessageButton(page); + + await gate.waitForHeldStoppedAgentUpdate(); + await expect(page.getByRole("button", { name: /stop|cancel/i })).toHaveCount(0, { + timeout: 15_000, + }); + await expectQueuedMessageButton(page); + const queuedUserMessage = page.getByTestId("user-message").filter({ hasText: queuedPrompt }); + await expect(queuedUserMessage).toHaveCount(0); + + gate.releaseHeldStoppedAgentUpdate(); + + await expect(queuedUserMessage).toBeVisible({ timeout: 15_000 }); + await expect(page.getByRole("button", { name: "Send queued message now" })).toHaveCount(0); + await expect(page.getByRole("button", { name: /stop|cancel/i }).first()).toBeVisible({ + timeout: 15_000, + }); + await expect(page.getByTestId("user-message")).toHaveCount(2, { timeout: 15_000 }); + } finally { + gate.releaseHeldStoppedAgentUpdate(); + await agent.cleanup(); + } + }); +}); diff --git a/packages/app/e2e/helpers/composer.ts b/packages/app/e2e/helpers/composer.ts index 68600f023f..9779079dbc 100644 --- a/packages/app/e2e/helpers/composer.ts +++ b/packages/app/e2e/helpers/composer.ts @@ -174,6 +174,7 @@ export async function selectGithubOption( } export interface MockAgentSetup { + agentId: string; client: SeedDaemonClient; repo: Awaited>; cleanup: () => Promise; @@ -209,6 +210,7 @@ export async function startRunningMockAgent( timeout: 30_000, }); return { + agentId: agent.id, client, repo, cleanup: async () => { diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index 860dae5be5..d1d447de56 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -23,6 +23,11 @@ interface ClientRequest { interface ServerMessage { type?: unknown; payload?: { + kind?: unknown; + agent?: { + id?: unknown; + status?: unknown; + }; initial?: { path?: unknown; }; @@ -65,6 +70,15 @@ function directoryForRequest(request: ClientRequest): keyof DirectoryBootstrapCo return null; } +function isStoppedAgentUpdate(message: ServerMessage | null, agentId: string | null): boolean { + return ( + message?.type === "agent_update" && + message.payload?.kind === "upsert" && + message.payload.agent?.id === agentId && + message.payload.agent.status !== "running" + ); +} + export async function installDaemonWebSocketGate(page: Page) { let acceptingConnections = true; const activeSockets = new Set(); @@ -84,6 +98,23 @@ export async function installDaemonWebSocketGate(page: Page) { let heldReadyFileUpdate: (() => void) | null = null; let resolveHeldReadyFileUpdate: (() => void) | null = null; let heldReadyFileUpdatePromise = Promise.resolve(); + let stoppedAgentUpdateIdToHold: string | null = null; + let heldStoppedAgentUpdate: (() => void) | null = null; + let resolveHeldStoppedAgentUpdate: (() => void) | null = null; + let heldStoppedAgentUpdatePromise = Promise.resolve(); + + const recordFileSubscription = (message: ServerMessage | null): void => { + if ( + message?.type !== "fs.file.subscribe.response" || + typeof message.payload?.initial?.path !== "string" + ) { + return; + } + const path = message.payload.initial.path; + subscribedFilePaths.add(path); + fileSubscriptionWaiters.get(path)?.(); + fileSubscriptionWaiters.delete(path); + }; await page.routeWebSocket(daemonWsRoutePattern(), (ws) => { if (!acceptingConnections) { @@ -126,14 +157,14 @@ export async function installDaemonWebSocketGate(page: Page) { if (!acceptingConnections) return; const serverMessage = readServerMessage(message); if ( - serverMessage?.type === "fs.file.subscribe.response" && - typeof serverMessage.payload?.initial?.path === "string" + isStoppedAgentUpdate(serverMessage, stoppedAgentUpdateIdToHold) && + !heldStoppedAgentUpdate ) { - const path = serverMessage.payload.initial.path; - subscribedFilePaths.add(path); - fileSubscriptionWaiters.get(path)?.(); - fileSubscriptionWaiters.delete(path); + heldStoppedAgentUpdate = () => ws.send(message); + resolveHeldStoppedAgentUpdate?.(); + return; } + recordFileSubscription(serverMessage); if ( serverMessage?.type === "fs.file.update" && serverMessage.payload?.version?.status === "ready" && @@ -191,6 +222,23 @@ export async function installDaemonWebSocketGate(page: Page) { resolveHeldReadyFileUpdate = null; forward?.(); }, + holdNextStoppedAgentUpdate(agentId: string): void { + stoppedAgentUpdateIdToHold = agentId; + heldStoppedAgentUpdate = null; + heldStoppedAgentUpdatePromise = new Promise((resolve) => { + resolveHeldStoppedAgentUpdate = resolve; + }); + }, + waitForHeldStoppedAgentUpdate(): Promise { + return heldStoppedAgentUpdatePromise; + }, + releaseHeldStoppedAgentUpdate(): void { + const forward = heldStoppedAgentUpdate; + heldStoppedAgentUpdate = null; + stoppedAgentUpdateIdToHold = null; + resolveHeldStoppedAgentUpdate = null; + forward?.(); + }, async drop(): Promise { acceptingConnections = false; const sockets = Array.from(activeSockets); diff --git a/packages/app/src/runtime/directory-sync/agent-replica.test.ts b/packages/app/src/runtime/directory-sync/agent-replica.test.ts index 4456ecc0a8..15c2365bed 100644 --- a/packages/app/src/runtime/directory-sync/agent-replica.test.ts +++ b/packages/app/src/runtime/directory-sync/agent-replica.test.ts @@ -2,18 +2,24 @@ import { describe, expect, it } from "vitest"; import type { DaemonClient, FetchAgentsEntry } from "@getpaseo/client/internal/daemon-client"; import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { useSessionStore } from "@/stores/session-store"; +import { processAgentStreamEvent } from "@/timeline/session-stream-reducers"; import { AgentDirectoryReplica } from "./agent-replica"; -function payload(title: string): AgentSnapshotPayload { +function payload(input: { + title: string; + status?: AgentSnapshotPayload["status"]; + updatedAt?: string; + archivedAt?: string | null; +}): AgentSnapshotPayload { return { id: "agent", provider: "codex", cwd: "/repo", model: null, createdAt: "2026-07-17T00:00:00.000Z", - updatedAt: "2026-07-17T00:01:00.000Z", + updatedAt: input.updatedAt ?? "2026-07-17T00:01:00.000Z", lastUserMessageAt: null, - status: "idle", + status: input.status ?? "idle", capabilities: { supportsStreaming: true, supportsSessionPersistence: true, @@ -26,8 +32,9 @@ function payload(title: string): AgentSnapshotPayload { availableModes: [], pendingPermissions: [], persistence: null, - title, + title: input.title, labels: {}, + ...(input.archivedAt !== undefined ? { archivedAt: input.archivedAt } : {}), }; } @@ -51,12 +58,83 @@ function entry(agent: AgentSnapshotPayload): FetchAgentsEntry { } describe("AgentDirectoryReplica", () => { + it("detects an authoritative stop after the timeline optimistically becomes idle", () => { + const serverId = "agent-replica-authoritative-transition"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + const stoppedAgentIds: string[] = []; + const replica = new AgentDirectoryReplica(serverId, (agentId) => stoppedAgentIds.push(agentId)); + replica.commitSnapshot([entry(payload({ title: "running", status: "running" }))], []); + + const runningAgent = useSessionStore.getState().sessions[serverId]?.agents.get("agent"); + if (!runningAgent) throw new Error("Expected running agent after authoritative snapshot"); + const timelineResult = processAgentStreamEvent({ + event: { type: "turn_completed", provider: "codex" }, + seq: undefined, + epoch: undefined, + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: { + status: runningAgent.status, + updatedAt: runningAgent.updatedAt, + lastActivityAt: runningAgent.lastActivityAt, + }, + timestamp: new Date("2026-07-17T00:02:00.000Z"), + }); + const timelineAgent = timelineResult.agent; + if (!timelineAgent) throw new Error("Expected turn completion to update the agent"); + expect(timelineAgent.status).toBe("idle"); + store.setAgents(serverId, (current) => { + const next = new Map(current); + next.set("agent", { ...runningAgent, ...timelineAgent }); + return next; + }); + + replica.applyDelta({ + kind: "upsert", + agent: payload({ + title: "idle", + status: "idle", + updatedAt: "2026-07-17T00:03:00.000Z", + }), + project: entry(payload({ title: "idle" })).project, + }); + + expect(stoppedAgentIds).toEqual(["agent"]); + store.clearSession(serverId); + }); + + it("does not report an authoritative stop when the agent is archived", () => { + const serverId = "agent-replica-archived-transition"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + const stoppedAgentIds: string[] = []; + const replica = new AgentDirectoryReplica(serverId, (agentId) => stoppedAgentIds.push(agentId)); + const running = payload({ title: "running", status: "running" }); + replica.commitSnapshot([entry(running)], []); + + replica.applyDelta({ + kind: "upsert", + agent: payload({ + title: "archived", + status: "idle", + updatedAt: "2026-07-17T00:03:00.000Z", + archivedAt: "2026-07-17T00:02:00.000Z", + }), + project: entry(running).project, + }); + + expect(stoppedAgentIds).toEqual([]); + store.clearSession(serverId); + }); + it("keeps membership authoritative across remove, stale timeline, and re-add", () => { const serverId = "agent-replica"; const store = useSessionStore.getState(); store.initializeSession(serverId, null as unknown as DaemonClient); const replica = new AgentDirectoryReplica(serverId, () => undefined); - replica.commitSnapshot([entry(payload("directory"))], []); + replica.commitSnapshot([entry(payload({ title: "directory" }))], []); const directoryPlacement = useSessionStore .getState() .sessions[serverId]?.agents.get("agent")?.projectPlacement; @@ -64,17 +142,17 @@ describe("AgentDirectoryReplica", () => { const staleToken = replica.captureTimeline("agent"); replica.remove("agent"); - expect(replica.submitTimelineAgent(staleToken, payload("stale"))).toBe(false); + expect(replica.submitTimelineAgent(staleToken, payload({ title: "stale" }))).toBe(false); expect(useSessionStore.getState().sessions[serverId]?.agents.has("agent")).toBe(false); replica.applyDelta({ kind: "upsert", - agent: payload("re-added"), - project: entry(payload("x")).project, + agent: payload({ title: "re-added" }), + project: entry(payload({ title: "x" })).project, }); - expect(replica.submitTimelineAgent(staleToken, payload("still stale"))).toBe(false); + expect(replica.submitTimelineAgent(staleToken, payload({ title: "still stale" }))).toBe(false); const currentToken = replica.captureTimeline("agent"); - expect(replica.submitTimelineAgent(currentToken, payload("current"))).toBe(true); + expect(replica.submitTimelineAgent(currentToken, payload({ title: "current" }))).toBe(true); expect(useSessionStore.getState().sessions[serverId]?.agents.get("agent")?.title).toBe( "current", ); diff --git a/packages/app/src/runtime/directory-sync/agent-replica.ts b/packages/app/src/runtime/directory-sync/agent-replica.ts index 62de843cdc..3d8cc5b5d5 100644 --- a/packages/app/src/runtime/directory-sync/agent-replica.ts +++ b/packages/app/src/runtime/directory-sync/agent-replica.ts @@ -4,6 +4,7 @@ import { clearArchiveAgentPending } from "@/hooks/use-archive-agent"; import { queryClient } from "@/data/query-client"; import { useSessionStore, type Agent } from "@/stores/session-store"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; +import { acceptAgentDirectoryUpdate } from "@/utils/agent-directory-update-policy"; import { applyAgentDirectoryDelta, type AgentDirectoryDelta, @@ -20,14 +21,24 @@ export interface AgentLifecycleToken { readonly version: number; } +interface AuthoritativeAgentState { + status: AgentSnapshotPayload["status"]; + updatedAt: Date | string; + lastUsage?: AgentSnapshotPayload["lastUsage"]; + archivedAt?: Date | string | null; +} + export class AgentDirectoryReplica { private readonly lifecycleVersions = new Map(); - private readonly members = new Set(); + private readonly authoritativeAgents = new Map(); constructor( private readonly serverId: string, private readonly onStoppedRunning: (agentId: string) => void, - ) {} + ) { + const existing = useSessionStore.getState().sessions[serverId]?.agents ?? new Map(); + for (const [agentId, agent] of existing) this.authoritativeAgents.set(agentId, agent); + } captureTimeline(agentId: string): AgentLifecycleToken { return { agentId, version: this.lifecycleVersions.get(agentId) ?? 0 }; @@ -35,7 +46,7 @@ export class AgentDirectoryReplica { submitTimelineAgent(token: AgentLifecycleToken, payload: AgentSnapshotPayload): boolean { if ( - !this.members.has(token.agentId) || + !this.authoritativeAgents.has(token.agentId) || token.version !== (this.lifecycleVersions.get(token.agentId) ?? 0) ) { return false; @@ -59,46 +70,58 @@ export class AgentDirectoryReplica { } applyDelta(delta: AgentDirectoryDelta): void { - const before = this.members.has(delta.kind === "remove" ? delta.agentId : delta.agent.id); - const result = applyAgentDirectoryDelta({ serverId: this.serverId, delta }); if (delta.kind === "remove") { - this.members.delete(delta.agentId); + applyAgentDirectoryDelta({ serverId: this.serverId, delta }); + this.authoritativeAgents.delete(delta.agentId); this.advance(delta.agentId); - } else { - this.members.add(delta.agent.id); - if (!before) this.advance(delta.agent.id); + return; + } + + const previous = this.authoritativeAgents.get(delta.agent.id); + const accepted = acceptAgentDirectoryUpdate(previous, delta.agent); + applyAgentDirectoryDelta({ serverId: this.serverId, delta }); + this.authoritativeAgents.set(delta.agent.id, accepted); + if (!previous) this.advance(delta.agent.id); + if (previous?.status === "running" && accepted.status !== "running" && !accepted.archivedAt) { + this.onStoppedRunning(delta.agent.id); } - if (result.stoppedRunning) this.onStoppedRunning(result.agentId); } commitSnapshot( entries: FetchAgentsEntry[], deltas: readonly AgentDirectoryDelta[], ): Map { - const previous = useSessionStore.getState().sessions[this.serverId]?.agents ?? new Map(); + const previous = this.authoritativeAgents; const reconciled = reconcileAgentDirectory({ previous, snapshot: entries, deltas }); const nextIds = new Set(reconciled.entries.map((entry) => entry.agent.id)); - for (const agentId of this.members) { + for (const agentId of this.authoritativeAgents.keys()) { if (!nextIds.has(agentId)) this.advance(agentId); } for (const agentId of nextIds) { - if (!this.members.has(agentId)) this.advance(agentId); + if (!this.authoritativeAgents.has(agentId)) this.advance(agentId); } - for (const agentId of previous.keys()) { + const replicaAgents = useSessionStore.getState().sessions[this.serverId]?.agents ?? new Map(); + for (const agentId of replicaAgents.keys()) { if (!nextIds.has(agentId)) removeAgentDirectoryReplica(this.serverId, agentId); } - this.members.clear(); - for (const agentId of nextIds) this.members.add(agentId); + this.authoritativeAgents.clear(); + for (const { agent } of reconciled.entries) this.authoritativeAgents.set(agent.id, agent); const { agents } = replaceFetchedAgentDirectory({ serverId: this.serverId, entries: reconciled.entries, }); - for (const agentId of reconciled.stoppedRunningAgentIds) this.onStoppedRunning(agentId); + for (const agentId of reconciled.stoppedRunningAgentIds) { + if (!this.authoritativeAgents.get(agentId)?.archivedAt) this.onStoppedRunning(agentId); + } return agents; } archive(agentId: string, archivedAt: string): void { this.advance(agentId); + const authoritative = this.authoritativeAgents.get(agentId); + if (authoritative) { + this.authoritativeAgents.set(agentId, { ...authoritative, archivedAt }); + } useSessionStore.getState().setAgents(this.serverId, (current) => { const agent = current.get(agentId); if (!agent) return current; @@ -110,7 +133,7 @@ export class AgentDirectoryReplica { } remove(agentId: string): void { - this.members.delete(agentId); + this.authoritativeAgents.delete(agentId); this.advance(agentId); removeAgentDirectoryReplica(this.serverId, agentId); } diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 0f4946a4b9..ca47c9859c 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -8,6 +8,8 @@ import type { import type { ConnectionOffer } from "@getpaseo/protocol/connection-offer"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; import type { AgentPermissionRequest } from "@getpaseo/protocol/agent-types"; +import type { AttachmentStore } from "@/attachments/types"; +import { __setAttachmentStoreForTests } from "@/attachments/store"; import type { HostConnection, HostProfile } from "@/types/host-connection"; import { useSessionStore, type Agent } from "@/stores/session-store"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; @@ -35,6 +37,7 @@ class FakeDaemonClient { Awaited> | ReturnType > = []; public sentAgentMessages: Array> = []; + private completedAgentMessageSends = 0; public sendAgentMessageFailures: Error[] = []; public sendAgentMessageResponses: Promise[] = []; private agentUpdateListeners = new Set< @@ -43,6 +46,7 @@ class FakeDaemonClient { private fetchWaiters = new Set<() => void>(); private agentListenerWaiters = new Set<() => void>(); private sentMessageWaiters = new Set<() => void>(); + private completedSentMessageWaiters = new Set<() => void>(); on( type: "agent_update", @@ -83,10 +87,15 @@ class FakeDaemonClient { async sendAgentMessage(...args: Parameters): Promise { this.sentAgentMessages.push(args); for (const waiter of this.sentMessageWaiters) waiter(); - const response = this.sendAgentMessageResponses.shift(); - if (response) await response; - const failure = this.sendAgentMessageFailures.shift(); - if (failure) throw failure; + try { + const response = this.sendAgentMessageResponses.shift(); + if (response) await response; + const failure = this.sendAgentMessageFailures.shift(); + if (failure) throw failure; + } finally { + this.completedAgentMessageSends += 1; + for (const waiter of this.completedSentMessageWaiters) waiter(); + } } async waitForSentMessages(count: number): Promise { @@ -101,6 +110,18 @@ class FakeDaemonClient { }); } + async waitForCompletedSentMessages(count: number): Promise { + if (this.completedAgentMessageSends >= count) return; + await new Promise((resolve) => { + const waiter = () => { + if (this.completedAgentMessageSends < count) return; + this.completedSentMessageWaiters.delete(waiter); + resolve(); + }; + this.completedSentMessageWaiters.add(waiter); + }); + } + ensureConnected(): void { if (this.state.status !== "connected") { this.setConnectionState({ status: "connected" }); @@ -202,6 +223,7 @@ class FakeDaemonClient { afterEach(() => { vi.useRealTimers(); + __setAttachmentStoreForTests(null); delete (globalThis as Record).__PASEO_INITIAL_DAEMON_CONNECTION__; delete (globalThis as { window?: unknown }).window; }); @@ -262,6 +284,21 @@ async function waitForDirectoryReady(store: HostRuntimeStore, serverId: string): }); } +function applyOptimisticTurnCompleted(serverId: string, agentId: string, timestamp: string): void { + const store = useSessionStore.getState(); + store.setAgents(serverId, (current) => { + const currentAgent = current.get(agentId); + if (!currentAgent) throw new Error(`Missing agent ${agentId}`); + const completedAt = new Date(timestamp); + return new Map(current).set(agentId, { + ...currentAgent, + status: "idle", + updatedAt: completedAt, + lastActivityAt: completedAt, + }); + }); +} + function makeFetchAgentsEntry(input: { id: string; cwd: string; @@ -2259,6 +2296,111 @@ describe("HostRuntimeStore", () => { useSessionStore.getState().clearSession(host.serverId); }); + it("drains one queued message per authoritative turn completion in FIFO order", async () => { + const host = makeHost({ + serverId: "srv_queued_fifo", + connections: [{ id: "direct:lan:6767", type: "directTcp", endpoint: "lan:6767" }], + }); + const fakeClient = new FakeDaemonClient(); + fakeClient.setConnectionState({ status: "connected" }); + const agentEntry = makeFetchAgentsEntry({ + id: "agent", + cwd: "/repo", + updatedAt: "2026-07-12T10:00:00.000Z", + }); + fakeClient.fetchAgentsResponses.push( + makeFetchAgentsPayload({ + entries: [{ ...agentEntry, agent: { ...agentEntry.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_queued_fifo", + }, + }); + const sessionStore = useSessionStore.getState(); + sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + sessionStore.setQueuedMessages( + host.serverId, + new Map([ + [ + "agent", + [ + { id: "message-first", text: "first queued", attachments: [] }, + { id: "message-second", text: "second queued", attachments: [] }, + ], + ], + ]), + ); + store.syncHosts([host]); + await waitForDirectoryReady(store, host.serverId); + + applyOptimisticTurnCompleted(host.serverId, "agent", "2026-07-12T10:01:00.000Z"); + fakeClient.agentUpdate({ + kind: "upsert", + agent: { + ...agentEntry.agent, + status: "idle", + updatedAt: "2026-07-12T10:02:00.000Z", + }, + project: agentEntry.project, + }); + await vi.waitFor(() => expect(fakeClient.sentAgentMessages).toHaveLength(1)); + await vi.waitFor(() => { + expect( + useSessionStore.getState().sessions[host.serverId]?.queuedMessages.get("agent"), + ).toEqual([{ id: "message-second", text: "second queued", attachments: [] }]); + }); + expect(fakeClient.sentAgentMessages).toHaveLength(1); + expect(fakeClient.sentAgentMessages[0]).toEqual([ + "agent", + "first queued", + { messageId: "message-first", attachments: [] }, + ]); + await fakeClient.waitForCompletedSentMessages(1); + await new Promise((resolve) => setImmediate(resolve)); + + fakeClient.agentUpdate({ + kind: "upsert", + agent: { + ...agentEntry.agent, + status: "running", + updatedAt: "2026-07-12T10:03:00.000Z", + }, + project: agentEntry.project, + }); + expect(fakeClient.sentAgentMessages).toHaveLength(1); + + applyOptimisticTurnCompleted(host.serverId, "agent", "2026-07-12T10:04:00.000Z"); + fakeClient.agentUpdate({ + kind: "upsert", + agent: { + ...agentEntry.agent, + status: "idle", + updatedAt: "2026-07-12T10:05:00.000Z", + }, + project: agentEntry.project, + }); + await vi.waitFor(() => expect(fakeClient.sentAgentMessages).toHaveLength(2)); + + expect(fakeClient.sentAgentMessages).toEqual([ + ["agent", "first queued", { messageId: "message-first", attachments: [] }], + ["agent", "second queued", { messageId: "message-second", attachments: [] }], + ]); + expect(useSessionStore.getState().sessions[host.serverId]?.queuedMessages.get("agent")).toEqual( + [], + ); + + store.syncHosts([]); + useSessionStore.getState().clearSession(host.serverId); + }); + it("restores an automatically drained message when sending fails", async () => { const host = makeHost({ serverId: "srv_failed_queue_drain" }); const fakeClient = new FakeDaemonClient(); @@ -2412,6 +2554,139 @@ describe("HostRuntimeStore", () => { sessionStore.clearSession(host.serverId); }); + it("preserves images, uploaded files, and Forge attachments when draining a queue", async () => { + const host = makeHost({ serverId: "srv_queue_attachments" }); + const fakeClient = new FakeDaemonClient(); + 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_attachments", + }, + }); + const sessionStore = useSessionStore.getState(); + sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + sessionStore.updateSessionServerInfo(host.serverId, { + serverId: host.serverId, + hostname: null, + version: "0.1.106", + features: { forgeSearch: true }, + }); + __setAttachmentStoreForTests({ + storageType: "web-indexeddb", + async save() { + throw new Error("not used in this test"); + }, + async encodeBase64({ attachment }) { + return `${attachment.id}:base64`; + }, + async resolvePreviewUrl() { + throw new Error("not used in this test"); + }, + async delete() {}, + async garbageCollect() {}, + } satisfies AttachmentStore); + sessionStore.setQueuedMessages( + host.serverId, + new Map([ + [ + "agent", + [ + { + id: "queued-modern-attachments", + text: "review these", + attachments: [ + { + kind: "image" as const, + metadata: { + id: "image-1", + mimeType: "image/png", + storageType: "web-indexeddb" as const, + storageKey: "image-1", + fileName: "screenshot.png", + byteSize: 128, + createdAt: 1, + }, + }, + { + kind: "file" as const, + attachment: { + type: "uploaded_file" as const, + id: "file-1", + fileName: "context.json", + mimeType: "application/json", + size: 42, + path: "/tmp/context.json", + }, + }, + { + kind: "forge_change_request" as const, + item: { + kind: "change_request" as const, + forge: "gitlab" as const, + projectPath: "acme/repo", + number: 42, + title: "Queue fix", + url: "https://gitlab.com/acme/repo/-/merge_requests/42", + state: "open" as const, + body: "Details", + labels: [], + baseRefName: "main", + headRefName: "fix/queue", + }, + }, + ], + }, + ], + ], + ]), + ); + + store.drainQueuedAgentMessage(host.serverId, "agent"); + await fakeClient.waitForCompletedSentMessages(1); + + expect(fakeClient.sentAgentMessages).toEqual([ + [ + "agent", + "review these", + { + messageId: "queued-modern-attachments", + images: [{ data: "image-1:base64", mimeType: "image/png" }], + attachments: [ + { + type: "uploaded_file", + id: "file-1", + fileName: "context.json", + mimeType: "application/json", + size: 42, + path: "/tmp/context.json", + }, + { + type: "forge_change_request", + mimeType: "application/paseo-forge-change-request", + forge: "gitlab", + projectPath: "acme/repo", + number: 42, + title: "Queue fix", + url: "https://gitlab.com/acme/repo/-/merge_requests/42", + body: "Details", + baseRefName: "main", + headRefName: "fix/queue", + }, + ], + }, + ], + ]); + expect(useSessionStore.getState().sessions[host.serverId]?.queuedMessages.get("agent")).toEqual( + [], + ); + sessionStore.clearSession(host.serverId); + }); + it("applies buffered stale side effects from the accepted page agent", async () => { const host = makeHost({ serverId: "srv_buffered_stale_side_effects", diff --git a/packages/app/src/utils/agent-directory-reconciliation.ts b/packages/app/src/utils/agent-directory-reconciliation.ts index 86130f5d2a..8f853fd66d 100644 --- a/packages/app/src/utils/agent-directory-reconciliation.ts +++ b/packages/app/src/utils/agent-directory-reconciliation.ts @@ -1,10 +1,10 @@ import type { FetchAgentsEntry } from "@getpaseo/client/internal/daemon-client"; -import type { Agent } from "@/stores/session-store"; +import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import type { AgentDirectoryDelta } from "./agent-directory-sync"; import { acceptAgentDirectoryUpdate } from "./agent-directory-update-policy"; export function reconcileAgentDirectory(input: { - previous: ReadonlyMap; + previous: ReadonlyMap>; snapshot: FetchAgentsEntry[]; deltas: readonly AgentDirectoryDelta[]; }): { entries: FetchAgentsEntry[]; stoppedRunningAgentIds: string[] } { diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index 1ec16bde26..29e56abc4b 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -211,7 +211,7 @@ describe("replaceFetchedAgentDirectory", () => { store.flushAgentLastActivity(); setAgentArchiving({ queryClient, serverId, agentId, isArchiving: true }); - const staleResult = applyAgentDirectoryDelta({ + applyAgentDirectoryDelta({ serverId, delta: { kind: "upsert", @@ -236,7 +236,6 @@ describe("replaceFetchedAgentDirectory", () => { status: agent?.status, usage: agent?.lastUsage, workspaceId: agent?.workspaceId, - stoppedRunning: staleResult.stoppedRunning, permissions: Array.from(state.sessions[serverId]?.pendingPermissions.values() ?? []).map( ({ request }) => request.id, ), @@ -247,7 +246,6 @@ describe("replaceFetchedAgentDirectory", () => { status: "running", usage: { inputTokens: 20, outputTokens: 8 }, workspaceId: "legacy-workspace", - stoppedRunning: false, permissions: ["current-permission"], archivePending: true, activity: "2026-07-12T11:00:00.000Z", diff --git a/packages/app/src/utils/agent-directory-sync.ts b/packages/app/src/utils/agent-directory-sync.ts index 24e14c4208..e096c08036 100644 --- a/packages/app/src/utils/agent-directory-sync.ts +++ b/packages/app/src/utils/agent-directory-sync.ts @@ -18,11 +18,10 @@ export type AgentDirectoryDelta = Extract< export function applyAgentDirectoryDelta(input: { serverId: string; delta: AgentDirectoryDelta }): { agentId: string; - stoppedRunning: boolean; } { if (input.delta.kind === "remove") { removeAgentDirectoryReplica(input.serverId, input.delta.agentId); - return { agentId: input.delta.agentId, stoppedRunning: false }; + return { agentId: input.delta.agentId }; } return upsertAgentDirectoryReplica(input.serverId, input.delta); } @@ -32,7 +31,7 @@ type AgentUpsertDelta = Extract; function upsertAgentDirectoryReplica( serverId: string, delta: AgentUpsertDelta, -): { agentId: string; stoppedRunning: boolean } { +): { agentId: string } { const normalized = normalizeAgentSnapshot(delta.agent, serverId); const session = useSessionStore.getState().sessions[serverId]; const previousAgent = @@ -59,7 +58,6 @@ function upsertAgentDirectoryReplica( useSessionStore.getState().setAgentLastActivity(acceptedAgent.id, acceptedAgent.lastActivityAt); return { agentId: acceptedAgent.id, - stoppedRunning: previousAgent?.status === "running" && acceptedAgent.status !== "running", }; }