fix(web): gate session hydration on subscription ack (#2287) (#2296)

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<HTMLDivElement>(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<HTMLElement>(`[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 (
-    <div ref={scrollRef}>
+    <div ref={scrollRef} data-testid={DIVIDER_SCROLL_CONTAINER_TEST_ID}>
       <div id="msg-m1" />
       <div id={`msg-${DIVIDER_KEY}`} />
     </div>
   );
 }
 
+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(<Harness itemCount={2} anchoredBarOffsetPx={0} />);
-    expect(scrollIntoView).toHaveBeenCalledTimes(1);
+    const scrollContainer = document.querySelector<HTMLElement>(
+      `[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(<Harness itemCount={2} anchoredBarOffsetPx={76} />);
 
-    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(
+      <Harness itemCount={2} anchoredBarOffsetPx={0} scrollLayoutKey="loading" />,
+    );
+    const scrollContainer = document.querySelector<HTMLElement>(
+      `[data-testid="${DIVIDER_SCROLL_CONTAINER_TEST_ID}"]`,
+    );
+    if (!scrollContainer) throw new Error(MISSING_SCROLL_CONTAINER_ERROR);
+    expect(scrollContainer.scrollTop).toBe(150);
+
+    rerender(
+      <Harness
+        itemCount={2}
+        anchoredBarOffsetPx={0}
+        scrollLayoutKey="settled"
+        dividerDocumentTop={166}
+      />,
+    );
+
+    expect(scrollContainer.scrollTop).toBe(66);
   });
 
   it("resynchronizes auto-scroll state after placing the divider", () => {
-    HTMLElement.prototype.scrollIntoView = vi.fn();
     const onDividerScroll = vi.fn();
 
     render(<Harness itemCount={2} anchoredBarOffsetPx={0} onDividerScroll={onDividerScroll} />);
@@ -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(<Harness itemCount={2} anchoredBarOffsetPx={0} />);
-    expect(scrollIntoView).toHaveBeenCalledTimes(1);
+    const scrollContainer = container.querySelector<HTMLElement>(
+      `[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(<Harness itemCount={2} anchoredBarOffsetPx={76} />);
 
-    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(
       <Harness itemCount={2} anchoredBarOffsetPx={0} dividerKey={null} />,
     );
-    expect(scrollIntoView).not.toHaveBeenCalled();
+    const scrollContainer = document.querySelector<HTMLElement>(
+      `[data-testid="${DIVIDER_SCROLL_CONTAINER_TEST_ID}"]`,
+    );
+    if (!scrollContainer) throw new Error(MISSING_SCROLL_CONTAINER_ERROR);
+    expect(scrollContainer.scrollTop).toBe(0);
 
     rerender(<Harness itemCount={2} anchoredBarOffsetPx={76} dividerKey={null} />);
 
-    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(<Harness itemCount={2} anchoredBarOffsetPx={0} />);
-    expect(scrollIntoView).toHaveBeenCalledTimes(1);
+    const scrollContainer = document.querySelector<HTMLElement>(
+      `[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(<Harness itemCount={2} anchoredBarOffsetPx={76} />);
 
-    expect(scrollIntoView).toHaveBeenCalledTimes(1);
+    expect(scrollContainer.scrollTop).toBe(150);
     vi.useRealTimers();
   });
 });

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<HTMLElement>(`[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 (

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.

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);
   });
 });

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<typeof attachGatewayTrafficCapture>,
+  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: <task-notification>" 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("<task-notification>");
-    await expect(session.chat).not.toContainText("Human: <task");
+    await expect(session.activeChat()).not.toContainText("<task-notification>");
+    await expect(session.activeChat()).not.toContainText("Human: <task");
   });
 
   test("singular event count uses 'event' not 'events'", async ({
@@ -79,35 +135,29 @@ test.describe("Claude-acp Monitor tool", () => {
     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);
   });
 });

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);
   });
 

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",

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<typeof useAppStoreApi>;
+
+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<number> {
+  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<BackfillStep> {
+  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<void> {
+  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,
+  });
+}

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<T>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
+function makeParams(
+  fetchAndStoreMessages: (
+    sessionId: string,
+    store: never,
+    isActive?: () => boolean,
+  ) => Promise<Message[]>,
+  setMessagesLoading: ReturnType<typeof vi.fn>,
+) {
+  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<Message[]>();
+    const second = deferred<Message[]>();
+    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);
+  });
+});

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<typeof useAppStoreApi>;
+
+// 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<string, number>();
+
+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<number | null>;
+  lastFetchedSessionIdRef: MutableRefObject<string | null>;
+  fetchAndStoreMessages: (
+    sessionId: string,
+    store: SessionMessageStore,
+    isActive?: () => boolean,
+  ) => Promise<Message[]>;
+  autoBackfillUntilUserMessage: (
+    sessionId: string,
+    store: SessionMessageStore,
+    isActive?: () => boolean,
+  ) => Promise<void>;
+  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<void> {
+  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);
+  }
+}

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<T>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
+describe("session subscription hydration ordering", () => {
+  it("does not request messages before subscription acknowledgement", async () => {
+    const readiness = deferred<void>();
+    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<void>();
+    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);
+  });
+});

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<TaskSessionState> = 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<void>;
+  promise: Promise<MessageListResponse>;
+};
 
 const EMPTY_MESSAGES: Message[] = [];
 const EMPTY_META = { isLoading: false, hasMore: false, oldestCursor: null };
+const inFlightMessageRequests = new Map<string, InFlightMessageRequest>();
 
 /** 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<ReturnType<typeof getWebSocketClient>>,
   sessionId: string,
-  store: ReturnType<typeof useAppStoreApi>,
-): Promise<Message[]> {
-  const client = getWebSocketClient();
-  if (!client) {
-    return [];
-  }
-  const seq = nextFetchSeq();
+  readiness: Promise<void>,
+): Promise<MessageListResponse> {
+  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<MessageListResponse>("message.list", requestParams, 10000);
+  const promise = client.request<MessageListResponse>("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<typeof useAppStoreApi>,
+  isActive?: () => boolean,
+): Promise<Message[]> {
+  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<typeof useAppStoreApi>,
-  oldestCursor: string,
-): Promise<number> {
-  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<typeof useAppStoreApi>,
-  round: number,
-): Promise<BackfillStep> {
-  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<typeof useAppStoreApi>,
-): Promise<void> {
-  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<typeof useAppStoreApi>;
-  setIsLoading: (v: boolean) => void;
-  setIsWaitingForInitialMessages: (v: boolean) => void;
-  initialFetchStartRef: MutableRefObject<number | null>;
-  lastFetchedSessionIdRef: MutableRefObject<string | null>;
-  onError?: (error: unknown) => void;
-};
-
-async function doFetchMessages({
-  taskSessionId,
-  store,
-  setIsLoading,
-  setIsWaitingForInitialMessages,
-  initialFetchStartRef,
-  lastFetchedSessionIdRef,
-  onError,
-}: FetchMessagesParams): Promise<void> {
-  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);

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();
   });
 });

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]);
 }

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" }),

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<typeof WebSocketClient>[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();
+  });
+});

apps/web/lib/ws/client.ts

@@ -19,9 +19,7 @@ const DISPATCH_LOG_DENYLIST = new Set<string>([
 
 type MessageHandler<T extends BackendMessageType> = (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<void>;
+  unsubscribe: () => void;
+}
+
+type SessionSubscriptionReadiness = {
+  promise: Promise<void>;
+  resolve: () => void;
+  reject: (reason: unknown) => void;
+  requestStarted: boolean;
+  settled: boolean;
+};
+
 const DEFAULT_RECONNECT_OPTIONS: Required<ReconnectOptions> = {
   enabled: true,
   maxAttempts: 10,
@@ -39,6 +50,7 @@ const DEFAULT_RECONNECT_OPTIONS: Required<ReconnectOptions> = {
   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<string, number>();
   private sessionSubscriptions = new Map<string, number>();
+  private sessionSubscriptionReadiness = new Map<string, SessionSubscriptionReadiness>();
   // 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<void> {
+    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<void> {
+    if (!this.sessionSubscriptions.has(sessionId)) return Promise.resolve();
+    if (
+      !this.sessionSubscriptionReadiness.has(sessionId) &&
+      (this.status === "disconnected" || this.status === "error")
+    ) {
+      const unavailable = Promise.reject<void>(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<void>((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({

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.

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.

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 |

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.