diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 847000fc6..a4d612506 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -120,7 +120,6 @@ interface BufferedAudioChunk { interface WorkspaceHydrationSnapshot { workspaces: Map; emptyProjects: Map; - deltas: readonly WorkspaceUpdatePayload[]; } interface WorkspaceHydrationTransaction { @@ -130,6 +129,10 @@ interface WorkspaceHydrationTransaction { deltas: WorkspaceUpdatePayload[]; } +function isWorkspaceHydrationCancelled(isCancelled?: () => boolean): boolean { + return isCancelled?.() ?? false; +} + type WorkspaceUpdatePayload = Extract< SessionOutboundMessage, { type: "workspace_update" } @@ -181,7 +184,6 @@ async function fetchWorkspaceHydrationSnapshot(input: { return { workspaces: new Map(input.transaction.workspaces), emptyProjects, - deltas: [...input.transaction.deltas], }; } @@ -406,11 +408,6 @@ function finalizeTimelineApplication(input: { if (shouldMarkAuthoritativeHistoryApplied) { setAgentAuthoritativeHistoryApplied(serverId, agentId, true); useCreateFlowStore.getState().clearByAgent({ serverId, agentId }); - } - if (result.initResolution === "resolve") { - resolveInitDeferred(initKey); - } - if (result.clearInitializing) { markAgentHistorySynchronized(serverId, agentId); const session = useSessionStore.getState().sessions[serverId]; const agent = session?.agents.get(agentId) ?? session?.agentDetails.get(agentId); @@ -418,6 +415,9 @@ function finalizeTimelineApplication(input: { getHostRuntimeStore().drainQueuedAgentMessage(serverId, agentId); } } + if (result.initResolution === "resolve") { + resolveInitDeferred(initKey); + } } function applyToolResultToMessages( @@ -603,7 +603,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } throw error; } - if (!snapshot || options?.isCancelled?.()) { + if (!snapshot || isWorkspaceHydrationCancelled(options?.isCancelled)) { if (workspaceHydrationRef.current === transaction) workspaceHydrationRef.current = null; return; } @@ -620,18 +620,25 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider return; } + if ( + workspaceHydrationRef.current !== transaction || + isWorkspaceHydrationCancelled(options?.isCancelled) + ) { + return; + } + const deltas = [...transaction.deltas]; + workspaceHydrationRef.current = null; setWorkspaces( serverId, reconcileWorkspaceDirectory({ serverId, snapshot: snapshot.workspaces, - deltas: snapshot.deltas, + deltas, }), ); setEmptyProjects(serverId, snapshot.emptyProjects.values()); setHasHydratedWorkspaces(serverId, true); - for (const delta of snapshot.deltas) applyWorkspaceUpdatePayload(delta); - if (workspaceHydrationRef.current === transaction) workspaceHydrationRef.current = null; + for (const delta of deltas) applyWorkspaceUpdatePayload(delta); }, [ applyWorkspaceUpdatePayload, diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index 77e78c184..dcf938378 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -13,6 +13,7 @@ import { useStoreWithEqualityFn } from "zustand/traditional"; import { AgentStreamView, type AgentStreamViewHandle } from "@/agent-stream/view"; import { ArchivedAgentCallout } from "@/components/archived-agent-callout"; import { FileDropZone } from "@/components/file-drop/file-drop-zone"; +import { useRetainedPanelActive } from "@/components/retained-panel"; import { Composer } from "@/composer"; import { AgentModeControl } from "@/composer/agent-controls/mode-control"; import { RewindComposerRestoreProvider } from "@/components/rewind/composer-restore"; @@ -701,6 +702,7 @@ function ChatAgentContent({ onOpenWorkspaceFile?: (request: WorkspaceFileOpenRequest) => void; }) { const { t } = useTranslation(); + const isPaneVisible = useRetainedPanelActive(); const { api: toastApi, toast: toastState, dismiss: dismissToast } = useToastHost(); const { isArchivingAgent } = useArchiveAgent(); const streamViewRef = useRef(null); @@ -912,7 +914,7 @@ function ChatAgentContent({ } return; } - if (!isPaneFocused || !isConnected || !hasSession) { + if (!isPaneVisible || !isConnected || !hasSession) { return; } if ( @@ -973,7 +975,7 @@ function ChatAgentContent({ ensureAgentIsInitialized, hasSession, isConnected, - isPaneFocused, + isPaneVisible, missingAgentState.kind, serverId, ]); diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index db64b571d..cf570292d 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -34,6 +34,7 @@ class FakeDaemonClient { Awaited> | ReturnType > = []; public sentAgentMessages: Array> = []; + public sendAgentMessageFailures: Error[] = []; private agentUpdateListeners = new Set< (message: Extract) => void >(); @@ -80,6 +81,8 @@ class FakeDaemonClient { async sendAgentMessage(...args: Parameters): Promise { this.sentAgentMessages.push(args); for (const waiter of this.sentMessageWaiters) waiter(); + const failure = this.sendAgentMessageFailures.shift(); + if (failure) throw failure; } async waitForSentMessages(count: number): Promise { @@ -2105,6 +2108,51 @@ describe("HostRuntimeStore", () => { 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(); + fakeClient.sendAgentMessageFailures.push(new Error("connection lost")); + 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_failed_queue_drain", + }, + }); + const sessionStore = useSessionStore.getState(); + sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + sessionStore.setQueuedMessages( + host.serverId, + new Map([ + [ + "agent", + [ + { id: "first", text: "retry me", attachments: [] }, + { id: "second", text: "keep me behind", attachments: [] }, + ], + ], + ]), + ); + + store.drainQueuedAgentMessage(host.serverId, "agent"); + + await vi.waitFor(() => { + expect(fakeClient.sentAgentMessages).toHaveLength(1); + expect( + useSessionStore.getState().sessions[host.serverId]?.queuedMessages.get("agent"), + ).toEqual([ + { id: "first", text: "retry me", attachments: [] }, + { id: "second", text: "keep me behind", attachments: [] }, + ]); + }); + + useSessionStore.getState().clearSession(host.serverId); + }); + it("applies buffered stale side effects from the accepted page agent", async () => { const host = makeHost({ serverId: "srv_buffered_stale_side_effects" }); const fakeClient = new FakeDaemonClient(); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index c303b2b58..6d80f87ea 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -61,6 +61,7 @@ import { } from "@/data/push-router"; import { mountBrowserAutomationDaemonClientHandler } from "@/browser-automation/handler"; import { schedulesQueryBaseKey } from "@/schedules/aggregated-schedules"; +import { sendQueuedComposerMessageNow } from "@/composer/actions"; import { splitComposerAttachmentsForSubmit } from "@/composer/attachments/submit"; import { encodeImages } from "@/utils/encode-images"; @@ -2126,31 +2127,38 @@ export class HostRuntimeStore { const store = useSessionStore.getState(); const session = store.sessions[serverId]; const queue = session?.queuedMessages.get(agentId); - if (!session?.client || !queue?.length || session.initializingAgents.get(agentId) === true) { + const client = session?.client; + if (!client || !queue?.length || session.initializingAgents.get(agentId) === true) { return; } - const [next, ...rest] = queue; - const wirePayload = splitComposerAttachmentsForSubmit(next.attachments); - store.setQueuedMessages(serverId, (current) => { - const updated = new Map(current); - updated.set(agentId, rest); - return updated; - }); - void encodeImages(wirePayload.images) - .then((images) => - session.client?.sendAgentMessage(agentId, next.text, { + const next = queue[0]; + void sendQueuedComposerMessageNow({ + agentId, + messageId: next.id, + queue: { + read: (queuedAgentId) => + useSessionStore.getState().sessions[serverId]?.queuedMessages.get(queuedAgentId) ?? [], + write: (update) => useSessionStore.getState().setQueuedMessages(serverId, update), + }, + submitMessage: async ({ text, attachments }) => { + const wirePayload = splitComposerAttachmentsForSubmit(attachments); + const images = await encodeImages(wirePayload.images); + await client.sendAgentMessage(agentId, text, { messageId: next.id, ...(images && images.length > 0 ? { images } : {}), attachments: wirePayload.attachments, - }), - ) - .catch((error) => { + }); + }, + }).then((result) => { + if (result.status === "failed") { console.error("[HostRuntime] failed to drain queued agent message", { serverId, agentId, - error: toErrorMessage(error), + error: result.errorMessage, }); - }); + } + return result; + }); } getSnapshot(serverId: string): HostRuntimeSnapshot | null {