dbg-cli 0.6.0

use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::os::unix::fs::OpenOptionsExt;
use std::os::unix::net::UnixListener;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicI32, AtomicUsize, Ordering};
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant};

use anyhow::{Context, Result, bail};
use dbg_cli::session_db::{self, CreateOptions, SessionDb, SessionKind, TargetClass};
use rusqlite::params;

use crate::backend::Backend;
use crate::commands::{
    self, Dispatched, crosstrack, debug as debug_cmds, lifecycle as lifecycle_cmds,
};
use crate::profile::ProfileData;
use crate::pty::{DebuggerIo, DebuggerProcess};
use dbg_cli::session_db::LiveDebugger;

const CMD_TIMEOUT: Duration = Duration::from_secs(60);
const MAX_IPC_CONNECTIONS: usize = 32;
const MAX_IPC_REQUEST_BYTES: u64 = 64 * 1024;
const IPC_IDLE_TIMEOUT: Duration = Duration::from_secs(5);

fn should_wake_accept_after_release(previous: usize) -> bool {
    previous == MAX_IPC_CONNECTIONS
}

fn cleanup_and_exit(keep_session_tmp: bool) -> ! {
    cleanup_runtime_state(keep_session_tmp);
    std::process::exit(0);
}

fn cleanup_runtime_state(keep_session_tmp: bool) {
    let me = session_slug();
    let _ = std::fs::remove_file(socket_path());
    let _ = std::fs::remove_file(pid_path());
    // Keep the reservation inode. Its flock is released when this process
    // exits, so a later starter can safely reuse the same inode. Unlinking
    // it here would let a new starter create a replacement inode while this
    // process was still shutting down.
    // Clear the "latest" pointer only if it still names us — another
    // daemon may have become the newest since we started, and we
    // mustn't blow away its pointer.
    clear_latest_pointer_if_owner(&me);
    cleanup_session_tmp(keep_session_tmp);
}

/// Remove state published before backend initialization completed.
/// Called by the forked child when `run_daemon` returns an error.
pub fn cleanup_after_startup_error() {
    cleanup_runtime_state(false);
}

fn cleanup_session_tmp(keep_session_tmp: bool) {
    cleanup_session_tmp_at(&session_tmp_dir(), keep_session_tmp);
}

fn cleanup_session_tmp_at(session_dir: &Path, keep_session_tmp: bool) {
    if keep_session_tmp {
        eprintln!(
            "[dbg] warning: raw session evidence was not removed; recover it from {}",
            session_dir.display()
        );
    } else {
        let _ = std::fs::remove_dir_all(&session_dir);
    }
}

/// Return the session-scoped temp directory (without appending a filename).
fn session_tmp_dir() -> PathBuf {
    // session_tmp("x") gives <dir>/x — take the parent to get <dir>
    let with_file = session_tmp("x");
    with_file.parent().unwrap_or(&with_file).to_path_buf()
}

fn users_uid() -> u32 {
    // Do not trust the conventional `$UID` environment variable. A caller
    // can set it to another user's value and otherwise redirect the IPC
    // pathname into a directory that is not theirs. `/proc` reports the
    // effective UID assigned by the kernel.
    std::fs::read_to_string("/proc/self/status")
        .ok()
        .and_then(|s| {
            s.lines()
                .find(|l| l.starts_with("Uid:"))
                .and_then(|l| l.split_ascii_whitespace().nth(1)?.parse().ok())
        })
        .unwrap_or(0)
}

/// Per-user IPC directory. Prefers `$XDG_RUNTIME_DIR/dbg-<uid>` (tmpfs,
/// already private), falls back to `/tmp/dbg-<uid>` with `0700` perms.
/// Containing the socket+pid in a `0700` dir keeps other local users
/// from racing the socket creation or hijacking the pid file — Unix
/// socket permissions aren't portable enough to rely on the socket
/// inode itself.
fn runtime_dir() -> PathBuf {
    use std::os::unix::fs::PermissionsExt;
    // Unix user — used to scope the IPC dir per-user so collisions
    // and permission races are impossible on shared /tmp.
    let uid = users_uid();
    let base = std::env::var("XDG_RUNTIME_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|_| PathBuf::from("/tmp"));
    let dir = base.join(format!("dbg-{uid}"));
    if let Err(e) = std::fs::create_dir_all(&dir) {
        panic!(
            "dbg: cannot create secure runtime directory {}: {e}",
            dir.display()
        );
    }
    let metadata = std::fs::symlink_metadata(&dir).unwrap_or_else(|e| {
        panic!(
            "dbg: cannot inspect runtime directory {}: {e}",
            dir.display()
        )
    });
    use std::os::unix::fs::MetadataExt;
    if !metadata.is_dir() || metadata.uid() != uid {
        panic!("dbg: refusing insecure runtime directory {}", dir.display());
    }
    if let Err(e) = std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)) {
        panic!(
            "dbg: cannot secure runtime directory {}: {e}",
            dir.display()
        );
    }
    let mode = std::fs::symlink_metadata(&dir).unwrap().mode();
    if mode & 0o077 != 0 {
        panic!(
            "dbg: runtime directory {} is writable by another user",
            dir.display()
        );
    }
    dir
}

/// Sanitize a user-supplied `DBG_SESSION` value to the slug charset.
fn sanitize_slug(v: &str) -> String {
    v.chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// Stable hash of the current working directory. Every slug in a given
/// cwd starts with this prefix, so scanning the runtime dir for peers
/// is a simple `starts_with` check.
fn cwd_base_slug() -> String {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/"));
    let mut h = DefaultHasher::new();
    cwd.hash(&mut h);
    format!("{:016x}", h.finish())
}

/// Path to the "most recent daemon for this cwd" pointer.
/// Clients with no `DBG_SESSION` read this to find a live daemon.
pub fn latest_pointer_path() -> PathBuf {
    runtime_dir().join(format!("dbg-{}.latest", cwd_base_slug()))
}

fn latest_pointer_lock_path() -> PathBuf {
    runtime_dir().join(format!("dbg-{}.latest.lock", cwd_base_slug()))
}

fn with_latest_pointer_lock<T>(f: impl FnOnce() -> T) -> T {
    let path = latest_pointer_lock_path();
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .open(path)
        .expect("cannot open latest-session pointer lock");
    let lock = nix::fcntl::Flock::lock(file, nix::fcntl::FlockArg::LockExclusive)
        .expect("cannot lock latest-session pointer");
    let result = f();
    drop(lock);
    result
}

/// Update the pointer to name `slug` as the newest daemon for this cwd.
pub fn write_latest_pointer(slug: &str) {
    with_latest_pointer_lock(|| {
        let path = latest_pointer_path();
        let tmp = path.with_extension(format!("latest.{}", std::process::id()));
        if std::fs::write(&tmp, format!("{slug}\n")).is_ok() {
            let _ = std::fs::rename(tmp, path);
        }
    });
}

fn clear_latest_pointer_if_owner(slug: &str) {
    with_latest_pointer_lock(|| {
        let path = latest_pointer_path();
        if std::fs::read_to_string(&path)
            .ok()
            .is_some_and(|contents| contents.trim() == slug)
        {
            let _ = std::fs::remove_file(path);
        }
    });
}

/// Read the pointer. Returns `None` if missing, empty, or pointing at
/// a daemon that is no longer alive.
fn read_live_pointer() -> Option<String> {
    with_latest_pointer_lock(|| {
        let raw = std::fs::read_to_string(latest_pointer_path()).ok()?;
        let slug = raw.trim();
        if slug.is_empty() || !is_slug_live(slug) {
            return None;
        }
        Some(slug.to_string())
    })
}

/// Slug the current process should act on. Priority:
///   1. `DBG_SESSION` env var (explicit caller choice)
///   2. `.latest` pointer file (newest live daemon in this cwd)
///   3. Bare `cwd_base_slug()` (legacy fallback; also the slug the
///      very first daemon in a fresh cwd gets)
fn session_slug() -> String {
    if let Ok(v) = std::env::var("DBG_SESSION") {
        let safe = sanitize_slug(&v);
        if !safe.is_empty() {
            return safe;
        }
    }
    if let Some(s) = read_live_pointer() {
        return s;
    }
    cwd_base_slug()
}

pub fn socket_path_for(slug: &str) -> PathBuf {
    runtime_dir().join(format!("dbg-{slug}.sock"))
}

pub fn pid_path_for(slug: &str) -> PathBuf {
    runtime_dir().join(format!("dbg-{slug}.pid"))
}

fn reservation_path_for(slug: &str) -> PathBuf {
    runtime_dir().join(format!("dbg-{slug}.lock"))
}

fn socket_path() -> PathBuf {
    socket_path_for(&session_slug())
}

fn pid_path() -> PathBuf {
    pid_path_for(&session_slug())
}

/// Same liveness check as `is_running`, scoped to a specific slug.
pub fn is_slug_live(slug: &str) -> bool {
    let Ok(pid_str) = std::fs::read_to_string(pid_path_for(slug)) else {
        return false;
    };
    let Some(pid) = pid_str
        .lines()
        .next()
        .and_then(|s| s.trim().parse::<i32>().ok())
    else {
        return false;
    };
    if nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_err() {
        return false;
    }
    daemon_identity_matches(&pid_str, pid) && proc_comm_is_dbg(pid)
}

/// Pick a slug for a new daemon. With `DBG_SESSION` set we honor the
/// explicit choice and fail loudly on collision — the user asked for a
/// named slot. Without `DBG_SESSION` we prefer the bare cwd slug and,
/// if that slot is busy, append `-<pid>` to coexist with the existing
/// daemon rather than evict it.
pub fn allocate_slug() -> Result<String> {
    let explicit = std::env::var("DBG_SESSION")
        .ok()
        .map(|v| sanitize_slug(&v))
        .filter(|s| !s.is_empty());
    if let Some(name) = explicit {
        if is_slug_live(&name) {
            anyhow::bail!(
                "DBG_SESSION={name} is already live — pick another name or \
                 `DBG_SESSION={name} dbg kill` first"
            );
        }
        claim_slug(&name)?;
        return Ok(name);
    }
    let base = cwd_base_slug();
    if !is_slug_live(&base) && claim_slug(&base).is_ok() {
        return Ok(base);
    }
    for suffix in 0..1000u32 {
        let name = if suffix == 0 {
            format!("{base}-{}", std::process::id())
        } else {
            format!("{base}-{}-{suffix}", std::process::id())
        };
        if !is_slug_live(&name) && claim_slug(&name).is_ok() {
            return Ok(name);
        }
    }
    anyhow::bail!("cannot allocate a free debugger session slot")
}

type SlugReservation = nix::fcntl::Flock<File>;

/// The reservation file is deliberately never removed. Ownership is the
/// kernel's exclusive flock on this stable inode, not a PID written into a
/// pathname. This makes stale recovery safe: after an owner dies, the next
/// starter acquires the same inode; it can never remove a path that another
/// starter acquired in the meantime.
static SLUG_RESERVATIONS: OnceLock<Mutex<HashMap<PathBuf, SlugReservation>>> = OnceLock::new();

fn slug_reservations() -> &'static Mutex<HashMap<PathBuf, SlugReservation>> {
    SLUG_RESERVATIONS.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Atomically reserve a slug before the caller forks. This closes the
/// check-then-create window between `allocate_slug` and daemon startup.
fn claim_slug(slug: &str) -> Result<()> {
    let path = reservation_path_for(slug);
    let mut reservations = slug_reservations()
        .lock()
        .unwrap_or_else(|e| e.into_inner());
    if let Entry::Occupied(_) = reservations.entry(path.clone()) {
        anyhow::bail!("session slot `{slug}` is already reserved")
    }

    // O_NOFOLLOW prevents a malicious or damaged runtime entry from
    // redirecting this open outside the per-user runtime directory.
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .custom_flags(nix::libc::O_NOFOLLOW)
        .open(&path)
        .with_context(|| format!("cannot open slug reservation {}", path.display()))?;
    #[cfg(test)]
    test_reservation_contender_ready();
    let lock = match nix::fcntl::Flock::lock(file, nix::fcntl::FlockArg::LockExclusiveNonblock) {
        Ok(lock) => lock,
        Err((_file, _err)) => {
            anyhow::bail!("session slot `{slug}` is already reserved")
        }
    };
    #[cfg(test)]
    test_reservation_lock_acquired();

    // A daemon started by an older binary may have a live PID marker but no
    // flock. Check it only while we own the stable reservation inode. No
    // pathname can be removed and replaced between this check and claim.
    if is_slug_live(slug) {
        anyhow::bail!("session slot `{slug}` is already live")
    }

    reservations.insert(path, lock);
    Ok(())
}

#[cfg(test)]
fn test_reservation_contender_ready() {
    let Ok(raw) = std::env::var("DBG_TEST_RESERVATION_BARRIER") else {
        return;
    };
    let barrier = PathBuf::from(raw);
    std::fs::create_dir_all(&barrier).expect("create reservation test barrier");
    std::fs::write(
        barrier.join(format!("contender-{}", std::process::id())),
        b"ready",
    )
    .expect("publish reservation contender");
    while !barrier.join("begin").exists() {
        std::thread::sleep(Duration::from_millis(2));
    }
}

#[cfg(test)]
fn test_reservation_lock_acquired() {
    let Ok(raw) = std::env::var("DBG_TEST_RESERVATION_BARRIER") else {
        return;
    };
    let barrier = PathBuf::from(raw);
    std::fs::write(
        barrier.join(format!("locked-{}", std::process::id())),
        b"locked",
    )
    .expect("publish reservation lock");
    while !barrier.join("release").exists() {
        std::thread::sleep(Duration::from_millis(2));
    }
}

/// Every slug that currently has a live daemon whose pid file lives in
/// `runtime_dir()` and whose slug begins with the current cwd's base.
/// Used by `dbg sessions` to enumerate peer daemons.
pub fn live_slugs_in_cwd() -> Vec<String> {
    let base = cwd_base_slug();
    let Ok(entries) = std::fs::read_dir(runtime_dir()) else {
        return Vec::new();
    };
    let mut out = Vec::new();
    for entry in entries.flatten() {
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        let Some(rest) = name.strip_prefix("dbg-") else {
            continue;
        };
        let Some(slug) = rest.strip_suffix(".pid") else {
            continue;
        };
        if !slug.starts_with(&base) {
            continue;
        }
        if is_slug_live(slug) {
            out.push(slug.to_string());
        }
    }
    out.sort();
    out
}

/// File the forked daemon's stderr is redirected to during startup.
/// The parent reads it back when the daemon dies before binding the
/// socket, so the agent sees the real failure reason instead of
/// "daemon failed to start".
pub fn startup_log_path() -> PathBuf {
    runtime_dir().join(format!("dbg-{}.startup.log", session_slug()))
}

/// Detach the current (post-fork) process from its parent's stdio.
///
/// Single-fork + setsid alone is not sufficient when the parent captures
/// stdout/stderr via pipes (e.g. Claude Code's Bash tool): the captured
/// pipe stays open as long as the daemon inherits fd 1/2, the parent
/// blocks waiting for EOF, and when the harness tears the tool call down
/// it kills the whole process tree — taking the daemon with it.
///
/// Redirect fd 0/1 to /dev/null and fd 2 to `log_path` so the parent's
/// pipe can close and the daemon survives the tool call.
///
/// Must be called from the forked daemon child, after `setsid()`.
pub fn detach_stdio(log_path: &std::path::Path) {
    use std::os::unix::io::AsRawFd;
    let devnull = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open("/dev/null")
        .ok();
    if let Some(dn) = &devnull {
        let _ = nix::unistd::dup2(dn.as_raw_fd(), 0);
        let _ = nix::unistd::dup2(dn.as_raw_fd(), 1);
    }
    if let Ok(f) = std::fs::OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(true)
        .open(log_path)
    {
        let _ = nix::unistd::dup2(f.as_raw_fd(), 2);
    } else if let Some(dn) = &devnull {
        // Never retain the parent's capture pipe merely because the
        // preferred startup log could not be opened.
        let _ = nix::unistd::dup2(dn.as_raw_fd(), 2);
    }
}

/// Session-scoped temp directory for profile data etc.
/// Uses a random ID to avoid collisions between concurrent sessions.
pub fn session_tmp(filename: &str) -> PathBuf {
    use std::sync::OnceLock;
    static SESSION_ID: OnceLock<String> = OnceLock::new();
    let id = SESSION_ID.get_or_init(|| {
        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};
        use std::time::SystemTime;
        let mut h = DefaultHasher::new();
        SystemTime::now().hash(&mut h);
        std::process::id().hash(&mut h);
        format!("{:08x}", h.finish() as u32)
    });
    let dir = runtime_dir().join(format!("dbg-{id}"));
    let _ = std::fs::create_dir_all(&dir);
    dir.join(filename)
}

struct Session {
    proc: Box<dyn DebuggerIo>,
    profile: Option<ProfileData>,
    /// Per-run SessionDb. `None` if DB creation failed on startup — we
    /// keep the debugger session alive either way; capture is best-effort.
    db: Option<SessionDb>,
    /// Monotonic hit counter per canonical `location_key`. Drives
    /// `breakpoint_hits.hit_seq` so agents can reference "the 5th hit
    /// at main.c:42".
    hit_seq: HashMap<String, u32>,
    /// Where to back the DB up on graceful shutdown (if the session
    /// accumulated any captured data and is still marked auto).
    save_to: Option<PathBuf>,
    /// Private raw-evidence directory for this session.
    tmp_dir: PathBuf,
    /// Target path — used by crosstrack collectors (dbg disasm, etc.).
    target: String,
    /// Working directory at session start — used to resolve
    /// `.dbg/sessions/<label>.db` paths and on-demand collectors.
    cwd: PathBuf,
    /// Classification chosen at startup from the backend name; picks
    /// which collector (lldb / jitdasm / go-objdump / …) runs.
    target_class: TargetClass,
}

/// LiveDebugger impl that hands `crosstrack::run` a narrow handle onto
/// the session PTY. Only used while the daemon holds the session lock
/// — so this is called synchronously from `handle_command`.
struct ProcLive<'a> {
    proc: &'a dyn DebuggerIo,
    tool_name: &'static str,
}

impl LiveDebugger for ProcLive<'_> {
    fn send(&self, cmd: &str) -> anyhow::Result<String> {
        self.proc.send_and_wait(cmd, CMD_TIMEOUT)
    }
    fn tool_name(&self) -> &'static str {
        self.tool_name
    }
}

/// Infer which `TargetClass` to stamp on the session row from the
/// backend name. Profilers and backends that don't fit the debug
/// taxonomy fall through to `NativeCpu` — the class is informational
/// for those, not driving any per-class domain tables.
fn backend_target_class(name: &str) -> TargetClass {
    match name {
        "pdb" | "debugpy-proto" => TargetClass::Python,
        "netcoredbg" | "netcoredbg-proto" | "dotnet-trace" | "jitdasm" => {
            TargetClass::ManagedDotnet
        }
        "jdb" => TargetClass::Jvm,
        "node-inspect" | "nodeprof" | "node-proto" => TargetClass::JsNode,
        "rdbg" | "stackprof" => TargetClass::Ruby,
        "phpdbg" | "xdebug-profile" => TargetClass::Php,
        // lldb, delve, perf, callgrind, massif, memcheck,
        // ghci, ocamldebug, etc. — all native-style by default.
        _ => TargetClass::NativeCpu,
    }
}

fn backend_session_kind(backend: &dyn Backend) -> SessionKind {
    if backend.is_profile_backend() {
        SessionKind::Profile
    } else {
        SessionKind::Debug
    }
}

/// Does this canonical operation cause an execution-state transition?
/// Only these ops can produce a *new* breakpoint hit. Inspection ops
/// (`stack`, `locals`, `print`, `breaks`) echo the current stop state
/// without transitioning and must NOT feed into `capture_hit_if_stopped`
/// or they create duplicate hit rows.
fn op_may_stop(canonical_op: &str) -> bool {
    matches!(
        canonical_op,
        "run" | "continue" | "step" | "next" | "finish" | "restart" | "pause"
    )
}

/// Same idea for raw/fallthrough commands that bypass the canonical
/// dispatcher — match on the native command verb.
fn command_may_stop(cmd: &str) -> bool {
    let first = cmd.trim().split_whitespace().next().unwrap_or("");
    matches!(
        first,
        // Canonical verbs that may cause a stop...
        "run" | "continue" | "step" | "next" | "finish"
            | "c" | "s" | "n" | "r"
            // ... and the most common native forms across backends.
            | "process" | "thread" | "stepout" | "return" | "restart"
    )
}

