Skip to content
Merged
Show file tree
Hide file tree
Changes from 10 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
19 changes: 6 additions & 13 deletions desktop/src-tauri/src/commands/agent_discovery.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ use crate::managed_agents::{
DEFAULT_ACP_COMMAND,
};

mod forced_single_flight;
mod post_install_verification;

fn active_installs() -> &'static std::sync::Mutex<std::collections::HashSet<String>> {
Expand Down Expand Up @@ -49,23 +50,15 @@ pub(crate) fn plan_adapter_install<'c>(
}
}

/// Discover the ACP runtime catalog. `force: false` (the default) serves the
/// cheap cached path; `force: true` runs the expensive re-discovery. See
/// [`forced_single_flight`] for the split and single-flight coalescing.
#[tauri::command]
pub async fn discover_acp_providers(
app: tauri::AppHandle,
force: Option<bool>,
) -> Result<Vec<AcpRuntimeCatalogEntry>, String> {
tokio::task::spawn_blocking(move || {
use tauri::Manager;
crate::managed_agents::clear_resolve_cache();
crate::managed_agents::refresh_login_shell_path();
let custom_dir = app
.path()
.app_data_dir()
.ok()
.map(|d| d.join("custom_harnesses"));
crate::managed_agents::discover_acp_runtimes_from(custom_dir.as_deref())
})
.await
.map_err(|e| format!("spawn_blocking failed: {e}"))
forced_single_flight::discover(app, force.unwrap_or(false)).await
}

/// Write a user-defined harness definition to `<app-data>/custom_harnesses/<id>.json`.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
//! Discovery execution + single-flight coalescing for the ACP runtime catalog.
//!
//! `force: false` serves from the process caches (no clear, no PATH re-fetch, no
//! CLI auth probes) — the low-millisecond path hot surfaces render from.
//!
//! `force: true` runs the expensive probe pipeline. React Query already dedups
//! the hook consumers; the single-flight here is the seatbelt for non-hook
//! invoke paths, so a burst of forced triggers coalesces onto one in-flight run
//! instead of stacking the pipeline.

use super::AcpRuntimeCatalogEntry;

type BoxedDiscovery = std::pin::Pin<
Box<dyn std::future::Future<Output = Result<Vec<AcpRuntimeCatalogEntry>, String>> + Send>,
>;
type SharedDiscovery = futures_util::future::Shared<BoxedDiscovery>;

fn inflight() -> &'static std::sync::Mutex<Option<SharedDiscovery>> {
use std::sync::{Mutex, OnceLock};
static INFLIGHT: OnceLock<Mutex<Option<SharedDiscovery>>> = OnceLock::new();
INFLIGHT.get_or_init(|| Mutex::new(None))
}

/// Discover the ACP runtime catalog. Cheap calls run directly; forced calls
/// coalesce onto a single shared run (see module docs).
pub(super) async fn discover(
app: tauri::AppHandle,
force: bool,
) -> Result<Vec<AcpRuntimeCatalogEntry>, String> {
if !force {
return run(app, false).await;
}

let shared = {
let mut guard = inflight().lock().unwrap_or_else(|e| e.into_inner());
match guard.as_ref() {
Some(existing) => existing.clone(),
None => {
let fut: BoxedDiscovery = Box::pin(run(app, true));
let shared = futures_util::FutureExt::shared(fut);
*guard = Some(shared.clone());
shared
}
}
};

let result = shared.clone().await;

// Clear the slot so the next forced call re-runs — but only if it still
// points at the future we just awaited (a newer run may have replaced it).
{
let mut guard = inflight().lock().unwrap_or_else(|e| e.into_inner());
if guard
.as_ref()
.is_some_and(|current| current.ptr_eq(&shared))
{
*guard = None;
}
}

result
}

async fn run(app: tauri::AppHandle, force: bool) -> Result<Vec<AcpRuntimeCatalogEntry>, String> {
tokio::task::spawn_blocking(move || {
use tauri::Manager;
if force {
crate::managed_agents::clear_resolve_cache();
crate::managed_agents::refresh_login_shell_path();
}
let custom_dir = app
.path()
.app_data_dir()
.ok()
.map(|d| d.join("custom_harnesses"));
crate::managed_agents::discover_acp_runtimes_from(custom_dir.as_deref(), force)
})
.await
.map_err(|e| format!("spawn_blocking failed: {e}"))
}
132 changes: 78 additions & 54 deletions desktop/src-tauri/src/managed_agents/discovery.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ use crate::managed_agents::{
AcpAvailabilityStatus, AcpRuntimeCatalogEntry, AuthStatus, CommandAvailabilityInfo,
HarnessSource,
};
mod auth_status_cache;
mod presets;
mod runtime_metadata;
#[macro_use]
Expand Down Expand Up @@ -558,25 +559,50 @@ pub fn resolve_command(command: &str) -> Option<PathBuf> {
}
}

