Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 77 additions & 25 deletions cmux-tui/crates/acpmux/src/hub/list_guard.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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<String, RpcError> {
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<String, RpcError> {
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)
Expand Down Expand Up @@ -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<usize>],
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:<id>`) 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));
Expand Down Expand Up @@ -573,12 +625,12 @@ mod tests {
json!({"peer": "cmux16", "limit": 1}),
] {
assert_eq!(
list.page(&params).unwrap(),
list.page(None, &params, usize::MAX).unwrap(),
reply(&params).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,
Expand Down Expand Up @@ -656,7 +708,7 @@ mod tests {
[json!({}), json!({"limit": 2}), json!({"status": "closed"}), json!({"peer": "remote"})]
{
assert_eq!(
list.page(&params).unwrap(),
list.page(None, &params, usize::MAX).unwrap(),
super::super::session_pages::session_list_reply(rows.clone(), &params)
.unwrap()
.to_string()
Expand Down
1 change: 1 addition & 0 deletions cmux-tui/crates/acpmux/src/hub/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
199 changes: 199 additions & 0 deletions cmux-tui/crates/acpmux/src/hub/reply_fit.rs
Original file line number Diff line number Diff line change
@@ -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<Item = String>,
mut replace: impl FnMut(&mut Value, usize),
) -> Vec<String> {
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<String> = 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());
}
}
Loading
Loading