cmux-next: a chat's first frame waits on one daemon round trip

cmux-tui/crates/acpmux/spec/acpmux-protocol.json

@@ -1789,6 +1789,17 @@
         },
         "session": {
           "$ref": "#/types/SessionDetail"
+        },
+        "turnOpener": {
+          "anyOf": [
+            {
+              "$ref": "#/types/SessionEvent"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "description": "The prompt that opened the turn the page starts inside, when the page starts inside one\nwith older records before it (`server/requests/turn_opener.rs`); null otherwise. An older\nacpmux leaves it out, and its client reads it with `_acpmux/events`."
         }
       },
       "required": [
@@ -1797,7 +1808,8 @@
         "lastSeq",
         "pageThrough",
         "presence",
-        "session"
+        "session",
+        "turnOpener"
       ],
       "type": "object"
     },

cmux-tui/crates/acpmux/src/protocol/events.rs

@@ -98,6 +98,10 @@ pub struct AttachResult {
     pub page_through: u64,
     /// Who has the chat open now, this attachment included (`hub/presence.rs`).
     pub presence: super::presence::PresenceNote,
+    /// The prompt that opened the turn the page starts inside, when the page starts inside one
+    /// with older records before it (`server/requests/turn_opener.rs`); null otherwise. An older
+    /// acpmux leaves it out, and its client reads it with `_acpmux/events`.
+    pub turn_opener: Option<SessionEvent>,
 }
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]

cmux-tui/crates/acpmux/src/server/requests.rs

@@ -12,6 +12,7 @@ mod mailbox;
 pub(super) mod person_stop;
 mod presets;
 mod successor;
+mod turn_opener;
 #[path = "watches.rs"]
 mod watches;
 
@@ -2014,11 +2015,13 @@ async fn dispatch_request(
             } else {
                 events::newest_seq(&page.events)
             };