// Slow path: resolve and cache.
// Slow path: resolve and cache. Negative results are cached too: an absent
// command must not re-run `resolve_command_uncached` (which spawns a login
// shell via `find_via_login_shell`) on every cheap discovery — that spawn
// on the channel-switch/composer hot path is exactly what this cache exists
// to prevent. `clear_resolve_cache` (run by every forced discovery) is the
// invalidation seam, so a newly-installed binary is still found on refresh.
let result = resolve_command_uncached(command);

if result.is_some() {
if let Ok(mut guard) = cache.lock() {
guard.insert(command.to_string(), result.clone());
}
if let Ok(mut guard) = cache.lock() {
guard.insert(command.to_string(), result.clone());
}

result
}

/// Cache-only command resolution for the cheap discovery path.
///
/// Consults the Buzz-managed shim dir (a filesystem stat, never a spawn) and
/// the resolve cache; on a miss it reports the command absent rather than
/// resolving live via `resolve_command_uncached` → `find_via_login_shell`,
/// which spawns a login shell on the channel-switch / composer hot path — the
/// freeze the cheap path exists to avoid. `resolve_command` (the forced path)
/// is the sole prober and cache populator.
pub fn resolve_command_cached(command: &str) -> Option<PathBuf> {
if let Some(managed) = resolve_buzz_managed_command(command) {
return Some(managed);
}
resolve_cache()
.lock()
.ok()
.and_then(|guard| guard.get(command).cloned())
.flatten()
}

/// Clear the resolve_command cache so that newly-installed binaries are detected.
pub fn clear_resolve_cache() {
let mut guard = resolve_cache().lock().unwrap_or_else(|e| e.into_inner());
guard.clear();
// Also invalidate the adapter-availability cache so a freshly-installed
// adapter is reflected the next time the summary builder checks the badge.
clear_adapter_availability_cache();
// And the auth-status cache so a forced re-discovery re-probes rather than
// reusing stale login state.
auth_status_cache::clear();
}

// ── Adapter availability cache (Phase-2 badge fallback) ─────────────────────
Expand Down Expand Up @@ -773,9 +799,16 @@ fn login_shell_candidates() -> Vec<PathBuf> {
}
}

/// Test-only counter for login-shell spawn attempts (see submodule).
#[cfg(test)]
#[path = "discovery/login_shell_spawn_probe.rs"]
pub(crate) mod login_shell_spawn_probe;

