Skip to content

acpmux: move the hub's startup into hub/startup.rs (move only; hub/mod.rs under 1000 lines) #20686

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
teamleaderleo merged 1 commit into from
Oct 14, 2026
Merged

acpmux: move the hub's startup into hub/startup.rs (move only; hub/mod.rs under 1000 lines) #20686

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
117 changes: 1 addition & 116 deletions cmux-tui/crates/acpmux/src/hub/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ use shutdown::ShutdownPlan;
mod spawn;
mod start_defaults;
pub(crate) mod start_mode;
mod startup;
mod steer_end;
mod stream;
mod tap;
Expand Down Expand Up @@ -432,122 +433,6 @@ impl Hub {
self.events.subscribe()
}

/// Hold session creation and agent spawns until `finish_startup`.
pub fn begin_startup(&self, login_env: bool) {
self.login_env_requested.store(login_env, Ordering::SeqCst);
self.startup_ready.send_replace(false);
self.catalog_ready.send_replace(false);
}

/// Background startup: import the login environment (when requested),
/// reload the catalog so PATH discovery sees it, check launchers, then
/// let spawns through and probe models.
pub async fn finish_startup(self: &Arc<Self>) {
// Prompts queued under the previous daemon, after any adopted turn (`queue.rs`).
self.resume_queues();
// The last live model lists, before anything slow.
self.load_live_cache().await;
let login_env = self.login_env_requested.load(Ordering::SeqCst);
let mut reloaded = false;
if login_env {
// A cache whose fingerprint matches applies at once; the shell
// runs only when needed, and agent spawns wait for it (at most
// `login_env::TIMEOUT`) only when nothing else can stand in.
let plan = tokio::task::spawn_blocking(crate::login_env::begin)
.await
.unwrap_or(crate::login_env::Plan { run_shell: true, wait: true });
let cached = crate::login_env::outcome() == crate::login_env::Outcome::Cached;
if cached {
reloaded = self.after_login_env().await;
}
if plan.run_shell && plan.wait {
if crate::login_env::import().await {
reloaded = self.after_login_env().await || reloaded;
}
} else if plan.run_shell {
let hub = self.clone();
tokio::spawn(async move {
if crate::login_env::import().await {
hub.after_login_env().await;
}
});
}
}
self.catalog_ready.send_replace(true);
self.verify_launchers().await;
self.startup_ready.send_replace(true);
tracing::info!("startup complete; agents may spawn");
// Probes of earlier runs left transcripts in the unattended folder: remove them.
let _ = tokio::task::spawn_blocking(|| {
crate::probe_cleanup::remove_probe_transcripts(
&crate::protected_folders::unattended_cwd(),
)
})
.await;
// A catalog reload already started fresh probes.
if !reloaded {
self.probe_models().await;
}
}

/// The login environment changed: `CLAUDE_CONFIG_DIR` / `CODEX_HOME`
/// and harnesses on the login PATH may come from it only. Watchers hear
/// the new catalog, so a client that listed the old one catches up.
async fn after_login_env(self: &Arc<Self>) -> bool {
self.set_harness_homes(crate::adopt::HarnessHomes::from_env());
self.reload_and_announce().await
}

pub fn startup_complete(&self) -> bool {
*self.startup_ready.borrow()
}

/// Wait until background startup has finished (immediate outside a daemon).
pub(super) async fn wait_startup(&self) {
let mut rx = self.startup_ready.subscribe();
let _ = rx.wait_for(|ready| *ready).await;
}

/// Run `--version` on proxy launchers off the executor and mark the ones
/// that fail unavailable (and drop fallbacks that point at them).
async fn verify_launchers(&self) {
let mut probe = self.config.read().await.clone();
let before = probe.harnesses.clone();
let Ok(probe) = tokio::task::spawn_blocking(move || {
crate::config::verify_launchers(&mut probe);
probe
})
.await
else {
return;
};
let mut cfg = self.config.write().await;
// Launchers the probe rerouted (claude-sr through the subrouter server), unless a
// reload changed them meanwhile.
for (name, routed) in &probe.harnesses {
let unchanged =
cfg.harnesses.get(name).map(|p| &p.argv) == before.get(name).map(|p| &p.argv);
if unchanged
&& before.get(name).map(|p| &p.argv) != Some(&routed.argv)
&& before.contains_key(name)
{
cfg.harnesses.insert(name.clone(), routed.clone());
}
}
for (name, reason) in &probe.unavailable {
let argv = |c: &Config| c.harnesses.get(name).map(|p| p.argv.clone());
if argv(&cfg) != argv(&probe) || cfg.unavailable.contains_key(name) {
continue;
}
for p in cfg.harnesses.values_mut() {
if p.fallback.as_deref() == Some(name.as_str()) {
p.fallback = None;
}
}
cfg.unavailable.insert(name.clone(), reason.clone());
}
}

