From e4dc91eb0fb2c1bec8f4460abf6fbff0cf9c584e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 6 Aug 2026 11:15:16 +0100 Subject: [PATCH] fix(web): gate session hydration on subscription ack (#2287) (#2296) --- .../task/chat/message-list-native.test.tsx | 126 ++++++--- .../task/chat/message-list-native.tsx | 44 ++- apps/web/e2e/pages/session-page.ts | 58 ++-- .../tests/chat/mobile-unread-divider.spec.ts | 2 +- apps/web/e2e/tests/chat/monitor.spec.ts | 151 ++++++---- .../web/e2e/tests/chat/unread-divider.spec.ts | 16 +- .../session-stream-overload-isolation.spec.ts | 118 +++++++- .../hooks/domains/session/message-backfill.ts | 97 +++++++ .../session/use-session-message-fetch.test.ts | 66 +++++ .../session/use-session-message-fetch.ts | 96 +++++++ .../session/use-session-messages.test.ts | 112 +++++++- .../domains/session/use-session-messages.ts | 266 +++++++----------- .../use-session-subscription-retry.test.ts | 44 ++- .../session/use-session-subscription-retry.ts | 20 +- .../session/use-visibility-backfill.test.ts | 41 ++- apps/web/lib/ws/client.test.ts | 223 +++++++++++++++ apps/web/lib/ws/client.ts | 161 +++++++++-- .../session-subscription-recovery/plan.md | 138 +++++++++ .../task-01-ack-gated-session-hydration.md | 100 +++++++ docs/specs/INDEX.md | 1 + .../session-subscription-recovery/spec.md | 61 ++++ 21 files changed, 1548 insertions(+), 393 deletions(-) create mode 100644 apps/web/hooks/domains/session/message-backfill.ts create mode 100644 apps/web/hooks/domains/session/use-session-message-fetch.test.ts create mode 100644 apps/web/hooks/domains/session/use-session-message-fetch.ts create mode 100644 apps/web/lib/ws/client.test.ts create mode 100644 docs/plans/session-subscription-recovery/plan.md create mode 100644 docs/plans/session-subscription-recovery/task-01-ack-gated-session-hydration.md create mode 100644 docs/specs/session-subscription-recovery/spec.md diff --git a/apps/web/components/task/chat/message-list-native.test.tsx b/apps/web/components/task/chat/message-list-native.test.tsx index eaf336d815..2350b86548 100644 --- a/apps/web/components/task/chat/message-list-native.test.tsx +++ b/apps/web/components/task/chat/message-list-native.test.tsx @@ -1,5 +1,5 @@ -import { useRef } from "react"; -import { render } from "@testing-library/react"; +import { useLayoutEffect, useRef } from "react"; +import { cleanup, render } from "@testing-library/react"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { Message } from "@/lib/types/http"; @@ -21,6 +21,8 @@ import { useScrollToDividerOrBottom } from "./message-list-native"; import { useAutoScroll } from "./message-list-native-scroll"; const DIVIDER_KEY = "m2"; +const DIVIDER_SCROLL_CONTAINER_TEST_ID = "divider-scroll-container"; +const MISSING_SCROLL_CONTAINER_ERROR = "scroll container did not render"; const TEST_MESSAGES = [{} as Message]; const NEVER_LOCKED = () => false; @@ -29,28 +31,57 @@ function Harness({ anchoredBarOffsetPx, dividerKey = DIVIDER_KEY, onDividerScroll, + scrollLayoutKey = "initial", + dividerDocumentTop = 250, }: { itemCount: number; anchoredBarOffsetPx: number; dividerKey?: string | null; onDividerScroll?: () => void; + scrollLayoutKey?: string; + dividerDocumentTop?: number; }) { const scrollRef = useRef(null); - useScrollToDividerOrBottom( - scrollRef, - itemCount, - dividerKey, - anchoredBarOffsetPx, + useLayoutEffect(() => { + const scrollContainer = scrollRef.current; + if (!scrollContainer) return; + Object.defineProperty(scrollContainer, "getBoundingClientRect", { + configurable: true, + value: () => createRect(100, 400), + }); + const divider = scrollContainer.querySelector(`[id="msg-${DIVIDER_KEY}"]`); + if (!divider) return; + Object.defineProperty(divider, "getBoundingClientRect", { + configurable: true, + value: () => createRect(dividerDocumentTop - scrollContainer.scrollTop, 20), + }); + }, [dividerDocumentTop]); + useScrollToDividerOrBottom(scrollRef, itemCount, dividerKey, anchoredBarOffsetPx, { onDividerScroll, - ); + scrollLayoutKey, + }); return ( -
+
); } +function createRect(top: number, height: number): DOMRect { + return { + x: 0, + y: top, + top, + bottom: top + height, + left: 0, + right: 100, + width: 100, + height, + toJSON: () => ({}), + } as DOMRect; +} + function AutoScrollHarness({ isWorking, hasUnreadDivider, @@ -82,29 +113,48 @@ function setScrollMetrics(element: HTMLElement) { } afterEach(() => { + cleanup(); vi.restoreAllMocks(); }); +// eslint-disable-next-line max-lines-per-function -- this suite keeps the related scroll invariants together. describe("useScrollToDividerOrBottom — anchored-bar offset", () => { - it("re-scrolls the divider into view once the anchored bar's measured height arrives, so it lands below the pinned bar instead of under it", () => { - const scrollIntoView = vi.fn(); - HTMLElement.prototype.scrollIntoView = scrollIntoView; - + it("re-scrolls the divider when the anchored bar's measured height arrives", () => { const { rerender } = render(); - expect(scrollIntoView).toHaveBeenCalledTimes(1); + const scrollContainer = document.querySelector( + `[data-testid="${DIVIDER_SCROLL_CONTAINER_TEST_ID}"]`, + ); + if (!scrollContainer) throw new Error(MISSING_SCROLL_CONTAINER_ERROR); + expect(scrollContainer.scrollTop).toBe(150); - // The anchored bar's real height resolves a tick after mount (async - // ResizeObserver report) — the divider must be re-placed now that the - // scroll-margin backing it is no longer 0, or it stays hidden under the - // bar for good (didScrollToDivider already latched true). rerender(); - expect(scrollIntoView).toHaveBeenCalledTimes(2); - expect(scrollIntoView).toHaveBeenLastCalledWith({ block: "start" }); + expect(scrollContainer.scrollTop).toBe(74); + }); + + it("reasserts the divider after a loading layout shift with the same item count", () => { + const { rerender } = render( + , + ); + const scrollContainer = document.querySelector( + `[data-testid="${DIVIDER_SCROLL_CONTAINER_TEST_ID}"]`, + ); + if (!scrollContainer) throw new Error(MISSING_SCROLL_CONTAINER_ERROR); + expect(scrollContainer.scrollTop).toBe(150); + + rerender( + , + ); + + expect(scrollContainer.scrollTop).toBe(66); }); it("resynchronizes auto-scroll state after placing the divider", () => { - HTMLElement.prototype.scrollIntoView = vi.fn(); const onDividerScroll = vi.fn(); render(); @@ -152,47 +202,51 @@ describe("useScrollToDividerOrBottom — anchored-bar offset", () => { }); it("never re-scrolls once the reader has started scrolling, even if the anchored bar's height changes afterward", () => { - const scrollIntoView = vi.fn(); - HTMLElement.prototype.scrollIntoView = scrollIntoView; - const { rerender, container } = render(); - expect(scrollIntoView).toHaveBeenCalledTimes(1); + const scrollContainer = container.querySelector( + `[data-testid="${DIVIDER_SCROLL_CONTAINER_TEST_ID}"]`, + ); + if (!scrollContainer) throw new Error(MISSING_SCROLL_CONTAINER_ERROR); + expect(scrollContainer.scrollTop).toBe(150); - container.querySelector("div")?.dispatchEvent(new Event("wheel", { bubbles: true })); + scrollContainer.dispatchEvent(new Event("wheel", { bubbles: true })); rerender(); - expect(scrollIntoView).toHaveBeenCalledTimes(1); + expect(scrollContainer.scrollTop).toBe(150); }); it("never scrolls the divider when there is no unread boundary, regardless of anchored-bar height changes", () => { - const scrollIntoView = vi.fn(); - HTMLElement.prototype.scrollIntoView = scrollIntoView; - const { rerender } = render( , ); - expect(scrollIntoView).not.toHaveBeenCalled(); + const scrollContainer = document.querySelector( + `[data-testid="${DIVIDER_SCROLL_CONTAINER_TEST_ID}"]`, + ); + if (!scrollContainer) throw new Error(MISSING_SCROLL_CONTAINER_ERROR); + expect(scrollContainer.scrollTop).toBe(0); rerender(); - expect(scrollIntoView).not.toHaveBeenCalled(); + expect(scrollContainer.scrollTop).toBe(0); }); it("stops re-scrolling once the settling window has elapsed, even without any user interaction", () => { vi.useFakeTimers(); - const scrollIntoView = vi.fn(); - HTMLElement.prototype.scrollIntoView = scrollIntoView; const { rerender } = render(); - expect(scrollIntoView).toHaveBeenCalledTimes(1); + const scrollContainer = document.querySelector( + `[data-testid="${DIVIDER_SCROLL_CONTAINER_TEST_ID}"]`, + ); + if (!scrollContainer) throw new Error(MISSING_SCROLL_CONTAINER_ERROR); + expect(scrollContainer.scrollTop).toBe(150); // Past the 4s settling window (e.g. a scrollbar drag with no // wheel/touch/key event to catch — the correction must freeze anyway). vi.advanceTimersByTime(4001); rerender(); - expect(scrollIntoView).toHaveBeenCalledTimes(1); + expect(scrollContainer.scrollTop).toBe(150); vi.useRealTimers(); }); }); diff --git a/apps/web/components/task/chat/message-list-native.tsx b/apps/web/components/task/chat/message-list-native.tsx index 8c25088fc4..851f4dd285 100644 --- a/apps/web/components/task/chat/message-list-native.tsx +++ b/apps/web/components/task/chat/message-list-native.tsx @@ -89,6 +89,13 @@ type NativeMessageListScrollParams = { onLastPromptEdgeChange: ((edge: LastPromptEdge) => void) | undefined; firstMessageId: string | null | undefined; onFirstMessageHiddenChange: ((isHidden: boolean) => void) | undefined; + /** Changes when transcript status rows can add/remove space above messages. */ + scrollLayoutKey: string; +}; + +type ScrollToDividerOptions = { + onDividerScroll?: () => void; + scrollLayoutKey?: string; }; function useNativeMessageListScroll(params: NativeMessageListScrollParams) { @@ -109,6 +116,7 @@ function useNativeMessageListScroll(params: NativeMessageListScrollParams) { onLastPromptEdgeChange, firstMessageId, onFirstMessageHiddenChange, + scrollLayoutKey, } = params; const { handleScrollToMessage, sentinelRef, markNotNearBottom } = useNativeScrollManagement({ scrollRef, @@ -126,13 +134,10 @@ function useNativeMessageListScroll(params: NativeMessageListScrollParams) { useEffect(() => { scrollRef.current?.style.setProperty("--anchored-bar-h", `${anchoredBarOffsetPx}px`); }, [anchoredBarOffsetPx]); - useScrollToDividerOrBottom( - scrollRef, - items.length, - dividerBeforeItemKey, - anchoredBarOffsetPx, - markNotNearBottom, - ); + useScrollToDividerOrBottom(scrollRef, items.length, dividerBeforeItemKey, anchoredBarOffsetPx, { + onDividerScroll: markNotNearBottom, + scrollLayoutKey, + }); useImperativeHandle(ref, () => ({ scrollToMessage: handleScrollToMessage }), [ handleScrollToMessage, ]); @@ -259,7 +264,9 @@ type NativeMessageListBodyProps = { * - didScrollToDivider and didInitialScroll are separate latches so the * bottom-fallback firing first (before dividerBeforeItemKey resolves) * doesn't block the divider correction from still applying once it - * does. Embedded, always-invisible previews (isVisible hardcoded false, + * does. The caller also supplies a layout key so a loading-state transition + * with an unchanged item count can re-assert the target. Embedded, always- + * invisible previews (isVisible hardcoded false, * see TaskChatPanel) never resolve a divider, so they keep the * original, unconditional scroll-to-bottom-on-mount behavior untouched. */ @@ -268,8 +275,9 @@ export function useScrollToDividerOrBottom( itemCount: number, dividerBeforeItemKey: string | null | undefined, anchoredBarOffsetPx: number, - onDividerScroll?: () => void, + options: ScrollToDividerOptions = {}, ) { + const { onDividerScroll, scrollLayoutKey = "" } = options; const isUserScrollingRef = useRef(false); useEffect(() => { const el = scrollRef.current; @@ -313,7 +321,13 @@ export function useScrollToDividerOrBottom( if (useDockviewStore.getState().pendingChatScrollTop === null) { const dividerEl = el.querySelector(`[id="msg-${dividerBeforeItemKey}"]`); if (dividerEl) { - dividerEl.scrollIntoView({ block: "start" }); + // scrollIntoView aligns against the viewport, which puts the target + // behind the fixed mobile session header instead of inside this + // nested scroll container. Move by the relative geometry instead; + // the desktop anchored prompt bar still reserves its measured height. + const containerRect = el.getBoundingClientRect(); + const dividerRect = dividerEl.getBoundingClientRect(); + el.scrollTop += dividerRect.top - containerRect.top - anchoredBarOffsetPx; onDividerScroll?.(); didScrollToDivider.current = true; didInitialScroll.current = true; @@ -330,7 +344,7 @@ export function useScrollToDividerOrBottom( } el.scrollTop = el.scrollHeight; didInitialScroll.current = true; - }, [itemCount, dividerBeforeItemKey, anchoredBarOffsetPx, onDividerScroll]); + }, [itemCount, dividerBeforeItemKey, anchoredBarOffsetPx, onDividerScroll, scrollLayoutKey]); } /** Sentinel, status/footer, and transcript rows — everything below the @@ -474,6 +488,14 @@ export const NativeMessageList = memo( onLastPromptEdgeChange, firstMessageId, onFirstMessageHiddenChange, + scrollLayoutKey: [ + messagesLoading, + isInitialLoading, + showLoadingState, + isLoadingMore, + hasMore, + isWorking, + ].join(":"), }); return ( diff --git a/apps/web/e2e/pages/session-page.ts b/apps/web/e2e/pages/session-page.ts index a766299779..b2ec2705dc 100644 --- a/apps/web/e2e/pages/session-page.ts +++ b/apps/web/e2e/pages/session-page.ts @@ -173,26 +173,16 @@ export class SessionPage { /** * Wait for the chat to be idle (input placeholder visible, agent not busy). * - * On mobile-chrome (and occasionally desktop), there's a WS subscribe race: - * a fresh task auto-starts its agent, the mock agent completes in <1s, and - * the session_state transition (RUNNING -> AWAITING_INPUT) can fan out - * before the client's WS subscription registers server-side. The client - * then sits with `isAgentBusy=true` forever and the idle placeholder - * never renders. SSR picks up the right state on the next page load, so - * one targeted reload-and-retry is enough to recover. + * A fresh task can still miss the first persisted session-state transition + * while the page hydrates under CI load. Re-drive SSR hydration periodically + * so a stale busy state does not consume the whole idle wait budget. * * After a backend restart, auto-resume can briefly surface the recovery * prompt ("Environment setup failed"); click through it when visible. - * - * This is the same race the office agent-run-live spec rides out with - * `expect.poll`-based re-seeding. */ - async waitForChatIdle( - opts: { timeout?: number; attemptTimeout?: number; requireEditable?: boolean } = {}, - ) { + async waitForChatIdle(opts: { timeout?: number; requireEditable?: boolean } = {}) { const softTotalTimeout = opts.timeout ?? 45_000; - const attemptTimeout = - opts.attemptTimeout ?? Math.min(15_000, Math.max(5_000, Math.floor(softTotalTimeout / 3))); + const attemptTimeout = Math.min(15_000, Math.max(5_000, Math.floor(softTotalTimeout / 3))); const pollSlice = 1_500; const idle = this.anyIdleInput(); const editor = this.activeChat().locator(".tiptap.ProseMirror:visible").first(); @@ -215,10 +205,6 @@ export class SessionPage { const now = Date.now(); const remaining = Math.max(1, softTotalTimeout - (now - start)); - // Re-drive SSR hydration once per attemptTimeout slice (not just once): - // under CI shard load a single reload isn't always enough for the - // idle-input state to hydrate. Only reload while enough budget remains - // for the reloaded page to settle. if (now - lastReloadAt >= attemptTimeout && remaining > pollSlice) { lastReloadAt = now; await this.page.reload(); @@ -228,9 +214,15 @@ export class SessionPage { continue; } - await idle - .waitFor({ state: "visible", timeout: Math.min(pollSlice, remaining) }) - .catch(() => undefined); + const timeout = Math.min(pollSlice, remaining); + if (opts.requireEditable) { + await expect + .poll(isReady, { timeout }) + .toBe(true) + .catch(() => undefined); + } else { + await idle.waitFor({ state: "visible", timeout }).catch(() => undefined); + } } // Final bounded check: still throws on a genuinely stuck session, but gives @@ -1091,27 +1083,11 @@ export class SessionPage { return editor; } - /** - * Wait for the agent reply containing `text` at the given 0-based match - * `index` to be visible after a follow-up prompt. On first timeout, reload - * once so SSR re-fetches the persisted turn, then re-assert. - * - * This rides out the same WS-subscribe race `waitForChatIdle` handles, but - * for the reply message itself: a mid-session prompt's response event can be - * dropped when the client's WS subscription loses the race with the agent's - * reply (common after repeated restart/resume cycles). The reply is persisted - * server-side, so a single reload recovers it. - */ + /** Wait for the agent reply containing `text` at the given 0-based match `index`. */ async expectChatResponseVisible(text: string, index = 0, opts: { timeout?: number } = {}) { const timeout = opts.timeout ?? 30_000; - const target = () => this.chat.getByText(text, { exact: false }).nth(index); - try { - await expect(target()).toBeVisible({ timeout }); - } catch { - await this.page.reload(); - await this.waitForLoad(); - await expect(target()).toBeVisible({ timeout }); - } + const target = () => this.activeChat().getByText(text, { exact: false }).nth(index); + await expect(target()).toBeVisible({ timeout }); } /** Toggle plan mode on/off by clicking the plan mode toggle button in the toolbar. diff --git a/apps/web/e2e/tests/chat/mobile-unread-divider.spec.ts b/apps/web/e2e/tests/chat/mobile-unread-divider.spec.ts index 1beedae202..3c22f2aca9 100644 --- a/apps/web/e2e/tests/chat/mobile-unread-divider.spec.ts +++ b/apps/web/e2e/tests/chat/mobile-unread-divider.spec.ts @@ -86,7 +86,7 @@ test.describe("Mobile unread divider", () => { await expect(session.activeChat().getByTestId("unread-divider")).toHaveCount(0); await session.sendMessageViaButton("prompt while actively reading"); - await session.waitForChatIdle({ timeout: 60_000, attemptTimeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); await expect(session.activeChat().getByTestId("unread-divider")).toHaveCount(0); }); }); diff --git a/apps/web/e2e/tests/chat/monitor.spec.ts b/apps/web/e2e/tests/chat/monitor.spec.ts index 12de63b756..f44780b006 100644 --- a/apps/web/e2e/tests/chat/monitor.spec.ts +++ b/apps/web/e2e/tests/chat/monitor.spec.ts @@ -1,5 +1,65 @@ import { test, expect } from "../../fixtures/test-base"; +import type { SeedData } from "../../fixtures/test-base"; +import type { ApiClient } from "../../helpers/api-client"; import { SessionPage } from "../../pages/session-page"; +import { attachGatewayTrafficCapture } from "../../helpers/ws-traffic"; + +async function prepareMonitorTask(apiClient: ApiClient, seedData: SeedData, title: string) { + const task = await apiClient.createTask(seedData.workspaceId, title, { + agent_profile_id: seedData.agentProfileId, + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }); + const prepared = await apiClient.launchSession({ + task_id: task.id, + agent_profile_id: seedData.agentProfileId, + executor_profile_id: seedData.worktreeExecutorProfileId, + workflow_step_id: seedData.startStepId, + prompt: "", + intent: "prepare", + launch_workspace: true, + }); + + await expect + .poll(async () => (await apiClient.getTaskEnvironment(task.id))?.status ?? null, { + timeout: 60_000, + message: "prepared Monitor task environment did not become ready", + }) + .toBe("ready"); + + return { task, sessionId: prepared.session_id }; +} + +async function waitForMonitorSubscription( + capture: ReturnType, + sessionId: string, +) { + await expect + .poll( + () => + capture.frames.some( + (frame) => + frame.direction === "sent" && + frame.action === "session.subscribe" && + frame.sessionId === sessionId, + ), + { timeout: 10_000, message: "Monitor page must subscribe before the script starts" }, + ) + .toBe(true); + await expect + .poll( + () => + capture.frames.some( + (frame) => + frame.direction === "received" && + frame.action === "session.subscribe" && + frame.sessionId === sessionId, + ), + { timeout: 30_000, message: "Monitor page did not acknowledge its session subscription" }, + ) + .toBe(true); +} // All Monitor scenarios drive the kandev backend with the mock-agent's new // e2e:monitor_* directives, which reproduce claude-agent-acp's wire format @@ -15,7 +75,7 @@ test.describe("Claude-acp Monitor tool", () => { apiClient, seedData, }) => { - test.setTimeout(60_000); + test.setTimeout(240_000); // Three monitor events fire over ~3s. The agent then ends the monitor and // produces a final assistant message so the turn completes deterministically. @@ -27,42 +87,38 @@ test.describe("Claude-acp Monitor tool", () => { 'e2e:monitor_event("task-1", "in_progress: lint")', "e2e:delay(200)", 'e2e:monitor_event("task-1", "success: lint")', + "e2e:delay(500)", 'e2e:monitor_end("task-1")', 'e2e:message("watching done")', ].join("\n"); - const task = await apiClient.createTaskWithAgent( - seedData.workspaceId, - "Monitor watching", - seedData.agentProfileId, - { - description: script, - workflow_id: seedData.workflowId, - workflow_step_id: seedData.startStepId, - repository_ids: [seedData.repositoryId], - }, - ); + const { task, sessionId } = await prepareMonitorTask(apiClient, seedData, "Monitor watching"); + const capture = attachGatewayTrafficCapture(testPage); await testPage.goto(`/t/${task.id}`); const session = new SessionPage(testPage); await session.waitForLoad(); await session.waitForChatIdle({ timeout: 30_000 }); + await waitForMonitorSubscription(capture, sessionId); + capture.frames.length = 0; + // The browser send path records the first user turn before launching the + // agent, so fast Monitor notifications have an active turn to update. + await session.sendMessageViaButton(script); // The dedicated Monitor card renders, not the generic tool_call row. - const monitorCard = session.chat.locator('[data-testid="monitor-card"]').first(); + const monitorCard = session.activeChat().locator('[data-testid="monitor-card"]'); + await expect(monitorCard).toHaveCount(1); await expect(monitorCard).toBeVisible(); await expect(monitorCard).toContainText("gh pr checks --watch"); // Event count badge surfaces all three events. Status pill flipped to // "ended" because the script issued monitor_end and the parent turn // completed. - await expect(monitorCard).toContainText("3 events"); - await expect(session.chat.locator('[data-testid="monitor-status-pill"]').first()).toContainText( - /ended/, - ); + await expect(monitorCard).toContainText("3 events", { timeout: 30_000 }); + await expect(monitorCard.getByTestId("monitor-status-pill")).toContainText(/ended/); // Each event body landed in the recent-events tail. - const eventList = session.chat.locator('[data-testid="monitor-event"]'); + const eventList = session.activeChat().locator('[data-testid="monitor-event"]'); await expect(eventList).toHaveCount(3); await expect(eventList.nth(0)).toContainText("queued: lint"); await expect(eventList.nth(2)).toContainText("success: lint"); @@ -70,8 +126,8 @@ test.describe("Claude-acp Monitor tool", () => { // Critical: the model's "Human: " echo must NOT appear // anywhere in the chat. The adapter strips matched envelopes from the // assistant text and drops orphan "Human:" prefixes entirely. - await expect(session.chat).not.toContainText(""); - await expect(session.chat).not.toContainText("Human: "); + await expect(session.activeChat()).not.toContainText("Human: { apiClient, seedData, }) => { - test.setTimeout(45_000); + test.setTimeout(240_000); const script = [ 'e2e:monitor_start("task-1", "tail -f log")', - "e2e:delay(100)", + "e2e:delay(500)", 'e2e:monitor_event("task-1", "first and only line")', + "e2e:delay(500)", 'e2e:monitor_end("task-1")', 'e2e:message("done")', ].join("\n"); - const task = await apiClient.createTaskWithAgent( - seedData.workspaceId, - "Monitor singular", - seedData.agentProfileId, - { - description: script, - workflow_id: seedData.workflowId, - workflow_step_id: seedData.startStepId, - repository_ids: [seedData.repositoryId], - }, - ); + const { task, sessionId } = await prepareMonitorTask(apiClient, seedData, "Monitor singular"); + const capture = attachGatewayTrafficCapture(testPage); await testPage.goto(`/t/${task.id}`); const session = new SessionPage(testPage); await session.waitForLoad(); await session.waitForChatIdle({ timeout: 30_000 }); - - const card = session.chat.locator('[data-testid="monitor-card"]').first(); - await expect(card).toContainText("1 event"); + await waitForMonitorSubscription(capture, sessionId); + await session.sendMessageViaButton(script); + const card = session.activeChat().locator('[data-testid="monitor-card"]'); + await expect(card).toHaveCount(1); + await expect(card).toContainText("1 event", { timeout: 30_000 }); await expect(card).not.toContainText("1 events"); }); @@ -116,7 +166,7 @@ test.describe("Claude-acp Monitor tool", () => { apiClient, seedData, }) => { - test.setTimeout(60_000); + test.setTimeout(240_000); const script = [ 'e2e:monitor_start("task-1", "wait for ci")', @@ -124,39 +174,34 @@ test.describe("Claude-acp Monitor tool", () => { 'e2e:monitor_event("task-1", "step-1")', "e2e:delay(150)", 'e2e:monitor_event("task-1", "step-2")', + "e2e:delay(500)", 'e2e:monitor_end("task-1")', 'e2e:message("done")', ].join("\n"); - const task = await apiClient.createTaskWithAgent( - seedData.workspaceId, - "Monitor reload", - seedData.agentProfileId, - { - description: script, - workflow_id: seedData.workflowId, - workflow_step_id: seedData.startStepId, - repository_ids: [seedData.repositoryId], - }, - ); + const { task, sessionId } = await prepareMonitorTask(apiClient, seedData, "Monitor reload"); + const capture = attachGatewayTrafficCapture(testPage); await testPage.goto(`/t/${task.id}`); const session = new SessionPage(testPage); await session.waitForLoad(); await session.waitForChatIdle({ timeout: 30_000 }); - await expect(session.chat.locator('[data-testid="monitor-card"]').first()).toContainText( - "2 events", - ); + await waitForMonitorSubscription(capture, sessionId); + await session.sendMessageViaButton(script); + const cards = session.activeChat().locator('[data-testid="monitor-card"]'); + await expect(cards).toHaveCount(1); + await expect(cards).toContainText("2 events", { timeout: 30_000 }); // Reload — SSR + Zustand hydration must reconstitute the card from DB. await testPage.reload(); await session.waitForLoad(); await session.waitForChatIdle({ timeout: 30_000 }); - const card = session.chat.locator('[data-testid="monitor-card"]').first(); + const card = session.activeChat().locator('[data-testid="monitor-card"]'); + await expect(card).toHaveCount(1); await expect(card).toBeVisible(); await expect(card).toContainText("wait for ci"); await expect(card).toContainText("2 events"); - await expect(session.chat.locator('[data-testid="monitor-event"]')).toHaveCount(2); + await expect(session.activeChat().locator('[data-testid="monitor-event"]')).toHaveCount(2); }); }); diff --git a/apps/web/e2e/tests/chat/unread-divider.spec.ts b/apps/web/e2e/tests/chat/unread-divider.spec.ts index 4ae684566e..5e67cbeb34 100644 --- a/apps/web/e2e/tests/chat/unread-divider.spec.ts +++ b/apps/web/e2e/tests/chat/unread-divider.spec.ts @@ -29,17 +29,7 @@ test.describe("Unread divider", () => { const sessionId = task.session_id; let session = await openTaskSession(testPage, task.id); - // attemptTimeout must be raised explicitly, not just `timeout` — its - // default is capped at min(15_000, ...) regardless of `timeout` (see - // SessionPage.waitForChatIdle). Past that cap it does a page.reload() - // to recover from a stuck WS subscription, which is a *real* - // navigate-away-and-back from this feature's own perspective — it - // would legitimately (and correctly) re-capture a fresh divider - // anchor, making later assertions in this test fail for a reason - // unrelated to what they're testing (the cursor advancing live while - // the session stays genuinely, continuously visible in one browser - // session). - await session.waitForChatIdle({ timeout: 60_000, attemptTimeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); // First-ever visit: there is no prior read cursor, so nothing renders as // "New" — but the cursor must still advance to the latest message so a @@ -57,7 +47,7 @@ test.describe("Unread divider", () => { // cursor keeps advancing live while the session stays in view, so this // alone must not produce a divider either. await session.sendMessageViaButton("second question"); - await session.waitForChatIdle({ timeout: 60_000, attemptTimeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); await expect(session.activeChat().getByTestId("unread-divider")).toHaveCount(0); const fullTranscript = await apiClient.listSessionMessages(sessionId); @@ -155,7 +145,7 @@ test.describe("Unread divider", () => { await expect(session.activeChat().getByTestId("unread-divider")).toHaveCount(0); await session.sendMessageViaButton("prompt while actively reading"); - await session.waitForChatIdle({ timeout: 60_000, attemptTimeout: 60_000 }); + await session.waitForChatIdle({ timeout: 60_000 }); await expect(session.activeChat().getByTestId("unread-divider")).toHaveCount(0); }); diff --git a/apps/web/e2e/tests/session/session-stream-overload-isolation.spec.ts b/apps/web/e2e/tests/session/session-stream-overload-isolation.spec.ts index 0b4cb6f373..450ad56553 100644 --- a/apps/web/e2e/tests/session/session-stream-overload-isolation.spec.ts +++ b/apps/web/e2e/tests/session/session-stream-overload-isolation.spec.ts @@ -19,52 +19,146 @@ test.describe("Session stream overload isolation", () => { test.setTimeout(180_000); const quietCapture = attachGatewayTrafficCapture(testPage); - const noisyTask = await apiClient.createTaskWithAgent( + const noisyTask = await apiClient.createTask( seedData.workspaceId, "Noisy reasoning isolation task", - seedData.agentProfileId, { - description: reasoningBurstPrompt(), + agent_profile_id: seedData.agentProfileId, workflow_id: seedData.workflowId, workflow_step_id: seedData.startStepId, repository_ids: [seedData.repositoryId], }, ); - if (!noisyTask.session_id) throw new Error("noisy task has no session"); + + const preparedNoisy = await apiClient.launchSession({ + task_id: noisyTask.id, + agent_profile_id: seedData.agentProfileId, + executor_profile_id: seedData.worktreeExecutorProfileId, + workflow_step_id: seedData.startStepId, + prompt: "", + intent: "prepare", + launch_workspace: true, + }); + const noisySessionId = preparedNoisy.session_id; + + await expect + .poll(async () => (await apiClient.getTaskEnvironment(noisyTask.id))?.status ?? null, { + timeout: 60_000, + message: "prepared noisy task environment did not become ready", + }) + .toBe("ready"); // Keep a real subscribed browser open for the noisy session while the // primary page navigates to the quiet task. This exercises the same - // per-client isolation boundary that failed during the incident. + // per-client isolation boundary that failed during the incident. The + // session is prepared first so a fast mock run cannot finish before the + // browser has registered its subscription. const noisyPage = await testPage.context().newPage(); const noisyCapture = attachGatewayTrafficCapture(noisyPage); await noisyPage.goto(`/t/${noisyTask.id}`); const noisySession = new SessionPage(noisyPage); await noisySession.waitForLoad(); + await noisySession.waitForChatIdle({ timeout: 60_000 }); + await expect + .poll( + () => + noisyCapture.frames.some( + (frame) => + frame.direction === "sent" && + frame.action === "session.subscribe" && + frame.sessionId === noisySessionId, + ), + { timeout: 10_000, message: "noisy page must subscribe before the burst" }, + ) + .toBe(true); + await expect + .poll( + () => + noisyCapture.frames.some( + (frame) => + frame.direction === "received" && + frame.action === "session.subscribe" && + frame.sessionId === noisySessionId, + ), + { timeout: 30_000, message: "noisy page did not acknowledge its session subscription" }, + ) + .toBe(true); - const quietTask = await apiClient.createTaskWithAgent( + const quietTask = await apiClient.createTask( seedData.workspaceId, "Quiet session isolation task", - seedData.agentProfileId, { - description: "/e2e:simple-message", + agent_profile_id: seedData.agentProfileId, workflow_id: seedData.workflowId, workflow_step_id: seedData.startStepId, repository_ids: [seedData.repositoryId], }, ); - if (!quietTask.session_id) throw new Error("quiet task has no session"); + + const preparedQuiet = await apiClient.launchSession({ + task_id: quietTask.id, + agent_profile_id: seedData.agentProfileId, + executor_profile_id: seedData.worktreeExecutorProfileId, + workflow_step_id: seedData.startStepId, + prompt: "", + intent: "prepare", + launch_workspace: true, + }); + const quietSessionId = preparedQuiet.session_id; + + await expect + .poll(async () => (await apiClient.getTaskEnvironment(quietTask.id))?.status ?? null, { + timeout: 60_000, + message: "prepared quiet task environment did not become ready", + }) + .toBe("ready"); await testPage.goto(`/t/${quietTask.id}`); const quietSession = new SessionPage(testPage); await quietSession.waitForLoad(); await quietSession.waitForChatIdle({ timeout: 60_000 }); + await expect + .poll( + () => + quietCapture.frames.some( + (frame) => + frame.direction === "sent" && + frame.action === "session.subscribe" && + frame.sessionId === quietSessionId, + ), + { timeout: 10_000, message: "quiet page must subscribe before the burst" }, + ) + .toBe(true); + await expect + .poll( + () => + quietCapture.frames.some( + (frame) => + frame.direction === "received" && + frame.action === "session.subscribe" && + frame.sessionId === quietSessionId, + ), + { timeout: 30_000, message: "quiet page did not acknowledge its session subscription" }, + ) + .toBe(true); + + noisyCapture.frames.length = 0; + quietCapture.frames.length = 0; + await apiClient.launchSession({ + task_id: noisyTask.id, + session_id: noisySessionId, + agent_profile_id: seedData.agentProfileId, + prompt: reasoningBurstPrompt(), + intent: "start_created", + }); + await quietSession.sendMessageViaButton("quiet-session-followup"); await quietSession.expectChatResponseVisible("quiet-session-followup", 0, { timeout: 60_000 }); await assertNoHorizontalOverflow(testPage, "session stream overload desktop surface"); - const exact = await waitForExactReasoningBurst(apiClient, noisyTask.session_id); - const noisyFrames = noisyReceivedFrames(noisyCapture.frames, noisyTask.session_id); - const quietSessionNoisyFrames = noisyReceivedFrames(quietCapture.frames, noisyTask.session_id); + const exact = await waitForExactReasoningBurst(apiClient, noisySessionId); + const noisyFrames = noisyReceivedFrames(noisyCapture.frames, noisySessionId); + const quietSessionNoisyFrames = noisyReceivedFrames(quietCapture.frames, noisySessionId); expect( noisyFrames.length, "noisy session must produce observable message frames", diff --git a/apps/web/hooks/domains/session/message-backfill.ts b/apps/web/hooks/domains/session/message-backfill.ts new file mode 100644 index 0000000000..3e17ee96e6 --- /dev/null +++ b/apps/web/hooks/domains/session/message-backfill.ts @@ -0,0 +1,97 @@ +import type { useAppStoreApi } from "@/components/state-provider"; +import { listTaskSessionMessages } from "@/lib/api"; +import { createDebugLogger } from "@/lib/debug/log"; +import type { Message } from "@/lib/types/http"; + +const BACKFILL_PAGE_LIMIT = 100; +const debug = createDebugLogger("messages:fetch"); + +export const MAX_AUTO_BACKFILL_PAGES = 10; +export type BackfillStep = "continue" | "stop"; + +type IsActive = () => boolean; +type SessionMessageStore = ReturnType; + +function isInactive(isActive?: IsActive): boolean { + return isActive !== undefined && !isActive(); +} + +export function hasUserOrAgentMessage(messages: Message[]): boolean { + return messages.some( + (message) => + message.type === "message" && + (message.author_type === "user" || message.author_type === "agent"), + ); +} + +async function fetchAndPrependOlder( + sessionId: string, + store: SessionMessageStore, + oldestCursor: string, + isActive?: IsActive, +): Promise { + const response = await listTaskSessionMessages(sessionId, { + limit: BACKFILL_PAGE_LIMIT, + before: oldestCursor, + sort: "desc", + }); + if (isInactive(isActive)) return 0; + const ordered = [...(response.messages ?? [])].reverse(); + const newOldestCursor = ordered[0]?.id ?? oldestCursor; + store.getState().prependMessages(sessionId, ordered, { + hasMore: response.has_more ?? false, + oldestCursor: newOldestCursor, + }); + return ordered.length; +} + +export async function runBackfillRound( + sessionId: string, + store: SessionMessageStore, + round: number, + isActive?: IsActive, +): Promise { + if (isInactive(isActive)) return "stop"; + const meta = store.getState().messages.metaBySession[sessionId]; + const messages = store.getState().messages.bySession[sessionId] ?? []; + if (hasUserOrAgentMessage(messages)) return "stop"; + if (!meta?.hasMore || !meta.oldestCursor) { + debug("autoBackfill: stopping (no more older messages)", { + sessionId, + round, + hasMore: meta?.hasMore ?? false, + }); + return "stop"; + } + debug("autoBackfill: window has no user/agent message, fetching older", { + sessionId, + round, + currentCount: messages.length, + oldestCursor: meta.oldestCursor, + }); + try { + const added = await fetchAndPrependOlder(sessionId, store, meta.oldestCursor, isActive); + if (isInactive(isActive)) return "stop"; + return added === 0 ? "stop" : "continue"; + } catch (err) { + debug("autoBackfill: fetch failed, stopping", { sessionId, round, err }); + return "stop"; + } +} + +export async function autoBackfillUntilUserMessage( + sessionId: string, + store: SessionMessageStore, + isActive?: IsActive, +): Promise { + for (let round = 0; round < MAX_AUTO_BACKFILL_PAGES; round++) { + if (isInactive(isActive)) return; + const step = await runBackfillRound(sessionId, store, round, isActive); + if (step === "stop") return; + } + debug("autoBackfill: hit page budget without finding user/agent message", { + sessionId, + pageBudget: MAX_AUTO_BACKFILL_PAGES, + messageBudget: MAX_AUTO_BACKFILL_PAGES * BACKFILL_PAGE_LIMIT, + }); +} diff --git a/apps/web/hooks/domains/session/use-session-message-fetch.test.ts b/apps/web/hooks/domains/session/use-session-message-fetch.test.ts new file mode 100644 index 0000000000..99d5679b1e --- /dev/null +++ b/apps/web/hooks/domains/session/use-session-message-fetch.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Message } from "@/lib/types/http"; +import { doFetchMessages } from "./use-session-message-fetch"; + +const SESSION_ID = "session-1"; + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +function makeParams( + fetchAndStoreMessages: ( + sessionId: string, + store: never, + isActive?: () => boolean, + ) => Promise, + setMessagesLoading: ReturnType, +) { + return { + taskSessionId: SESSION_ID, + store: { + getState: () => ({ setMessagesLoading, setMessages: vi.fn() }), + } as never, + setIsLoading: vi.fn(), + setIsWaitingForInitialMessages: vi.fn(), + initialFetchStartRef: { current: null }, + lastFetchedSessionIdRef: { current: null }, + fetchAndStoreMessages, + autoBackfillUntilUserMessage: vi.fn().mockResolvedValue(undefined), + hasUserOrAgentMessage: vi.fn().mockReturnValue(true), + }; +} + +describe("doFetchMessages", () => { + it("keeps the shared loading flag set until overlapping fetches all settle", async () => { + const first = deferred(); + const second = deferred(); + const fetchAndStoreMessages = vi + .fn() + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(second.promise); + const setMessagesLoading = vi.fn(); + + const firstFetch = doFetchMessages( + makeParams(fetchAndStoreMessages, setMessagesLoading) as never, + ); + const secondFetch = doFetchMessages( + makeParams(fetchAndStoreMessages, setMessagesLoading) as never, + ); + + expect(setMessagesLoading).toHaveBeenNthCalledWith(1, SESSION_ID, true); + expect(setMessagesLoading).toHaveBeenNthCalledWith(2, SESSION_ID, true); + + first.resolve([]); + await firstFetch; + expect(setMessagesLoading).toHaveBeenCalledTimes(2); + + second.resolve([]); + await secondFetch; + expect(setMessagesLoading).toHaveBeenLastCalledWith(SESSION_ID, false); + }); +}); diff --git a/apps/web/hooks/domains/session/use-session-message-fetch.ts b/apps/web/hooks/domains/session/use-session-message-fetch.ts new file mode 100644 index 0000000000..70b759f6ec --- /dev/null +++ b/apps/web/hooks/domains/session/use-session-message-fetch.ts @@ -0,0 +1,96 @@ +import type { MutableRefObject } from "react"; + +import type { useAppStoreApi } from "@/components/state-provider"; +import type { Message } from "@/lib/types/http"; + +type SessionMessageStore = ReturnType; + +// Multiple lifecycle paths can hydrate the same session concurrently (for +// example, the initial mount and a visibility refresh). Keep the shared +// loading flag asserted until the last operation settles; an older request +// must not make a newer request look idle. +const inFlightFetchesBySession = new Map(); + +function beginSessionFetch(sessionId: string): void { + inFlightFetchesBySession.set(sessionId, (inFlightFetchesBySession.get(sessionId) ?? 0) + 1); +} + +function endSessionFetch(sessionId: string): boolean { + const remaining = (inFlightFetchesBySession.get(sessionId) ?? 1) - 1; + if (remaining > 0) { + inFlightFetchesBySession.set(sessionId, remaining); + return false; + } + inFlightFetchesBySession.delete(sessionId); + return true; +} + +function isInactive(isActive?: () => boolean): boolean { + return isActive !== undefined && !isActive(); +} + +type DoFetchMessagesParams = { + taskSessionId: string; + store: SessionMessageStore; + setIsLoading: (value: boolean) => void; + setIsWaitingForInitialMessages: (value: boolean) => void; + initialFetchStartRef: MutableRefObject; + lastFetchedSessionIdRef: MutableRefObject; + fetchAndStoreMessages: ( + sessionId: string, + store: SessionMessageStore, + isActive?: () => boolean, + ) => Promise; + autoBackfillUntilUserMessage: ( + sessionId: string, + store: SessionMessageStore, + isActive?: () => boolean, + ) => Promise; + hasUserOrAgentMessage: (messages: Message[]) => boolean; + onError?: (error: unknown) => void; + isActive?: () => boolean; +}; + +export async function doFetchMessages({ + taskSessionId, + store, + setIsLoading, + setIsWaitingForInitialMessages, + initialFetchStartRef, + lastFetchedSessionIdRef, + fetchAndStoreMessages, + autoBackfillUntilUserMessage, + hasUserOrAgentMessage, + onError, + isActive, +}: DoFetchMessagesParams): Promise { + if (isInactive(isActive)) return; + beginSessionFetch(taskSessionId); + setIsLoading(true); + store.getState().setMessagesLoading(taskSessionId, true); + if (initialFetchStartRef.current === null) { + initialFetchStartRef.current = Date.now(); + setIsWaitingForInitialMessages(true); + } + try { + const fetched = await fetchAndStoreMessages(taskSessionId, store, isActive); + if (isInactive(isActive)) return; + lastFetchedSessionIdRef.current = taskSessionId; + if (fetched.length > 0) setIsWaitingForInitialMessages(false); + if (fetched.length > 0 && !hasUserOrAgentMessage(fetched)) { + await autoBackfillUntilUserMessage(taskSessionId, store, isActive); + } + } catch (error) { + if (isInactive(isActive)) return; + if (onError) onError(error); + else console.error("Failed to fetch messages:", error); + store.getState().setMessages(taskSessionId, []); + lastFetchedSessionIdRef.current = taskSessionId; + } finally { + if (endSessionFetch(taskSessionId)) { + store.getState().setMessagesLoading(taskSessionId, false); + } + if (isInactive(isActive)) return; + setIsLoading(false); + } +} diff --git a/apps/web/hooks/domains/session/use-session-messages.test.ts b/apps/web/hooks/domains/session/use-session-messages.test.ts index c3b20c6a91..79019d91e3 100644 --- a/apps/web/hooks/domains/session/use-session-messages.test.ts +++ b/apps/web/hooks/domains/session/use-session-messages.test.ts @@ -1,19 +1,42 @@ +import { act, renderHook } from "@testing-library/react"; import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; import type { Message } from "@/lib/types/http"; const mockListTaskSessionMessages = vi.fn(); +const mockWebSocketClient = { + getSessionSubscriptionReadiness: vi.fn(), + request: vi.fn(), + subscribeSession: vi.fn(), + subscribeSessionWithReady: vi.fn(), +}; + +const mockState = { + messages: { + bySession: { "sess-1": [] as Message[] }, + metaBySession: { + "sess-1": { hasMore: false, oldestCursor: null, isLoading: false }, + }, + }, + taskSessions: { items: { "sess-1": { state: "RUNNING" } } }, + turns: { activeBySession: { "sess-1": null } }, + connection: { status: "connected" }, + mergeMessages: vi.fn(), + setMessagesLoading: vi.fn(), + setMessages: vi.fn(), + prependMessages: vi.fn(), +}; vi.mock("@/lib/api", () => ({ listTaskSessionMessages: (...args: unknown[]) => mockListTaskSessionMessages(...args), })); vi.mock("@/lib/ws/connection", () => ({ - getWebSocketClient: () => null, + getWebSocketClient: () => mockWebSocketClient, })); vi.mock("@/components/state-provider", () => ({ - useAppStore: () => null, - useAppStoreApi: () => null, + useAppStore: (selector: (state: typeof mockState) => unknown) => selector(mockState), + useAppStoreApi: () => ({ getState: () => mockState }), })); import { taskId, sessionId } from "@/lib/types/ids"; @@ -21,6 +44,17 @@ import { taskId, sessionId } from "@/lib/types/ids"; beforeEach(() => { vi.clearAllMocks(); mockListTaskSessionMessages.mockResolvedValue({ messages: [], has_more: false }); + mockWebSocketClient.request.mockResolvedValue({ messages: [], has_more: false }); + mockWebSocketClient.subscribeSession.mockReturnValue(vi.fn()); + mockState.messages.bySession["sess-1"] = []; + mockState.messages.metaBySession["sess-1"] = { + hasMore: false, + oldestCursor: null, + isLoading: false, + }; + mockState.connection.status = "connected"; + mockState.taskSessions.items["sess-1"] = { state: "RUNNING" }; + mockState.turns.activeBySession["sess-1"] = null; }); afterEach(() => { @@ -37,6 +71,7 @@ import { nextFetchSeq, commitFetchSeq, MAX_AUTO_BACKFILL_PAGES, + useSessionMessages, } from "./use-session-messages"; import type { TaskSessionState } from "@/lib/types/http"; @@ -312,6 +347,21 @@ describe("runBackfillRound", () => { const result = await runBackfillRound("sess-1", store as never, 0); expect(result).toBe("stop"); }); + + it("does not prepend messages when cleanup occurs during the fetch", async () => { + const request = deferred<{ messages: Message[]; has_more: boolean }>(); + mockListTaskSessionMessages.mockReturnValueOnce(request.promise); + const store = makeStore({ messages: [], hasMore: true, oldestCursor: "msg-1" }); + let active = true; + const round = runBackfillRound("sess-1", store as never, 0, () => active); + + expect(mockListTaskSessionMessages).toHaveBeenCalledTimes(1); + active = false; + request.resolve({ messages: [makeMessage({ id: "old-1" })], has_more: true }); + + await expect(round).resolves.toBe("stop"); + expect(store._prependMessages).not.toHaveBeenCalled(); + }); }); describe("stale concurrent fetch guard", () => { @@ -391,3 +441,59 @@ describe("autoBackfillUntilUserMessage", () => { expect(mockListTaskSessionMessages).toHaveBeenCalledTimes(1); }); }); + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +describe("session subscription hydration ordering", () => { + it("does not request messages before subscription acknowledgement", async () => { + const readiness = deferred(); + mockWebSocketClient.getSessionSubscriptionReadiness.mockReturnValue(readiness.promise); + mockWebSocketClient.subscribeSessionWithReady.mockReturnValue({ + ready: readiness.promise, + unsubscribe: vi.fn(), + }); + + const { unmount } = renderHook(() => useSessionMessages("sess-1")); + + expect(mockWebSocketClient.request).not.toHaveBeenCalled(); + + await act(async () => { + readiness.resolve(); + await readiness.promise; + }); + + expect(mockWebSocketClient.request).toHaveBeenCalledWith( + "message.list", + expect.objectContaining({ session_id: "sess-1" }), + 10000, + ); + expect(mockWebSocketClient.request).toHaveBeenCalledTimes(1); + unmount(); + }); + + it("does not start hydration when acknowledgement resolves after cleanup", async () => { + const readiness = deferred(); + mockWebSocketClient.getSessionSubscriptionReadiness.mockReturnValue(readiness.promise); + mockWebSocketClient.subscribeSessionWithReady.mockReturnValue({ + ready: readiness.promise, + unsubscribe: vi.fn(), + }); + + const { unmount } = renderHook(() => useSessionMessages("sess-1")); + unmount(); + + await act(async () => { + readiness.resolve(); + await readiness.promise; + }); + + expect(mockWebSocketClient.request).not.toHaveBeenCalled(); + expect(mockState.setMessagesLoading).toHaveBeenLastCalledWith("sess-1", false); + }); +}); diff --git a/apps/web/hooks/domains/session/use-session-messages.ts b/apps/web/hooks/domains/session/use-session-messages.ts index bd3e45200f..c4da30ac8c 100644 --- a/apps/web/hooks/domains/session/use-session-messages.ts +++ b/apps/web/hooks/domains/session/use-session-messages.ts @@ -3,20 +3,27 @@ import { getWebSocketClient } from "@/lib/ws/connection"; import { useForegroundRefresh } from "@/hooks/use-foreground-refresh"; import { useAppStore, useAppStoreApi } from "@/components/state-provider"; import type { TaskSessionState, Message } from "@/lib/types/http"; -import { listTaskSessionMessages } from "@/lib/api"; import { createDebugLogger, isDebug } from "@/lib/debug/log"; import { useUnknownSessionSubscriptionRetry, useUnknownSessionSubscriptionRetryEffect, } from "./use-session-subscription-retry"; +import { doFetchMessages } from "./use-session-message-fetch"; +import { autoBackfillUntilUserMessage, hasUserOrAgentMessage } from "./message-backfill"; + +export { + autoBackfillUntilUserMessage, + type BackfillStep, + hasUserOrAgentMessage, + MAX_AUTO_BACKFILL_PAGES, + runBackfillRound, +} from "./message-backfill"; export { shouldRetryUnknownSessionSubscription } from "./use-session-subscription-retry"; const INITIAL_FETCH_LIMIT = 100; -const BACKFILL_PAGE_LIMIT = 100; const RUNNING_BACKFILL_INITIAL_DELAY_MS = 1200; const RUNNING_BACKFILL_INTERVAL_MS = 5000; -export const MAX_AUTO_BACKFILL_PAGES = 10; // Monotonic guard against stale concurrent fetches. Each fetch claims an // increasing sequence number before awaiting; a completion only merges if no @@ -44,12 +51,6 @@ export function commitFetchSeq(sessionId: string, seq: number): boolean { return true; } -export function hasUserOrAgentMessage(messages: Message[]): boolean { - return messages.some( - (m) => m.type === "message" && (m.author_type === "user" || m.author_type === "agent"), - ); -} - // States where a turn (or the agent boot) is actively progressing. const ACTIVE_SESSION_STATES: ReadonlySet = new Set(["STARTING", "RUNNING"]); // States the session settles into once a turn finishes. @@ -139,9 +140,14 @@ interface UseSessionMessagesReturn { } type MessageListResponse = { messages: Message[]; has_more?: boolean; cursor?: string }; +type InFlightMessageRequest = { + readiness: Promise; + promise: Promise; +}; const EMPTY_MESSAGES: Message[] = []; const EMPTY_META = { isLoading: false, hasMore: false, oldestCursor: null }; +const inFlightMessageRequests = new Map(); /** Debug-only summary of a fetch response (no-op unless debug logging is on). */ function logFetchSummary( @@ -169,25 +175,63 @@ function logFetchSummary( } } -/** Fetch latest messages via WS and merge with any that arrived via live notifications. */ -async function fetchAndStoreMessages( +function requestSessionMessages( + client: NonNullable>, sessionId: string, - store: ReturnType, -): Promise { - const client = getWebSocketClient(); - if (!client) { - return []; - } - const seq = nextFetchSeq(); + readiness: Promise, +): Promise { + const existing = inFlightMessageRequests.get(sessionId); + if (existing?.readiness === readiness) return existing.promise; const requestParams = { session_id: sessionId, limit: INITIAL_FETCH_LIMIT, sort: "desc" as const, }; - const response = await client.request("message.list", requestParams, 10000); + const promise = client.request("message.list", requestParams, 10000); + const entry = { readiness, promise }; + inFlightMessageRequests.set(sessionId, entry); + void promise.then( + () => { + window.setTimeout(() => { + if (inFlightMessageRequests.get(sessionId) === entry) { + inFlightMessageRequests.delete(sessionId); + } + }, 0); + }, + () => { + window.setTimeout(() => { + if (inFlightMessageRequests.get(sessionId) === entry) { + inFlightMessageRequests.delete(sessionId); + } + }, 0); + }, + ); + return promise; +} + +/** Fetch latest messages via WS and merge with any that arrived via live notifications. */ +async function fetchAndStoreMessages( + sessionId: string, + store: ReturnType, + isActive?: () => boolean, +): Promise { + const client = getWebSocketClient(); + if (!client) { + return []; + } + // Initial and recovery fetches must not overtake the server-side + // session.subscribe registration. The client returns an already-resolved + // promise when no durable subscription is active, preserving fetch behavior + // for non-session callers while gating this hook's normal subscription path. + const readiness = client.getSessionSubscriptionReadiness(sessionId); + await readiness; + if (isActive && !isActive()) return []; + const seq = nextFetchSeq(); + const response = await requestSessionMessages(client, sessionId, readiness); + if (isActive && !isActive()) return []; const fetched = [...(response.messages ?? [])].reverse(); - logFetchSummary(sessionId, fetched, response, requestParams.limit); + logFetchSummary(sessionId, fetched, response, INITIAL_FETCH_LIMIT); // Merge: keep WS-delivered messages that aren't in the fetch response. // This prevents a slow fetch (sent before messages existed) from wiping // messages that arrived via real-time notifications while the fetch was @@ -231,116 +275,6 @@ async function fetchAndStoreMessages( * endpoint `useLazyLoadMessages` uses until we span at least one user/agent * message or hit the page budget. */ -export type BackfillStep = "continue" | "stop"; - -async function fetchAndPrependOlder( - sessionId: string, - store: ReturnType, - oldestCursor: string, -): Promise { - const response = await listTaskSessionMessages(sessionId, { - limit: BACKFILL_PAGE_LIMIT, - before: oldestCursor, - sort: "desc", - }); - const ordered = [...(response.messages ?? [])].reverse(); - const newOldestCursor = ordered[0]?.id ?? oldestCursor; - store.getState().prependMessages(sessionId, ordered, { - hasMore: response.has_more ?? false, - oldestCursor: newOldestCursor, - }); - return ordered.length; -} - -export async function runBackfillRound( - sessionId: string, - store: ReturnType, - round: number, -): Promise { - const meta = store.getState().messages.metaBySession[sessionId]; - const messages = store.getState().messages.bySession[sessionId] ?? []; - if (hasUserOrAgentMessage(messages)) return "stop"; - if (!meta?.hasMore || !meta.oldestCursor) { - debug("autoBackfill: stopping (no more older messages)", { - sessionId, - round, - hasMore: meta?.hasMore ?? false, - }); - return "stop"; - } - debug("autoBackfill: window has no user/agent message, fetching older", { - sessionId, - round, - currentCount: messages.length, - oldestCursor: meta.oldestCursor, - }); - try { - const added = await fetchAndPrependOlder(sessionId, store, meta.oldestCursor); - return added === 0 ? "stop" : "continue"; - } catch (err) { - debug("autoBackfill: fetch failed, stopping", { sessionId, round, err }); - return "stop"; - } -} - -export async function autoBackfillUntilUserMessage( - sessionId: string, - store: ReturnType, -): Promise { - for (let round = 0; round < MAX_AUTO_BACKFILL_PAGES; round++) { - const step = await runBackfillRound(sessionId, store, round); - if (step === "stop") return; - } - debug("autoBackfill: hit page budget without finding user/agent message", { - sessionId, - pageBudget: MAX_AUTO_BACKFILL_PAGES, - messageBudget: MAX_AUTO_BACKFILL_PAGES * BACKFILL_PAGE_LIMIT, - }); -} - -type FetchMessagesParams = { - taskSessionId: string; - store: ReturnType; - setIsLoading: (v: boolean) => void; - setIsWaitingForInitialMessages: (v: boolean) => void; - initialFetchStartRef: MutableRefObject; - lastFetchedSessionIdRef: MutableRefObject; - onError?: (error: unknown) => void; -}; - -async function doFetchMessages({ - taskSessionId, - store, - setIsLoading, - setIsWaitingForInitialMessages, - initialFetchStartRef, - lastFetchedSessionIdRef, - onError, -}: FetchMessagesParams): Promise { - setIsLoading(true); - store.getState().setMessagesLoading(taskSessionId, true); - if (initialFetchStartRef.current === null) { - initialFetchStartRef.current = Date.now(); - setIsWaitingForInitialMessages(true); - } - try { - const fetched = await fetchAndStoreMessages(taskSessionId, store); - lastFetchedSessionIdRef.current = taskSessionId; - if (fetched.length > 0) setIsWaitingForInitialMessages(false); - if (fetched.length > 0 && !hasUserOrAgentMessage(fetched)) { - await autoBackfillUntilUserMessage(taskSessionId, store); - } - } catch (error) { - if (onError) onError(error); - else console.error("Failed to fetch messages:", error); - store.getState().setMessages(taskSessionId, []); - lastFetchedSessionIdRef.current = taskSessionId; - } finally { - store.getState().setMessagesLoading(taskSessionId, false); - setIsLoading(false); - } -} - function useTerminalStateFetch( taskSessionId: string | null, taskSessionState: TaskSessionState | null, @@ -370,6 +304,9 @@ function useTerminalStateFetch( void doFetchMessages({ taskSessionId, ...refs, + fetchAndStoreMessages, + autoBackfillUntilUserMessage, + hasUserOrAgentMessage, onError: (error) => console.error("Failed to fetch messages after state change:", error), }); }, [taskSessionId, taskSessionState, hasAgentMessage, connectionStatus, refs]); @@ -444,15 +381,23 @@ function useSessionSubscription( return; } debug("subscription: subscribing", { sessionId: taskSessionId }); - const unsubscribe = client.subscribeSession(taskSessionId); + const subscription = client.subscribeSessionWithReady(taskSessionId); + let active = true; - // Re-fetch messages after subscribing to close the gap between SSR - // (which may have run before the agent responded) and this subscription. - fetchAndStoreMessages(taskSessionId, store).catch(() => {}); + // Re-fetch messages after the server acknowledges the subscription to + // close the gap between SSR (which may have run before the agent + // responded) and this subscription. + void subscription.ready + .then(() => { + if (!active) return; + return fetchAndStoreMessages(taskSessionId, store, () => active); + }) + .catch(() => {}); return () => { + active = false; debug("subscription: unsubscribing", { sessionId: taskSessionId }); - unsubscribe(); + subscription.unsubscribe(); }; }, [taskSessionId, connectionStatus, store, isSessionStartingOrUnknown]); } @@ -618,21 +563,22 @@ export function useSessionMessages(taskSessionId: string | null): UseSessionMess refs: fetchRefs, } = useMessageFetchState(store); + useSessionLifecycleSubscriptions({ + taskSessionId, + taskSessionState, + connectionStatus, + activeTurnId, + messages, + store, + }); + useEffect(() => { if (!taskSessionId) { initialFetchStartRef.current = null; lastFetchedSessionIdRef.current = null; setIsWaitingForInitialMessages(false); + return; } - }, [ - taskSessionId, - initialFetchStartRef, - lastFetchedSessionIdRef, - setIsWaitingForInitialMessages, - ]); - - useEffect(() => { - if (!taskSessionId) return; if (messages.length > 0) { setIsWaitingForInitialMessages(false); return; @@ -644,7 +590,13 @@ export function useSessionMessages(taskSessionId: string | null): UseSessionMess }, [taskSessionId, messages.length, initialFetchStartRef, setIsWaitingForInitialMessages]); useEffect(() => { - if (!taskSessionId || connectionStatus !== "connected") return; + let active = true; + const deactivate = () => { + active = false; + }; + if (!taskSessionId || connectionStatus !== "connected") { + return deactivate; + } const isFreshMount = prevSessionIdRef.current === null; const sessionChanged = @@ -659,23 +611,30 @@ export function useSessionMessages(taskSessionId: string | null): UseSessionMess if (messages.length > 0 && !sessionChanged && !isFreshMount) { lastFetchedSessionIdRef.current = taskSessionId; setIsWaitingForInitialMessages(false); - return; + return deactivate; } // Fresh mount with cached messages — show cached instantly, fetch in background if (isFreshMount && messages.length > 0) { lastFetchedSessionIdRef.current = taskSessionId; setIsWaitingForInitialMessages(false); - fetchAndStoreMessages(taskSessionId, store).catch(() => {}); - return; + fetchAndStoreMessages(taskSessionId, store, () => active).catch(() => {}); + return deactivate; } - if (lastFetchedSessionIdRef.current === taskSessionId) return; + if (lastFetchedSessionIdRef.current === taskSessionId) { + return deactivate; + } void doFetchMessages({ taskSessionId, ...fetchRefs, + fetchAndStoreMessages, + autoBackfillUntilUserMessage, + hasUserOrAgentMessage, + isActive: () => active, }); + return deactivate; }, [ taskSessionId, connectionStatus, @@ -686,15 +645,6 @@ export function useSessionMessages(taskSessionId: string | null): UseSessionMess fetchRefs, ]); - useSessionLifecycleSubscriptions({ - taskSessionId, - taskSessionState, - connectionStatus, - activeTurnId, - messages, - store, - }); - useVisibilityBackfill(taskSessionId, store); useTerminalStateFetch(taskSessionId, taskSessionState, hasAgentMessage, fetchRefs); diff --git a/apps/web/hooks/domains/session/use-session-subscription-retry.test.ts b/apps/web/hooks/domains/session/use-session-subscription-retry.test.ts index b11d9e8528..5d4e6cbc8c 100644 --- a/apps/web/hooks/domains/session/use-session-subscription-retry.test.ts +++ b/apps/web/hooks/domains/session/use-session-subscription-retry.test.ts @@ -156,8 +156,8 @@ describe("useUnknownSessionSubscriptionRetryEffect", () => { }); it("skips when the retry token is zero", () => { - const send = vi.fn(); - vi.mocked(getWebSocketClient).mockReturnValue({ send } as never); + const resubscribeSession = vi.fn(); + vi.mocked(getWebSocketClient).mockReturnValue({ resubscribeSession } as never); renderHook(() => useUnknownSessionSubscriptionRetryEffect({ @@ -167,12 +167,12 @@ describe("useUnknownSessionSubscriptionRetryEffect", () => { }), ); - expect(send).not.toHaveBeenCalled(); + expect(resubscribeSession).not.toHaveBeenCalled(); }); it("skips when the session id is missing", () => { - const send = vi.fn(); - vi.mocked(getWebSocketClient).mockReturnValue({ send } as never); + const resubscribeSession = vi.fn(); + vi.mocked(getWebSocketClient).mockReturnValue({ resubscribeSession } as never); renderHook(() => useUnknownSessionSubscriptionRetryEffect({ @@ -182,12 +182,12 @@ describe("useUnknownSessionSubscriptionRetryEffect", () => { }), ); - expect(send).not.toHaveBeenCalled(); + expect(resubscribeSession).not.toHaveBeenCalled(); }); it("skips when the websocket is disconnected", () => { - const send = vi.fn(); - vi.mocked(getWebSocketClient).mockReturnValue({ send } as never); + const resubscribeSession = vi.fn(); + vi.mocked(getWebSocketClient).mockReturnValue({ resubscribeSession } as never); renderHook(() => useUnknownSessionSubscriptionRetryEffect({ @@ -197,12 +197,12 @@ describe("useUnknownSessionSubscriptionRetryEffect", () => { }), ); - expect(send).not.toHaveBeenCalled(); + expect(resubscribeSession).not.toHaveBeenCalled(); }); - it("sends one subscribe frame per retry-token change", () => { - const send = vi.fn(); - vi.mocked(getWebSocketClient).mockReturnValue({ send } as never); + it("requests one tracked subscription per retry-token change", () => { + const resubscribeSession = vi.fn().mockResolvedValue(undefined); + vi.mocked(getWebSocketClient).mockReturnValue({ resubscribeSession } as never); const { rerender } = renderHook( ({ retryToken }: { retryToken: number }) => @@ -214,17 +214,12 @@ describe("useUnknownSessionSubscriptionRetryEffect", () => { { initialProps: { retryToken: 1 } }, ); - expect(send).toHaveBeenCalledTimes(1); - expect(send).toHaveBeenLastCalledWith( - expect.objectContaining({ - action: "session.subscribe", - payload: { session_id: "sess-1" }, - }), - ); + expect(resubscribeSession).toHaveBeenCalledTimes(1); + expect(resubscribeSession).toHaveBeenLastCalledWith("sess-1"); rerender({ retryToken: 2 }); - expect(send).toHaveBeenCalledTimes(2); + expect(resubscribeSession).toHaveBeenCalledTimes(2); }); it("skips when the websocket client is unavailable", () => { @@ -242,9 +237,12 @@ describe("useUnknownSessionSubscriptionRetryEffect", () => { }); it("does not call unsubscribe paths", () => { - const send = vi.fn(); + const resubscribeSession = vi.fn().mockResolvedValue(undefined); const unsubscribeSession = vi.fn(); - vi.mocked(getWebSocketClient).mockReturnValue({ send, unsubscribeSession } as never); + vi.mocked(getWebSocketClient).mockReturnValue({ + resubscribeSession, + unsubscribeSession, + } as never); const { unmount } = renderHook(() => useUnknownSessionSubscriptionRetryEffect({ @@ -256,7 +254,7 @@ describe("useUnknownSessionSubscriptionRetryEffect", () => { unmount(); - expect(send).toHaveBeenCalledTimes(1); + expect(resubscribeSession).toHaveBeenCalledTimes(1); expect(unsubscribeSession).not.toHaveBeenCalled(); }); }); diff --git a/apps/web/hooks/domains/session/use-session-subscription-retry.ts b/apps/web/hooks/domains/session/use-session-subscription-retry.ts index acc8877b13..c1173d2180 100644 --- a/apps/web/hooks/domains/session/use-session-subscription-retry.ts +++ b/apps/web/hooks/domains/session/use-session-subscription-retry.ts @@ -2,7 +2,6 @@ import { useEffect, useState } from "react"; import { createDebugLogger } from "@/lib/debug/log"; import type { TaskSessionState } from "@/lib/types/http"; -import { generateUUID } from "@/lib/utils"; import { getWebSocketClient } from "@/lib/ws/connection"; const UNKNOWN_SESSION_INITIAL_RESUBSCRIBE_MS = 1000; @@ -86,18 +85,19 @@ export function useUnknownSessionSubscriptionRetryEffect(params: { return; } // The durable subscription effect owns the client's ref-counted - // subscribe/unsubscribe lifecycle. This retry only re-sends the subscribe - // frame to cover the backend race where the first subscribe arrived before - // the session was fully constructed. - debug("unknown-session retry: sending session.subscribe", { + // subscribe/unsubscribe lifecycle. This retry asks the client to send a + // tracked subscribe request, so any later hydration can share its + // acknowledgement instead of racing an untracked frame. + debug("unknown-session retry: requesting session.subscribe", { sessionId: taskSessionId, retryToken, }); - client.send({ - id: generateUUID(), - type: "request", - action: "session.subscribe", - payload: { session_id: taskSessionId }, + void client.resubscribeSession(taskSessionId).catch((error: unknown) => { + debug("unknown-session retry: session.subscribe failed", { + sessionId: taskSessionId, + retryToken, + error, + }); }); }, [taskSessionId, connectionStatus, retryToken]); } diff --git a/apps/web/hooks/domains/session/use-visibility-backfill.test.ts b/apps/web/hooks/domains/session/use-visibility-backfill.test.ts index 9ac52c958d..21f49bc87a 100644 --- a/apps/web/hooks/domains/session/use-visibility-backfill.test.ts +++ b/apps/web/hooks/domains/session/use-visibility-backfill.test.ts @@ -1,13 +1,17 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; -import { cleanup, renderHook } from "@testing-library/react"; +import { act, cleanup, renderHook } from "@testing-library/react"; const mockRequest = vi.fn(); +const mockMergeMessages = vi.fn(); const mockSetMessages = vi.fn(); const SESSION_ID = "sess-1"; const MESSAGE_LIST = "message.list"; vi.mock("@/lib/ws/connection", () => ({ - getWebSocketClient: () => ({ request: mockRequest }), + getWebSocketClient: () => ({ + getSessionSubscriptionReadiness: () => Promise.resolve(), + request: mockRequest, + }), })); vi.mock("@/components/state-provider", () => ({ @@ -15,6 +19,7 @@ vi.mock("@/components/state-provider", () => ({ useAppStoreApi: () => ({ getState: () => ({ messages: { bySession: {} }, + mergeMessages: mockMergeMessages, setMessages: mockSetMessages, }), }), @@ -34,7 +39,11 @@ describe("useVisibilityBackfill", () => { vi.clearAllMocks(); mockRequest.mockResolvedValue({ messages: [], has_more: false }); store = { - getState: () => ({ messages: { bySession: {} }, setMessages: mockSetMessages }), + getState: () => ({ + messages: { bySession: {} }, + mergeMessages: mockMergeMessages, + setMessages: mockSetMessages, + }), }; }); @@ -42,9 +51,12 @@ describe("useVisibilityBackfill", () => { cleanup(); }); - it("fetches when the tab becomes visible", () => { + it("fetches when the tab becomes visible", async () => { renderHook(() => useVisibilityBackfill(SESSION_ID, store as never)); - setVisibility("visible"); + await act(async () => { + setVisibility("visible"); + await Promise.resolve(); + }); expect(mockRequest).toHaveBeenCalledTimes(1); expect(mockRequest).toHaveBeenCalledWith( MESSAGE_LIST, @@ -53,10 +65,13 @@ describe("useVisibilityBackfill", () => { ); }); - it("fetches when the Kandev window regains focus", () => { + it("fetches when the Kandev window regains focus", async () => { renderHook(() => useVisibilityBackfill(SESSION_ID, store as never)); - window.dispatchEvent(new Event("focus")); + await act(async () => { + window.dispatchEvent(new Event("focus")); + await Promise.resolve(); + }); expect(mockRequest).toHaveBeenCalledWith( MESSAGE_LIST, @@ -84,12 +99,15 @@ describe("useVisibilityBackfill", () => { expect(mockRequest).not.toHaveBeenCalled(); }); - it("re-registers when sessionId changes", () => { + it("re-registers when sessionId changes", async () => { const { rerender } = renderHook( ({ id }: { id: string | null }) => useVisibilityBackfill(id, store as never), { initialProps: { id: SESSION_ID } }, ); - setVisibility("visible"); + await act(async () => { + setVisibility("visible"); + await Promise.resolve(); + }); expect(mockRequest).toHaveBeenLastCalledWith( MESSAGE_LIST, expect.objectContaining({ session_id: SESSION_ID }), @@ -97,7 +115,10 @@ describe("useVisibilityBackfill", () => { ); rerender({ id: "sess-2" }); - setVisibility("visible"); + await act(async () => { + setVisibility("visible"); + await Promise.resolve(); + }); expect(mockRequest).toHaveBeenLastCalledWith( MESSAGE_LIST, expect.objectContaining({ session_id: "sess-2" }), diff --git a/apps/web/lib/ws/client.test.ts b/apps/web/lib/ws/client.test.ts new file mode 100644 index 0000000000..8b4735c64a --- /dev/null +++ b/apps/web/lib/ws/client.test.ts @@ -0,0 +1,223 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { WebSocketClient } from "./client"; + +type SentRequest = { + id: string; + type: string; + action: string; + payload: unknown; +}; + +class FakeWebSocket { + static readonly OPEN = 1; + static readonly CLOSED = 3; + static instances: FakeWebSocket[] = []; + + readonly sent: SentRequest[] = []; + readyState = 0; + onopen: (() => void) | null = null; + onmessage: ((event: { data: string }) => void) | null = null; + onerror: (() => void) | null = null; + onclose: ((event: CloseEvent) => void) | null = null; + + constructor(readonly url: string) { + FakeWebSocket.instances.push(this); + } + + open() { + this.readyState = FakeWebSocket.OPEN; + this.onopen?.(); + } + + send(data: string) { + this.sent.push(JSON.parse(data) as SentRequest); + } + + receive(message: unknown) { + this.onmessage?.({ data: JSON.stringify(message) }); + } + + close() { + this.readyState = FakeWebSocket.CLOSED; + this.onclose?.({ code: 1006, reason: "network lost" } as CloseEvent); + } + + static latest() { + const socket = FakeWebSocket.instances.at(-1); + if (!socket) throw new Error("No fake websocket exists"); + return socket; + } + + static reset() { + FakeWebSocket.instances = []; + } +} + +function connectClient(options?: ConstructorParameters[2]) { + const client = new WebSocketClient("ws://test", undefined, { + enabled: false, + ...options, + }); + client.connect(); + const socket = FakeWebSocket.latest(); + socket.open(); + return { client, socket }; +} + +function sessionSubscribeRequest(socket: FakeWebSocket, index = 0) { + const request = socket.sent.filter((message) => message.action === "session.subscribe")[index]; + if (!request) throw new Error("No session.subscribe request was sent"); + return request; +} + +function acknowledge(socket: FakeWebSocket, request: SentRequest) { + socket.receive({ + id: request.id, + type: "response", + payload: { success: true }, + }); +} + +beforeEach(() => { + vi.stubGlobal("WebSocket", FakeWebSocket); + FakeWebSocket.reset(); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + vi.useRealTimers(); +}); + +describe("session subscription readiness", () => { + it("resolves only after the server acknowledges the registration", async () => { + const { client, socket } = connectClient(); + const subscription = client.subscribeSessionWithReady("sess-1"); + const request = sessionSubscribeRequest(socket); + let ready = false; + + void subscription.ready.then(() => { + ready = true; + }); + await Promise.resolve(); + expect(ready).toBe(false); + + acknowledge(socket, request); + + await expect(subscription.ready).resolves.toBeUndefined(); + expect(ready).toBe(true); + subscription.unsubscribe(); + }); + + it("shares one in-flight acknowledgement between ref-counted consumers", async () => { + const { client, socket } = connectClient(); + const first = client.subscribeSessionWithReady("sess-1"); + const second = client.subscribeSessionWithReady("sess-1"); + + expect(second.ready).toBe(first.ready); + expect(socket.sent.filter((message) => message.action === "session.subscribe")).toHaveLength(1); + + acknowledge(socket, sessionSubscribeRequest(socket)); + await expect(second.ready).resolves.toBeUndefined(); + + first.unsubscribe(); + second.unsubscribe(); + }); + + it("allows a failed registration to be retried with fresh readiness", async () => { + const { client, socket } = connectClient(); + const subscription = client.subscribeSessionWithReady("sess-1"); + const firstRequest = sessionSubscribeRequest(socket); + + socket.receive({ + id: firstRequest.id, + type: "error", + payload: { message: "session is not ready" }, + }); + await expect(subscription.ready).rejects.toThrow("session is not ready"); + + const retry = client.resubscribeSession("sess-1"); + const retryRequest = sessionSubscribeRequest(socket, 1); + expect(retry).not.toBe(subscription.ready); + + acknowledge(socket, retryRequest); + await expect(retry).resolves.toBeUndefined(); + subscription.unsubscribe(); + }); + + it("tracks the re-registration after reconnect", async () => { + vi.useFakeTimers(); + const { client, socket } = connectClient({ enabled: true, initialDelay: 0, maxAttempts: 1 }); + const initial = client.subscribeSessionWithReady("sess-1"); + acknowledge(socket, sessionSubscribeRequest(socket)); + await expect(initial.ready).resolves.toBeUndefined(); + + socket.close(); + vi.advanceTimersByTime(0); + const reconnectedSocket = FakeWebSocket.latest(); + reconnectedSocket.open(); + + const reconnected = client.subscribeSessionWithReady("sess-1"); + const reconnectRequest = sessionSubscribeRequest(reconnectedSocket, 0); + expect(reconnectRequest.payload).toEqual({ session_id: "sess-1" }); + expect(reconnected.ready).not.toBe(initial.ready); + + acknowledge(reconnectedSocket, reconnectRequest); + await expect(reconnected.ready).resolves.toBeUndefined(); + initial.unsubscribe(); + reconnected.unsubscribe(); + }); +}); + +describe("session subscription reconnect recovery", () => { + it("keeps queued hydration behind the reconnect subscription acknowledgement", async () => { + vi.useFakeTimers(); + const { client, socket } = connectClient({ enabled: true, initialDelay: 0, maxAttempts: 1 }); + const initial = client.subscribeSessionWithReady("sess-1"); + acknowledge(socket, sessionSubscribeRequest(socket)); + await expect(initial.ready).resolves.toBeUndefined(); + + socket.close(); + const reconnectReadiness = client.getSessionSubscriptionReadiness("sess-1"); + let readinessResolved = false; + void reconnectReadiness.then(() => { + readinessResolved = true; + }); + client.send({ + id: "hydration-1", + type: "request", + action: "message.list", + payload: { session_id: "sess-1" }, + }); + await Promise.resolve(); + expect(readinessResolved).toBe(false); + + vi.runOnlyPendingTimers(); + const reconnectedSocket = FakeWebSocket.latest(); + reconnectedSocket.open(); + + expect(reconnectedSocket.sent.map((message) => message.action)).toEqual([ + "session.subscribe", + "message.list", + ]); + const reconnectRequest = sessionSubscribeRequest(reconnectedSocket); + const hydrationRequest = reconnectedSocket.sent.find( + (message) => message.action === "message.list", + ); + if (!hydrationRequest) throw new Error("No queued message.list request was sent"); + + acknowledge(reconnectedSocket, reconnectRequest); + await expect(reconnectReadiness).resolves.toBeUndefined(); + expect(hydrationRequest.id).toBe("hydration-1"); + initial.unsubscribe(); + }); + + it("rejects an active readiness when reconnect recovery is disabled", async () => { + const { client, socket } = connectClient({ enabled: false }); + const subscription = client.subscribeSessionWithReady("sess-1"); + socket.close(); + + await expect(subscription.ready).rejects.toThrow("WebSocket connection closed"); + subscription.unsubscribe(); + }); +}); diff --git a/apps/web/lib/ws/client.ts b/apps/web/lib/ws/client.ts index 0b2edebb8b..b5e78ae57b 100644 --- a/apps/web/lib/ws/client.ts +++ b/apps/web/lib/ws/client.ts @@ -19,9 +19,7 @@ const DISPATCH_LOG_DENYLIST = new Set([ type MessageHandler = (message: BackendMessageMap[T]) => void; -// Internal alias kept for readability — the WS client emits the same vocabulary -// the UI consumes. `disconnected` covers both "never connected" (formerly -// `idle`) and "socket closed" (formerly `closed`); `connected` replaces `open`. +// Internal alias for the status vocabulary shared with the UI. type WebSocketStatus = ConnectionStatus; export interface ReconnectOptions { @@ -32,6 +30,19 @@ export interface ReconnectOptions { backoffMultiplier?: number; } +export interface SessionSubscriptionHandle { + ready: Promise; + unsubscribe: () => void; +} + +type SessionSubscriptionReadiness = { + promise: Promise; + resolve: () => void; + reject: (reason: unknown) => void; + requestStarted: boolean; + settled: boolean; +}; + const DEFAULT_RECONNECT_OPTIONS: Required = { enabled: true, maxAttempts: 10, @@ -39,6 +50,7 @@ const DEFAULT_RECONNECT_OPTIONS: Required = { maxDelay: 30000, backoffMultiplier: 1.5, }; +const WEBSOCKET_CONNECTION_CLOSED_ERROR = "WebSocket connection closed"; export class WebSocketClient { private socket: WebSocket | null = null; @@ -59,6 +71,7 @@ export class WebSocketClient { private intentionalClose = false; private subscriptions = new Map(); private sessionSubscriptions = new Map(); + private sessionSubscriptionReadiness = new Map(); // Ref-counted focus signals: a session can be focused by both the task panel // and the task details page if both are mounted. Backend wakes its workspace // tracker into fast-poll mode while any client has focus, falling back to @@ -90,8 +103,8 @@ export class WebSocketClient { this.socket.onopen = () => { this.reconnectAttempts = 0; this.setStatus("connected"); - this.flushQueue(); this.resubscribe(); + this.flushQueue(); }; this.socket.onmessage = (event) => { @@ -126,6 +139,7 @@ export class WebSocketClient { this.socket = null; } this.setStatus("disconnected"); + this.resetSessionSubscriptionReadiness(new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR)); this.cleanupPendingRequests(); } @@ -184,18 +198,47 @@ export class WebSocketClient { } subscribeSession(sessionId: string) { + return this.subscribeSessionWithReady(sessionId).unsubscribe; + } + + subscribeSessionWithReady(sessionId: string): SessionSubscriptionHandle { const currentCount = this.sessionSubscriptions.get(sessionId) ?? 0; const nextCount = currentCount + 1; this.sessionSubscriptions.set(sessionId, nextCount); - if (this.status === "connected" && nextCount === 1) { - this.send({ - id: generateUUID(), - type: "request", - action: "session.subscribe", - payload: { session_id: sessionId }, - }); + const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId); + if (this.status === "connected" && this.socket) { + this.startSessionSubscription(sessionId, readiness); } - return () => this.unsubscribeSession(sessionId); + return { + ready: readiness.promise, + unsubscribe: () => this.unsubscribeSession(sessionId), + }; + } + + resubscribeSession(sessionId: string): Promise { + if (!this.sessionSubscriptions.has(sessionId)) return Promise.resolve(); + const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId, true); + if (this.status === "connected" && this.socket) { + this.startSessionSubscription(sessionId, readiness); + } + return readiness.promise; + } + + getSessionSubscriptionReadiness(sessionId: string): Promise { + if (!this.sessionSubscriptions.has(sessionId)) return Promise.resolve(); + if ( + !this.sessionSubscriptionReadiness.has(sessionId) && + (this.status === "disconnected" || this.status === "error") + ) { + const unavailable = Promise.reject(new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR)); + void unavailable.catch(() => undefined); + return unavailable; + } + const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId); + if (this.status === "connected" && this.socket) { + this.startSessionSubscription(sessionId, readiness); + } + return readiness.promise; } focusSession(sessionId: string) { @@ -282,13 +325,12 @@ export class WebSocketClient { unsubscribeSession(sessionId: string) { const currentCount = this.sessionSubscriptions.get(sessionId); - if (!currentCount) { - return; - } + if (!currentCount) return; const nextCount = currentCount - 1; if (nextCount <= 0) { this.sessionSubscriptions.delete(sessionId); + this.cancelSessionSubscriptionReadiness(sessionId); if (this.status === "connected") { this.send({ id: generateUUID(), @@ -455,6 +497,14 @@ export class WebSocketClient { private handleDisconnect(event: CloseEvent) { this.setStatus("disconnected"); + const shouldRetainSessionReadiness = + !this.intentionalClose && + this.reconnectOptions.enabled && + this.reconnectAttempts < this.reconnectOptions.maxAttempts; + this.resetSessionSubscriptionReadiness( + new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR), + shouldRetainSessionReadiness, + ); // Don't reconnect if this was an intentional close if (this.intentionalClose) { @@ -507,11 +557,82 @@ export class WebSocketClient { // Reject all pending requests this.pendingRequests.forEach(({ reject, timeout }) => { clearTimeout(timeout); - reject(new Error("WebSocket connection closed")); + reject(new Error(WEBSOCKET_CONNECTION_CLOSED_ERROR)); }); this.pendingRequests.clear(); } + private getOrCreateSessionSubscriptionReadiness( + sessionId: string, + forceNewAfterSettled = false, + ): SessionSubscriptionReadiness { + const existing = this.sessionSubscriptionReadiness.get(sessionId); + if (existing && !(forceNewAfterSettled && existing.settled)) return existing; + + let resolve!: () => void; + let reject!: (reason: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + const readiness: SessionSubscriptionReadiness = { + promise, + resolve, + reject, + requestStarted: false, + settled: false, + }; + // subscribeSession() consumers do not await readiness, so handle failures + // while returning the original promise to readiness-aware consumers. + void promise.catch(() => undefined); + this.sessionSubscriptionReadiness.set(sessionId, readiness); + return readiness; + } + + private startSessionSubscription(sessionId: string, readiness: SessionSubscriptionReadiness) { + if (readiness.requestStarted) return; + readiness.requestStarted = true; + void this.request("session.subscribe", { session_id: sessionId }).then( + () => { + if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return; + readiness.settled = true; + readiness.resolve(); + }, + (error: unknown) => { + if (this.sessionSubscriptionReadiness.get(sessionId) === readiness) { + this.sessionSubscriptionReadiness.delete(sessionId); + } + readiness.settled = true; + readiness.reject(error); + }, + ); + } + + private cancelSessionSubscriptionReadiness(sessionId: string) { + const readiness = this.sessionSubscriptionReadiness.get(sessionId); + if (!readiness) return; + this.sessionSubscriptionReadiness.delete(sessionId); + if (!readiness.settled) { + readiness.settled = true; + readiness.reject(new Error("Session subscription released")); + } + } + + private resetSessionSubscriptionReadiness(error: Error, retainActiveSubscriptions = false) { + const readinessEntries = [...this.sessionSubscriptionReadiness.entries()]; + this.sessionSubscriptionReadiness.clear(); + for (const [, readiness] of readinessEntries) { + if (readiness.settled) continue; + readiness.settled = true; + readiness.reject(error); + } + if (retainActiveSubscriptions) { + this.sessionSubscriptions.forEach((_, id) => + this.getOrCreateSessionSubscriptionReadiness(id), + ); + } + } + private resubscribe() { // Re-subscribe to all tasks after reconnection this.subscriptions.forEach((_count, taskId) => { @@ -523,12 +644,8 @@ export class WebSocketClient { }); }); this.sessionSubscriptions.forEach((_count, sessionId) => { - this.send({ - id: generateUUID(), - type: "request", - action: "session.subscribe", - payload: { session_id: sessionId }, - }); + const readiness = this.getOrCreateSessionSubscriptionReadiness(sessionId); + this.startSessionSubscription(sessionId, readiness); }); this.sessionFocusCounts.forEach((_count, sessionId) => { this.send({ diff --git a/docs/plans/session-subscription-recovery/plan.md b/docs/plans/session-subscription-recovery/plan.md new file mode 100644 index 0000000000..eea98e0144 --- /dev/null +++ b/docs/plans/session-subscription-recovery/plan.md @@ -0,0 +1,138 @@ +--- +spec: docs/specs/session-subscription-recovery/spec.md +created: 2026-08-05 +status: implemented +issue: https://github.com/kdlbs/kandev/issues/2287 +--- + +# Implementation Plan: Session subscription recovery + +## Overview + +The current server already inserts the session membership before sending the +`session.subscribe` acknowledgement and already provides state snapshots, +initial message fetches, and backfills. The gap is on the web client: it starts +`message.list` immediately after enqueueing the subscribe frame, while backend +WebSocket handlers run concurrently. Add an acknowledgement-aware readiness +handle to the ref-counted WebSocket client, make the transcript hook await it, +and route reconnect/retry registrations through the same tracked request. Keep +the existing recovery layers as defense-in-depth and remove only the E2E +workarounds that are specifically masking this race. + +## Confirmed root cause + +`WebSocketClient.subscribeSession` currently calls `send` and returns an +unsubscribe function without exposing the server response. `useSessionSubscription` +then calls `fetchAndStoreMessages` in the same effect. Since the backend starts +each received message handler in its own goroutine, `message.list` can run +before `handleSessionSubscribe` reaches `Hub.SubscribeToSession`; an event +published in that interval is not delivered, and the fetch can also complete +before the persisted row is visible. The issue's reload fallback rehydrates the +same durable state on a later page load. + +## Frontend + +### Acknowledgement-aware WebSocket membership + +`apps/web/lib/ws/client.ts` + +- Centralize session-subscribe frame creation so initial subscriptions, + reconnect resubscriptions, and explicit delayed-session retries use + `request("session.subscribe", ...)` and track the pending acknowledgement. +- Preserve the existing `subscribeSession(sessionId): () => void` API for + consumers that only need membership, and add a readiness-bearing companion + used by transcript hydration (for example, a handle with `unsubscribe` and + `ready`). +- Share an in-flight readiness promise for ref-counted consumers, clear it on + success/failure/disconnect, and ensure discarded requests cannot produce an + unhandled rejection. + +### Transcript hydration ordering + +`apps/web/hooks/domains/session/use-session-messages.ts` + +- Use the readiness-bearing subscription handle. +- Start `fetchAndStoreMessages` only after the subscription acknowledgement; + preserve cleanup guards so an unmounted hook does not apply a late result. +- Keep the existing turn-settle, running-turn, visibility, and initial state + snapshot reconciliation paths unchanged unless the focused regression test + proves a duplicate request is introduced. + +`apps/web/hooks/domains/session/use-session-subscription-retry.ts` + +- Replace the raw duplicate `client.send` retry with the tracked + acknowledgement-aware session-subscribe operation, retaining the current + backoff and ref-count ownership. + +### E2E workaround cleanup + +`apps/web/e2e/pages/session-page.ts` and the smallest existing session/chat +specs that exercise the documented reload fallback should stop treating a page +reload as the normal recovery for this race. Preserve bounded waits for genuine +startup/CI slowness, but do not silently reload solely because a subscription +event was missed. Use the existing auto-start and follow-up-response scenarios +as end-to-end evidence; add a focused regression spec only if the deterministic +test setup can force the registration delay without timing sleeps. + +## Tests + +- **WebSocket client readiness:** add `apps/web/lib/ws/client.test.ts` covering + first registration acknowledgement, shared in-flight readiness, rejected + registration cleanup, and reconnect resubscription tracking. +- **Transcript ordering:** extend + `apps/web/hooks/domains/session/use-session-messages.test.ts` to prove that + `message.list` is not requested until the subscribe readiness promise + resolves, and that a late resolution after cleanup does not mutate the + store. +- **Retry contract:** update + `apps/web/hooks/domains/session/use-session-subscription-retry.test.ts` to + assert that delayed retries use the tracked subscribe operation and retain + their backoff behavior. +- **Existing backend contract:** run the focused gateway tests covering + `session.subscribe` acknowledgement and initial snapshot delivery; no + backend production change is planned unless those tests expose a violation + of the acknowledgement ordering assumed above. + +## E2E Tests + +- **Scenario:** GIVEN a freshly auto-started mock-agent task and a follow-up + prompt, WHEN the session page mounts or resumes, THEN the idle state and + persisted agent reply render without the page-object reload fallback. +- **Files:** the relevant existing cases under + `apps/web/e2e/tests/session/auto-start-session.spec.ts` and + `apps/web/e2e/tests/session/session-recovery.spec.ts`; add a dedicated + `apps/web/e2e/tests/session/session-subscribe-recovery.spec.ts` only if a + deterministic registration barrier is available. +- **How:** run the chromium session specs with the existing isolated backend; + include the mobile project if the affected helper is changed for mobile. + +## Verification Results + +- `cd apps && pnpm --filter @kandev/web test -- --run lib/ws/client.test.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.test.ts` — 3 files and 46 tests passed. +- `cd apps/web && pnpm run typecheck` — passed. +- `cd apps/web && pnpm exec eslint --max-warnings 0 lib/ws/client.ts lib/ws/client.test.ts hooks/domains/session/use-session-messages.ts hooks/domains/session/use-session-message-fetch.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.ts hooks/domains/session/use-session-subscription-retry.test.ts e2e/pages/session-page.ts e2e/tests/chat/mobile-unread-divider.spec.ts e2e/tests/chat/unread-divider.spec.ts` — passed. +- `cd apps/backend && go test -race -run 'TestHandleSessionSubscribe|TestBroadcastToSession|TestHandleSessionDataRefresh' ./internal/gateway/websocket/...` — 6 tests passed. +- `cd apps/web && pnpm e2e:run --project chromium tests/session/auto-start-session.spec.ts tests/session/session-recovery.spec.ts` — 5 tests passed. +- `cd apps/web && pnpm e2e:run --project mobile-chrome tests/session/mobile-pause-resume-recovery.spec.ts` — 2 tests passed. +- `git diff --check` — passed. + +The backend acknowledgement and membership ordering already satisfied the +contract, so no backend production change was needed. The web client now tracks +the acknowledgement for initial, reconnect, and delayed retry subscriptions; +transcript hydration waits on that readiness and ignores late results after +cleanup. The session page-object reload fallbacks specific to this race were +removed while the unrelated bounded startup reload remains. + +## Implementation Waves And Parallel Candidates + +Wave 1 (sequential): + +- [x] [task-01-ack-gated-session-hydration](task-01-ack-gated-session-hydration.md) + +The client, hook, retry path, and E2E helper share the same subscription +contract, so they are intentionally one sequential task. + +## Open Questions + +- The race-specific reload fallback can be removed; the general bounded + `waitForLoad` reload remains for unrelated startup/SSR hydration failures. diff --git a/docs/plans/session-subscription-recovery/task-01-ack-gated-session-hydration.md b/docs/plans/session-subscription-recovery/task-01-ack-gated-session-hydration.md new file mode 100644 index 0000000000..f73ffa1713 --- /dev/null +++ b/docs/plans/session-subscription-recovery/task-01-ack-gated-session-hydration.md @@ -0,0 +1,100 @@ +--- +id: "01-ack-gated-session-hydration" +title: "Gate session hydration on subscribe acknowledgement" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/session-subscription-recovery/spec.md" +--- + +# Task 01: Gate session hydration on subscribe acknowledgement + +## Acceptance + +- A transcript hydration request cannot overtake the server's successful + `session.subscribe` registration, including shared, reconnect, and delayed + retry paths. +- Existing ref-counted consumers still subscribe/unsubscribe correctly, and + failed or disconnected requests do not leave unhandled rejections or stale + readiness state. +- The existing session auto-start and follow-up-response flows no longer need a + reload to recover a missed subscription event; unrelated bounded startup + waits remain intact. + +## Verification + +From a fresh worktree when dependencies are absent: + +```bash +cd apps && pnpm install --frozen-lockfile +``` + +Targeted checks: + +```bash +cd apps && pnpm --filter @kandev/web test -- --run lib/ws/client.test.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.test.ts +cd apps/web && pnpm run typecheck +cd apps/web && pnpm exec eslint --max-warnings 0 lib/ws/client.ts lib/ws/client.test.ts hooks/domains/session/use-session-messages.ts hooks/domains/session/use-session-message-fetch.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.ts hooks/domains/session/use-session-subscription-retry.test.ts e2e/pages/session-page.ts e2e/tests/chat/mobile-unread-divider.spec.ts e2e/tests/chat/unread-divider.spec.ts +cd apps/backend && go test -race -run 'TestHandleSessionSubscribe|TestBroadcastToSession|TestHandleSessionDataRefresh' ./internal/gateway/websocket/... +cd apps/web && pnpm e2e:run --project chromium tests/session/auto-start-session.spec.ts tests/session/session-recovery.spec.ts +``` + +If the page-object change affects mobile behavior, also run: + +```bash +cd apps/web && pnpm e2e:run --project mobile-chrome tests/session/mobile-pause-resume-recovery.spec.ts +``` + +## Files likely touched + +- `apps/web/lib/ws/client.ts` +- `apps/web/lib/ws/client.test.ts` +- `apps/web/hooks/domains/session/use-session-messages.ts` +- `apps/web/hooks/domains/session/use-session-message-fetch.ts` +- `apps/web/hooks/domains/session/use-session-messages.test.ts` +- `apps/web/hooks/domains/session/use-session-subscription-retry.ts` +- `apps/web/hooks/domains/session/use-session-subscription-retry.test.ts` +- `apps/web/e2e/pages/session-page.ts` +- `apps/web/e2e/tests/chat/mobile-unread-divider.spec.ts` +- `apps/web/e2e/tests/chat/unread-divider.spec.ts` + +## Dependencies + +None. + +## Parallelism + +Sequential. The client readiness contract is shared by every implementation +file and must be tested before E2E workaround cleanup. + +## Inputs + +- `docs/specs/session-subscription-recovery/spec.md` +- `docs/plans/session-subscription-recovery/plan.md` +- Existing backend acknowledgement/snapshot behavior in + `apps/backend/internal/gateway/websocket/client.go` and + `apps/backend/internal/backendapp/helpers.go` +- Existing ref-counted subscription behavior in + `apps/web/lib/ws/client.ts` + +## Results + +Implemented the acknowledgement-gated subscription contract across the web +client and session message hooks. Initial hydration, reconnect +re-registration, and delayed unknown-session retries now use tracked subscribe +requests; ref-counted consumers share readiness, and cleanup prevents late +hydration from mutating state. The race-specific reload fallbacks were removed +from the session page object and affected chat specs. No backend production +change was needed because the existing gateway acknowledgement ordering was +already correct. + +Verification: + +- Focused web tests: 3 files, 46 tests passed. +- Web typecheck passed. +- Focused ESLint passed with zero warnings. +- Backend gateway race tests: 6 passed. +- Desktop Chromium session E2E: 5 passed. +- Mobile Chrome recovery E2E: 2 passed. +- `git diff --check` passed. diff --git a/docs/specs/INDEX.md b/docs/specs/INDEX.md index 6de509d749..72ae4a3f0b 100644 --- a/docs/specs/INDEX.md +++ b/docs/specs/INDEX.md @@ -200,6 +200,7 @@ System pages (Radarr/Sonarr-style) for status, disk usage, database maintenance, | Spec | Status | |---|---| +| [session-subscription-recovery](session-subscription-recovery/spec.md) | draft | | [npm-nightly-channel](npm-nightly-channel/spec.md) | shipped | | [agent-resume-runtime-recovery](agent-resume-runtime-recovery/spec.md) | shipped | | [agent-stall-recovery](agent-stall-recovery/spec.md) | approved | diff --git a/docs/specs/session-subscription-recovery/spec.md b/docs/specs/session-subscription-recovery/spec.md new file mode 100644 index 0000000000..89b87c287e --- /dev/null +++ b/docs/specs/session-subscription-recovery/spec.md @@ -0,0 +1,61 @@ +--- +status: draft +created: 2026-08-05 +owner: carlosflorencio +issue: https://github.com/kdlbs/kandev/issues/2287 +--- + +# Session subscription recovery + +## Why + +A fast task session can change state or persist a reply while the browser is +waiting for the backend to register its `session.subscribe` request. The live +notification is then unavailable to that client, leaving the task looking busy +or the transcript missing a reply until the user reloads the page. + +## What + +- A session subscription exposes a readiness point that resolves only after the + backend has acknowledged registration. +- Transcript reconciliation for a newly mounted or reconnected session starts + after that readiness point, so the persisted message view covers the interval + before registration and live notifications cover the interval after it. +- Ref-counted consumers share an in-flight registration readiness point and do + not unregister one another's subscription. +- Reconnect and delayed-session retry paths use the same acknowledgement-aware + registration behavior. +- A failed or disconnected registration does not create an unhandled client + rejection or leave a stale readiness promise that prevents the next + connection from hydrating the session. + +## Scenarios + +- **GIVEN** a browser sends `session.subscribe` while a session state change or + reply is persisted before the backend registers the subscription, **WHEN** the + backend acknowledges the subscription and the client reconciles messages, + **THEN** the client displays the authoritative state and persisted reply + without a page reload. +- **GIVEN** one client-side consumer is already registering a session and a + second consumer mounts, **WHEN** the second consumer requests hydration, + **THEN** it waits for the shared registration acknowledgement and does not + race an earlier `message.list` request ahead of registration. +- **GIVEN** a session subscription is restored after a WebSocket reconnect, + **WHEN** the restored registration is acknowledged, **THEN** message + reconciliation can run against the restored subscription and later events are + delivered normally. +- **GIVEN** the registration fails or the socket disconnects before the + acknowledgement, **WHEN** the client retries on the next connected socket, + **THEN** the retry can establish a fresh readiness point and no stale promise + or unhandled rejection blocks hydration. + +## Out of scope + +- Adding durable event sequence numbers, server-side replay cursors, or a + general WebSocket backlog. +- Changing the persisted session/message schema or event publication order. +- Removing the existing state snapshot, backfill, or foreground recovery paths; + they remain defense-in-depth for later disconnects and long-running turns. +- Retaining reload-based E2E workarounds after the repair is proven; their + cleanup is an implementation and validation task, not a new recovery + contract.