/// Verbs owned by the in-memory Profile REPL (see `ProfileData::handle_command`).
/// Used to decide whether a command during a profile-mode session should
/// be handled by the profile REPL or fall through to the normal dispatch
/// path — meta verbs like `sessions`, `status`, `save`, `kill`, `replay`
/// must still work during a profile session.
pub fn is_profile_repl_verb(cmd: &str) -> bool {
    let first = cmd.trim().split_whitespace().next().unwrap_or("");
    matches!(
        first,
        "top"
            | "callers"
            | "callees"
            | "traces"
            | "tree"
            | "hotpath"
            | "threads"
            | "stats"
            | "search"
            | "focus"
            | "ignore"
            | "reset"
    )
}

/// Read the profile source file at `src` and copy it into the session
/// DB's meta table so a subsequent `dbg replay` can rebuild a
/// `ProfileData` from it. Errors are returned so the caller can record
/// them in the same durable session. The format hint comes
/// from the file extension so V8 cpuprofile vs speedscope JSON
/// disambiguation matches what `ProfileData::load` would have used on
/// the original path.
fn persist_profile_source(db: &SessionDb, src: &Path) -> Result<()> {
    const MAX_PROFILE_SOURCE_BYTES: u64 = 256 * 1024 * 1024;
    let metadata =
        std::fs::metadata(src).with_context(|| format!("stat profile source {}", src.display()))?;
    if metadata.len() > MAX_PROFILE_SOURCE_BYTES {
        anyhow::bail!(
            "profile source {} is too large ({} bytes; maximum is {} bytes)",
            src.display(),
            metadata.len(),
            MAX_PROFILE_SOURCE_BYTES
        );
    }
    let content = crate::profile::read_bounded_text(src, MAX_PROFILE_SOURCE_BYTES)
        .with_context(|| format!("reading profile source {}", src.display()))?;
    db.set_meta("profile_raw", &content)?;
    if let Some(ext) = src.extension().and_then(|e| e.to_str()) {
        db.set_meta("profile_raw_ext", ext)?;
    }
    Ok(())
}

/// Start the daemon: spawn the debugger, listen on socket.
/// This function does NOT return on success — it runs the event loop.
pub fn run_daemon(
    backend: &dyn Backend,
    target: &str,
    args: &[String],
    attach: Option<&crate::backend::AttachSpec>,
) -> Result<()> {
    // Branch on transport kind. Most backends use the default PTY
    // transport; `node-proto` uses the V8 Inspector WebSocket
    // transport; DAP-capable backends (delve-proto, debugpy-proto, …)
    // use the Debug Adapter Protocol over TCP. All paths produce a
    // `Box<dyn DebuggerIo>`. Protocol backends have no post-spawn
    // init commands — their handshake happens inside the transport.
    let (proc, init_commands, fail_fast_init): (Box<dyn DebuggerIo>, Vec<String>, bool) = if backend
        .uses_inspector()
    {
        let t = crate::inspector::InspectorTransport::spawn(target, args)
            .context("failed to spawn inspector transport")?;
        (Box::new(t), Vec::new(), false)
    } else if backend.uses_dap() {
        let cfg = match attach {
            Some(spec) => backend.dap_attach(spec)?,
            None => backend.dap_launch(target, args)?,
        };
        let t = crate::dap::DapTransport::spawn(cfg).context("failed to spawn DAP transport")?;
        (Box::new(t), Vec::new(), false)
    } else {
        let config = backend.spawn_config(target, args)?;
        let p = DebuggerProcess::spawn(
            &config.bin,
            &config.args,
            &config.env,
            backend.prompt_pattern(),
        )
        .context("failed to spawn debugger")?;
        let fail_fast = config.bin == "bash" || config.bin.ends_with("/bash");
        (Box::new(p), config.init_commands, fail_fast)
    };

    // Publish the jitdasm capture path so on-demand collectors (which
    // live in the lib crate and can't reach `daemon::session_tmp`) can
    // serve `dbg disasm` from the pre-captured file rather than
    // re-spawning dotnet on every call.
    if backend.name() == "jitdasm" {
        // SAFETY: set_var is only unsafe in multi-threaded contexts due
        // to libc env races. We're still single-threaded here — this
        // runs before the socket listener spawns request handlers.
        unsafe {
            std::env::set_var(
                "DBG_JITDASM_CAPTURE",
                session_tmp("jitdasm").join("capture.asm"),
            );
        }
    }

    // Expose the child PID outside the session mutex so the quit handler
    // can SIGINT the child to interrupt a blocked `send_and_wait` (e.g.
    // jdb waiting on a `continue` that hasn't hit a breakpoint yet).
    let child_pid = AtomicI32::new(proc.child_pid().as_raw());
    let child_identity = process_identity(proc.child_pid().as_raw());

    // Pre-clone the event-log handle outside the session mutex. `dbg
    // events --wait` blocks on the log's condvar without touching the
    // session lock, so live-tailing works concurrently even while a
    // `continue` holds the mutex for minutes.
    let log_handle = proc.log();

    // Wait for initial prompt
    let startup_output = proc
        .wait_for_prompt(Duration::from_secs(120))
        .context("debugger did not produce prompt")?;

    // Publish pid + socket BEFORE running init commands. Backends
    // like jitdasm have multi-minute init flows (dotnet build +
    // disasm capture + exec-into-REPL); keeping pid/socket gated on
    // init completion made `dbg status` lie ("no session") the whole
    // time, and `dbg start` appeared to hang silently. With pid
    // written and socket bound up-front, connections queue into the
    // listener backlog and are served as soon as the accept loop
    // below picks them up — which happens right after init finishes.
    std::fs::write(&pid_path(), daemon_pid_record(std::process::id() as i32))?;
    let _ = std::fs::remove_file(&socket_path());
    let listener = UnixListener::bind(&socket_path()).context("failed to bind socket")?;
    // Point the cwd's "latest" pointer at ourselves so env-less clients
    // in other shells find this daemon by default. Parent also writes
    // this pre-fork; re-asserting here is idempotent and covers the
    // case where the parent's write was lost/raced.
    write_latest_pointer(&session_slug());

    let shutdown_requested = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let active_connections = std::sync::Arc::new(AtomicUsize::new(0));
    let shutdown_flag = shutdown_requested.clone();
    let wake_path = socket_path();
    let signal_wake_path = wake_path.clone();
    let signal_pid = child_pid.load(Ordering::Relaxed);
    let signal_identity = child_identity.clone();
    ctrlc::set_handler(move || {
        shutdown_flag.store(true, Ordering::Release);
        let _ = signal_child_identity(signal_pid, signal_identity.as_deref(), nix::libc::SIGINT);
        let _ = std::os::unix::net::UnixStream::connect(&signal_wake_path);
    })
    .ok();

    // Run init commands (PTY backends only; inspector has none).
    // Profiling backends (pstats, massif, …) may wrap a target that
    // takes minutes to finish, so the backend's `init_timeout()`
    // overrides the default CMD_TIMEOUT here.
    let init_deadline = backend.init_timeout();
    // Shell-backed profilers must stop at the first failed preparation
    // command. `send_and_wait` only waits for the prompt, so it does not
    // report a shell command's exit status. Append a private status marker
    // and inspect it after every command. This also covers interactive bash
    // contexts where `set -e` can be suppressed or changed by a command.
    for cmd in &init_commands {
        let is_exec = cmd.trim_start().starts_with("exec ");
        let response = if fail_fast_init && !is_exec {
            proc.send_and_wait(
                &format!("{cmd}; printf '\\n__DBG_INIT_STATUS__%s\\n' \"$?\""),
                init_deadline,
            )?
        } else {
            proc.send_and_wait(cmd, init_deadline)?
        };
        if fail_fast_init && !is_exec {
            let status = parse_init_status(&response);
            if status != Some(0) {
                bail!("init command failed (exit status {status:?}): {cmd}");
            }
        }
    }

    // Cache help output now while the debugger is idle and responsive.
    // Stored outside the session mutex so it can be served even when
    // the debugger is busy running a command.
    // Protocol transports expose their command surface through structured
    // requests. They do not accept the native PTY help command, so use the
    // backend's static description without sending an unsupported request.
    // PTY transports keep the native help query and its parsed output.
    let cached_help = cache_help(backend, || {
        proc.send_and_wait(backend.help_command(), CMD_TIMEOUT)
            .map(|raw| backend.parse_help(&raw))
            .context("failed to initialize debugger help")
    })?;

    let profile_output_path = backend.profile_output();
    let (profile, profile_error) = match profile_output_path.as_ref() {
        Some(path) => match ProfileData::load(Path::new(path)) {
            Ok(profile) => (Some(profile), None),
            Err(error) => (None, Some(format!("{}: {error:#}", path))),
        },
        None => (None, None),
    };

    // Build a per-run SessionDb alongside the debugger session. Create
    // A live session must always have a durable owner and failure record.
    let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
    let target_class = backend_target_class(backend.name());
    let tmp_db = session_tmp("session.db");
    let _ = std::fs::remove_file(&tmp_db); // fresh DB on every start
    let session_kind = backend_session_kind(backend);
    // Honour a user-chosen session label (e.g. `dbg import --label foo`).
    // The cmd_* layer sets DBG_LABEL before forking the daemon; we
    // pass it through to the session DB so `dbg sessions` and
    // `dbg replay <name>` resolve against the human-friendly name
    // instead of the auto-generated `<basename>-<timestamp>` form.
    let user_label = std::env::var("DBG_LABEL").ok().filter(|s| !s.is_empty());
    let (db, save_to) = match SessionDb::create(CreateOptions {
        kind: session_kind,
        target,
        target_class,
        cwd: &cwd,
        db_path: Some(&tmp_db),
        label: user_label,
        target_hash: None,
    }) {
        Ok(db) => {
            db.set_meta("backend", backend.name())?;
            // Profile sessions need their source content stashed in the
            // DB so `dbg replay` can rehydrate top/callers/callees from
            // a saved session. The raw file is on a tmp path that won't
            // survive the daemon — copy it into the SessionDb's meta
            // table now while we still know where it is.
            if let Some(ref src) = profile_output_path {
                if let Err(error) = persist_profile_source(&db, Path::new(src)) {
                    db.add_failure("profile_source", &format!("{error:#}"))?;
                }
            }
            if let Some(error) = profile_error {
                db.add_failure("profile_parse", &error)?;
            }
            let final_path = session_save_path(&cwd, db.label());
            (Some(db), final_path)
        }
        Err(e) => bail!("session DB initialization failed: {e:#}"),
    };

    let session = Mutex::new(Session {
        proc,
        profile,
        db,
        hit_seq: HashMap::new(),
        save_to,
        tmp_dir: tmp_db
            .parent()
            .map(Path::to_path_buf)
            .unwrap_or_else(session_tmp_dir),
        target: target.to_string(),
        cwd: cwd.clone(),
        target_class,
    });

    // Consume a protocol entry stop (or a PTY stop banner) at startup. This
    // makes the entry stop observable exactly once and prevents the first
    // user execution command from treating it as its own result.
    if let Ok(mut session_guard) = session.lock() {
        capture_hit_if_stopped(&mut session_guard, backend, &startup_output);
    }

    if shutdown_requested.load(Ordering::Acquire) {
        let (_, keep_session_tmp) =
            handle_quit(backend, &session, &child_pid, child_identity.as_deref());
        cleanup_and_exit(keep_session_tmp);
    }

    // Blocking accept — the daemon does nothing useful between
    // connections, and ctrl-c is delivered to the signal handler
    // regardless of whether accept is blocked. Polling burned ~100
    // wakeups/sec idle for no benefit.
    std::thread::scope(|scope| {
        loop {
            let stream = match listener.accept() {
                Ok((s, _)) => s,
                Err(_) => continue,
            };

            // The signal-handler wake must be processed even when all
            // request slots are occupied. Otherwise the wake connection is
            // rejected at the limit and shutdown waits for a blocked worker.
            if shutdown_requested.load(Ordering::Acquire) {
                let (_, keep_session_tmp) =
                    handle_quit(backend, &session, &child_pid, child_identity.as_deref());
                cleanup_and_exit(keep_session_tmp);
            }

            let previous =
                active_connections.fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| {
                    (n < MAX_IPC_CONNECTIONS).then_some(n + 1)
                });
            if previous.is_err() {
                // Refuse excess clients without allocating a handler
                // thread. The client can retry after an active request
                // completes.
                drop(stream);
                continue;
            }

            if shutdown_requested.load(Ordering::Acquire) {
                let (_, keep_session_tmp) =
                    handle_quit(backend, &session, &child_pid, child_identity.as_deref());
                cleanup_and_exit(keep_session_tmp);
            }

            let session = &session;
            let cached_help = &cached_help;
            let child_pid = &child_pid;
            let log_handle = &log_handle;
            let active_connections = active_connections.clone();
            let connection_wake_path = wake_path.clone();
            let connection_child_identity = child_identity.clone();

            scope.spawn(move || {
                struct ConnectionGuard {
                    active: std::sync::Arc<AtomicUsize>,
                    wake_path: PathBuf,
                }
                impl Drop for ConnectionGuard {
                    fn drop(&mut self) {
                        let previous = self.active.fetch_sub(1, Ordering::AcqRel);
                        // A queued client may be waiting in accept while all
                        // request slots are occupied. Wake accept only when
                        // this drop frees one of those full-limit slots.
                        // Empty wake clients must not wake themselves again.
                        if should_wake_accept_after_release(previous) {
                            let _ = std::os::unix::net::UnixStream::connect(&self.wake_path);
                        }
                    }
                }
                let _connection_guard = ConnectionGuard {
                    active: active_connections,
                    wake_path: connection_wake_path,
                };
                let mut stream = stream;
                let _ = stream.set_read_timeout(Some(IPC_IDLE_TIMEOUT));
                let _ = stream.set_write_timeout(Some(IPC_IDLE_TIMEOUT));
                let (read_result, bytes) = read_ipc_request(&mut stream);
                if read_result.is_err() || bytes.len() > MAX_IPC_REQUEST_BYTES as usize {
                    let _ = stream.write_all(b"[error: IPC request exceeds size or idle limit]");
                    return;
                }
                let Ok(data) = String::from_utf8(bytes) else {
                    let _ = stream.write_all(b"[error: IPC request is not valid UTF-8]");
                    return;
                };
                let cmd = data.trim().to_string();

                if cmd.is_empty() {
                    return;
                }

                if cmd == "quit" {
                    let (response, keep_session_tmp) = handle_quit(
                        backend,
                        session,
                        child_pid,
                        connection_child_identity.as_deref(),
                    );
                    let _ = stream.write_all(response.as_bytes());
                    let _ = stream.flush();
                    // Close the write half so the client's read_to_string
                    // returns EOF cleanly before we drop the socket —
                    // without this the client sees SIGPIPE/EPIPE and
                    // exits 144.
                    let _ = stream.shutdown(std::net::Shutdown::Write);
                    // Give the kernel a brief moment to deliver the last
                    // bytes to the client's socket buffer.
                    std::thread::sleep(Duration::from_millis(20));
                    // Exit immediately — scoped threads would otherwise wait
                    // for any blocked command (e.g. `continue`) to finish.
                    cleanup_and_exit(keep_session_tmp);
                }

                if cmd == "cancel" {
                    let response = handle_cancel(child_pid, connection_child_identity.as_deref());
                    let _ = stream.write_all(response.as_bytes());
                    return;
                }

                let response = handle_command(&cmd, backend, session, cached_help, log_handle);
                let _ = stream.write_all(response.as_bytes());
            });
        }
    });

    cleanup_and_exit(false);
}

fn read_ipc_request(stream: &mut std::os::unix::net::UnixStream) -> (std::io::Result<()>, Vec<u8>) {
    read_ipc_request_with_timeout(stream, IPC_IDLE_TIMEOUT)
}

fn parse_init_status(response: &str) -> Option<i32> {
    response
        .lines()
        .rev()
        .find_map(|line| line.strip_prefix("__DBG_INIT_STATUS__"))
        .and_then(|value| value.trim().parse::<i32>().ok())
}

fn cache_help(
    backend: &dyn Backend,
    native_help: impl FnOnce() -> Result<String>,
) -> Result<String> {
    if !backend.supports_help_command() {
        Ok(backend.parse_help(""))
    } else {
        native_help()
    }
}

fn native_help_topic_command(backend: &dyn Backend, topic: &str) -> Option<String> {
    backend
        .supports_help_command()
        .then(|| format!("{} {topic}", backend.help_command()))
}

fn jdb_run_needs_vm_prompt(backend: &dyn Backend, canonical_op: &str, output: &str) -> bool {
    backend.name() == "jdb"
        && canonical_op == "run"
        && !output.contains("VM Started:")
        && !output.contains("The application exited")
        && !output.contains("The application has been disconnected")
}

fn read_ipc_request_with_timeout(
    stream: &mut std::os::unix::net::UnixStream,
    idle_timeout: Duration,
) -> (std::io::Result<()>, Vec<u8>) {
    let deadline = Instant::now() + idle_timeout;
    let _ = stream.set_read_timeout(Some(idle_timeout));
    let mut bytes = Vec::new();
    let mut chunk = [0u8; 8192];
    let result = loop {
        if Instant::now() >= deadline {
            break Err(std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                "IPC request exceeded idle limit",
            ));
        }
        match stream.read(&mut chunk) {
            Ok(0) => break Ok(()),
            Ok(n) => {
                bytes.extend_from_slice(&chunk[..n]);
                if bytes.len() > MAX_IPC_REQUEST_BYTES as usize {
                    break Err(std::io::Error::new(
                        std::io::ErrorKind::InvalidData,
                        "IPC request is too large",
                    ));
                }
                // Successful reads must not refresh the deadline. A
                // slow-drip client must not retain a handler forever.
                if Instant::now() >= deadline {
                    break Err(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        "IPC request exceeded idle limit",
                    ));
                }
            }
            Err(e)
                if matches!(
                    e.kind(),
                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
                ) && Instant::now() < deadline =>
            {
                continue;
            }
            Err(e) => break Err(e),
        }
    };
    (result, bytes)
}

/// Lock the session mutex, recovering from poisoning so the daemon
/// stays responsive after a thread panic.
fn lock_session(session: &Mutex<Session>) -> std::sync::MutexGuard<'_, Session> {
    session.lock().unwrap_or_else(|e| e.into_inner())
}

/// Graceful quit: interrupt the child so any blocked `send_and_wait`
/// returns, acquire the session lock, persist the DB, then quit the
/// debugger. This fixes the jdb timing issue where the 2s lock timeout
/// wasn't enough for a slow `continue` to finish.
fn handle_quit(
    backend: &dyn Backend,
    session: &Mutex<Session>,
    child_pid: &AtomicI32,
    child_identity: Option<&str>,
) -> (String, bool) {
    handle_quit_with(backend, session, child_pid, child_identity, &|path| {
        Ok(Box::new(std::fs::read_dir(path)?)
            as Box<
                dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>,
            >)
    })
}

fn handle_quit_with(
    backend: &dyn Backend,
    session: &Mutex<Session>,
    child_pid: &AtomicI32,
    child_identity: Option<&str>,
    read_dir: &dyn Fn(
        &Path,
    ) -> std::io::Result<
        Box<dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>>,
    >,
) -> (String, bool) {
    // SIGINT the child to interrupt any blocked PTY read (e.g. jdb
    // waiting for a breakpoint during `continue`). This causes the
    // `send_and_wait` in the command thread to return promptly,
    // releasing the session lock so we can acquire it.
    let pid = child_pid.load(Ordering::Relaxed);
    let _ = signal_child_identity(pid, child_identity, nix::libc::SIGINT);

    // Give the interrupted command thread time to release the lock. Do not
    // report success and remove the temp directory while it is still held:
    // the database is the only durable copy of captured evidence.
    loop {
        if let Ok(mut guard) = session.try_lock() {
            let keep_session_tmp = !persist_session_on_exit_with(&mut guard, read_dir);
            // Preserve raw evidence after a failed durable copy, but always
            // stop the owned debugger before the daemon exits. Once the
            // socket is removed there is no way to recover a live child.
            guard.proc.quit(backend.quit_command());
            break ("stopped".to_string(), keep_session_tmp);
        }
        std::thread::sleep(Duration::from_millis(50));
    }
}

