Keep large routed timelines connected when the receiver falls behind (#9109)

* fix(server): pace routed output to its receiving client

A routed transfer previously waited only for the destination-to-intermediary write, so a slower final hop overflowed the route queue. Negotiate bounded per-channel acknowledgments after final forwarding drains, preserving concurrent small sends and legacy peers.

* fix(transport): keep small routed writes concurrent

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

docs/agent-stream-disclosure.md

@@ -1,8 +1,8 @@
 # Agent stream disclosures
 
 Expanding or collapsing something in the chat timeline (a turn's summary line, a tool-call group,
-a tool call's inline details, a todo card) keeps the tapped trigger at the same screen y in the
-first painted frame. No animation.
+a tool call's inline details, a todo card, a long user message's show more, a plan card) keeps the
+tapped trigger at the same screen y in the first painted frame. No animation.
 
 ## The contract
 
@@ -14,12 +14,20 @@ const handleToggle = () => hold(() => setExpanded((value) => !value));
 ```
 
 `view.tsx` wraps every timeline row in `DisclosureRow`, which routes `hold` to the strategy's
-`holdRow(rowId, apply)` (`packages/app/src/agent-stream/disclosure.tsx`). Outside a timeline row,
-`apply` runs as is. The strategy owns when `apply` runs and how the row keeps its place.
+`holdRow(rowId, apply, triggerEdge)` (`packages/app/src/agent-stream/disclosure.tsx`). Outside a
+timeline row, `apply` runs as is. The strategy owns when `apply` runs and how the row keeps its place.
+A toggle that calls `setState` directly skips the hold: on Android the expanded content paints
+over the next row for a frame before the list learns the new size.
+
+`triggerEdge` names the edge of the row the trigger sits on, and the strategy keeps that edge in
+place. Header triggers sit on the top edge, the default. A long user message's show more sits under
+the text it opens, so `MessagePreview` passes `{ triggerEdge: "bottom" }`: the text grows and shrinks
+above the trigger, and the rows below stay put.
 
 Following output:
 
-- A hold always detaches the bottom anchor controller (`holdForDisclosure`).
+- A hold detaches the bottom anchor controller (`holdForDisclosure`), except a bottom-edge hold on
+  native while following: there the end pin already keeps the row's bottom in place.
 - When the hold ends, the controller re-sticks only if the list sits at its end and no user scroll
   owns the viewport (`releaseDisclosure({ isAtEnd })`). Expanding at the bottom therefore leaves
   the list detached with the scroll-to-bottom pill showing; collapsing back to the end resumes
@@ -33,13 +41,16 @@ ends at its end and the trigger moves by the difference.
 
 ## Web
 
-`holdRow` pins the row in the reading anchor and runs `apply`; the next reading reconcile restores
-the row's top after layout, before paint (`strategy-web.tsx`).
+`holdRow` pins the row's trigger edge in the reading anchor and runs `apply`; the next reading
+reconcile restores that edge after layout, before paint (`strategy-web.tsx`). The pin records the
+scroll offset it was taken at, because the browser clamps `scrollTop` when a collapse shrinks the
+content past the offset, before the reconcile reads it. Restoring from the clamped offset would
+apply the shrink twice.
 
 ## Native
 
 On compact layouts, tool calls and tool-call groups open a sheet instead of expanding, so they do
-not hold. Summary lines and todo cards expand inline.
+not hold. Summary lines, todo cards, plan cards and show more expand inline.
 
 The timeline is a LegendList in forward order: one `data` array of history rows followed by the
 live head, oldest first (`strategy-native.tsx`). LegendList keeps a row in place across a size
@@ -51,14 +62,14 @@ toggle; its applied commit took up to 3.5 s of JS on a long turn.
 
 One owner keeps the scroll offset at any moment:
 
-| State                  | Owner                                                                                     |
-| ---------------------- | ----------------------------------------------------------------------------------------- |
-| Before rows are placed | LegendList's initial scroll to the end. It keeps the rows hidden until they are measured. |
-| Following the end      | The end pin (`native-end-pin.tsx`).                                                       |
-| Disclosure hold        | LegendList's `maintainVisibleContentPosition`, restoring only the held row.               |
-| User scroll            | The finger. The controller detaches at drag start.                                        |
-| Detached               | LegendList's `maintainVisibleContentPosition` on the first visible row.                   |
-| Keyboard motion        | Nobody. The keyboard translates the list with the composer and leaves the offset alone.   |
+| State                  | Owner                                                                                                             |
+| ---------------------- | ----------------------------------------------------------------------------------------------------------------- |
+| Before rows are placed | LegendList's initial scroll to the end. It keeps the rows hidden until they are measured.                         |
+| Following the end      | The end pin (`native-end-pin.tsx`), also for a bottom-edge hold.                                                  |
+| Disclosure hold        | LegendList's `maintainVisibleContentPosition` on the hold's anchor row, or the end pin for the last row's bottom. |
+| User scroll            | The finger. The controller detaches at drag start.                                                                |
+| Detached               | LegendList's `maintainVisibleContentPosition` on the first visible row.                                           |
+| Keyboard motion        | Nobody. The keyboard translates the list with the composer and leaves the offset alone.                           |
 
 The end pin is a zero-height view at the bottom of the content and the content's first native
 child, with React Native's `maintainVisibleContentPosition` set to `{ minIndexForVisible: 0 }`. The
@@ -78,15 +89,38 @@ jumps by the keyboard height after the keyboard settles.
 
 ### The hold
 
-`holdRow(rowId, apply)` runs in the toggle's event:
+`holdRow(rowId, apply, triggerEdge)` runs in the toggle's event:
 
-1. Mark the row held and detach the controller (`holdForDisclosure`). The pin goes away and
-   LegendList's `maintainVisibleContentPosition` takes over, restoring only the held row
-   (`shouldRestorePosition`). Upstream LegendList anchors only rows from the first fully visible
-   one down, and a size change with no data change keeps the first row in view whatever
+1. Pick the anchor and detach the controller (`holdForDisclosure`). LegendList restores row
+   tops only, so a top-edge hold anchors the toggled row and a bottom-edge hold anchors the row
+   below it, whose top is the toggled row's bottom (`disclosureAnchor`). The pin goes away
+   and LegendList's `maintainVisibleContentPosition` takes over, restoring only the anchor row
+   (`shouldRestorePosition`). A bottom-edge hold while following skips this step: the end pin
+   keeps the content end, and the row's bottom with it, in the frame the new height mounts.
+
+   A bottom-edge hold on the last row, such as a prompt the agent has not answered yet, has no
+   row below it: its bottom is the content end (the running indicator is the list footer, not a
+   row). The hold mounts the end pin while detached, and `apply` runs a frame later. The pin
+   anchors only once it is mounted, and a pin that mounts in the same commit as the new height
+   anchors nothing. The pin's commit mounts before that frame's JS callbacks run, so `apply`'s
+   commit mounts with the pin already in place. The pin keeps the distance from the end, and each
+   content size change clamps it to the scroll room left, so the hold re-sticks if the list was
+   at its end when tapped or a collapse used up the room. The release tracks that distance
+   instead of reading the offset: the new content size reaches JS before the pin's scroll does.
+   A reply landing below the row ends the pin in the commit that mounts the reply, so the reply
+   grows below the row and the trigger stays where it was. A reply that lands before `apply` runs
+   moves the hold to the reply's row.
+
+   Upstream LegendList anchors only rows from the first fully visible one down to the last fully
+   visible one, and a size change with no data change keeps the first row in view whatever
    `shouldRestorePosition` says. A row cut off at the top would then let the row below it hold
-   instead, moving the trigger by the row's size change. The LegendList patch counts rows cut off
-   at the top as in view and applies `shouldRestorePosition` to size changes too.
+   instead, and a bottom-edge anchor just under the viewport would not hold at all. The LegendList
+   patch counts the rows cut off at the top, and the row just past the bottom, as in view and
+   applies `shouldRestorePosition` to size changes too. It also drops LegendList's Android clamp
+   prediction: the scroll view clamps after the adjustment
+   ([scroll view corrections](#scroll-view-corrections)), so the prediction moved a collapse near
+   the end to the end.
+
 2. Run `apply`.
 3. Call the row's `useSyncLayout` trigger. The toggle changes the row from its own state, so
    LegendList's container does not re-render and would learn the new height from `onLayout` a task
@@ -130,7 +164,7 @@ With every turn collapsed, flinging from a cold open into history did not blank
 
 ### LegendList patch
 
-`patches/@legendapp+list+3.7.1.patch` carries ten changes for the timeline. Its header lists
+`patches/@legendapp+list+3.7.1.patch` carries twelve changes for the timeline. Its header lists
 them by purpose, and
 [legend-list #593](https://github.com/LegendApp/legend-list/issues/593) reports each upstream.
 Drop a hunk when the installed LegendList ships its fix.

packages/app/e2e/browser/new-workspace-live-pin.spec.ts

@@ -1,6 +1,7 @@
 import type { Page } from "@playwright/test";
 import { expect, test } from "../support/fixtures";
 import { gotoAppShell } from "../support/helpers/app";
+import { expectComposerModel } from "../support/helpers/agent-profiles";
 import {
   delayBrowserAgentCreatedStatus,
   openNewWorkspaceComposer,
@@ -69,6 +70,8 @@ test.describe("New workspace live pin", () => {
       await gotoAppShell(page);
       await waitForSidebarHydration(page);
       await openNewWorkspaceComposer(page, seeded);
+      // The composer appears before its provider catalog has hydrated.
+      await expectComposerModel(page, "Ten second stream");
       await submitNewWorkspacePrompt(page, "Pin me while I start");
 
       await creation.waitForCreateRequest();

packages/app/src/agent-stream/disclosure.browser.test.tsx

@@ -0,0 +1,147 @@
+import React, { act, createRef, type ReactNode } from "react";
+import { createRoot, type Root } from "react-dom/client";
+import { within } from "@testing-library/dom";
+import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { i18n as testI18n } from "@/i18n/i18next";
+import { PlanCard } from "@/components/plan-card";
+import { MessagePreview } from "@/components/user-message";
+import { DisclosureRow } from "./disclosure";
+import type { DisclosureTriggerEdge, StreamViewportHandle } from "./strategy";
+
+// Load translations so the toggles expose their real accessible names.
+void testI18n;
+
+// App sources compile against the classic JSX runtime, which expects React on the global.
+beforeEach(() => vi.stubGlobal("React", React));
+
+let root: Root;
+let container: HTMLDivElement;
+// The plan body renders through the app's Markdown renderer, which reads through React Query.
+let queryClient: QueryClient;
+
+beforeEach(() => {
+  container = document.createElement("div");
+  document.body.appendChild(container);
+  root = createRoot(container);
+  queryClient = new QueryClient();
+});
+
+afterEach(() => {
+  act(() => root.unmount());
+  queryClient.clear();
+  container.remove();
+});
+
+interface Hold {
+  rowId: string;
+  triggerEdge: DisclosureTriggerEdge;
+  /** Whether the expanded content was already on screen when the strategy received the hold. */
+  appliedBefore: boolean;
+}
+
+/**
+ * A timeline row whose viewport records each hold. Like the strategies, it runs `apply` itself, so
+ * a toggle that skips the hold still changes state but leaves no record.
+ */
+function mountRow(rowId: string, isApplied: () => boolean, children: ReactNode) {
+  const holds: Hold[] = [];
+  const viewport: StreamViewportHandle = {
+    scrollToBottom: vi.fn(),
+    prepareForViewportChange: vi.fn(),
+    holdRow: (heldRowId, apply, triggerEdge) => {
+      holds.push({ rowId: heldRowId, triggerEdge, appliedBefore: isApplied() });
+      apply();
+    },
+  };
+  const viewportRef = createRef<StreamViewportHandle>();
+  viewportRef.current = viewport;
+  act(() =>
+    root.render(
+      <QueryClientProvider client={queryClient}>
+        <DisclosureRow rowId={rowId} viewportRef={viewportRef}>
+          {children}
+        </DisclosureRow>
+      </QueryClientProvider>,
+    ),
+  );
+  return holds;
+}
+
+const view = () => within(container);
+
+const preLine = { whiteSpace: "pre-line" } as const;
+
+async function waitForLayout() {
+  await act(async () => {
+    for (let frame = 0; frame < 2; frame += 1) {
+      await new Promise((resolve) => requestAnimationFrame(resolve));
+    }
+  });
+}
+
+describe("timeline disclosures hold their row", () => {
+  it("routes user message show more and show less through the row's hold", async () => {
+    const longText = Array.from({ length: 60 }, (_, line) => `Line ${line + 1}`).join("\n");
+    const holds = mountRow(
+      "message-1",
+      () => view().queryByRole("button", { name: "Show less" }) !== null,
+      <MessagePreview>
+        <span style={preLine}>{longText}</span>
+      </MessagePreview>,
+    );
+    // The preview learns it overflows from its first layout.
+    await waitForLayout();
+
+    act(() => view().getByRole("button", { name: "Show more" }).click());
+    expect(view().getByRole("button", { name: "Show less" })).toBeTruthy();
+
+    act(() => view().getByRole("button", { name: "Show less" }).click());
+    expect(view().getByRole("button", { name: "Show more" })).toBeTruthy();
+
+    expect(holds).toEqual([
+      // Show more sits under the text it opens.
+      { rowId: "message-1", triggerEdge: "bottom", appliedBefore: false },
+      { rowId: "message-1", triggerEdge: "bottom", appliedBefore: true },
+    ]);
+  });
+
+  it("routes plan card collapse and expand through the row's hold", () => {
+    const header = () => view().getByRole("button", { name: "Approved plan" });
+    const isExpanded = () => header().getAttribute("aria-expanded") === "true";
+    const holds = mountRow(
+      "plan-1",
+      isExpanded,
+      <PlanCard title="Plan" text="Step one" outcome="approved" />,
+    );
+
+    act(() => header().click());
+    expect(isExpanded()).toBe(false);
+
+    act(() => header().click());
+    expect(isExpanded()).toBe(true);
+
+    expect(holds).toEqual([
+      { rowId: "plan-1", triggerEdge: "top", appliedBefore: true },
+      { rowId: "plan-1", triggerEdge: "top", appliedBefore: false },
+    ]);
+  });
+
+  it("toggles a plan card outside a timeline row, as the pending permission card renders", () => {
+    act(() =>
+      root.render(
+        <QueryClientProvider client={queryClient}>
+          <PlanCard title="Plan" text="Step one" outcome="pending" />
+        </QueryClientProvider>,
+      ),
+    );
+    const header = () => view().getByRole("button", { name: "Plan" });
+    const isExpanded = () => header().getAttribute("aria-expanded") === "true";
+
+    act(() => header().click());
+    expect(isExpanded()).toBe(false);
+
+    act(() => header().click());
+    expect(isExpanded()).toBe(true);
+  });
+});

packages/app/src/agent-stream/disclosure.tsx

@@ -1,11 +1,15 @@
 import { createContext, useContext, useMemo, type ReactNode, type RefObject } from "react";
-import type { StreamViewportHandle } from "./strategy";
+import type { DisclosureTriggerEdge, StreamViewportHandle } from "./strategy";
 
 /**
- * Runs a disclosure's state change. Inside a timeline row, the row that owns the
- * trigger keeps its screen position through the change; elsewhere `apply` runs as is.
+ * Runs a disclosure's state change. Inside a timeline row, the trigger keeps its screen
+ * position through the change; elsewhere `apply` runs as is. A trigger below the content it
+ * opens passes `triggerEdge: "bottom"`.
  */
-export type DisclosureHold = (apply: () => void) => void;
+export type DisclosureHold = (
+  apply: () => void,
+  options?: { triggerEdge?: DisclosureTriggerEdge },
+) => void;
 
 const applyInPlace: DisclosureHold = (apply) => apply();
 
@@ -21,10 +25,10 @@ export function DisclosureRow({
   children: ReactNode;
 }) {
   const hold = useMemo<DisclosureHold>(
-    () => (apply) => {
+    () => (apply, options) => {
       const viewport = viewportRef.current;
       if (viewport) {
-        viewport.holdRow(rowId, apply);
+        viewport.holdRow(rowId, apply, options?.triggerEdge ?? "top");
       } else {
         apply();
       }

packages/app/src/agent-stream/legend-list-held-row.browser.test.tsx

@@ -3,6 +3,7 @@ import React, { act, useSyncExternalStore } from "react";
 import { createRoot, type Root } from "react-dom/client";
 import { View } from "react-native";
 import { afterEach, beforeEach, expect, it } from "vitest";
+import { disclosureAnchor } from "./native-disclosure-hold";
 
 // The native chat timeline runs on the patched LegendList (patches/@legendapp+list+3.7.1.patch).
 // A disclosure hold keeps only the held row in place: `shouldRestorePosition` accepts no other row.
@@ -130,7 +131,16 @@ function mountList(heldRowId: string) {
     return position - state.scroll;
   };
   const rowSize = (rowId: string) => listRef.current?.getState().sizes.get(rowId);
-  return { render, scrollTo, heldRowTop, rowSize };
+  const rowBottom = (rowId: string) => {
+    const state = listRef.current?.getState();
+    const position = state?.positionByKey(rowId);
+    const size = state?.sizes.get(rowId);
+    if (state === undefined || position === undefined || size === undefined) {
+      throw new Error(`Row ${rowId} is not laid out`);
+    }
+    return position + size - state.scroll;
+  };
+  return { render, scrollTo, heldRowTop, rowSize, rowBottom };
 }
 
 it("anchors a data change on a held row whose top is above the viewport", async () => {
@@ -171,6 +181,72 @@ it("anchors a size change on a held row whose top is above the viewport", async
   expect(list.heldRowTop()).toBe(-20);
 });
 
+// A long message's show more sits under the text it opens, on the bottom edge of its row. The
+// strategy anchors the row below it, whose top is that bottom edge.
+const bottomTriggerRows: Row[] = Array.from({ length: 10 }, (_, index) => ({
+  id: `row-${index}`,
+  height: index === 3 ? 60 : 40,
+}));
+
+function mountBottomTriggerHold(rowId: string) {
+  const anchor = disclosureAnchor(bottomTriggerRows, rowId, "bottom");
+  if (anchor.kind !== "row") throw new Error(`Row ${rowId} has no row below it`);
+  return mountList(anchor.rowId);
+}
+
+it("keeps the bottom of a row whose trigger sits under its growing content", async () => {
+  const list = mountBottomTriggerHold("row-3");
+  list.render(bottomTriggerRows);
+  await wait(150);
+  await list.scrollTo(100);
+  const triggerBottom = list.rowBottom("row-3");
+  expect(triggerBottom).toBe(80);
+
+  act(() => growRow("row-3", 260));
+  await wait(150);
+  expect(list.rowBottom("row-3")).toBe(triggerBottom);
+
+  act(() => growRow("row-3", 60));
+  await wait(150);
+  expect(list.rowBottom("row-3")).toBe(triggerBottom);
+});
+
+it("keeps the bottom of a bottom-trigger row whose top is above the viewport", async () => {
+  const list = mountBottomTriggerHold("row-3");
+  list.render(bottomTriggerRows);
+  await wait(150);
+  await list.scrollTo(140);
+  const triggerBottom = list.rowBottom("row-3");
+  expect(triggerBottom).toBe(40);
+
+  act(() => growRow("row-3", 260));
+  await wait(150);
+  expect(list.rowBottom("row-3")).toBe(triggerBottom);
+
+  act(() => growRow("row-3", 60));
+  await wait(150);
+  expect(list.rowBottom("row-3")).toBe(triggerBottom);
+});
+
+it("keeps the bottom of a bottom-trigger row that ends below the viewport", async () => {
+  // In the app the row ends with the message's timestamp under the trigger, so a trigger near the
+  // bottom of the viewport has its row's bottom, and the row below, out of view.
+  const list = mountBottomTriggerHold("row-3");
+  list.render(bottomTriggerRows);
+  await wait(150);
+  await list.scrollTo(40);
+  const rowBottom = list.rowBottom("row-3");
+  expect(rowBottom).toBe(140);
+
+  act(() => growRow("row-3", 260));
+  await wait(150);
+  expect(list.rowBottom("row-3")).toBe(rowBottom);
+
+  act(() => growRow("row-3", 60));
+  await wait(150);
+  expect(list.rowBottom("row-3")).toBe(rowBottom);
+});
+
 it("keeps measured row sizes while the list is not laid out", async () => {
   // A hidden retained tab is display: none. Its rows still render, and a row measured there
   // reports 0 x 0. When the tab shows again, a row whose frame did not change gets no new layout

packages/app/src/agent-stream/native-disclosure-hold.test.ts

@@ -1,5 +1,10 @@
 import { describe, expect, it } from "vitest";
-import { createNativeDisclosureHold } from "./native-disclosure-hold";
+import {
+  createNativeDisclosureHold,
+  disclosureAnchor,
+  createEndHoldDistance,
+  type DisclosureAnchor,
+} from "./native-disclosure-hold";
 
 function createFrames() {
   let nextId = 1;
@@ -39,22 +44,22 @@ describe("native disclosure hold", () => {
   it("holds the row until two frames pass without an item size change", () => {
     const { frames, hold, releases } = createHarness();
 
-    hold.hold("turn-1");
+    hold.hold({ kind: "row", rowId: "turn-1" });
     frames.runFrame();
 
-    expect(hold.heldRowId()).toBe("turn-1");
+    expect(hold.anchorRowId()).toBe("turn-1");
     expect(releases).toEqual([]);
 
     frames.runFrame();
 
-    expect(hold.heldRowId()).toBeNull();
+    expect(hold.anchorRowId()).toBeNull();
     expect(releases).toEqual(["released"]);
   });
 
   it("restarts the quiet frames on every item size change", () => {
     const { frames, hold, releases } = createHarness();
 
-    hold.hold("turn-1");
+    hold.hold({ kind: "row", rowId: "turn-1" });
     frames.runFrame();
     hold.noteItemSizeChanged();
     frames.runFrame();
@@ -79,11 +84,11 @@ describe("native disclosure hold", () => {
   it("moves the hold to the latest toggled row and releases once", () => {
     const { frames, hold, releases } = createHarness();
 
-    hold.hold("turn-1");
+    hold.hold({ kind: "row", rowId: "turn-1" });
     frames.runFrame();
-    hold.hold("turn-2");
+    hold.hold({ kind: "row", rowId: "turn-2" });
 
-    expect(hold.heldRowId()).toBe("turn-2");
+    expect(hold.anchorRowId()).toBe("turn-2");
 
     frames.runFrame();
     frames.runFrame();
@@ -94,12 +99,137 @@ describe("native disclosure hold", () => {
   it("drops the hold without releasing when cancelled", () => {
     const { frames, hold, releases } = createHarness();
 
-    hold.hold("turn-1");
+    hold.hold({ kind: "row", rowId: "turn-1" });
     hold.cancel();
     frames.runFrame();
     frames.runFrame();
 
-    expect(hold.heldRowId()).toBeNull();
+    expect(hold.anchorRowId()).toBeNull();
     expect(releases).toEqual([]);
   });
+
+  it("holds the content end until two quiet frames pass, with no row to anchor on", () => {
+    const { frames, hold, releases } = createHarness();
+
+    hold.hold({ kind: "end" });
+    frames.runFrame();
+    hold.noteItemSizeChanged();
+    frames.runFrame();
+
+    expect(hold.anchorRowId()).toBeNull();
+    expect(releases).toEqual([]);
+
+    frames.runFrame();
+
+    expect(releases).toEqual(["released"]);
+  });
+});
+
+describe("next-frame disclosure hold", () => {
+  it("anchors the row a reply landed as while the toggle waited for its frame", () => {
+    const { frames, hold } = createHarness();
+    let anchor: DisclosureAnchor = { kind: "end" };
+    const applied: DisclosureAnchor[] = [];
+
+    hold.holdNextFrame(
+      () => anchor,
+      (resolved) => applied.push(resolved),
+    );
+    expect(applied).toEqual([]);
+
+    anchor = { kind: "row", rowId: "turn-2" };
+    frames.runFrame();
+
+    expect(hold.anchorRowId()).toBe("turn-2");
+    expect(applied).toEqual([{ kind: "row", rowId: "turn-2" }]);
+  });
+
+  it("still lands the toggle when the hold is cancelled before its frame", () => {
+    const { frames, hold, releases } = createHarness();
+    const applied: DisclosureAnchor[] = [];
+
+    hold.holdNextFrame(
+      () => ({ kind: "end" }),
+      (resolved) => applied.push(resolved),
+    );
+    hold.cancel();
+    frames.runFrame();
+    frames.runFrame();
+
+    expect(applied).toEqual([{ kind: "end" }]);
+    expect(releases).toEqual([]);
+  });
+
+  it("lands a waiting toggle before the next hold starts", () => {
+    const { frames, hold } = createHarness();
+    const applied: DisclosureAnchor[] = [];
+
+    hold.holdNextFrame(
+      () => ({ kind: "end" }),
+      (resolved) => applied.push(resolved),
+    );
+    hold.hold({ kind: "row", rowId: "turn-1" });
+
+    expect(applied).toEqual([{ kind: "end" }]);
+    frames.runFrame();
+    expect(applied).toEqual([{ kind: "end" }]);
+    expect(hold.anchorRowId()).toBe("turn-1");
+  });
+});
+
+describe("disclosure anchor", () => {
+  const rows = [{ id: "message-1" }, { id: "turn-1" }, { id: "message-2" }];
+
+  it("keeps the top of a row whose trigger sits on its top edge", () => {
+    expect(disclosureAnchor(rows, "turn-1", "top")).toEqual({ kind: "row", rowId: "turn-1" });
+  });
+
+  it("keeps the top of the row below a row whose trigger sits on its bottom edge", () => {
+    expect(disclosureAnchor(rows, "message-1", "bottom")).toEqual({
+      kind: "row",
+      rowId: "turn-1",
+    });
+  });
+
+  // A prompt the agent has not answered yet is the last row; the running indicator is the footer.
+  it("keeps the content end for a bottom-edge trigger on the last row", () => {
+    expect(disclosureAnchor(rows, "message-2", "bottom")).toEqual({ kind: "end" });
+  });
+
+  // The strategy re-derives this each render, so a reply lets go of the end in the commit it lands.
+  it("stops keeping the content end once a reply lands below the row", () => {
+    expect(disclosureAnchor([...rows, { id: "turn-2" }], "message-2", "bottom")).toEqual({
+      kind: "row",
+      rowId: "turn-2",
+    });
+  });
+});
+
+describe("end hold distance", () => {
+  const viewportHeight = 600;
+
+  it("keeps the distance from the end the toggle started at", () => {
+    const distance = createEndHoldDistance(20);
+
+    distance.noteContentSize({ contentHeight: 2000, viewportHeight });
+
+    expect(distance.current()).toBe(20);
+  });
+
+  it("shrinks to the scroll room a collapse leaves", () => {
+    const distance = createEndHoldDistance(20);
+
+    distance.noteContentSize({ contentHeight: 610, viewportHeight });
+
+    expect(distance.current()).toBe(10);
+  });
+
+  it("stays at the end when output grows the content after a collapse reached the end", () => {
+    const distance = createEndHoldDistance(20);
+
+    distance.noteContentSize({ contentHeight: 600, viewportHeight });
+    distance.noteContentSize({ contentHeight: 700, viewportHeight });
+
+    expect(distance.current()).toBe(0);
+  });
 });

packages/app/src/agent-stream/native-disclosure-hold.ts

@@ -1,13 +1,30 @@
+import type { DisclosureTriggerEdge } from "./strategy";
+
+/** What a disclosure hold keeps in place: the top of a row, or the end of the content. */
+export type DisclosureAnchor = { kind: "row"; rowId: string } | { kind: "end" };
+
 /**
- * The native viewport's disclosure hold. While a row is held, it is the only row the list may
- * anchor on. The hold ends once item sizes stop changing for `quietFrames` frames: the list
- * measures the toggled row and any rows it inserted over one or more frames, and end following
- * must not resume between the change and the list's last layout correction.
+ * The native viewport's disclosure hold. While it holds a row anchor, that row is the only row the
+ * list may anchor on (`disclosureAnchor`). The hold ends once item sizes stop changing for
+ * `quietFrames` frames: the list measures the toggled row and any rows it inserted over one or more
+ * frames, and end following must not resume between the change and the list's last layout
+ * correction.
  */
 export interface NativeDisclosureHold {
-  heldRowId(): string | null;
-  hold(rowId: string): void;
+  /** The row the list may anchor on; null outside a hold and while the hold keeps the end. */
+  anchorRowId(): string | null;
+  hold(anchor: DisclosureAnchor): void;
+  /**
+   * Holds on the next frame and runs `apply` there, for a toggle that has to land after a commit
+   * already on its way. `resolveAnchor` runs in that frame, because rows may land while it waits.
+   * A later hold or a cancel lands a waiting toggle right away.
+   */
+  holdNextFrame(
+    resolveAnchor: () => DisclosureAnchor,
+    apply: (anchor: DisclosureAnchor) => void,
+  ): void;
   noteItemSizeChanged(): void;
+  /** Drops the hold without releasing it. */
   cancel(): void;
 }
 
@@ -17,9 +34,10 @@ export function createNativeDisclosureHold(input: {
   cancelFrame: (id: number) => void;
   onRelease: () => void;
 }): NativeDisclosureHold {
-  let heldRowId: string | null = null;
+  let anchor: DisclosureAnchor | null = null;
   let frameId: number | null = null;
   let remainingFrames = 0;
+  let waitingToggle: { frameId: number; land: () => void } | null = null;
 
   const stopCountdown = () => {
     if (frameId !== null) {
@@ -35,7 +53,7 @@ export function createNativeDisclosureHold(input: {
       return;
     }
     frameId = null;
-    heldRowId = null;
+    anchor = null;
     input.onRelease();
   };
 
@@ -45,20 +63,80 @@ export function createNativeDisclosureHold(input: {
     frameId = input.requestFrame(tick);
   };
 
+  const landWaitingToggle = () => {
+    if (!waitingToggle) return;
+    const { frameId: waitingFrameId, land } = waitingToggle;
+    waitingToggle = null;
+    input.cancelFrame(waitingFrameId);
+    land();
+  };
+
+  const startHold = (nextAnchor: DisclosureAnchor) => {
+    anchor = nextAnchor;
+    restartCountdown();
+  };
+
   return {
-    heldRowId: () => heldRowId,
-    hold(rowId) {
-      heldRowId = rowId;
-      restartCountdown();
+    anchorRowId: () => (anchor?.kind === "row" ? anchor.rowId : null),
+    hold(nextAnchor) {
+      landWaitingToggle();
+      startHold(nextAnchor);
+    },
+    holdNextFrame(resolveAnchor, apply) {
+      landWaitingToggle();
+      const waitingFrameId = input.requestFrame(() => {
+        waitingToggle = null;
+        const nextAnchor = resolveAnchor();
+        startHold(nextAnchor);
+        apply(nextAnchor);
+      });
+      waitingToggle = { frameId: waitingFrameId, land: () => apply(resolveAnchor()) };
     },
     noteItemSizeChanged() {
-      if (heldRowId !== null) {
+      if (anchor !== null) {
         restartCountdown();
       }
     },
     cancel() {
+      landWaitingToggle();
       stopCountdown();
-      heldRowId = null;
+      anchor = null;
     },
   };
 }
+
+/**
+ * What the list keeps in place while a disclosure toggles `rowId`. LegendList restores row tops, so
+ * a trigger on the bottom edge of its row anchors the row below: that row's top is the toggled
+ * row's bottom. The last row's bottom is the end of the content.
+ */
+export function disclosureAnchor(
+  rows: readonly { id: string }[],
+  rowId: string,
+  triggerEdge: DisclosureTriggerEdge,
+): DisclosureAnchor {
+  if (triggerEdge === "top") return { kind: "row", rowId };
+  const index = rows.findIndex((row) => row.id === rowId);
+  const below = rows[index + 1];
+  return below ? { kind: "row", rowId: below.id } : { kind: "end" };
+}
+
+/**
+ * The list's distance from the end during an end hold. The end pin keeps the distance through every
+ * content size change, and the scroll view clamps it to the scroll room the content has at that
+ * moment. A clamp is permanent: content that grows afterwards does not bring the distance back.
+ */
+export function createEndHoldDistance(distanceAtToggle: number): EndHoldDistance {
+  let distance = distanceAtToggle;
+  return {
+    noteContentSize({ contentHeight, viewportHeight }) {
+      distance = Math.min(distance, Math.max(0, contentHeight - viewportHeight));
+    },
+    current: () => distance,
+  };
+}
+
+export interface EndHoldDistance {
+  noteContentSize(input: { contentHeight: number; viewportHeight: number }): void;
+  current(): number;
+}

packages/app/src/agent-stream/reading-anchor.test.ts

@@ -7,7 +7,7 @@ describe("reading anchor", () => {
     anchor.reconcile(0, [{ id: "tool", top: 48, height: 200 }]);
     const replacements = new Map([["tool", "prompt"]]);
     expect(anchor.getRowId(replacements)).toBe("prompt");
-    expect(anchor.project(0, { id: "prompt", top: 1048 }, replacements)).toBe(1000);
+    expect(anchor.project(0, { id: "prompt", top: 1048, height: 100 }, replacements)).toBe(1000);
     // Projection does not consume the correction; an unmounted host keeps the pin.
     expect(anchor.reconcile(0, [], false, replacements)).toBe(0);
     expect(anchor.getRowId()).toBe("tool");
@@ -59,13 +59,32 @@ describe("reading anchor", () => {
   it("keeps a pinned row in place instead of the row under the reading line", () => {
     const anchor = createReadingAnchor();
     anchor.reconcile(600, rows);
-    anchor.pin(rows[2]!);
+    anchor.pin(rows[2]!, "top", 600);
     // The row being read grows; only the pinned row below it holds its position.
     expect(
       anchor.reconcile(600, [rows[0]!, { ...rows[1]!, height: 800 }, { ...rows[2]!, top: 1300 }]),
     ).toBe(900);
   });
 
+  it("keeps the bottom of a row pinned by a trigger below its content", () => {
+    const anchor = createReadingAnchor();
+    anchor.reconcile(600, rows);
+    anchor.pin(rows[2]!, "bottom", 600);
+    // The pinned row grows downward; its bottom, where the trigger sits, holds its position.
+    const grown = { ...rows[2]!, height: rows[2]!.height + 300 };
+    expect(anchor.reconcile(600, [rows[0]!, rows[1]!, grown])).toBe(900);
+  });
+
+  it("measures a pinned row's correction from the offset at the toggle", () => {
+    const anchor = createReadingAnchor();
+    anchor.reconcile(900, rows);
+    anchor.pin(rows[2]!, "bottom", 900);
+    // The row shrinks and the browser clamps the offset to the shorter content before the
+    // reconcile reads it.
+    const shrunk = { ...rows[2]!, height: rows[2]!.height - 300 };
+    expect(anchor.reconcile(700, [rows[0]!, rows[1]!, shrunk])).toBe(600);
+  });
+
   it("leaves expansion of the row being read in place", () => {
     const anchor = createReadingAnchor();
     anchor.reconcile(600, rows);
@@ -109,8 +128,8 @@ describe("reading anchor", () => {
   it("projects a prepend range without consuming the correction before layout commits", () => {
     const anchor = createReadingAnchor();
     anchor.reconcile(600, rows);
-    expect(anchor.project(440, { id: "reading", top: 2500 })).toBe(2440);
-    expect(anchor.project(440, { id: "reading", top: 2500 })).toBe(2440);
+    expect(anchor.project(440, { id: "reading", top: 2500, height: 100 })).toBe(2440);
+    expect(anchor.project(440, { id: "reading", top: 2500, height: 100 })).toBe(2440);
     expect(
       anchor.reconcile(
         440,

packages/app/src/agent-stream/reading-anchor.ts

@@ -1,9 +1,14 @@
+import type { DisclosureTriggerEdge } from "./strategy";
+
 interface RowGeometry {
   id: string;
   top: number;
   height: number;
 }
 
+const edgeOf = (row: RowGeometry, edge: DisclosureTriggerEdge) =>
+  edge === "top" ? row.top : row.top + row.height;
+
 // A row must clear the reading line before the next row takes ownership.
 const READING_POSITION_OFFSET_PX = 8;
 export interface ReadingRowReplacements {
@@ -14,7 +19,16 @@ const NO_REPLACEMENTS: ReadingRowReplacements = new Map();
 // Content coordinates survive user scrolling. Layout commits replace the geometry;
 // scroll events reuse it, without another DOM measurement pass.
 export function createReadingAnchor() {
-  let anchor: { id: string; top: number } | null = null;
+  // The row being read keeps its top. A disclosure pins the edge of its row the trigger sits on,
+  // with the offset at the toggle: a toggle that shrinks the content lets the browser clamp the
+  // offset before the next reconcile reads it.
+  let anchor: {
+    id: string;
+    edge: DisclosureTriggerEdge;
+    position: number;
+    pinnedScrollTop?: number;
+  } | null = null;
+  const anchorTop = (row: RowGeometry) => ({ id: row.id, edge: "top" as const, position: row.top });
   let geometry: readonly RowGeometry[] = [];
   const readingRow = (scrollTop: number) =>
     geometry.find((row) => row.top + row.height > scrollTop + READING_POSITION_OFFSET_PX);
@@ -22,9 +36,12 @@ export function createReadingAnchor() {
     anchor ? (replacements.get(anchor.id) ?? anchor.id) : null;
   const project = (
     scrollTop: number,
-    row: Pick<RowGeometry, "id" | "top"> | undefined,
+    row: RowGeometry | undefined,
     replacements = NO_REPLACEMENTS,
-  ) => scrollTop + (anchor && row && row.id === getRowId(replacements) ? row.top - anchor.top : 0);
+  ) =>
+    anchor && row && row.id === getRowId(replacements)
+      ? (anchor.pinnedScrollTop ?? scrollTop) + edgeOf(row, anchor.edge) - anchor.position
+      : scrollTop;
   return {
     getRowId,
     getReadingRowId: (scrollTop: number) =>
@@ -33,14 +50,14 @@ export function createReadingAnchor() {
     reset() {
       anchor = null;
     },
-    // A disclosure names the row that owns its trigger. That row keeps its top
-    // through the toggle, even when another row sits under the reading line.
-    pin(row: Pick<RowGeometry, "id" | "top">) {
-      anchor = { id: row.id, top: row.top };
+    // A disclosure names the row that owns its trigger and the edge the trigger sits on. That
+    // edge stays in place through the toggle, even when another row sits under the reading line.
+    pin(row: RowGeometry, edge: DisclosureTriggerEdge, scrollTop: number) {
+      anchor = { id: row.id, edge, position: edgeOf(row, edge), pinnedScrollTop: scrollTop };
     },
     scroll(scrollTop: number) {
       const next = readingRow(scrollTop);
-      anchor = next ? { id: next.id, top: next.top } : null;
+      anchor = next ? anchorTop(next) : null;
     },
     reconcile(
       scrollTop: number,
@@ -57,7 +74,7 @@ export function createReadingAnchor() {
       // A prepend can expose the bottom of an estimated row above the reader.
       // Do not transfer ownership to it until the user moves the reading position.
       const next = previous && !userScrolled ? previous : readingRow(correctedTop);
-      anchor = next ? { id: next.id, top: next.top } : null;
+      anchor = next ? anchorTop(next) : null;
       return correctedTop;
     },
   };

packages/app/src/agent-stream/strategy-native.tsx

@@ -35,10 +35,20 @@ import { shouldAnimateNativeScrollToBottom } from "./native-scroll-animation";
 import { useRevisedHistoryRows } from "./history-row-revision";
 import { useBottomAnchorController } from "./bottom-anchor-controller";
 import { useScrollKeyboardDismiss } from "./scroll-keyboard-dismiss/use-scroll-keyboard-dismiss";
-import { createNativeDisclosureHold } from "./native-disclosure-hold";
+import {
+  createNativeDisclosureHold,
+  disclosureAnchor,
+  createEndHoldDistance,
+  type EndHoldDistance,
+} from "./native-disclosure-hold";
 import { EndPinnedScrollView } from "./native-end-pin";
 import { createNewRowEstimates } from "./new-row-estimates";
-import type { StreamRenderInput, StreamStrategy, StreamViewportHandle } from "./strategy";
+import type {
+  DisclosureTriggerEdge,
+  StreamRenderInput,
+  StreamStrategy,
+  StreamViewportHandle,
+} from "./strategy";
 import { createStreamStrategy, isNearBottomForStreamRenderStrategy } from "./strategy";
 import {
   isHistoryStartLoadingOperation,
@@ -306,8 +316,14 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat
   const isAtEnd = useStableEvent(
     () => distanceFromEnd(streamViewportMetricsRef.current) <= AT_END_OFFSET_PX,
   );
+  // A hold on the content end mounts the end pin while detached (`holdRow`). The release reads the
+  // tracked distance instead of the offset: the content size reaches JS before the pin's scroll does.
+  const [endHold, setEndHold] = useState<{ rowId: string; distance: EndHoldDistance } | null>(null);
   const releaseDisclosure = useStableEvent(() => {
-    bottomAnchorController.releaseDisclosure({ isAtEnd: isAtEnd() });
+    setEndHold(null);
+    bottomAnchorController.releaseDisclosure({
+      isAtEnd: endHold ? endHold.distance.current() <= AT_END_OFFSET_PX : isAtEnd(),
+    });
   });
   const [disclosureHold] = useState(() =>
     createNativeDisclosureHold({
@@ -317,16 +333,54 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat
       onRelease: releaseDisclosure,
     }),
   );
-  useEffect(() => () => disclosureHold.cancel(), [disclosureHold]);
-  // The hold detaches the controller in the same event as `apply`, so the toggle's commit already
-  // has end following off and LegendList anchoring on the held row. The row's measurement trigger
-  // makes that same commit carry the row's new height.
-  const holdRow = useStableEvent((rowId: string, apply: () => void) => {
-    disclosureHold.hold(rowId);
-    bottomAnchorController.holdForDisclosure();
-    apply();
-    rowLayoutSyncRegistry.get(rowId)?.();
+  const cancelDisclosureHold = useStableEvent(() => {
+    disclosureHold.cancel();
+    setEndHold(null);
   });
+  useEffect(() => () => cancelDisclosureHold(), [cancelDisclosureHold]);
+  // The hold detaches the controller in the same event as `apply`, so the toggle's commit already
+  // has end following off and LegendList anchoring on the held row, or on the row below it for a
+  // trigger on the row's bottom edge. The row's measurement trigger makes that same commit carry
+  // the row's new height. While following, a bottom trigger keeps following: the end pin holds the
+  // content end, and with it the bottom of the row, in the frame the new height mounts.
+  // The last row's bottom is the content end, which LegendList cannot anchor. The end pin can, from
+  // any offset, but only once it is mounted: a pin that mounts with the new height anchors nothing.
+  // So the toggle lands a frame after the pin's commit, whose mount runs before the frame's timers.
+  const latestDisclosureAnchor = useStableEvent(
+    (rowId: string, triggerEdge: DisclosureTriggerEdge) =>
+      disclosureAnchor(rows, rowId, triggerEdge),
+  );
+  const holdRow = useStableEvent(
+    (rowId: string, apply: () => void, triggerEdge: DisclosureTriggerEdge) => {
+      const applyAndMeasure = () => {
+        apply();
+        rowLayoutSyncRegistry.get(rowId)?.();
+      };
+      if (triggerEdge === "bottom" && bottomAnchorController.mode === "sticky-bottom") {
+        applyAndMeasure();
+        return;
+      }
+      const anchor = disclosureAnchor(rows, rowId, triggerEdge);
+      bottomAnchorController.holdForDisclosure();
+      if (anchor.kind === "row" || !isListPlacedRef.current) {
+        setEndHold(null);
+        disclosureHold.hold(anchor);
+        applyAndMeasure();
+        return;
+      }
+      setEndHold({
+        rowId,
+        distance: createEndHoldDistance(distanceFromEnd(streamViewportMetricsRef.current)),
+      });
+      disclosureHold.holdNextFrame(
+        () => latestDisclosureAnchor(rowId, triggerEdge),
+        (latestAnchor) => {
+          if (latestAnchor.kind === "row") setEndHold(null);
+          applyAndMeasure();
+        },
+      );
+    },
+  );
   const handleItemSizeChanged = useStableEvent(
     (info: { size: number; previous: number; itemKey: string; itemData: StreamItem }) => {
       newRowEstimates.noteMeasured(info.itemData, info.size);
@@ -334,8 +388,8 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat
     },
   );
   const shouldRestorePosition = useStableEvent((item: StreamItem) => {
-    const heldRowId = disclosureHold.heldRowId();
-    return heldRowId === null || item.id === heldRowId;
+    const anchorRowId = disclosureHold.anchorRowId();
+    return anchorRowId === null || item.id === anchorRowId;
   });
   // One owner keeps the scroll offset at any moment:
   // - Before the list has placed its rows for the agent: LegendList's initial scroll to the end. It
@@ -349,16 +403,23 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat
   //   scrolls to the end its JS sizes predict, which runs ahead of the native mount, so it lands
   //   short of the real end and the controller reads that as the user scrolling away.
   // - A disclosure hold: the held row. The hold detaches the controller, which removes the pin, and
-  //   LegendList keeps only the held row in place.
+  //   LegendList keeps only the held row in place, or the row below it when the trigger sits on the
+  //   held row's bottom edge. A bottom-edge hold while following leaves the end pin in charge, and
+  //   one on the last row mounts the pin for the hold: the row's bottom is the content end.
   // - A user scroll: the finger. The controller detaches for the gesture, so streamed lines grow below
   //   the rows the user is dragging, and resumes following if the gesture ends at the end.
   // - Detached: the visible rows. LegendList keeps the first visible row in place, so history loading
   //   above or rows growing below leave what the user is reading where it is.
   const isFollowingEnd = bottomAnchorController.mode === "sticky-bottom";
-  const pinsEnd = isFollowingEnd && isListPlaced;
+  // An end hold pins the end only while its row's bottom is the end. A reply landing below the row
+  // drops the pin in the commit that mounts the reply, so the reply grows below the held row.
+  const isEndHeld =
+    endHold !== null && disclosureAnchor(rows, endHold.rowId, "bottom").kind === "end";
+  const isEndPinned = isFollowingEnd || isEndHeld;
+  const pinsEnd = isEndPinned && isListPlaced;
   const maintainVisibleContentPosition = useMemo(
-    () => (isFollowingEnd ? false : { data: true, size: true, shouldRestorePosition }),
-    [isFollowingEnd, shouldRestorePosition],
+    () => (isEndPinned ? false : { data: true, size: true, shouldRestorePosition }),
+    [isEndPinned, shouldRestorePosition],
   );
   // A new identity re-renders LegendList's scroll component, which is how the pin follows the mode.
   const renderScrollComponent = useCallback(
@@ -505,7 +566,7 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat
     // The gesture takes the viewport from a disclosure hold. Size changes from a streaming row can
     // keep the hold's quiet frames from ever passing, and a held row that scrolls away would leave
     // no row LegendList may anchor a history page on. The gesture's end decides following.
-    disclosureHold.cancel();
+    cancelDisclosureHold();
     bottomAnchorController.beginUserScroll();
     historyStartPagination.rearm();
   });
@@ -581,6 +642,7 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat
       contentHeight: nextContentHeight,
       contentMeasuredForKey: isListPlacedRef.current ? "native-virtualized" : null,
     };
+    endHold?.distance.noteContentSize(streamViewportMetricsRef.current);
     bottomAnchorController.handleContentSizeChange();
     // Content growing below a detached viewport moves the end away without a scroll event.
     if (

packages/app/src/agent-stream/strategy-web.browser.test.tsx

@@ -3,7 +3,7 @@ import { createRoot, type Root } from "react-dom/client";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
 import { DEFAULT_CONTENT_MAX_WIDTH } from "@/styles/theme";
 import type { StreamItem } from "@/types/stream";
-import type { StreamRenderInput, StreamViewportHandle } from "./strategy";
+import type { DisclosureTriggerEdge, StreamRenderInput, StreamViewportHandle } from "./strategy";
 import { createWebStreamStrategy } from "./strategy-web";
 
 Object.assign(globalThis, { IS_REACT_ACT_ENVIRONMENT: true });
@@ -91,10 +91,15 @@ function mountTimeline(historyMounted: StreamItem[]) {
 
   return {
     render: (segments: Partial<StreamRenderInput["segments"]>) => act(() => render(segments)),
-    hold: (rowId: string, apply: () => void) =>
-      act(() => viewportRef.current!.holdRow(rowId, apply)),
+    hold: (rowId: string, apply: () => void, triggerEdge: DisclosureTriggerEdge = "top") =>
+      act(() => viewportRef.current!.holdRow(rowId, apply, triggerEdge)),
     rowTop: (id: string) =>
       rowElement(id).getBoundingClientRect().top - scrollContainer.getBoundingClientRect().top,
+    rowBottom: (id: string) =>
+      rowElement(id).getBoundingClientRect().bottom - scrollContainer.getBoundingClientRect().top,
+    scrollTo: (top: number) => {
+      scrollContainer.scrollTop = top;
+    },
     distanceFromEnd: () =>
       scrollContainer.scrollHeight - scrollContainer.clientHeight - scrollContainer.scrollTop,
     // Resize observers and the hold's settle run on animation frames.
@@ -139,6 +144,53 @@ describe("web disclosure hold", () => {
     expect(timeline.rowTop("message-5")).toBe(triggerTop);
   });
 
+  // A long message's show more sits under the text it opens, on the bottom edge of its row.
+  it("keeps the bottom of a bottom-trigger row through show more and show less while following output", async () => {
+    const history = [1, 2, 3, 4, 5, 6].map((index) => row(index));
+    const timeline = mountTimeline(history);
+    await timeline.settle();
+    const triggerBottom = timeline.rowBottom("message-5");
+
+    timeline.hold(
+      "message-5",
+      () => timeline.render({ historyMounted: [...history.slice(0, 4), row(5, 240), history[5]!] }),
+      "bottom",
+    );
+    await timeline.settle();
+    expect(timeline.rowBottom("message-5")).toBe(triggerBottom);
+
+    timeline.hold("message-5", () => timeline.render({ historyMounted: history }), "bottom");
+    await timeline.settle();
+    expect(timeline.rowBottom("message-5")).toBe(triggerBottom);
+  });
+
+  it("keeps the bottom of a bottom-trigger row through show more and show less when scrolled up", async () => {
+    const history = [1, 2, 3, 4, 5, 6].map((index) => row(index));
+    const timeline = mountTimeline(history);
+    await timeline.settle();
+    timeline.scrollTo(60);
+    await timeline.settle();
+    // message-2 is cut off at the top; message-3 is fully in view.
+    for (const index of [2, 3]) {
+      const triggerBottom = timeline.rowBottom(`message-${index}`);
+      const grown = history.map((item, position) =>
+        position === index - 1 ? row(index, 240) : item,
+      );
+
+      timeline.hold(`message-${index}`, () => timeline.render({ historyMounted: grown }), "bottom");
+      await timeline.settle();
+      expect(timeline.rowBottom(`message-${index}`)).toBe(triggerBottom);
+
+      timeline.hold(
+        `message-${index}`,
+        () => timeline.render({ historyMounted: history }),
+        "bottom",
+      );
+      await timeline.settle();
+      expect(timeline.rowBottom(`message-${index}`)).toBe(triggerBottom);
+    }
+  });
+
   it("resumes following when collapsing an anchored row lands back at the end", async () => {
     const history = [1, 2, 3, 4, 5, 6].map((index) => row(index));
     const timeline = mountTimeline(history);

packages/app/src/agent-stream/strategy-web.tsx

@@ -418,6 +418,7 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool
               {
                 id: String(measurement.key),
                 top: measurement.start,
+                height: measurement.size,
               },
               props.readingRowReplacements,
             )
@@ -1172,11 +1173,12 @@ function WebStreamViewport(props: StreamRenderInput & { isMobileBreakpoint: bool
         }
         scheduleStickToBottom();
       },
-      holdRow: (rowId, apply) => {
+      holdRow: (rowId, apply, triggerEdge) => {
         cancelPendingStickToBottom();
         setFollowOutput(false);
         const row = measureRows()?.find((candidate) => candidate.id === rowId);
-        if (row) readingAnchor.pin(row);
+        const scrollTop = scrollContainerRef.current?.scrollTop;
+        if (row && scrollTop !== undefined) readingAnchor.pin(row, triggerEdge, scrollTop);
         disclosurePendingRef.current = true;
         apply();
       },

packages/app/src/agent-stream/strategy.ts

@@ -46,16 +46,23 @@ export interface ScrollToMessageOccurrence {
   targetTop(): number | null;
 }
 
+/**
+ * The edge of its row a disclosure's trigger sits on. A header trigger sits on the top: the content
+ * it opens grows below it. A trigger under the content it opens, such as a long message's show
+ * more, sits on the bottom.
+ */
+export type DisclosureTriggerEdge = "top" | "bottom";
+
 export interface StreamViewportHandle {
   /** Give keyboard navigation back to the scrolling surface on web. */
   focus?: () => void;
   scrollToBottom: (reason?: BottomAnchorLocalRequest["reason"]) => void;
   prepareForViewportChange: () => void;
   /**
-   * Run a disclosure's state change so the row that owns its trigger keeps its visual top.
-   * Following output stops; it resumes if the change leaves the list at its end.
+   * Run a disclosure's state change so the row that owns its trigger keeps the edge the trigger
+   * sits on in place. Following output stops; it resumes if the change leaves the list at its end.
    */
-  holdRow: (rowId: string, apply: () => void) => void;
+  holdRow: (rowId: string, apply: () => void, triggerEdge: DisclosureTriggerEdge) => void;
   scrollToMessage?: (messageId: string, occurrence?: ScrollToMessageOccurrence) => void;
 }
 

packages/app/src/components/plan-card.tsx

@@ -1,3 +1,4 @@
+import { useDisclosureHold } from "@/agent-stream/disclosure";
 import { MarkdownRenderer } from "@/components/markdown/renderer";
 import { isWeb } from "@/constants/platform";
 import type { Theme } from "@/styles/theme";
@@ -50,7 +51,8 @@ function PlanCardContent({
     () => (isWeb ? ({ "aria-expanded": expanded } as const) : null),
     [expanded],
   );
-  const toggleExpanded = useCallback(() => setExpanded((value) => !value), []);
+  const hold = useDisclosureHold();
+  const toggleExpanded = useCallback(() => hold(() => setExpanded((value) => !value)), [hold]);
   const containerStyle = useMemo(
     () => [styles.container, disableOuterSpacing && styles.containerCompact],
     [disableOuterSpacing],

packages/app/src/components/user-message/index.tsx

@@ -9,8 +9,11 @@ import { useProviderIcon } from "@/components/provider-icons";
 import { useSessionStore } from "@/stores/session-store";
 import { navigateToAgent } from "@/utils/navigate-to-agent";
 import { isNative } from "@/constants/platform";
+import { useDisclosureHold } from "@/agent-stream/disclosure";
 
 const PREVIEW_HEIGHT = 240;
+// Show more and show less sit under the text they open and close.
+const TRIGGER_BELOW_CONTENT = { triggerEdge: "bottom" } as const;
 // Native View still clips with "scroll", but Yoga measures its children without
 // the height cap. "hidden" caps the measured body too, hiding the expand control.
 const previewStyle: ViewStyle = {
@@ -26,7 +29,11 @@ export function MessagePreview({ children }: { children: ReactNode }) {
     (event: LayoutChangeEvent) => setOverflowing(event.nativeEvent.layout.height > PREVIEW_HEIGHT),
     [],
   );
-  const toggle = useCallback(() => setExpanded((value) => !value), []);
+  const hold = useDisclosureHold();
+  const toggle = useCallback(
+    () => hold(() => setExpanded((value) => !value), TRIGGER_BELOW_CONTENT),
+    [hold],
+  );
   const accessibilityState = useMemo(() => ({ expanded }), [expanded]);
   return (
     <>

packages/client/src/daemon-client-transport-types.ts

@@ -4,7 +4,7 @@ export interface DaemonTransport {
   maxFrameBytes?: number;
   send: (data: string | Uint8Array | ArrayBuffer) => void | Promise<void>;
   close: (code?: number, reason?: string) => void;
-  onMessage: (handler: (data: unknown, isBinary: boolean) => void) => () => void;
+  onMessage: (handler: (data: unknown, isBinary: boolean) => void | Promise<void>) => () => void;
   onOpen: (handler: () => void) => () => void;
   onClose: (handler: (event?: unknown) => void) => () => void;
   onError: (handler: (event?: unknown) => void) => () => void;

packages/client/src/daemon-client.ts

@@ -1923,18 +1923,34 @@ export class DaemonClient {
       });
     }
   }
-  async openHostLink(channelId: string): Promise<string> {
+  openHostLink(channelId: string): Promise<string>;
+  openHostLink(
+    channelId: string,
+    options: { flowControl: true },
+  ): Promise<{ routeId: string; flowControl: boolean }>;
+  async openHostLink(
+    channelId: string,
+    options?: { flowControl: true },
+  ): Promise<string | { routeId: string; flowControl: boolean }> {
     this.requireConnectedHosts();
     if (this.getLastServerInfoMessage()?.features?.hostMultiplexing !== true)
       throw Object.assign(new Error("Update the destination daemon"), {
         code: "unsupported_capability",
       });
-    return (
-      await this.sendCorrelatedSessionRequest({
-        message: { type: "hosts.link.open.request", channelId },
-        responseType: "hosts.link.open.response",
-      })
-    ).routeId;
+    const response = await this.sendCorrelatedSessionRequest({
+      message: {
+        type: "hosts.link.open.request",
+        channelId,
+        ...(options?.flowControl ? { flowControl: true } : {}),
+      },
+      responseType: "hosts.link.open.response",
+    });
+    return options?.flowControl
+      ? { routeId: response.routeId, flowControl: response.flowControl === true }
+      : response.routeId;
+  }
+  acknowledgeHostLink(routeId: string, sequence: number): Promise<void> {
+    return this.sendSessionMessageStrict({ type: "hosts.link.ack.request", routeId, sequence });
   }
   sendHostFrame(routeId: string, data: string | Uint8Array): Promise<void> {
     if (typeof data === "string") {

packages/client/src/hosts/index.ts

@@ -26,11 +26,13 @@ export function createHostChannelTransport(
   return () => {
     const opens = new Set<() => void>();
     const closes = new Set<(event?: unknown) => void>();
-    const messages = new Set<(data: unknown, binary: boolean) => void>();
+    const messages = new Set<(data: unknown, binary: boolean) => void | Promise<void>>();
     const errors = new Set<(error: unknown) => void>();
     let routeId: string | null = null;
     let closed = false;
     const cleanups: (() => void)[] = [];
+    let flowControl = false;
+    let receivedFrames = 0;
     const close = (error?: unknown) => {
       if (closed) return;
       closed = true;
@@ -45,10 +47,28 @@ export function createHostChannelTransport(
       if (error) for (const handler of errors) handler(error);
       for (const handler of closes) handler(error);
     };
+    const deliver = (data: unknown, binary: boolean) => {
+      if (closed) return;
+      const sequence = ++receivedFrames;
+      try {
+        const deliveries = [...messages].map(async (handler) => {
+          await handler(data, binary);
+        });
+        void Promise.all(deliveries)
+          .then(async () => {
+            if (flowControl && !closed && routeId)
+              await origin.acknowledgeHostLink(routeId, sequence);
+            return undefined;
+          })
+          .catch(close);
+      } catch (error) {
+        close(error);
+      }
+    };
     cleanups.push(
       origin.subscribeRawMessages((message) => {
         if (message.type === "hosts.route.frame" && message.payload.subscriptionId === routeId) {
-          for (const handler of messages) handler(message.payload.data, false);
+          deliver(message.payload.data, false);
         }
         if (message.type === "hosts.route.closed" && message.payload.subscriptionId === routeId)
           close(
@@ -61,7 +81,7 @@ export function createHostChannelTransport(
     );
     cleanups.push(
       origin.subscribeHostFrames((frame) => {
-        if (frame.routeId === routeId) for (const handler of messages) handler(frame.payload, true);
+        if (frame.routeId === routeId) deliver(frame.payload, true);
       }),
     );
     cleanups.push(
@@ -71,7 +91,15 @@ export function createHostChannelTransport(
     );
     let opening: Promise<string>;
     if (first) opening = Promise.resolve(first);
-    else opening = link ? origin.openHostLink(scopeId) : origin.openHostRoute(serverId, scopeId);
+    else if (link) {
+      opening = origin.openHostLink(scopeId, { flowControl: true }).then((opened) => {
+        // COMPAT(hostLinkFlowControl): added after v0.15.3 (#9109); remove after
+        // 2027-05-08 once the daemon floor supports forwarding acknowledgments.
+        // Older destinations omit flowControl and receive no acknowledgment messages.
+        flowControl = opened.flowControl;
+        return opened.routeId;
+      });
+    } else opening = origin.openHostRoute(serverId, scopeId);
     first = null;
     void opening
       .then((id) => {

packages/protocol/src/chunked-transport.test.ts

@@ -390,3 +390,65 @@ it("unpaced transports retain ordinary outbound admission and accept inbound chu
   expect(delivered).toEqual(["A"]);
   await transport.send("small");
 });
+
+it("fills an ordered writer's window while keeping chunk sequences contiguous", async () => {
+  const delivered: (string | Uint8Array)[] = [];
+  const receiver = new ChunkedTransport({
+    frameBytes: 8,
+    send() {},
+    deliver: (frame) => {
+      delivered.push(frame);
+    },
+  });
+  receiver.enabled = true;
+  const drain = Promise.withResolvers<void>();
+  const frames: (string | Uint8Array)[] = [];
+  const sender = new ChunkedTransport({
+    frameBytes: 8,
+    supportsConcurrentWrites: true,
+    send: (frame) => {
+      frames.push(frame);
+      receiver.receive(frame);
+      return drain.promise;
+    },
+    deliver() {},
+  });
+  sender.enabled = true;
+  const first = sender.send("one");
+  const second = sender.send("two");
+  expect(delivered).toEqual(["one", "two"]);
+  expect(sender.queuedByteLength).toBe(0);
+  const large = sender.send("large message");
+  const after = sender.send("after");
+  expect(frames).toHaveLength(3);
+  expect(sender.queuedByteLength).toBe(5);
+  drain.resolve();
+  await Promise.all([first, second, large, after]);
+  expect(delivered).toEqual(["one", "two", "large message", "after"]);
+  expect(sender.queuedByteLength).toBe(0);
+});
+
+it("accounts for independently draining writes and a rejected admission", async () => {
+  const firstDrain = Promise.withResolvers<void>();
+  const secondDrain = Promise.withResolvers<void>();
+  const sender = new ChunkedTransport({
+    frameBytes: 8,
+    supportsConcurrentWrites: true,
+    send: (frame) => {
+      if (frame === "reject") throw new Error("capacity");
+      return frame === "one" ? firstDrain.promise : secondDrain.promise;
+    },
+    deliver() {},
+  });
+  const first = sender.send("one");
+  const second = sender.send("two");
+  expect(() => sender.send("reject")).toThrow("capacity");
+  expect(sender.queuedByteLength).toBe(0);
+  secondDrain.resolve();
+  await second;
+  expect(sender.queuedByteLength).toBe(0);
+  firstDrain.resolve();
+  await first;
+  expect(sender.queuedByteLength).toBe(0);
+  await sender.send("usable");
+});

packages/protocol/src/chunked-transport.ts

@@ -23,6 +23,8 @@ export class ChunkedTransport {
     private readonly io: {
       frameBytes: number;
       supportsDrain?: boolean;
+      /** Writes are admitted synchronously in order; promises track only their drain. */
+      supportsConcurrentWrites?: boolean;
       send: (frame: Frame, waitForDrain: boolean) => void | Promise<void>;
       deliver: (frame: Frame) => void;
       onActivity?: () => void;
@@ -46,9 +48,11 @@ export class ChunkedTransport {
     const chunk =
       this.enabled && this.io.supportsDrain !== false && byteLength > this.io.frameBytes;
     if (this.closed) throw new Error("Transport closed");
+    let started = false;
     const write = () => {
       if (this.closed) throw new Error("Transport closed");
-      this.activeBytes = byteLength;
+      started = true;
+      this.activeBytes += byteLength;
       // A synchronous rejection did not admit a frame. Only accepted writes can
       // fail the connection; queued admission failures still belong to their sender.
       const result = chunk
@@ -76,27 +80,31 @@ export class ChunkedTransport {
       result = this.pending ? this.pending.then(write) : write();
     } catch (error) {
       this.queuedBytes -= byteLength;
-      this.activeBytes = 0;
+      if (started) this.activeBytes -= byteLength;
       throw error;
     }
     if (!result) {
       this.queuedBytes -= byteLength;
-      this.activeBytes = 0;
+      if (started) this.activeBytes -= byteLength;
       return Promise.resolve();
     }
     const completion = Promise.resolve(result).finally(() => {
       this.queuedBytes -= byteLength;
-      this.activeBytes = 0;
-    });
-    const tail = completion.then(
-      () => undefined,
-      () => undefined,
-    );
-    this.pending = tail;
-    void tail.then(() => {
-      if (this.pending === tail) this.pending = undefined;
-      return undefined;
+      if (started) this.activeBytes -= byteLength;
     });
+    // Ordinary writes already entered an ordered writer and can use its bounded
+    // window. Chunk sequences remain a barrier so their frames never interleave.
+    if (chunk || !this.io.supportsConcurrentWrites) {
+      const tail = completion.then(
+        () => undefined,
+        () => undefined,
+      );
+      this.pending = tail;
+      void tail.then(() => {
+        if (this.pending === tail) this.pending = undefined;
+        return undefined;
+      });
+    }
     return completion;
   }
 

packages/protocol/src/hosts.ts

@@ -116,8 +116,19 @@ export const HostsLinkOpenRequestSchema = z.object({
   type: z.literal("hosts.link.open.request"),
   requestId: z.string(),
   channelId: z.string().min(1).max(256),
+  flowControl: z.boolean().optional(),
 });
 export const HostsLinkOpenResponseSchema = z.object({
   type: z.literal("hosts.link.open.response"),
-  payload: z.object({ requestId: z.string(), routeId: z.string() }),
+  payload: z.object({
+    requestId: z.string(),
+    routeId: z.string(),
+    flowControl: z.boolean().optional(),
+  }),
+});
+
+export const HostsLinkAckSchema = z.object({
+  type: z.literal("hosts.link.ack.request"),
+  routeId: z.string(),
+  sequence: z.number().int().positive(),
 });

packages/protocol/src/messages.ts

@@ -5,6 +5,7 @@ import {
   HostsUpdateRequestSchema,
   HostsRemoveRequestSchema,
   HostsLinkOpenRequestSchema,
+  HostsLinkAckSchema,
   HostsRouteOpenRequestSchema,
   HostsRouteFrameRequestSchema,
   HostsRouteCloseRequestSchema,
@@ -3397,6 +3398,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
   HostsUpdateRequestSchema,
   HostsRemoveRequestSchema,
   HostsLinkOpenRequestSchema,
+  HostsLinkAckSchema,
   HostsRouteOpenRequestSchema,
   HostsRouteFrameRequestSchema,
   HostsRouteCloseRequestSchema,

packages/server/src/server/agent/agent-manager.test.ts

@@ -469,7 +469,7 @@ class TestAgentSession implements AgentSession {
 
   async startTurn(): Promise<{ turnId: string }> {
     this.interrupted = false;
-    const turnId = `turn-${++this.turnIdCounter}`;
+    const turnId = `turn-${this.id}-${++this.turnIdCounter}`;
     // Use setTimeout so events arrive after the caller sets up the foreground waiter
     setTimeout(() => {
       this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
@@ -13487,3 +13487,35 @@ test("stops only a currently owned public background item with its current token
   expect(stopped).toEqual(["one"]);
   await manager.closeAgent(agent.id);
 });
+
+test("turns keep distinct identities across an idle runtime collection and resume", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-turn-identity-"));
+  const storage = new AgentStorage(join(workdir, "agents"), logger);
+  const client = new TestAgentClient();
+  const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
+  const collect = runtimeCleanup(manager);
+  try {
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    const consume = async (prompt: string, clientMessageId: string) => {
+      for await (const _event of manager.streamAgent(agent.id, prompt, { clientMessageId })) {
+      }
+    };
+    await consume("first", "first");
+    const later = new Date(Date.now() + 61 * 60_000);
+    expect((await collect({ now: later })).collected).toHaveLength(1);
+    await ensureAgentLoaded(agent.id, { agentManager: manager, agentStorage: storage, logger });
+    await consume("second", "second");
+    const rows = manager.fetchTimeline(agent.id, { limit: 0 }).rows;
+    const userTurnIds = rows
+      .filter((row) => row.item.type === "user_message")
+      .map((row) => row.turnId);
+    expect(userTurnIds).toHaveLength(2);
+    expect(new Set(userTurnIds).size).toBe(2);
+  } finally {
+    await manager.closeAgent(manager.listAgents()[0]?.id ?? "missing").catch(() => undefined);
+    await storage.flush();
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});

packages/server/src/server/agent/agent-sdk-types.ts

@@ -690,6 +690,7 @@ export interface AgentSession {
    * replay them at their original timestamps; restored sessions omit old rows. */
   readonly initialTimeline?: ImportedTimelineEntry[];
   run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult>;
+  /** Returns an ID unique across the agent lifetime, including resumed sessions and daemon restarts. */
   startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>;
   steerActiveTurn?(prompt: AgentPromptInput, options: SteerActiveTurnOptions): Promise<SteerResult>;
   subscribe(callback: (event: AgentStreamEvent) => void): () => void;

packages/server/src/server/agent/providers/claude/agent.redesign.test.ts

@@ -1206,7 +1206,7 @@ test("reuses one autonomous run for unbound stream_event bursts with no foregrou
   );
 
   const firstRunId = internal.autonomousTurn?.id ?? null;
-  expect(firstRunId).toBe("autonomous-turn-1");
+  expect(firstRunId).toEqual(expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/));
   expect(internal.nextTurnOrdinal).toBe(2);
 
   await internal.routeSdkMessageFromPump(
@@ -1360,7 +1360,9 @@ test("a main-session frame opens no turn until Claude starts one", async () => {
   // session start a turn of its own, with no prompt from the user.
   await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
   await internal.routeSdkMessageFromPump(unboundTextDelta("picking this up"), queryMock);
-  expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
+  expect(internal.autonomousTurn?.id).toEqual(
+    expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
+  );
 
   await internal.routeSdkMessageFromPump(successResult(), queryMock);
   expect(internal.autonomousTurn).toBeNull();
@@ -1410,7 +1412,9 @@ test("Claude waking up after a helper finishes gets an autonomous turn that its
   expect(internal.autonomousTurn).toBeNull();
 
   await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
-  expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
+  expect(internal.autonomousTurn?.id).toEqual(
+    expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
+  );
   expect(internal.turnState).toBe("autonomous");
 
   await internal.routeSdkMessageFromPump(unboundTextDelta("the helper is done"), queryMock);
@@ -1446,7 +1450,9 @@ test("a steer folded into a turn keeps one turn, and a turn Claude starts after
 
   // A message that arrived while Claude wrote its final answer runs as a turn of its own.
   await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
-  expect(internal.autonomousTurn?.id).toBe("autonomous-turn-2");
+  expect(internal.autonomousTurn?.id).toEqual(
+    expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-2$/),
+  );
   await internal.routeSdkMessageFromPump(successResult(), queryMock);
   expect(internal.autonomousTurn).toBeNull();
   expect(internal.turnState).toBe("idle");
@@ -1483,7 +1489,9 @@ test("an idle session state closes an autonomous turn that no result closed", as
   session.subscribe((event) => events.push(event));
 
   await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
-  expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
+  expect(internal.autonomousTurn?.id).toEqual(
+    expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
+  );
 
   // Safety net: Claude says nothing at all is running any more.
   await internal.routeSdkMessageFromPump(sessionState("idle"), queryMock);
@@ -1685,7 +1693,9 @@ test("a new turn clears an interrupt flag no result will consume", async () => {
 
   await internal.routeSdkMessageFromPump(claudeTurnInit(), queryMock);
   expect(internal.pendingInterruptAbort).toBe(false);
-  expect(internal.autonomousTurn?.id).toBe("autonomous-turn-1");
+  expect(internal.autonomousTurn?.id).toEqual(
+    expect.stringMatching(/^autonomous-turn-[0-9a-f-]{36}-1$/),
+  );
 
   // This turn's own failure is reported, not swallowed as the interrupted turn's leftover.
   await internal.routeSdkMessageFromPump(

packages/server/src/server/agent/providers/claude/agent.test.ts

@@ -1345,6 +1345,30 @@ describe("ClaudeAgentSession features", () => {
     await session.close();
   });
 
+  test("keeps turn identities distinct across resumed session objects", async () => {
+    const { queryFactory } = createQueryMock();
+    const client = new ClaudeAgentClient({
+      logger,
+      queryFactory,
+      resolveBinary: async () => "/test/claude/bin",
+    });
+    const handle = {
+      provider: "claude" as const,
+      sessionId: "saved-session",
+      metadata: { cwd: process.cwd() },
+    };
+    const first = await client.resumeSession(handle);
+    const firstTurn = await first.startTurn("first");
+    await first.close();
+    const resumed = await client.resumeSession(handle);
+    try {
+      const nextTurn = await resumed.startTurn("second");
+      expect(nextTurn.turnId).not.toBe(firstTurn.turnId);
+    } finally {
+      await resumed.close();
+    }
+  });
+
   test("disables Claude hooks for internal agents only", async () => {
     const { queryFactory } = createQueryMock();
     const client = new ClaudeAgentClient({

packages/server/src/server/agent/providers/claude/agent.ts

@@ -2378,6 +2378,7 @@ class ClaudeAgentSession implements AgentSession {
   private historySessionId: string | null = null;
   private replayingHistory = false;
   private turnState: TurnState = "idle";
+  private readonly turnNamespace = randomUUID();
   private nextTurnOrdinal = 1;
   private cancelCurrentTurn: (() => void) | null = null;
   private cachedRuntimeInfo: AgentRuntimeInfo | null = null;
@@ -3983,7 +3984,7 @@ class ClaudeAgentSession implements AgentSession {
   }
 
   private createTurnId(owner: "foreground" | "autonomous"): string {
-    return `${owner}-turn-${this.nextTurnOrdinal++}`;
+    return `${owner}-turn-${this.turnNamespace}-${this.nextTurnOrdinal++}`;
   }
 
   private isTerminalTurnEvent(event: AgentStreamEvent): boolean {

packages/server/src/server/agent/providers/codex-app-server-agent.test.ts

@@ -1359,7 +1359,7 @@ describe("Codex app-server provider", () => {
     const firstSession = await createProviderWithFakeAppServer(beforeRestart).createSession(
       createConfig({ thinkingOptionId: undefined }),
     );
-    await firstSession.startTurn("first turn");
+    const firstTurn = await firstSession.startTurn("first turn");
     await expect(beforeRestart.waitForTurnStart()).resolves.toMatchObject({ effort: "medium" });
     const saved = waitForNextEvent(firstSession, "thread_started");
     beforeRestart.startsTurn({ threadId: "thread-1" });
@@ -1378,7 +1378,8 @@ describe("Codex app-server provider", () => {
       await expect(resumed.getRuntimeInfo()).resolves.toMatchObject({
         thinkingOptionId: "medium",
       });
-      await resumed.startTurn("turn after restart");
+      const resumedTurn = await resumed.startTurn("turn after restart");
+      expect(resumedTurn.turnId).not.toBe(firstTurn.turnId);
       await expect(afterRestart.waitForTurnStart()).resolves.toMatchObject({ effort: "medium" });
       beforeRestart.assertNoErrors();
       afterRestart.assertNoErrors();
@@ -4853,6 +4854,7 @@ describe("Codex app-server provider", () => {
     );
 
     try {
+      const started = waitForNextEvent(session, "turn_started");
       const resultPromise = session.run("Delegate the investigation.");
       await appServer.waitForTurnStart();
       const child = waitForProviderSubagent(session, "legacy-only-child-thread");
@@ -4870,7 +4872,7 @@ describe("Codex app-server provider", () => {
       await expect(child).resolves.toMatchObject({
         type: "provider_subagent",
         provider: "codex",
-        turnId: "codex-turn-0",
+        turnId: (await started).turnId,
         event: {
           type: "upsert",
           id: "legacy-only-child-thread",
@@ -4880,7 +4882,7 @@ describe("Codex app-server provider", () => {
       await expect(spawn).resolves.toMatchObject({
         type: "timeline",
         provider: "codex",
-        turnId: "codex-turn-0",
+        turnId: (await started).turnId,
         item: {
           type: "tool_call",
           callId: "spawn-legacy-only-child",

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -3529,6 +3529,7 @@ export class CodexAppServerAgentSession implements AgentSession {
   private threadRollbackAvailable = true;
   private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
   private startupEvents: AgentStreamEvent[] | null = null;
+  private readonly turnNamespace = randomUUID();
   private nextTurnOrdinal = 0;
   private activeForegroundTurnId: string | null = null;
   private activeClientMessageId: string | null = null;
@@ -5719,7 +5720,7 @@ export class CodexAppServerAgentSession implements AgentSession {
   }
 
   private createTurnId(): string {
-    return `codex-turn-${this.nextTurnOrdinal++}`;
+    return `codex-turn-${this.turnNamespace}-${this.nextTurnOrdinal++}`;
   }
 
   private pendingNotifications: Promise<void> = Promise.resolve();

packages/server/src/server/agent/providers/opencode-agent.test.ts

@@ -32,6 +32,16 @@ import type {
 const EXPECTED_STARTUP_BUDGET_MS = 30_000;
 const EXPECTED_EVENT_READINESS_BUDGET_MS = 45_000;
 
+// Keep event correlation and turn ordering assertions exact while accepting a fresh namespace.
+function turnIdAt(events: AgentStreamEvent[], ordinal: number): string {
+  const started = events.find((event) => event.type === "turn_started");
+  if (!started || started.type !== "turn_started" || typeof started.turnId !== "string") {
+    throw new Error("Expected a turn_started event with a turn ID");
+  }
+  const namespace = started.turnId.slice(0, started.turnId.lastIndexOf("-"));
+  return `${namespace}-${ordinal}`;
+}
+
 function tmpCwd(): string {
   const dir = mkdtempSync(path.join(os.tmpdir(), "opencode-agent-test-"));
   try {
@@ -757,6 +767,19 @@ describe("OpenCodeAgentClient adapter smoke tests", () => {
     }
   });
 
+  test("keeps turn identities distinct for the same persisted session", async () => {
+    const first = await createParentSession("ses_persisted");
+    const firstTurn = await first.parent.startTurn("first");
+    await first.parent.close();
+    const resumed = await createParentSession("ses_persisted");
+    try {
+      const nextTurn = await resumed.parent.startTurn("second");
+      expect(nextTurn.turnId).not.toBe(firstTurn.turnId);
+    } finally {
+      await resumed.parent.close();
+    }
+  });
+
   test.each([
     { saved: "default", variant: undefined },
     { saved: "high", variant: "high" },
@@ -1638,10 +1661,10 @@ describe("OpenCode adapter startTurn error handling", () => {
       item: { type: "user_message", clientMessageId: "client-message-1" },
     });
     expect(turn.events.map((event) => ("turnId" in event ? event.turnId : undefined))).toEqual([
-      "opencode-turn-0",
-      "opencode-turn-0",
-      "opencode-turn-0",
-      "opencode-turn-0",
+      turnIdAt(turn.events, 0),
+      turnIdAt(turn.events, 0),
+      turnIdAt(turn.events, 0),
+      turnIdAt(turn.events, 0),
     ]);
   });
 
@@ -1833,7 +1856,7 @@ describe("OpenCode adapter startTurn error handling", () => {
           type: "error",
           message: "Provider retry (attempt 1): model does not exist",
         },
-        turnId: "opencode-turn-0",
+        turnId: turnIdAt(events, 0),
       });
       expect(events.some((event) => event.type === "turn_failed")).toBe(false);
       await session.close();
@@ -2641,11 +2664,11 @@ describe("OpenCode adapter startTurn error handling", () => {
       expect(
         turnEventSignatures(events.filter((event) => event.type !== "provider_subagent")),
       ).toEqual([
-        ["turn_started", "opencode-turn-0"],
-        ["turn_canceled", "opencode-turn-0"],
-        ["turn_started", "opencode-turn-1"],
-        ["timeline", "opencode-turn-1"],
-        ["turn_completed", "opencode-turn-1"],
+        ["turn_started", turnIdAt(events, 0)],
+        ["turn_canceled", turnIdAt(events, 0)],
+        ["turn_started", turnIdAt(events, 1)],
+        ["timeline", turnIdAt(events, 1)],
+        ["turn_completed", turnIdAt(events, 1)],
       ]);
     } finally {
       settleRetry.resolve();
@@ -2706,7 +2729,7 @@ describe("OpenCode adapter startTurn error handling", () => {
         expect(events).toContainEqual(
           expect.objectContaining({
             type: "timeline",
-            turnId: "opencode-turn-1",
+            turnId: turnIdAt(events, 1),
             item: expect.objectContaining({ text: "Wake output streams live." }),
           }),
         ),
@@ -2726,11 +2749,11 @@ describe("OpenCode adapter startTurn error handling", () => {
       });
       await vi.waitFor(() =>
         expect(events).toContainEqual(
-          expect.objectContaining({ type: "turn_completed", turnId: "opencode-turn-1" }),
+          expect.objectContaining({ type: "turn_completed", turnId: turnIdAt(events, 1) }),
         ),
       );
       await expect(session.startTurn("after the wake")).resolves.toEqual({
-        turnId: "opencode-turn-2",
+        turnId: turnIdAt(events, 2),
       });
     } finally {
       settleAbort.resolve();
@@ -2843,13 +2866,13 @@ describe("OpenCode adapter startTurn error handling", () => {
       expect(events).toContainEqual({
         type: "turn_started",
         provider: "opencode",
-        turnId: "opencode-turn-1",
+        turnId: turnIdAt(events, 1),
       });
       await vi.waitFor(() =>
         expect(events).toContainEqual({
           type: "turn_completed",
           provider: "opencode",
-          turnId: "opencode-turn-1",
+          turnId: turnIdAt(events, 1),
           usage: expect.objectContaining({
             inputTokens: 7,
             outputTokens: 3,
@@ -2858,7 +2881,7 @@ describe("OpenCode adapter startTurn error handling", () => {
         }),
       );
       const autonomousCompletion = events.find(
-        (event) => event.type === "turn_completed" && event.turnId === "opencode-turn-1",
+        (event) => event.type === "turn_completed" && event.turnId === turnIdAt(events, 1),
       );
       expect(autonomousCompletion?.type).toBe("turn_completed");
       if (autonomousCompletion?.type === "turn_completed") {
@@ -2867,7 +2890,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       expect(events).toContainEqual(
         expect.objectContaining({
           type: "timeline",
-          turnId: "opencode-turn-1",
+          turnId: turnIdAt(events, 1),
           item: expect.objectContaining({
             type: "assistant_message",
             text: "Autonomous wake completed.",
@@ -2875,7 +2898,7 @@ describe("OpenCode adapter startTurn error handling", () => {
         }),
       );
       await expect(session.startTurn("after autonomous wake")).resolves.toEqual({
-        turnId: "opencode-turn-2",
+        turnId: turnIdAt(events, 2),
       });
     } finally {
       settleAbort.resolve();
@@ -2954,7 +2977,7 @@ describe("OpenCode adapter startTurn error handling", () => {
         type: "turn_canceled",
         provider: "opencode",
         reason: "interrupted",
-        turnId: "opencode-turn-0",
+        turnId: turnIdAt(events, 0),
       });
 
       openCode.emitEvent({
@@ -2962,7 +2985,7 @@ describe("OpenCode adapter startTurn error handling", () => {
         properties: { sessionID: "ses_retried_stop" },
       });
       await expect(session.startTurn("replacement")).resolves.toEqual({
-        turnId: "opencode-turn-1",
+        turnId: turnIdAt(events, 1),
       });
       expect(openCode.calls.sessionPromptAsync).toHaveLength(2);
     } finally {
@@ -3058,7 +3081,7 @@ describe("OpenCode adapter startTurn error handling", () => {
         expect(events).toContainEqual({
           type: "turn_started",
           provider: "opencode",
-          turnId: "opencode-turn-1",
+          turnId: turnIdAt(events, 1),
         }),
       );
 
@@ -3767,7 +3790,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       ];
       const replacement = session.startTurn("second");
       await vi.advanceTimersByTimeAsync(10_000);
-      await expect(replacement).resolves.toEqual({ turnId: "opencode-turn-1" });
+      await expect(replacement).resolves.toEqual({ turnId: turnIdAt(events, 1) });
       expect(countEvents(events, "turn_canceled")).toBe(1);
     } finally {
       vi.useRealTimers();
@@ -4009,6 +4032,9 @@ describe("OpenCode adapter startTurn error handling", () => {
       return { data: true };
     };
 
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+
     try {
       await session.startTurn("first");
       const interrupt = session.interrupt();
@@ -4022,7 +4048,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       expect(openCode.calls.sessionStatus).toHaveLength(1);
 
       releaseIdle.resolve();
-      await expect(replacement).resolves.toEqual({ turnId: "opencode-turn-1" });
+      await expect(replacement).resolves.toEqual({ turnId: turnIdAt(events, 1) });
       expect(openCode.calls.sessionPromptAsync).toHaveLength(2);
       expect(openCode.calls.sessionStatus.length).toBeGreaterThanOrEqual(2);
     } finally {
@@ -4722,9 +4748,9 @@ describe("OpenCode provider subagent contract", () => {
       "turn_completed",
     ]);
     expect(events.map((event) => ("turnId" in event ? event.turnId : undefined))).toEqual([
-      "opencode-turn-0",
-      "opencode-turn-0",
-      "opencode-turn-0",
+      turnIdAt(events, 0),
+      turnIdAt(events, 0),
+      turnIdAt(events, 0),
     ]);
     expect(events).toContainEqual(
       expect.objectContaining({
@@ -4760,7 +4786,7 @@ describe("OpenCode provider subagent contract", () => {
           expect(events).toContainEqual({
             type: "turn_started",
             provider: "opencode",
-            turnId: "opencode-turn-0",
+            turnId: turnIdAt(events, 0),
           });
         });
         for (const event of assistantTurnEvents({
@@ -4774,9 +4800,9 @@ describe("OpenCode provider subagent contract", () => {
 
         expect(childClient.calls.sessionStatus).toEqual([{ directory: "/workspace/repo" }]);
         expect(turnEventSignatures(events)).toEqual([
-          ["turn_started", "opencode-turn-0"],
-          ["timeline", "opencode-turn-0"],
-          ["turn_completed", "opencode-turn-0"],
+          ["turn_started", turnIdAt(events, 0)],
+          ["timeline", turnIdAt(events, 0)],
+          ["turn_completed", turnIdAt(events, 0)],
         ]);
         expect(events).toContainEqual(
           expect.objectContaining({
@@ -4881,9 +4907,9 @@ describe("OpenCode provider subagent contract", () => {
 
       await vi.waitFor(() => {
         expect(turnEventSignatures(events)).toEqual([
-          ["turn_started", "opencode-turn-0"],
-          ["timeline", "opencode-turn-0"],
-          ["turn_completed", "opencode-turn-0"],
+          ["turn_started", turnIdAt(events, 0)],
+          ["timeline", turnIdAt(events, 0)],
+          ["turn_completed", turnIdAt(events, 0)],
         ]);
       });
       expect(events).toContainEqual(
@@ -4927,9 +4953,9 @@ describe("OpenCode provider subagent contract", () => {
 
       await vi.waitFor(() => {
         expect(turnEventSignatures(events)).toEqual([
-          ["turn_started", "opencode-turn-0"],
-          ["timeline", "opencode-turn-0"],
-          ["turn_completed", "opencode-turn-0"],
+          ["turn_started", turnIdAt(events, 0)],
+          ["timeline", turnIdAt(events, 0)],
+          ["turn_completed", turnIdAt(events, 0)],
         ]);
       });
     } finally {
@@ -5003,7 +5029,7 @@ describe("OpenCode provider subagent contract", () => {
       });
       await vi.waitFor(() => {
         expect(events).toEqual([
-          { type: "turn_started", provider: "opencode", turnId: "opencode-turn-0" },
+          { type: "turn_started", provider: "opencode", turnId: turnIdAt(events, 0) },
         ]);
       });
 
@@ -5118,12 +5144,12 @@ describe("OpenCode provider subagent contract", () => {
 
       await streamDrained.promise;
       expect(events.filter((event) => event.type !== "provider_subagent")).toEqual([
-        { type: "turn_started", provider: "opencode", turnId: "opencode-turn-0" },
+        { type: "turn_started", provider: "opencode", turnId: turnIdAt(events, 0) },
         {
           type: "turn_canceled",
           provider: "opencode",
           reason: "interrupted",
-          turnId: "opencode-turn-0",
+          turnId: turnIdAt(events, 0),
         },
       ]);
     } finally {
@@ -5174,7 +5200,7 @@ describe("OpenCode provider subagent contract", () => {
       expect.objectContaining({
         type: "permission_requested",
         request: expect.objectContaining({ id: "perm_child_external" }),
-        turnId: "opencode-turn-0",
+        turnId: turnIdAt(events, 0),
       }),
     );
   });

packages/server/src/server/agent/providers/opencode-agent.ts

@@ -3409,6 +3409,7 @@ class OpenCodeAgentSession implements AgentSession {
   private partTypes = new Map<string, string>();
   private availableModesCache: AgentMode[] | null = null;
   private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
+  private readonly turnNamespace = randomUUID();
   private nextTurnOrdinal = 0;
   private turnState: OpenCodeTurnState = { status: "idle" };
   /**
@@ -4886,7 +4887,7 @@ class OpenCodeAgentSession implements AgentSession {
 
   private createTurnId(): string {
     this.gapRepairRevision += 1;
-    return `opencode-turn-${this.nextTurnOrdinal++}`;
+    return `opencode-turn-${this.turnNamespace}-${this.nextTurnOrdinal++}`;
   }
 
   private traceOpenCode(msg: OpenCodeTraceMessage, data: OpenCodeTraceData = {}): void {

packages/server/src/server/authorization/operation-permissions.ts

@@ -11,6 +11,7 @@ const INBOUND_PERMISSION = {
   "hosts.update.request": "tunnel.manage",
   "hosts.remove.request": "tunnel.manage",
   "hosts.link.open.request": null,
+  "hosts.link.ack.request": null,
   "hosts.route.open.request": null,
   "hosts.route.frame.request": null,
   "hosts.route.close.request": null,

packages/server/src/server/connected-hosts/flow-control.ts

@@ -0,0 +1,51 @@
+export class HostChannelOutputLimitError extends Error {
+  constructor() {
+    super("Host channel output limit reached");
+  }
+}
+
+/** Completion belongs to the receiver's forwarding write, not just our physical socket. */
+export class HostChannelFlowControl {
+  private sequence = 0;
+  private bytes = 0;
+  private closed = false;
+  private readonly pending = new Map<
+    number,
+    {
+      resolve(): void;
+      reject(error: Error): void;
+    }
+  >();
+
+  constructor(private readonly reserveShared: (bytes: number) => () => void) {}
+
+  async send(bytes: number, write: () => Promise<void>): Promise<void> {
+    if (this.closed) throw new Error("Host channel closed");
+    // Bound both payload retention and metadata for a producer of tiny frames.
+    if (this.bytes + bytes > 32 * 1024 * 1024 || this.pending.size >= 4096)
+      throw new HostChannelOutputLimitError();
+    const release = this.reserveShared(bytes);
+    this.bytes += bytes;
+    const sequence = ++this.sequence;
+    const acknowledgment = new Promise<void>((resolve, reject) => {
+      this.pending.set(sequence, { resolve, reject });
+    });
+    try {
+      await Promise.all([acknowledgment, write()]);
+    } finally {
+      this.pending.delete(sequence);
+      this.bytes -= bytes;
+      release();
+    }
+  }
+
+  acknowledge(sequence: number): void {
+    this.pending.get(sequence)?.resolve();
+  }
+
+  close(): void {
+    this.closed = true;
+    for (const pending of this.pending.values()) pending.reject(new Error("Host channel closed"));
+    this.pending.clear();
+  }
+}

packages/server/src/server/connected-hosts/hosts.e2e.test.ts

@@ -1141,7 +1141,7 @@ test("source-key registration remains idempotent at directory capacity", async (
 });
 
 /** Real WebSocket proxy records the A–B physical wire without replacing either daemon. */
-async function hostWire(b: TestPaseoDaemon) {
+async function hostWire(b: TestPaseoDaemon, options: { legacyFlowControl?: boolean } = {}) {
   const http = createServer();
   const wss = new WebSocketServer({ server: http });
   const frames: { binary: boolean; bytes: Buffer; direction: "toB" | "fromB" }[] = [];
@@ -1154,7 +1154,14 @@ async function hostWire(b: TestPaseoDaemon) {
       for (const send of pending.splice(0)) send();
     });
     downstream.on("message", (data, binary) => {
-      const bytes = Buffer.from(data as Buffer);
+      let bytes = Buffer.from(data as Buffer);
+      if (options.legacyFlowControl && !binary) {
+        const envelope = JSON.parse(bytes.toString());
+        if (envelope.message?.type === "hosts.link.open.request") {
+          delete envelope.message.flowControl;
+          bytes = Buffer.from(JSON.stringify(envelope));
+        }
+      }
       frames.push({ binary, bytes, direction: "toB" });
       const send = () => {
         if (upstream.readyState === 1) upstream.send(bytes, { binary });
@@ -2390,6 +2397,169 @@ test(
   },
 );
 
+test("a host channel waits for its downstream consumer without blocking sibling requests", async () => {
+  const b = await daemon();
+  const destination = await api(b);
+  const workspace = await destination.paseo.workspaces.create({
+    source: { kind: "directory", path: b.staticDir },
+  });
+  const agent = await workspace.agents.create({
+    config: { provider: "claude/claude-test", modeId: "full-access" },
+    title: "Flow-controlled history",
+  });
+  const text = "x".repeat(240 * 1024);
+  for (let index = 0; index < 50; index++)
+    await b.daemon.agentManager.appendTimelineItem(agent.id, { type: "user_message", text });
+
+  const channel = createHostChannelTransport(
+    destination.client,
+    b.daemon.getServerId()!,
+    "held-consumer",
+    null,
+    true,
+  )({ url: "ws://host-link/ws" });
+  const received = Promise.withResolvers<void>();
+  const forward = Promise.withResolvers<void>();
+  let binaryFrames = 0;
+  channel.onMessage((_data, binary) => {
+    if (!binary) return;
+    binaryFrames++;
+    received.resolve();
+    return forward.promise;
+  });
+  const consumer = new DaemonClient({
+    url: "ws://host-link/ws",
+    clientId: "held-consumer",
+    reconnect: { enabled: false },
+    transportFactory: () => channel,
+  });
+  const remote = createPaseoApi(consumer);
+  cleanup.push(async () => {
+    forward.resolve();
+    await remote.dispose();
+    await consumer.close();
+  });
+  await consumer.connect();
+  const history = remote.agents.ref(agent.id).timeline.refetch({ limit: 0 });
+  void history.catch(() => {}); // Cleanup can close the request after an earlier assertion fails.
+  await received.promise;
+  // A round trip on the same physical link must progress while this channel is held.
+  await destination.client.ping();
+  expect(binaryFrames).toBe(1);
+  forward.resolve();
+  const response = await history;
+  const messages = response.entries.flatMap((row) =>
+    row.item.type === "user_message" ? [row.item.text] : [],
+  );
+  expect(messages).toHaveLength(50);
+  expect(messages.every((message) => message === text)).toBe(true);
+  await consumer.ping();
+});
+
+test("small host-channel responses fill the window before earlier forwarding completes", async () => {
+  const b = await daemon();
+  const destination = await api(b);
+  const channel = createHostChannelTransport(
+    destination.client,
+    b.daemon.getServerId()!,
+    "concurrent-consumer",
+    null,
+    true,
+  )({ url: "ws://host-link/ws" });
+  const first = Promise.withResolvers<void>();
+  const forward = Promise.withResolvers<void>();
+  let responses = 0;
+  channel.onMessage((data, binary) => {
+    if (binary || JSON.parse(String(data)).message?.type !== "pong") return;
+    responses++;
+    first.resolve();
+    return forward.promise;
+  });
+  const consumer = new DaemonClient({
+    url: "ws://host-link/ws",
+    clientId: "concurrent-consumer",
+    reconnect: { enabled: false },
+    transportFactory: () => channel,
+  });
+  cleanup.push(async () => {
+    forward.resolve();
+    await consumer.close();
+  });
+  await consumer.connect();
+  const requests = Promise.all(Array.from({ length: 10 }, () => consumer.ping()));
+  void requests.catch(() => {});
+  await first.promise;
+  await destination.client.ping();
+  await destination.client.ping();
+  expect(responses).toBe(10);
+  forward.resolve();
+  await requests;
+});
+
+test("legacy host links receive no acknowledgment messages", async () => {
+  const b = await daemon();
+  const wire = await hostWire(b, { legacyFlowControl: true });
+  const origin = createHostClient(wire.connection, {
+    clientId: "legacy-flow-origin",
+    reconnect: { enabled: false },
+  });
+  cleanup.push(() => origin.close());
+  await origin.connect();
+  const consumer = new DaemonClient({
+    url: "ws://host-link/ws",
+    clientId: "legacy-flow-consumer",
+    reconnect: { enabled: false },
+    transportFactory: createHostChannelTransport(
+      origin,
+      b.daemon.getServerId()!,
+      "legacy",
+      null,
+      true,
+    ),
+  });
+  cleanup.push(() => consumer.close());
+  await consumer.connect();
+  await consumer.ping();
+  await origin.ping();
+  const operations = wire.frames
+    .filter((frame) => !frame.binary && frame.direction === "toB")
+    .map((frame) => JSON.parse(frame.bytes.toString()).message?.type);
+  expect(operations).toContain("hosts.link.open.request");
+  expect(operations).not.toContain("hosts.link.ack.request");
+});
+
+test("mixed asynchronous and synchronous receiver failures close only their host channel", async () => {
+  const b = await daemon();
+  const destination = await api(b);
+  const channel = createHostChannelTransport(
+    destination.client,
+    b.daemon.getServerId()!,
+    "failing-consumer",
+    null,
+    true,
+  )({ url: "ws://host-link/ws" });
+  const failed = Promise.withResolvers<unknown>();
+  channel.onError(failed.resolve);
+  channel.onMessage(async () => {
+    throw new Error("asynchronous receiver failure");
+  });
+  channel.onMessage(() => {
+    throw new Error("synchronous receiver failure");
+  });
+  const consumer = new DaemonClient({
+    url: "ws://host-link/ws",
+    clientId: "failing-consumer",
+    reconnect: { enabled: false },
+    transportFactory: () => channel,
+  });
+  cleanup.push(() => consumer.close());
+  const connecting = consumer.connect().catch((error: unknown) => error);
+  expect(await failed.promise).toBeInstanceOf(Error);
+  await connecting;
+  expect(consumer.isConnected).toBe(false);
+  await destination.client.ping();
+});
+
 test("a pre-write host budget rejection leaves sibling routes connected", async () => {
   const { HostTransportBudget } = await import("@getpaseo/client/node");
   const b = await daemon();

packages/server/src/server/connected-hosts/index.ts

@@ -384,7 +384,7 @@ export class ConnectedHosts {
     originSessionId: string;
     permissions: readonly DaemonPermission[];
     signal: AbortSignal;
-    frame(data: string | Uint8Array): void;
+    frame(data: string | Uint8Array): void | Promise<void>;
     ended(code: string, message: string): void;
   }): Promise<{ send(data: string | Uint8Array): void; close(): void }> {
     await this.ready;
@@ -499,7 +499,7 @@ export class ConnectedHosts {
           return;
         }
       }
-      input.frame(binary ? (data as Uint8Array) : String(data));
+      return input.frame(binary ? (data as Uint8Array) : String(data));
     });
     await ready;
     if (closed) throw new HostError("host_offline", "Host closed during connection");

packages/server/src/server/connected-hosts/link.test.ts

@@ -0,0 +1,118 @@
+import { afterEach, expect, test } from "vitest";
+import { DaemonLinkEndpoint, type VirtualHostSocket } from "./index.js";
+import { SessionDelivery } from "../session/owned-subscriptions/index.js";
+import type { SessionInboundMessage, SessionOutboundMessage } from "@getpaseo/protocol/messages";
+
+const cleanup: (() => Promise<void>)[] = [];
+afterEach(async () => {
+  for (const close of cleanup.splice(0).toReversed()) await close();
+});
+
+async function link() {
+  const source = {};
+  const messages: SessionOutboundMessage[] = [];
+  const sockets: VirtualHostSocket[] = [];
+  const delivery = new SessionDelivery((_source, message) => {
+    messages.push(message);
+  });
+  delivery.attach(source, true);
+  const endpoint = new DaemonLinkEndpoint({
+    identity: "physical",
+    permissions: () => ["daemon.read"],
+    routed: () => false,
+    bufferedAmount: () => 0,
+    delivery,
+    admit: async (socket) => {
+      sockets.push(socket);
+    },
+  });
+  cleanup.push(async () => {
+    endpoint.close();
+    await delivery.close();
+  });
+  const dispatch = (message: SessionInboundMessage) =>
+    delivery.request(source, message, () => endpoint.dispatch(message));
+  const open = async (channelId: string, flowControl = true) => {
+    await dispatch({
+      type: "hosts.link.open.request",
+      requestId: channelId,
+      channelId,
+      flowControl,
+    });
+    const response = messages.find(
+      (message) =>
+        message.type === "hosts.link.open.response" && message.payload.requestId === channelId,
+    );
+    expect(response).toMatchObject({
+      type: "hosts.link.open.response",
+      payload: flowControl ? { flowControl: true } : {},
+    });
+    if (response?.type !== "hosts.link.open.response") throw new Error("Missing opened channel");
+    return { socket: sockets.at(-1)!, id: response.payload.routeId };
+  };
+  const ack = (routeId: string, sequence: number) =>
+    dispatch({ type: "hosts.link.ack.request", routeId, sequence });
+  const frames = () =>
+    messages
+      .filter((message) => message.type === "hosts.route.frame")
+      .map((message) => message.payload.data);
+  return { open, ack, frames };
+}
+
+test("a channel waits for forwarding acknowledgments while its sibling keeps sending", async () => {
+  const host = await link();
+  const slow = await host.open("slow");
+  const sibling = await host.open("sibling");
+  let completed = false;
+  const first = slow.socket.send("first").then(() => {
+    completed = true;
+    return undefined;
+  });
+  let secondCompleted = false;
+  const second = slow.socket.send("second").then(() => {
+    secondCompleted = true;
+    return undefined;
+  });
+  await expect.poll(host.frames).toEqual(["first", "second"]);
+  expect(completed).toBe(false);
+  const independent = sibling.socket.send("independent");
+  await expect.poll(host.frames).toEqual(["first", "second", "independent"]);
+  await host.ack(sibling.id, 1);
+  await independent;
+  await host.ack(slow.id, 1);
+  await first;
+  await expect.poll(host.frames).toEqual(["first", "second", "independent"]);
+  await host.ack(slow.id, 1); // A duplicate acknowledgment cannot release the next frame.
+  expect(completed).toBe(true);
+  expect(secondCompleted).toBe(false);
+  await host.ack(slow.id, 2);
+  await second;
+});
+
+test("closing a blocked channel rejects its pending writes and leaves a sibling usable", async () => {
+  const host = await link();
+  const slow = await host.open("slow");
+  const sibling = await host.open("sibling", false);
+  const first = slow.socket.send("first");
+  const second = slow.socket.send("second");
+  const results = Promise.allSettled([first, second]);
+  await expect.poll(host.frames).toEqual(["first", "second"]);
+  slow.socket.close();
+  expect((await results).map((result) => result.status)).toEqual(["rejected", "rejected"]);
+  await sibling.socket.send("legacy sibling");
+  expect(host.frames()).toEqual(["first", "second", "legacy sibling"]);
+});
+
+test("an unpaced producer exhausting its acknowledgment window closes only its channel", async () => {
+  const host = await link();
+  const slow = await host.open("slow");
+  const sibling = await host.open("sibling", false);
+  const writes: Promise<unknown>[] = [];
+  // A producer ignoring send completion must remain bounded even for tiny messages.
+  for (let index = 0; index < 5000; index++)
+    writes.push(slow.socket.send("x").catch((error: unknown) => error));
+  await Promise.all(writes);
+  expect(slow.socket.readyState).toBe(3);
+  await sibling.socket.send("still connected");
+  expect(host.frames().at(-1)).toBe("still connected");
+});

packages/server/src/server/connected-hosts/link.ts

@@ -1,3 +1,4 @@
+import { HostChannelFlowControl, HostChannelOutputLimitError } from "./flow-control.js";
 import type { SessionDelivery } from "../session/owned-subscriptions/index.js";
 import { EventEmitter } from "node:events";
 import { createHash } from "node:crypto";
@@ -15,6 +16,7 @@ import type {
 /** A logical socket has no network resources or authority of its own. */
 export class VirtualHostSocket extends EventEmitter {
   readonly maxFrameBytes = MAX_HOST_CHANNEL_FRAME_BYTES;
+  readonly supportsConcurrentWrites = true;
   readyState = 1;
   constructor(
     private readonly output: (data: string | Uint8Array) => Promise<void>,
@@ -61,7 +63,11 @@ export class VirtualHostSocket extends EventEmitter {
 
 /** One authenticated physical source owns all its channels and their admission ceiling. */
 export class DaemonLinkEndpoint {
-  private readonly channels = new Map<string, { scopeId: string; socket: VirtualHostSocket }>();
+  private readonly channels = new Map<
+    string,
+    { scopeId: string; socket: VirtualHostSocket; flowControl?: HostChannelFlowControl }
+  >();
+  private pendingOutputBytes = 0;
   private closed = false;
   constructor(
     private readonly parent: {
@@ -76,6 +82,7 @@ export class DaemonLinkEndpoint {
   handles(message: SessionInboundMessage): boolean {
     return (
       message.type === "hosts.link.open.request" ||
+      message.type === "hosts.link.ack.request" ||
       (message.type === "hosts.route.frame.request" && this.channels.has(message.routeId)) ||
       (message.type === "hosts.route.close.request" &&
         ((!!message.routeId && this.channels.has(message.routeId)) ||
@@ -83,6 +90,10 @@ export class DaemonLinkEndpoint {
     );
   }
   async dispatch(message: SessionInboundMessage): Promise<void> {
+    if (message.type === "hosts.link.ack.request") {
+      this.channels.get(message.routeId)?.flowControl?.acknowledge(message.sequence);
+      return;
+    }
     if (message.type === "hosts.link.open.request") {
       const error = (code: string, text: string) =>
         this.parent.delivery.reply({
@@ -103,8 +114,19 @@ export class DaemonLinkEndpoint {
         return;
       }
       let socket: VirtualHostSocket;
+      const flowControl = message.flowControl
+        ? new HostChannelFlowControl((bytes) => {
+            if (this.pendingOutputBytes + bytes > 256 * 1024 * 1024)
+              throw new HostChannelOutputLimitError();
+            this.pendingOutputBytes += bytes;
+            return () => {
+              this.pendingOutputBytes -= bytes;
+            };
+          })
+        : undefined;
       const owner = this.parent.delivery.begin("host-link", undefined, () => {
         this.channels.delete(owner.id);
+        flowControl?.close();
         socket?.close();
       });
       const id = owner.id;
@@ -130,15 +152,25 @@ export class DaemonLinkEndpoint {
             throw new Error("Host channel output limit reached");
           }
           try {
-            if (output) await owner.emitAndWait(output);
-            else await owner.emitBinaryAndWait(frame!);
+            const write = async () => {
+              if (output) await owner.emitAndWait(output);
+              else await owner.emitBinaryAndWait(frame!);
+            };
+            if (flowControl) await flowControl.send(bytes, write);
+            else await write();
           } catch (failure) {
-            socket.close();
+            socket.close(
+              failure instanceof HostChannelOutputLimitError ? 1009 : 1000,
+              failure instanceof HostChannelOutputLimitError
+                ? failure.message
+                : "Host channel closed",
+            );
             throw failure;
           }
         },
         (code, reason) => {
           this.channels.delete(id);
+          flowControl?.close();
           let failure = "host_offline";
           if (code === 1009) failure = "resource_limit";
           if (code === 1008) failure = "permission_denied";
@@ -167,7 +199,7 @@ export class DaemonLinkEndpoint {
           });
         },
       );
-      this.channels.set(id, { scopeId: message.channelId, socket });
+      this.channels.set(id, { scopeId: message.channelId, socket, flowControl });
       try {
         await this.parent.admit(socket, ceiling);
         if (this.closed || socket.readyState !== 1) {
@@ -176,7 +208,11 @@ export class DaemonLinkEndpoint {
         }
         this.parent.delivery.reply({
           type: "hosts.link.open.response",
-          payload: { requestId: message.requestId, routeId: id },
+          payload: {
+            requestId: message.requestId,
+            routeId: id,
+            ...(flowControl ? { flowControl: true } : {}),
+          },
         });
       } catch {
         socket.close();

packages/server/src/server/connected-hosts/session.test.ts

@@ -0,0 +1,70 @@
+import { expect, test } from "vitest";
+import { ConnectedHostSession, type ConnectedHosts } from "./index.js";
+import { SessionDelivery } from "../session/owned-subscriptions/index.js";
+
+// The upstream host and final writer are independently controlled transport peers.
+test.each(["text", "binary"] as const)(
+  "routed %s completion waits for the final writer",
+  async (kind) => {
+    const source = {};
+    const writing = Promise.withResolvers<void>();
+    const drain = Promise.withResolvers<void>();
+    const write = () => {
+      writing.resolve();
+      return drain.promise;
+    };
+    const delivery = new SessionDelivery((_source, message) => {
+      if (message.type === "hosts.route.frame") return write();
+    }, write);
+    delivery.attach(source, true);
+    let receive!: Parameters<ConnectedHosts["openRoute"]>[0]["frame"];
+    const unexpected = (): never => {
+      throw new Error("Unexpected host operation");
+    };
+    const session = new ConnectedHostSession(
+      {
+        list: unexpected,
+        observe: unexpected,
+        add: unexpected,
+        update: unexpected,
+        remove: unexpected,
+        openRoute: async (input) => {
+          receive = input.frame;
+          return { send: () => {}, close: () => {} };
+        },
+      },
+      delivery,
+      { sourceId: "owner", manageSources: false },
+      () => [],
+      () => {},
+      () => false,
+      "origin",
+      () => 0,
+    );
+    const request = {
+      type: "hosts.route.open.request" as const,
+      requestId: "open",
+      serverId: "remote",
+      scopeId: "scope",
+    };
+    await delivery.request(source, request, () => session.dispatch(request)!);
+    let completed = false;
+    const forwarded = Promise.resolve(
+      receive(kind === "text" ? "payload" : new Uint8Array([1, 2])),
+    ).then(() => {
+      completed = true;
+      return undefined;
+    });
+    try {
+      await writing.promise;
+      expect(completed).toBe(false);
+      drain.resolve();
+      await forwarded;
+      expect(completed).toBe(true);
+    } finally {
+      drain.resolve();
+      await forwarded;
+      await delivery.close();
+    }
+  },
+);

packages/server/src/server/connected-hosts/session.ts

@@ -23,7 +23,10 @@ interface SessionRoute {
 export class ConnectedHostSession {
   private readonly routes = new Map<string, SessionRoute>();
   constructor(
-    private readonly hosts: ConnectedHosts,
+    private readonly hosts: Pick<
+      ConnectedHosts,
+      "list" | "observe" | "add" | "update" | "remove" | "openRoute"
+    >,
     private readonly delivery: SessionDelivery,
     private readonly actor: HostActor,
     private readonly permissions: () => DaemonPermission[],
@@ -206,7 +209,7 @@ export class ConnectedHostSession {
         originSessionId: this.originSessionId,
         permissions: this.permissions(),
         signal: owner.signal,
-        frame: (data) => {
+        frame: async (data) => {
           const output: SessionOutboundMessage | null =
             typeof data === "string"
               ? {
@@ -223,8 +226,8 @@ export class ConnectedHostSession {
             fail("resource_limit", "The receiving client cannot keep up with host output");
             return;
           }
-          if (output) owner.emit(output);
-          else owner.emitBinary(binary!);
+          if (output) await owner.emitAndWait(output);
+          else await owner.emitBinaryAndWait(binary!);
         },
         ended: fail,
       });

packages/server/src/server/websocket-server.ts

@@ -451,6 +451,7 @@ function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffe
 
 export interface WebSocketLike {
   maxFrameBytes?: number;
+  supportsConcurrentWrites?: boolean;
   readyState: number;
   bufferedAmount?: number;
   send: (
@@ -1307,6 +1308,7 @@ export class VoiceAssistantWebSocketServer {
     if (!transport) {
       transport = new ChunkedTransport({
         frameBytes: ws.maxFrameBytes ?? 32 * 1024 * 1024,
+        supportsConcurrentWrites: ws.supportsConcurrentWrites,
         onFailure: (err) =>
           this.closePhysicalSocket({
             ws,

patches/@legendapp+list+3.7.1.patch

@@ -19,15 +19,21 @@
 # 7. Header and footer revealed with the rows (RevealedLayoutView, ListComponent).
 # 8. Top-padding shrink hold on iOS only (setPaddingTop): Android paints the held height as blank
 #    space at the end.
-# 9. Rows cut off at the top count as in view (getIdsInVisibleRange): they follow the fully
-#    visible rows in idsInView, so a disclosure hold on a row whose top is above the viewport
-#    still anchors on that row instead of the row below it.
+# 9. Rows cut off at the top count as in view, and so does the row just past the end of the
+#    viewport (getIdsInVisibleRange): they follow the fully visible rows in idsInView, so a
+#    disclosure hold on a row whose top is above the viewport still anchors on that row instead
+#    of the row below it, and a hold on the row below a bottom-edge trigger anchors even when
+#    that row starts below the viewport.
 # 10. A size change keeps only a row shouldRestorePosition accepts in place (prepareMVCP), as a
 #    data change already does, so a disclosure that only grows (a todo card) holds the cut-off
 #    row instead of the first row in view.
 # 11. A 0 x 0 measurement is not a size (updateItemSizesBatch): rows of a hidden retained tab
 #    (display: none) still render and measure 0 x 0. Recording it collapsed the row, and a row
 #    whose frame was unchanged when the tab showed again got no layout event to correct it.
+# 12. No Android clamp prediction for a size change (prepareMVCP): upstream measures the adjustment
+#    of a shrink past the end from the new end, assuming the scroll view clamped first. The app's
+#    scroll view clamps after the adjustment on both platforms (modules/paseo-scroll), so a
+#    bottom-edge disclosure collapsing near the end landed at the end instead of in place.
 #
 # Upgrading: drop each hunk whose fix the new version ships, regenerate with
 # `npx patch-package @legendapp/list`, and copy this header back on top.
@@ -48,7 +54,7 @@ index 983e937..970ba51 100644
       * Returns a stable item type used for pooling and size estimation.
       */
 diff --git a/node_modules/@legendapp/list/react-native.js b/node_modules/@legendapp/list/react-native.js
-index eeb3fb0..5fe940d 100644
+index eeb3fb0..5ebd8e9 100644
 --- a/node_modules/@legendapp/list/react-native.js
 +++ b/node_modules/@legendapp/list/react-native.js
 @@ -1497,7 +1497,15 @@ function getItemSize(ctx, key, index, data, useAverageSize, preferCachedSize, no
@@ -135,7 +141,28 @@ index eeb3fb0..5fe940d 100644
      }
      if (dataChanged && idsInView.length > 0 && state.didContainersLayout) {
        for (let i = 0; i < idsInView.length; i++) {
-@@ -2371,7 +2394,10 @@ function finishScrollTo(ctx) {
+@@ -1874,15 +1897,11 @@ function prepareMVCP(ctx, dataChanged, adjustmentSource) {
+         now,
+         positionDiff
+       });
+-      if (Platform.OS === "android" && !dataChanged && scrollTarget === void 0) {
+-        const maxScroll = getScrollRangeEnd(ctx);
+-        if (state.scroll > maxScroll + MVCP_POSITION_EPSILON) {
+-          const targetScroll = Math.max(0, Math.min(state.scroll + positionDiff, maxScroll));
+-          state.scroll = maxScroll;
+-          state.scrollPending = maxScroll;
+-          positionDiff = targetScroll - maxScroll;
+-        }
+-      }
++      // Paseo patch: no Android clamp prediction. Upstream assumes the scroll view clamped to the new
++      // end before React Native adds the anchor's movement, and measures the adjustment from that
++      // end. The app's scroll view defers the clamp until after the adjustment, as iOS does
++      // (modules/paseo-scroll), so an adjustment measured from the end moved a shrinking row's
++      // anchor back to the end instead of keeping it in place.
+       if (shouldQueueNativeMVCPAdjust(dataChanged, state, positionDiff, prevTotalSize, prevScroll, scrollTarget)) {
+         state.pendingNativeMVCPAdjust = {
+           amount: positionDiff,
+@@ -2371,7 +2390,10 @@ function finishScrollTo(ctx) {
      };
      if (scrollingTo.isInitialScroll || state.initialScroll) {
        const isOffsetSession = ((_b = state.initialScrollSession) == null ? void 0 : _b.kind) === "offset";
@@ -147,7 +174,7 @@ index eeb3fb0..5fe940d 100644
        finishInitialScroll(ctx, {
          onFinished: complete,
          preserveTarget: isOffsetSession && state.props.data.length === 0 || shouldPreserveResizeTarget,
-@@ -2515,11 +2541,12 @@ function checkFinishedScrollFallback(ctx) {
+@@ -2515,11 +2537,12 @@ function checkFinishedScrollFallback(ctx) {
          const targetOffset = (_d = (_c = (_b = initialScrollWatchdog.get(state)) == null ? void 0 : _b.targetOffset) != null ? _c : isStillScrollingTo.targetOffset) != null ? _d : 0;
          const jiggleOffset = targetOffset >= SILENT_INITIAL_SCROLL_TARGET_EPSILON ? targetOffset - SILENT_INITIAL_SCROLL_TARGET_EPSILON : targetOffset + SILENT_INITIAL_SCROLL_TARGET_EPSILON;
          initialScrollCompletion.markSilentInitialScrollRetry(state);
@@ -164,7 +191,7 @@ index eeb3fb0..5fe940d 100644
          scheduleFallbackCheck(SILENT_INITIAL_SCROLL_RETRY_DELAY_MS);
        } else if (shouldRetryUnalignedEndScroll) {
          scrollToFallbackOffset(ctx, completionState.clampedTargetOffset, !!isStillScrollingTo.animated);
-@@ -3378,7 +3405,8 @@ function finishBootstrapInitialScrollWithoutScroll(ctx, resolvedOffset) {
+@@ -3378,7 +3401,8 @@ function finishBootstrapInitialScrollWithoutScroll(ctx, resolvedOffset) {
    var _a3;
    const state = ctx.state;
    clearBootstrapInitialScrollSession(state);
@@ -174,7 +201,7 @@ index eeb3fb0..5fe940d 100644
    finishInitialScroll(ctx, {
      preserveTarget: shouldPreserveResizeTarget,
      recalculateItems: true,
-@@ -4569,6 +4597,17 @@ function getIdsInVisibleRange(state, range) {
+@@ -4569,6 +4593,24 @@ function getIdsInVisibleRange(state, range) {
        const id = (_b = state.idCache[i]) != null ? _b : getId(state, i);
        idsInView.push(id);
      }
@@ -188,11 +215,18 @@ index eeb3fb0..5fe940d 100644
 +        const id = (_b = state.idCache[i]) != null ? _b : getId(state, i);
 +        idsInView.push(id);
 +      }
++    }
++    // Paseo patch: the row that starts at or below the end of the viewport goes last. A disclosure
++    // whose trigger sits on the bottom edge of its row holds the row below it, whose top is that
++    // edge, and that row is below the viewport whenever the held row's bottom is.
++    const belowIndex = range.endNoBuffer + 1;
++    if (belowIndex < state.props.data.length) {
++      idsInView.push((_b = state.idCache[belowIndex]) != null ? _b : getId(state, belowIndex));
 +    }
    }
    return idsInView;
  }
-@@ -5176,8 +5215,13 @@ var NATIVE_LAYOUT_MEASUREMENT_EPSILON = 1 / PixelRatio.get() + FLOATING_POINT_SL
+@@ -5176,8 +5218,13 @@ var NATIVE_LAYOUT_MEASUREMENT_EPSILON = 1 / PixelRatio.get() + FLOATING_POINT_SL
  function isWithinEpsilon(delta) {
    return Math.abs(delta) <= NATIVE_LAYOUT_MEASUREMENT_EPSILON;
  }
@@ -208,7 +242,7 @@ index eeb3fb0..5fe940d 100644
  }
  
  // src/core/updateItemSizes.ts
-@@ -5291,6 +5335,11 @@ function updateItemSizesBatch(ctx, measurements) {
+@@ -5291,6 +5338,11 @@ function updateItemSizesBatch(ctx, measurements) {
    const result = {};
    const previousTotalSize = state.totalSize;
    for (const measurement of measurements) {
@@ -220,7 +254,7 @@ index eeb3fb0..5fe940d 100644
      const ownsMeasuredItem = measurement.containerId === void 0 || isContainerItemCurrent(ctx, measurement.containerId, measurement.itemKey);
      if (ownsMeasuredItem) {
        const index = state.indexByKey.get(measurement.itemKey);
-@@ -5414,7 +5463,7 @@ function updateOneItemSize(ctx, itemKey, sizeObj, resolvedMeasurementItem, notif
+@@ -5414,7 +5466,7 @@ function updateOneItemSize(ctx, itemKey, sizeObj, resolvedMeasurementItem, notif
      state.fixedItemSizeKeys.add(itemKey);
    }
    const prevSizeKnown = sizesKnown.get(itemKey);
@@ -229,7 +263,7 @@ index eeb3fb0..5fe940d 100644
      return 0;
    }
    const size = Platform.OS === "web" ? Math.round(rawSize) : roundSize(rawSize);
-@@ -6283,6 +6332,10 @@ function useFreshDataTransitionVisibility(readyToRender, transitionEpoch) {
+@@ -6283,6 +6335,10 @@ function useFreshDataTransitionVisibility(readyToRender, transitionEpoch) {
  
  // src/components/Containers.native.tsx
  var useCommittedMinSize = IsNewArchitecture ? () => useArr$(["totalSize"])[0] : () => void 0;
@@ -240,7 +274,7 @@ index eeb3fb0..5fe940d 100644
  var ContainersLayer = typedMemo(function ContainersLayer2({
    children,
    freshDataTransitionEpoch,
-@@ -6290,7 +6343,7 @@ var ContainersLayer = typedMemo(function ContainersLayer2({
+@@ -6290,7 +6346,7 @@ var ContainersLayer = typedMemo(function ContainersLayer2({
  }) {
    const ctx = useStateContext();
    const columnWrapperStyle = ctx.columnWrapperStyle;
@@ -249,7 +283,7 @@ index eeb3fb0..5fe940d 100644
    const minSize = useCommittedMinSize();
    const [readyToRender, numColumns, otherAxisSize = 0] = useArr$(["readyToRender", "numColumns", "otherAxisSize"]);
    const isVisible = useFreshDataTransitionVisibility(!!readyToRender, freshDataTransitionEpoch);
-@@ -6513,6 +6566,14 @@ var AlignItemsAtEndSpacer = typedMemo(function AlignItemsAtEndSpacer2({ horizont
+@@ -6513,6 +6569,14 @@ var AlignItemsAtEndSpacer = typedMemo(function AlignItemsAtEndSpacer2({ horizont
      null
    );
  });
@@ -264,7 +298,7 @@ index eeb3fb0..5fe940d 100644
  var ListComponent = typedMemo(function ListComponent2({
    canRender,
    freshDataTransitionEpoch,
-@@ -6651,7 +6712,7 @@ var ListComponent = typedMemo(function ListComponent2({
+@@ -6651,7 +6715,7 @@ var ListComponent = typedMemo(function ListComponent2({
        style: autoOtherAxisStyle ? [autoOtherAxisStyle, style] : style
      },
      /* @__PURE__ */ React2__namespace.createElement(ScrollAdjust, null),
@@ -273,7 +307,7 @@ index eeb3fb0..5fe940d 100644
      ListEmptyComponent && getComponent(ListEmptyComponent),
      shouldRenderAlignItemsAtEndSpacer && /* @__PURE__ */ React2__namespace.createElement(AlignItemsAtEndSpacer, { horizontal }),
      canRender && !ListEmptyComponent && /* @__PURE__ */ React2__namespace.createElement(
-@@ -6665,7 +6726,7 @@ var ListComponent = typedMemo(function ListComponent2({
+@@ -6665,7 +6729,7 @@ var ListComponent = typedMemo(function ListComponent2({
          stickyHeaderConfig
        }
      ),
@@ -282,7 +316,7 @@ index eeb3fb0..5fe940d 100644
      IS_DEV && ENABLE_DEVMODE
    );
  });
-@@ -7886,7 +7947,10 @@ function setPaddingTop(ctx, { stylePaddingTop }) {
+@@ -7886,7 +7950,10 @@ function setPaddingTop(ctx, { stylePaddingTop }) {
    const state = ctx.state;
    if (stylePaddingTop !== void 0) {
      const prevStylePaddingTop = peek$(ctx, "stylePaddingTop") || 0;
@@ -294,7 +328,7 @@ index eeb3fb0..5fe940d 100644
        let prevTotalSize = peek$(ctx, "totalSize") || 0;
        set$(ctx, "totalSize", prevTotalSize + prevStylePaddingTop);
        state.timeoutSetPaddingTop = setTimeout(() => {
-@@ -7989,6 +8053,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
+@@ -7989,6 +8056,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
      estimatedListSize,
      extraData,
      getFixedItemSize,
@@ -302,7 +336,7 @@ index eeb3fb0..5fe940d 100644
      getItemType,
      horizontal,
      rtl,
-@@ -8282,6 +8347,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
+@@ -8282,6 +8350,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
      dataVersion,
      drawDistance,
      estimatedItemSize,
@@ -311,7 +345,7 @@ index eeb3fb0..5fe940d 100644
      getItemType: useWrapIfItem(getItemType),
      hasExternalScroll,
 diff --git a/node_modules/@legendapp/list/react-native.mjs b/node_modules/@legendapp/list/react-native.mjs
-index fe5f415..7f1b9dd 100644
+index fe5f415..9788ce6 100644
 --- a/node_modules/@legendapp/list/react-native.mjs
 +++ b/node_modules/@legendapp/list/react-native.mjs
 @@ -1476,7 +1476,15 @@ function getItemSize(ctx, key, index, data, useAverageSize, preferCachedSize, no
@@ -398,7 +432,28 @@ index fe5f415..7f1b9dd 100644
      }
      if (dataChanged && idsInView.length > 0 && state.didContainersLayout) {
        for (let i = 0; i < idsInView.length; i++) {
-@@ -2350,7 +2373,10 @@ function finishScrollTo(ctx) {
+@@ -1853,15 +1876,11 @@ function prepareMVCP(ctx, dataChanged, adjustmentSource) {
+         now,
+         positionDiff
+       });
+-      if (Platform.OS === "android" && !dataChanged && scrollTarget === void 0) {
+-        const maxScroll = getScrollRangeEnd(ctx);
+-        if (state.scroll > maxScroll + MVCP_POSITION_EPSILON) {
+-          const targetScroll = Math.max(0, Math.min(state.scroll + positionDiff, maxScroll));
+-          state.scroll = maxScroll;
+-          state.scrollPending = maxScroll;
+-          positionDiff = targetScroll - maxScroll;
+-        }
+-      }
++      // Paseo patch: no Android clamp prediction. Upstream assumes the scroll view clamped to the new
++      // end before React Native adds the anchor's movement, and measures the adjustment from that
++      // end. The app's scroll view defers the clamp until after the adjustment, as iOS does
++      // (modules/paseo-scroll), so an adjustment measured from the end moved a shrinking row's
++      // anchor back to the end instead of keeping it in place.
+       if (shouldQueueNativeMVCPAdjust(dataChanged, state, positionDiff, prevTotalSize, prevScroll, scrollTarget)) {
+         state.pendingNativeMVCPAdjust = {
+           amount: positionDiff,
+@@ -2350,7 +2369,10 @@ function finishScrollTo(ctx) {
      };
      if (scrollingTo.isInitialScroll || state.initialScroll) {
        const isOffsetSession = ((_b = state.initialScrollSession) == null ? void 0 : _b.kind) === "offset";
@@ -410,7 +465,7 @@ index fe5f415..7f1b9dd 100644
        finishInitialScroll(ctx, {
          onFinished: complete,
          preserveTarget: isOffsetSession && state.props.data.length === 0 || shouldPreserveResizeTarget,
-@@ -2494,11 +2520,12 @@ function checkFinishedScrollFallback(ctx) {
+@@ -2494,11 +2516,12 @@ function checkFinishedScrollFallback(ctx) {
          const targetOffset = (_d = (_c = (_b = initialScrollWatchdog.get(state)) == null ? void 0 : _b.targetOffset) != null ? _c : isStillScrollingTo.targetOffset) != null ? _d : 0;
          const jiggleOffset = targetOffset >= SILENT_INITIAL_SCROLL_TARGET_EPSILON ? targetOffset - SILENT_INITIAL_SCROLL_TARGET_EPSILON : targetOffset + SILENT_INITIAL_SCROLL_TARGET_EPSILON;
          initialScrollCompletion.markSilentInitialScrollRetry(state);
@@ -427,7 +482,7 @@ index fe5f415..7f1b9dd 100644
          scheduleFallbackCheck(SILENT_INITIAL_SCROLL_RETRY_DELAY_MS);
        } else if (shouldRetryUnalignedEndScroll) {
          scrollToFallbackOffset(ctx, completionState.clampedTargetOffset, !!isStillScrollingTo.animated);
-@@ -3357,7 +3384,8 @@ function finishBootstrapInitialScrollWithoutScroll(ctx, resolvedOffset) {
+@@ -3357,7 +3380,8 @@ function finishBootstrapInitialScrollWithoutScroll(ctx, resolvedOffset) {
    var _a3;
    const state = ctx.state;
    clearBootstrapInitialScrollSession(state);
@@ -437,7 +492,7 @@ index fe5f415..7f1b9dd 100644
    finishInitialScroll(ctx, {
      preserveTarget: shouldPreserveResizeTarget,
      recalculateItems: true,
-@@ -4548,6 +4576,17 @@ function getIdsInVisibleRange(state, range) {
+@@ -4548,6 +4572,24 @@ function getIdsInVisibleRange(state, range) {
        const id = (_b = state.idCache[i]) != null ? _b : getId(state, i);
        idsInView.push(id);
      }
@@ -451,11 +506,18 @@ index fe5f415..7f1b9dd 100644
 +        const id = (_b = state.idCache[i]) != null ? _b : getId(state, i);
 +        idsInView.push(id);
 +      }
++    }
++    // Paseo patch: the row that starts at or below the end of the viewport goes last. A disclosure
++    // whose trigger sits on the bottom edge of its row holds the row below it, whose top is that
++    // edge, and that row is below the viewport whenever the held row's bottom is.
++    const belowIndex = range.endNoBuffer + 1;
++    if (belowIndex < state.props.data.length) {
++      idsInView.push((_b = state.idCache[belowIndex]) != null ? _b : getId(state, belowIndex));
 +    }
    }
    return idsInView;
  }
-@@ -5155,8 +5194,13 @@ var NATIVE_LAYOUT_MEASUREMENT_EPSILON = 1 / PixelRatio.get() + FLOATING_POINT_SL
+@@ -5155,8 +5197,13 @@ var NATIVE_LAYOUT_MEASUREMENT_EPSILON = 1 / PixelRatio.get() + FLOATING_POINT_SL
  function isWithinEpsilon(delta) {
    return Math.abs(delta) <= NATIVE_LAYOUT_MEASUREMENT_EPSILON;
  }
@@ -471,7 +533,7 @@ index fe5f415..7f1b9dd 100644
  }
  
  // src/core/updateItemSizes.ts
-@@ -5270,6 +5314,11 @@ function updateItemSizesBatch(ctx, measurements) {
+@@ -5270,6 +5317,11 @@ function updateItemSizesBatch(ctx, measurements) {
    const result = {};
    const previousTotalSize = state.totalSize;
    for (const measurement of measurements) {
@@ -483,7 +545,7 @@ index fe5f415..7f1b9dd 100644
      const ownsMeasuredItem = measurement.containerId === void 0 || isContainerItemCurrent(ctx, measurement.containerId, measurement.itemKey);
      if (ownsMeasuredItem) {
        const index = state.indexByKey.get(measurement.itemKey);
-@@ -5393,7 +5442,7 @@ function updateOneItemSize(ctx, itemKey, sizeObj, resolvedMeasurementItem, notif
+@@ -5393,7 +5445,7 @@ function updateOneItemSize(ctx, itemKey, sizeObj, resolvedMeasurementItem, notif
      state.fixedItemSizeKeys.add(itemKey);
    }
    const prevSizeKnown = sizesKnown.get(itemKey);
@@ -492,7 +554,7 @@ index fe5f415..7f1b9dd 100644
      return 0;
    }
    const size = Platform.OS === "web" ? Math.round(rawSize) : roundSize(rawSize);
-@@ -6262,6 +6311,10 @@ function useFreshDataTransitionVisibility(readyToRender, transitionEpoch) {
+@@ -6262,6 +6314,10 @@ function useFreshDataTransitionVisibility(readyToRender, transitionEpoch) {
  
  // src/components/Containers.native.tsx
  var useCommittedMinSize = IsNewArchitecture ? () => useArr$(["totalSize"])[0] : () => void 0;
@@ -503,7 +565,7 @@ index fe5f415..7f1b9dd 100644
  var ContainersLayer = typedMemo(function ContainersLayer2({
    children,
    freshDataTransitionEpoch,
-@@ -6269,7 +6322,7 @@ var ContainersLayer = typedMemo(function ContainersLayer2({
+@@ -6269,7 +6325,7 @@ var ContainersLayer = typedMemo(function ContainersLayer2({
  }) {
    const ctx = useStateContext();
    const columnWrapperStyle = ctx.columnWrapperStyle;
@@ -512,7 +574,7 @@ index fe5f415..7f1b9dd 100644
    const minSize = useCommittedMinSize();
    const [readyToRender, numColumns, otherAxisSize = 0] = useArr$(["readyToRender", "numColumns", "otherAxisSize"]);
    const isVisible = useFreshDataTransitionVisibility(!!readyToRender, freshDataTransitionEpoch);
-@@ -6492,6 +6545,14 @@ var AlignItemsAtEndSpacer = typedMemo(function AlignItemsAtEndSpacer2({ horizont
+@@ -6492,6 +6548,14 @@ var AlignItemsAtEndSpacer = typedMemo(function AlignItemsAtEndSpacer2({ horizont
      null
    );
  });
@@ -527,7 +589,7 @@ index fe5f415..7f1b9dd 100644
  var ListComponent = typedMemo(function ListComponent2({
    canRender,
    freshDataTransitionEpoch,
-@@ -6630,7 +6691,7 @@ var ListComponent = typedMemo(function ListComponent2({
+@@ -6630,7 +6694,7 @@ var ListComponent = typedMemo(function ListComponent2({
        style: autoOtherAxisStyle ? [autoOtherAxisStyle, style] : style
      },
      /* @__PURE__ */ React2.createElement(ScrollAdjust, null),
@@ -536,7 +598,7 @@ index fe5f415..7f1b9dd 100644
      ListEmptyComponent && getComponent(ListEmptyComponent),
      shouldRenderAlignItemsAtEndSpacer && /* @__PURE__ */ React2.createElement(AlignItemsAtEndSpacer, { horizontal }),
      canRender && !ListEmptyComponent && /* @__PURE__ */ React2.createElement(
-@@ -6644,7 +6705,7 @@ var ListComponent = typedMemo(function ListComponent2({
+@@ -6644,7 +6708,7 @@ var ListComponent = typedMemo(function ListComponent2({
          stickyHeaderConfig
        }
      ),
@@ -545,7 +607,7 @@ index fe5f415..7f1b9dd 100644
      IS_DEV && ENABLE_DEVMODE
    );
  });
-@@ -7865,7 +7926,10 @@ function setPaddingTop(ctx, { stylePaddingTop }) {
+@@ -7865,7 +7929,10 @@ function setPaddingTop(ctx, { stylePaddingTop }) {
    const state = ctx.state;
    if (stylePaddingTop !== void 0) {
      const prevStylePaddingTop = peek$(ctx, "stylePaddingTop") || 0;
@@ -557,7 +619,7 @@ index fe5f415..7f1b9dd 100644
        let prevTotalSize = peek$(ctx, "totalSize") || 0;
        set$(ctx, "totalSize", prevTotalSize + prevStylePaddingTop);
        state.timeoutSetPaddingTop = setTimeout(() => {
-@@ -7968,6 +8032,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
+@@ -7968,6 +8035,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
      estimatedListSize,
      extraData,
      getFixedItemSize,
@@ -565,7 +627,7 @@ index fe5f415..7f1b9dd 100644
      getItemType,
      horizontal,
      rtl,
-@@ -8261,6 +8326,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
+@@ -8261,6 +8329,7 @@ var LegendListInner = typedForwardRef(function LegendListInner2(props, forwarded
      dataVersion,
      drawDistance,
      estimatedItemSize,

public-docs/plugins/hooks.md

@@ -94,7 +94,7 @@ type PluginTurnOutcome =
 
 | Field        | Shape / meaning                                                                        |
 | ------------ | -------------------------------------------------------------------------------------- |
-| `turnId`     | Provider-reported `string` or `null`; can repeat after a session reopens               |
+| `turnId`     | Provider-reported `string` or `null`; unique across resumed sessions                   |
 | `timeline`   | Deprecated, always `[]`; fetch history with the SDK.                                   |
 | `request`    | SDK `AgentPermissionRequest`; `kind` is `tool`, `plan`, `question`, `mode`, or `other` |
 | `resolution` | SDK `AgentPermissionResponse`                                                          |

public-docs/plugins/hooks.zh-CN.md

@@ -92,7 +92,7 @@ type PluginTurnOutcome =
 
 | 字段         | 结构 / 含义                                                                              |
 | ------------ | ---------------------------------------------------------------------------------------- |
-| `turnId`     | Provider 报告的 `string` 或 `null`；会话重新打开后可能重复                               |
+| `turnId`     | Provider 报告的 `string` 或 `null`；在会话恢复后仍保持唯一                               |
 | `timeline`   | 已弃用，始终为 `[]`；通过 SDK 获取历史。                                                 |
 | `request`    | SDK 的 `AgentPermissionRequest`；`kind` 为 `tool`、`plan`、`question`、`mode` 或 `other` |
 | `resolution` | SDK 的 `AgentPermissionResponse`                                                         |