Add detached engine process titles

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"

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" }

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<Arc<dyn Fn(&str) + Send + Sync>> =
+        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<dyn Fn(&str) + Send + Sync>
+        });
+
     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,
     };
 

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"
+        ));
+    }
 }

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,

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"

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");
+    }
+}

lib/crates/fabro-proctitle/c/capture_argv.c

@@ -0,0 +1,16 @@
+#include <string.h>
+
+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; }

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<Mutex<Buffer>> = 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<Buffer> {
+    // 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<Buffer> {
+    // 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<Buffer> {
+    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]);
+    }
+}

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,

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<Arc<dyn Fn(&str) + Send + Sync>>,
     /// Set in on_edge_selected when loop_restart approved; read+cleared by EventLifecycle::on_run_start
     restarted_from: Arc<Mutex<Option<(String, String)>>>,
     /// Shared git checkpoint result (written by git, read by event)
@@ -84,6 +85,7 @@ impl WorkflowLifecycle {
         run_dir: &PathBuf,
         run_options: &Arc<RunOptions>,
         is_resume: bool,
+        on_node: Option<Arc<dyn Fn(&str) + Send + Sync>>,
     ) -> Self {
         let runtime_state = RuntimeState::new(run_dir);
         let restarted_from: Arc<Mutex<Option<(String, String)>>> = 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<WorkflowGraph> for WorkflowLifecycle {
         node: &WorkflowNode,
         state: &WfRunState,
     ) -> CoreResult<WfNodeDecision> {
+        if let Some(on_node) = &self.on_node {
+            on_node(node.id());
+        }
         self.fidelity.before_node(node, state).await
     }
 

lib/crates/fabro-workflows/src/operations/start.rs

@@ -42,6 +42,7 @@ struct RunSession {
     sandbox: SandboxSpec,
     llm: LlmSpec,
     interviewer: Arc<dyn Interviewer>,
+    on_node: Option<Arc<dyn Fn(&str) + Send + Sync>>,
     lifecycle: LifecycleOptions,
     hooks: fabro_hooks::HookConfig,
     sandbox_env: SandboxEnvSpec,
@@ -66,6 +67,7 @@ pub struct StartServices {
     pub interviewer: Arc<dyn Interviewer>,
     pub git_author: GitAuthor,
     pub github_app: Option<fabro_github::GitHubAppCredentials>,
+    pub on_node: Option<Arc<dyn Fn(&str) + Send + Sync>>,
     pub registry_override: Option<Arc<HandlerRegistry>>,
 }
 
@@ -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<Checkpoint>,
     ) -> Result<Started, FabroError> {
         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();

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 {

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,

lib/crates/fabro-workflows/src/pipeline/types.rs

@@ -251,6 +251,7 @@ pub struct Initialized {
     pub emitter: Arc<EventEmitter>,
     pub sandbox: Arc<dyn Sandbox>,
     pub registry: Arc<HandlerRegistry>,
+    pub on_node: Option<Arc<dyn Fn(&str) + Send + Sync>>,
     pub hook_runner: Option<Arc<HookRunner>>,
     pub env: HashMap<String, String>,
     pub dry_run: bool,

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(),