mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
Refactor managed-agent runtime into cohesive modules (#2974)
## Summary - split the managed-agent runtime warehouse into cohesive modules for process ownership/termination, orphan sweeping, dead-instance reaping, lifecycle synchronization, and runtime metadata - preserve the existing `runtime` API through narrow re-exports; helper bodies and platform `cfg` branches are unchanged apart from module-qualified visibility - reduce `runtime.rs` from 2,220 lines on `main` to 908 lines and remove its temporary file-size override, restoring the standard 1,000-line ceiling ## Why `main` failed after stale successful PR checks allowed independent growth to combine above `runtime.rs`'s 2,216-line override. The earlier fix in #2974 extracted only 55 lines and left the monolith on a special ratchet. This replacement includes that extraction but establishes responsibility boundaries and removes the exception entirely. ## Module boundaries - `process.rs` — process identity, ownership markers, receipt validation, and termination primitives - `orphan_sweep.rs` — same-instance orphan discovery and cleanup - `instance_reaper.rs` — foreign/dead desktop instance detection and agent reaping - `lifecycle.rs` — tracked runtime synchronization and stale record cleanup - `metadata.rs` — model/provider metadata resolution - `runtime.rs` — summary/config/spawn orchestration and composition ## Validation At `a824fda31eff6ecc0d39ca1b8ea5602a108897e6`: - pre-push `desktop-check` - pre-push `desktop-test` - pre-push full `desktop-tauri-test`: 1,637 passed, 0 failed, 14 ignored; integration + doc tests passed - `cargo check --manifest-path desktop/src-tauri/Cargo.toml --lib` - `cargo fmt --manifest-path desktop/src-tauri/Cargo.toml --all -- --check` - `node desktop/scripts/check-file-sizes.mjs` Supersedes #2974 and #2930. Signed-off-by: Wes <wesbillman@users.noreply.github.com> Co-authored-by: Wes <wesbillman@users.noreply.github.com> Co-authored-by: Princess Donut <68157ebd23b3897c1991015c3038658ea916200c67d3a54620b0754d1b92f6e0@buzz.block.builderlab.xyz>
This commit is contained in:
klopez4212
co-authored by
Wes
Princess Donut
Wes
Princess Donut
parent
dd222a509b
commit
74b63e1846
@@ -119,19 +119,6 @@ const overrides = new Map([
|
||||
// helpers) replace the pubkey-keyed PID file, plus the hashed pair-scoped
|
||||
// runtime log path. Load-bearing crash-recovery surface; queued to split.
|
||||
["src-tauri/src/managed_agents/storage.rs", 1383],
|
||||
// harness-persona-sync: persona-runtime resolution threaded into the spawn
|
||||
// path here. Load-bearing feature growth; queued to split in the resolver
|
||||
// unify refactor followup. +26 for resolve_effective_prompt_model_provider
|
||||
// re-introduced after 826d735fe removal (config-bridge caller still needs it).
|
||||
// PGID resolution helper + PID-recycling safety guard added for orphan sweep.
|
||||
// activity-feed threads avatar_url into build_managed_agent_summary for the
|
||||
// assistant-bubble pinned snapshot.
|
||||
// +1 for agent_pubkey field in setup payload (config-nudge card wire).
|
||||
// persona-blank-fallback: resolve_effective_prompt_model_provider gains a
|
||||
// record_provider param + applies persona_field_with_record_fallback. +5 lines.
|
||||
// global-agent-config: spawn_agent_child loads global config and merges as
|
||||
// lowest env layer (+8 lines). Queued to split.
|
||||
["src-tauri/src/managed_agents/runtime.rs", 2216],
|
||||
// config-bridge setup-payload env-boundary fix adds readiness wiring in
|
||||
// spawn_agent_child; load-bearing security fix, queued to split.
|
||||
["src-tauri/src/managed_agents/config_bridge/reader.rs", 1016],
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,359 @@
|
||||
use super::*;
|
||||
|
||||
/// Binary names for the Buzz desktop/Tauri process. Used by dead-instance
|
||||
/// detection to confirm the owning desktop is still alive.
|
||||
const DESKTOP_BINARY_NAMES: &[&str] = &[
|
||||
"Buzz",
|
||||
"buzz-desktop",
|
||||
"buzz_desktop",
|
||||
// Linux limits /proc/<pid>/comm to 15 visible bytes, truncating the
|
||||
// AppImage shim's real executable name, `buzz-desktop.bin`.
|
||||
"buzz-desktop.bi",
|
||||
];
|
||||
|
||||
/// Check if a process name matches a known Buzz desktop binary.
|
||||
pub(super) fn is_desktop_binary(name: &str) -> bool {
|
||||
DESKTOP_BINARY_NAMES.contains(&name)
|
||||
}
|
||||
|
||||
/// Check whether `buf` contains `id` as a complete identifier — not as a
|
||||
/// prefix of a longer dotted name. The identifier appears in the Tauri config
|
||||
/// JSON as `"identifier":"xyz.block.buzz.app.dev"` and in environment entries
|
||||
/// as `KEY=...app.dev\0`, so a valid match is followed by a non-identifier byte
|
||||
/// (not `[A-Za-z0-9._-]`) or sits at the end of the buffer. This prevents
|
||||
/// `xyz.block.buzz.app` from matching inside `xyz.block.buzz.app.dev`.
|
||||
pub(super) fn buffer_contains_identifier(buf: &[u8], id: &[u8]) -> bool {
|
||||
if id.is_empty() {
|
||||
return false;
|
||||
}
|
||||
buf.windows(id.len()).enumerate().any(|(i, w)| {
|
||||
if w != id {
|
||||
return false;
|
||||
}
|
||||
// Boundary check on the byte immediately after the match: end-of-buffer
|
||||
// or any byte that can't continue a dotted reverse-DNS identifier.
|
||||
match buf.get(i + id.len()) {
|
||||
None => true,
|
||||
Some(&next) => {
|
||||
!next.is_ascii_alphanumeric() && next != b'.' && next != b'_' && next != b'-'
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Extract the `BUZZ_MANAGED_AGENT` value from a process's environment.
|
||||
/// Returns `None` if the process doesn't have the marker or can't be read.
|
||||
#[cfg(target_os = "macos")]
|
||||
fn extract_buzz_marker_value(pid: u32) -> Option<String> {
|
||||
let prefix = b"BUZZ_MANAGED_AGENT=";
|
||||
let buf = sweep::procargs2_buffer(pid)?;
|
||||
|
||||
if buf.len() < std::mem::size_of::<libc::c_int>() {
|
||||
return None;
|
||||
}
|
||||
let mut n_args: libc::c_int = 0;
|
||||
unsafe {
|
||||
std::ptr::copy_nonoverlapping(
|
||||
buf.as_ptr(),
|
||||
&mut n_args as *mut libc::c_int as *mut u8,
|
||||
std::mem::size_of::<libc::c_int>(),
|
||||
);
|
||||
}
|
||||
let mut pos = std::mem::size_of::<libc::c_int>();
|
||||
|
||||
// Skip exec path.
|
||||
while pos < buf.len() && buf[pos] != 0 {
|
||||
pos += 1;
|
||||
}
|
||||
while pos < buf.len() && buf[pos] == 0 {
|
||||
pos += 1;
|
||||
}
|
||||
// Skip argc argument strings.
|
||||
let mut args_remaining = n_args;
|
||||
while args_remaining > 0 && pos < buf.len() {
|
||||
while pos < buf.len() && buf[pos] != 0 {
|
||||
pos += 1;
|
||||
}
|
||||
while pos < buf.len() && buf[pos] == 0 {
|
||||
pos += 1;
|
||||
}
|
||||
args_remaining -= 1;
|
||||
}
|
||||
// Search environment entries for our marker.
|
||||
for entry in buf[pos..].split(|&b| b == 0) {
|
||||
if entry.starts_with(prefix) {
|
||||
return String::from_utf8(entry[prefix.len()..].to_vec()).ok();
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
fn extract_buzz_marker_value(pid: u32) -> Option<String> {
|
||||
let prefix = b"BUZZ_MANAGED_AGENT=";
|
||||
let data = std::fs::read(format!("/proc/{pid}/environ")).ok()?;
|
||||
for entry in data.split(|&b| b == 0) {
|
||||
if entry.starts_with(prefix) {
|
||||
return String::from_utf8(entry[prefix.len()..].to_vec()).ok();
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
fn extract_buzz_marker_value(_pid: u32) -> Option<String> {
|
||||
None
|
||||
}
|
||||
|
||||
/// Check if a Buzz desktop process is still alive for the given instance ID.
|
||||
/// Scans all user-owned processes named "Buzz" or "buzz-desktop" and checks
|
||||
/// whether any has the identifier in its command-line args (KERN_PROCARGS2 buffer
|
||||
/// includes both argv and environ — the `--config` JSON from `tauri dev` contains
|
||||
/// the identifier string).
|
||||
#[cfg(target_os = "macos")]
|
||||
fn desktop_is_alive_for_instance(instance_id: &str) -> bool {
|
||||
extern "C" {
|
||||
fn proc_name(pid: libc::c_int, buffer: *mut libc::c_void, buffersize: u32) -> libc::c_int;
|
||||
}
|
||||
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let identifier_bytes = instance_id.as_bytes();
|
||||
|
||||
let pids = sweep::collect_all_pids();
|
||||
if pids.is_empty() {
|
||||
return false;
|
||||
}
|
||||
|
||||
for &pid in &pids {
|
||||
if pid <= 0 {
|
||||
continue;
|
||||
}
|
||||
// Check binary name — only look at desktop binaries.
|
||||
let mut name_buf = [0u8; 1024];
|
||||
let len = unsafe {
|
||||
proc_name(
|
||||
pid,
|
||||
name_buf.as_mut_ptr() as *mut libc::c_void,
|
||||
name_buf.len() as u32,
|
||||
)
|
||||
};
|
||||
if len <= 0 {
|
||||
continue;
|
||||
}
|
||||
let name = String::from_utf8_lossy(&name_buf[..len as usize]);
|
||||
if !is_desktop_binary(&name) {
|
||||
continue;
|
||||
}
|
||||
// Verify UID.
|
||||
let mut info = std::mem::MaybeUninit::<BSDInfo>::zeroed();
|
||||
let ret = unsafe {
|
||||
proc_pidinfo(
|
||||
pid,
|
||||
PROC_PIDTBSDINFO,
|
||||
0,
|
||||
info.as_mut_ptr() as *mut libc::c_void,
|
||||
std::mem::size_of::<BSDInfo>() as libc::c_int,
|
||||
)
|
||||
};
|
||||
if ret <= 0 {
|
||||
continue;
|
||||
}
|
||||
let info = unsafe { info.assume_init() };
|
||||
if info.pbi_uid != my_uid {
|
||||
continue;
|
||||
}
|
||||
// Check if this desktop process's args/env contain the identifier.
|
||||
// The KERN_PROCARGS2 buffer holds argv + environ as null-delimited strings.
|
||||
let Some(args_buf) = sweep::procargs2_buffer(pid as u32) else {
|
||||
continue;
|
||||
};
|
||||
// Boundary-anchored search: the identifier in the config JSON is
|
||||
// followed by a non-identifier char (typically `"`). A raw substring
|
||||
// match would let `...app` match inside `...app.dev`.
|
||||
if buffer_contains_identifier(&args_buf, identifier_bytes) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
fn desktop_is_alive_for_instance(instance_id: &str) -> bool {
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let Ok(entries) = std::fs::read_dir("/proc") else {
|
||||
return false;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let name = entry.file_name();
|
||||
let Some(name_str) = name.to_str() else {
|
||||
continue;
|
||||
};
|
||||
let Ok(pid) = name_str.parse::<u32>() else {
|
||||
continue;
|
||||
};
|
||||
// Check ownership.
|
||||
let Ok(meta) = entry.metadata() else {
|
||||
continue;
|
||||
};
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
if meta.uid() != my_uid {
|
||||
continue;
|
||||
}
|
||||
// Check binary name via /proc/<pid>/comm.
|
||||
let Ok(comm) = std::fs::read_to_string(format!("/proc/{pid}/comm")) else {
|
||||
continue;
|
||||
};
|
||||
if !is_desktop_binary(comm.trim()) {
|
||||
continue;
|
||||
}
|
||||
// Check cmdline for the identifier with boundary anchoring.
|
||||
let Ok(cmdline) = std::fs::read(format!("/proc/{pid}/cmdline")) else {
|
||||
continue;
|
||||
};
|
||||
if buffer_contains_identifier(&cmdline, instance_id.as_bytes()) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
fn desktop_is_alive_for_instance(_instance_id: &str) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// Reap agent processes belonging to dead Buzz desktop instances.
|
||||
///
|
||||
/// Scans all user processes for `BUZZ_MANAGED_AGENT=*`, groups them by
|
||||
/// instance ID, and for each foreign instance (≠ `our_instance_id`) checks
|
||||
/// whether a Buzz desktop binary is still alive for that instance. If not,
|
||||
/// all agents from that dead instance are reaped.
|
||||
#[cfg(target_os = "macos")]
|
||||
pub(crate) fn reap_dead_instance_agents(our_instance_id: &str, skip_pids: &[u32]) {
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let my_pid = std::process::id() as i32;
|
||||
|
||||
let pids = sweep::collect_all_pids();
|
||||
if pids.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
// Collect (pid, instance_id) for all foreign agent processes.
|
||||
let mut foreign_agents: HashMap<String, Vec<i32>> = HashMap::new();
|
||||
|
||||
for &pid in &pids {
|
||||
if pid <= 0 || pid == my_pid {
|
||||
continue;
|
||||
}
|
||||
let upid = pid as u32;
|
||||
if skip_pids.contains(&upid) {
|
||||
continue;
|
||||
}
|
||||
if !process_belongs_to_us(upid) {
|
||||
continue;
|
||||
}
|
||||
// Verify UID.
|
||||
let mut info = std::mem::MaybeUninit::<BSDInfo>::zeroed();
|
||||
let ret = unsafe {
|
||||
proc_pidinfo(
|
||||
pid,
|
||||
PROC_PIDTBSDINFO,
|
||||
0,
|
||||
info.as_mut_ptr() as *mut libc::c_void,
|
||||
std::mem::size_of::<BSDInfo>() as libc::c_int,
|
||||
)
|
||||
};
|
||||
if ret <= 0 {
|
||||
continue;
|
||||
}
|
||||
let info = unsafe { info.assume_init() };
|
||||
if info.pbi_uid != my_uid {
|
||||
continue;
|
||||
}
|
||||
// Extract the instance ID from this agent's env.
|
||||
let Some(agent_instance_id) = extract_buzz_marker_value(upid) else {
|
||||
continue;
|
||||
};
|
||||
// Skip agents belonging to our own instance (handled by sweep_system_agent_processes).
|
||||
if agent_instance_id == our_instance_id {
|
||||
continue;
|
||||
}
|
||||
foreign_agents
|
||||
.entry(agent_instance_id)
|
||||
.or_default()
|
||||
.push(pid);
|
||||
}
|
||||
|
||||
// For each foreign instance, check if its desktop is still alive.
|
||||
for (instance_id, agent_pids) in &foreign_agents {
|
||||
if desktop_is_alive_for_instance(instance_id) {
|
||||
continue;
|
||||
}
|
||||
eprintln!(
|
||||
"buzz-desktop: reaping {} orphaned agent(s) from dead instance '{instance_id}'",
|
||||
agent_pids.len()
|
||||
);
|
||||
resolve_pgids_and_kill(agent_pids);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
pub(crate) fn reap_dead_instance_agents(our_instance_id: &str, skip_pids: &[u32]) {
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let my_pid = std::process::id() as i32;
|
||||
let mut foreign_agents: HashMap<String, Vec<i32>> = HashMap::new();
|
||||
|
||||
let Ok(entries) = std::fs::read_dir("/proc") else {
|
||||
return;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let name = entry.file_name();
|
||||
let Some(name_str) = name.to_str() else {
|
||||
continue;
|
||||
};
|
||||
let Ok(pid) = name_str.parse::<i32>() else {
|
||||
continue;
|
||||
};
|
||||
if pid <= 0 || pid == my_pid {
|
||||
continue;
|
||||
}
|
||||
let upid = pid as u32;
|
||||
if skip_pids.contains(&upid) {
|
||||
continue;
|
||||
}
|
||||
let Ok(meta) = entry.metadata() else {
|
||||
continue;
|
||||
};
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
if meta.uid() != my_uid {
|
||||
continue;
|
||||
}
|
||||
if !process_belongs_to_us(upid) {
|
||||
continue;
|
||||
}
|
||||
let Some(agent_instance_id) = extract_buzz_marker_value(upid) else {
|
||||
continue;
|
||||
};
|
||||
if agent_instance_id == our_instance_id {
|
||||
continue;
|
||||
}
|
||||
foreign_agents
|
||||
.entry(agent_instance_id)
|
||||
.or_default()
|
||||
.push(pid);
|
||||
}
|
||||
|
||||
for (instance_id, agent_pids) in &foreign_agents {
|
||||
if desktop_is_alive_for_instance(instance_id) {
|
||||
continue;
|
||||
}
|
||||
eprintln!(
|
||||
"buzz-desktop: reaping {} orphaned agent(s) from dead instance '{instance_id}'",
|
||||
agent_pids.len()
|
||||
);
|
||||
resolve_pgids_and_kill(agent_pids);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn reap_dead_instance_agents(_our_instance_id: &str, _skip_pids: &[u32]) {}
|
||||
@@ -0,0 +1,105 @@
|
||||
use super::*;
|
||||
|
||||
/// Kill stale agent processes from a previous session whose PID is still alive
|
||||
/// but not tracked in the current `runtimes` map. Updates the record fields and
|
||||
/// returns `true` if any records were modified.
|
||||
pub fn kill_stale_tracked_processes(
|
||||
records: &mut [ManagedAgentRecord],
|
||||
runtimes: &HashMap<ManagedAgentRuntimeKey, ManagedAgentPairRuntime>,
|
||||
instance_id: &str,
|
||||
) -> bool {
|
||||
use crate::managed_agents::BackendKind;
|
||||
|
||||
let mut changed = false;
|
||||
for record in records.iter_mut() {
|
||||
if record.backend != BackendKind::Local {
|
||||
continue;
|
||||
}
|
||||
let Some(pid) = record.runtime_pid else {
|
||||
continue;
|
||||
};
|
||||
if !runtimes.keys().any(|key| key.pubkey == record.pubkey) {
|
||||
if process_belongs_to_us(pid) && process_has_buzz_marker(pid, instance_id) {
|
||||
let _ = terminate_process(pid);
|
||||
}
|
||||
record.runtime_pid = None;
|
||||
record.last_stopped_at = Some(crate::util::now_iso());
|
||||
record.updated_at = crate::util::now_iso();
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
changed
|
||||
}
|
||||
|
||||
pub fn sync_managed_agent_processes(
|
||||
records: &mut [ManagedAgentRecord],
|
||||
runtimes: &mut HashMap<ManagedAgentRuntimeKey, ManagedAgentPairRuntime>,
|
||||
_instance_id: &str,
|
||||
) -> (bool, Vec<String>) {
|
||||
let mut changed = false;
|
||||
let mut exited = Vec::new();
|
||||
|
||||
for (key, runtime) in runtimes.iter_mut() {
|
||||
let status = match runtime.child.try_wait() {
|
||||
Ok(status) => status,
|
||||
Err(error) => {
|
||||
if let Some(record) = records
|
||||
.iter_mut()
|
||||
.find(|record| record.pubkey == key.pubkey)
|
||||
{
|
||||
record.updated_at = now_iso();
|
||||
record.last_error = Some(format!("failed to inspect process state: {error}"));
|
||||
record.last_error_code = None;
|
||||
}
|
||||
changed = true;
|
||||
exited.push(key.clone());
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let Some(status) = status else {
|
||||
continue;
|
||||
};
|
||||
|
||||
if let Some(record) = records
|
||||
.iter_mut()
|
||||
.find(|record| record.pubkey == key.pubkey)
|
||||
{
|
||||
record.updated_at = now_iso();
|
||||
record.last_stopped_at = Some(now_iso());
|
||||
record.last_exit_code = status.code();
|
||||
let log_err = if status.success() {
|
||||
None
|
||||
} else {
|
||||
Some(
|
||||
super::super::meaningful_agent_error_from_log(&runtime.log_path)
|
||||
.unwrap_or_else(|| super::super::storage::AgentLogError {
|
||||
message: format!("harness exited with status {status}"),
|
||||
code: None,
|
||||
}),
|
||||
)
|
||||
};
|
||||
record.last_error = log_err.as_ref().map(|e| e.message.clone());
|
||||
record.last_error_code = log_err.as_ref().and_then(|e| e.code);
|
||||
}
|
||||
|
||||
changed = true;
|
||||
exited.push(key.clone());
|
||||
}
|
||||
|
||||
let exited_pubkeys: Vec<String> = exited.iter().map(|key| key.pubkey.clone()).collect();
|
||||
for key in exited {
|
||||
runtimes.remove(&key);
|
||||
}
|
||||
|
||||
// `runtime_pid` is legacy bookkeeping. Pair runtimes and receipts are the
|
||||
// authoritative lifecycle source; migration cleanup is handled separately.
|
||||
for record in records.iter_mut() {
|
||||
if record.runtime_pid.take().is_some() {
|
||||
record.updated_at = now_iso();
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
|
||||
(changed, exited_pubkeys)
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
/// Returns the (key, value) env var pairs that should be forwarded to the
|
||||
/// agent process for model and provider selection.
|
||||
///
|
||||
/// Model injection is unconditional — even agents that support ACP model
|
||||
/// switching need the initial bootstrap value. Provider injection is skipped
|
||||
/// when `provider_locked` is true (e.g. Claude runtimes that only work with
|
||||
/// Anthropic).
|
||||
pub(crate) fn runtime_metadata_env_vars<'a>(
|
||||
model_env_var: Option<&'a str>,
|
||||
provider_env_var: Option<&'a str>,
|
||||
provider_locked: bool,
|
||||
effective_model: Option<&'a str>,
|
||||
effective_provider: Option<&'a str>,
|
||||
) -> Vec<(&'a str, &'a str)> {
|
||||
let mut vars = Vec::new();
|
||||
if let (Some(env_key), Some(model)) = (model_env_var, effective_model) {
|
||||
vars.push((env_key, model));
|
||||
}
|
||||
if !provider_locked {
|
||||
if let (Some(env_key), Some(provider)) = (provider_env_var, effective_provider) {
|
||||
vars.push((env_key, provider));
|
||||
}
|
||||
}
|
||||
vars
|
||||
}
|
||||
|
||||
/// Resolve the effective (prompt, model, provider) triple for a persona-linked agent.
|
||||
///
|
||||
/// Given a persona_id, finds the persona in the list and returns its system_prompt,
|
||||
/// model, and provider as the authoritative values. When the persona leaves `model`
|
||||
/// or `provider` blank (None or whitespace-only), falls back to the record's own
|
||||
/// field using the same precedence rule as `persona_snapshot_with_agent_config_fallback`
|
||||
/// so the display surface matches spawn behavior. Falls back to the record's own
|
||||
/// prompt/model/provider when no persona is linked or found.
|
||||
///
|
||||
/// Used by `agent_config.rs` to inject persona defaults into the config surface
|
||||
/// before running the reader, so BuzzExplicit-tagged fields can be re-tagged to
|
||||
/// PersonaDefault for fields the record did not independently set.
|
||||
pub(crate) fn resolve_effective_prompt_model_provider(
|
||||
persona_id: Option<&str>,
|
||||
personas: &[crate::managed_agents::types::AgentDefinition],
|
||||
record_prompt: Option<String>,
|
||||
record_model: Option<String>,
|
||||
record_provider: Option<String>,
|
||||
) -> (Option<String>, Option<String>, Option<String>) {
|
||||
let fallback = crate::managed_agents::persona_events::persona_field_with_record_fallback;
|
||||
match persona_id.and_then(|pid| personas.iter().find(|p| p.id == pid)) {
|
||||
Some(p) => (
|
||||
Some(p.system_prompt.clone()),
|
||||
fallback(p.model.as_deref(), record_model.as_deref()), // fallback: record.model
|
||||
fallback(p.provider.as_deref(), record_provider.as_deref()), // fallback: record.provider
|
||||
),
|
||||
None => (record_prompt, record_model, record_provider),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,371 @@
|
||||
use super::*;
|
||||
|
||||
/// Kill orphaned agent processes using PID file receipts. Reads all files from
|
||||
/// `agent-pids/`, verifies each PID still belongs to a known agent binary,
|
||||
/// then resolves each candidate's actual PGID and signals the process group.
|
||||
/// Deletes the PID file after killing.
|
||||
///
|
||||
/// `skip_pids` are PIDs already handled by the tracked-agent path.
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn sweep_orphaned_agent_processes(app: &AppHandle, skip_pids: &[u32]) {
|
||||
let legacy_entries = super::super::read_all_agent_pid_files(app);
|
||||
let instance_id = current_instance_id(app);
|
||||
let receipt_entries: Vec<_> = super::super::read_all_agent_runtime_receipts(app)
|
||||
.into_iter()
|
||||
.filter_map(|(path, receipt)| {
|
||||
if valid_agent_runtime_receipt(&path, &receipt, &instance_id) {
|
||||
Some((path, receipt))
|
||||
} else {
|
||||
super::super::remove_agent_runtime_receipt_path(&path);
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
// Collect live orphans AND dead-leader groups into a single kill batch.
|
||||
// Dead leaders: PGID may have been recycled, but the window is narrow
|
||||
// (PID files are from this session) and the cost of missing surviving
|
||||
// group members outweighs the recycling risk.
|
||||
let targets: Vec<i32> = legacy_entries
|
||||
.iter()
|
||||
.map(|(_, pid)| *pid)
|
||||
.chain(receipt_entries.iter().map(|(_, receipt)| receipt.pid))
|
||||
.filter(|pid| {
|
||||
if skip_pids.contains(pid) {
|
||||
return false;
|
||||
}
|
||||
(process_is_running(*pid) && process_belongs_to_us(*pid)) || !process_is_running(*pid)
|
||||
})
|
||||
.map(|pid| pid as i32)
|
||||
.collect();
|
||||
|
||||
if !targets.is_empty() {
|
||||
resolve_pgids_and_kill(&targets);
|
||||
}
|
||||
|
||||
// Clean up PID files for processes we just killed or that are already gone.
|
||||
for (pubkey, pid) in &legacy_entries {
|
||||
if skip_pids.contains(pid) {
|
||||
continue;
|
||||
}
|
||||
if !process_is_running(*pid) || !process_belongs_to_us(*pid) {
|
||||
super::super::remove_agent_pid_file(app, pubkey);
|
||||
}
|
||||
}
|
||||
for (_, receipt) in &receipt_entries {
|
||||
if skip_pids.contains(&receipt.pid) {
|
||||
continue;
|
||||
}
|
||||
if !process_is_running(receipt.pid) || !process_belongs_to_us(receipt.pid) {
|
||||
super::super::remove_agent_runtime_receipt(app, &receipt.key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn sweep_orphaned_agent_processes(app: &AppHandle, _skip_pids: &[u32]) {
|
||||
let _ = app;
|
||||
}
|
||||
|
||||
// ── macOS process-info FFI (shared by all sweep/reap functions) ──────────
|
||||
//
|
||||
// `proc_listallpids` lives in `sweep.rs` (which owns `collect_all_pids`).
|
||||
// All callers in this file reach it through `sweep::collect_all_pids()`.
|
||||
// `proc_pidinfo` and `BSDInfo` are declared here as `pub(super)` so that
|
||||
// `sweep.rs` can call `super::proc_pidinfo` / use `super::BSDInfo` without
|
||||
// redefining the struct layout in two places.
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
extern "C" {
|
||||
pub(super) fn proc_pidinfo(
|
||||
pid: libc::c_int,
|
||||
flavor: libc::c_int,
|
||||
arg: u64,
|
||||
buffer: *mut libc::c_void,
|
||||
buffersize: libc::c_int,
|
||||
) -> libc::c_int;
|
||||
}
|
||||
|
||||
/// Subset of `struct proc_bsdinfo` from `<sys/proc_info.h>`. Layout verified
|
||||
/// against the macOS SDK — total size 136 bytes.
|
||||
#[cfg(target_os = "macos")]
|
||||
#[repr(C)]
|
||||
pub(super) struct BSDInfo {
|
||||
_flags_status_xstatus: [u8; 12], // pbi_flags + pbi_status + pbi_xstatus
|
||||
pub(super) pbi_pid: u32, // offset 12
|
||||
pub(super) pbi_ppid: u32, // offset 16
|
||||
pub(super) pbi_uid: u32, // offset 20
|
||||
_rest: [u8; 112],
|
||||
}
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
const _: () = assert!(std::mem::size_of::<BSDInfo>() == 136);
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
pub(super) const PROC_PIDTBSDINFO: libc::c_int = 3;
|
||||
|
||||
/// Enumerate all processes on the system owned by the current user and kill any
|
||||
/// agent binary stamped with *this* instance's `BUZZ_MANAGED_AGENT` marker
|
||||
/// (`instance_id`) that isn't in `skip_pids`. This catches orphans that escaped
|
||||
/// PID-file-based cleanup (e.g. agent workers spawned with their own process
|
||||
/// group whose parent harness already exited and had its PID file removed),
|
||||
/// while leaving another live Buzz instance's agents untouched.
|
||||
#[cfg(target_os = "macos")]
|
||||
pub(crate) fn sweep_system_agent_processes(instance_id: &str, skip_pids: &[u32]) {
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let pids = sweep::collect_all_pids();
|
||||
if pids.is_empty() {
|
||||
return;
|
||||
}
|
||||
let my_pid = std::process::id() as i32;
|
||||
let mut orphans: Vec<i32> = Vec::new();
|
||||
|
||||
for &pid in &pids {
|
||||
if pid <= 0 {
|
||||
continue;
|
||||
}
|
||||
let upid = pid as u32;
|
||||
if skip_pids.contains(&upid) || pid == my_pid {
|
||||
continue;
|
||||
}
|
||||
// Check binary name first (cheap proc_name call) before UID lookup.
|
||||
if !process_belongs_to_us(upid) {
|
||||
continue;
|
||||
}
|
||||
// Verify UID and PPID via proc_pidinfo.
|
||||
let mut info = std::mem::MaybeUninit::<BSDInfo>::zeroed();
|
||||
let ret = unsafe {
|
||||
proc_pidinfo(
|
||||
pid,
|
||||
PROC_PIDTBSDINFO,
|
||||
0,
|
||||
info.as_mut_ptr() as *mut libc::c_void,
|
||||
std::mem::size_of::<BSDInfo>() as libc::c_int,
|
||||
)
|
||||
};
|
||||
if ret <= 0 {
|
||||
continue;
|
||||
}
|
||||
let info = unsafe { info.assume_init() };
|
||||
if info.pbi_uid != my_uid {
|
||||
continue;
|
||||
}
|
||||
if !process_has_buzz_marker(upid, instance_id) {
|
||||
continue;
|
||||
}
|
||||
// Live descendants of a tracked harness are exempt — see sweep::is_live_descendant_*.
|
||||
if sweep::is_live_descendant_macos(upid, info.pbi_ppid, skip_pids) {
|
||||
continue;
|
||||
}
|
||||
orphans.push(pid);
|
||||
}
|
||||
|
||||
if !orphans.is_empty() {
|
||||
eprintln!(
|
||||
"buzz-desktop: system sweep found {} orphaned agent process(es), cleaning up",
|
||||
orphans.len()
|
||||
);
|
||||
resolve_pgids_and_kill(&orphans);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
pub(crate) fn sweep_system_agent_processes(instance_id: &str, skip_pids: &[u32]) {
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let mut orphans: Vec<i32> = Vec::new();
|
||||
let my_pid = std::process::id() as i32;
|
||||
|
||||
let Ok(entries) = std::fs::read_dir("/proc") else {
|
||||
return;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let name = entry.file_name();
|
||||
let Some(name_str) = name.to_str() else {
|
||||
continue;
|
||||
};
|
||||
let Ok(pid) = name_str.parse::<i32>() else {
|
||||
continue;
|
||||
};
|
||||
if pid <= 0 || pid == my_pid {
|
||||
continue;
|
||||
}
|
||||
let upid = pid as u32;
|
||||
if skip_pids.contains(&upid) {
|
||||
continue;
|
||||
}
|
||||
// Check ownership via /proc/<pid> metadata.
|
||||
let Ok(meta) = entry.metadata() else {
|
||||
continue;
|
||||
};
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
if meta.uid() != my_uid {
|
||||
continue;
|
||||
}
|
||||
if !process_belongs_to_us(upid) || !process_has_buzz_marker(upid, instance_id) {
|
||||
continue;
|
||||
}
|
||||
// Live descendants of a tracked harness are exempt — see sweep::is_live_descendant_*.
|
||||
if sweep::is_live_descendant_linux(upid, skip_pids) {
|
||||
continue;
|
||||
}
|
||||
orphans.push(pid);
|
||||
}
|
||||
|
||||
if !orphans.is_empty() {
|
||||
eprintln!(
|
||||
"buzz-desktop: system sweep found {} orphaned agent process(es), cleaning up",
|
||||
orphans.len()
|
||||
);
|
||||
resolve_pgids_and_kill(&orphans);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn sweep_system_agent_processes(_instance_id: &str, _skip_pids: &[u32]) {}
|
||||
|
||||
/// Periodic-sweep variant with two-tick grace: only reaps same-instance orphans
|
||||
/// that were also seen orphaned on the previous tick. This prevents killing a
|
||||
/// legitimately-starting agent that spawned between the skip-list snapshot and
|
||||
/// the process scan. Returns the current orphan set for use as `prev_orphans`
|
||||
/// on the next tick.
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn sweep_system_agent_processes_with_grace(
|
||||
instance_id: &str,
|
||||
skip_pids: &[u32],
|
||||
prev_orphans: &std::collections::HashSet<u32>,
|
||||
) -> std::collections::HashSet<u32> {
|
||||
let current = collect_same_instance_orphans(instance_id, skip_pids);
|
||||
// Only reap PIDs seen orphaned on two consecutive ticks.
|
||||
let confirmed: Vec<i32> = current
|
||||
.iter()
|
||||
.filter(|pid| prev_orphans.contains(pid))
|
||||
.map(|&pid| pid as i32)
|
||||
.collect();
|
||||
if !confirmed.is_empty() {
|
||||
eprintln!(
|
||||
"buzz-desktop: periodic sweep confirmed {} orphaned agent process(es), cleaning up",
|
||||
confirmed.len()
|
||||
);
|
||||
resolve_pgids_and_kill(&confirmed);
|
||||
}
|
||||
current
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn sweep_system_agent_processes_with_grace(
|
||||
_instance_id: &str,
|
||||
_skip_pids: &[u32],
|
||||
_prev_orphans: &std::collections::HashSet<u32>,
|
||||
) -> std::collections::HashSet<u32> {
|
||||
std::collections::HashSet::new()
|
||||
}
|
||||
|
||||
/// Collect PIDs of same-instance agent processes that appear orphaned (not in
|
||||
/// `skip_pids`). Returns the set for use in two-tick grace logic — does NOT
|
||||
/// kill anything.
|
||||
#[cfg(target_os = "macos")]
|
||||
pub(crate) fn collect_same_instance_orphans(
|
||||
instance_id: &str,
|
||||
skip_pids: &[u32],
|
||||
) -> std::collections::HashSet<u32> {
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let my_pid = std::process::id() as i32;
|
||||
let mut orphans = std::collections::HashSet::new();
|
||||
|
||||
let pids = sweep::collect_all_pids();
|
||||
if pids.is_empty() {
|
||||
return orphans;
|
||||
}
|
||||
|
||||
for &pid in &pids {
|
||||
if pid <= 0 || pid == my_pid {
|
||||
continue;
|
||||
}
|
||||
let upid = pid as u32;
|
||||
if skip_pids.contains(&upid) {
|
||||
continue;
|
||||
}
|
||||
if !process_belongs_to_us(upid) {
|
||||
continue;
|
||||
}
|
||||
let mut info = std::mem::MaybeUninit::<BSDInfo>::zeroed();
|
||||
let ret = unsafe {
|
||||
proc_pidinfo(
|
||||
pid,
|
||||
PROC_PIDTBSDINFO,
|
||||
0,
|
||||
info.as_mut_ptr() as *mut libc::c_void,
|
||||
std::mem::size_of::<BSDInfo>() as libc::c_int,
|
||||
)
|
||||
};
|
||||
if ret <= 0 {
|
||||
continue;
|
||||
}
|
||||
let info = unsafe { info.assume_init() };
|
||||
if info.pbi_uid != my_uid {
|
||||
continue;
|
||||
}
|
||||
if !process_has_buzz_marker(upid, instance_id) {
|
||||
continue;
|
||||
}
|
||||
// Live descendants of a tracked harness are exempt — see sweep::is_live_descendant_*.
|
||||
if sweep::is_live_descendant_macos(upid, info.pbi_ppid, skip_pids) {
|
||||
continue;
|
||||
}
|
||||
orphans.insert(upid);
|
||||
}
|
||||
orphans
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
pub(crate) fn collect_same_instance_orphans(
|
||||
instance_id: &str,
|
||||
skip_pids: &[u32],
|
||||
) -> std::collections::HashSet<u32> {
|
||||
let my_uid = unsafe { libc::getuid() };
|
||||
let my_pid = std::process::id() as i32;
|
||||
let mut orphans = std::collections::HashSet::new();
|
||||
|
||||
let Ok(entries) = std::fs::read_dir("/proc") else {
|
||||
return orphans;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let name = entry.file_name();
|
||||
let Some(name_str) = name.to_str() else {
|
||||
continue;
|
||||
};
|
||||
let Ok(pid) = name_str.parse::<i32>() else {
|
||||
continue;
|
||||
};
|
||||
if pid <= 0 || pid == my_pid {
|
||||
continue;
|
||||
}
|
||||
let upid = pid as u32;
|
||||
if skip_pids.contains(&upid) {
|
||||
continue;
|
||||
}
|
||||
let Ok(meta) = entry.metadata() else {
|
||||
continue;
|
||||
};
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
if meta.uid() != my_uid {
|
||||
continue;
|
||||
}
|
||||
if !process_belongs_to_us(upid) || !process_has_buzz_marker(upid, instance_id) {
|
||||
continue;
|
||||
}
|
||||
// Live descendants of a tracked harness are exempt — see sweep::is_live_descendant_*.
|
||||
if sweep::is_live_descendant_linux(upid, skip_pids) {
|
||||
continue;
|
||||
}
|
||||
orphans.insert(upid);
|
||||
}
|
||||
orphans
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn collect_same_instance_orphans(
|
||||
_instance_id: &str,
|
||||
_skip_pids: &[u32],
|
||||
) -> std::collections::HashSet<u32> {
|
||||
std::collections::HashSet::new()
|
||||
}
|
||||
@@ -0,0 +1,447 @@
|
||||
use super::*;
|
||||
|
||||
/// Binary name fragments for all known agent/harness processes that Buzz
|
||||
/// may spawn. Used by `process_belongs_to_us()` and the orphan sweep to
|
||||
/// identify processes we should clean up. Both hyphenated and underscored
|
||||
/// variants are listed because macOS `proc_name()` and Linux `/proc/comm`
|
||||
/// may report either form depending on how the binary was built.
|
||||
pub(crate) const KNOWN_AGENT_BINARIES: &[&str] = &[
|
||||
"buzz-acp",
|
||||
"buzz_acp",
|
||||
"buzz-agent",
|
||||
"buzz_agent",
|
||||
"claude-agent-acp",
|
||||
"claude_agent_acp",
|
||||
"claude-code-acp",
|
||||
"claude_code_acp",
|
||||
"codex-acp",
|
||||
"codex_acp",
|
||||
"goose",
|
||||
// buzz-dev-mcp's multicall personalities (rg, tree, buzz,
|
||||
// git-credential-nostr, git-sign-nostr) are short-lived per-tool-call
|
||||
// invocations — not listed here.
|
||||
"buzz-dev-mcp",
|
||||
"buzz_dev_mcp",
|
||||
];
|
||||
|
||||
/// Script interpreters that may host managed agent wrappers (e.g. npm shims).
|
||||
/// A process whose name matches here is NOT immediately claimed — it must also
|
||||
/// carry `BUZZ_MANAGED_AGENT` in its environment (checked by the caller via
|
||||
/// `process_has_buzz_marker()`). This avoids sweeping unrelated node processes.
|
||||
pub(crate) const KNOWN_SCRIPT_INTERPRETERS: &[&str] = &["node"];
|
||||
|
||||
/// Check if a process name matches any of our known agent binaries.
|
||||
/// Uses exact match or prefix-with-separator to avoid false positives
|
||||
/// (e.g. `"goose"` must not match `"mongoose"`).
|
||||
pub(super) fn name_matches_known_binary(name: &str) -> bool {
|
||||
KNOWN_AGENT_BINARIES.iter().any(|&binary| {
|
||||
name == binary || {
|
||||
name.starts_with(binary) && {
|
||||
let rest = &name[binary.len()..];
|
||||
rest.starts_with('-') || rest.starts_with('_') || rest.starts_with('.')
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Check if a process name is a known script interpreter that may be hosting
|
||||
/// a managed agent wrapper (e.g. `node` running an npm shim for `codex-acp`).
|
||||
/// Callers must additionally verify `BUZZ_MANAGED_AGENT` ownership.
|
||||
pub(super) fn name_matches_interpreter(name: &str) -> bool {
|
||||
KNOWN_SCRIPT_INTERPRETERS.contains(&name)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn process_is_running(pid: u32) -> bool {
|
||||
// Use libc::kill with signal 0 instead of forking a subprocess.
|
||||
// Returns true only if the process exists AND we can signal it.
|
||||
// Returns false for non-existent PIDs (ESRCH) and PIDs owned by
|
||||
// other users (EPERM) — callers should not interact with those.
|
||||
unsafe { libc::kill(pid as i32, 0) == 0 }
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn process_is_running(_pid: u32) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// Check if a PID belongs to a known agent process we spawned.
|
||||
/// Returns false for recycled PIDs that now belong to other processes.
|
||||
#[cfg(target_os = "macos")]
|
||||
pub(crate) fn process_belongs_to_us(pid: u32) -> bool {
|
||||
// Use proc_name() from libproc to get the process name without spawning
|
||||
// a subprocess.
|
||||
extern "C" {
|
||||
fn proc_name(pid: libc::c_int, buffer: *mut libc::c_void, buffersize: u32) -> libc::c_int;
|
||||
}
|
||||
let mut buf = [0u8; 1024];
|
||||
let len = unsafe {
|
||||
proc_name(
|
||||
pid as i32,
|
||||
buf.as_mut_ptr() as *mut libc::c_void,
|
||||
buf.len() as u32,
|
||||
)
|
||||
};
|
||||
if len <= 0 {
|
||||
return false;
|
||||
}
|
||||
let name = String::from_utf8_lossy(&buf[..len as usize]);
|
||||
// Fall through for script interpreters (e.g. `node` hosting an npm shim):
|
||||
// the caller's `process_has_buzz_marker()` check decides true ownership.
|
||||
name_matches_known_binary(&name) || name_matches_interpreter(&name)
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
pub(crate) fn process_belongs_to_us(pid: u32) -> bool {
|
||||
// First try /proc/<pid>/comm. Note: comm is truncated to 15 bytes on Linux,
|
||||
// so binaries with names longer than 15 chars (e.g. "claude-agent-acp")
|
||||
// will never match here.
|
||||
if let Ok(name) = std::fs::read_to_string(format!("/proc/{pid}/comm")) {
|
||||
if name_matches_known_binary(name.trim()) {
|
||||
return true;
|
||||
}
|
||||
// Interpreter check: `node` is 4 bytes, never truncated.
|
||||
if name_matches_interpreter(name.trim()) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: read /proc/<pid>/exe which is a symlink to the full binary path.
|
||||
// This is not subject to the 15-byte truncation limit.
|
||||
if let Ok(exe_path) = std::fs::read_link(format!("/proc/{pid}/exe")) {
|
||||
if let Some(basename) = exe_path.file_name().and_then(|n| n.to_str()) {
|
||||
// Fall through for script interpreters — caller checks the marker.
|
||||
return name_matches_known_binary(basename) || name_matches_interpreter(basename);
|
||||
}
|
||||
}
|
||||
|
||||
false
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn process_belongs_to_us(_pid: u32) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// The value stamped into the `BUZZ_MANAGED_AGENT` env var of every agent we
|
||||
/// spawn, identifying *which* desktop instance owns it. We use the app's bundle
|
||||
/// identifier (`xyz.block.buzz.app` for release, `xyz.block.buzz.app.dev`
|
||||
/// for `just dev`) because it is stable across restarts — a relaunched dev
|
||||
/// instance still recognizes its own previously-spawned agents as reclaimable,
|
||||
/// while never matching another instance's (e.g. a dev build never reaps a DMG
|
||||
/// build's agents, and vice versa). This is what lets two Buzzs coexist on
|
||||
/// one machine without one's cleanup nuking the other's agents.
|
||||
pub(crate) fn current_instance_id(app: &AppHandle) -> String {
|
||||
app.config().identifier.clone()
|
||||
}
|
||||
|
||||
/// Build the full `BUZZ_MANAGED_AGENT=<instance-id>` env entry we match
|
||||
/// against when scanning processes. Kept here so the spawn stamp and the sweep
|
||||
/// matcher can never drift apart.
|
||||
pub(super) fn buzz_marker_entry(instance_id: &str) -> Vec<u8> {
|
||||
format!("BUZZ_MANAGED_AGENT={instance_id}").into_bytes()
|
||||
}
|
||||
|
||||
/// Check if a running process is one of *our* managed agents: it must carry
|
||||
/// `BUZZ_MANAGED_AGENT=<instance_id>` in its environment, where `instance_id`
|
||||
/// is this desktop instance's id. A process stamped with a *different* instance
|
||||
/// id belongs to another live Buzz app and must never be reaped here.
|
||||
#[cfg(target_os = "macos")]
|
||||
pub(crate) fn process_has_buzz_marker(pid: u32, instance_id: &str) -> bool {
|
||||
let marker = buzz_marker_entry(instance_id);
|
||||
let Some(buf) = sweep::procargs2_buffer(pid) else {
|
||||
return false;
|
||||
};
|
||||
|
||||
// Buffer layout: [i32 argc][exec_path\0][null padding][argv\0...][env\0...]
|
||||
if buf.len() < std::mem::size_of::<libc::c_int>() {
|
||||
return false;
|
||||
}
|
||||
let mut n_args: libc::c_int = 0;
|
||||
unsafe {
|
||||
std::ptr::copy_nonoverlapping(
|
||||
buf.as_ptr(),
|
||||
&mut n_args as *mut libc::c_int as *mut u8,
|
||||
std::mem::size_of::<libc::c_int>(),
|
||||
);
|
||||
}
|
||||
let mut pos = std::mem::size_of::<libc::c_int>();
|
||||
|
||||
// Skip exec path (scan to first null).
|
||||
while pos < buf.len() && buf[pos] != 0 {
|
||||
pos += 1;
|
||||
}
|
||||
// Skip null padding between exec path and argv[0].
|
||||
while pos < buf.len() && buf[pos] == 0 {
|
||||
pos += 1;
|
||||
}
|
||||
// Skip argc argument strings.
|
||||
let mut args_remaining = n_args;
|
||||
while args_remaining > 0 && pos < buf.len() {
|
||||
while pos < buf.len() && buf[pos] != 0 {
|
||||
pos += 1;
|
||||
}
|
||||
while pos < buf.len() && buf[pos] == 0 {
|
||||
pos += 1;
|
||||
}
|
||||
args_remaining -= 1;
|
||||
}
|
||||
// Remaining bytes are null-delimited environment strings.
|
||||
buf[pos..].split(|&b| b == 0).any(|entry| entry == marker)
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
pub(crate) fn process_has_buzz_marker(pid: u32, instance_id: &str) -> bool {
|
||||
let marker = buzz_marker_entry(instance_id);
|
||||
let Ok(data) = std::fs::read(format!("/proc/{pid}/environ")) else {
|
||||
return false;
|
||||
};
|
||||
data.split(|&b| b == 0).any(|entry| entry == marker)
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
pub(crate) fn process_has_buzz_marker(_pid: u32, _instance_id: &str) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn signal_process_group_or_leader(pid: u32, signal: i32, action: &str) -> Result<(), String> {
|
||||
let pgid = -(pid as i32);
|
||||
|
||||
if unsafe { libc::kill(pgid, signal) } == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let group_err = std::io::Error::last_os_error();
|
||||
if !process_is_running(pid) {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Some local agent trees can no longer be signalled as a process group
|
||||
// (for example if the leader changed groups, or macOS returns EPERM for one
|
||||
// descendant). Fall back to the leader PID so stop/delete can still recover.
|
||||
if matches!(
|
||||
group_err.raw_os_error(),
|
||||
Some(libc::EPERM) | Some(libc::ESRCH)
|
||||
) {
|
||||
if unsafe { libc::kill(pid as i32, signal) } == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let leader_err = std::io::Error::last_os_error();
|
||||
if leader_err.raw_os_error() == Some(libc::ESRCH) || !process_is_running(pid) {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
return Err(format!("failed to {action} process {pid}: {leader_err}"));
|
||||
}
|
||||
|
||||
Err(format!(
|
||||
"failed to {action} process group {pid}: {group_err}"
|
||||
))
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn terminate_process(pid: u32) -> Result<(), String> {
|
||||
// Try graceful shutdown first (SIGTERM to the group).
|
||||
signal_process_group_or_leader(pid, libc::SIGTERM, "terminate")?;
|
||||
|
||||
// Wait up to 1s for graceful exit.
|
||||
for _ in 0..10 {
|
||||
if !process_is_running(pid) {
|
||||
return Ok(());
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(100));
|
||||
}
|
||||
|
||||
// Escalate to SIGKILL on the entire group.
|
||||
signal_process_group_or_leader(pid, libc::SIGKILL, "kill")?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
pub(crate) fn terminate_process(pid: u32) -> Result<(), String> {
|
||||
// No job handle is available on this path (e.g. after an app restart, when
|
||||
// we only recovered the PID from the record), so fall back to taskkill on
|
||||
// the whole tree.
|
||||
super::super::process_lifecycle::taskkill_tree(pid)
|
||||
}
|
||||
|
||||
#[cfg(not(any(unix, windows)))]
|
||||
pub(crate) fn terminate_process(_pid: u32) -> Result<(), String> {
|
||||
Err("managed agent shutdown after app restart is not supported on this platform".to_string())
|
||||
}
|
||||
|
||||
/// Send SIGTERM to all given PIDs (as process groups), wait, then SIGKILL
|
||||
/// any survivors. Uses `-pid` to kill the entire process group — if an
|
||||
/// orphaned agent called `setsid()`, it IS the group leader, so this
|
||||
/// reaches its children too.
|
||||
#[cfg(unix)]
|
||||
fn sigterm_then_sigkill(pids: &[i32]) {
|
||||
// Send SIGTERM to each process group. Track whether any signal was
|
||||
// actually delivered so we can skip the sleep when everything is
|
||||
// already gone.
|
||||
let mut any_signalled = false;
|
||||
for &pid in pids {
|
||||
if unsafe { libc::kill(-pid, libc::SIGTERM) } == 0 {
|
||||
any_signalled = true;
|
||||
}
|
||||
}
|
||||
|
||||
if !any_signalled {
|
||||
return;
|
||||
}
|
||||
|
||||
std::thread::sleep(std::time::Duration::from_millis(200));
|
||||
|
||||
for &pid in pids {
|
||||
// Check if the group has any living members, not just the leader.
|
||||
// kill(-pid, 0) returns 0 if ANY member of the group is signalable.
|
||||
if unsafe { libc::kill(-pid, 0) } == 0 {
|
||||
unsafe {
|
||||
libc::kill(-pid, libc::SIGKILL);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve orphan candidate PIDs to their actual process group IDs, dedupe,
|
||||
/// and signal the groups. An orphaned grandchild (e.g. `goose` or `buzz-dev-mcp`)
|
||||
/// whose harness has exited retains the harness's PGID — signaling that PGID
|
||||
/// kills the entire orphaned subtree. Falls back to the candidate PID itself
|
||||
/// when PGID resolution fails (process may have exited between detection and
|
||||
/// kill).
|
||||
#[cfg(target_os = "macos")]
|
||||
pub(super) fn resolve_pgids_and_kill(candidate_pids: &[i32]) {
|
||||
let candidate_set: std::collections::HashSet<i32> = candidate_pids.iter().copied().collect();
|
||||
let mut pgids = std::collections::HashSet::new();
|
||||
for &pid in candidate_pids {
|
||||
let pgid = unsafe { libc::getpgid(pid) };
|
||||
if pgid > 0 {
|
||||
pgids.insert(pgid);
|
||||
} else {
|
||||
// Process may have exited; try signaling it directly as a group.
|
||||
pgids.insert(pid);
|
||||
}
|
||||
}
|
||||
// PID-recycling guard: if a resolved PGID is alive but isn't one of our
|
||||
// orphan candidates, the old harness PID was recycled by a new process
|
||||
// that called setsid() — skip it to avoid killing an unrelated group.
|
||||
let candidate_groups = pgids.len();
|
||||
pgids.retain(|&pgid| {
|
||||
if candidate_set.contains(&pgid) {
|
||||
return true;
|
||||
}
|
||||
let alive = unsafe { libc::kill(pgid, 0) } == 0;
|
||||
!alive
|
||||
});
|
||||
if pgids.is_empty() && candidate_groups > 0 {
|
||||
eprintln!(
|
||||
"buzz-desktop: orphan sweep: skipped all {candidate_groups} candidate group(s) (live foreign group leader or candidate already exited); nothing signalled"
|
||||
);
|
||||
}
|
||||
let unique: Vec<i32> = pgids.into_iter().collect();
|
||||
sigterm_then_sigkill(&unique);
|
||||
}
|
||||
|
||||
/// Resolve orphan candidate PIDs to their actual process group IDs, dedupe,
|
||||
/// and signal the groups. Linux variant reads PGID from /proc/<pid>/stat.
|
||||
#[cfg(all(unix, not(target_os = "macos")))]
|
||||
pub(super) fn resolve_pgids_and_kill(candidate_pids: &[i32]) {
|
||||
let candidate_set: std::collections::HashSet<i32> = candidate_pids.iter().copied().collect();
|
||||
let mut pgids = std::collections::HashSet::new();
|
||||
for &pid in candidate_pids {
|
||||
if let Some((_, pgid)) = sweep::proc_stat_ppid_pgid_linux(pid as u32) {
|
||||
pgids.insert(pgid as i32);
|
||||
} else {
|
||||
// Process may have exited; try signaling it directly as a group.
|
||||
pgids.insert(pid);
|
||||
}
|
||||
}
|
||||
// PID-recycling guard: if a resolved PGID is alive but isn't one of our
|
||||
// orphan candidates, the old harness PID was recycled by a new process
|
||||
// that called setsid() — skip it to avoid killing an unrelated group.
|
||||
let candidate_groups = pgids.len();
|
||||
pgids.retain(|&pgid| {
|
||||
if candidate_set.contains(&pgid) {
|
||||
return true;
|
||||
}
|
||||
let alive = unsafe { libc::kill(pgid, 0) } == 0;
|
||||
!alive
|
||||
});
|
||||
if pgids.is_empty() && candidate_groups > 0 {
|
||||
eprintln!(
|
||||
"buzz-desktop: orphan sweep: skipped all {candidate_groups} candidate group(s) (live foreign group leader or candidate already exited); nothing signalled"
|
||||
);
|
||||
}
|
||||
let unique: Vec<i32> = pgids.into_iter().collect();
|
||||
sigterm_then_sigkill(&unique);
|
||||
}
|
||||
|
||||
pub(crate) fn valid_agent_runtime_receipt(
|
||||
path: &std::path::Path,
|
||||
receipt: &super::super::ManagedAgentRuntimeReceipt,
|
||||
instance_id: &str,
|
||||
) -> bool {
|
||||
let Ok(canonical) =
|
||||
ManagedAgentRuntimeKey::new(receipt.key.pubkey.clone(), &receipt.key.relay_url)
|
||||
else {
|
||||
return false;
|
||||
};
|
||||
canonical == receipt.key
|
||||
&& path.file_name().and_then(|name| name.to_str())
|
||||
== Some(&format!("{}.json", receipt.key.runtime_id()))
|
||||
&& receipt.desktop_instance_id == instance_id
|
||||
&& process_is_running(receipt.pid)
|
||||
&& process_belongs_to_us(receipt.pid)
|
||||
&& process_has_buzz_marker(receipt.pid, &receipt.desktop_instance_id)
|
||||
}
|
||||
|
||||
pub(super) fn terminate_runtime_receipt_with(
|
||||
path: &std::path::Path,
|
||||
receipt: &super::super::ManagedAgentRuntimeReceipt,
|
||||
terminate: impl FnOnce(u32) -> Result<(), String>,
|
||||
mut is_running: impl FnMut(u32) -> bool,
|
||||
remove: impl FnOnce(&std::path::Path),
|
||||
) -> Result<(), String> {
|
||||
terminate(receipt.pid)?;
|
||||
for _ in 0..20 {
|
||||
if !is_running(receipt.pid) {
|
||||
remove(path);
|
||||
return Ok(());
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(100));
|
||||
}
|
||||
Err(format!(
|
||||
"prior runtime {} for pair {} on {} did not exit",
|
||||
receipt.pid, receipt.key.pubkey, receipt.key.relay_url
|
||||
))
|
||||
}
|
||||
|
||||
/// Replace a valid prior-session process before registering a new child for
|
||||
/// the same pair. The caller must hold the runtime transition lock so receipt
|
||||
/// inspection, termination, spawn, and registration cannot race shutdown or
|
||||
/// another start.
|
||||
pub(crate) fn terminate_untracked_pair_runtime(
|
||||
app: &AppHandle,
|
||||
key: &ManagedAgentRuntimeKey,
|
||||
) -> Result<(), String> {
|
||||
let instance_id = current_instance_id(app);
|
||||
let Some((path, receipt)) = super::super::read_all_agent_runtime_receipts(app)
|
||||
.into_iter()
|
||||
.find(|(path, receipt)| {
|
||||
receipt.key == *key && valid_agent_runtime_receipt(path, receipt, &instance_id)
|
||||
})
|
||||
else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
terminate_runtime_receipt_with(
|
||||
&path,
|
||||
&receipt,
|
||||
terminate_process,
|
||||
process_is_running,
|
||||
super::super::remove_agent_runtime_receipt_path,
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user