/// Interrupt the running command by SIGINT'ing the child without
/// tearing down the session. Unlike `quit`, this keeps the debugger
/// alive: the blocked `send_and_wait` returns (prompt reappears after
/// the interrupt), the session lock drops, and subsequent commands
/// proceed normally. Used to break out of a `continue` that hasn't
/// hit a breakpoint, an infinite loop in the target, etc.
fn handle_cancel(child_pid: &AtomicI32, child_identity: Option<&str>) -> String {
    match signal_child_identity(
        child_pid.load(Ordering::Relaxed),
        child_identity,
        nix::libc::SIGINT,
    ) {
        Ok(()) => "interrupted".to_string(),
        Err(e) => format!("[error: {e}]"),
    }
}

fn handle_command(
    cmd: &str,
    backend: &dyn Backend,
    session: &Mutex<Session>,
    cached_help: &str,
    log_handle: &crate::pty::LogHandle,
) -> String {
    // Serve help from cache — no lock needed, works even when the debugger is busy.
    if cmd == "help" {
        return cached_help.to_string();
    }

    if cmd == "events" || cmd.starts_with("events ") {
        // Uses the pre-cloned log handle — does NOT touch the session
        // mutex, so live-tailing is concurrent with a blocked command.
        return handle_events(cmd, log_handle);
    }

    // Profile mode: handle commands from in-memory profile data.
    // Only intercept verbs the profile REPL actually owns — meta-commands
    // (sessions/status/save/kill/replay/…) and crosstrack verbs
    // (hits/cross/…) must still reach the normal dispatcher. Previously
    // `profile.handle_command` caught every command and returned
    // "unknown command: sessions" for meta verbs during a profile session.
    if is_profile_repl_verb(cmd) {
        let mut guard = lock_session(session);
        if let Some(ref mut profile) = guard.profile {
            return profile.handle_command(cmd);
        }
    }

    if let Some(topic) = cmd.strip_prefix("help ") {
        // dbg's own verbs (start, kill, hits, break, …) — served from
        // a static registry so `dbg help start` actually returns text
        // even when the backend doesn't know about it.
        let topic = topic.trim();
        if let Some(text) = dbg_verb_help(topic) {
            return text.to_string();
        }
        let Some(help_command) = native_help_topic_command(backend, topic) else {
            return cached_help.to_string();
        };
        let guard = match session.try_lock() {
            Ok(g) => g,
            Err(_) => return "[busy] debugger is running a command — try again".to_string(),
        };
        return guard
            .proc
            .send_and_wait(&help_command, CMD_TIMEOUT)
            .unwrap_or_else(|e| format!("[error: {e}]"));
    }

    // Route through the unified dispatcher: cross-track verbs first
    // (hits/hit-diff/cross/disasm/source/at-hit), canonical debug
    // verbs next (break/step/continue/…), then Fallthrough to the
    // legacy passthrough path for anything else.
    match commands::dispatch(cmd, backend) {
        Dispatched::Immediate(resp) => {
            let mut guard = lock_session(session);
            // Drain deferred stop banners even for immediate commands.
            // node-inspect's `locals` dispatches as Immediate (unsupported),
            // but a prior `continue` may have left a banner in the PTY.
            drain_pending_events(&mut guard, backend);
            log_command(&mut guard, cmd, &resp, Some("meta"));
            resp
        }
        Dispatched::Native {
            canonical_op,
            native_cmd,
            decorate,
            structured,
        } => {
            let mut guard = lock_session(session);
            // Drain any deferred stop banner from a prior async
            // execution command before sending the next one.
            drain_pending_events(&mut guard, backend);

            // Post-mortem fallback: once the debuggee has exited, live
            // inspection verbs (stack/locals) can still be answered
            // from the last captured breakpoint hit in the session DB.
            // Execution verbs (step/continue/run/…) have no useful
            // fallback — return a directed exit message instead.
            if !guard.proc.is_alive() {
                // First post-mortem access also flushes the DB to
                // `.dbg/sessions/<label>.db` so `dbg kill` isn't
                // required to make the run discoverable via
                // `dbg sessions` / `dbg replay`.
                persist_session_on_exit(&mut guard);
                let fallback = post_mortem_fallback(&guard, canonical_op);
                log_command(&mut guard, cmd, &fallback, Some(canonical_op));
                return fallback;
            }

            let send_result = match structured.as_ref() {
                Some(req) => match guard.proc.dispatch_structured(req, CMD_TIMEOUT) {
                    Some(r) => r,
                    None => guard.proc.send_and_wait(&native_cmd, CMD_TIMEOUT),
                },
                None => guard.proc.send_and_wait(&native_cmd, CMD_TIMEOUT),
            };
            let send_result = match send_result {
                Ok(mut raw) if jdb_run_needs_vm_prompt(backend, canonical_op, &raw) => {
                    // JDB emits one transient prompt while it starts the VM.
                    // Wait for the next prompt so the run command returns at
                    // the breakpoint or exit boundary.
                    guard.proc.wait_for_prompt(CMD_TIMEOUT).map(|late| {
                        raw.push_str(&late);
                        raw
                    })
                }
                result => result,
            };
            match send_result {
                Ok(raw) => {
                    // A stop banner can arrive after the command that
                    // caused it has already acknowledged its prompt. It
                    // can therefore be consumed by an inspection request
                    // that was queued next. Parse every response so that
                    // such a stop is not discarded just because the
                    // inspection verb itself is non-executing.
                    if op_may_stop(canonical_op) || !raw.trim().is_empty() {
                        capture_hit_if_stopped(&mut guard, backend, &raw);
                    }
                    // When the agent explicitly runs `dbg locals`,
                    // retroactively fill in the most recent hit's
                    // locals_json if it was NULL. This enables sparklines
                    // for backends where auto_capture_locals=false: the
                    // agent's natural workflow (continue → locals →
                    // continue → locals) populates the data.
                    if canonical_op == "locals" {
                        backfill_locals(&mut guard, backend, &raw);
                    }

                    let mut cleaned = backend.clean(&native_cmd, &raw);
                    if decorate {
                        cleaned =
                            debug_cmds::decorate_output_for_op(backend, canonical_op, &cleaned);
                    }
                    log_command(&mut guard, cmd, &cleaned, Some(canonical_op));
                    cleaned
                }
                Err(e) => format!("[error: {e}]"),
            }
        }
        Dispatched::Query(q) => {
            let canonical_op = q.canonical_op();
            let mut guard = lock_session(session);
            // Drain first: a deferred stop banner from a prior async
            // execution may still be pending in the channel. Without
            // draining here, the query would report stale hit counts.
            drain_pending_events(&mut guard, backend);
            let response = run_crosstrack(&mut guard, backend, &q);
            log_command(&mut guard, cmd, &response, Some(canonical_op));
            response
        }
        Dispatched::Lifecycle(l) => {
            let canonical_op = l.canonical_op();
            let mut guard = lock_session(session);
            drain_pending_events(&mut guard, backend);
            let ctx = lifecycle_cmds::LifeCtx {
                cwd: &guard.cwd,
                active: guard.db.as_ref(),
            };
            let response = lifecycle_cmds::run(&l, &ctx);
            log_command(&mut guard, cmd, &response, Some(canonical_op));
            response
        }
        Dispatched::InsnHits(req) => {
            // Capture what the collector needs under the lock, then
            // drop the guard before running the collector. Holding
            // the session mutex for the full --window would block
            // every other command (status, kill, ...) for the whole
            // capture, which can run for minutes.
            let captured = {
                let mut guard = lock_session(session);
                drain_pending_events(&mut guard, backend);
                guard
                    .db
                    .as_ref()
                    .and_then(|db| db.db_path())
                    .map(|p| (p.to_path_buf(), guard.target.clone(), guard.cwd.clone()))
            };
            let response = match captured {
                Some((db_path, target, cwd)) => {
                    let response = match SessionDb::open(&db_path) {
                        Ok(db) => {
                            let probe = crate::commands::insnhits::HostProbeAuto;
                            let ctx = crate::commands::insnhits::CollectCtx {
                                target_binary: &target,
                                cwd: &cwd,
                            };
                            crate::commands::insnhits::run(&req, &db, &probe, &ctx)
                        }
                        Err(e) => format!(
                            "[insn-hits: could not reopen session DB at {}: {e}]",
                            db_path.display()
                        ),
                    };
                    let mut guard = lock_session(session);
                    log_command(&mut guard, cmd, &response, Some("insn-hits"));
                    response
                }
                None => "[insn-hits unavailable: SessionDb failed to initialize at \
                         start; restart the session to collect a fresh capture]"
                    .to_string(),
            };
            response
        }
        Dispatched::Fallthrough => {
            let mut guard = lock_session(session);
            drain_pending_events(&mut guard, backend);

            // For profiling backends (massif/memcheck/pprof/…) which
            // lack canonical_ops and run atop a bash PTY, `dbg run`
            // reaches the shell as the literal word `run` → "command
            // not found". Translate the bare `run` verb into the
            // backend-declared recipe (e.g. `ms_print $MASSIF_OUT`)
            // when one is available. Any other verb falls through
            // unchanged.
            let effective_cmd: String = if cmd.trim() == "run" {
                let recipe = backend.run_command();
                if recipe.is_empty() {
                    cmd.to_string()
                } else {
                    recipe.to_string()
                }
            } else {
                cmd.to_string()
            };

            match guard.proc.send_and_wait(&effective_cmd, CMD_TIMEOUT) {
                Ok(raw) => {
                    if command_may_stop(cmd) || !raw.trim().is_empty() {
                        capture_hit_if_stopped(&mut guard, backend, &raw);
                    }
                    let cleaned = backend.clean(cmd, &raw);
                    log_command(&mut guard, cmd, &cleaned, None);
                    cleaned
                }
                Err(e) => format!("[error: {e}]"),
            }
        }
    }
}

/// Record a `commands` row for the just-issued command. `canonical_op`
/// is `Some(...)` when the command went through the canonical
/// dispatcher (e.g. `"break"`, `"step"`, `"raw"`, `"meta"`) and `None`
/// for legacy passthrough.
fn log_command(session: &mut Session, input: &str, output: &str, canonical_op: Option<&str>) {
    let db = match session.db.as_ref() {
        Some(d) => d,
        None => return,
    };
    let head: String = output.chars().take(4096).collect();
    let output_bytes = output.len() as i64;
    let _ = db.conn().execute(
        "INSERT INTO commands (session_id, input, output_head, output_bytes, ts, canonical_op)
         VALUES (
             (SELECT id FROM sessions LIMIT 1),
             ?1, ?2, ?3, datetime('now'), ?4
         )",
        params![input, head, output_bytes, canonical_op],
    );
}

/// Help text for dbg-level verbs. Returns `None` when the verb is
/// unknown — callers fall through to the backend's own help.
pub fn dbg_verb_help(verb: &str) -> Option<&'static str> {
    Some(match verb {
        "start" => {
            "\
dbg start <type> <target> [--break SPEC] [--args ...] [--run]
dbg start <dap-type> <target-hint> --attach-pid N
  Spawn a debugger session. <type> may be omitted when <target>'s
  extension unambiguously identifies a backend (.py, .go, .java, .rb,
  .php, .csproj, .js, .ts, .hs, .ml).
  --break SPEC        set a breakpoint before run (repeatable)
  --args a b c        forward args to the debuggee
  --run               continue past the debugger's startup prompt;
                      breakpoints still fire.
  --attach-pid N      attach to a running process with an explicit DAP
                      backend; put this after <type> <target-hint>.
                      Use the debuggee child PID, not a launcher shell.
                      Relative paths resolve from dbg start's cwd.
  --attach-port H:P   attach via host:port (DAP backends)"
        }
        "kill" | "quit" => {
            "\
dbg kill
  Stop the active session, persisting captured data to .dbg/sessions/
  before exit. Idempotent — safe to run when no session is live."
        }
        "status" => {
            "\
dbg status
  Show live session info: backend, target, PID, elapsed time. Returns
  \"no session\" when nothing is running."
        }
        "sessions" => {
            "\
dbg sessions
  List persisted sessions under .dbg/sessions/. The currently-live
  session (if any) is marked with *."
        }
        "replay" => {
            "\
dbg replay <label>
  Open a persisted session read-only and run crosstrack queries against
  it. Same vocabulary as a live session: hits, hit-trend, hit-diff,
  cross, disasm, source. No live debugger — exec verbs are rejected."
        }
        "insn-hits" => {
            "\
dbg insn-hits <symbol|0xADDR> [--live | --window <duration>]
              [--with-stack] [--with-regs]
              [--backend auto|pt|etm|uprobe|hwbp|ibs|pebs|spe] [--why]
  Count how many times an instruction (or symbol entry) executed.
  The planner picks the cheapest backend that satisfies the requested
  flags from those available on the host. Pass --why to see the
  trail (chosen backend, eligible alternatives, why each rejected
  backend lost)."
        }
        "run" => {
            "\
dbg run
  Start the debuggee (or restart it). Stops at the first breakpoint
  that fires."
        }
        "continue" | "c" => {
            "\
dbg continue
  Resume the debuggee until the next breakpoint, signal, or exit."
        }
        "step" | "s" => {
            "\
dbg step
  Step into the next line (source-level)."
        }
        "next" | "n" => {
            "\
dbg next
  Step over the next line, not descending into function calls."
        }
        "finish" => {
            "\
dbg finish
  Run until the current function returns."
        }
        "break" | "b" => {
            "\
dbg break <spec>
  Set a breakpoint. <spec> is one of:
    file:line               — e.g. main.c:42, broken.py:20
    function                — e.g. MyClass.run, main
    /abs/file:line          — absolute paths always work unambiguously
  For pdb/delve/others, breaking on a `def` / function-header line
  may not fire until the first body line — dbg auto-advances when it
  can detect this; otherwise use file:line with a body line."
        }
        "locals" => {
            "\
dbg locals
  Print local variables at the current stop. Also backfills
  locals_json on the most recent captured hit, enabling
  `dbg hit-trend`/`--group-by` sparklines progressively."
        }
        "stack" | "bt" => {
            "\
dbg stack
  Print the call stack at the current stop."
        }
        "print" | "p" => {
            "\
dbg print <expr>
  Evaluate <expr> in the debuggee's current frame."
        }
        "hits" => {
            "\
dbg hits <loc> [--group-by FIELD] [--count-by FIELD --top N]
  List captured breakpoint hits at <loc>. With --group-by, aggregate
  by a locals field: `dbg hits broken.py:20 --group-by n` → count per
  distinct n. --count-by is an alias; --top N truncates to the most
  frequent. Dotted paths supported (self.x)."
        }
        "hit-diff" => {
            "\
dbg hit-diff <loc> <seq_a> <seq_b>
  Show a field-by-field diff of captured locals between two hits."
        }
        "hit-trend" => {
            "\
dbg hit-trend <loc> <field>
  Render a sparkline of a locals field across hits. Accepts dotted
  paths (self.x). When the field isn't captured, enumerates available
  field names in the error so you can pick one."
        }
        "cross" => {
            "\
dbg cross <symbol>
  Show everything known about <symbol>: hit count, profile samples,
  jit events, disassembly rows, source snapshots."
        }
        "disasm" => {
            "\
dbg disasm [<symbol>] [--refresh]
  Capture and show disassembly for <symbol>. Omit <symbol> at a stop
  point to disasm the current frame. --refresh forces recollection."
        }
        "source" => {
            "\
dbg source <symbol> [radius=5]
  Show the source of <symbol> with ±radius lines of context."
        }
        "events" => {
            "\
dbg events [--since=SEQ] [--tail=N] [--kind=stop,stdout,...] [--wait=MS]
  Live-tail the session's PTY event log. Does not touch the session
  mutex, so it works while a `continue` is blocked."
        }
        "save" => {
            "\
dbg save [<label> | --label <label>]
  Copy the live session DB to .dbg/sessions/<label>.db so it
  survives `dbg kill` and can be reopened with `dbg replay`.
  Without a label, uses the session's auto-label (see `dbg status`).
  Marks the persisted DB as user-owned so `dbg prune` won't reap it."
        }
        "cancel" => {
            "\
dbg cancel
  Interrupt the currently-running debugger command without tearing
  down the session (sends SIGINT to the debuggee). Use to break out
  of a `continue` that hasn't hit a breakpoint."
        }
        "finalize" => {
            "\
dbg finalize
  Stop a profile/sample collector cleanly so its trace is flushed and
  the daemon can move into query mode. Walks the daemon's child tree
  (daemon → bash → collector) and SIGINTs the collector (dotnet-trace,
  perf, callgrind, ...). Works while the daemon is still blocked in
  init waiting for the target to exit — `dbg cancel` cannot, because
  the accept loop hasn't started yet.

  Typical flow for a long-running server profile:
    1. dbg start dotnet-trace ./Server.dll
    2. drive load against it
    3. dbg finalize          (collector flushes, daemon proceeds)
    4. dbg top / dbg callers ...
    5. dbg kill"
        }
        _ => return None,
    })
}

/// Format the reader's event log for `dbg events`. Supports:
///   * `events`                     — full retained log
///   * `events --since=<N>`         — only entries with seq > N
///   * `events --tail=<N>`          — only last N entries
///   * `events --kind=stop[,output]` — filter by event kind
///   * `events --wait=<ms>`         — block up to this many ms for
///                                     new entries matching the filter
///                                     (capped at 50s).
///
/// Combined `--kind=stop --wait=N --since=M` means "wake me on the
/// next breakpoint hit past seq M, or timeout after N ms."
///
/// Agents use this to tail a session's PTY timeline live. The log is
/// populated by the reader thread + daemon (Stop events), so it updates
/// even while a command is blocked — `handle_events` never touches the
/// session mutex, so live-tailing is concurrent with a blocked
/// `continue`.
fn handle_events(cmd: &str, log_handle: &crate::pty::LogHandle) -> String {
    let mut since: u64 = 0;
    let mut tail: Option<usize> = None;
    let mut wait_ms: Option<u64> = None;
    let mut kinds: Option<Vec<crate::pty::EventKind>> = None;
    for tok in cmd.split_whitespace().skip(1) {
        if let Some(v) = tok.strip_prefix("--since=") {
            since = v.parse().unwrap_or(0);
        } else if let Some(v) = tok.strip_prefix("--tail=") {
            tail = v.parse().ok();
        } else if let Some(v) = tok.strip_prefix("--wait=") {
            wait_ms = v.parse().ok();
        } else if let Some(v) = tok.strip_prefix("--kind=") {
            let mut parsed = Vec::new();
            for name in v.split(',') {
                match crate::pty::EventKind::parse(name) {
                    Some(k) => parsed.push(k),
                    None => {
                        return format!(
                            "unknown event kind '{name}' in --kind={v} (valid: stdout, output, prompt, exit, stop)"
                        );
                    }
                }
            }
            if !parsed.is_empty() {
                kinds = Some(parsed);
            }
        }
    }

    let matches_filter = |kind: crate::pty::EventKind| -> bool {
        match &kinds {
            None => true,
            Some(ks) => ks.contains(&kind),
        }
    };

    // Snapshot + filter.
    let mut entries: Vec<_> = log_handle
        .since(since)
        .into_iter()
        .filter(|e| matches_filter(e.kind))
        .collect();

    // If no matching entries and --wait is set, loop-wait on the
    // condvar until a matching event arrives or the deadline expires.
    // Plain wait (no filter) wakes on any new event — fine; wake-check
    // is cheap. With a filter, unrelated events can wake the wait, so
    // we re-check and re-wait against the remaining deadline.
    if entries.is_empty() {
        if let Some(ms) = wait_ms {
            let capped = ms.min(50_000);
            let deadline = std::time::Instant::now() + Duration::from_millis(capped);
            let mut cursor = since;
            loop {
                let remaining = deadline.saturating_duration_since(std::time::Instant::now());
                if remaining.is_zero() {
                    break;
                }
                let woke = log_handle.since_wait(cursor, remaining);
                if woke.is_empty() {
                    break;
                }
                cursor = woke.last().map(|e| e.seq).unwrap_or(cursor);
                entries.extend(woke.into_iter().filter(|e| matches_filter(e.kind)));
                if !entries.is_empty() {
                    break;
                }
            }
        }
    }
    let last_seq = log_handle.last_seq();

    if let Some(n) = tail {
        if entries.len() > n {
            entries.drain(..entries.len() - n);
        }
    }

    if entries.is_empty() {
        return format!("no events (last seq={last_seq})");
    }

    let mut out = String::new();
    for e in &entries {
        let kind = e.kind.as_str();
        let ts = format_ms(e.ts_ms);
        match e.kind {
            crate::pty::EventKind::Output | crate::pty::EventKind::Stdout => {
                let preview = output_preview(&e.bytes);
                let nbytes = e.bytes.len();
                out.push_str(&format!(
                    "#{seq:<6} {ts:>9}  {kind:<7} {nbytes:>5}B  {preview}\n",
                    seq = e.seq,
                    ts = ts,
                    kind = kind,
                    nbytes = nbytes,
                    preview = preview,
                ));
            }
            crate::pty::EventKind::Prompt | crate::pty::EventKind::Exit => {
                out.push_str(&format!(
                    "#{seq:<6} {ts:>9}  {kind}\n",
                    seq = e.seq,
                    ts = ts,
                    kind = kind,
                ));
            }
            crate::pty::EventKind::Stop => {
                out.push_str(&format!(
                    "#{seq:<6} {ts:>9}  {kind:<7}   {payload}\n",
                    seq = e.seq,
                    ts = ts,
                    kind = kind,
                    payload = format_stop_payload(&e.bytes),
                ));
            }
        }
    }
    out.push_str(&format!(
        "({n} entries, last seq={last_seq})\n",
        n = entries.len(),
        last_seq = last_seq,
    ));
    out
}

