From 93fbd17f4f47018e56a1e2679e7e4f21bc910c09 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 11:00:52 +0000 Subject: [PATCH 1/2] Chief: tests for an acpmux session list read in pages acpmux answers `_acpmux/sessions` and `_acpmux/watch` with no `limit` in pages, each with the `nextCursor` of the rest, when the whole list is too large for one message. A fake daemon answers in two pages: the link comes up on both, a session only the second page lists is found, and a cursor the daemon answers with again is refused. These fail on the Chief as it is, which reads the first page only. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-c11d09c9f41988fc1d93 --- .../tests/acpmux_session_info.rs | 136 ++++++++++++++++++ 1 file changed, 136 insertions(+) diff --git a/Native/OptChat/optchat-chief/tests/acpmux_session_info.rs b/Native/OptChat/optchat-chief/tests/acpmux_session_info.rs index e61be3b5a772..5e8a94cb9bae 100644 --- a/Native/OptChat/optchat-chief/tests/acpmux_session_info.rs +++ b/Native/OptChat/optchat-chief/tests/acpmux_session_info.rs @@ -213,3 +213,139 @@ fn the_list_answers_only_what_info_cannot() { ); assert_eq!(lists(), 4); } + +/// A daemon whose session list is too large for one reply (acpmux pages a +/// list with no `limit` past its largest message): the first page holds `s-1` +/// and the `nextCursor` of the rest, read by `_acpmux/sessions` from it; the +/// rest holds `p-2`, a peer's session whose link is down, so `_acpmux/info` +/// refuses it and only the list knows it. With `cursor_repeats`, the rest +/// answers with its own cursor again (a list that would never end) and the +/// watch answers with the first page alone and no cursor, as a list that fits. +fn serve_pages(listener: UnixListener, requests: Arc>>, cursor_repeats: bool) { + std::thread::spawn(move || { + for conn in listener.incoming().flatten() { + let requests = requests.clone(); + std::thread::spawn(move || { + let mut out = conn.try_clone().unwrap(); + for line in BufReader::new(conn).lines() { + let Ok(line) = line else { return }; + let req: Value = serde_json::from_str(&line).unwrap(); + requests.lock().unwrap().push(req.clone()); + let id = req["id"].clone(); + let ok = |result: Value| json!({"jsonrpc": "2.0", "id": id, "result": result}); + let row = |id: &str, name: &str| json!({"sessionId": id, "name": name, "harness": "codex", "status": "idle"}); + let first = json!({"sessions": [row("s-1", "one")], "nextCursor": "c1"}); + let mut rest = json!({"sessions": [row("p-2", "builder/two")]}); + if cursor_repeats { + rest["nextCursor"] = json!("c1"); + } + let reply = match req["method"].as_str().unwrap() { + "_acpmux/watch" if cursor_repeats => { + ok(json!({"sessions": [row("s-1", "one")]})) + } + "_acpmux/watch" => ok(first), + "_acpmux/sessions" => match req["params"]["cursor"].as_str() { + None => ok(first), + Some("c1") => ok(rest), + Some(other) => { + json!({"jsonrpc": "2.0", "id": id, "error": {"code": -32602, "message": format!("no cursor {other:?}")}}) + } + }, + "_acpmux/info" => { + json!({"jsonrpc": "2.0", "id": id, "error": {"code": -32000, "message": "peer builder is not connected"}}) + } + "_acpmux/harnesses" => ok(common::catalog()), + _ => ok(json!({})), + }; + if req.get("id").is_some() { + writeln!(out, "{reply}").unwrap(); + } + } + }); + } + }); +} + +/// `serve_pages` linked: the link's acpmux once it is up, the list it came up +/// on, and every request the daemon read. +fn link_pages( + dir: &std::path::Path, + cursor_repeats: bool, +) -> (Arc, Vec, Arc>>) { + let socket = dir.join("acpmux.sock"); + let requests = Arc::new(Mutex::new(Vec::new())); + serve_pages( + UnixListener::bind(&socket).unwrap(), + requests.clone(), + cursor_repeats, + ); + let acpmux = Acpmux::new(socket, None, Vec::new()); + let (tx, rx) = channel(); + let sink_tx = Mutex::new(tx); + acpmux.spawn_link( + Arc::new(move |e| { + let _ = sink_tx.lock().unwrap().send(e); + }), + Arc::new(|_: &str| {}), + ); + let up = loop { + if let AgentEvent::Up(list) = rx.recv_timeout(common::WAIT).unwrap() { + break list; + } + }; + let ids = up.into_iter().map(|s| s.session_id).collect(); + (acpmux, ids, requests) +} + +/// The `_acpmux/sessions` requests the daemon read, by their params. +fn list_params(requests: &Mutex>) -> Vec { + requests + .lock() + .unwrap() + .iter() + .filter(|r| r["method"] == "_acpmux/sessions") + .map(|r| r["params"].clone()) + .collect() +} + +/// A watch list too large for one reply: the link comes up on every page of +/// it, the rest read from the watch's cursor with the watch's params less +/// `enabled`. +#[test] +fn the_link_comes_up_on_every_page_of_the_watch_list() { + let dir = tempfile::tempdir().unwrap(); + let (_acpmux, up, requests) = link_pages(dir.path(), false); + assert_eq!(up, ["s-1", "p-2"]); + assert_eq!(list_params(&requests), [json!({"cursor": "c1"})]); +} + +/// A session on a later page of the list is found when only the list knows it. +#[test] +fn a_session_on_a_later_page_of_the_list_is_found() { + let dir = tempfile::tempdir().unwrap(); + let (acpmux, _, requests) = link_pages(dir.path(), false); + requests.lock().unwrap().clear(); + let two = acpmux.session("p-2").unwrap().expect("listed on page two"); + assert_eq!(two.name, "builder/two"); + assert_eq!(acpmux.find("builder/two").unwrap().as_deref(), Some("p-2")); + assert_eq!( + list_params(&requests), + [ + json!({}), + json!({"cursor": "c1"}), + json!({}), + json!({"cursor": "c1"}) + ] + ); +} + +/// A cursor the daemon answers with again is refused, never read in a loop. +#[test] +fn a_cursor_that_comes_again_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let (acpmux, up, requests) = link_pages(dir.path(), true); + assert_eq!(up, ["s-1"]); + let error = acpmux.session("p-2").unwrap_err(); + assert!(error.contains("c1"), "{error}"); + assert_eq!(list_params(&requests), [json!({}), json!({"cursor": "c1"})]); +} From 952bc6014db50a427e9aec17ccda2f1383c5b6dc Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 11:02:40 +0000 Subject: [PATCH 2/2] Chief: read a paged acpmux session list whole acpmux answers `_acpmux/sessions` and `_acpmux/watch` with no `limit` in byte-bounded pages, each with the `nextCursor` of the rest, when the whole list is too large for one message. The Chief read only the first page. One helper, `whole_list`, now reads every page by `_acpmux/sessions` with the same params (less the watch's `enabled`) from each cursor and joins the rows; a cursor that comes again is refused so it never loops, and a call with a `limit` stays one request. The link's watch, `sessions()` and the bounded list in agents.rs all go through it. A list that fits is one reply with no cursor, so older daemons work unchanged. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-c11d09c9f41988fc1d93 --- Native/OptChat/optchat-chief/src/acpmux.rs | 90 ++++++++++++++++++++-- Native/OptChat/optchat-chief/src/agents.rs | 12 +-- 2 files changed, 92 insertions(+), 10 deletions(-) diff --git a/Native/OptChat/optchat-chief/src/acpmux.rs b/Native/OptChat/optchat-chief/src/acpmux.rs index d8abe4321855..512fadfa60d6 100644 --- a/Native/OptChat/optchat-chief/src/acpmux.rs +++ b/Native/OptChat/optchat-chief/src/acpmux.rs @@ -688,9 +688,12 @@ impl Acpmux { .map_err(|e| format!("initialize: {e}"))?; // The watch answers with the session list: reading it again at once spent // the connection's second paced list (acpmux `hub/list_guard.rs`). - let watched = client - .request("_acpmux/watch", json!({"enabled": true})) - .map_err(|e| format!("watch: {e}"))?; + let watched = whole_list( + |m, p| client.request(m, p), + "_acpmux/watch", + json!({"enabled": true}), + ) + .map_err(|e| format!("watch: {e}"))?; let list = match session_rows(&watched) { Some(list) => list, None => sessions(&client)?, @@ -863,12 +866,60 @@ fn route(turns: &Mutex>, sink: &Sink, n: Notification /// `_acpmux/sessions`, rows that do not parse skipped. pub fn sessions(client: &RpcClient) -> Result, String> { - let result = client - .request("_acpmux/sessions", json!({})) + let result = whole_list(|m, p| client.request(m, p), "_acpmux/sessions", json!({})) .map_err(|e| format!("sessions: {e}"))?; Ok(session_rows(&result).unwrap_or_default()) } +/// The whole session list `method` (`_acpmux/sessions` or `_acpmux/watch`) +/// answers for `params`, each request asked by `ask`. With no `limit`, a list +/// too large for one acpmux message comes in pages, each with the `nextCursor` +/// of the rest, read by `_acpmux/sessions` with the same params (less the +/// watch's `enabled`) from that cursor; the answer is one `sessions` of every +/// page. A list that fits is one answer with no cursor, as from an acpmux that +/// never pages. A cursor that comes again is refused, never read in a loop. +/// With a `limit`, the one page asked for (acpmux `Client::whole_list`). +pub(crate) fn whole_list( + ask: impl Fn(&str, Value) -> Result, + method: &str, + params: Value, +) -> Result { + let ask = |m: &str, p: Value| ask(m, p).map_err(|e| e.to_string()); + if params.get("limit").is_some_and(|limit| !limit.is_null()) { + return ask(method, params); + } + let mut reply = ask(method, params.clone())?; + let mut rest = params; + if let Some(p) = rest.as_object_mut() { + p.remove("enabled"); + } + let mut read = HashSet::new(); + while let Some(cursor) = reply.as_object_mut().and_then(|r| r.remove("nextCursor")) { + let Value::String(cursor) = cursor else { break }; + if !read.insert(cursor.clone()) { + return Err(format!( + "the session list answered cursor {cursor:?} again; it would never end" + )); + } + rest["cursor"] = Value::String(cursor); + let mut page = ask("_acpmux/sessions", rest.clone())?; + let rows = page + .get_mut("sessions") + .map(Value::take) + .unwrap_or_default(); + if let (Some(all), Value::Array(rows)) = ( + reply.get_mut("sessions").and_then(Value::as_array_mut), + rows, + ) { + all.extend(rows); + } + if let Some(next) = page.get_mut("nextCursor").map(Value::take) { + reply["nextCursor"] = next; + } + } + Ok(reply) +} + /// A list answer's rows (`_acpmux/sessions`, `_acpmux/watch`), those that do /// not parse skipped; None when it has no `sessions` (an acpmux whose watch /// answers without the list). @@ -1482,3 +1533,32 @@ const PICKER_FIELDS: &[&str] = &[ "releaseDate", "description", ]; + +#[cfg(test)] +mod whole_list_tests { + use std::cell::RefCell; + + use serde_json::{Value, json}; + + use super::whole_list; + + /// With a `limit` the caller pages itself: one request, its cursor kept. + #[test] + fn a_list_with_a_limit_is_one_request() { + let asked = RefCell::new(Vec::new()); + let reply = whole_list( + |m: &str, p: Value| { + asked.borrow_mut().push((m.to_owned(), p)); + Ok::<_, String>(json!({"sessions": [{"sessionId": "s-1"}], "nextCursor": "c1"})) + }, + "_acpmux/sessions", + json!({"limit": 1}), + ) + .unwrap(); + assert_eq!(reply["nextCursor"], "c1"); + assert_eq!( + asked.into_inner(), + [("_acpmux/sessions".to_owned(), json!({"limit": 1}))] + ); + } +} diff --git a/Native/OptChat/optchat-chief/src/agents.rs b/Native/OptChat/optchat-chief/src/agents.rs index 5f9be5d754df..9caa65a5d8d0 100644 --- a/Native/OptChat/optchat-chief/src/agents.rs +++ b/Native/OptChat/optchat-chief/src/agents.rs @@ -178,12 +178,14 @@ fn ask_bounded(client: &RpcClient, method: &str, params: Value) -> Result Result, String> { - Ok( - crate::acpmux::session_rows(&ask_bounded(client, "_acpmux/sessions", json!({}))?) - .unwrap_or_default(), - ) + let list = crate::acpmux::whole_list( + |m, p| ask_bounded(client, m, p), + "_acpmux/sessions", + json!({}), + )?; + Ok(crate::acpmux::session_rows(&list).unwrap_or_default()) } /// How long [`end_own_sessions`] waits for all its kills.