/// Run a command in a login shell (tries zsh then bash on Unix, Git Bash on Windows).
/// Returns trimmed stdout if the command succeeds with non-empty output.
fn run_in_login_shell(args: &[&str]) -> Option<String> {
#[cfg(test)]
login_shell_spawn_probe::record();
for shell in login_shell_candidates() {
let mut cmd = Command::new(&shell);
cmd.args(args);
Expand Down Expand Up @@ -1295,27 +1328,39 @@ struct PartialEntry {
entry: AcpRuntimeCatalogEntry,
}

fn discover_acp_runtime_phase1(runtime: &'static KnownAcpRuntime) -> PartialEntry {
fn discover_acp_runtime_phase1(runtime: &'static KnownAcpRuntime, force: bool) -> PartialEntry {
// Cheap path is cache-only (no login-shell spawn); forced path resolves live.
let resolve = if force {
resolve_command
} else {
resolve_command_cached
};
let adapter_result = runtime
.commands
.iter()
.find_map(|command| find_command(command).map(|path| (*command, path)));
.find_map(|command| resolve(command).map(|path| (*command, path)));

let underlying_cli_found = runtime
.underlying_cli
.map(|cli| find_command(cli).is_some())
.map(|cli| resolve(cli).is_some())
.unwrap_or(false);
let (mut availability, command, binary_path) =
classify_runtime(adapter_result, runtime.underlying_cli, underlying_cli_found);

// For codex-acp: when the adapter resolves as Available, probe its full
// version. An adapter below MIN_CODEX_ACP_VERSION is treated as outdated.
// For codex-acp: when the adapter resolves as Available, determine its full
// version. A forced discovery probes the binary (spawns a subprocess); the
// cheap default path reuses the last cached availability so it stays
// process-free. An adapter below MIN_CODEX_ACP_VERSION is treated as outdated.
if runtime.id == "codex"
&& availability == AcpAvailabilityStatus::Available
&& command.as_deref() == Some("codex-acp")
{
if let Some(path_str) = &binary_path {
availability = codex_adapter_availability(&PathBuf::from(path_str));
if force {
if let Some(path_str) = &binary_path {
availability = codex_adapter_availability(&PathBuf::from(path_str));
}
} else if let Some(cached) = adapter_availability_cached() {
availability = cached;
}
}

Expand All @@ -1328,7 +1373,7 @@ fn discover_acp_runtime_phase1(runtime: &'static KnownAcpRuntime) -> PartialEntr

let underlying_cli_path = runtime
.underlying_cli
.and_then(find_command)
.and_then(resolve)
.map(|p| p.display().to_string());

let default_args = command
Expand Down Expand Up @@ -1373,8 +1418,8 @@ fn discover_acp_runtime_phase1(runtime: &'static KnownAcpRuntime) -> PartialEntr
AcpAvailabilityStatus::AdapterMissing | AcpAvailabilityStatus::NotInstalled
) && runtime_needs_npm(runtime)
&& buzz_managed_node_bin_dir().is_none()
&& resolve_command("npm").is_none()
&& resolve_command("node").is_none();
&& resolve("npm").is_none()
&& resolve("node").is_none();

PartialEntry {
runtime,
Expand Down Expand Up @@ -1415,7 +1460,9 @@ fn discover_acp_runtime_phase1(runtime: &'static KnownAcpRuntime) -> PartialEntr
/// resolves, so it should not pay the cost of authenticating every catalog entry.
pub(crate) fn discover_acp_runtime_availability(runtime_id: &str) -> Option<AcpAvailabilityStatus> {
known_acp_runtime_exact(runtime_id)
.map(discover_acp_runtime_phase1)
// Post-install verification wants fresh filesystem/version state, so
// probe rather than trust the cheap-path cache.
.map(|runtime| discover_acp_runtime_phase1(runtime, true))
.map(|partial| partial.entry.availability)
}

Expand All @@ -1438,47 +1485,24 @@ pub(crate) fn discover_acp_runtime_availability(runtime_id: &str) -> Option<AcpA
/// re-running discovery.
pub fn discover_acp_runtimes_from(
custom_harnesses_dir: Option<&Path>,
force: bool,
) -> Vec<AcpRuntimeCatalogEntry> {
// Cheap path is cache-only (no login-shell spawn); forced path resolves live.
let resolve = if force {
resolve_command
} else {
resolve_command_cached
};

// Phase 1: build all builtin entries (fast — no probes yet).
let mut partials: Vec<PartialEntry> = KNOWN_ACP_RUNTIMES
.iter()
.map(discover_acp_runtime_phase1)
.collect();

// Phase 2: run auth probes in parallel for entries that need them.
// Spawn one thread per probeable entry; total cost = max(probe latency).
let probe_handles: Vec<(usize, std::thread::JoinHandle<AuthStatus>)> = partials
.iter()
.enumerate()
.filter_map(|(idx, partial)| {
if partial.entry.availability != AcpAvailabilityStatus::Available {
return None;
}
let probe_args = partial.runtime.auth_probe_args?;
// Need the resolved binary path for the CLI (e.g. the actual `claude` binary).
let binary_path = resolve_command(probe_args[0])?;
let probe_args_owned: Vec<String> = probe_args.iter().map(|s| s.to_string()).collect();

let handle = std::thread::spawn(move || {
let refs: Vec<&str> = probe_args_owned.iter().map(String::as_str).collect();
probe_auth_status(&binary_path, &refs)
});
Some((idx, handle))
})
.map(|runtime| discover_acp_runtime_phase1(runtime, force))
.collect();

// Collect probe results and patch entries.
for (idx, handle) in probe_handles {
let status = handle.join().unwrap_or(AuthStatus::Unknown);
let partial = &mut partials[idx];
partial.entry.login_hint =
if matches!(status, AuthStatus::LoggedIn | AuthStatus::NotApplicable) {
None
} else {
partial.runtime.login_hint.map(str::to_string)
};
partial.entry.auth_status = status;
}
// Phase 2: resolve each available runtime's auth status (forced discovery
// spawns parallel CLI probes and warms the cache; the cheap path reuses it).
auth_status_cache::resolve_auth_statuses(&mut partials, force);

// Fill NotApplicable / Unknown for non-probed entries.
for partial in &mut partials {
Expand Down Expand Up @@ -1508,7 +1532,7 @@ pub fn discover_acp_runtimes_from(
}
seen_ids.insert(def.id.to_string());

entries.push(preset_catalog_entry(def, find_command));
entries.push(preset_catalog_entry(def, resolve));
}

// Phase 3: load and append custom harness definitions.
Expand All @@ -1523,8 +1547,8 @@ pub fn discover_acp_runtimes_from(
continue;
}

// Availability: command on PATH → Available, else NotInstalled.
let (availability, command, binary_path) = match find_command(&def.command) {
// Availability: command resolves → Available, else NotInstalled.
let (availability, command, binary_path) = match resolve(&def.command) {
Some(path) => (
AcpAvailabilityStatus::Available,
Some(def.command.clone()),
Expand Down
Loading
Loading