fn format_ms(ms: u64) -> String {
    if ms < 1000 {
        format!("+{ms}ms")
    } else {
        format!("+{:.2}s", ms as f64 / 1000.0)
    }
}

/// Format a Stop event's JSON payload as a readable one-line summary.
/// Emitted by `capture_hit_if_stopped`; fields are location_key,
/// hit_seq, thread, file, line, frame_symbol.
fn format_stop_payload(bytes: &[u8]) -> String {
    let s = String::from_utf8_lossy(bytes);
    let v: serde_json::Value = match serde_json::from_str(&s) {
        Ok(v) => v,
        Err(_) => return format!("(unparseable stop: {s})"),
    };
    let loc = v
        .get("location_key")
        .and_then(|x| x.as_str())
        .unwrap_or("?");
    let seq = v.get("hit_seq").and_then(|x| x.as_i64()).unwrap_or(0);
    let frame = v
        .get("frame_symbol")
        .and_then(|x| x.as_str())
        .map(|s| format!(" in {s}"))
        .unwrap_or_default();
    let thread = v
        .get("thread")
        .and_then(|x| x.as_str())
        .map(|s| format!(" thread={s}"))
        .unwrap_or_default();
    format!("{loc} #{seq}{frame}{thread}")
}

/// One-line preview of an Output event. Strips ANSI, collapses
/// whitespace, truncates to 80 chars. Used by `dbg events`.
fn output_preview(bytes: &[u8]) -> String {
    let s = String::from_utf8_lossy(bytes);
    let cleaned = s.replace(['\r', '\n'], "⏎").replace('\t', " ");
    let single: String = cleaned.split_whitespace().collect::<Vec<_>>().join(" ");
    if single.chars().count() > 80 {
        let truncated: String = single.chars().take(77).collect();
        format!("{truncated}…")
    } else {
        single
    }
}

/// Drain any events that arrived on the reader channel since the last
/// command. For async debuggers (node-inspect) the stop banner for a
/// prior `continue` can arrive *after* the ack prompt, while the daemon
/// is idle. The reader thread captures those bytes regardless; this
/// helper pulls them from the channel and feeds them through
/// `capture_hit_if_stopped` before the next command runs. Non-blocking.
/// After the debuggee exits, live inspection verbs have no live
/// debugger to ask. For stack/locals we serve the last captured hit
/// from the session DB so the agent can still look at the program's
/// final observable state. Execution verbs get a clear "you're
/// post-mortem" message pointing at `dbg hits` / `dbg cross`.
fn post_mortem_fallback(session: &Session, canonical_op: &str) -> String {
    let db = match session.db.as_ref() {
        Some(d) => d,
        None => {
            return "(debuggee has exited and no session DB is available — \
                    start a fresh session with `dbg start`)"
                .to_string();
        }
    };
    match canonical_op {
        "locals" => {
            let row: Option<(i64, String, Option<String>)> = db
                .conn()
                .query_row(
                    "SELECT hit_seq, location_key, locals_json
                     FROM breakpoint_hits
                     WHERE locals_json IS NOT NULL
                     ORDER BY id DESC LIMIT 1",
                    [],
                    |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
                )
                .ok();
            match row {
                Some((seq, loc, Some(json))) => format!(
                    "[post-mortem] locals at last captured hit {loc} #{seq}\n{json}\n\
                     (debuggee has exited; use `dbg hits <loc>` to browse all hits)"
                ),
                _ => "[post-mortem] debuggee has exited and no locals were \
                      captured — run a new session with `dbg start`"
                    .to_string(),
            }
        }
        "stack" | "bt" | "backtrace" => {
            let row: Option<(i64, String, Option<String>)> = db
                .conn()
                .query_row(
                    "SELECT hit_seq, location_key, stack_json
                     FROM breakpoint_hits
                     WHERE stack_json IS NOT NULL
                     ORDER BY id DESC LIMIT 1",
                    [],
                    |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
                )
                .ok();
            match row {
                Some((seq, loc, Some(json))) => {
                    format!("[post-mortem] stack at last captured hit {loc} #{seq}\n{json}")
                }
                _ => "[post-mortem] debuggee has exited and no stack was \
                      captured — run a new session with `dbg start`"
                    .to_string(),
            }
        }
        "run" | "continue" | "step" | "next" | "finish" | "restart" | "pause" => {
            "debuggee has exited — cannot resume. Use `dbg hits <loc>`, \
             `dbg stack`, `dbg locals`, `dbg cross <sym>` to inspect \
             captured state, or `dbg start` for a fresh run."
                .to_string()
        }
        _ => "debuggee has exited — live inspection unavailable. Use \
              `dbg hits`, `dbg cross`, `dbg replay <label>` against the \
              captured DB, or `dbg start` for a new session."
            .to_string(),
    }
}

fn drain_pending_events(session: &mut Session, backend: &dyn Backend) {
    if let Some(pending) = session.proc.drain_pending() {
        if !pending.trim().is_empty() {
            capture_hit_if_stopped(session, backend, &pending);
        }
    }
}

/// Parse `output` through the active backend's `CanonicalOps::parse_hit`
/// and, on a match, synthesize `op_locals` + `op_stack` roundtrips so
/// we can attach a rich hit record to the SessionDb without asking the
/// agent to run extra commands.
///
/// With PTY echo disabled, each `send_and_wait` returns exactly the
/// debugger's response for that command.  Stop banners always appear
/// in the execution command's response — no deferred/merged banners.
fn capture_hit_if_stopped(session: &mut Session, backend: &dyn Backend, output: &str) {
    let ops = match backend.canonical_ops() {
        Some(o) => o,
        None => return,
    };
    // Structured path first: protocol transports (V8 Inspector, DAP)
    // deliver paused events with frame data inline; we skip text
    // banner parsing entirely. Falls back to `parse_hit` for PTY
    // transports that only surface stop info as debugger output.
    let hit = match session.proc.pending_hit() {
        Some(h) => h,
        None => match ops.parse_hit(output) {
            Some(h) => h,
            None => return,
        },
    };

    let seq = session.hit_seq.entry(hit.location_key.clone()).or_insert(0);
    *seq += 1;
    let hit_seq = *seq as i64;

    // Emit a structured Stop event on the timeline so `dbg events
    // --kind=stop` can tail program stops. Lean payload — callers who
    // need locals/stack query the session DB by location_key + hit_seq.
    let stop_payload = serde_json::json!({
        "location_key": hit.location_key,
        "hit_seq": hit_seq,
        "thread": hit.thread,
        "file": hit.file,
        "line": hit.line,
        "frame_symbol": hit.frame_symbol,
    })
    .to_string();
    session
        .proc
        .log()
        .push(crate::pty::EventKind::Stop, stop_payload.into_bytes());

    let db = match session.db.as_ref() {
        Some(d) => d,
        None => return,
    };

    // Auto-capture locals + stack when the backend says it's safe.
    // Fragile backends (jdb, ghci, netcoredbg CLI) set
    // `auto_capture_locals = false` because the synthesized roundtrip
    // can destroy their PTY state. Agents use `dbg locals` explicitly.
    let (locals_json, stack_json) = if ops.auto_capture_locals() {
        const CAPTURE_TIMEOUT: Duration = Duration::from_secs(1);
        let lj = ops.op_locals().ok().and_then(|op| {
            match session.proc.send_and_wait(&op, CAPTURE_TIMEOUT) {
                Ok(raw) => ops.parse_locals(&raw).map(|v| v.to_string()),
                Err(_) => None,
            }
        });
        let sj = ops.op_stack(Some(20)).ok().and_then(|op| {
            match session.proc.send_and_wait(&op, CAPTURE_TIMEOUT) {
                Ok(raw) => Some(
                    serde_json::json!({
                        "raw": raw,
                        "frame_symbol": hit.frame_symbol,
                    })
                    .to_string(),
                ),
                Err(_) => None,
            }
        });
        (lj, sj)
    } else {
        let sj = hit
            .frame_symbol
            .as_ref()
            .map(|s| serde_json::json!({ "frame_symbol": s }).to_string());
        (None, sj)
    };

    let _ = db.conn().execute(
        "INSERT INTO breakpoint_hits
            (session_id, location_key, hit_seq, thread, ts, locals_json, stack_json)
         VALUES (
             (SELECT id FROM sessions LIMIT 1),
             ?1, ?2, ?3, datetime('now'), ?4, ?5
         )",
        params![
            hit.location_key,
            hit_seq,
            hit.thread,
            locals_json,
            stack_json
        ],
    );
}

/// Execute a cross-track query against the session's DB, threading
/// a LiveDebugger handle so collectors can reuse the existing PTY
/// when they want to. Returns a formatted report ready for the agent.
fn run_crosstrack(session: &mut Session, backend: &dyn Backend, q: &crosstrack::Query) -> String {
    let db = match session.db.as_ref() {
        Some(d) => d,
        None => return "[session DB unavailable — crosstrack queries need capture]".into(),
    };
    let tool_name = backend
        .canonical_ops()
        .map(|c| c.tool_name())
        .unwrap_or("unknown");
    let live = ProcLive {
        proc: &*session.proc,
        tool_name,
    };
    let ctx = crosstrack::RunCtx {
        target: &session.target,
        target_class: session.target_class,
        cwd: &session.cwd,
        live: Some(&live),
    };
    crosstrack::run(q, db, &ctx)
}

/// Discard any pending PTY output so the next `send_and_wait` starts
/// from a clean buffer. Called after a timed-out synthesized roundtrip
/// to prevent stale data from poisoning the next agent command.
/// When the agent explicitly runs `dbg locals`, retroactively fill in
/// the most recent breakpoint_hits row's `locals_json` if it was NULL.
/// This enables `dbg hit-trend <loc> <field>` sparklines for backends
/// where `auto_capture_locals=false`: the agent's natural workflow
/// (`continue` → `locals` → repeat) populates the data progressively.
fn backfill_locals(session: &mut Session, backend: &dyn Backend, raw_output: &str) {
    let ops = match backend.canonical_ops() {
        Some(o) => o,
        None => return,
    };
    let locals_json = match ops.parse_locals(raw_output) {
        Some(v) => v.to_string(),
        None => return,
    };
    let db = match session.db.as_ref() {
        Some(d) => d,
        None => return,
    };
    // Update the most recent hit that has NULL locals_json.
    let _ = db.conn().execute(
        "UPDATE breakpoint_hits
         SET locals_json = ?1
         WHERE id = (
             SELECT id FROM breakpoint_hits
             WHERE locals_json IS NULL
             ORDER BY id DESC LIMIT 1
         )",
        params![locals_json],
    );
}

/// Back up the session DB to `.dbg/sessions/<label>.db` iff the
/// session is still marked auto and has captured any data. If the
/// session is empty we leave nothing behind — the tmpdir is wiped by
/// `cleanup_and_exit`.
/// Persist all session evidence before the daemon removes its private temp
/// directory. Returns false when any durable write failed; callers must then
/// keep that directory so the only remaining evidence is recoverable.
fn persist_session_on_exit(session: &mut Session) -> bool {
    persist_session_on_exit_with(session, &|path| {
        Ok(Box::new(std::fs::read_dir(path)?)
            as Box<
                dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>,
            >)
    })
}

fn persist_session_on_exit_with(
    session: &mut Session,
    read_dir: &dyn Fn(
        &Path,
    ) -> std::io::Result<
        Box<dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>>,
    >,
) -> bool {
    let db = match session.db.as_ref() {
        Some(d) => d,
        None => return false,
    };
    let has_data = match db.has_captured_data() {
        Ok(has_data) => has_data,
        Err(e) => {
            // An inspection failure is not evidence that the session is
            // empty. Keep the private directory because it may contain the
            // only copy of a native capture.
            eprintln!("[dbg] warning: failed to inspect captured session data: {e}");
            return false;
        }
    };
    let has_raw_evidence = match raw_evidence_present_with(&session.tmp_dir, read_dir) {
        Ok(present) => present,
        Err(e) => {
            // An inspection failure is not evidence that the session is
            // empty. Keep the private directory and the live debugger.
            eprintln!("[dbg] warning: failed to inspect raw session evidence: {e}");
            return false;
        }
    };
    if !has_data && !has_raw_evidence {
        return true;
    }
    let Some(path) = session.save_to.as_ref() else {
        eprintln!("[dbg] warning: session has captured data but no safe save destination");
        return false;
    };
    if let Err(e) = persist_raw_evidence(db, &session.cwd, db.label(), &session.tmp_dir) {
        eprintln!("[dbg] warning: failed to preserve raw session evidence: {e}");
        return false;
    }
    if let Err(e) = db.save_to(path) {
        eprintln!(
            "[dbg] warning: failed to save session DB to {}: {e}",
            path.display()
        );
        return false;
    }
    true
}

#[cfg(test)]
fn raw_evidence_present(root: &Path) -> std::io::Result<bool> {
    raw_evidence_present_with(root, &|path| {
        Ok(Box::new(std::fs::read_dir(path)?)
            as Box<
                dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>,
            >)
    })
}

fn raw_evidence_present_with(
    root: &Path,
    read_dir: &dyn Fn(
        &Path,
    ) -> std::io::Result<
        Box<dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>>,
    >,
) -> std::io::Result<bool> {
    raw_evidence_present_inner(root, read_dir, true)
}

fn raw_evidence_present_inner(
    root: &Path,
    read_dir: &dyn Fn(
        &Path,
    ) -> std::io::Result<
        Box<dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>>,
    >,
    skip_session_db: bool,
) -> std::io::Result<bool> {
    let metadata = match std::fs::symlink_metadata(root) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(error) => return Err(error),
    };
    if metadata.file_type().is_symlink() || !metadata.is_dir() {
        return Ok(true);
    }
    let entries = read_dir(root)?;
    let mut present = false;
    for entry in entries {
        let entry = entry?;
        let path = entry.path();
        let metadata = std::fs::symlink_metadata(&path)?;
        let is_session_db = entry.file_name() == "session.db";
        match metadata {
            // A rejected symlink is still evidence. Treating it as empty
            // would let shutdown delete the only copy after persistence
            // correctly refused to follow the link.
            meta if meta.file_type().is_symlink() => present = true,
            meta if skip_session_db && is_session_db && meta.is_file() => {}
            meta if meta.is_dir() => {
                present |= raw_evidence_present_inner(&path, read_dir, false)?;
            }
            _ => present = true,
        }
    }
    Ok(present)
}

fn session_save_path(cwd: &Path, label: &str) -> Option<PathBuf> {
    if label.is_empty()
        || label == "."
        || label == ".."
        || label.contains('/')
        || label.contains('\\')
    {
        eprintln!("[dbg] warning: refusing unsafe session label `{label}`");
        return None;
    }
    let dir = session_db::sessions_dir(cwd);
    if let Err(e) = std::fs::create_dir_all(&dir) {
        eprintln!("[dbg] warning: cannot create {}: {e}", dir.display());
        return None;
    }
    let root = match std::fs::canonicalize(&dir) {
        Ok(p) => p,
        Err(e) => {
            eprintln!("[dbg] warning: cannot resolve {}: {e}", dir.display());
            return None;
        }
    };
    let path = root.join(format!("{label}.db"));
    if std::fs::symlink_metadata(&path).is_ok_and(|m| m.file_type().is_symlink()) {
        eprintln!(
            "[dbg] warning: refusing symbolic-link session destination {}",
            path.display()
        );
        return None;
    }
    Some(path)
}

/// Copy raw profiler/debugger artifacts beside the durable SQLite session.
/// SQLite metadata alone cannot regenerate native captures such as perf.data
/// or .nettrace files, so cleanup must happen only after this copy succeeds.
fn persist_raw_evidence(db: &SessionDb, cwd: &Path, label: &str, source: &Path) -> Result<()> {
    if label.is_empty()
        || label.contains('/')
        || label.contains('\\')
        || label == "."
        || label == ".."
    {
        anyhow::bail!("unsafe session label {label:?}");
    }
    let destination = session_db::raw_dir(cwd, label);
    copy_raw_tree(&source, &destination)?;

    // Saved databases must refer to the durable copy, not the daemon's
    // private directory that cleanup removes after shutdown.
    let mut stmt = db
        .conn()
        .prepare("SELECT id, file FROM layers WHERE file IS NOT NULL")?;
    let rows: Vec<(i64, String)> = stmt
        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
        .collect::<rusqlite::Result<_>>()?;
    for (id, file) in rows {
        if let Ok(relative) = Path::new(&file).strip_prefix(&source) {
            let durable = destination.join(relative);
            db.conn().execute(
                "UPDATE layers SET file = ?1 WHERE id = ?2",
                rusqlite::params![durable.display().to_string(), id],
            )?;
        }
    }
    Ok(())
}

fn copy_raw_tree(src: &Path, dst: &Path) -> std::io::Result<()> {
    #[cfg(unix)]
    {
        return copy_raw_tree_unix(src, dst);
    }

    #[cfg(not(unix))]
    copy_raw_tree_portable(src, dst)
}

#[cfg(not(unix))]
fn copy_raw_tree_portable(src: &Path, dst: &Path) -> std::io::Result<()> {
    copy_raw_tree_portable_inner(src, dst, true)
}

#[cfg(not(unix))]
fn copy_raw_tree_portable_inner(
    src: &Path,
    dst: &Path,
    skip_session_db: bool,
) -> std::io::Result<()> {
    let src_meta = std::fs::symlink_metadata(src)?;
    if src_meta.file_type().is_symlink() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            format!("raw evidence source is a symbolic link: {}", src.display()),
        ));
    }
    if !src_meta.is_dir() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::NotADirectory,
            format!("raw evidence source is not a directory: {}", src.display()),
        ));
    }

    match std::fs::symlink_metadata(dst) {
        Ok(meta) if meta.file_type().is_symlink() => {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!(
                    "raw evidence destination is a symbolic link: {}",
                    dst.display()
                ),
            ));
        }
        Ok(meta) if !meta.is_dir() => {
            return Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                format!(
                    "raw evidence destination is not a directory: {}",
                    dst.display()
                ),
            ));
        }
        Ok(_) => {}
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => std::fs::create_dir_all(dst)?,
        Err(e) => return Err(e),
    }

    for entry in std::fs::read_dir(src)? {
        let entry = entry?;
        let from = entry.path();
        let to = dst.join(entry.file_name());
        let meta = std::fs::symlink_metadata(&from)?;
        if meta.file_type().is_symlink() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!("raw evidence contains a symbolic link: {}", from.display()),
            ));
        }
        if skip_session_db
            && from.file_name().and_then(|n| n.to_str()) == Some("session.db")
            && meta.is_file()
        {
            continue;
        }
        if meta.is_dir() {
            copy_raw_tree_portable_inner(&from, &to, false)?;
        } else if meta.is_file() {
            if std::fs::symlink_metadata(&to)
                .is_ok_and(|existing| existing.file_type().is_symlink())
            {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    format!(
                        "raw evidence destination is a symbolic link: {}",
                        to.display()
                    ),
                ));
            }
            std::fs::copy(&from, &to)?;
        } else {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                format!(
                    "raw evidence contains an unsupported file: {}",
                    from.display()
                ),
            ));
        }
    }
    Ok(())
}

