diff --git a/apps/examples/desktop-app/sidecar/cloud-sessions.test.ts b/apps/examples/desktop-app/sidecar/cloud-sessions.test.ts index 2967d9db80..aab82e3eae 100644 --- a/apps/examples/desktop-app/sidecar/cloud-sessions.test.ts +++ b/apps/examples/desktop-app/sidecar/cloud-sessions.test.ts @@ -67,6 +67,7 @@ class FakeHubClient { onFailedSend?: () => void; commandHook?: (command: string) => void | Promise; invalidMessagesSnapshot = false; + malformedQueueReply = false; listedModel?: string; messages: unknown[] = [{ role: "user", content: "hi" }]; prompts: Array> = [ @@ -146,6 +147,9 @@ class FakeHubClient { payload: { approvals: this.pendingApprovals }, }; } + if (command === "session.pending_prompts" && this.malformedQueueReply) { + return { ok: true, payload: {} }; + } if ( command === "session.pending_prompts" || command === "session.update_pending_prompt" || @@ -944,6 +948,50 @@ describe("CloudSessionManager", () => { ).toBe(true); }); + it("keeps buffered queue state when the queue snapshot reply is malformed", async () => { + const { ctx } = createContext(); + const hub = new FakeHubClient(); + hub.malformedQueueReply = true; + // A queue event arrives while the rehydration snapshot is in flight, + // so it lands in the reconnect buffer. + hub.commandHook = async (command) => { + if (command !== "session.messages") return; + hub.events?.({ + version: "v1", + event: "session.pending_prompts", + eventId: "evt-queue-buffered", + timestamp: Date.now(), + sessionId: "inner-1", + payload: { + prompts: [ + { + id: "q-buffered", + prompt: "still queued", + delivery: "queue", + attachmentCount: 0, + }, + ], + }, + }); + }; + const manager = new CloudSessionManager(ctx, { + api: { list: async () => [REMOTE_SESSION] } as CloudSessionApi, + apiBaseUrl: "https://api.example", + getAuthToken: async () => "workos:fresh", + createHubClient: () => hub as never, + }); + await manager.list(); + await manager.attach("ses-outer"); + // Reading with an unknown transcript forces the rehydration snapshot. + await manager.readMessages("ses-outer"); + + // The malformed reply must not count as an authoritative snapshot; + // the buffered queue event is replayed and keeps the queued prompt. + expect(ctx.liveSessions.get("ses-outer")?.promptsInQueue).toMatchObject([ + { id: "q-buffered", prompt: "still queued" }, + ]); + }); + it("queues one rerun when a second sync overlaps the active snapshot", async () => { const { ctx } = createContext(); const hub = new FakeHubClient(); diff --git a/apps/examples/desktop-app/sidecar/cloud-sessions.ts b/apps/examples/desktop-app/sidecar/cloud-sessions.ts index 0c427f24cb..4174fbecbe 100644 --- a/apps/examples/desktop-app/sidecar/cloud-sessions.ts +++ b/apps/examples/desktop-app/sidecar/cloud-sessions.ts @@ -1432,7 +1432,15 @@ export class CloudSessionManager { } } } - const prompts = queueReply + // A reply is only an authoritative queue snapshot when it succeeded + // and actually carries a prompts array; treating an unsuccessful or + // malformed reply as authoritative would publish an empty queue and + // drop the buffered queue events that still hold the real state. + const queueSnapshotValid = + queueReply !== undefined && + queueReply.ok !== false && + Array.isArray(queueReply.payload?.prompts); + const prompts = queueSnapshotValid ? this.applyQueueSnapshot(outerSessionId, queueReply) : undefined; await this.refreshPendingApprovals(outerSessionId, connection); @@ -1454,7 +1462,7 @@ export class CloudSessionManager { const buffered = reconcileBufferedCloudEvents( connection.bufferedEvents, messages, - { queueSnapshotApplied: Boolean(queueReply) }, + { queueSnapshotApplied: queueSnapshotValid }, ); connection.bufferedEvents = []; connection.bufferingEvents = false;