Chief: read a paged acpmux session list whole

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<HashMap<String, TurnRoute>>, sink: &Sink, n: Notification
 
 /// `_acpmux/sessions`, rows that do not parse skipped.
 pub fn sessions(client: &RpcClient) -> Result<Vec<SessionSummary>, 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<E: std::fmt::Display>(
+    ask: impl Fn(&str, Value) -> Result<Value, E>,
+    method: &str,
+    params: Value,
+) -> Result<Value, String> {
+    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}))]
+        );
+    }
+}

Native/OptChat/optchat-chief/src/agents.rs

@@ -178,12 +178,14 @@ fn ask_bounded(client: &RpcClient, method: &str, params: Value) -> Result<Value,
         .map_err(|e| format!("{method}: {e}"))
 }
 
-/// `_acpmux/sessions`, bounded.
+/// `_acpmux/sessions`, every page, each bounded.
 fn sessions_bounded(client: &RpcClient) -> Result<Vec<SessionSummary>, 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.

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<Mutex<Vec<Value>>>, 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<Acpmux>, Vec<String>, Arc<Mutex<Vec<Value>>>) {
+    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<Value>>) -> Vec<Value> {
+    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"})]);
+}