pub(super) fn load_from_store(self: &Arc<Self>) {
let metas = match self.store.list() {
Ok(m) => m,
Expand Down
126 changes: 126 additions & 0 deletions cmux-tui/crates/acpmux/src/hub/startup.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
//! The hub's startup: hold spawns until the login environment and the
//! catalog are ready, check launchers, then let agents start (moved from
//! `hub/mod.rs` unchanged).

use super::Hub;
use crate::config::Config;
use std::sync::Arc;
use std::sync::atomic::Ordering;

impl Hub {
/// Hold session creation and agent spawns until `finish_startup`.
pub fn begin_startup(&self, login_env: bool) {
self.login_env_requested.store(login_env, Ordering::SeqCst);
self.startup_ready.send_replace(false);
self.catalog_ready.send_replace(false);
}

/// Background startup: import the login environment (when requested),
/// reload the catalog so PATH discovery sees it, check launchers, then
/// let spawns through and probe models.
pub async fn finish_startup(self: &Arc<Self>) {
// Prompts queued under the previous daemon, after any adopted turn (`queue.rs`).
self.resume_queues();
// The last live model lists, before anything slow.
self.load_live_cache().await;
let login_env = self.login_env_requested.load(Ordering::SeqCst);
let mut reloaded = false;
if login_env {
// A cache whose fingerprint matches applies at once; the shell
// runs only when needed, and agent spawns wait for it (at most
// `login_env::TIMEOUT`) only when nothing else can stand in.
let plan = tokio::task::spawn_blocking(crate::login_env::begin)
.await
.unwrap_or(crate::login_env::Plan { run_shell: true, wait: true });
let cached = crate::login_env::outcome() == crate::login_env::Outcome::Cached;
if cached {
reloaded = self.after_login_env().await;
}
if plan.run_shell && plan.wait {
if crate::login_env::import().await {
reloaded = self.after_login_env().await || reloaded;
}
} else if plan.run_shell {
let hub = self.clone();
tokio::spawn(async move {
if crate::login_env::import().await {
hub.after_login_env().await;
}
});
}
}
self.catalog_ready.send_replace(true);
self.verify_launchers().await;
self.startup_ready.send_replace(true);
tracing::info!("startup complete; agents may spawn");
// Probes of earlier runs left transcripts in the unattended folder: remove them.
let _ = tokio::task::spawn_blocking(|| {
crate::probe_cleanup::remove_probe_transcripts(
&crate::protected_folders::unattended_cwd(),
)
})
.await;
// A catalog reload already started fresh probes.
if !reloaded {
self.probe_models().await;
}
}

/// The login environment changed: `CLAUDE_CONFIG_DIR` / `CODEX_HOME`
/// and harnesses on the login PATH may come from it only. Watchers hear
/// the new catalog, so a client that listed the old one catches up.
async fn after_login_env(self: &Arc<Self>) -> bool {
self.set_harness_homes(crate::adopt::HarnessHomes::from_env());
self.reload_and_announce().await
}

pub fn startup_complete(&self) -> bool {
*self.startup_ready.borrow()
}

/// Wait until background startup has finished (immediate outside a daemon).
pub(crate) async fn wait_startup(&self) {
let mut rx = self.startup_ready.subscribe();
let _ = rx.wait_for(|ready| *ready).await;
}

/// Run `--version` on proxy launchers off the executor and mark the ones
/// that fail unavailable (and drop fallbacks that point at them).
async fn verify_launchers(&self) {
let mut probe = self.config.read().await.clone();
let before = probe.harnesses.clone();
let Ok(probe) = tokio::task::spawn_blocking(move || {
crate::config::verify_launchers(&mut probe);
probe
})
.await
else {
return;
};
let mut cfg = self.config.write().await;
// Launchers the probe rerouted (claude-sr through the subrouter server), unless a
// reload changed them meanwhile.
for (name, routed) in &probe.harnesses {
let unchanged =
cfg.harnesses.get(name).map(|p| &p.argv) == before.get(name).map(|p| &p.argv);
if unchanged
&& before.get(name).map(|p| &p.argv) != Some(&routed.argv)
&& before.contains_key(name)
{
cfg.harnesses.insert(name.clone(), routed.clone());
}
}
for (name, reason) in &probe.unavailable {
let argv = |c: &Config| c.harnesses.get(name).map(|p| p.argv.clone());
if argv(&cfg) != argv(&probe) || cfg.unavailable.contains_key(name) {
continue;
}
for p in cfg.harnesses.values_mut() {
if p.fallback.as_deref() == Some(name.as_str()) {
p.fallback = None;
}
}
cfg.unavailable.insert(name.clone(), reason.clone());
}
}
}
Loading