#[cfg(unix)]
fn copy_raw_tree_unix(src: &Path, dst: &Path) -> std::io::Result<()> {
    use std::ffi::CStr;
    use std::os::fd::{AsRawFd, FromRawFd};
    use std::os::unix::ffi::OsStrExt;

    let source = open_source_directory(src, None)?;
    let destination = open_destination_directory(dst)?;
    copy_raw_directory_fd(source.as_raw_fd(), src, destination.as_raw_fd(), dst, true)?;
    // Keep the source directory fd alive until all children have been
    // opened. This makes each child lookup relative to the directory inode,
    // not to a pathname which an attacker can replace.
    drop(source);

    fn open_source_directory(path: &Path, parent_fd: Option<i32>) -> std::io::Result<File> {
        let raw_path = path.as_os_str().as_bytes();
        let c_path = std::ffi::CString::new(raw_path).map_err(|_| {
            std::io::Error::new(std::io::ErrorKind::InvalidInput, "source path contains NUL")
        })?;
        let fd = unsafe {
            match parent_fd {
                Some(parent) => nix::libc::openat(
                    parent,
                    c_path.as_ptr(),
                    nix::libc::O_RDONLY
                        | nix::libc::O_DIRECTORY
                        | nix::libc::O_NOFOLLOW
                        | nix::libc::O_CLOEXEC,
                    0,
                ),
                None => nix::libc::open(
                    c_path.as_ptr(),
                    nix::libc::O_RDONLY
                        | nix::libc::O_DIRECTORY
                        | nix::libc::O_NOFOLLOW
                        | nix::libc::O_CLOEXEC,
                ),
            }
        };
        if fd < 0 {
            let error = std::io::Error::last_os_error();
            let source_is_symlink = std::fs::symlink_metadata(path)
                .is_ok_and(|metadata| metadata.file_type().is_symlink());
            if error.raw_os_error() == Some(nix::libc::ELOOP) || source_is_symlink {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    format!("raw evidence source is a symbolic link: {}", path.display()),
                ));
            }
            return Err(error);
        }
        // SAFETY: open/openat returned a unique owned descriptor.
        Ok(unsafe { File::from_raw_fd(fd) })
    }

    fn open_source_entry(parent_fd: i32, name: &std::ffi::OsStr) -> std::io::Result<File> {
        let c_name = std::ffi::CString::new(name.as_bytes()).map_err(|_| {
            std::io::Error::new(std::io::ErrorKind::InvalidInput, "source name contains NUL")
        })?;
        let fd = unsafe {
            nix::libc::openat(
                parent_fd,
                c_name.as_ptr(),
                nix::libc::O_RDONLY
                    | nix::libc::O_NONBLOCK
                    | nix::libc::O_NOFOLLOW
                    | nix::libc::O_CLOEXEC,
                0,
            )
        };
        if fd < 0 {
            let error = std::io::Error::last_os_error();
            if error.raw_os_error() == Some(nix::libc::ELOOP) {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    format!("raw evidence contains a symbolic link: {}", name.display()),
                ));
            }
            return Err(error);
        }
        // SAFETY: openat returned a unique owned descriptor.
        Ok(unsafe { File::from_raw_fd(fd) })
    }

    fn open_directory_at(parent_fd: i32, name: &std::ffi::OsStr) -> std::io::Result<File> {
        let c_name = std::ffi::CString::new(name.as_bytes()).map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "destination name contains NUL",
            )
        })?;
        let fd = unsafe {
            nix::libc::openat(
                parent_fd,
                c_name.as_ptr(),
                nix::libc::O_RDONLY
                    | nix::libc::O_DIRECTORY
                    | nix::libc::O_NOFOLLOW
                    | nix::libc::O_CLOEXEC,
                0,
            )
        };
        if fd < 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: openat returned a unique owned descriptor.
        Ok(unsafe { File::from_raw_fd(fd) })
    }

    fn open_or_create_directory_at(
        parent_fd: i32,
        name: &std::ffi::OsStr,
    ) -> std::io::Result<File> {
        match open_directory_at(parent_fd, name) {
            Ok(directory) => Ok(directory),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                let c_name = std::ffi::CString::new(name.as_bytes()).map_err(|_| {
                    std::io::Error::new(
                        std::io::ErrorKind::InvalidInput,
                        "destination name contains NUL",
                    )
                })?;
                let mkdir_result = unsafe { nix::libc::mkdirat(parent_fd, c_name.as_ptr(), 0o700) };
                if mkdir_result < 0 {
                    let mkdir_error = std::io::Error::last_os_error();
                    if mkdir_error.kind() != std::io::ErrorKind::AlreadyExists {
                        return Err(mkdir_error);
                    }
                }
                // Reopen with O_NOFOLLOW after mkdir. If a race installed a
                // symlink or a non-directory, this fails without following
                // it. The returned descriptor pins the directory inode.
                open_directory_at(parent_fd, name)
            }
            Err(error) => Err(error),
        }
    }

    fn open_destination_directory(path: &Path) -> std::io::Result<File> {
        let mut current = if path.is_absolute() {
            open_directory_at(nix::libc::AT_FDCWD, std::ffi::OsStr::new("/"))?
        } else {
            open_directory_at(nix::libc::AT_FDCWD, std::ffi::OsStr::new("."))?
        };
        for component in path.components() {
            use std::path::Component;
            let name = match component {
                Component::RootDir | Component::CurDir => continue,
                Component::Normal(name) => name,
                Component::ParentDir => std::ffi::OsStr::new(".."),
                Component::Prefix(_) => continue,
            };
            current = open_or_create_directory_at(current.as_raw_fd(), name)?;
        }
        Ok(current)
    }

    fn open_destination_file(parent_fd: i32, name: &std::ffi::OsStr) -> std::io::Result<File> {
        let c_name = std::ffi::CString::new(name.as_bytes()).map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "destination name contains NUL",
            )
        })?;
        let fd = unsafe {
            nix::libc::openat(
                parent_fd,
                c_name.as_ptr(),
                nix::libc::O_WRONLY
                    | nix::libc::O_CREAT
                    | nix::libc::O_TRUNC
                    | nix::libc::O_NOFOLLOW
                    | nix::libc::O_CLOEXEC,
                0o600,
            )
        };
        if fd < 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: openat returned a unique owned descriptor.
        Ok(unsafe { File::from_raw_fd(fd) })
    }

    fn visit_source_entries(
        parent_fd: i32,
        mut visit: impl FnMut(std::ffi::OsString) -> std::io::Result<()>,
    ) -> std::io::Result<()> {
        // Duplicate the descriptor because fdopendir takes ownership. This
        // enumerates the held source directory directly and does not require
        // /proc/self/fd to be mounted.
        let duplicate = unsafe { nix::libc::dup(parent_fd) };
        if duplicate < 0 {
            return Err(std::io::Error::last_os_error());
        }
        let directory = unsafe { nix::libc::fdopendir(duplicate) };
        if directory.is_null() {
            let error = std::io::Error::last_os_error();
            unsafe { nix::libc::close(duplicate) };
            return Err(error);
        }
        struct Directory(*mut nix::libc::DIR);

        impl Drop for Directory {
            fn drop(&mut self) {
                if !self.0.is_null() {
                    unsafe { nix::libc::closedir(self.0) };
                }
            }
        }

        let directory = Directory(directory);
        loop {
            nix::errno::Errno::clear();
            let entry = unsafe { nix::libc::readdir(directory.0) };
            if entry.is_null() {
                let error = nix::errno::Errno::last_raw();
                if error != 0 {
                    return Err(std::io::Error::from_raw_os_error(error));
                }
                break;
            }
            let name = unsafe { CStr::from_ptr((*entry).d_name.as_ptr()) }.to_bytes();
            if name == b"." || name == b".." {
                continue;
            }
            visit(std::ffi::OsStr::from_bytes(name).to_os_string())?;
        }
        Ok(())
    }

    fn copy_raw_directory_fd(
        parent_fd: i32,
        source_path: &Path,
        destination_fd: i32,
        destination_path: &Path,
        skip_session_db: bool,
    ) -> std::io::Result<()> {
        visit_source_entries(parent_fd, |name| {
            let source_entry = open_source_entry(parent_fd, &name)?;
            let metadata = source_entry.metadata()?;
            if skip_session_db && name.to_str() == Some("session.db") && metadata.is_file() {
                // A regular session.db is the SQLite metadata already saved
                // separately. Inspect it through the no-follow descriptor,
                // but do not copy it as raw evidence.
                return Ok(());
            }
            let target = destination_path.join(&name);
            if metadata.is_dir() {
                let destination_entry = open_or_create_directory_at(destination_fd, &name)?;
                copy_raw_directory_fd(
                    source_entry.as_raw_fd(),
                    &source_path.join(&name),
                    destination_entry.as_raw_fd(),
                    &target,
                    false,
                )?;
            } else if metadata.is_file() {
                let mut output = open_destination_file(destination_fd, &name)?;
                let mut input = source_entry;
                std::io::copy(&mut input, &mut output)?;
            } else {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    format!(
                        "raw evidence contains an unsupported file: {}",
                        source_path.join(&name).display()
                    ),
                ));
            }
            Ok(())
        })
    }

    Ok(())
}

/// Send a command to a running daemon. Returns the response.
///
/// Retries the initial connect a few times over ~150ms. This masks a
/// short-lived race immediately after `dbg start ... --run` returns:
/// the socket is bound before the first in-flight command finishes,
/// but a follow-up client invocation issued microseconds later can
/// still see the listener in a transient not-accepting state (seen on
/// short-lived targets where the backend's program thread is racing
/// with our accept loop). Without retry the second call surfaces as
/// a misleading "no session running".
pub fn send_command(cmd: &str) -> Result<String> {
    send_command_for_slug(&session_slug(), cmd)
}

fn send_command_for_slug(slug: &str, cmd: &str) -> Result<String> {
    let mut stream = connect_with_retry_for_slug(slug)?;
    stream.set_read_timeout(Some(CMD_TIMEOUT))?;

    stream.write_all(format!("{cmd}\n").as_bytes())?;
    stream.shutdown(std::net::Shutdown::Write)?;

    let mut response = String::new();
    stream.read_to_string(&mut response)?;
    Ok(response)
}

fn connect_with_retry() -> Result<std::os::unix::net::UnixStream> {
    connect_with_retry_for_slug(&session_slug())
}

fn connect_with_retry_for_slug(slug: &str) -> Result<std::os::unix::net::UnixStream> {
    use std::os::unix::net::UnixStream;
    let mut attempts = 0;
    loop {
        match UnixStream::connect(socket_path_for(slug)) {
            Ok(s) => return Ok(s),
            Err(e) => {
                // Two failure modes to distinguish:
                //  (a) no daemon — pid file gone or names a
                //      dead/non-dbg pid. Fail fast with the start hint.
                //  (b) daemon alive but its socket path is gone. The
                //      bound `UnixListener` keeps serving the inode,
                //      so the daemon can't tell anything is wrong;
                //      the client can't reach it via path. There's
                //      no in-process recovery — `dbg kill` + restart
                //      is the only path back to a connectable
                //      socket.
                if !is_slug_live(slug) || attempts >= 5 {
                    if is_slug_live(slug) && !socket_path_for(slug).exists() {
                        let path = socket_path_for(slug);
                        return Err(e).context(format!(
                            "daemon is running but its socket file is gone \
                             ({}). Something unlinked the path while the \
                             daemon held its bound listener — recover with \
                             `dbg kill` then `dbg start ...`",
                            path.display()
                        ));
                    }
                    let hint = foreign_daemon_hint();
                    return Err(e).context(format!("no session running — use: dbg start{hint}"));
                }
                attempts += 1;
                std::thread::sleep(Duration::from_millis(30));
            }
        }
    }
}

/// When the resolved slug has no live daemon but *other* daemons are
/// reachable, surface them so the caller can pick one. Distinguishes
/// peers in the same cwd (matching `cwd_base_slug()` prefix — the user
/// likely wants one of these) from daemons in other cwds (same user,
/// different bucket — usually a shell that cd'd away).
fn foreign_daemon_hint() -> String {
    let peers = live_slugs_in_cwd();
    if !peers.is_empty() {
        let list = peers.join(", ");
        return format!("\n  (peers in this cwd: {list} — select with DBG_SESSION=<slug>)");
    }
    let dir = runtime_dir();
    let entries = match std::fs::read_dir(&dir) {
        Ok(e) => e,
        Err(_) => return String::new(),
    };
    let other_count = entries
        .flatten()
        .filter(|e| e.path().extension().and_then(|s| s.to_str()) == Some("sock"))
        .count();
    if other_count == 0 {
        String::new()
    } else {
        format!(
            "\n  (note: {other_count} daemon socket(s) exist in {} for other cwds \
             — `dbg` is cwd-scoped; run from the same directory where you did `dbg start`)",
            dir.display(),
        )
    }
}

/// Return a process identity that remains stable while a PID is alive.
/// The kernel start time is the important part: a recycled PID has a
/// different value even when the replacement executable has the same name.
fn process_identity(pid: i32) -> Option<String> {
    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
    let after_comm = stat.rsplit_once(") ")?.1;
    let fields: Vec<&str> = after_comm.split_whitespace().collect();
    let start_time = fields.get(19)?; // field 22, after pid + comm
    let exe = std::fs::read_link(format!("/proc/{pid}/exe"))
        .ok()
        .map(|p| p.to_string_lossy().into_owned())
        .unwrap_or_default();
    Some(format!("{start_time}\n{exe}"))
}

fn daemon_pid_record(pid: i32) -> String {
    match process_identity(pid) {
        Some(identity) => format!("{pid}\n{identity}\n"),
        None => format!("{pid}\n"),
    }
}

fn recorded_process_identity(pid_file: &str) -> Option<String> {
    let mut lines = pid_file.lines();
    let _pid = lines.next()?;
    let start = lines.next()?.trim();
    let exe = lines.next().unwrap_or("").trim();
    if start.is_empty() {
        None
    } else {
        Some(format!("{start}\n{exe}"))
    }
}

fn daemon_identity_matches(pid_file: &str, pid: i32) -> bool {
    daemon_identity_matches_with_procfs(pid_file, pid, procfs_available())
}

fn daemon_identity_matches_with_procfs(pid_file: &str, pid: i32, procfs_available: bool) -> bool {
    match recorded_process_identity(pid_file) {
        Some(recorded) => process_identity(pid).is_some_and(|current| current == recorded),
        // A PID-only marker is unsafe against PID reuse when procfs is
        // available. If procfs is unavailable, it is the only identity
        // published by older daemons and must still identify a live daemon.
        None => !procfs_available || pid == std::process::id() as i32,
    }
}

fn procfs_available() -> bool {
    std::fs::read_to_string("/proc/self/stat").is_ok()
}

/// List the direct child PIDs of `pid` via `/proc/<pid>/task/*/children`.
/// Linux-only — returns an empty vec on other platforms or when the
/// procfs entry is missing. Used by `dbg finalize` to walk from the
/// daemon down to the profile collector (daemon → bash → collector).
fn proc_children(pid: i32) -> Vec<i32> {
    let task_dir = format!("/proc/{pid}/task");
    let Ok(tasks) = std::fs::read_dir(&task_dir) else {
        return Vec::new();
    };
    let mut children = Vec::new();
    for task in tasks.flatten() {
        let path = task.path().join("children");
        let Ok(content) = std::fs::read_to_string(path) else {
            continue;
        };
        children.extend(parse_proc_children(&content));
    }
    children.sort_unstable();
    children.dedup();
    children
}

fn parse_proc_children(content: &str) -> Vec<i32> {
    content
        .split_whitespace()
        .filter_map(|t| t.parse().ok())
        .collect()
}

#[cfg(target_os = "linux")]
#[derive(Debug)]
struct IdentitySafePidFd {
    pid: i32,
    identity: String,
    fd: std::os::fd::OwnedFd,
}

/// Open a handle to one process instance and verify the instance after the
/// open. If the PID changes after this function returns, the pidfd still
/// refers to the original process and cannot address the replacement.
#[cfg(target_os = "linux")]
fn open_identity_safe_pidfd(pid: i32, expected_identity: &str) -> Result<IdentitySafePidFd> {
    use std::os::fd::{FromRawFd, OwnedFd};

    if expected_identity.is_empty() {
        bail!("process {pid} has no verified start identity")
    }

    // SAFETY: this is the Linux pidfd_open(2) syscall. On kernels without
    // pidfd support it returns an error; never fall back to kill(pid, ...).
    let raw_fd = unsafe { nix::libc::syscall(nix::libc::SYS_pidfd_open, pid, 0) } as i32;
    if raw_fd < 0 {
        return Err(std::io::Error::last_os_error())
            .with_context(|| format!("cannot obtain identity-safe handle for PID {pid}"));
    }
    // SAFETY: a successful pidfd_open returns a new owned file descriptor.
    let fd = unsafe { OwnedFd::from_raw_fd(raw_fd) };
    let Some(identity) = process_identity(pid) else {
        bail!("cannot verify start identity for PID {pid} after opening its pidfd")
    };
    if identity != expected_identity {
        bail!("PID {pid} changed process instance while opening its pidfd; refusing to signal it")
    }
    Ok(IdentitySafePidFd { pid, identity, fd })
}

#[cfg(target_os = "linux")]
fn open_current_identity_safe_pidfd(pid: i32) -> Result<IdentitySafePidFd> {
    let expected = process_identity(pid)
        .ok_or_else(|| anyhow::anyhow!("cannot obtain start identity for PID {pid}"))?;
    open_identity_safe_pidfd(pid, &expected)
}

#[cfg(target_os = "linux")]
fn pidfd_send_sigint(handle: &IdentitySafePidFd) -> std::io::Result<()> {
    pidfd_send_signal(handle, nix::libc::SIGINT)
}

#[cfg(target_os = "linux")]
fn pidfd_send_signal(handle: &IdentitySafePidFd, signal: i32) -> std::io::Result<()> {
    use std::os::fd::AsRawFd;

    // SAFETY: pidfd_send_signal addresses the process instance bound to the
    // fd, never a later occupant of handle.pid.
    let rc = unsafe {
        nix::libc::syscall(
            nix::libc::SYS_pidfd_send_signal,
            handle.fd.as_raw_fd(),
            signal,
            std::ptr::null::<nix::libc::siginfo_t>(),
            0,
        )
    };
    if rc == 0 {
        Ok(())
    } else {
        Err(std::io::Error::last_os_error())
    }
}

fn signal_child_identity(
    pid: i32,
    expected_identity: Option<&str>,
    signal: i32,
) -> anyhow::Result<()> {
    #[cfg(target_os = "linux")]
    {
        let identity = expected_identity
            .ok_or_else(|| anyhow::anyhow!("child PID {pid} has no verified identity"))?;
        let handle = open_identity_safe_pidfd(pid, identity)?;
        pidfd_send_signal(&handle, signal)?;
        return Ok(());
    }
    #[cfg(not(target_os = "linux"))]
    {
        if let Some(identity) = expected_identity {
            if process_identity(pid).as_deref() != Some(identity) {
                anyhow::bail!("child PID {pid} no longer names the original process");
            }
        }
        nix::sys::signal::kill(
            nix::unistd::Pid::from_raw(pid),
            nix::sys::signal::Signal::try_from(signal)
                .map_err(|_| anyhow::anyhow!("invalid signal {signal}"))?,
        )?;
        Ok(())
    }
}

#[cfg(target_os = "linux")]
fn process_parent_pid(pid: i32) -> Option<i32> {
    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
    let after_comm = stat.rsplit_once(") ")?.1;
    let fields: Vec<&str> = after_comm.split_whitespace().collect();
    fields.get(1)?.parse().ok() // field 4: state, then parent PID
}

#[cfg(target_os = "linux")]
fn process_is_child_of(pid: i32, parent: i32) -> bool {
    let Some(actual_parent) = process_parent_pid(pid) else {
        return false;
    };
    if actual_parent == parent {
        return true;
    }
    // A child forked by a non-leader thread has that thread's TID as its
    // parent in /proc. It is still a child of the daemon process here.
    std::fs::read_dir(format!("/proc/{parent}/task"))
        .ok()
        .into_iter()
        .flatten()
        .filter_map(|entry| entry.ok()?.file_name().to_str()?.parse::<i32>().ok())
        .any(|tid| tid == actual_parent)
}

#[cfg(target_os = "linux")]
fn open_verified_child(parent: &IdentitySafePidFd, pid: i32) -> Result<IdentitySafePidFd> {
    let child = open_current_identity_safe_pidfd(pid)
        .with_context(|| format!("cannot bind child PID {pid} to an identity-safe handle"))?;
    // Check the process-tree boundary while both instance-bound handles are
    // held. A replacement or a reparented process is not a descendant.
    if !process_is_child_of(pid, parent.pid)
        || process_identity(parent.pid).as_deref() != Some(parent.identity.as_str())
    {
        anyhow::bail!(
            "PID {pid} is no longer a verified child of process instance {}",
            parent.pid
        );
    }
    Ok(child)
}

