diff --git a/Cargo.lock b/Cargo.lock index e69c6af61..6b998fff7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1553,6 +1553,7 @@ dependencies = [ "fabro-mcp", "fabro-model", "fabro-openai-oauth", + "fabro-proctitle", "fabro-retro", "fabro-sandbox", "fabro-store", @@ -1802,6 +1803,14 @@ dependencies = [ "tracing", ] +[[package]] +name = "fabro-proctitle" +version = "0.176.2" +dependencies = [ + "cc", + "libc", +] + [[package]] name = "fabro-retro" version = "0.176.2" diff --git a/lib/crates/fabro-cli/Cargo.toml b/lib/crates/fabro-cli/Cargo.toml index 0310bf3af..9d7d1c854 100644 --- a/lib/crates/fabro-cli/Cargo.toml +++ b/lib/crates/fabro-cli/Cargo.toml @@ -29,6 +29,7 @@ fabro-devcontainer = { path = "../fabro-devcontainer" } fabro-hooks = { path = "../fabro-hooks" } fabro-interview = { path = "../fabro-interview" } fabro-mcp = { path = "../fabro-mcp" } +fabro-proctitle = { path = "../fabro-proctitle" } fabro-retro = { path = "../fabro-retro" } fabro-sandbox = { path = "../fabro-sandbox", features = ["ssh", "daytona"] } fabro-git-storage = { path = "../fabro-git-storage" } diff --git a/lib/crates/fabro-cli/src/commands/run/detached.rs b/lib/crates/fabro-cli/src/commands/run/detached.rs index a9216bf99..32c2b7bda 100644 --- a/lib/crates/fabro-cli/src/commands/run/detached.rs +++ b/lib/crates/fabro-cli/src/commands/run/detached.rs @@ -7,21 +7,33 @@ use fabro_store::RuntimeState; use fabro_workflows::event::EventEmitter; use fabro_workflows::git::GitAuthor; use fabro_workflows::operations::{StartServices, resume as resume_run, start as start_run}; +use fabro_workflows::records::{RunRecord, RunRecordExt}; use crate::cli_config; use crate::shared; pub(crate) async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bool) -> Result<()> { + let _ = fabro_proctitle::init(); + + let _launcher_guard = scopeguard::guard(launcher_path.clone(), |path| { + super::launcher::remove_launcher_record(&path); + }); + let cli_settings = cli_config::load_cli_settings(None)?; + let on_node: Option> = + RunRecord::load(&run_dir).ok().map(|record| { + let short_id = super::short_run_id(&record.run_id).to_string(); + fabro_proctitle::set(&format!("fabro: {short_id}")); + Arc::new(move |node_id: &str| { + fabro_proctitle::set(&format!("fabro: {short_id} {node_id}")); + }) as Arc + }); + let github_app = shared::github::build_github_app_credentials(cli_settings.app_id()); let git_author = GitAuthor::from_options( cli_settings.git_author().and_then(|a| a.name.clone()), cli_settings.git_author().and_then(|a| a.email.clone()), ); - - let _launcher_guard = scopeguard::guard(launcher_path.clone(), |path| { - super::launcher::remove_launcher_record(&path); - }); let runtime_state = RuntimeState::new(&run_dir); let services = StartServices { @@ -34,6 +46,7 @@ pub(crate) async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bo )), git_author, github_app, + on_node, registry_override: None, }; diff --git a/lib/crates/fabro-cli/src/commands/run/launcher.rs b/lib/crates/fabro-cli/src/commands/run/launcher.rs index c7356d8b3..f2c4a5567 100644 --- a/lib/crates/fabro-cli/src/commands/run/launcher.rs +++ b/lib/crates/fabro-cli/src/commands/run/launcher.rs @@ -86,8 +86,15 @@ fn launcher_process_matches(record: &LauncherRecord) -> bool { }; let command = String::from_utf8_lossy(&output.stdout); + command_matches_launcher(record, &command) +} + +#[cfg(unix)] +fn command_matches_launcher(record: &LauncherRecord, command: &str) -> bool { let run_dir = record.run_dir.to_string_lossy(); - command.contains("__detached") && command.contains(run_dir.as_ref()) + let old_match = command.contains("__detached") && command.contains(run_dir.as_ref()); + let new_match = command.contains(&format!("fabro: {}", super::short_run_id(&record.run_id))); + old_match || new_match } #[cfg(not(unix))] @@ -144,4 +151,44 @@ mod tests { assert!(active_launcher_record_for_run(&run_dir).is_none()); assert!(!launcher_path.exists()); } + + #[cfg(unix)] + #[test] + fn command_matches_launcher_accepts_old_detached_format() { + let dir = tempfile::tempdir().unwrap(); + let record = LauncherRecord { + run_id: "01JGABCDEF12ZYXW".to_string(), + run_dir: dir.path().join("run"), + pid: 42, + resume: false, + log_path: dir.path().join("launcher.log"), + started_at: Utc::now(), + }; + + let command = format!( + "/usr/local/bin/fabro __detached --run-dir {} --launcher-path /tmp/launcher.json", + record.run_dir.display() + ); + + assert!(command_matches_launcher(&record, &command)); + } + + #[cfg(unix)] + #[test] + fn command_matches_launcher_accepts_new_title_format() { + let dir = tempfile::tempdir().unwrap(); + let record = LauncherRecord { + run_id: "01JGABCDEF12ZYXW".to_string(), + run_dir: dir.path().join("run"), + pid: 42, + resume: false, + log_path: dir.path().join("launcher.log"), + started_at: Utc::now(), + }; + + assert!(command_matches_launcher( + &record, + "fabro: 01JGABCDEF12 plan" + )); + } } diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index b2706ee7f..278b368c6 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -26,6 +26,10 @@ pub(crate) mod ssh; pub(crate) mod start; pub(crate) mod wait; +pub(super) fn short_run_id(id: &str) -> &str { + if id.len() > 12 { &id[..12] } else { id } +} + pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<()> { match cmd { RunCommands::Run(args) => command::execute(args, globals).await, diff --git a/lib/crates/fabro-proctitle/Cargo.toml b/lib/crates/fabro-proctitle/Cargo.toml new file mode 100644 index 000000000..a7138f277 --- /dev/null +++ b/lib/crates/fabro-proctitle/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "fabro-proctitle" +edition.workspace = true +version.workspace = true +license.workspace = true +description = "In-place process title updates for Fabro detached engines" + +[lints] +workspace = true + +[target.'cfg(unix)'.dependencies] +libc = "0.2" + +[build-dependencies] +cc = "1" diff --git a/lib/crates/fabro-proctitle/build.rs b/lib/crates/fabro-proctitle/build.rs new file mode 100644 index 000000000..2c97ef50c --- /dev/null +++ b/lib/crates/fabro-proctitle/build.rs @@ -0,0 +1,9 @@ +fn main() { + println!("cargo:rerun-if-changed=c/capture_argv.c"); + + if std::env::var("CARGO_CFG_TARGET_OS").as_deref() == Ok("linux") { + cc::Build::new() + .file("c/capture_argv.c") + .compile("capture_argv"); + } +} diff --git a/lib/crates/fabro-proctitle/c/capture_argv.c b/lib/crates/fabro-proctitle/c/capture_argv.c new file mode 100644 index 000000000..b719a78fc --- /dev/null +++ b/lib/crates/fabro-proctitle/c/capture_argv.c @@ -0,0 +1,16 @@ +#include + +static char *g_argv_start = 0; +static unsigned long g_argv_len = 0; + +__attribute__((constructor)) +static void capture_argv(int argc, char **argv, char **envp) { + (void)envp; + if (argc <= 0 || !argv || !argv[0]) return; + g_argv_start = argv[0]; + char *end = argv[argc - 1] + strlen(argv[argc - 1]) + 1; + g_argv_len = (unsigned long)(end - argv[0]); +} + +char *fabro_proctitle_argv_start(void) { return g_argv_start; } +unsigned long fabro_proctitle_argv_len(void) { return g_argv_len; } diff --git a/lib/crates/fabro-proctitle/src/lib.rs b/lib/crates/fabro-proctitle/src/lib.rs new file mode 100644 index 000000000..f244a9e1f --- /dev/null +++ b/lib/crates/fabro-proctitle/src/lib.rs @@ -0,0 +1,157 @@ +#![allow(unsafe_code)] + +use std::sync::{Mutex, OnceLock}; + +struct Buffer { + start: *mut u8, + len: usize, +} + +// Safety: after init() we treat the captured argv region as exclusively writable +// by this crate, and serialize writes with the mutex below. +unsafe impl Send for Buffer {} +// Safety: the raw pointer metadata is immutable after capture. +unsafe impl Sync for Buffer {} + +static STATE: OnceLock> = OnceLock::new(); + +#[cfg(target_os = "linux")] +unsafe extern "C" { + fn fabro_proctitle_argv_start() -> *mut libc::c_char; + fn fabro_proctitle_argv_len() -> libc::c_ulong; +} + +#[cfg(target_os = "macos")] +unsafe extern "C" { + fn _NSGetArgv() -> *mut *mut *mut libc::c_char; + fn _NSGetArgc() -> *mut libc::c_int; +} + +/// Capture the argv buffer. Call once early in the process. +#[must_use] +pub fn init() -> usize { + if let Some(state) = STATE.get() { + return state.lock().map_or(0, |buffer| buffer.len); + } + + let Some(buffer) = platform_init() else { + return 0; + }; + let len = buffer.len; + let _ = STATE.set(Mutex::new(buffer)); + + STATE + .get() + .and_then(|state| state.lock().ok().map(|buffer| buffer.len)) + .unwrap_or(len) +} + +/// Overwrite the process title shown by `ps`. +pub fn set(title: &str) { + let Some(state) = STATE.get() else { + return; + }; + let Ok(buffer) = state.lock() else { + return; + }; + if buffer.start.is_null() || buffer.len == 0 { + return; + } + + // SAFETY: init() captured a writable argv byte range for this process, and + // the mutex guard above provides exclusive access while we rewrite it. + let dst = unsafe { std::slice::from_raw_parts_mut(buffer.start, buffer.len) }; + write_title(dst, title.as_bytes()); +} + +fn write_title(dst: &mut [u8], title: &[u8]) { + if dst.is_empty() { + return; + } + + dst.fill(0); + let copy_len = title.len().min(dst.len().saturating_sub(1)); + dst[..copy_len].copy_from_slice(&title[..copy_len]); +} + +#[cfg(target_os = "linux")] +fn platform_init() -> Option { + // SAFETY: these symbols are provided by the Linux-only C object compiled in build.rs. + let start = unsafe { fabro_proctitle_argv_start() }; + // SAFETY: paired with the symbol above. + let len = unsafe { fabro_proctitle_argv_len() }; + let len = usize::try_from(len).ok()?; + if start.is_null() || len == 0 { + return None; + } + + Some(Buffer { + start: start.cast(), + len, + }) +} + +#[cfg(target_os = "macos")] +fn platform_init() -> Option { + // SAFETY: macOS exposes argc/argv through crt_externs for the current process. + let argc_ptr = unsafe { _NSGetArgc() }; + // SAFETY: paired with _NSGetArgc above. + let argv_ptr = unsafe { _NSGetArgv() }; + if argc_ptr.is_null() || argv_ptr.is_null() { + return None; + } + + // SAFETY: the pointers above are process globals owned by libc. + let argc = unsafe { *argc_ptr }; + // SAFETY: same as above. + let argv = unsafe { *argv_ptr }; + if argc <= 0 || argv.is_null() { + return None; + } + + // SAFETY: argc > 0 and argv is non-null, so argv[0] and argv[argc - 1] are valid to read. + let start = unsafe { *argv }; + // SAFETY: same bound check as above. + let last = unsafe { *argv.add(usize::try_from(argc).ok()?.saturating_sub(1)) }; + if start.is_null() || last.is_null() { + return None; + } + + // SAFETY: last points to a C string owned by the process image. + let last_len = unsafe { libc::strlen(last) }; + // SAFETY: advancing by the string length plus trailing NUL stays within the captured argv span. + let end = unsafe { last.add(last_len + 1) }; + let len = (end as usize).checked_sub(start as usize)?; + if len == 0 { + return None; + } + + Some(Buffer { + start: start.cast(), + len, + }) +} + +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +fn platform_init() -> Option { + None +} + +#[cfg(test)] +mod tests { + use super::write_title; + + #[test] + fn write_title_zero_fills_remainder() { + let mut buffer = [b'x'; 8]; + write_title(&mut buffer, b"fabro"); + assert_eq!(buffer, [b'f', b'a', b'b', b'r', b'o', 0, 0, 0]); + } + + #[test] + fn write_title_truncates_to_leave_nul() { + let mut buffer = [b'x'; 6]; + write_title(&mut buffer, b"toolong"); + assert_eq!(buffer, [b't', b'o', b'o', b'l', b'o', 0]); + } +} diff --git a/lib/crates/fabro-workflows/src/handler/manager_loop.rs b/lib/crates/fabro-workflows/src/handler/manager_loop.rs index 00e5692be..6b2474f2d 100644 --- a/lib/crates/fabro-workflows/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflows/src/handler/manager_loop.rs @@ -208,6 +208,7 @@ impl Handler for SubWorkflowHandler { emitter, sandbox, registry, + on_node: None, hook_runner, env, dry_run, diff --git a/lib/crates/fabro-workflows/src/lifecycle/mod.rs b/lib/crates/fabro-workflows/src/lifecycle/mod.rs index 5969a7852..8bf3a315c 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/mod.rs @@ -61,6 +61,7 @@ pub(crate) struct WorkflowLifecycle { disk: DiskLifecycle, git: GitLifecycle, artifact: ArtifactLifecycle, + on_node: Option>, /// Set in on_edge_selected when loop_restart approved; read+cleared by EventLifecycle::on_run_start restarted_from: Arc>>, /// Shared git checkpoint result (written by git, read by event) @@ -84,6 +85,7 @@ impl WorkflowLifecycle { run_dir: &PathBuf, run_options: &Arc, is_resume: bool, + on_node: Option>, ) -> Self { let runtime_state = RuntimeState::new(run_dir); let restarted_from: Arc>> = Arc::new(Mutex::new(None)); @@ -177,6 +179,7 @@ impl WorkflowLifecycle { disk, git, artifact, + on_node, restarted_from, checkpoint_git_result, is_initial_resume: AtomicBool::new(is_resume), @@ -261,6 +264,9 @@ impl RunLifecycle for WorkflowLifecycle { node: &WorkflowNode, state: &WfRunState, ) -> CoreResult { + if let Some(on_node) = &self.on_node { + on_node(node.id()); + } self.fidelity.before_node(node, state).await } diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index 4675d39bf..6efbf426a 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/lib/crates/fabro-workflows/src/operations/start.rs @@ -42,6 +42,7 @@ struct RunSession { sandbox: SandboxSpec, llm: LlmSpec, interviewer: Arc, + on_node: Option>, lifecycle: LifecycleOptions, hooks: fabro_hooks::HookConfig, sandbox_env: SandboxEnvSpec, @@ -66,6 +67,7 @@ pub struct StartServices { pub interviewer: Arc, pub git_author: GitAuthor, pub github_app: Option, + pub on_node: Option>, pub registry_override: Option>, } @@ -284,6 +286,7 @@ impl RunSession { dry_run: settings.dry_run_enabled(), }, interviewer, + on_node: services.on_node, lifecycle: LifecycleOptions { setup_commands: settings.setup_commands().to_vec(), setup_command_timeout_ms: settings.setup_timeout_ms().unwrap_or(300_000), @@ -372,6 +375,7 @@ impl RunSession { checkpoint: Option, ) -> Result { let preserve_sandbox = self.preserve_sandbox; + let on_node = self.on_node.clone(); let record = persisted.run_record(); let run_options = RunOptions { @@ -429,7 +433,8 @@ impl RunSession { checkpoint, seed_context: self.seed_context, }; - let initialized = pipeline::initialize(persisted, init_options).await?; + let mut initialized = pipeline::initialize(persisted, init_options).await?; + initialized.on_node = on_node; let sandbox_for_cleanup = Arc::clone(&initialized.sandbox); let cleanup_guard = scopeguard::guard((), move |()| { @@ -739,6 +744,7 @@ mod tests { interviewer: Arc::new(fabro_interview::AutoApproveInterviewer), git_author: crate::git::GitAuthor::default(), github_app: None, + on_node: None, registry_override: Some(registry), } } @@ -801,6 +807,35 @@ mod tests { assert!(run_dir.join("conclusion.json").exists()); } + #[tokio::test] + async fn start_invokes_on_node_callback_before_execution() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + let emitter = Arc::new(EventEmitter::new()); + let registry = Arc::new(test_registry()); + let visited = Arc::new(Mutex::new(Vec::new())); + + persisted_workflow(MINIMAL_DOT, &run_dir); + + let started = start( + &run_dir, + StartServices { + on_node: Some(Arc::new({ + let visited = Arc::clone(&visited); + move |node_id: &str| { + visited.lock().unwrap().push(node_id.to_string()); + } + })), + ..test_start_services(emitter, registry) + }, + ) + .await + .unwrap(); + + assert_eq!(started.finalized.conclusion.status, StageStatus::Success); + assert_eq!(*visited.lock().unwrap(), vec!["start".to_string()]); + } + #[tokio::test] async fn start_errors_when_checkpoint_exists() { let temp = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-workflows/src/pipeline/execute.rs b/lib/crates/fabro-workflows/src/pipeline/execute.rs index c84ce20ed..19c98539e 100644 --- a/lib/crates/fabro-workflows/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflows/src/pipeline/execute.rs @@ -43,6 +43,7 @@ pub async fn execute(init: Initialized) -> Executed { emitter, sandbox, registry, + on_node, hook_runner, env, dry_run, @@ -92,6 +93,7 @@ pub async fn execute(init: Initialized) -> Executed { &run_options.run_dir, &settings_arc, checkpoint.is_some(), + on_node, ); if let Some(ref cp) = checkpoint { diff --git a/lib/crates/fabro-workflows/src/pipeline/initialize.rs b/lib/crates/fabro-workflows/src/pipeline/initialize.rs index 6ea1c039a..79524387e 100644 --- a/lib/crates/fabro-workflows/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflows/src/pipeline/initialize.rs @@ -635,6 +635,7 @@ pub async fn initialize( emitter: options.emitter, sandbox, registry, + on_node: None, hook_runner, env, dry_run: options.dry_run, diff --git a/lib/crates/fabro-workflows/src/pipeline/types.rs b/lib/crates/fabro-workflows/src/pipeline/types.rs index 0099a7aa2..02a017beb 100644 --- a/lib/crates/fabro-workflows/src/pipeline/types.rs +++ b/lib/crates/fabro-workflows/src/pipeline/types.rs @@ -251,6 +251,7 @@ pub struct Initialized { pub emitter: Arc, pub sandbox: Arc, pub registry: Arc, + pub on_node: Option>, pub hook_runner: Option>, pub env: HashMap, pub dry_run: bool, diff --git a/lib/crates/fabro-workflows/src/test_support.rs b/lib/crates/fabro-workflows/src/test_support.rs index cb5e4ce31..eb17a8585 100644 --- a/lib/crates/fabro-workflows/src/test_support.rs +++ b/lib/crates/fabro-workflows/src/test_support.rs @@ -37,6 +37,7 @@ fn initialized( emitter, sandbox, registry: Arc::new(registry), + on_node: None, hook_runner: options.hook_runner, env: options.env, dry_run: run_options.dry_run_enabled(),