fix(tauri-shell): proactively kill stale openhuman RPC on startup (#1166)

app/src-tauri/src/core_process.rs

@@ -2,9 +2,22 @@
 //!
 //! The core's HTTP/JSON-RPC server runs as a tokio task inside the Tauri host
 //! so its lifetime is tied to the GUI process — there is no sidecar to leak
-//! on Cmd+Q. If something is already listening on the configured port (e.g.
-//! a manual `openhuman-core run` harness for debugging), `ensure_running`
-//! attaches to it instead of spawning a duplicate listener.
+//! on Cmd+Q.
+//!
+//! Stale-listener policy (see issue #1130): if something is already listening
+//! on the configured port when `ensure_running` runs, we probe `GET /` to see
+//! whether it is an OpenHuman core. If it is, we treat it as a stale process
+//! left behind by a previous build/dev session and proactively terminate it
+//! (graceful signal, then a force-kill that *revalidates* the pid is still
+//! the same listener — guards against PID reuse if the original exits inside
+//! the grace window) before spawning a fresh embedded server — otherwise the
+//! new UI would silently bind to an older RPC implementation. If the listener
+//! is something else (or unreachable), we refuse to attach and surface the
+//! conflict so it can be diagnosed instead of producing 401s and version
+//! drift downstream.
+//! Set `OPENHUMAN_CORE_REUSE_EXISTING=1` to opt back into the legacy
+//! attach-to-whatever-is-listening behavior (e.g. a manual `openhuman-core
+//! run` harness for debugging).
 
 use std::sync::Arc;
 use std::sync::LazyLock;
@@ -82,27 +95,37 @@ impl CoreProcessHandle {
     }
 
     async fn is_rpc_port_open(&self) -> bool {
-        matches!(
-            timeout(
-                Duration::from_millis(150),
-                TcpStream::connect(("127.0.0.1", self.port)),
-            )
-            .await,
-            Ok(Ok(_))
-        )
+        is_port_open(self.port).await
     }
 
     pub async fn ensure_running(&self) -> Result<(), String> {
         if self.is_rpc_port_open().await {
-            log::info!(
-                "[core] found existing core rpc endpoint at {}",
-                self.rpc_url()
-            );
-            log::warn!(
-                "[core] reusing port {} — another `openhuman-core` instance is already listening; this Tauri host will not spawn an embedded server. Authenticated Tauri-side calls will 401 unless the listener was started with this process's OPENHUMAN_CORE_TOKEN.",
-                self.port
-            );
-            return Ok(());
+            if reuse_existing_listener_enabled() {
+                log::warn!(
+                    "[core] OPENHUMAN_CORE_REUSE_EXISTING=1 — attaching to whatever is listening on port {} without identification (legacy behavior)",
+                    self.port
+                );
+                return Ok(());
+            }
+
+            match identify_listener(self.port).await {
+                ListenerKind::OpenHuman => {
+                    log::warn!(
+                        "[core] found stale OpenHuman listener on port {} — taking over (issue #1130)",
+                        self.port
+                    );
+                    self.takeover_stale_listener().await?;
+                    // Fall through to spawn-and-wait below.
+                }
+                ListenerKind::Unknown { reason } => {
+                    let msg = format!(
+                        "Core RPC port {} is in use by something that is not an OpenHuman core ({reason}). Refusing to attach (set OPENHUMAN_CORE_REUSE_EXISTING=1 to override) — quit the other process or set OPENHUMAN_CORE_PORT to a different port and relaunch.",
+                        self.port
+                    );
+                    log::error!("[core] {msg}");
+                    return Err(msg);
+                }
+            }
         }
 
         {
@@ -160,6 +183,92 @@ impl CoreProcessHandle {
         Err("core process did not become ready".to_string())
     }
 
+    /// Identify the OS pid currently bound to our port and terminate it,
+    /// then wait for the port to free. Used when the listener has been
+    /// fingerprinted as an OpenHuman core (via `GET /`) so killing it is safe.
+    async fn takeover_stale_listener(&self) -> Result<(), String> {
+        let pid = match find_pid_on_port(self.port) {
+            Some(pid) => pid,
+            None => {
+                return Err(format!(
+                    "could not determine pid bound to port {} — refusing to take over",
+                    self.port
+                ));
+            }
+        };
+        let self_pid = std::process::id();
+        if pid == self_pid {
+            // Defensive — `ensure_running` checks the port before spawning,
+            // so this branch should be unreachable. If it ever hits, killing
+            // ourselves would be catastrophic.
+            return Err(format!(
+                "stale-listener pid {pid} matches the Tauri host pid; refusing to self-terminate"
+            ));
+        }
+        log::warn!(
+            "[core] terminating stale OpenHuman process pid={pid} on port {} (issue #1130)",
+            self.port
+        );
+        if let Err(e) = kill_pid_term(pid) {
+            return Err(format!("failed to signal stale openhuman pid {pid}: {e}"));
+        }
+
+        // Wait for the graceful exit, then revalidate ownership before any
+        // force-kill — between the SIGTERM and a delayed SIGKILL the original
+        // pid could have exited and been reused by an unrelated process. If
+        // the port is now bound to a different pid (or to nothing), we do
+        // NOT escalate to a force-kill against the originally-resolved pid.
+        // (CodeRabbit feedback on #1166.)
+        const GRACE_MS: u64 = 750;
+        tokio::time::sleep(Duration::from_millis(GRACE_MS)).await;
+
+        if is_port_open(self.port).await {
+            match find_pid_on_port(self.port) {
+                Some(current) if current == pid => {
+                    log::warn!(
+                        "[core] pid {pid} still bound to port {} after SIGTERM — escalating to SIGKILL",
+                        self.port
+                    );
+                    if let Err(e) = kill_pid_force(pid) {
+                        return Err(format!(
+                            "failed to force-kill stale openhuman pid {pid}: {e}"
+                        ));
+                    }
+                }
+                Some(current) => {
+                    return Err(format!(
+                        "port {} rebounded to pid {current} after terminating pid {pid}; refusing to kill a different process",
+                        self.port
+                    ));
+                }
+                None => {
+                    // Port still showed open in `is_port_open` but pid lookup
+                    // returned nothing — likely a transient race with the
+                    // listener tearing down. Fall through to the poll loop.
+                }
+            }
+        }
+
+        const POLL_MS: u64 = 100;
+        const MAX_WAIT_MS: u64 = 5_000;
+        let mut waited_ms: u64 = GRACE_MS;
+        while is_port_open(self.port).await {
+            if waited_ms >= MAX_WAIT_MS {
+                return Err(format!(
+                    "signaled pid {pid} but port {} remained bound after {MAX_WAIT_MS}ms",
+                    self.port
+                ));
+            }
+            tokio::time::sleep(Duration::from_millis(POLL_MS)).await;
+            waited_ms += POLL_MS;
+        }
+        log::info!(
+            "[core] stale listener cleared (pid={pid}, port={}) after {waited_ms}ms",
+            self.port
+        );
+        Ok(())
+    }
+
     /// Restart the embedded core to pick up updated macOS permission grants.
     ///
     /// macOS caches permission state per-process; restarting forces a fresh
@@ -244,6 +353,215 @@ pub fn default_core_port() -> u16 {
         .unwrap_or(7788)
 }
 
+/// Whether `OPENHUMAN_CORE_REUSE_EXISTING` is set to a truthy value. Opts
+/// back into the pre-#1130 behavior of attaching to whatever is listening
+/// on the port without identification — useful for manual harnesses.
+fn reuse_existing_listener_enabled() -> bool {
+    std::env::var("OPENHUMAN_CORE_REUSE_EXISTING")
+        .map(|v| matches!(v.trim(), "1" | "true" | "TRUE" | "yes" | "YES"))
+        .unwrap_or(false)
+}
+
+async fn is_port_open(port: u16) -> bool {
+    matches!(
+        timeout(
+            Duration::from_millis(150),
+            TcpStream::connect(("127.0.0.1", port)),
+        )
+        .await,
+        Ok(Ok(_))
+    )
+}
+
+/// What is currently listening on the core RPC port.
+#[derive(Debug)]
+enum ListenerKind {
+    /// `GET /` returned a JSON body with `"name": "openhuman"` — i.e. a
+    /// stale OpenHuman core process from a previous build/session.
+    OpenHuman,
+    /// Either the listener didn't speak HTTP, didn't respond, or returned
+    /// a body that doesn't identify as openhuman.
+    Unknown { reason: String },
+}
+
+/// Probe `GET http://127.0.0.1:<port>/` to fingerprint the listener.
+/// Unauthenticated — the core's root handler does not require a token.
+async fn identify_listener(port: u16) -> ListenerKind {
+    let url = format!("http://127.0.0.1:{port}/");
+    let client = match reqwest::Client::builder()
+        .timeout(Duration::from_millis(750))
+        .build()
+    {
+        Ok(c) => c,
+        Err(e) => {
+            return ListenerKind::Unknown {
+                reason: format!("reqwest client build failed: {e}"),
+            };
+        }
+    };
+    let resp = match client.get(&url).send().await {
+        Ok(r) => r,
+        Err(e) => {
+            return ListenerKind::Unknown {
+                reason: format!("probe GET / failed: {e}"),
+            };
+        }
+    };
+    if !resp.status().is_success() {
+        return ListenerKind::Unknown {
+            reason: format!("probe GET / returned status {}", resp.status()),
+        };
+    }
+    let body = match resp.text().await {
+        Ok(b) => b,
+        Err(e) => {
+            return ListenerKind::Unknown {
+                reason: format!("probe GET / body read failed: {e}"),
+            };
+        }
+    };
+    if is_openhuman_root_body(&body) {
+        log::info!("[core] listener on port {port} identified as openhuman core");
+        ListenerKind::OpenHuman
+    } else {
+        let preview: String = body.chars().take(80).collect();
+        ListenerKind::Unknown {
+            reason: format!("probe GET / body did not identify as openhuman ({preview:?})"),
+        }
+    }
+}
+
+/// Pure parse of the root-handler JSON. Public-by-test so the fingerprinting
+/// logic stays unit-testable without standing up an HTTP server.
+fn is_openhuman_root_body(body: &str) -> bool {
+    let value: serde_json::Value = match serde_json::from_str(body) {
+        Ok(v) => v,
+        Err(_) => return false,
+    };
+    value
+        .get("name")
+        .and_then(|v| v.as_str())
+        .map(|s| s == "openhuman")
+        .unwrap_or(false)
+}
+
+#[cfg(unix)]
+fn find_pid_on_port(port: u16) -> Option<u32> {
+    let output = std::process::Command::new("lsof")
+        .args(["-nP", "-iTCP", &format!("-i:{port}"), "-sTCP:LISTEN", "-t"])
+        .output()
+        .ok()?;
+    if !output.status.success() {
+        return None;
+    }
+    parse_lsof_pid(&String::from_utf8_lossy(&output.stdout))
+}
+
+#[cfg(windows)]
+fn find_pid_on_port(port: u16) -> Option<u32> {
+    let output = std::process::Command::new("netstat")
+        .args(["-ano", "-p", "TCP"])
+        .output()
+        .ok()?;
+    if !output.status.success() {
+        return None;
+    }
+    parse_netstat_pid(&String::from_utf8_lossy(&output.stdout), port)
+}
+
+/// Pure parse of `lsof -t` output (one pid per line; first wins).
+fn parse_lsof_pid(stdout: &str) -> Option<u32> {
+    stdout
+        .lines()
+        .map(str::trim)
+        .find(|l| !l.is_empty())
+        .and_then(|l| l.parse::<u32>().ok())
+}
+
+/// Pure parse of `netstat -ano` output for a LISTENING entry on `port`.
+#[allow(dead_code)] // exercised only on windows builds
+fn parse_netstat_pid(stdout: &str, port: u16) -> Option<u32> {
+    let needle = format!(":{port}");
+    for line in stdout.lines() {
+        let trimmed = line.trim();
+        if !trimmed.contains("LISTENING") {
+            continue;
+        }
+        let parts: Vec<&str> = trimmed.split_whitespace().collect();
+        // Expected: ["TCP", "127.0.0.1:7788", "0.0.0.0:0", "LISTENING", "1234"]
+        if parts.len() >= 5 && parts[1].ends_with(&needle) {
+            if let Ok(pid) = parts[parts.len() - 1].parse::<u32>() {
+                return Some(pid);
+            }
+        }
+    }
+    None
+}
+
+/// Send the graceful-shutdown signal to `pid`. Returns `Ok` if the process
+/// exited cleanly, was already gone, or accepted the signal. Callers must
+/// re-check ownership of the resource (e.g. that the same pid is still bound
+/// to the port) before escalating to `kill_pid_force` — see the PID-reuse
+/// hazard discussed on #1166.
+#[cfg(unix)]
+fn kill_pid_term(pid: u32) -> Result<(), String> {
+    use nix::sys::signal::{kill, Signal};
+    use nix::unistd::Pid;
+    let target = Pid::from_raw(pid as i32);
+    if let Err(e) = kill(target, Signal::SIGTERM) {
+        // ESRCH means already gone — treat as success.
+        if e != nix::errno::Errno::ESRCH {
+            return Err(format!("SIGTERM pid {pid}: {e}"));
+        }
+    }
+    Ok(())
+}
+
+/// Force-kill `pid` after `kill_pid_term` failed to free the resource. Caller
+/// is responsible for revalidating that `pid` still owns the resource we're
+/// trying to free — see `takeover_stale_listener` for the revalidation step.
+#[cfg(unix)]
+fn kill_pid_force(pid: u32) -> Result<(), String> {
+    use nix::sys::signal::{kill, Signal};
+    use nix::unistd::Pid;
+    let target = Pid::from_raw(pid as i32);
+    match kill(Pid::from_raw(pid as i32), Signal::SIGKILL) {
+        Ok(()) => Ok(()),
+        // ESRCH means the process exited between our re-validation and the
+        // SIGKILL — the resource is freeing on its own, treat as success.
+        Err(e) if e == nix::errno::Errno::ESRCH => {
+            let _ = target;
+            Ok(())
+        }
+        Err(e) => Err(format!("SIGKILL pid {pid}: {e}")),
+    }
+}
+
+/// Windows has no graceful equivalent for a windowless RPC server — `taskkill`
+/// without `/F` only delivers `WM_CLOSE` to GUI apps. Send the WM_CLOSE first
+/// (best-effort) so console subprocesses can run shutdown handlers; the
+/// follow-up `kill_pid_force` does the actual termination.
+#[cfg(windows)]
+fn kill_pid_term(pid: u32) -> Result<(), String> {
+    // Best-effort — ignore non-zero exit (e.g. process is windowless).
+    let _ = std::process::Command::new("taskkill")
+        .args(["/PID", &pid.to_string()])
+        .status();
+    Ok(())
+}
+
+#[cfg(windows)]
+fn kill_pid_force(pid: u32) -> Result<(), String> {
+    let status = std::process::Command::new("taskkill")
+        .args(["/F", "/T", "/PID", &pid.to_string()])
+        .status()
+        .map_err(|e| format!("taskkill spawn: {e}"))?;
+    if !status.success() {
+        return Err(format!("taskkill exited with {status}"));
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 #[path = "core_process_tests.rs"]
 mod tests;

app/src-tauri/src/core_process_tests.rs

@@ -1,4 +1,7 @@
-use super::{current_rpc_token, default_core_port, generate_rpc_token, CoreProcessHandle};
+use super::{
+    current_rpc_token, default_core_port, generate_rpc_token, is_openhuman_root_body,
+    parse_lsof_pid, parse_netstat_pid, CoreProcessHandle,
+};
 use std::sync::{Mutex, MutexGuard, OnceLock};
 
 fn env_lock() -> MutexGuard<'static, ()> {
@@ -54,8 +57,36 @@ fn core_process_handle_new_creates_instance() {
     assert_eq!(handle.rpc_url(), "http://127.0.0.1:9999/rpc");
 }
 
+/// Issue #1130: a non-OpenHuman listener on the RPC port must NOT be
+/// silently attached to. The test binds a bare `TcpListener` (which never
+/// answers HTTP) so the identification probe sees an unknown listener and
+/// `ensure_running` must surface the conflict instead of returning Ok.
 #[test]
-fn ensure_running_returns_ok_when_rpc_port_already_open() {
+fn ensure_running_refuses_unknown_listener_on_port() {
+    let _env_lock = env_lock();
+    let _unset = EnvGuard::unset("OPENHUMAN_CORE_REUSE_EXISTING");
+    let rt = tokio::runtime::Runtime::new().expect("runtime");
+    let result = rt.block_on(async {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind test listener");
+        let port = listener.local_addr().expect("local addr").port();
+        let handle = CoreProcessHandle::new(port);
+        handle.ensure_running().await
+    });
+    let err = result.expect_err("ensure_running must refuse an unidentified listener");
+    assert!(
+        err.contains("not an OpenHuman core") || err.contains("port"),
+        "error should explain the conflict, got: {err}"
+    );
+}
+
+/// Escape hatch: setting `OPENHUMAN_CORE_REUSE_EXISTING=1` opts back into
+/// the legacy attach-to-anything behavior for manual harnesses.
+#[test]
+fn ensure_running_reuses_unknown_listener_when_override_set() {
+    let _env_lock = env_lock();
+    let _override = EnvGuard::set("OPENHUMAN_CORE_REUSE_EXISTING", "1");
     let rt = tokio::runtime::Runtime::new().expect("runtime");
     let result = rt.block_on(async {
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
@@ -67,10 +98,60 @@ fn ensure_running_returns_ok_when_rpc_port_already_open() {
     });
     assert!(
         result.is_ok(),
-        "ensure_running should fast-path: {result:?}"
+        "override should restore legacy fast-path: {result:?}"
     );
 }
 
+// ---------------------------------------------------------------------------
+// Listener fingerprinting (issue #1130)
+// ---------------------------------------------------------------------------
+
+#[test]
+fn is_openhuman_root_body_matches_canonical_root_response() {
+    // Mirrors the JSON shape produced by `core/jsonrpc.rs::root_handler`.
+    let body = r#"{
+        "name": "openhuman",
+        "ok": true,
+        "endpoints": {"health": "/health", "rpc": "/rpc"}
+    }"#;
+    assert!(is_openhuman_root_body(body));
+}
+
+#[test]
+fn is_openhuman_root_body_rejects_other_services() {
+    assert!(!is_openhuman_root_body(r#"{"name": "something-else"}"#));
+    assert!(!is_openhuman_root_body(r#"{"ok": true}"#));
+    assert!(!is_openhuman_root_body("not json at all"));
+    assert!(!is_openhuman_root_body(""));
+    // Wrong type for `name`.
+    assert!(!is_openhuman_root_body(r#"{"name": 42}"#));
+}
+
+#[test]
+fn parse_lsof_pid_picks_first_pid() {
+    assert_eq!(parse_lsof_pid("12345\n"), Some(12345));
+    // Multiple pids — pick the first non-empty line. lsof can emit several
+    // when multiple sockets share the port (IPv4/IPv6).
+    assert_eq!(parse_lsof_pid("\n  9876  \n12345\n"), Some(9876));
+    assert_eq!(parse_lsof_pid(""), None);
+    assert_eq!(parse_lsof_pid("not-a-pid\n"), None);
+}
+
+#[test]
+fn parse_netstat_pid_finds_listening_entry() {
+    // Sample shape from `netstat -ano -p TCP` on Windows.
+    let stdout = "\
+Active Connections
+
+  Proto  Local Address          Foreign Address        State           PID
+  TCP    0.0.0.0:135            0.0.0.0:0              LISTENING       1024
+  TCP    127.0.0.1:7788         0.0.0.0:0              LISTENING       4242
+  TCP    127.0.0.1:50000        127.0.0.1:7788         ESTABLISHED     5555
+";
+    assert_eq!(parse_netstat_pid(stdout, 7788), Some(4242));
+    assert_eq!(parse_netstat_pid(stdout, 9999), None);
+}
+
 // ---------------------------------------------------------------------------
 // Token generation tests
 // ---------------------------------------------------------------------------

docs/BUILDING.md

@@ -182,3 +182,24 @@ If the lock is left behind by a crashed process (PID no longer alive), the prefl
 **Known limitation**
 
 Dev and release builds still share `com.openhuman.app` as the cache identifier. Isolating dev to a separate `com.openhuman.app.dev` cache requires changes to the vendored `tauri-runtime-cef` (cache path is built inside the runtime from the bundle identifier, not exposed to the openhuman shell). Tracked as a follow-up to #864.
+
+### Stale `openhuman` RPC process on the core port
+
+**Symptom**
+
+A previous Tauri build or `openhuman-core run` harness left a process listening on `OPENHUMAN_CORE_PORT` (default `7788`). Until issue #1130 the new Tauri build would silently attach to that listener — leading to version drift and 401s when the new build's `OPENHUMAN_CORE_TOKEN` didn't match.
+
+**Current behavior (issue #1130)**
+
+`core_process::ensure_running` now probes the port at startup:
+
+- If `GET /` identifies the listener as an OpenHuman core (JSON body with `"name": "openhuman"`), it is treated as a stale process from a previous run and proactively terminated (`SIGTERM`, then `SIGKILL` after 750ms on Unix; `taskkill /F /T /PID` on Windows). The Tauri host then spawns its own fresh embedded core.
+- If the listener is something else (or doesn't speak HTTP), startup fails loudly with the conflict surfaced in the log instead of silently attaching.
+- Set `OPENHUMAN_CORE_REUSE_EXISTING=1` to opt back into the legacy attach-to-anything behavior — useful when running `openhuman-core run` as a manual debugging harness.
+
+**Manual cleanup (still works)**
+
+```bash
+pkill -f "OpenHuman.app/Contents"
+pkill -f "openhuman-core"
+```