#[cfg(target_os = "linux")]
fn open_verified_children_from_pids(
    parent: &IdentitySafePidFd,
    pids: Vec<i32>,
) -> Result<(Vec<IdentitySafePidFd>, bool)> {
    let had_children = !pids.is_empty();
    let children = pids
        .into_iter()
        .map(|pid| open_verified_child(parent, pid))
        .collect::<Result<Vec<_>>>()?;
    Ok((children, had_children))
}

#[cfg(target_os = "linux")]
fn open_verified_children(parent: &IdentitySafePidFd) -> Result<(Vec<IdentitySafePidFd>, bool)> {
    open_verified_children_from_pids(parent, proc_children(parent.pid))
}

#[cfg(target_os = "linux")]
fn append_verified_descendants(
    root: IdentitySafePidFd,
    output: &mut Vec<IdentitySafePidFd>,
) -> Result<()> {
    let (children, _) = open_verified_children(&root)?;
    output.push(root);
    for child in children {
        append_verified_descendants(child, output)?;
    }
    Ok(())
}

#[cfg(target_os = "linux")]
fn collector_identity_safe_targets(daemon: &IdentitySafePidFd) -> Result<Vec<IdentitySafePidFd>> {
    let (direct, had_direct) = open_verified_children(daemon)?;
    if !had_direct {
        anyhow::bail!(
            "daemon {} has no child processes — has the collector already exited? Try `dbg status` to see the current state.",
            daemon.pid
        );
    }
    // The normal tree is daemon → shell → collector. Preserve the old
    // direct-child fallback for backends that do not use a shell, but only
    // with handles verified as this daemon's children.
    let mut collectors = Vec::new();
    let mut had_grandchildren = false;
    for child in &direct {
        let (grandchildren, had) = open_verified_children(child)?;
        had_grandchildren |= had;
        collectors.extend(grandchildren);
    }
    if !had_grandchildren {
        collectors = direct;
    }

    let mut targets = Vec::new();
    for collector in collectors {
        append_verified_descendants(collector, &mut targets)?;
    }
    if targets.is_empty() {
        anyhow::bail!("finalize: collector tree has no identity-safe signal targets");
    }
    Ok(targets)
}

#[cfg(test)]
fn collector_signal_targets<F>(daemon_pid: i32, mut children_of: F) -> Result<Vec<i32>>
where
    F: FnMut(i32) -> Vec<i32>,
{
    // Daemon → bash (depth 1) → collector (depth 2). The collector is
    // what we want to SIGINT; bash itself ignores SIGINT in interactive
    // mode. Fall back to depth-1 children if depth-2 is empty (covers
    // backends that don't go through bash, or races where the
    // collector hasn't spawned yet).
    let depth1 = children_of(daemon_pid);
    if depth1.is_empty() {
        anyhow::bail!(
            "daemon {daemon_pid} has no child processes — has the collector \
             already exited? Try `dbg status` to see the current state."
        );
    }
    let mut targets: Vec<i32> = depth1.iter().flat_map(|&p| children_of(p)).collect();
    if targets.is_empty() {
        targets = depth1;
    }
    targets.sort_unstable();
    targets.dedup();
    Ok(targets)
}

#[derive(Debug)]
pub struct FinalizeReport {
    pub signalled: Vec<i32>,
    pub failed: Vec<i32>,
}

/// Signal the profile/sample collector process tree for the active daemon so
/// blocking collectors can flush their trace and let daemon init complete.
pub fn finalize_collector() -> Result<FinalizeReport> {
    #[cfg(not(target_os = "linux"))]
    {
        // There is no portable process-instance-bound signal primitive here.
        // A bare PID or process-group fallback would make stale runtime data
        // dangerous, so unsupported platforms fail closed.
        anyhow::bail!("finalize: identity-safe process handles are unavailable on this platform");
    }

    #[cfg(target_os = "linux")]
    {
        let pid_file =
            std::fs::read_to_string(pid_path()).context("cannot read daemon identity marker")?;
        // Parse PID and identity from one marker snapshot. Reading the path
        // twice could combine an old identity with a replacement marker's PID.
        let daemon_pid = pid_file
            .lines()
            .next()
            .and_then(|line| line.trim().parse::<i32>().ok())
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "no daemon registered for this cwd — `dbg start` must be running. \
                     If `dbg start` is mid-init the pid file should exist; check \
                     $XDG_RUNTIME_DIR/dbg-$UID/ for stale state."
                )
            })?;
        let daemon_identity = recorded_process_identity(&pid_file).ok_or_else(|| {
            anyhow::anyhow!(
                "finalize: daemon identity marker has no verified process start identity; refusing to signal PID {daemon_pid}"
            )
        })?;

        // Open the daemon handle before inspecting descendants. This binds
        // the operation to this daemon instance if its PID is recycled.
        let daemon = open_identity_safe_pidfd(daemon_pid, &daemon_identity)?;
        let targets = collector_identity_safe_targets(&daemon)?;

        let mut signalled = Vec::new();
        let mut failed = Vec::new();
        for target in &targets {
            match pidfd_send_sigint(target) {
                Ok(()) => signalled.push(target.pid),
                Err(_) => failed.push(target.pid),
            }
        }

        if signalled.is_empty() {
            anyhow::bail!("finalize: failed to signal any identity-safe collector ({failed:?})");
        }

        return Ok(FinalizeReport { signalled, failed });
    }
}

pub fn is_running() -> bool {
    // Purely observational — never mutates filesystem state. A
    // previous version auto-deleted pid/socket files it judged
    // stale, but that raced with our own daemon's startup (pid is
    // written a few microseconds before `UnixListener::bind`; a
    // concurrent `dbg status` would delete the pid in the gap and
    // then every subsequent client saw "no session"). Orphan
    // cleanup belongs in `dbg start`/`dbg kill`, not here.
    let Ok(pid_str) = std::fs::read_to_string(&pid_path()) else {
        return false;
    };
    let Some(pid) = pid_str
        .lines()
        .next()
        .and_then(|s| s.trim().parse::<i32>().ok())
    else {
        return false;
    };
    let alive = nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_ok();
    if !alive {
        return false;
    }
    // PID recycling defense: the kernel may hand pid N to a
    // different program after our daemon dies, and a bare
    // `kill(pid, 0)` would happily green-light that stranger.
    // /proc/<pid>/comm names the binary backing the live pid, so
    // checking it equals "dbg" rejects the recycled-pid case
    // without depending on any file we wrote ourselves. We don't
    // gate on the socket file's presence: the daemon's
    // `UnixListener` keeps serving on its bound inode even after
    // the directory entry is unlinked, so a missing path is not
    // proof the daemon is gone — only that something removed the
    // path. Diagnosing that case (alive daemon, dead socket path)
    // is `connect_with_retry`'s job, not ours.
    daemon_identity_matches(&pid_str, pid) && proc_comm_is_dbg(pid)
}

fn proc_comm_is_dbg(pid: i32) -> bool {
    let Ok(comm) = std::fs::read_to_string(format!("/proc/{pid}/comm")) else {
        // No /proc on this platform (non-Linux). Fall back to
        // trusting raw pid liveness. PID-recycling within the
        // sub-second window between our daemon dying and a client
        // checking is rare enough to accept here.
        return true;
    };
    // Prefix match rather than equality: cargo's test binary is
    // named `dbg-<hash>` (and so is any future `dbg-foo` helper we
    // might ship). The kernel truncates /proc/<pid>/comm to 15
    // chars but the prefix is preserved. The only way this
    // false-positives is a recycled pid landing on a different
    // binary that happens to start with "dbg" — vastly less likely
    // than a bare `kill(pid, 0)` accepting any random process.
    comm.trim().starts_with("dbg")
}

/// Clean up stale pid/socket files from a dead or foreign daemon.
/// Safe to call before `dbg start` / `dbg replay` — those used to
/// rely on `is_running()` doing it as a side-effect, which caused
/// the race described above.
///
/// Preserve files only when the pid is live *and* still looks like a dbg
/// process. A recycled pid owned by an unrelated program is stale state and
/// should not pin this runtime slot forever.
pub fn clean_stale_runtime_files() {
    let slug = session_slug();
    let Some(reservation) = try_reserve_for_cleanup(&slug) else {
        // A starter owns this slug. It may publish the PID/socket between
        // any two pathname observations, so leave its files alone.
        return;
    };
    let pid_path = pid_path_for(&slug);
    let socket_path = socket_path_for(&slug);
    if let Ok(pid_str) = std::fs::read_to_string(&pid_path) {
        if let Some(pid) = pid_str
            .lines()
            .next()
            .and_then(|s| s.trim().parse::<i32>().ok())
        {
            if nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_ok()
                && daemon_identity_matches(&pid_str, pid)
                && proc_comm_is_dbg(pid)
            {
                drop(reservation);
                return;
            }
        }
    }
    let _ = std::fs::remove_file(pid_path);
    let _ = std::fs::remove_file(socket_path);
    drop(reservation);
}

fn try_reserve_for_cleanup(slug: &str) -> Option<nix::fcntl::Flock<File>> {
    let path = reservation_path_for(slug);
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .custom_flags(nix::libc::O_NOFOLLOW)
        .open(path)
        .ok()?;
    nix::fcntl::Flock::lock(file, nix::fcntl::FlockArg::LockExclusiveNonblock).ok()
}

/// Kill the running daemon. Blocks until the process is gone and
/// socket/pid files are cleared — callers that immediately spawn a
/// new daemon need this to be synchronous.
pub fn kill_daemon() -> Result<String> {
    // Resolve the target slug *once* up front. The daemon's own
    // cleanup clears the `.latest` pointer on graceful exit, so
    // subsequent re-evaluations of `session_slug()` would fall through
    // to the bare cwd slug — and nuke the pid/socket of a *different*
    // peer daemon coexisting in the same cwd.
    let slug = session_slug();
    if !is_slug_live(&slug) {
        if let Some(lock) = try_reserve_for_cleanup(&slug) {
            let _ = std::fs::remove_file(socket_path_for(&slug));
            let _ = std::fs::remove_file(pid_path_for(&slug));
            drop(lock);
        }
        return Ok("stopped".into());
    }
    let response = send_command_for_slug(&slug, "quit").unwrap_or_else(|_| "stopped".into());
    // Wait for the daemon to actually exit before reaping its runtime
    // files. The previous 5s window was too tight for backends with
    // long finalize phases — dotnet-trace flushing a multi-GB
    // `.nettrace`, callgrind/massif writing their outputs, jdb
    // persisting hits — and the unconditional unlink that followed
    // would orphan a still-alive daemon: socket gone, listener inode
    // still held open by the running process, every subsequent
    // `connect()` failing because the path no longer resolves.
    //
    // Override via `DBG_KILL_TIMEOUT_SECS` for backends with truly
    // long shutdowns. The unlink only happens if the daemon is dead
    // by the deadline; otherwise we leave the files intact and tell
    // the caller, so they can retry or escalate to SIGKILL rather
    // than silently corrupting a live session.
    let timeout_secs = std::env::var("DBG_KILL_TIMEOUT_SECS")
        .ok()
        .and_then(|s| s.parse::<u64>().ok())
        .unwrap_or(30);
    let deadline = std::time::Instant::now() + Duration::from_secs(timeout_secs);
    while std::time::Instant::now() < deadline && is_slug_live(&slug) {
        std::thread::sleep(Duration::from_millis(20));
    }
    if is_slug_live(&slug) {
        let pid = std::fs::read_to_string(pid_path_for(&slug))
            .unwrap_or_default()
            .lines()
            .next()
            .unwrap_or("")
            .trim()
            .to_string();
        return Ok(format!(
            "{response} (warning: daemon pid={pid} still alive after {timeout_secs}s — \
             runtime files left intact to avoid orphaning the session. Retry \
             `dbg kill`, raise DBG_KILL_TIMEOUT_SECS, or send SIGKILL manually)"
        ));
    }
    if let Some(lock) = try_reserve_for_cleanup(&slug) {
        let _ = std::fs::remove_file(socket_path_for(&slug));
        let _ = std::fs::remove_file(pid_path_for(&slug));
        drop(lock);
    }
    Ok(response)
}

