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
Original file line number Diff line number Diff line change
Expand Up @@ -122607,7 +122607,7 @@
"en": {
"stringUnit": {
"state": "translated",
"value": "The most sub-sessions one session can have open at the same time. Closing a sub-session frees its place."
"value": "The most live sub-sessions one session can have at the same time. Running, waiting and harness-alive children count; finished, dead and closed children do not use a slot."
}
},
"ar": {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ extension SubsessionSettings {
/// `agents.subsessions.answerQuestions`: On: a session answers the questions (Claude Code AskUserQuestion, Grok and Cursor asks) of the sub-sessions under it (_acpmux/subsession/answer), never their permission prompts or plans. Off (the default, since it loosens the person-only answer rule): only the person answers them (subsessions.answers_off).
public nonisolated static let subsessionsAnswerQuestionsPath = ["agents", "subsessions", "answerQuestions"]
public nonisolated static let subsessionsAnswerQuestionsDefault = false
/// `agents.subsessions.maxChildren`: Open (not closed) sub-sessions one session may have at a time. A session or an ancestor can set a lower value for a subtree.
/// `agents.subsessions.maxChildren`: Live sub-sessions (running, waiting for input, or with a harness process alive) one session may have at a time. Finished, dead and closed sub-sessions do not use a slot. A session or an ancestor can set a lower value for a subtree.
public nonisolated static let subsessionsMaxChildrenPath = ["agents", "subsessions", "maxChildren"]
public nonisolated static let subsessionsMaxChildrenDefault: Double = 50
public nonisolated static let subsessionsMaxChildrenRange: ClosedRange<Double> = 1...1000
Expand Down Expand Up @@ -53,7 +53,7 @@ nonisolated enum SubsessionSettingsSchema {
section: .general,
group: group,
title: SettingsText.keyed("settings.agents.subsessions.maxChildren", "Open Sub-sessions per Session"),
help: SettingsText.keyed("settings.agents.subsessions.maxChildren.help", "The most sub-sessions one session can have open at the same time. Closing a sub-session frees its place."),
help: SettingsText.keyed("settings.agents.subsessions.maxChildren.help", "The most live sub-sessions one session can have at the same time. Running, waiting and harness-alive children count; finished, dead and closed children do not use a slot."),
kind: .number(SettingNumber(SubsessionSettings.subsessionsMaxChildrenRange, step: 1, unit: .count)),
default: .number(SubsessionSettings.subsessionsMaxChildrenDefault),
keywords: ["subagent", "sub-session", "limit", "width", "children", "parallel"]
Expand Down
10 changes: 9 additions & 1 deletion cmux-tui/crates/acpmux/src/hub/log_load.rs
Original file line number Diff line number Diff line change
Expand Up @@ -223,7 +223,15 @@ impl Hub {
if kept && gone && waiting {
// A wait the restart cut: its `status idle` record says so (`mark_lost_wait`), now
// or when the background pass reads its log.
session.meta.lock().unwrap_or_else(|e| e.into_inner()).status = SessionStatus::Idle;
{
let mut m = session.meta.lock().unwrap_or_else(|e| e.into_inner());
m.status = SessionStatus::Idle;
// Its harness is gone: a sub-session frees its parent's width slot, under the
// same lock as the status, as `write_status` does (`subsessions.rs`).
if let Some(link) = &m.subsession {
self.subsessions.note_status(&link.started_by, &session.id, m.status);
}
}
session.mark_waiting(false);
if *loaded {
self.mark_lost_wait(session);
Expand Down
15 changes: 5 additions & 10 deletions cmux-tui/crates/acpmux/src/hub/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -925,11 +925,10 @@ impl Hub {
if let Some(link) = &meta.subsession {
self.subsessions.note_name(&link.started_by, &link.short_name);
}
// A sub-session that is not closed holds a slot of its parent (`subsessions.rs`).
if let Some(link) =
meta.subsession.as_ref().filter(|_| meta.status != SessionStatus::Closed)
{
self.subsessions.set_open(&link.started_by, &meta.id, true);
// A live sub-session holds a slot of its parent (`subsessions.rs`). Idle-finished,
// disconnected and closed children are retained in the store but do not use a slot.
if let Some(link) = meta.subsession.as_ref() {
self.subsessions.note_status(&link.started_by, &meta.id, meta.status);
}
let session = self.make_session(meta);
session.caught_up.store(false, Ordering::SeqCst);
Expand Down Expand Up @@ -1554,11 +1553,7 @@ impl Hub {
// concurrent status writes and the index never disagree (the
// index is a leaf lock, `subsessions.rs`).
if let Some(link) = &m.subsession {
self.subsessions.set_open(
&link.started_by,
&session.id,
status != SessionStatus::Closed,
);
self.subsessions.note_status(&link.started_by, &session.id, status);
}
(was, m.subsession.clone())
};
Expand Down
4 changes: 1 addition & 3 deletions cmux-tui/crates/acpmux/src/hub/offers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -209,9 +209,7 @@ impl Hub {
if meta.lineage.is_none() {
meta.lineage = lineage;
}
if meta.status != super::SessionStatus::Closed {
self.subsessions.set_open(&session.id, &child.id, true);
}
self.subsessions.note_status(&session.id, &child.id, meta.status);
}
} else if !self.with_open_offers(session, |open| Ok(open.remove(offer_id).is_some()))? {
return Err(no_offer());
Expand Down
164 changes: 164 additions & 0 deletions cmux-tui/crates/acpmux/src/hub/subsession_index.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,164 @@
//! Part of `subsessions.rs`: each parent's children that are not closed, the live ones among
//! them, which count toward its width limit (`maxChildren`), and the slots starts in flight
//! reserve. A child is live while its harness is alive (`holds_width_slot`); its entry follows
//! every status write, under the child's meta lock (`Hub::set_status`). A child that is not live
//! but has a prompt on its way to a turn counts too (`Hub::waking_children`).
//!
//! The index is a leaf lock: nothing else is locked while it is held, so it may be taken under
//! any other lock.

use super::*;
use std::collections::BTreeSet;

/// The open and live children of each parent, and the slots reserved by starts in flight.
#[derive(Default)]
pub struct Index {
state: StdMutex<IndexState>,
}

#[derive(Default)]
struct IndexState {
/// Children not closed.
open: HashMap<String, BTreeSet<String>>,
/// Children whose harness is alive: a subset of `open`.
live: HashMap<String, BTreeSet<String>>,
reserved: HashMap<String, u32>,
/// Every short name a child of each parent has or had, open or closed;
/// a failed start keeps its name, so a name is never given twice.
names: HashMap<String, BTreeSet<String>>,
}

fn note(sets: &mut HashMap<String, BTreeSet<String>>, parent: &str, child: &str, on: bool) {
if on {
sets.entry(parent.to_owned()).or_default().insert(child.to_owned());
} else if let Some(set) = sets.get_mut(parent) {
set.remove(child);
if set.is_empty() {
sets.remove(parent);
}
}
}

impl Index {
fn lock(&self) -> std::sync::MutexGuard<'_, IndexState> {
self.state.lock().unwrap_or_else(|e| e.into_inner())
}

/// Notes `child` of `parent` as open or closed, live or not, from its status.
pub(in crate::hub) fn note_status(&self, parent: &str, child: &str, status: SessionStatus) {
let mut s = self.lock();
note(&mut s.open, parent, child, status != SessionStatus::Closed);
note(&mut s.live, parent, child, holds_width_slot(status));
}

/// Notes a loaded child's short name as taken.
pub(in crate::hub) fn note_name(&self, parent: &str, short: &str) {
self.lock().names.entry(parent.to_owned()).or_default().insert(short.to_owned());
}

/// Takes the first free short name of `parent` from `base`, `base-2`, ...
pub(in crate::hub) fn claim_name(&self, parent: &str, base: &str) -> String {
let mut s = self.lock();
let taken = s.names.entry(parent.to_owned()).or_default();
let short = (1..)
.map(|n| if n == 1 { base.to_owned() } else { format!("{base}-{n}") })
.find(|n| !taken.contains(n))
.unwrap_or_else(|| base.to_owned());
taken.insert(short.clone());
short
}

/// The open (not closed) children of `parent`.
pub(in crate::hub) fn open_children(&self, parent: &str) -> Vec<String> {
self.lock().open.get(parent).map(|s| s.iter().cloned().collect()).unwrap_or_default()
}

/// How many children of `parent` are live.
pub fn live_count(&self, parent: &str) -> u32 {
self.lock().live.get(parent).map_or(0, |s| s.len() as u32)
}

/// How many children of `parent` use a width slot: live, or among `waking` (children with a
/// prompt on its way, `Hub::waking_children`) and open but not live now.
pub(super) fn used(&self, parent: &str, waking: &[String]) -> u32 {
Self::used_in(&self.lock(), parent, waking)
}

fn used_in(s: &IndexState, parent: &str, waking: &[String]) -> u32 {
let live = s.live.get(parent);
let open = s.open.get(parent);
let resting = |c: &&String| {
open.is_some_and(|o| o.contains(*c)) && !live.is_some_and(|l| l.contains(*c))
};
live.map_or(0, |set| set.len() as u32) + waking.iter().filter(resting).count() as u32
}

/// How many children of `parent` are open.
pub fn open_count(&self, parent: &str) -> u32 {
self.lock().open.get(parent).map_or(0, |s| s.len() as u32)
}

/// Reserves one child slot of `parent` under `max`, or returns how many are used ([`used`])
/// or reserved.
///
/// [`used`]: Self::used
pub(super) fn reserve<'a>(
&'a self,
parent: &'a str,
max: u32,
waking: &[String],
) -> Result<Reservation<'a>, u32> {
let mut s = self.lock();
let used = Self::used_in(&s, parent, waking) + s.reserved.get(parent).copied().unwrap_or(0);
if used >= max {
return Err(used);
}
*s.reserved.entry(parent.to_owned()).or_default() += 1;
Ok(Reservation { index: self, parent })
}

fn release(&self, parent: &str) {
let mut s = self.lock();
if let Some(n) = s.reserved.get_mut(parent) {
*n = n.saturating_sub(1);
if *n == 0 {
s.reserved.remove(parent);
}
}
}
}

/// One width slot held by a start in flight; dropped (an error, a panic, a
/// cancelled request) it frees the slot.
pub(super) struct Reservation<'a> {
index: &'a Index,
parent: &'a str,
}

impl Reservation<'_> {
/// The start made `child`: it holds the slot from now on while it is
/// live. Under the child's meta lock, like every other
/// change of its entry ([`Hub::set_status`]), so the index follows the
/// order of the status writes.
pub(super) fn settle(self, child: &Session) {
let m = child.meta.lock().unwrap_or_else(|e| e.into_inner());
self.index.note_status(self.parent, &child.id, m.status);
}
}

impl Drop for Reservation<'_> {
fn drop(&mut self) {
self.index.release(self.parent);
}
}

/// A live child: its harness can still do work. `Ready` means the harness process
/// is alive between turns; `Running` and `Waiting` cover an active turn and a prompt waiting for
/// input. `Idle` has no harness process, and `Disconnected` records a dead one.
pub(super) fn holds_width_slot(status: SessionStatus) -> bool {
matches!(status, SessionStatus::Ready | SessionStatus::Running | SessionStatus::Waiting)
}

#[cfg(test)]
#[path = "subsession_width_tests.rs"]
mod width_tests;
119 changes: 119 additions & 0 deletions cmux-tui/crates/acpmux/src/hub/subsession_width_tests.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
//! Which sub-sessions hold a width slot of their parent (`subsession_index.rs`, Leo:
//! "0 open sub-sessions and may have at most 50 open at once ... WTF IS IT COUNTING"). A child
//! counts while its harness is alive (ready, running, waiting); a hibernated (idle), dead
//! (disconnected) or closed one keeps its history but frees its slot.

use super::*;
use crate::config::{Config, StoreMode};
use crate::store::SubsessionLink;
use std::path::Path;

async fn hub_with_parent() -> (Arc<Hub>, Arc<Session>) {
let fake = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fake_agent.py");
let mut cfg: Config = serde_json::from_value(json!({
"harnesses": {"fake": {"argv": ["python3", fake]}},
"defaultHarness": "fake",
}))
.unwrap();
cfg.store.mode = StoreMode::Memory;
let store = crate::store::open(&cfg.store, Path::new("/nonexistent")).unwrap();
let hub = Hub::new(cfg, store);
let req = NewRequest { cwd: Some(std::env::temp_dir()), ..NewRequest::default() };
let parent = hub.new_session(req).await.unwrap();
(hub, parent)
}

async fn child_of(hub: &Arc<Hub>, parent: &Session, short: &str) -> Arc<Session> {
let req = NewRequest {
cwd: Some(std::env::temp_dir()),
subsession: Some(SubsessionLink {
started_by: parent.id.clone(),
short_name: short.to_owned(),
depth: 1,
offered: false,
}),
..NewRequest::default()
};
hub.new_session(req).await.unwrap()
}

/// The `subsession` record events about `child` on `parent`'s log.
fn child_events(hub: &Hub, parent: &Session, child: &Session) -> Vec<String> {
let events = hub.events(&parent.id, 0, 10_000).unwrap();
events
.into_iter()
.filter(|e| e.dir == "mux" && e.kind == RECORD_KIND)
.filter(|e| e.msg["childSessionId"] == child.id.as_str())
.map(|e| e.msg["event"].as_str().unwrap_or_default().to_owned())
.collect()
}

#[test]
fn width_slots_count_live_children_only() {
let index = Index::default();
index.note_status("parent", "finished", SessionStatus::Idle);
index.note_status("parent", "dead", SessionStatus::Disconnected);
index.note_status("parent", "closed", SessionStatus::Closed);
index.note_status("parent", "live", SessionStatus::Running);

assert_eq!(index.live_count("parent"), 1);
// Not closed: the work below a parent looks at them all (`work_below.rs`).
assert_eq!(index.open_children("parent"), vec!["dead", "finished", "live"]);
assert!(index.reserve("parent", 1, &[]).is_err(), "the live child still binds the cap");

index.note_status("parent", "live", SessionStatus::Idle);
assert_eq!(index.live_count("parent"), 0);
assert!(index.reserve("parent", 1, &[]).is_ok(), "finished children free the slot");
// A waking child that is not live counts; a live one counts once, a closed one or a name
// it does not know counts for nothing.
let waking = ["finished".to_owned(), "closed".to_owned(), "gone".to_owned()];
assert_eq!(index.reserve("parent", 1, &waking).err(), Some(1), "a waking child binds it");
index.note_status("parent", "finished", SessionStatus::Running);
assert_eq!(index.used("parent", &waking), 1, "a live waking child counts once");
}

#[tokio::test]
async fn only_live_children_hold_a_width_slot() {
let (hub, parent) = hub_with_parent().await;
let child = child_of(&hub, &parent, "a").await;
let count = || hub.width_used(&parent.id);
let reserve = || hub.subsessions.reserve(&parent.id, 1, &hub.waking_children(&parent.id));
assert_eq!(child.status(), SessionStatus::Ready);
assert_eq!(count(), 1, "a child whose harness is up holds a slot");

for live in [SessionStatus::Running, SessionStatus::Waiting, SessionStatus::Ready] {
hub.set_status(&child, live);
assert_eq!(count(), 1, "a {live} child holds a slot");
}

// Hibernated after its turn: its harness exited, its history stays.
hub.set_status(&child, SessionStatus::Idle);
assert_eq!(count(), 0, "a finished, hibernated child frees its slot");
drop(reserve().expect("the freed slot takes a new start"));

// A prompt queued for it (behind a turn, or for one of the parent's run slots) will wake it.
child.queued.fetch_add(1, Ordering::SeqCst);
assert_eq!(count(), 1, "a child with a queued prompt holds a slot");
assert!(reserve().is_err(), "a start never takes the slot its wake needs");
child.queued.fetch_sub(1, Ordering::SeqCst);
// Its prompt holds the turn lock while its harness starts.
let starting = child.turn_lock.lock().await;
assert_eq!(count(), 1, "a child starting its harness for a prompt holds a slot");
assert!(reserve().is_err());
hub.set_status(&child, SessionStatus::Running);
drop(starting);
assert_eq!(count(), 1, "a woken child holds its slot");

// Its harness died.
hub.set_status(&child, SessionStatus::Disconnected);
assert_eq!(count(), 0, "a dead child frees its slot");

hub.set_status(&child, SessionStatus::Ready);
hub.kill(&child, false).await.unwrap();
assert_eq!(child.status(), SessionStatus::Closed);
assert_eq!(count(), 0, "a closed child frees its slot");
assert_eq!(hub.subsessions.open_count(&parent.id), 0);

// The parent's log says closed only for the real close: hibernating and dying are not one.
assert_eq!(child_events(&hub, &parent, &child), ["closed"]);
}
Loading
Loading