+            // The prompt of the turn the page starts inside, read with it (`turn_opener.rs`).
+            let opener = turn_opener::turn_opener(hub, &s.id, &page);
             let events: Vec<Value> = events::values(&s.id, page.events);
             // Who has the chat open now (`hub/presence.rs`), this attachment included.
             let presence = hub.presence.snapshot(&s.id);
             Ok(
-                json!({"session": detail, "events": events, "hasMore": page.has_more, "lastSeq": last, "pageThrough": through, "presence": presence}),
+                json!({"session": detail, "events": events, "hasMore": page.has_more, "lastSeq": last, "pageThrough": through, "presence": presence, "turnOpener": opener}),
             )
         }
         method::MUX_DETACH => {

cmux-tui/crates/acpmux/src/server/requests/turn_opener.rs

@@ -0,0 +1,73 @@
+//! The prompt that opened the turn an attach page starts inside (`_acpmux/attach`'s
+//! `turnOpener`). A capped page of a long turn holds none of the turn's prompt; read with the page
+//! here, the pane opening the chat shows what the turn was asked without a round trip of its own
+//! after the attach (its `fetchTurnOpener`, which an older acpmux still leaves it).
+
+use super::*;
+
+/// The user messages read back before the page: steers into the turn are user messages too, and
+/// the opener is the newest that is not one. The pane's own read asks the same.
+const PROMPTS_READ: usize = 16;
+
+/// The opener as an `_acpmux/event` value, `None` when the page needs none (it starts a turn, or
+/// holds the log from its start) or the read back finds none.
+pub(super) fn turn_opener(
+    hub: &Hub,
+    session_id: &str,
+    page: &crate::hub::EventPage,
+) -> Option<Value> {
+    if !page.has_more {
+        return None;
+    }
+    let first = page.events.first()?;
+    if first.seq <= 1 || starts_turn(first) {
+        return None;
+    }
+    let filter = crate::hub::EventFilter::parse(Some(&json!(["user_message"]))).ok()?;
+    let prompts = hub
+        .events_page(
+            session_id,
+            0,
+            Some(first.seq),
+            PROMPTS_READ,
+            Some(crate::hub::page_bytes(&Value::Null)),
+            true,
+            &filter,
+        )
+        .ok()?;
+    let opener = prompts.events.into_iter().rev().find(starts_turn)?;
+    Some(into_event_value(session_id, opener))
+}
+
+/// A prompt that starts a turn: acpmux's own `user_message`, not a steer into a running turn.
+fn starts_turn(rec: &crate::store::EventRecord) -> bool {
+    rec.dir == "mux"
+        && rec.kind == "user_message"
+        && rec.msg.get("steer") != Some(&Value::Bool(true))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn rec(seq: u64, kind: &str, msg: Value) -> crate::store::EventRecord {
+        crate::store::EventRecord {
+            seq,
+            at: 0,
+            dir: "mux".into(),
+            kind: kind.into(),
+            msg,
+            host_seq: None,
+        }
+    }
+
+    #[test]
+    fn a_steer_does_not_start_a_turn() {
+        assert!(starts_turn(&rec(3, "user_message", json!({"text": "go"}))));
+        assert!(!starts_turn(&rec(4, "user_message", json!({"text": "also", "steer": true}))));
+        assert!(!starts_turn(&rec(5, "turn_result", json!({}))));
+        let mut agent = rec(6, "user_message", json!({}));
+        agent.dir = "in".into();
+        assert!(!starts_turn(&agent));
+    }
+}

cmux-tui/crates/acpmux/tests/hub_integration/transcript_events.rs

@@ -739,3 +739,46 @@ async fn orchestration_session_changed_carries_effort() {
         c.wait_for(method::MUX_SESSION_CHANGED, |p| p["session"]["effort"] == "high").await;
     assert_eq!(changed["sessionId"], id);
 }
+
+/// An attach page that starts inside a turn names the prompt that opened it (`turnOpener`), the
+/// record a read of the user messages before the page finds: the pane shows what the turn was
+/// asked with no read of its own after the attach. A page that starts a turn, or holds the log
+/// from its start, names none.
+#[tokio::test]
+async fn an_attach_page_names_the_prompt_of_the_turn_it_starts_inside() {
+    let (hub, mut c) = setup(PermissionPolicy::ApproveAll).await;
+    let id = new_session(&mut c, "opener").await;
+    for text in ["one", "two", "three"] {
+        c.request(method::SESSION_PROMPT, prompt(&id, text, None)).await.unwrap();
+    }
+    let mut a = connect(&hub).await;
+    let attach = |limit: u64| json!({"sessionId": id, "kinds": ["transcript"], "limit": limit});
+    let whole = a.request(method::MUX_ATTACH, attach(10_000)).await.unwrap();
+    assert_eq!(whole["hasMore"], false);
+    assert_eq!(whole["turnOpener"], Value::Null, "{whole}");
+    let starts_turn =
+        |e: &Value| e["dir"] == "mux" && e["kind"] == "user_message" && e["msg"]["steer"] != true;
+    let transcript = whole["events"].as_array().unwrap().clone();
+    let last_prompt = transcript.iter().rposition(starts_turn).expect("a prompt in the log");
+    // Pages that start right after the newest prompt, inside its turn.
+    let inside = transcript.len() - last_prompt - 1;
+    assert!(inside >= 1, "the last turn has records after its prompt: {whole}");
+    for limit in 1..=inside as u64 {
+        let page = a.request(method::MUX_ATTACH, attach(limit)).await.unwrap();
+        let first = page["events"][0]["seq"].as_u64().unwrap();
+        let read = a
+            .request(
+                method::MUX_EVENTS,
+                json!({"sessionId": id, "beforeSeq": first, "limit": 16, "kinds": ["user_message"]}),
+            )
+            .await
+            .unwrap();
+        let expected =
+            read["events"].as_array().unwrap().iter().rev().find(|e| starts_turn(e)).cloned();
+        assert_eq!(page["turnOpener"], expected.unwrap(), "limit {limit}: {page}");
+    }
+    // The page that starts with the prompt itself needs none.
+    let at_prompt = a.request(method::MUX_ATTACH, attach(inside as u64 + 1)).await.unwrap();
+    assert!(starts_turn(&at_prompt["events"][0]), "{at_prompt}");
+    assert_eq!(at_prompt["turnOpener"], Value::Null);
+}

webviews/src/agent-session/acpmux/chatOpenTrips.test.ts

@@ -0,0 +1,167 @@
+// A chat's first frame waits on one daemon round trip after the handshake, not one per read (Leo,
+// 2026-10-29: "as in i want ours to load fast too :("). The open used to ask, one after another,
+// `_acpmux/status`, the chat list, the attach, and then the turn opener, permissions and handoff:
+// five waits on a busy daemon before the first row. These tests hold every reply until the test
+// answers the requests out as one wave, and count the waves before the first frame with rows.
+import { afterAll, beforeAll, expect, test } from "bun:test";
+import { JSDOM } from "jsdom";
+
+const dom = new JSDOM("<!doctype html>", { url: "http://localhost/" });
+const scope = globalThis as Record<string, unknown>;
+const saved = {
+  window: "window" in scope,
+  document: "document" in scope,
+  values: { window: scope.window, document: scope.document },
+};
+
+beforeAll(() => {
+  scope.window = dom.window;
+  scope.document = dom.window.document;
+});
+afterAll(() => {
+  for (const key of ["window", "document"] as const)
+    if (saved[key]) scope[key] = saved.values[key];
+    else delete scope[key];
+});
+
+type Daemon = {
+  emit: (target: string, step: unknown, at?: number) => void;
+  send: (raw: string) => void;
+};
+type Sent = { method: string; wave: number; params?: Record<string, unknown> };
+
+const turns = () => new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+/// Opens `mockHost`'s chat on `socket` with every reply held, answering what is out one wave at a
+/// time. Answers the waves the first frame with rows waited on, and every request with its wave.
+async function openCountingWaves(
+  socket: unknown,
+  rewrite: (method: string, reply: Record<string, unknown>) => void = () => {},
+): Promise<{ waves: number; sent: Sent[]; close: () => void }> {
+  const { AcpmuxDirectClient } = await import("./direct");
+  const { mockHost } = await import("./mock");
+  // The mock opened before the page asked (`open`, below): the page is told so once it listens.
+  await turns();
+  const daemon = socket as Daemon & {
+    onmessage: ((message: { data: string }) => void) | null;
+    onopen: (() => void) | null;
+  };
+  const answer = daemon.send.bind(socket);
+  const held: string[] = [];
+  const sent: Sent[] = [];
+  const methods = new Map<number, string>();
+  let wave = 0;
+  daemon.send = (raw) => {
+    const request = JSON.parse(raw) as { id?: number; method: string; params?: Record<string, unknown> };
+    if (request.id === undefined) return answer(raw);
+    methods.set(request.id, request.method);
+    sent.push({ method: request.method, wave, params: request.params });
+    held.push(raw);
+  };
+  let painted = false;
+  const connecting = AcpmuxDirectClient.connect(
+    mockHost,
+    (snapshot) => {
+      const rows = (snapshot as { rows?: unknown[] }).rows;
+      if (Array.isArray(rows) && rows.length > 0) painted = true;
+    },
+    undefined,
+    () => {
+      // The page's own handler, with each reply rewritten as the test asks (an older daemon).
+      queueMicrotask(() => {
+        daemon.onopen?.();
+        const page = daemon.onmessage;
+        daemon.onmessage = (message) => {
+          const parsed = JSON.parse(message.data) as { id?: number; result?: Record<string, unknown> };
+          const method = parsed.id === undefined ? undefined : methods.get(parsed.id);
+          if (method && parsed.result && typeof parsed.result === "object") {
+            rewrite(method, parsed.result);
+            page?.({ data: JSON.stringify(parsed) });
+          } else page?.(message);
+        };
+      });
+      return socket as WebSocket;
+    },
+  );
+  // The host's tail read (none outside the app) and the socket's open settle with no reply out.
+  for (let idle = 0; !painted && wave < 12; ) {
+    for (let i = 0; i < 20; i += 1) await turns();
+    if (painted) break;
+    if (held.length === 0) {
+      // Nothing out: what the page waits on is not the daemon (at most the host's 250 ms race).
+      idle += 1;
+      if (idle > 40) break;
+      await new Promise((resolve) => setTimeout(resolve, 10));
+      continue;
+    }
+    wave += 1;
+    for (const raw of held.splice(0)) answer(raw);
+  }
+  // Everything after the first frame answers at once, so the open completes.
+  daemon.send = answer;
+  for (const raw of held.splice(0)) answer(raw);
+  const client = await connecting;
+  expect(painted).toBe(true);
+  return { waves: wave, sent, close: () => client.close() };
+}
+
+test("a chat's first frame waits on the handshake and one round trip", async () => {
+  const { MockAcpmuxSocket } = await import("./mock");
+  const socket = new MockAcpmuxSocket(() => Promise.resolve());
+  const { waves, sent, close } = await openCountingWaves(socket);
+  try {
+    const before = sent.filter((request) => request.wave < waves).map((request) => `${request.wave} ${request.method}`);
+    // Wave 0: initialize alone (the app's bridge sends the person proof right after it). Wave 1:
+    // the machines, the list, the attach and the chat's own reads, side by side.
+    expect(waves).toBeLessThanOrEqual(2);
+    expect(before[0]).toBe("0 initialize");
+    const second = sent.filter((request) => request.wave === 1).map((request) => request.method);
+    expect(second).toEqual(expect.arrayContaining(["_acpmux/status", "_acpmux/watch", "_acpmux/attach"]));
+  } finally {
+    close();
+  }
+});
+
+/// A chat whose newest page starts inside a long turn: its prompt, then more records than a page holds.
+function longTurn(socket: unknown, id: string): void {
+  const daemon = socket as Daemon;
+  daemon.emit(id, { mux: "user_message", msg: { text: "Port the parser", promptId: "p-long" } });
+  for (let index = 0; index < 450; index += 1)
+    daemon.emit(id, {
+      update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: `step ${index} ` } },
+    });
+}
+
+test("the turn's opener comes with the attach page, not a read after it", async () => {
+  const { MockAcpmuxSocket, mockHost } = await import("./mock");
+  const socket = new MockAcpmuxSocket(() => Promise.resolve());
+  // Logged before any page listens, so no page hears it live.
+  await turns();
+  longTurn(socket, mockHost.sessionId!);
+  const { waves, sent, close } = await openCountingWaves(socket);
+  try {
+    expect(waves).toBeLessThanOrEqual(2);
+    // No `_acpmux/events` read of the turn's prompts: the attach named it.
+    expect(sent.filter((request) => request.method === "_acpmux/events")).toEqual([]);
+  } finally {
+    close();
+  }
+});
+
+test("an acpmux that does not name the opener still has it read", async () => {
+  const { MockAcpmuxSocket, mockHost } = await import("./mock");
+  const socket = new MockAcpmuxSocket(() => Promise.resolve());
+  // Logged before any page listens, so no page hears it live.
+  await turns();
+  longTurn(socket, mockHost.sessionId!);
+  const { sent, close } = await openCountingWaves(socket, (method, reply) => {
+    if (method === "_acpmux/attach") delete reply.turnOpener;
+  });
+  try {
+    const reads = sent.filter((request) => request.method === "_acpmux/events");
+    expect(reads.length).toBe(1);
+    expect(reads[0]!.params).toMatchObject({ kinds: ["user_message"], limit: 16 });
+  } finally {
+    close();
+  }
+});