diff --git a/cmux-tui/crates/acpmux/src/hub/list_guard.rs b/cmux-tui/crates/acpmux/src/hub/list_guard.rs index 6f4c5ce48567..54a343761602 100644 --- a/cmux-tui/crates/acpmux/src/hub/list_guard.rs +++ b/cmux-tui/crates/acpmux/src/hub/list_guard.rs @@ -205,15 +205,7 @@ impl Snapshot { let picked = super::session_pages::session_list_reply(keys.clone(), &filters)?; let rows = picked.get("sessions").and_then(Value::as_array).map(Vec::as_slice).unwrap_or_default(); - let at = |key: &Value| key.get("at").and_then(Value::as_u64).unwrap_or_default() as usize; - let row = |key: &Value| -> std::borrow::Cow<'_, str> { - let text = spans.get(at(key)).and_then(|span| text.get(span.clone())); - match text { - Some(text) if text.len() <= budget => text.into(), - Some(text) => too_large_stub(key, text.len()).into(), - None => "null".into(), - } - }; + let row = |key: &Value| row_text(text, spans, key, budget); let whole = cursor.is_none() && { let bytes: usize = rows.iter().map(|key| row(key).len() + 1).sum(); bytes <= budget @@ -254,23 +246,49 @@ impl Snapshot { rows.finish() } - /// The reply to a paged or filtered `params`, cut from the shared text: the bytes - /// `session_list_reply` over the summaries writes (`nextCursor` sorts first). - pub(crate) fn page(&self, params: &Value) -> Result { - let picked = super::session_pages::session_list_reply(self.keys.clone(), params)?; + /// The reply to a paged `params` (a `limit`, or a census's `orderBy`), cut from the shared + /// text, with tools when `full` is given: the rows `session_list_reply` picks, as many as fit + /// in `budget` bytes, at least one, and `nextCursor` first. A page cut short by its bytes ends + /// with the `nextCursor` of its last row, as a page of that many rows would: 1000 rows of a + /// busy host with usage were more than one message, refused whole on every retry. A row + /// larger than `budget` alone is trimmed ([`row_text`]). + pub(crate) fn page( + &self, + full: Option<&FullRows>, + params: &Value, + budget: usize, + ) -> Result { + let (text, spans, keys) = match full { + Some(full) => (full.whole.as_str(), &full.spans, &full.keys), + None => (self.whole.as_str(), &self.spans, &self.keys), + }; + let picked = super::session_pages::session_list_reply(keys.clone(), params)?; + let rows = + picked.get("sessions").and_then(Value::as_array).map(Vec::as_slice).unwrap_or_default(); + let mut next = picked.get("nextCursor").cloned(); + let (mut page, mut used) = (Vec::with_capacity(rows.len()), 0_usize); + for (n, key) in rows.iter().enumerate() { + let row = row_text(text, spans, key, budget); + // One more byte for the comma between rows. + let total = used.saturating_add(row.len()).saturating_add(1); + if n > 0 && total > budget { + let last = rows.get(n.saturating_sub(1)).unwrap_or(key); + next = Some(Value::String(page_cursor(params, last))); + break; + } + used = total; + page.push(row); + } let mut text = String::from("{"); - if let Some(next) = picked.get("nextCursor") { + if let Some(next) = next { text.push_str(&format!("\"nextCursor\":{next},")); } text.push_str("\"sessions\":["); - for (n, key) in picked["sessions"].as_array().into_iter().flatten().enumerate() { + for (n, row) in page.iter().enumerate() { if n > 0 { text.push(','); } - let at = key["at"].as_u64().unwrap_or_default() as usize; - text.push_str( - self.spans.get(at).and_then(|span| self.whole.get(span.clone())).unwrap_or("null"), - ); + text.push_str(row); } text.push_str("]}"); Ok(text) @@ -299,9 +317,43 @@ impl Snapshot { } } -/// A row larger than a page alone goes as this stub, with its size, as a record too large for a -/// reply goes (`hub/paging.rs`): a page holds at least one row, and one that large would be -/// refused, page and all. It keeps the row's paging keys (an id longer than 4 KiB as null), so +/// Row `key` of `text`: whole when it fits in `budget`, else trimmed to fit, its largest fields +/// first (`reply_fit.rs`), or as [`too_large_stub`] when even the fields it keeps do not. +fn row_text<'a>( + text: &'a str, + spans: &[Range], + key: &Value, + budget: usize, +) -> std::borrow::Cow<'a, str> { + let at = key.get("at").and_then(Value::as_u64).and_then(|at| usize::try_from(at).ok()); + let Some(row) = at.and_then(|at| spans.get(at)).and_then(|span| text.get(span.clone())) else { + return "null".into(); + }; + if row.len() <= budget { + return row.into(); + } + let mut value: Value = serde_json::from_str(row).unwrap_or_default(); + if super::reply_fit::fit_object(&mut value, budget) { + value.to_string().into() + } else { + too_large_stub(key, row.len()).into() + } +} + +/// The cursor of the page after `last`: a census's (`orderBy`, `s:`) or the list's +/// (`session_pages::cursor_after`). +fn page_cursor(params: &Value, last: &Value) -> String { + let census = params.get("orderBy").and_then(Value::as_str).is_some_and(|o| !o.is_empty()); + if census { + format!("s:{}", last.get("sessionId").and_then(Value::as_str).unwrap_or("")) + } else { + super::session_pages::cursor_after(last) + } +} + +/// A row larger than a page alone even trimmed (a kept field that large) goes as this stub, with +/// its size, as a record too large for a reply goes (`hub/paging.rs`): a page holds at least one +/// row, and one that large would be refused, page and all. It keeps the row's paging keys (an id longer than 4 KiB as null), so /// a client still sees the chat and can read it alone (`_acpmux/info`). fn too_large_stub(key: &Value, bytes: usize) -> String { let id = key.get("sessionId").filter(|v| v.as_str().is_some_and(|t| t.len() <= 4 << 10)); @@ -573,12 +625,12 @@ mod tests { json!({"peer": "cmux16", "limit": 1}), ] { assert_eq!( - list.page(¶ms).unwrap(), + list.page(None, ¶ms, usize::MAX).unwrap(), reply(¶ms).unwrap().to_string(), "{params}" ); } - assert!(list.page(&json!({"cursor": 3})).is_err()); + assert!(list.page(None, &json!({"cursor": 3}), usize::MAX).is_err()); } /// A list read from a session list cursor (`k...`) with no `limit`, too large for one page, @@ -656,7 +708,7 @@ mod tests { [json!({}), json!({"limit": 2}), json!({"status": "closed"}), json!({"peer": "remote"})] { assert_eq!( - list.page(¶ms).unwrap(), + list.page(None, ¶ms, usize::MAX).unwrap(), super::super::session_pages::session_list_reply(rows.clone(), ¶ms) .unwrap() .to_string() diff --git a/cmux-tui/crates/acpmux/src/hub/mod.rs b/cmux-tui/crates/acpmux/src/hub/mod.rs index aa3ac4003bac..63f2a22b519e 100644 --- a/cmux-tui/crates/acpmux/src/hub/mod.rs +++ b/cmux-tui/crates/acpmux/src/hub/mod.rs @@ -104,6 +104,7 @@ mod prompt_line_tests; mod protected_hold; mod queue; mod queue_edit; +pub(crate) mod reply_fit; pub use digest::{MAX_FROM_AGENT_CHARS, declared_sender}; pub use queue::{QueuedPrompt, Withdrawn, supersede::key as supersede_key}; mod resolve; diff --git a/cmux-tui/crates/acpmux/src/hub/reply_fit.rs b/cmux-tui/crates/acpmux/src/hub/reply_fit.rs new file mode 100644 index 000000000000..d448c9bdfd92 --- /dev/null +++ b/cmux-tui/crates/acpmux/src/hub/reply_fit.rs @@ -0,0 +1,199 @@ +//! Part of `Hub`; see `hub/mod.rs`. A reply fits one message (`rpc::MAX_MESSAGE_BYTES`) as a +//! whole: a session row or a chat's detail too large for it is trimmed, its largest fields +//! first, rather than refused, so the list page or the `_acpmux/info` it is in still goes. A +//! refused reply is refused again on every retry: `acpmux ls` on a fleet host failed with +//! "message too large to send: 16927852 bytes" on 2026-10-31, and no lead could list its +//! sessions. +//! +//! [`shrink_largest`] is the one rule: replace the largest parts, largest first, until the +//! value fits. [`fit_object`] applies it to an object's own fields and marks what it trimmed. + +use std::cmp::Reverse; + +use serde_json::{Map, Value, json}; + +/// The most one reply's result holds: a megabyte under the message limit, for the JSON-RPC +/// envelope, its id and what masking a peer's text may add. +pub(crate) const REPLY_BYTES: usize = crate::rpc::MAX_MESSAGE_BYTES - (1 << 20); + +/// A trimmed string keeps this much of its start, so a row still says what it was. +const PREVIEW_BYTES: usize = 1 << 10; + +/// Room kept for the `omitted` mark [`fit_object`] adds. +const MARK_BYTES: usize = 4 << 10; + +/// The fields a row is found, paged, filtered and opened by: never trimmed. The same set the +/// attach reply's session keeps (`_acpmux/attach`). +const KEEP: &[&str] = &[ + "sessionId", + "id", + "qualifiedSessionId", + "sessionReceipt", + "peer", + "status", + "harness", + "cwd", + "updatedAt", + "name", + // With `sessionId` and `harness`, what a peer's reply masks a summary by (`peer_text.rs`): + // a trimmed row without it would go to a peer unmasked. + "lastSeq", +]; + +fn size(value: &Value) -> usize { + serde_json::to_string(value).map_or(usize::MAX, |text| text.len()) +} + +/// Replaces the parts of `value` at `pointers` (JSON pointers), largest first, with +/// `replace(part, its bytes)` until `value` serializes in `budget` bytes, and returns the +/// pointers it replaced. `bytes` is `value`'s size now. +pub(crate) fn shrink_largest( + value: &mut Value, + mut bytes: usize, + budget: usize, + pointers: impl IntoIterator, + mut replace: impl FnMut(&mut Value, usize), +) -> Vec { + let mut parts: Vec<(String, usize)> = pointers + .into_iter() + .filter_map(|at| { + let part = size(value.pointer(&at)?); + Some((at, part)) + }) + .collect(); + parts.sort_by_key(|(_, part)| Reverse(*part)); + let mut replaced = Vec::new(); + for (at, part) in parts { + if bytes <= budget { + break; + } + let Some(target) = value.pointer_mut(&at) else { continue }; + replace(target, part); + bytes = bytes.saturating_sub(part.saturating_sub(size(target))); + replaced.push(at); + } + replaced +} + +/// Trims `object`'s largest fields until it serializes in `budget` bytes: a string keeps its +/// first [`PREVIEW_BYTES`], any other value goes. The fields of [`KEEP`] stay whole. What it +/// trimmed is marked `omitted: {bytes, reason: "too_large", fields}`, `bytes` the object's size +/// before, as a record too large for a reply is (`hub/paging.rs`). False when it still does not +/// fit (a kept field alone is that large): the caller sends a stub instead. +pub(crate) fn fit_object(object: &mut Value, budget: usize) -> bool { + let before = size(object); + if before <= budget { + return true; + } + let Some(map) = object.as_object() else { return false }; + let pointers: Vec = map + .keys() + .filter(|key| !KEEP.contains(&key.as_str())) + .map(|key| format!("/{}", key.replace('~', "~0").replace('/', "~1"))) + .collect(); + let room = budget.saturating_sub(MARK_BYTES); + let trimmed = shrink_largest(object, before, room, pointers, |part, _| { + *part = match part { + Value::String(text) => Value::String(preview(text)), + _ => Value::Null, + }; + }); + let Some(map) = object.as_object_mut() else { return false }; + let mut fields = Vec::with_capacity(trimmed.len()); + for at in &trimmed { + let key = at.get(1..).unwrap_or("").replace("~1", "/").replace("~0", "~"); + if map.get(&key).is_some_and(Value::is_null) { + map.remove(&key); + } + fields.push(Value::String(key)); + } + let mark = json!({"bytes": before, "reason": "too_large", "fields": fields}); + map.insert("omitted".to_owned(), mark); + size(object) <= budget +} + +/// [`fit_object`] for a whole reply (`_acpmux/info`'s detail): what still does not fit goes as +/// only its [`KEEP`] fields that are small, marked as [`fit_object`] marks. +pub(crate) fn fit_reply_object(object: &mut Value) { + let before = size(object); + if fit_object(object, REPLY_BYTES) { + return; + } + let mut stub = Map::new(); + for key in KEEP { + if let Some(value) = object.get(*key).filter(|v| size(v) <= PREVIEW_BYTES) { + stub.insert((*key).to_owned(), value.clone()); + } + } + stub.insert("omitted".to_owned(), json!({"bytes": before, "reason": "too_large"})); + *object = Value::Object(stub); +} + +/// The start of `text`, at most [`PREVIEW_BYTES`], masked (`secret_scan.rs`) before it is cut, +/// over [`SCAN_SLACK_BYTES`] more, as a lane's title is (`views.rs` `masked_cut`): a secret the +/// cut would split is masked whole, never left as a head a peer's masking no longer knows. +fn preview(text: &str) -> String { + let head = prefix(text, PREVIEW_BYTES.saturating_add(SCAN_SLACK_BYTES)); + prefix(&crate::secret_scan::mask(head), PREVIEW_BYTES).to_owned() +} + +/// How far past the cut a preview is scanned for secrets. +const SCAN_SLACK_BYTES: usize = 4 << 10; + +/// The start of `text`, at most `max` bytes, ending on a character boundary. +fn prefix(text: &str, max: usize) -> &str { + let end = text + .char_indices() + .map(|(at, c)| at.saturating_add(c.len_utf8())) + .take_while(|end| *end <= max) + .last() + .unwrap_or(0); + text.get(..end).unwrap_or("") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_row_too_large_keeps_its_keys_and_a_preview_and_says_what_it_trimmed() { + let mut row = json!({"sessionId": "s1", "status": "idle", "updatedAt": 5, "cwd": "/w", + "title": "é".repeat(1 << 20), "turn": {"plan": ["x".repeat(1 << 19)]}}); + let before = size(&row); + assert!(fit_object(&mut row, 64 << 10)); + assert!(size(&row) <= 64 << 10, "{}", size(&row)); + assert_eq!((row["sessionId"].as_str(), row["cwd"].as_str()), (Some("s1"), Some("/w"))); + let title = row["title"].as_str().unwrap(); + assert!(title.len() <= PREVIEW_BYTES && title.starts_with('é')); + assert!(row.get("turn").is_none()); + assert_eq!(row["omitted"]["bytes"], before); + assert_eq!(row["omitted"]["reason"], "too_large"); + assert_eq!(row["omitted"]["fields"], json!(["title", "turn"])); + // One that fits is left as it was. + let small = json!({"sessionId": "s2", "title": "t"}); + let mut same = small.clone(); + assert!(fit_object(&mut same, 1 << 10)); + assert_eq!(same, small); + } + + #[test] + fn a_reply_whose_kept_field_alone_is_too_large_goes_as_a_stub() { + let mut detail = json!({"sessionId": "s", "name": "n".repeat(REPLY_BYTES), "cwd": "/w"}); + fit_reply_object(&mut detail); + assert_eq!(detail["sessionId"], "s"); + // The kept fields that are small stay, the huge one goes. + assert!(detail.get("name").is_none()); + assert_eq!(detail["cwd"], "/w"); + assert_eq!(detail["omitted"]["reason"], "too_large"); + } + + #[test] + fn shrink_largest_replaces_largest_first_and_stops_once_it_fits() { + let mut v = json!({"a": "x".repeat(100), "b": "y".repeat(300), "c": "z".repeat(200)}); + let bytes = size(&v); + let ptrs = ["/a", "/b", "/c", "/missing"].map(str::to_owned); + let done = shrink_largest(&mut v, bytes, bytes - 250, ptrs, |p, _| *p = Value::Null); + assert_eq!(done, ["/b"]); + assert!(v["b"].is_null() && v["a"].is_string() && v["c"].is_string()); + } +} diff --git a/cmux-tui/crates/acpmux/src/server/list_fit_tests.rs b/cmux-tui/crates/acpmux/src/server/list_fit_tests.rs new file mode 100644 index 000000000000..d9d44f912144 --- /dev/null +++ b/cmux-tui/crates/acpmux/src/server/list_fit_tests.rs @@ -0,0 +1,200 @@ +//! Every session list page and session detail fits one message as a whole. On 2026-10-31 +//! `acpmux ls` failed on a fleet host with "message too large to send: 16927852 bytes, the +//! limit is 15728640": a page of `limit` rows was cut by count alone, and a reply refused as +//! too large is refused again on every retry. Here the list (and one row of it alone) is past +//! the limit, and every read still lists every id. + +use crate::config::{Config, StoreMode}; +use crate::peer::{Peer, PeerNotice}; +use crate::rpc::{MAX_MESSAGE_BYTES, Message, method}; +use serde_json::{Value, json}; +use std::collections::HashSet; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; + +const ROWS: usize = 1000; +const BIG: &str = "peer-chat-big"; +const LOCAL: &str = "local-long-title"; + +/// A hub with a chat of its own titled with 16 MiB, and mirroring `ROWS` chats of 20 KB each +/// from its peer, plus one of 16 MiB: the list and that one row are each past the limit. +async fn big_hub() -> Arc { + big_hub_with(Vec::new()).await +} + +/// [`big_hub`] with `extra` rows from the peer too. +async fn big_hub_with(extra: Vec) -> Arc { + let extras = extra.len(); + let mut config = Config::default(); + config.store.mode = StoreMode::Memory; + let store = crate::store::open(&config.store, std::path::Path::new("/nonexistent")).unwrap(); + let meta = serde_json::from_value(json!({ + "schema": crate::store::META_SCHEMA, "id": LOCAL, "name": LOCAL, "title": "t".repeat(16 << 20), + "harness": "fake", "cwd": "/tmp", "status": "idle", "createdAt": 1, "updatedAt": 1 + })) + .unwrap(); + store.save(&meta).unwrap(); + let hub = crate::hub::Hub::new(config, store); + let (tx, _rx) = mpsc::channel(8); + let peer = Peer::new("box", "ws://127.0.0.1:9", None, None, tx); + hub.peers.lock().unwrap().insert("box".to_owned(), peer.clone()); + let row = |id: String, at: usize, step: usize| { + json!({"sessionId": id, "name": "lane", "harness": "claude", "cwd": "/work", + "status": "idle", "updatedAt": at as u64 + 10, "planStep": "x".repeat(step)}) + }; + let mut list: Vec = + (0..ROWS).map(|i| row(format!("peer-chat-{i}"), i, 20_000)).collect(); + list.push(row(BIG.to_owned(), ROWS / 2, 16 << 20)); + list.extend(extra); + let mut settled = peer.settled(); + for notice in [PeerNotice::Sessions(list), PeerNotice::Connected] { + hub.peer_notices.send(("box".to_owned(), peer.generation, notice)).await.unwrap(); + } + tokio::time::timeout(Duration::from_secs(10), settled.wait_for(|n| *n >= 1)) + .await + .expect("the hub applies the notices") + .unwrap(); + assert_eq!(hub.remote_sessions().len(), ROWS + 1 + extras); + hub +} + +/// A connection to `hub` that sends requests and returns each reply with its line's size. +struct Client { + input: mpsc::Sender, + outgoing: mpsc::Receiver, + id: u64, +} + +impl Client { + async fn new(hub: &Arc) -> Self { + Self::on(hub, super::Origin::Local).await + } + + /// A connection of `origin`: a peer's reads its summaries masked (`peer_text.rs`). + async fn on(hub: &Arc, origin: super::Origin) -> Self { + let (input, incoming) = mpsc::channel(8); + let (output, outgoing) = mpsc::channel(8); + tokio::spawn(super::serve_connection_with(hub.clone(), incoming, output, origin)); + let mut client = Self { input, outgoing, id: 0 }; + let hello = json!({"protocolVersion": 1, "clientInfo": + {"name": crate::peer::CLIENT_NAME, "version": "0"}}); + client.call(method::INITIALIZE, hello).await; + client + } + + async fn call(&mut self, m: &str, params: Value) -> (Value, usize) { + self.id += 1; + let id = self.id; + self.input.send(Message::request(id, m, params).to_line()).await.unwrap(); + tokio::time::timeout(Duration::from_secs(60), async { + loop { + let line = self.outgoing.recv().await.unwrap(); + let reply: Value = serde_json::from_str(&line).unwrap(); + if reply["id"] == id { + return (reply, line.len()); + } + } + }) + .await + .expect("the reply comes") + } + + /// Every `_acpmux/sessions` row for `params`, following `nextCursor`; each reply under the + /// message limit. + async fn rows(&mut self, params: Value) -> Vec { + let mut rows = Vec::new(); + let mut cursor = Value::Null; + for page in 0..100 { + let mut p = params.clone(); + p["cursor"] = cursor; + let (reply, bytes) = self.call(method::MUX_SESSIONS, p).await; + assert!(reply.get("error").is_none(), "{params} page {page}: {}", reply["error"]); + assert!(bytes <= MAX_MESSAGE_BYTES, "{params} page {page}: {bytes} bytes"); + rows.extend(reply["result"]["sessions"].as_array().unwrap().iter().cloned()); + cursor = reply["result"]["nextCursor"].clone(); + if !cursor.is_string() { + return rows; + } + } + panic!("{params}: the pages never end"); + } +} + +#[tokio::test] +async fn a_list_and_a_row_each_past_the_message_limit_still_list_every_id() { + let hub = big_hub().await; + let mut client = Client::new(&hub).await; + // `acpmux ls`'s pages (limit 1000), with tools, and with no limit at all. + for params in [ + json!({"limit": 1000, "usage": true}), + json!({"limit": 1000, "withTools": true}), + json!({}), + json!({"localOnly": true, "orderBy": "sessionId", "limit": 1000}), + ] { + let rows = client.rows(params.clone()).await; + let ids: HashSet<&str> = rows.iter().filter_map(|r| r["sessionId"].as_str()).collect(); + if params.get("localOnly").is_some() { + assert_eq!(ids, HashSet::from([LOCAL]), "{params}"); + continue; + } + assert_eq!(ids.len(), ROWS + 2, "{params}: {} ids", ids.len()); + let big = rows.iter().find(|r| r["sessionId"] == BIG).unwrap(); + // The row is trimmed, not dropped: its keys stay, and it says what it lost. + assert_eq!((big["status"].as_str(), big["peer"].as_str()), (Some("idle"), Some("box"))); + assert_eq!(big["cwd"], "/work", "{params}"); + assert!(big["planStep"].as_str().is_some_and(|s| s.len() <= 1 << 10), "{params}"); + assert_eq!(big["omitted"]["reason"], "too_large", "{params}"); + assert_eq!(big["omitted"]["fields"], json!(["planStep"]), "{params}"); + let small = rows.iter().find(|r| r["sessionId"] == "peer-chat-7").unwrap(); + assert_eq!(small["planStep"].as_str().map(str::len), Some(20_000), "{params}"); + } +} + +#[tokio::test] +async fn a_session_info_past_the_message_limit_comes_trimmed() { + let hub = big_hub().await; + let mut client = Client::new(&hub).await; + let (reply, bytes) = + client.call(method::MUX_INFO, json!({"sessionId": LOCAL, "waitLogs": false})).await; + assert!(reply.get("error").is_none(), "{}", reply["error"]); + assert!(bytes <= MAX_MESSAGE_BYTES, "{bytes} bytes"); + let detail = &reply["result"]; + assert_eq!((detail["sessionId"].as_str(), detail["cwd"].as_str()), (Some(LOCAL), Some("/tmp"))); + assert!(detail["title"].as_str().is_some_and(|t| t.starts_with('t') && t.len() <= 1 << 10)); + assert!(detail["omitted"]["fields"].as_array().unwrap().contains(&json!("title"))); +} + +/// A fake token the secret scanner masks (`secret_scan_tests.rs`). +fn token(seed: &str) -> String { + let body: String = "aZ3kQ9mX7pL2vB8nR4tW6yC1dF5gH0jS".chars().cycle().take(36).collect(); + format!("gh{}_{seed}{}", "p", body.get(seed.len()..).unwrap()) +} + +/// A row trimmed to fit goes to a peer masked: it keeps what masking knows a summary by, and a +/// secret its preview's cut would split is masked before the cut, never left as a head. +#[tokio::test] +async fn a_trimmed_row_still_goes_to_a_peer_masked() { + let (head, cut) = (token("Head"), token("Cut")); + // The second starts 20 bytes before the 1 KiB preview ends: the cut would split it. + let mut prompt = format!("run {head} then "); + prompt.push_str(&"p".repeat(1004 - prompt.len())); + prompt.push_str(&cut); + prompt.push_str(&"q".repeat(16 << 20)); + let row = json!({"sessionId": "secret-chat", "harness": "claude", "lastSeq": 3, + "cwd": "/work", "status": "idle", "updatedAt": 99_999, "lastPrompt": prompt}); + let hub = big_hub_with(vec![row]).await; + let mut client = Client::on(&hub, super::Origin::Peer).await; + for params in [json!({"limit": 1000}), json!({})] { + let rows = client.rows(params.clone()).await; + let row = rows.iter().find(|r| r["sessionId"] == "secret-chat").expect("listed"); + assert_eq!(row["lastSeq"], 3, "{params}"); + assert_eq!(row["omitted"]["fields"], json!(["lastPrompt"]), "{params}"); + let shown = row["lastPrompt"].as_str().unwrap(); + assert!(shown.starts_with("run "), "{params}: {shown:.40}"); + for secret in [&head, &cut] { + let start = secret.get(..12).unwrap(); + assert!(!shown.contains(start), "{params}: {start} in {shown}"); + } + } +} diff --git a/cmux-tui/crates/acpmux/src/server/list_reply.rs b/cmux-tui/crates/acpmux/src/server/list_reply.rs index 7c0bc0b1a1d9..7a10728c2447 100644 --- a/cmux-tui/crates/acpmux/src/server/list_reply.rs +++ b/cmux-tui/crates/acpmux/src/server/list_reply.rs @@ -108,43 +108,22 @@ fn with_tools(params: &Value) -> Result { /// `_acpmux/sessions`: the whole list shared, or a page or filter cut from it. One with no /// `limit` goes whole, newest first, as before, when it fits in [`LIST_PAGE_BYTES`] as the /// client reads it; one larger goes in pages, each with the `nextCursor` of the rest -/// (`Snapshot::bounded`), rather than as one message too large to send. A census in its own -/// order (`orderBy`) pages by `limit` itself. +/// (`Snapshot::bounded`), rather than as one message too large to send. A page by `limit`, or a +/// census in its own order (`orderBy`), holds at most `limit` rows and at most +/// [`LIST_PAGE_BYTES`] (`Snapshot::page`): no reply here is too large to send. pub(super) async fn sessions(hub: &Hub, conn: &Conn, params: &Value) -> Result { let raw = conn.raw_text; let limit = params.get("limit").and_then(Value::as_u64); let unset = |k: &str| params.get(k).is_none_or(|v| v.is_null() || v.as_str() == Some("")); let by_bytes = unset("limit") && unset("orderBy"); - let shown = |text: String| { - Arc::new(if raw { text } else { masked(serde_json::from_str(&text).unwrap_or_default()) }) - }; - if with_tools(params)? { - return reply(hub, conn, limit, |list| { - if by_bytes { - return within_limit(raw, |budget| list.bounded(hub, params, true, budget)); - } - let full = list.full_whole(hub); - let mut result: Value = serde_json::from_str(&full).unwrap_or_default(); - if !whole(params) { - let rows = result - .get_mut("sessions") - .map(Value::take) - .unwrap_or_default() - .as_array() - .cloned() - .unwrap_or_default(); - result = crate::hub::session_pages::session_list_reply(rows, params)?; - } - Ok(Arc::new(if raw { result.to_string() } else { masked(result) })) - }) - .await; - } - if !whole(params) { + let tools = with_tools(params)?; + if tools || !whole(params) { return reply(hub, conn, limit, |list| { if by_bytes { - return within_limit(raw, |budget| list.bounded(hub, params, false, budget)); + return within_limit(raw, |budget| list.bounded(hub, params, tools, budget)); } - Ok(shown(list.page(params)?)) + let full = tools.then(|| list.full_rows(hub)); + within_limit(raw, |budget| list.page(full.as_deref(), params, budget)) }) .await; } @@ -294,10 +273,10 @@ fn acp_row_within(s: &Value, budget: usize) -> Value { /// The longest id or `cwd` a stub row keeps. const STUB_TEXT_BYTES: usize = 4 << 10; -/// The most a session list result may hold (`session/list`, and `_acpmux/sessions` or -/// `_acpmux/watch` with no `limit`): a megabyte under the message limit, for the reply's own -/// frame and what masking a peer's text may add. -const LIST_PAGE_BYTES: usize = crate::rpc::MAX_MESSAGE_BYTES - (1 << 20); +/// The most a session list result may hold (`session/list`, `_acpmux/sessions` and +/// `_acpmux/watch`): a megabyte under the message limit, for the reply's own frame and what +/// masking a peer's text may add, the bound of every reply fitted whole (`hub/reply_fit.rs`). +const LIST_PAGE_BYTES: usize = crate::hub::reply_fit::REPLY_BYTES; fn acp_row(s: &Value) -> Value { let text = |k: &str| s.get(k).and_then(Value::as_str).map(str::to_owned); diff --git a/cmux-tui/crates/acpmux/src/server/mod.rs b/cmux-tui/crates/acpmux/src/server/mod.rs index 42dae3ce88a0..dcce1cc12bea 100644 --- a/cmux-tui/crates/acpmux/src/server/mod.rs +++ b/cmux-tui/crates/acpmux/src/server/mod.rs @@ -2116,6 +2116,10 @@ mod request_admission_tests { #[path = "acp_list_tests.rs"] mod acp_list_tests; +#[cfg(test)] +#[path = "list_fit_tests.rs"] +mod list_fit_tests; + #[cfg(test)] #[path = "session_list_pages_tests.rs"] mod session_list_pages_tests; diff --git a/cmux-tui/crates/acpmux/src/server/requests/info.rs b/cmux-tui/crates/acpmux/src/server/requests/info.rs index f9d950dda2be..be30a43751cc 100644 --- a/cmux-tui/crates/acpmux/src/server/requests/info.rs +++ b/cmux-tui/crates/acpmux/src/server/requests/info.rs @@ -31,5 +31,8 @@ pub(super) async fn info(hub: &Arc, conn: &Conn, params: &Value) -> Result< { obj.insert("sessionEnv".into(), json!(crate::session_env::shown(&s.meta().session_env))); } + // One message carries it whole, its largest fields trimmed if need be: a detail refused + // as too large is refused again on every read (`hub/reply_fit.rs`). + crate::hub::reply_fit::fit_reply_object(&mut detail); Ok(detail) }