/// Wait for the socket file to appear.
pub fn wait_for_socket(timeout: Duration) -> bool {
    let start = std::time::Instant::now();
    while start.elapsed() < timeout {
        if Path::new(&socket_path()).exists() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(100));
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::delve_proto::DelveProtoBackend;
    use crate::backend::jdb::JdbBackend;
    use crate::backend::node_proto::NodeProtoBackend;
    use crate::dap::{DapLaunchConfig, DapTransport};
    use crate::pty::LogHandle;
    use tempfile::TempDir;

    #[test]
    fn import_profile_parse_init_errors_fail_closed() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("profile.json");
        std::fs::write(&source, "not valid profile data").unwrap();
        let db = SessionDb::create(CreateOptions {
            kind: SessionKind::Profile,
            target: source.to_str().unwrap(),
            target_class: TargetClass::NativeCpu,
            cwd: tmp.path(),
            db_path: None,
            label: Some("profile-error".into()),
            target_hash: Some("test".into()),
        })
        .unwrap();

        let parse_error = match ProfileData::load(&source) {
            Ok(_) => panic!("invalid profile input unexpectedly parsed"),
            Err(error) => error.to_string(),
        };
        db.add_failure("profile_parse", &parse_error).unwrap();
        persist_profile_source(&db, &source).unwrap();

        let failure: String = db
            .conn()
            .query_row(
                "SELECT error FROM failures WHERE phase = 'profile_parse'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert!(!failure.is_empty());
        assert_eq!(
            db.meta("profile_raw").unwrap().as_deref(),
            Some("not valid profile data")
        );
    }

    #[test]
    fn dap_adapter_child() {
        if std::env::var_os("DBG_DAP_ADAPTER_CHILD").is_none() {
            return;
        }
        use std::io::{Read, Write};
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        println!("LISTENING {address}");
        std::io::stdout().flush().unwrap();
        let (mut stream, _) = listener.accept().unwrap();

        let read_message = |stream: &mut std::net::TcpStream| {
            let mut header = Vec::new();
            while !header.windows(4).any(|window| window == b"\r\n\r\n") {
                let mut byte = [0u8; 1];
                stream.read_exact(&mut byte)?;
                header.push(byte[0]);
            }
            let header = String::from_utf8_lossy(&header);
            let length = header
                .lines()
                .find_map(|line| line.strip_prefix("Content-Length:"))
                .unwrap()
                .trim()
                .parse::<usize>()
                .unwrap();
            let mut body = vec![0u8; length];
            stream.read_exact(&mut body)?;
            Ok::<_, std::io::Error>(serde_json::from_slice::<serde_json::Value>(&body).unwrap())
        };
        let send_message =
            |stream: &mut std::net::TcpStream, message: serde_json::Value| -> std::io::Result<()> {
                let body = message.to_string();
                write!(stream, "Content-Length: {}\r\n\r\n{body}", body.len())?;
                stream.flush()
            };
        let response = |stream: &mut std::net::TcpStream,
                        request: &serde_json::Value,
                        body: serde_json::Value| {
            send_message(
                stream,
                serde_json::json!({
                    "seq": 1000 + request["seq"].as_i64().unwrap(),
                    "type": "response",
                    "request_seq": request["seq"],
                    "success": true,
                    "command": request["command"],
                    "body": body,
                }),
            )
            .unwrap();
        };
        let event = |stream: &mut std::net::TcpStream, name: &str, body: serde_json::Value| {
            send_message(
                stream,
                serde_json::json!({
                    "seq": 2000,
                    "type": "event",
                    "event": name,
                    "body": body,
                }),
            )
            .unwrap();
        };

        loop {
            let request = read_message(&mut stream).unwrap();
            match request["command"].as_str().unwrap_or_default() {
                "initialize" => response(
                    &mut stream,
                    &request,
                    serde_json::json!({"supportsConfigurationDoneRequest": true}),
                ),
                "launch" => {
                    event(&mut stream, "initialized", serde_json::json!({}));
                    response(&mut stream, &request, serde_json::json!({}));
                }
                "configurationDone" => {
                    response(&mut stream, &request, serde_json::json!({}));
                    event(
                        &mut stream,
                        "stopped",
                        serde_json::json!({"reason": "entry", "threadId": 1}),
                    );
                }
                "continue" => {
                    response(&mut stream, &request, serde_json::json!({}));
                    event(&mut stream, "continued", serde_json::json!({"threadId": 1}));
                    event(
                        &mut stream,
                        "stopped",
                        serde_json::json!({"reason": "breakpoint", "threadId": 1}),
                    );
                }
                "stackTrace" => response(
                    &mut stream,
                    &request,
                    serde_json::json!({"stackFrames": [{
                        "id": 7,
                        "name": "main.main",
                        "line": 7,
                        "column": 1,
                        "source": {"path": "/virtual/e2e.go"}
                    }]}),
                ),
                "scopes" => response(
                    &mut stream,
                    &request,
                    serde_json::json!({"scopes": [{
                        "name": "Locals",
                        "variablesReference": 1
                    }]}),
                ),
                "variables" => {
                    response(
                        &mut stream,
                        &request,
                        serde_json::json!({"variables": [{
                            "name": "answer",
                            "value": "42",
                            "variablesReference": 0
                        }]}),
                    );
                    event(&mut stream, "terminated", serde_json::json!({}));
                }
                "disconnect" => {
                    response(&mut stream, &request, serde_json::json!({}));
                    break;
                }
                _ => response(&mut stream, &request, serde_json::json!({})),
            }
        }
    }

    #[cfg(unix)]
    fn real_dap_session(tmp: &TempDir) -> Session {
        let tmp_dir = tmp.path().join("session-tmp");
        std::fs::create_dir_all(&tmp_dir).unwrap();
        // This is a small wire-level adapter. It is a child process with a
        // real TCP socket and DAP framing, so these tests exercise
        // DapTransport, the daemon dispatcher, and SessionDb together.
        // It does not provide prepared JSON to the daemon.
        let adapter = std::env::current_exe().unwrap();
        let transport = DapTransport::spawn(DapLaunchConfig {
            // Run the adapter in the background so it is reparented before
            // the other daemon-process tests inspect this test process's
            // children for signalling. The adapter exits after the
            // terminated event, so this does not leave a child behind.
            bin: "sh".into(),
            args: vec![
                "-c".into(),
                "env DBG_DAP_ADAPTER_CHILD=1 \"$1\" --exact daemon::tests::dap_adapter_child --nocapture &".into(),
                "dbg-dap-adapter".into(),
                adapter.display().to_string(),
            ],
            listen_marker: "LISTENING".into(),
            launch_verb: "launch".into(),
            launch_args: serde_json::json!({"request": "launch"}),
            preassigned_addr: None,
            connect_addr: None,
            debuggee_pid: None,
        })
        .unwrap();
        let db = SessionDb::create(CreateOptions {
            kind: SessionKind::Debug,
            target: "/virtual/e2e.go",
            target_class: TargetClass::NativeCpu,
            cwd: tmp.path(),
            db_path: None,
            label: Some("dap-e2e".into()),
            target_hash: Some("dap-e2e".into()),
        })
        .unwrap();
        Session {
            proc: Box::new(transport),
            profile: None,
            db: Some(db),
            hit_seq: HashMap::new(),
            save_to: None,
            tmp_dir,
            target: "/virtual/e2e.go".into(),
            cwd: tmp.path().to_path_buf(),
            target_class: TargetClass::NativeCpu,
        }
    }

    /// Regression: the accept loop used non-blocking accept + a
    /// 10ms sleep, which burned ~100 wakeups/sec on an idle daemon
    /// for no benefit (ctrl-c is signal-delivered regardless of
    /// whether accept is blocked, and the daemon has no other work
    /// to do between connections). Assert the polling pattern stays
    /// out of the source.
    #[test]
    fn accept_loop_does_not_busy_poll() {
        let src = include_str!("daemon.rs");
        // Confine the search to run_daemon's accept loop region: after
        // the listener bind, before std::thread::scope's closing.
        let bind_pos = src.find("UnixListener::bind").expect("bind site");
        let scope_end = src[bind_pos..]
            .find("cleanup_and_exit(false);\n}")
            .map(|p| bind_pos + p)
            .unwrap_or(src.len());
        let region = &src[bind_pos..scope_end];
        assert!(
            !region.contains("set_nonblocking(true)"),
            "listener must not be set non-blocking — that's the polling defect"
        );
        assert!(
            !region.contains("WouldBlock"),
            "WouldBlock retry loop signals the polling defect has returned"
        );
    }

    #[test]
    fn releasing_a_full_ipc_slot_wakes_accept_once() {
        assert!(should_wake_accept_after_release(MAX_IPC_CONNECTIONS));
        assert!(!should_wake_accept_after_release(MAX_IPC_CONNECTIONS - 1));
        assert!(!should_wake_accept_after_release(0));
    }

    #[test]
    fn pid_and_socket_bind_before_init_commands() {
        // Regression: jitdasm (and any other backend with a long-init
        // flow) made `dbg status` return "no session" and `dbg start`
        // appear to hang for minutes, because pid_path + socket bind
        // were gated on init_commands completing. Assert the source
        // order here: pid write + socket bind must precede the
        // `for cmd in &init_commands` loop inside run_daemon.
        let src = include_str!("daemon.rs");
        let pid_write = src
            .find("std::fs::write(&pid_path()")
            .expect("pid_path write");
        let bind = src
            .find("UnixListener::bind(&socket_path()")
            .expect("socket bind");
        let init_loop = src.find("for cmd in &init_commands").expect("init loop");
        assert!(
            pid_write < init_loop,
            "pid file write must precede init_commands loop"
        );
        assert!(
            bind < init_loop,
            "socket bind must precede init_commands loop"
        );
    }

    #[test]
    fn ipc_slow_drip_has_an_absolute_deadline() {
        use std::os::unix::net::UnixStream;

        let (mut writer, mut reader) = UnixStream::pair().unwrap();
        let producer = std::thread::spawn(move || {
            writer.write_all(b"x").unwrap();
            std::thread::sleep(Duration::from_millis(40));
            let _ = writer.write_all(b"y");
        });

        let started = Instant::now();
        let (result, bytes) = read_ipc_request_with_timeout(&mut reader, Duration::from_millis(10));
        assert!(result.is_err(), "slow-drip request must be rejected");
        assert_eq!(bytes, b"x");
        assert!(started.elapsed() < Duration::from_millis(100));
        producer.join().unwrap();
    }

    #[test]
    fn init_command_status_marker_surfaces_nonzero_exit() {
        assert_eq!(parse_init_status("output\n__DBG_INIT_STATUS__0\n"), Some(0));
        assert_eq!(parse_init_status("output\n__DBG_INIT_STATUS__7\n"), Some(7));
        assert_eq!(parse_init_status("output without marker"), None);
    }

    #[test]
    fn protocol_help_uses_static_surface_without_native_command() {
        let help = cache_help(&NodeProtoBackend, || {
            panic!("node-proto must not receive the native help command")
        })
        .unwrap();
        assert!(help.contains("node-proto"));
        assert_eq!(
            native_help_topic_command(&NodeProtoBackend, "unknown-topic"),
            None
        );
    }

    #[test]
    fn jdb_run_waits_past_the_transient_startup_prompt() {
        let backend = JdbBackend;
        assert!(jdb_run_needs_vm_prompt(
            &backend,
            "run",
            "run Broken\nSet deferred uncaught java.lang.Throwable\n"
        ));
        assert!(!jdb_run_needs_vm_prompt(
            &backend,
            "run",
            "VM Started:\nBreakpoint hit: Broken$Sku.equals"
        ));
        assert!(!jdb_run_needs_vm_prompt(&backend, "stack", "where\n"));
    }

    // These tests mutate process-global env vars (XDG_RUNTIME_DIR +
    // DBG_SESSION) to isolate pid/socket paths. Cargo runs tests in
    // parallel threads that share the env, so serialize them with
    // a shared Mutex or they'll stomp each other.
    static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    #[test]
    fn is_running_is_observational_only() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        // is_running() is read-only: it inspects pid_path and
        // /proc/<pid>/comm but never deletes either. Plant a pid
        // pointing at init (pid 1, comm = "systemd" or "init") —
        // alive, but its comm doesn't start with "dbg", so
        // is_running must reject it. The pid file must still be on
        // disk afterwards.
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "testobs");
        }
        std::fs::write(pid_path(), "1").unwrap();
        assert!(
            !is_running(),
            "live pid whose comm != dbg must read as not-running"
        );
        assert!(pid_path().exists(), "is_running must not mutate fs");
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn clean_stale_runtime_files_reaps_orphans() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "testreap");
        }
        // Dead PID + stray socket file from a crashed daemon.
        std::fs::write(pid_path(), "2147483000").unwrap();
        std::fs::write(socket_path(), "").unwrap();
        clean_stale_runtime_files();
        assert!(!pid_path().exists(), "dead-pid file should be reaped");
        assert!(
            !socket_path().exists(),
            "stray socket file should be reaped"
        );
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn clean_stale_runtime_files_reaps_foreign_live_pid() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "testforeign");
        }
        // PID 1 is alive in normal containers/hosts, but its comm is not dbg.
        // A stale dbg pid file recycled to such a foreign process must not pin
        // the runtime slot forever.
        std::fs::write(pid_path(), "1").unwrap();
        std::fs::write(socket_path(), "").unwrap();
        clean_stale_runtime_files();
        assert!(!pid_path().exists(), "foreign live pid should be reaped");
        assert!(
            !socket_path().exists(),
            "foreign live socket should be reaped"
        );
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn allocate_slug_uses_bare_when_free() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::remove_var("DBG_SESSION");
        }
        let base = cwd_base_slug();
        let got = allocate_slug().unwrap();
        assert_eq!(got, base, "free slot should return bare cwd slug");
    }

    #[test]
    fn allocate_slug_appends_pid_when_bare_is_live() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::remove_var("DBG_SESSION");
        }
        // Plant a live daemon at the bare slug (our own pid + socket).
        let base = cwd_base_slug();
        std::fs::write(pid_path_for(&base), std::process::id().to_string()).unwrap();
        std::fs::write(socket_path_for(&base), "").unwrap();
        let got = allocate_slug().unwrap();
        assert!(
            got.starts_with(&base),
            "allocated slug should share cwd prefix: {got}"
        );
        assert_ne!(got, base, "must not reuse the busy bare slot");
        assert!(got.contains('-'), "pid-suffixed slug expected: {got}");
    }

    #[test]
    fn allocate_slug_fails_on_explicit_collision() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "named-slot");
        }
        std::fs::write(pid_path_for("named-slot"), std::process::id().to_string()).unwrap();
        std::fs::write(socket_path_for("named-slot"), "").unwrap();
        let err = allocate_slug().unwrap_err();
        assert!(err.to_string().contains("already live"), "{err}");
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn concurrent_starters_cannot_share_a_slug_reservation() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::remove_var("DBG_SESSION");
        }

        // Both contenders start at the same synchronization point. One
        // owns the bare slot; the other must receive a different slot. A
        // stale-marker check followed by unlink would allow both to report
        // the same slot in this interleaving.
        let gate = std::sync::Arc::new(std::sync::Barrier::new(2));
        let results = std::thread::scope(|scope| {
            let mut workers = Vec::new();
            for _ in 0..2 {
                let gate = std::sync::Arc::clone(&gate);
                workers.push(scope.spawn(move || {
                    gate.wait();
                    allocate_slug().unwrap()
                }));
            }
            workers
                .into_iter()
                .map(|worker| worker.join().unwrap())
                .collect::<Vec<_>>()
        });
        assert_ne!(results[0], results[1], "starters must own distinct slugs");
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn stale_reservation_recovery_is_atomic_across_processes() {
        use std::process::Command;

        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let barrier = tmp.path().join("reservation-barrier");
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "stale-race");
        }

        // This is a stale reservation left by a crashed starter. Both new
        // starters must contend for this same inode. The test helper pauses
        // one owner while the other reaches flock, so the interleaving is
        // deterministic and uses independent processes.
        let reservation = reservation_path_for("stale-race");
        std::fs::write(&reservation, b"2147483000\n").unwrap();

        let executable = std::env::current_exe().unwrap();
        let mut children = Vec::new();
        for n in 0..2 {
            let result = tmp.path().join(format!("result-{n}"));
            children.push(
                Command::new(&executable)
                    .arg("--exact")
                    .arg("daemon::tests::claim_slug_helper")
                    .arg("--nocapture")
                    .env("XDG_RUNTIME_DIR", tmp.path())
                    .env("DBG_SESSION", "stale-race")
                    .env("DBG_CLAIM_HELPER", "1")
                    .env("DBG_CLAIM_RESULT", &result)
                    .env("DBG_TEST_RESERVATION_BARRIER", &barrier)
                    .spawn()
                    .unwrap(),
            );
        }

        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while std::fs::read_dir(&barrier)
            .ok()
            .map(|entries| {
                entries
                    .flatten()
                    .filter(|e| e.file_name().to_string_lossy().starts_with("contender-"))
                    .count()
            })
            .unwrap_or(0)
            < 2
            && std::time::Instant::now() < deadline
        {
            std::thread::sleep(Duration::from_millis(2));
        }
        std::fs::create_dir_all(&barrier).unwrap();
        std::fs::write(barrier.join("begin"), b"begin").unwrap();

        let locked = loop {
            let count = std::fs::read_dir(&barrier)
                .ok()
                .map(|entries| {
                    entries
                        .flatten()
                        .filter(|e| e.file_name().to_string_lossy().starts_with("locked-"))
                        .count()
                })
                .unwrap_or(0);
            if count > 0 || std::time::Instant::now() >= deadline {
                break count;
            }
            std::thread::sleep(Duration::from_millis(2));
        };

        // The losing process must already have observed the held lock before
        // the winner is allowed to release it. Otherwise a sequential stale
        // recovery could look like two simultaneous owners.
        let loser_reported = loop {
            let reported = (0..2).any(|n| {
                std::fs::read_to_string(tmp.path().join(format!("result-{n}")))
                    .is_ok_and(|result| result.starts_with("err:"))
            });
            if reported || std::time::Instant::now() >= deadline {
                break reported;
            }
            std::thread::sleep(Duration::from_millis(2));
        };
        std::fs::write(barrier.join("release"), b"release").unwrap();

        let mut statuses = Vec::new();
        for child in children {
            let output = child.wait_with_output().unwrap();
            statuses.push((output.status, output.stdout, output.stderr));
        }
        assert_eq!(locked, 1, "exactly one process must acquire the stale lock");
        assert!(
            loser_reported,
            "the second process must lose while the lock is held"
        );
        let outcomes: Vec<_> = (0..2)
            .map(|n| std::fs::read_to_string(tmp.path().join(format!("result-{n}"))).unwrap())
            .collect();
        assert!(statuses.iter().all(|(status, _, _)| status.success()));
        assert_eq!(
            outcomes
                .iter()
                .filter(|result| result.as_str() == "ok")
                .count(),
            1
        );
        assert_eq!(
            outcomes
                .iter()
                .filter(|result| result.starts_with("err:"))
                .count(),
            1
        );
        assert!(
            reservation.exists(),
            "stale recovery must retain the stable inode"
        );

        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn claim_slug_helper() {
        if std::env::var_os("DBG_CLAIM_HELPER").is_none() {
            return;
        }
        let result = std::env::var("DBG_CLAIM_RESULT").unwrap();
        let outcome = match allocate_slug() {
            Ok(_) => "ok".to_string(),
            Err(error) => format!("err:{error}"),
        };
        std::fs::write(result, outcome).unwrap();
    }

    #[test]
    fn pointer_file_is_honored_when_live() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::remove_var("DBG_SESSION");
        }
        let base = cwd_base_slug();
        let pid_slug = format!("{base}-9999999");
        std::fs::write(pid_path_for(&pid_slug), std::process::id().to_string()).unwrap();
        std::fs::write(socket_path_for(&pid_slug), "").unwrap();
        write_latest_pointer(&pid_slug);
        assert_eq!(
            session_slug(),
            pid_slug,
            "pointer should win over bare fallback"
        );
    }

    #[test]
    fn startup_error_cleanup_removes_published_runtime_state() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "failed-startup");
        }

        let slug = session_slug();
        std::fs::write(socket_path(), "socket").unwrap();
        std::fs::write(pid_path(), "123").unwrap();
        write_latest_pointer(&slug);
        let evidence = session_tmp("partial.data");
        std::fs::write(&evidence, "partial").unwrap();

        cleanup_after_startup_error();

        assert!(!socket_path().exists());
        assert!(!pid_path().exists());
        assert!(!latest_pointer_path().exists());
        assert!(!evidence.parent().unwrap().exists());
        unsafe {
            std::env::remove_var("DBG_SESSION");
            std::env::remove_var("XDG_RUNTIME_DIR");
        }
    }

    #[test]
    fn live_slugs_in_cwd_enumerates_peers() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::remove_var("DBG_SESSION");
        }
        let base = cwd_base_slug();
        let a = base.clone();
        let b = format!("{base}-111");
        for s in [&a, &b] {
            std::fs::write(pid_path_for(s), std::process::id().to_string()).unwrap();
            std::fs::write(socket_path_for(s), "").unwrap();
        }
        let peers = live_slugs_in_cwd();
        assert!(peers.contains(&a), "missing {a}: {peers:?}");
        assert!(peers.contains(&b), "missing {b}: {peers:?}");
    }

    /// Regression: during a profile-mode session (nodeprof/pprof/…),
    /// the daemon routed every command to `ProfileData::handle_command`,
    /// which returned "unknown command: sessions" for meta verbs like
    /// `sessions`, `status`, `save`, `kill`, `replay`. The dispatcher
    /// now checks `is_profile_repl_verb` first so only profile verbs
    /// are intercepted; everything else falls through to normal
    /// dispatch.
    #[test]
    fn profile_verbs_intercepted_meta_verbs_fall_through() {
        for v in [
            "top",
            "callers compute",
            "callees main",
            "traces 5",
            "tree",
            "hotpath",
            "threads",
            "stats",
            "search sort",
            "focus sort",
            "ignore gc",
            "reset",
        ] {
            assert!(
                is_profile_repl_verb(v),
                "`{v}` must route to the profile REPL"
            );
        }
        for v in [
            "sessions",
            "status",
            "save",
            "kill",
            "replay session-1",
            "hits main.py:1",
            "cross foo",
            "disasm main",
            "help",
            "break main.py:1",
        ] {
            assert!(
                !is_profile_repl_verb(v),
                "`{v}` must NOT be intercepted by the profile REPL"
            );
        }
    }

    /// Regression: between `pid_path` write and `socket_path` bind in
    /// `run_daemon`, a concurrent `dbg start` could call
    /// `clean_stale_runtime_files`, see "alive dbg pid, socket missing",
    /// and delete the live daemon's pid file. The fix: bail out when
    /// the pid file names a live dbg process, regardless of socket
    /// presence — that's a healthy daemon (or one mid-startup), either
    /// way its files are not orphans.
    #[test]
    fn proc_children_parser_ignores_invalid_tokens() {
        assert_eq!(
            parse_proc_children("101 202 nope 303\n"),
            vec![101, 202, 303]
        );
        assert!(parse_proc_children("").is_empty());
    }

    #[test]
    fn collector_signal_targets_prefers_grandchildren() {
        let targets = collector_signal_targets(10, |pid| match pid {
            10 => vec![20, 30],
            20 => vec![200, 201],
            30 => vec![300],
            _ => Vec::new(),
        })
        .unwrap();
        assert_eq!(targets, vec![200, 201, 300]);
    }

    #[test]
    fn collector_signal_targets_falls_back_to_direct_children() {
        let targets = collector_signal_targets(10, |pid| match pid {
            10 => vec![20, 30],
            _ => Vec::new(),
        })
        .unwrap();
        assert_eq!(targets, vec![20, 30]);
    }

    #[test]
    fn collector_signal_targets_rejects_daemon_without_children() {
        let err = collector_signal_targets(10, |_| Vec::new())
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("has no child processes"),
            "unexpected error: {err}"
        );
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn finalize_signals_a_live_collector_with_an_identity_safe_handle() {
        use std::process::Command;

        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "finalize-live");
        }
        let mut collector = Command::new("sleep").arg("30").spawn().unwrap();
        let collector_pid = collector.id() as i32;
        std::fs::write(pid_path(), daemon_pid_record(std::process::id() as i32)).unwrap();

        let report = finalize_collector().expect("live collector must be signalled");
        assert!(report.signalled.contains(&collector_pid));
        let status = collector.wait().unwrap();
        assert!(!status.success(), "SIGINT must stop the collector");

        let _ = std::fs::remove_file(pid_path());
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn finalize_does_not_signal_an_unrelated_process_in_the_collector_group() {
        use std::process::Command;

        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "finalize-group");
        }

        let mut collector = Command::new("sleep").arg("30").spawn().unwrap();
        let collector_pid = collector.id() as i32;
        // Both children inherit this test process's process group. The orphan
        // below is reparented to init, so it is not a collector-tree
        // descendant even though a negative-PGID kill would reach it.
        let orphan_pid_file = tmp.path().join("orphan.pid");
        let command = format!("sleep 30 & echo $! > {}; exit 0", orphan_pid_file.display());
        let mut helper = Command::new("sh").arg("-c").arg(command).spawn().unwrap();
        let deadline = Instant::now() + Duration::from_secs(2);
        while !orphan_pid_file.exists() && Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(5));
        }
        let orphan_pid: i32 = std::fs::read_to_string(&orphan_pid_file)
            .unwrap()
            .trim()
            .parse()
            .unwrap();
        helper.wait().unwrap();

        std::fs::write(pid_path(), daemon_pid_record(std::process::id() as i32)).unwrap();
        let report = finalize_collector().expect("collector must be signalled");
        assert!(report.signalled.contains(&collector_pid));
        assert!(
            nix::sys::signal::kill(nix::unistd::Pid::from_raw(orphan_pid), None).is_ok(),
            "an unrelated same-group orphan must survive finalize"
        );

        let _ = collector.wait();
        let _ = nix::sys::signal::kill(
            nix::unistd::Pid::from_raw(orphan_pid),
            nix::sys::signal::Signal::SIGKILL,
        );
        let _ = std::fs::remove_file(pid_path());
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn finalize_rejects_a_replaced_daemon_identity() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "finalize-daemon-replaced");
        }
        std::fs::write(
            pid_path(),
            format!("{}\nnot-the-current-instance\n", std::process::id()),
        )
        .unwrap();

        let error = finalize_collector().unwrap_err().to_string();
        assert!(
            error.contains("changed process instance") || error.contains("refusing to signal"),
            "unexpected replacement error: {error}"
        );

        let _ = std::fs::remove_file(pid_path());
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn identity_safe_handle_rejects_a_replaced_descendant_identity() {
        let pid = std::process::id() as i32;
        let current = process_identity(pid).unwrap();
        let replacement = format!("{current}\nreplacement-executable");
        assert!(
            open_identity_safe_pidfd(pid, &replacement).is_err(),
            "a PID with a different start identity must not be signalled"
        );
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn descendant_handle_failure_aborts_capture_before_any_signal() {
        use std::process::Command;

        let parent = open_current_identity_safe_pidfd(std::process::id() as i32).unwrap();
        let mut child = Command::new("sleep").arg("30").spawn().unwrap();
        let child_pid = child.id() as i32;

        let error = open_verified_children_from_pids(&parent, vec![child_pid, i32::MAX])
            .unwrap_err()
            .to_string();
        assert!(
            error.contains("identity-safe handle"),
            "unexpected handle failure: {error}"
        );
        assert!(
            nix::sys::signal::kill(nix::unistd::Pid::from_raw(child_pid), None).is_ok(),
            "capture failure must happen before a verified sibling is signalled"
        );

        let _ = child.kill();
        let _ = child.wait();
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn finalize_fails_closed_when_identity_safe_handle_cannot_be_verified() {
        let pid = std::process::id() as i32;
        assert!(
            open_identity_safe_pidfd(pid, "").is_err(),
            "missing identity must never fall back to a bare PID signal"
        );
        let src = include_str!("daemon.rs");
        assert!(src.contains("SYS_pidfd_send_signal"));
    }

    #[test]
    fn clean_stale_runtime_files_preserves_live_daemon_mid_startup() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "testmidstart");
        }
        // Mid-startup: pid file written, socket not yet bound.
        std::fs::write(pid_path(), std::process::id().to_string()).unwrap();
        assert!(!socket_path().exists());
        clean_stale_runtime_files();
        assert!(
            pid_path().exists(),
            "live daemon mid-startup must keep its pid file"
        );
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn pid_only_marker_is_accepted_when_procfs_is_unavailable() {
        assert!(daemon_identity_matches_with_procfs(
            &format!("{}\n", std::process::id()),
            std::process::id() as i32 + 1,
            false,
        ));
    }

    /// Regression: `kill_daemon` used to unlink socket+pid files
    /// unconditionally after a 5s shutdown wait, even when the daemon
    /// was still alive. That orphaned the live daemon — its listener
    /// inode was held open but the path was gone, so new clients
    /// couldn't connect. The fix: only reap files if the daemon is
    /// actually dead by the deadline; otherwise return a warning and
    /// leave runtime files intact.
    #[test]
    fn kill_daemon_does_not_unlink_files_when_daemon_still_alive() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "testkilltimeout");
            std::env::set_var("DBG_KILL_TIMEOUT_SECS", "1");
        }
        // Simulate a stuck daemon: pid file points to our own PID
        // (which is_slug_live recognizes as live + dbg-comm), but no
        // listener is bound — send_command will fail with EBADF/
        // ENOENT and unwrap_or_else absorbs it, sending us into the
        // poll loop. Loop exits at the 1s deadline with the daemon
        // still "alive" (us).
        std::fs::write(pid_path(), std::process::id().to_string()).unwrap();
        std::fs::write(socket_path(), "").unwrap();

        let result = kill_daemon().expect("kill_daemon must not error");
        assert!(
            result.contains("still alive"),
            "expected warning about stuck daemon, got: {result}"
        );
        assert!(
            pid_path().exists(),
            "pid file must survive when daemon is still alive at deadline"
        );
        assert!(
            socket_path().exists(),
            "socket file must survive when daemon is still alive at deadline"
        );
        // Cleanup so other tests aren't polluted.
        let _ = std::fs::remove_file(pid_path());
        let _ = std::fs::remove_file(socket_path());
        unsafe {
            std::env::remove_var("DBG_SESSION");
            std::env::remove_var("DBG_KILL_TIMEOUT_SECS");
        }
    }

    #[test]
    fn clean_stale_runtime_files_preserves_live_daemon() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        unsafe {
            std::env::set_var("XDG_RUNTIME_DIR", tmp.path());
            std::env::set_var("DBG_SESSION", "testkeep");
        }
        // Simulate a healthy daemon: our own PID + socket present.
        let my_pid = std::process::id();
        std::fs::write(pid_path(), my_pid.to_string()).unwrap();
        std::fs::write(socket_path(), "").unwrap();
        clean_stale_runtime_files();
        assert!(pid_path().exists(), "live daemon's pid file must survive");
        assert!(socket_path().exists(), "live daemon's socket must survive");
        unsafe {
            std::env::remove_var("DBG_SESSION");
        }
    }

    #[test]
    fn backend_target_class_mapping() {
        assert_eq!(backend_target_class("lldb"), TargetClass::NativeCpu);
        assert_eq!(backend_target_class("delve"), TargetClass::NativeCpu);
        assert_eq!(backend_target_class("pdb"), TargetClass::Python);
        assert_eq!(backend_target_class("debugpy-proto"), TargetClass::Python);
        assert_eq!(
            backend_target_class("netcoredbg"),
            TargetClass::ManagedDotnet
        );
        assert_eq!(
            backend_target_class("netcoredbg-proto"),
            TargetClass::ManagedDotnet
        );
        assert_eq!(backend_target_class("jdb"), TargetClass::Jvm);
        assert_eq!(backend_target_class("node-inspect"), TargetClass::JsNode);
        assert_eq!(backend_target_class("node-proto"), TargetClass::JsNode);
        assert_eq!(backend_target_class("rdbg"), TargetClass::Ruby);
        assert_eq!(backend_target_class("phpdbg"), TargetClass::Php);
        // Unknown backends fall through to native-cpu — informational only.
        assert_eq!(backend_target_class("perf"), TargetClass::NativeCpu);
    }

    #[test]
    fn report_oriented_profilers_are_profile_sessions() {
        let profilers: Vec<Box<dyn Backend>> = vec![
            Box::new(crate::backend::callgrind::CallgrindBackend),
            Box::new(crate::backend::pstats::PstatsBackend),
            Box::new(crate::backend::massif::MassifBackend),
            Box::new(crate::backend::memcheck::MemcheckBackend),
            Box::new(crate::backend::ghcprof::GhcProfBackend),
            Box::new(crate::backend::stackprof::StackprofBackend),
            Box::new(crate::backend::xdebug::XdebugProfileBackend),
        ];
        for backend in profilers {
            assert_eq!(
                backend_session_kind(backend.as_ref()),
                SessionKind::Profile,
                "{} must create a profile-kind session",
                backend.name()
            );
        }

        let debugger = crate::backend::lldb::LldbBackend;
        assert_eq!(
            backend_session_kind(&debugger),
            SessionKind::Debug,
            "debuggers must retain debug-kind sessions"
        );
    }

    #[test]
    fn command_may_stop_covers_canonical_and_native_verbs() {
        // Canonical verbs.
        assert!(command_may_stop("continue"));
        assert!(command_may_stop("step"));
        assert!(command_may_stop("next"));
        assert!(command_may_stop("finish"));
        assert!(command_may_stop("run"));
        assert!(command_may_stop("restart"));
        // Native / aliased forms the daemon is likely to see on raw.
        assert!(command_may_stop("process continue"));
        assert!(command_may_stop("thread step-in"));
        assert!(command_may_stop("stepout"));
        assert!(command_may_stop("return"));
        assert!(command_may_stop("c"));
        assert!(command_may_stop("s"));
        assert!(command_may_stop("n"));
        // Definitely non-stopping commands.
        assert!(!command_may_stop("print x"));
        assert!(!command_may_stop("breakpoint list"));
        assert!(!command_may_stop("frame variable"));
        assert!(!command_may_stop("help"));
    }

    #[test]
    fn raw_copy_failure_is_reported_without_removing_source() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("source");
        std::fs::create_dir(&source).unwrap();
        let artifact = source.join("perf.data");
        std::fs::write(&artifact, b"capture").unwrap();

        // A file at the destination makes the copy fail. The source must
        // remain available because cleanup is allowed only after a full
        // durable copy succeeds.
        let destination = tmp.path().join("destination");
        std::fs::write(&destination, b"not a directory").unwrap();
        assert!(copy_raw_tree(&source, &destination).is_err());
        assert!(artifact.exists());
    }

    #[test]
    fn daemon_cleanup_after_raw_persist() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("session-tmp");
        std::fs::create_dir_all(&source).unwrap();
        let artifact = source.join("cleanup-capture.data");
        std::fs::write(&artifact, b"capture").unwrap();
        let save_path = session_db::sessions_dir(tmp.path()).join("dap-e2e.db");
        let session = Mutex::new(real_dap_session(&tmp));
        session.lock().unwrap().save_to = Some(save_path.clone());
        let backend = DelveProtoBackend;

        assert!(raw_evidence_present(&source).unwrap());
        let (_, keep_session_tmp) = handle_quit(&backend, &session, &AtomicI32::new(0), None);
        assert!(!keep_session_tmp);
        assert!(save_path.exists());

        // This is the same cleanup operation used by the daemon after the
        // persistence-dependent decision in handle_quit.
        cleanup_session_tmp_at(&source, keep_session_tmp);
        assert!(!source.exists());
        let durable = session_db::raw_dir(tmp.path(), "dap-e2e");
        assert_eq!(
            std::fs::read(durable.join("cleanup-capture.data")).unwrap(),
            b"capture"
        );
        assert!(SessionDb::open(&save_path).is_ok());
    }

    #[cfg(unix)]
    #[test]
    fn daemon_cleanup_persists_nested_session_db_evidence() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("session-tmp");
        let nested = source.join("capture").join("session.db");
        std::fs::create_dir_all(nested.parent().unwrap()).unwrap();
        std::fs::write(&nested, b"nested raw evidence").unwrap();
        let save_path = session_db::sessions_dir(tmp.path()).join("nested-db-evidence.db");
        let session = Mutex::new(real_dap_session(&tmp));
        session.lock().unwrap().save_to = Some(save_path.clone());
        let backend = DelveProtoBackend;

        let (_, keep_session_tmp) = handle_quit(&backend, &session, &AtomicI32::new(0), None);
        assert!(
            !keep_session_tmp,
            "nested raw evidence was treated as empty"
        );
        cleanup_session_tmp_at(&source, keep_session_tmp);

        assert!(!source.exists());
        assert_eq!(
            std::fs::read(session_db::raw_dir(tmp.path(), "dap-e2e").join("capture/session.db"))
                .unwrap(),
            b"nested raw evidence"
        );
        assert!(SessionDb::open(&save_path).is_ok());
    }

    #[cfg(unix)]
    #[test]
    fn daemon_cleanup_preserves_evidence_on_failure() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("session-tmp");
        std::fs::create_dir_all(&source).unwrap();
        let outside = tmp.path().join("live-evidence");
        std::fs::write(&outside, b"live evidence").unwrap();
        std::os::unix::fs::symlink(&outside, source.join("cleanup-capture.data")).unwrap();
        let save_path = session_db::sessions_dir(tmp.path()).join("dap-e2e.db");
        let raw_destination = session_db::raw_dir(tmp.path(), "dap-e2e");
        let _ = std::fs::remove_dir_all(&raw_destination);
        let session = Mutex::new(real_dap_session(&tmp));
        session.lock().unwrap().save_to = Some(save_path.clone());
        let backend = DelveProtoBackend;

        // Rejected raw input is still evidence. A failed persistence attempt
        // must leave both the source entry and its target available.
        assert!(raw_evidence_present(&source).unwrap());
        let (_, keep_session_tmp) = handle_quit(&backend, &session, &AtomicI32::new(0), None);
        assert!(
            keep_session_tmp,
            "cleanup was allowed after raw-copy failure"
        );
        cleanup_session_tmp_at(&source, keep_session_tmp);
        assert!(source.join("cleanup-capture.data").exists());
        assert!(outside.exists());
        assert!(!session.lock().unwrap().proc.is_alive());
        assert!(!save_path.exists());
        assert!(!raw_destination.join("cleanup-capture.data").exists());

        // Keep the test process from leaving the intentionally preserved raw
        // source behind after proving the failure behavior.
        std::fs::remove_dir_all(source).unwrap();
    }

    #[cfg(unix)]
    #[test]
    fn raw_evidence_presence_fails_closed_for_invalid_root() {
        let tmp = TempDir::new().unwrap();
        let target = tmp.path().join("outside");
        std::fs::create_dir(&target).unwrap();
        std::fs::write(target.join("capture.data"), b"evidence").unwrap();
        let link = tmp.path().join("raw-link");
        std::os::unix::fs::symlink(&target, &link).unwrap();

        assert!(raw_evidence_present(&link).unwrap());
        assert!(!raw_evidence_present(&tmp.path().join("missing")).unwrap());
    }

    #[cfg(unix)]
    #[test]
    fn raw_evidence_does_not_ignore_session_db_symlink() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("session-tmp");
        std::fs::create_dir(&source).unwrap();
        let outside = tmp.path().join("outside.db");
        std::fs::write(&outside, b"not the session database").unwrap();
        std::os::unix::fs::symlink(&outside, source.join("session.db")).unwrap();

        assert!(raw_evidence_present(&source).unwrap());
    }

    #[test]
    fn raw_evidence_reports_directory_and_iterator_errors() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("session-tmp");
        std::fs::create_dir(&source).unwrap();
        let artifact = source.join("capture.data");
        std::fs::write(&artifact, b"capture").unwrap();

        let open_error = |_path: &Path| -> std::io::Result<
            Box<dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>>,
        > { Err(std::io::Error::other("directory inspection failed")) };
        assert!(raw_evidence_present_with(&source, &open_error).is_err());

        let source_for_reader = source.clone();
        let iterator_error = move |path: &Path| -> std::io::Result<
            Box<dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>>,
        > {
            if path == source_for_reader {
                let entry = std::fs::read_dir(path)?.next().unwrap()?;
                Ok(Box::new(
                    vec![
                        Ok(entry),
                        Err(std::io::Error::other("directory iterator failed")),
                    ]
                    .into_iter(),
                ))
            } else {
                Ok(Box::new(std::fs::read_dir(path)?))
            }
        };
        let error = raw_evidence_present_with(&source, &iterator_error).unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::Other);
    }

    #[cfg(unix)]
    #[test]
    fn daemon_cleanup_preserves_evidence_on_inspection_failure() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("session-tmp");
        std::fs::create_dir_all(&source).unwrap();
        let artifact = source.join("capture.data");
        std::fs::write(&artifact, b"capture").unwrap();
        let save_path = session_db::sessions_dir(tmp.path()).join("inspection-failure.db");
        let session = Mutex::new(real_dap_session(&tmp));
        session.lock().unwrap().save_to = Some(save_path.clone());
        let backend = DelveProtoBackend;
        let reader = |_path: &Path| -> std::io::Result<
            Box<dyn Iterator<Item = std::io::Result<std::fs::DirEntry>>>,
        > { Err(std::io::Error::other("raw evidence inspection failed")) };

        let (_, keep_session_tmp) =
            handle_quit_with(&backend, &session, &AtomicI32::new(0), None, &reader);
        assert!(
            keep_session_tmp,
            "inspection uncertainty must keep the raw source"
        );
        cleanup_session_tmp_at(&source, keep_session_tmp);
        assert!(artifact.exists());
        assert!(!session.lock().unwrap().proc.is_alive());
        assert!(!save_path.exists());
    }

    #[test]
    fn raw_copy_rejects_file_symlinks_without_following_them() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("source");
        let destination = tmp.path().join("destination");
        std::fs::create_dir(&source).unwrap();
        let outside = tmp.path().join("outside-secret");
        std::fs::write(&outside, b"must not be copied").unwrap();
        std::os::unix::fs::symlink(&outside, source.join("capture.data")).unwrap();

        let error = copy_raw_tree(&source, &destination).unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
        assert!(!destination.join("capture.data").exists());
        assert!(outside.exists(), "the symlink target must remain untouched");
    }

    #[test]
    fn raw_copy_rejects_directory_symlinks_without_traversing_them() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("source");
        let destination = tmp.path().join("destination");
        let outside = tmp.path().join("outside");
        std::fs::create_dir(&source).unwrap();
        std::fs::create_dir(&outside).unwrap();
        std::fs::write(outside.join("secret.nettrace"), b"must not be copied").unwrap();
        std::os::unix::fs::symlink(&outside, source.join("nested")).unwrap();

        let error = copy_raw_tree(&source, &destination).unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
        assert!(!destination.join("nested").exists());
        assert!(outside.join("secret.nettrace").exists());
    }

    #[test]
    fn raw_copy_rejects_source_root_symlinks_without_traversing_them() {
        let tmp = TempDir::new().unwrap();
        let real_source = tmp.path().join("real-source");
        let source = tmp.path().join("source-link");
        let destination = tmp.path().join("destination");
        std::fs::create_dir(&real_source).unwrap();
        std::fs::write(real_source.join("capture.data"), b"must not be copied").unwrap();
        std::os::unix::fs::symlink(&real_source, &source).unwrap();

        let error = copy_raw_tree(&source, &destination).unwrap_err();
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
        assert!(!destination.exists());
        assert!(real_source.join("capture.data").exists());
    }

    #[test]
    fn raw_copy_rejects_destination_directory_symlinks_without_writing_outside() {
        let tmp = TempDir::new().unwrap();
        let source = tmp.path().join("source");
        let destination = tmp.path().join("destination");
        let outside = tmp.path().join("outside");
        std::fs::create_dir_all(source.join("nested")).unwrap();
        std::fs::write(source.join("nested/capture.data"), b"must stay inside").unwrap();
        std::fs::create_dir(&destination).unwrap();
        std::fs::create_dir(&outside).unwrap();
        std::os::unix::fs::symlink(&outside, destination.join("nested")).unwrap();

        assert!(copy_raw_tree(&source, &destination).is_err());
        assert!(!outside.join("capture.data").exists());
    }

    /// End-to-end of the SessionDb capture path without spawning a real
    /// debugger: we drive the same SQL the daemon uses and assert the
    /// rows land + `has_captured_data` flips.
    #[test]
    fn capture_path_writes_rows() {
        let tmp = TempDir::new().unwrap();
        let db_path = tmp.path().join("s.db");
        let db = SessionDb::create(CreateOptions {
            kind: SessionKind::Debug,
            target: "./app",
            target_class: TargetClass::NativeCpu,
            cwd: tmp.path(),
            db_path: Some(&db_path),
            label: Some("t".into()),
            target_hash: Some("h".into()),
        })
        .unwrap();

        // Simulate `log_command`.
        db.conn()
            .execute(
                "INSERT INTO commands (session_id, input, output_head, output_bytes, ts, canonical_op)
                 VALUES ((SELECT id FROM sessions LIMIT 1), ?1, ?2, ?3, datetime('now'), NULL)",
                params!["continue", "stopped at main.c:42", 20_i64],
            )
            .unwrap();

        // Simulate `capture_hit_if_stopped`.
        db.conn()
            .execute(
                "INSERT INTO breakpoint_hits
                    (session_id, location_key, hit_seq, thread, ts, locals_json, stack_json)
                 VALUES ((SELECT id FROM sessions LIMIT 1), ?1, ?2, ?3, datetime('now'), ?4, ?5)",
                params![
                    "main.c:42",
                    1_i64,
                    "1",
                    r#"{"x":{"value":"42"}}"#,
                    r#"{"raw":"..."}"#
                ],
            )
            .unwrap();

        assert!(db.has_captured_data().unwrap());
        let (cmds, hits): (i64, i64) = db
            .conn()
            .query_row(
                "SELECT (SELECT COUNT(*) FROM commands),
                        (SELECT COUNT(*) FROM breakpoint_hits)",
                [],
                |r| Ok((r.get(0)?, r.get(1)?)),
            )
            .unwrap();
        assert_eq!(cmds, 1);
        assert_eq!(hits, 1);
    }

    #[cfg(unix)]
    #[test]
    fn dap_automatic_capture_persist_e2e() {
        let tmp = TempDir::new().unwrap();
        let session = Mutex::new(real_dap_session(&tmp));
        let backend = DelveProtoBackend;
        let log = LogHandle::new();
        let save_path = session_db::sessions_dir(tmp.path()).join("dap-e2e.db");
        session.lock().unwrap().save_to = Some(save_path.clone());

        let response = handle_command("continue", &backend, &session, "", &log);
        assert!(!response.starts_with("[error:"), "{response}");
        let (_, keep_session_tmp) = handle_quit(&backend, &session, &AtomicI32::new(0), None);
        assert!(!keep_session_tmp, "captured DAP data was not persisted");

        let saved = SessionDb::open(&save_path).unwrap();
        let (location, locals, stack): (String, Option<String>, Option<String>) = saved
            .conn()
            .query_row(
                "SELECT location_key, locals_json, stack_json
                 FROM breakpoint_hits LIMIT 1",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
        assert_eq!(location, "/virtual/e2e.go:7");
        assert_eq!(locals.as_deref(), Some(r#"{"answer":"42"}"#));
        assert!(stack.unwrap().contains("main.main"));
    }

    #[cfg(unix)]
    #[test]
    fn dap_post_exit_stack_locals_e2e() {
        let tmp = TempDir::new().unwrap();
        let session = Mutex::new(real_dap_session(&tmp));
        let backend = DelveProtoBackend;
        let log = LogHandle::new();
        let save_path = session_db::sessions_dir(tmp.path()).join("dap-post-mortem.db");
        session.lock().unwrap().save_to = Some(save_path.clone());

        let _ = handle_command("continue", &backend, &session, "", &log);
        for _ in 0..100 {
            if !session.lock().unwrap().proc.is_alive() {
                break;
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        let stack = handle_command("stack", &backend, &session, "", &log);
        let locals = handle_command("locals", &backend, &session, "", &log);
        assert!(stack.contains("[post-mortem]"), "{stack}");
        assert!(stack.contains("main.main"), "{stack}");
        assert!(locals.contains("[post-mortem]"), "{locals}");
        assert!(locals.contains("42"), "{locals}");

        let saved = SessionDb::open(&save_path).unwrap();
        let (stack, locals): (String, String) = saved
            .conn()
            .query_row(
                "SELECT stack_json, locals_json
                 FROM breakpoint_hits LIMIT 1",
                [],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert!(stack.contains("main.main"));
        assert!(locals.contains("42"));
    }
}