agentmux_launcher/

second_instance.rs

// Copyright 2025-2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

#[cfg(not(target_os = "windows"))]
use crate::logging::log;

/// Phase B.6 (post-fix) — forward an `open_new_window` request to
/// the already-running host and let this launcher exit 0.
///
/// The host writes `<data-dir>/ipc-port` after CEF init as
/// `port:token`. We open a TCP connection to 127.0.0.1:port, send a
/// minimal HTTP/1.1 POST to /ipc with the bearer token and a JSON
/// body, and bail. We deliberately do NOT pull in reqwest: the
/// launcher binary should stay tiny (~325 KB) and the protocol is
/// fixed, so a hand-rolled request is the right tool.
///
/// Failure classification (codex P2 PR #598):
/// - `Transient` — port file missing / unreadable / malformed.
///   The host is alive (pipe held) but mid-startup; caller exits
///   0 silently so the user isn't punished for double-clicking
///   quickly.
/// - `Fatal` — port file is readable, but the HTTP path failed
///   (connect refused, write failed, timeout). Either a hung
///   host or a non-running-instance source of
///   `ERROR_ACCESS_DENIED` (namespace conflict, security
///   descriptor failure). Caller surfaces the dialog so the user
///   sees a real problem rather than a silent no-op.
pub(crate) enum ForwardError {
    Transient(String),
    Fatal(String),
}

impl ForwardError {
    pub(crate) fn reason(&self) -> &str {
        match self {
            Self::Transient(r) | Self::Fatal(r) => r,
        }
    }
}

/// How long a relaunch keeps at it when the running instance's host didn't
/// take its `open_new_window`: long enough for an instance that is quitting
/// to let go of the single-instance socket/pipe (its launcher holds on until
/// srv has closed every agent, up to `SRV_EXIT_WAIT`, via
/// `upgrade::quiesce_srv`).
pub(crate) const QUITTING_INSTANCE_WAIT: std::time::Duration =
    std::time::Duration::from_secs(agentmux_common::process::SRV_EXIT_WAIT.as_secs() + 5);

/// What [`await_running_instance`] ended with.
pub(crate) enum Handoff<T> {
    /// The old instance quit; this launcher holds the socket/pipe now.
    Claimed(T),
    /// The host answered and took the request.
    Forwarded,
    /// Neither within [`QUITTING_INSTANCE_WAIT`]; the last forward error.
    GaveUp(ForwardError),
}

/// A relaunch whose forward failed with `first`: the running instance is
/// starting (no port file yet) or quitting (its host is gone; a clean exit
/// deletes the port file). Until [`QUITTING_INSTANCE_WAIT`] passes, take the
/// socket/pipe as soon as the old launcher lets go of it, or forward as soon
/// as the host answers. Shared by both supervisors.
pub(crate) fn await_running_instance<T>(
    mut try_claim: impl FnMut() -> Option<T>,
    mut try_forward: impl FnMut() -> Result<(), ForwardError>,
    first: ForwardError,
) -> Handoff<T> {
    let until = std::time::Instant::now() + QUITTING_INSTANCE_WAIT;
    let mut last = first;
    loop {
        if let Some(claimed) = try_claim() {
            return Handoff::Claimed(claimed);
        }
        if std::time::Instant::now() >= until {
            return Handoff::GaveUp(last);
        }
        std::thread::sleep(std::time::Duration::from_millis(250));
        match try_forward() {
            Ok(()) => return Handoff::Forwarded,
            Err(e) => last = e,
        }
    }
}

/// Forward an arbitrary host IPC command over the same authenticated
/// localhost channel `forward_open_new_window` uses.
///
/// Extracted (issue #2977 Workstream 1) so the tray can also send `quit_app`
/// without a second copy of the port-file read, bearer-token handshake, and
/// the read-the-response subtlety documented below — all of which are easy to
/// get subtly wrong and none of which are specific to opening a window.
/// `forward_open_new_window` is now a thin wrapper, so the long-standing
/// second-instance path keeps its exact behavior.
pub(crate) fn forward_host_cmd(
    data_dir: &std::path::Path,
    dir_hash: &str,
    cmd: &str,
) -> Result<(), ForwardError> {
    forward_host_cmd_impl(data_dir, dir_hash, cmd)
}

pub(crate) fn forward_open_new_window(
    data_dir: &std::path::Path,
    dir_hash: &str,
) -> Result<(), ForwardError> {
    forward_host_cmd_impl(data_dir, dir_hash, "open_new_window")
}

/// `forward_host_cmd` with an `args` object — the host's `/ipc` body is
/// `{"cmd", "args"}` (`crates/cef/src/ipc.rs` `IpcRequest`). Used by the
/// notification click path (`open_new_window {workspace_id}`,
/// `focus_window {label}`).
pub(crate) fn forward_host_cmd_with_args(
    data_dir: &std::path::Path,
    dir_hash: &str,
    cmd: &str,
    args: serde_json::Value,
) -> Result<(), ForwardError> {
    forward_host_body(data_dir, dir_hash, &serde_json::json!({ "cmd": cmd, "args": args }).to_string())
}

fn forward_host_cmd_impl(
    data_dir: &std::path::Path,
    dir_hash: &str,
    cmd: &str,
) -> Result<(), ForwardError> {
    // `cmd` is a fixed internal identifier chosen by call sites in this
    // crate, never user input, so a plain format is safe here — there is no
    // untrusted string to escape.
    forward_host_body(data_dir, dir_hash, &format!(r#"{{"cmd":"{}"}}"#, cmd))
}

fn forward_host_body(
    data_dir: &std::path::Path,
    dir_hash: &str,
    body: &str,
) -> Result<(), ForwardError> {
    // Read the version-scoped port file so we reach THIS version's host,
    // not a concurrent release's host that may have overwritten "ipc-port".
    let port_file_name = format!("ipc-port-{}", dir_hash);
    let port_file = data_dir.join(&port_file_name);
    let contents = std::fs::read_to_string(&port_file).map_err(|e| {
        ForwardError::Transient(format!("read {}: {}", port_file.display(), e))
    })?;
    let trimmed = contents.trim();
    let (port_str, token) = trimmed.split_once(':').ok_or_else(|| {
        ForwardError::Transient(format!(
            "malformed port file (expected port:token): {}",
            trimmed
        ))
    })?;
    let port: u16 = port_str
        .parse()
        .map_err(|e| ForwardError::Transient(format!("invalid port {:?}: {}", port_str, e)))?;

    // From here on the file was readable: any failure is a fatal
    // forward (the host got far enough to publish but isn't
    // serving the IPC port).
    let addr: std::net::SocketAddr = ([127, 0, 0, 1], port).into();
    let mut stream = std::net::TcpStream::connect_timeout(&addr, std::time::Duration::from_secs(2))
        .map_err(|e| ForwardError::Fatal(format!("connect 127.0.0.1:{}: {}", port, e)))?;
    stream
        .set_write_timeout(Some(std::time::Duration::from_secs(2)))
        .ok();

    let req = format!(
        "POST /ipc HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAuthorization: Bearer {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
        token,
        body.len(),
        body
    );
    use std::io::{Read, Write};
    stream
        .write_all(req.as_bytes())
        .map_err(|e| ForwardError::Fatal(format!("write request: {}", e)))?;
    // CRITICAL: read at least the status line. The host's axum
    // handler is async — if the launcher closes the TCP socket
    // before axum has finished parsing + dispatching to
    // `open_new_window`, the request can be dropped (smoke caught
    // exactly this on v0.33.481: the launcher logged "forwarded"
    // but no second window appeared because the process exited
    // before axum ran the handler). We don't care about the body
    // — `Connection: close` lets the server drop the socket once
    // the response is written, so a single short read is enough
    // to keep the connection alive past handler dispatch.
    stream
        .set_read_timeout(Some(std::time::Duration::from_secs(2)))
        .ok();
    let mut sink = [0u8; 64];
    let _ = stream.read(&mut sink);
    Ok(())
}

/// User-facing message for an actual second-instance collision (not a
/// bind-failure, not a stale socket — a real running launcher answered our
/// connect probe). Shared by both call sites in
/// `bind_socket_with_recovery` (fast-path and post-recovery-lock retry),
/// which previously duplicated this message and both got it wrong the same
/// way — see docs/specs/SPEC_DEV_INSTANCE_ISOLATION_DIAGNOSTICS_2026_09_19.md.
///
/// For a `dev-` channel, this used to say "Use `task dev:local` to launch a
/// second isolated session" — plausible-sounding, and wrong for the common
/// case. `task dev:local`'s version bump only changes the launcher's own
/// IPC socket key (`hash.rs`'s `data_dir_hash16`, keyed on
/// `CARGO_PKG_VERSION`/`AGENTMUX_BUILD_LABEL` at compile time) — it does
/// NOT change the CEF profile/data directory, which for dev mode is keyed
/// on `(branch, clone_id)` — NOT version — via
/// `crates/common/src/data_paths.rs`'s `resolve_channel_and_dir`
/// (`honor_env_channel=false` for dev launches). `clone_id`
/// (`runtime_mode.rs`'s `derive_clone_id`) is a hash of the CLONE'S OWN
/// CANONICAL WORKSPACE-ROOT PATH — git-worktree-aware by design, so a
/// second worktree of the SAME branch already gets a distinct `clone_id`
/// and therefore a distinct data dir, with no branch switch needed;
/// `resolve_channel_and_dir` folds both into the one `channel` string this
/// function receives (`format!("dev-{branch}-{clone_id}")` — confirmed by
/// reading that call site directly). Two dev instances collide on CEF's
/// own `SingletonLock` only when BOTH branch and clone/workspace path are
/// identical — reproduced live: a real rebuild under a bumped version, run
/// from the unchanged clone path on the unchanged branch, got a new
/// launcher socket and STILL hit `CEF early exit (process singleton or
/// similar)` on the unchanged data dir. The fix is a different branch OR a
/// different clone/worktree path (either one changes `channel`) — named
/// directly here instead of pointing at a command that cannot solve this
/// specific problem. (ReAgent P1 on this PR: an earlier revision of this
/// message claimed branch was the ONLY lever, which is what this revision
/// corrects — verified against `resolve_channel_and_dir`'s actual source,
/// not just re-asserted.)
#[cfg(not(target_os = "windows"))]
fn print_dev_instance_collision(channel: &str, data_dir: &std::path::Path, socket_path: &str) {
    eprintln!(
        "AgentMux dev instance already running for channel \"{channel}\"\n\
         (dev-<branch>-<clone_id>, where clone_id identifies THIS checkout's\n\
         own workspace-root path). Data-dir isolation is keyed on branch AND\n\
         clone/workspace path together, not build version — a `task dev:local`\n\
         version bump will NOT let a second instance run alongside this one\n\
         from this same checkout on this same branch (it only changes this\n\
         launcher's own IPC socket key, not the CEF profile directory below).\n\
         To run a second dev instance concurrently, either check out a\n\
         DIFFERENT branch, or run `task dev` from a DIFFERENT clone/worktree\n\
         path (even on this same branch — clone/worktree path alone is\n\
         enough to change the channel above).\n\
         Data dir: {}\n\
         Socket:   {socket_path}",
        data_dir.display()
    );
}

/// Best-effort `open_new_window` forward for the unix second-instance path.
/// Unlike the Windows path (which pops a dialog on a fatal forward), unix just
/// logs and lets the caller exit 0: the existing instance is alive (its socket
/// answered our connect probe), so a transient/fatal forward failure shouldn't
/// block — at worst the relaunch is a silent no-op instead of a new window.
/// SPEC_MACOS_LAUNCH_COHERENCE_2026_06_18.md.
#[cfg(target_os = "macos")]
pub(crate) fn forward_open_new_window_or_log(data_dir: &std::path::Path, dir_hash: &str) {
    match forward_open_new_window(data_dir, dir_hash) {
        Ok(()) => log("forwarded open_new_window to existing instance"),
        Err(ForwardError::Transient(reason)) => {
            log(&format!("open_new_window forward transient (host mid-startup?): {}", reason))
        }
        Err(ForwardError::Fatal(reason)) => {
            log(&format!("open_new_window forward failed: {}", reason))
        }
    }
}

/// What a launcher that lost the single-instance race does: open a window in
/// the running instance, unless it was started by a login entry, which only
/// asks that AgentMux be running (SPEC_START_WITH_OS_2026_09_25.md §3.6).
/// Not used by the macOS reopen handler, which is a user asking for a window.
/// Returns the forward's error, if any, for [`await_running_instance`].
#[cfg(not(target_os = "windows"))]
fn forward_for_second_instance(data_dir: &std::path::Path, dir_hash: &str) -> Option<ForwardError> {
    if !crate::autostart::second_instance_opens_window(&std::env::args().collect::<Vec<_>>()) {
        log("login start: AgentMux is already running — exiting without opening a window");
        return None;
    }
    match forward_open_new_window(data_dir, dir_hash) {
        Ok(()) => {
            log("forwarded open_new_window to existing instance");
            None
        }
        Err(e) => Some(e),
    }
}

/// The socket, once the launcher that held it has exited: unlink its file
/// and bind. `None` while it still answers. Called under the recovery lock.
#[cfg(not(target_os = "windows"))]
fn claim_after_quit(socket_path: &str) -> Option<tokio::net::UnixListener> {
    if std::os::unix::net::UnixStream::connect(socket_path).is_ok() {
        return None;
    }
    let _ = std::fs::remove_file(socket_path);
    crate::ipc::server::bind_first_unix_socket(socket_path).ok()
}

/// Bind the launcher's IPC socket with single-instance enforcement +
/// crash-safe stale-socket recovery, serialized across concurrent
/// launchers via `flock(2)`.
///
/// Returns a bound `UnixListener` on success. Calls `std::process::exit`
/// on:
///   * second-instance detection (exit code 0)
///   * a hard bind failure that isn't `EADDRINUSE` (exit code 2)
///   * unable to acquire the recovery lock (exit code 2)
///
/// Why the lockfile (codex P1 + reagent P1 on PR #1288): see the call-
/// site comment. Two-launcher concurrent stale-cleanup would otherwise
/// produce two live launchers for one data dir.
#[cfg(not(target_os = "windows"))]
pub(crate) fn bind_socket_with_recovery(
    socket_path: &str,
    data_dir: &std::path::Path,
    dir_hash: &str,
    channel: &str,
) -> tokio::net::UnixListener {
    use std::os::unix::io::AsRawFd as _;

    // Fast path: bind without contention.
    match crate::ipc::server::bind_first_unix_socket(socket_path) {
        Ok(l) => return l,
        Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => { /* slow path below */ }
        Err(e) => {
            log(&format!("FATAL: bind {} failed: {}", socket_path, e));
            eprintln!(
                "AgentMux failed to start: could not bind IPC socket.\n\nSocket: {}\nError: {}",
                socket_path, e
            );
            std::process::exit(2);
        }
    }

    // Slow path: contention. Acquire the recovery lock so only one
    // launcher at a time does the connect-probe + unlink + rebind.
    let lock_path = format!("{}.lock", socket_path);
    let lock_file = match std::fs::OpenOptions::new()
        .create(true)
        .read(true)
        .write(true)
        .truncate(false)
        .open(&lock_path)
    {
        Ok(f) => f,
        Err(e) => {
            log(&format!(
                "FATAL: could not open recovery lockfile {}: {}",
                lock_path, e
            ));
            eprintln!(
                "AgentMux failed to start: could not open IPC recovery lockfile.\n\nLockfile: {}\nError: {}",
                lock_path, e
            );
            std::process::exit(2);
        }
    };
    // Block until we have the lock — another launcher's recovery
    // window is bounded by a single bind + a single connect-probe;
    // we won't wait long. flock(2) is auto-released on close (when
    // the OS reaps the launcher process), so a SIGKILL'd holder
    // doesn't leak the lock.
    if unsafe { libc::flock(lock_file.as_raw_fd(), libc::LOCK_EX) } != 0 {
        let errno = std::io::Error::last_os_error();
        log(&format!(
            "FATAL: flock({}, LOCK_EX) failed: {}",
            lock_path, errno
        ));
        eprintln!(
            "AgentMux failed to start: could not acquire IPC recovery lock.\n\nLockfile: {}\nError: {}",
            lock_path, errno
        );
        std::process::exit(2);
    }

    // Retry the bind under the lock — another launcher may have
    // already cleaned up the stale file while we were waiting on
    // flock, leaving us free to bind directly.
    match crate::ipc::server::bind_first_unix_socket(socket_path) {
        Ok(l) => return l,
        Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => { /* probe below */ }
        Err(e) => {
            log(&format!(
                "FATAL: post-lock bind {} failed: {}",
                socket_path, e
            ));
            eprintln!(
                "AgentMux failed to start: post-lock IPC bind failed.\n\nSocket: {}\nError: {}",
                socket_path, e
            );
            std::process::exit(2);
        }
    }

    // Disambiguate: is the existing socket a real running launcher,
    // or a stale file?
    match std::os::unix::net::UnixStream::connect(socket_path) {
        Ok(_) => {
            // Real second-instance. Forward an `open_new_window` request to the
            // already-running launcher's host (Windows-parity: `run_windows`'s
            // `already_running` branch in supervisor/windows.rs), then exit
            // cleanly. SPEC_MACOS_LAUNCH_COHERENCE_2026_06_18.md.
            if channel.starts_with("dev-") {
                print_dev_instance_collision(channel, data_dir, socket_path);
            } else {
                eprintln!(
                    "AgentMux is already running for this data directory.\n\nSocket: {}",
                    socket_path
                );
            }
            if let Some(first) = forward_for_second_instance(data_dir, dir_hash) {
                log(&format!(
                    "[ipc] open_new_window not taken ({}) — waiting for the instance to start or finish quitting",
                    first.reason()
                ));
                match await_running_instance(
                    || claim_after_quit(socket_path),
                    || forward_open_new_window(data_dir, dir_hash),
                    first,
                ) {
                    Handoff::Claimed(listener) => {
                        log("[ipc] the previous instance has quit — starting");
                        return listener;
                    }
                    Handoff::Forwarded => log("forwarded open_new_window to existing instance"),
                    Handoff::GaveUp(e) => log(&format!("open_new_window forward failed: {}", e.reason())),
                }
            }
            log(&format!(
                "[ipc] second-instance detected — existing launcher owns {}",
                socket_path
            ));
            std::process::exit(0);
        }
        Err(connect_err)
            if connect_err.kind() == std::io::ErrorKind::ConnectionRefused
                || connect_err.raw_os_error() == Some(libc::ENOENT) =>
        {
            // Stale socket file from a crashed launcher. Unlink and
            // rebind. The lock serializes us against other launchers
            // ALSO doing recovery, but it does NOT block a fresh
            // launcher taking the fast-path bind() above — that
            // launcher is lock-free and can win the socket in the
            // microsecond window between our `remove_file` and our
            // `bind`. If that happens, AddrInUse means a real
            // launcher just claimed the socket and WE are now the
            // losing second instance, not a failed start.
            // (Reagent P2 on PR #1288.)
            log(&format!(
                "[ipc] stale socket file at {} — unlinking and rebinding (under recovery lock)",
                socket_path
            ));
            let _ = std::fs::remove_file(socket_path);
            match crate::ipc::server::bind_first_unix_socket(socket_path) {
                Ok(l) => l,
                Err(retry_e) if retry_e.kind() == std::io::ErrorKind::AddrInUse => {
                    if channel.starts_with("dev-") {
                        print_dev_instance_collision(channel, data_dir, socket_path);
                    } else {
                        eprintln!(
                            "AgentMux is already running for this data directory.\n\nSocket: {}",
                            socket_path
                        );
                    }
                    let _ = forward_for_second_instance(data_dir, dir_hash);
                    log(&format!(
                        "[ipc] post-recovery bind lost the race to a fresh launcher on {} — exiting as second instance",
                        socket_path
                    ));
                    std::process::exit(0);
                }
                Err(retry_e) => {
                    log(&format!(
                        "FATAL: bind retry after stale-socket unlink failed: {}",
                        retry_e
                    ));
                    eprintln!(
                        "AgentMux failed to start: IPC rebind after stale cleanup failed.\n\nSocket: {}\nError: {}",
                        socket_path, retry_e
                    );
                    std::process::exit(2);
                }
            }
        }
        Err(other) => {
            log(&format!(
                "[ipc] AddrInUse but connect probe failed in an unexpected way: {} — \
                 treating as second instance and exiting cleanly",
                other
            ));
            std::process::exit(0);
        }
    }
    // `lock_file` drops here; flock auto-released on close.
}

#[cfg(test)]
mod tests {
    use super::{await_running_instance, ForwardError, Handoff};

    /// A relaunch during a quit takes the socket/pipe as soon as the old
    /// instance lets go of it, not after the whole wait.
    #[test]
    fn a_relaunch_claims_as_soon_as_the_old_instance_lets_go() {
        let mut tries = 0;
        let started = std::time::Instant::now();
        let out = await_running_instance(
            || {
                tries += 1;
                (tries == 3).then_some("socket")
            },
            || Err(ForwardError::Transient("port file gone".into())),
            ForwardError::Transient("port file gone".into()),
        );
        assert!(matches!(out, Handoff::Claimed("socket")));
        assert!(started.elapsed() < std::time::Duration::from_secs(2));
    }

    /// A relaunch while the instance is still starting forwards once its
    /// host answers.
    #[test]
    fn a_relaunch_forwards_once_the_host_answers() {
        let mut forwards = 0;
        let out = await_running_instance(
            || None::<()>,
            || {
                forwards += 1;
                if forwards == 2 { Ok(()) } else { Err(ForwardError::Transient("not yet".into())) }
            },
            ForwardError::Transient("not yet".into()),
        );
        assert!(matches!(out, Handoff::Forwarded));
        assert_eq!(forwards, 2);
    }
}