zsh/extensions/

dap.rs

//! DAP server for zshrs — `zshrs --dap [HOST:PORT]` (TCP connect-back, or stdio
//! when no address is given).
//!
//! Mirrors strykelang's DAP architecture (`strykelang/dap.rs`):
//! connect TCP → spawn reader thread → wait for `launch` → run the
//! script IN-PROCESS so per-statement breakpoint checks have a live
//! interpreter to pause. The compiler emits `BUILTIN_SET_LINENO` at
//! every top-level statement; that builtin's handler in
//! `fusevm_bridge.rs` calls [`check_line`] which consults the
//! private `DAP_SHARED` static for matching breakpoints and
//! condvar-waits in [`DapShared::pause`].
//!
//! NOT subprocess-based — earlier v1 spawned the script as a child,
//! which made it impossible to honor breakpoints (no IPC channel to
//! pause the child). The new model keeps the script in-process so the
//! cv-wait literally blocks the executor thread until the IDE sends
//! `continue`.

use serde_json::{json, Value};
use std::collections::HashMap;
use std::io::{self, BufRead, BufReader, Read, Write};
use std::net::TcpStream;
use std::process::{Child, ChildStdout, Command, Stdio};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex, OnceLock};
use std::thread;
use std::time::Duration;

// ── Pause-and-resume shared state ───────────────────────────────────────

/// What the IDE asked the paused executor to do.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DebugAction {
    /// Resume normal execution.
    Continue,
    /// Step over the current statement (treat as Continue + pause at
    /// the next `check_line` call). v1 simplification — Continue is
    /// the same as StepOver in absence of frame depth tracking.
    StepOver,
    /// Step in — same simplification as StepOver in v1.
    StepIn,
    /// Step out — same simplification as Continue in v1.
    StepOut,
    /// Client disconnected — bail out of the script.
    Quit,
}

/// Snapshot captured when the executor pauses. Sent back to the IDE
/// as part of the `stopped` event body.
#[derive(Debug, Clone, Default)]
pub struct PauseSnapshot {
    pub reason: String, // "breakpoint" | "step" | "pause" | "entry"
    /// `file` field.
    pub file: String,
    /// `line` field.
    pub line: u32,
}

/// Breakpoint state shared between the reader thread (which updates
/// it on `setBreakpoints`) and the executor thread (which consults
/// it on every `check_line`).
#[derive(Debug, Default)]
pub struct BreakpointState {
    /// Absolute file path → set of 1-based line numbers.
    pub line_breakpoints: HashMap<String, Vec<u32>>,
}

struct DapSharedInner {
    pending_action: Option<DebugAction>,
    is_paused: bool,
    pause_request: bool, // client asked us to pause asap (via `pause`)
    step_mode: bool,     // true = pause at every check_line (StepOver/In)
}

/// Reader thread + executor thread coordinate through this. Owns the
/// TCP writer, the request seq counter, the pause-cv, and the
/// disconnected flag.
pub struct DapShared {
    /// `inner` field.
    inner: Mutex<DapSharedInner>,
    /// `cv` field.
    cv: Condvar,
    /// `seq` field.
    seq: AtomicU64,
    /// `writer` field — the DAP output sink (a TCP socket in connect-back mode,
    /// stdout in stdio mode).
    writer: Mutex<Box<dyn Write + Send>>,
    /// `configuration_done` field.
    pub configuration_done: AtomicBool,
    /// `disconnected` field.
    pub disconnected: AtomicBool,
    /// Set once the executor has been handed the launched program.
    /// `disconnect` / `terminate` consult it: before launch the reader
    /// thread simply exits and the main thread falls out of
    /// `launch_rx.recv()`, but afterwards the executor owns the thread
    /// and may be blocked in a `waitpid` on a long-running child, so
    /// the adapter needs the watchdog in `terminate_debuggee`.
    pub launched: AtomicBool,
    /// Absolute path of the launched program — set on `launch`, read
    /// by `check_line` to know which file's breakpoint set to check.
    pub program: Mutex<String>,
}

impl DapShared {
    fn new(writer: Box<dyn Write + Send>) -> Arc<Self> {
        Arc::new(Self {
            inner: Mutex::new(DapSharedInner {
                pending_action: None,
                is_paused: false,
                pause_request: false,
                step_mode: false,
            }),
            cv: Condvar::new(),
            seq: AtomicU64::new(1),
            writer: Mutex::new(writer),
            configuration_done: AtomicBool::new(false),
            disconnected: AtomicBool::new(false),
            launched: AtomicBool::new(false),
            program: Mutex::new(String::new()),
        })
    }

    /// Called by the executor thread when a `check_line` matches a
    /// breakpoint (or step mode is on). Captures the snapshot, emits
    /// a `stopped` event, then condvar-waits for the IDE to send
    /// `continue` / `next` / `stepIn` / `stepOut`.
    pub fn pause(&self, snap: PauseSnapshot) -> DebugAction {
        // Flush stdout/stderr so any `echo` / `print` output the user
        // produced before this breakpoint is visible in the IDE
        // Console BEFORE the suspend UI shows.
        let _ = io::Write::flush(&mut io::stdout());
        let _ = io::Write::flush(&mut io::stderr());
        tracing::info!(
            target: "zshrs::dap::pause",
            reason = %snap.reason,
            file = %snap.file,
            line = snap.line,
            "executor PAUSED (cv-wait)",
        );
        {
            let mut s = self.inner.lock().expect("dap lock");
            s.is_paused = true;
            s.pending_action = None;
            s.pause_request = false;
        }
        let _ = self.emit_event(
            "stopped",
            json!({
                "reason": snap.reason,
                "threadId": 1,
                "allThreadsStopped": true,
                "preserveFocusHint": false,
                "description": snap.reason,
                "text": format!("{}:{}", snap.file, snap.line),
            }),
        );
        let mut guard = self.inner.lock().expect("dap lock");
        while guard.pending_action.is_none() && !self.disconnected.load(Ordering::SeqCst) {
            guard = self.cv.wait(guard).expect("dap cv");
        }
        let action = guard.pending_action.take().unwrap_or(DebugAction::Continue);
        guard.is_paused = false;
        // Step mode persists until the IDE sends a plain `continue`.
        guard.step_mode = matches!(action, DebugAction::StepOver | DebugAction::StepIn);
        tracing::info!(
            target: "zshrs::dap::pause",
            ?action,
            step_mode = guard.step_mode,
            "executor RESUMED",
        );
        action
    }

    fn resume_with(&self, action: DebugAction) {
        let mut g = self.inner.lock().expect("dap lock");
        g.pending_action = Some(action);
        self.cv.notify_all();
    }

    fn request_pause(&self) {
        let mut g = self.inner.lock().expect("dap lock");
        g.pause_request = true;
    }

    fn want_pause(&self) -> bool {
        self.inner.lock().map(|g| g.pause_request).unwrap_or(false)
    }

    fn step_mode(&self) -> bool {
        self.inner.lock().map(|g| g.step_mode).unwrap_or(false)
    }

    fn next_seq(&self) -> u64 {
        self.seq.fetch_add(1, Ordering::SeqCst)
    }

    fn write_message(&self, body: Value) -> io::Result<()> {
        let s = serde_json::to_string(&body)?;
        let mut w = self.writer.lock().expect("dap writer");
        write!(w, "Content-Length: {}\r\n\r\n{}", s.len(), s)?;
        w.flush()
    }

    fn emit_response(
        &self,
        req_seq: i64,
        command: &str,
        success: bool,
        body: Value,
    ) -> io::Result<()> {
        let seq = self.next_seq();
        let msg = json!({
            "seq": seq,
            "type": "response",
            "request_seq": req_seq,
            "command": command,
            "success": success,
            "body": body,
        });
        tracing::trace!(target: "zshrs::dap::send", seq, %command, "response");
        self.write_message(msg)
    }
    /// `emit_event` — see implementation.
    pub fn emit_event(&self, event: &str, body: Value) -> io::Result<()> {
        let seq = self.next_seq();
        let milestone = matches!(
            event,
            "stopped" | "terminated" | "exited" | "initialized" | "process" | "breakpoint"
        );
        if milestone {
            tracing::info!(target: "zshrs::dap::send", seq, %event, "event (milestone)");
        } else {
            tracing::trace!(target: "zshrs::dap::send", seq, %event, "event");
        }
        let msg = json!({
            "seq": seq,
            "type": "event",
            "event": event,
            "body": body,
        });
        self.write_message(msg)
    }
}

/// Global handle to the live DAP server (set by `run_dap` before the
/// script starts). The `BUILTIN_SET_LINENO` handler in
/// `fusevm_bridge.rs` reads this on every statement; if unset
/// (normal shell mode) the lookup is a single atomic load and falls
/// through to no-op.
static DAP_SHARED: OnceLock<Arc<DapShared>> = OnceLock::new();

/// Per-thread breakpoint snapshot — read in the hot path of every
/// `check_line` call. Cloned from `BreakpointState` after each
/// `setBreakpoints` so the executor doesn't need to lock the global
/// state on every statement.
static DAP_BREAKPOINTS: OnceLock<Arc<Mutex<BreakpointState>>> = OnceLock::new();

/// Called by `BUILTIN_SET_LINENO` in `fusevm_bridge.rs` for every
/// top-level statement. O(1) when DAP is not active (single atomic
/// load). When active, checks the breakpoint set for the launched
/// program and pauses if the current line matches OR if step mode
/// is on OR if the client asked for a pause.
pub fn check_line(line: u32) {
    let Some(shared) = DAP_SHARED.get() else {
        return;
    };
    if shared.disconnected.load(Ordering::SeqCst) {
        return;
    }
    let program = shared.program.lock().map(|g| g.clone()).unwrap_or_default();
    let reason = if shared.want_pause() {
        "pause"
    } else if shared.step_mode() {
        "step"
    } else {
        // Breakpoint match check
        let bp_arc = match DAP_BREAKPOINTS.get() {
            Some(b) => b,
            None => return,
        };
        let bp = match bp_arc.lock() {
            Ok(g) => g,
            Err(_) => return,
        };
        let hit = bp
            .line_breakpoints
            .get(&program)
            .map(|lines| lines.contains(&line))
            .unwrap_or(false);
        if !hit {
            return;
        }
        "breakpoint"
    };
    let snap = PauseSnapshot {
        reason: reason.to_string(),
        file: program,
        line,
    };
    let action = shared.pause(snap);
    if matches!(action, DebugAction::Quit) {
        // Terminate the script. We don't have a clean unwind from
        // inside the VM here; signal via errflag so the executor
        // halts at the next safe point.
        crate::ported::utils::errflag
            .fetch_or(crate::ported::zsh_h::ERRFLAG_ERROR, Ordering::Relaxed);
    }
}

// ── Public entry point ──────────────────────────────────────────────────

/// Serve the DAP protocol until the client disconnects.
///
/// `addr` selects the transport (called from `bins/zshrs.rs` on `--dap`):
///   * `Some("127.0.0.1:55123")` — **TCP connect-back**: dial the IDE's DAP
///     listener (the path JetBrains uses; stdout stays free for the script).
///   * `None` — **stdio**: DAP over stdin/stdout, for clients that spawn the
///     adapter as an executable (e.g. VS Code). Unusable under IntelliJ, whose
///     OSProcessHandler reads stdout and would steal DAP bytes.
///
/// Stryke-mirror architecture:
///   1. Connect the transport (TCP socket or stdin/stdout).
///   2. Spawn reader thread to handle `initialize`, `setBreakpoints`,
///      `launch`, `continue`, etc. Each `launch` request goes through
///      a oneshot channel back to the main thread.
///   3. Block on the channel until `launch` arrives.
///   4. Install DAP_SHARED + DAP_BREAKPOINTS globals (the BUILTIN_SET_LINENO
///      hook in fusevm_bridge.rs consults them on every statement).
///   5. Redirect fd 1 / fd 2 into `output` events, then run the script
///      IN-PROCESS via `ShellExecutor::execute_script_file`.
///      The executor blocks inside `check_line → pause()` whenever a
///      breakpoint hits; the reader thread sends `continue` to resume.
///   6. After execution: emit `exited` + `terminated`, drop globals.
pub fn run_dap(addr: Option<&str>) -> i32 {
    tracing::info!(
        target: "zshrs::dap",
        pid = std::process::id(),
        mode = if addr.is_some() { "tcp" } else { "stdio" },
        "starting --dap",
    );

    // Own the process group before anything is forked, so every
    // descendant of the debugged script lands in it and `disconnect`
    // with `terminateDebuggee` can take them all down.
    #[cfg(unix)]
    claim_process_group();

    // Two transports. Both run the identical DAP server below — only the
    // reader/writer differ.
    let (reader, writer): (Box<dyn Read + Send>, Box<dyn Write + Send>) = match addr {
        // TCP connect-back: dial the IDE's DAP listener (the path JetBrains
        // uses). The protocol has its own socket, so the script's stdout /
        // stderr stay separate from it — they are captured at the fd level
        // during `launch` and re-emitted as `output` events (see
        // `install_output_capture`), which is what a DAP client reads.
        Some(addr) => {
            let stream = match TcpStream::connect(addr) {
                Ok(s) => s,
                Err(e) => {
                    tracing::error!(target: "zshrs::dap", %addr, %e, "tcp connect failed");
                    eprintln!("zshrs: --dap: connect {} failed: {}", addr, e);
                    return 1;
                }
            };
            if let Err(e) = stream.set_nodelay(true) {
                tracing::warn!(target: "zshrs::dap", %e, "TCP_NODELAY failed (non-fatal)");
            }
            let reader_stream = match stream.try_clone() {
                Ok(s) => s,
                Err(e) => {
                    tracing::error!(target: "zshrs::dap", %e, "tcp clone failed");
                    eprintln!("zshrs: --dap: clone socket: {}", e);
                    return 1;
                }
            };
            tracing::info!(target: "zshrs::dap", %addr, "tcp connected");
            (Box::new(reader_stream), Box::new(stream))
        }
        // Stdio mode: DAP traffic over stdin/stdout. For clients that spawn the
        // adapter as an executable and talk over its pipes (e.g. VS Code's
        // DebugAdapterExecutable). NOTE: unusable under IntelliJ, whose
        // OSProcessHandler reads stdout and would steal DAP bytes — use TCP there.
        None => {
            tracing::info!(target: "zshrs::dap", "stdio mode");
            // Take a PRIVATE dup of fd 1 for the protocol before
            // `install_output_capture` re-points fd 1 at a pipe. Without
            // it the launched script's own `echo` would land in the
            // middle of a `Content-Length` frame and desync the client
            // — and with it, that output correctly arrives as `output`
            // events instead.
            #[cfg(unix)]
            let writer: Box<dyn Write + Send> = {
                use std::os::unix::io::FromRawFd;
                // SAFETY: dup(2) on the process's own stdout; the
                // returned fd is owned solely by this File.
                let fd = unsafe { libc::dup(libc::STDOUT_FILENO) };
                if fd >= 0 {
                    Box::new(unsafe { std::fs::File::from_raw_fd(fd) })
                } else {
                    Box::new(io::stdout())
                }
            };
            #[cfg(not(unix))]
            let writer: Box<dyn Write + Send> = Box::new(io::stdout());
            (Box::new(io::stdin()), writer)
        }
    };

    let shared = DapShared::new(writer);
    let bp_state = Arc::new(Mutex::new(BreakpointState::default()));

    // Reader thread: parses DAP requests + dispatches. Sends a
    // LaunchParams down the channel when `launch` arrives.
    let (launch_tx, launch_rx) = std::sync::mpsc::channel::<LaunchParams>();
    let shared_reader = shared.clone();
    let bp_reader = bp_state.clone();
    let _reader = thread::spawn(move || {
        let mut br = BufReader::new(reader);
        loop {
            let msg = match read_message(&mut br) {
                Ok(Some(m)) => m,
                Ok(None) => {
                    tracing::info!(target: "zshrs::dap", "client disconnected (EOF)");
                    break;
                }
                Err(e) => {
                    tracing::error!(target: "zshrs::dap", %e, "read error");
                    break;
                }
            };
            handle_request(&shared_reader, &bp_reader, &launch_tx, msg);
            if shared_reader.disconnected.load(Ordering::SeqCst) {
                break;
            }
        }
        // Reader exiting — unblock any cv-wait so the executor can
        // see Quit and bail.
        shared_reader.disconnected.store(true, Ordering::SeqCst);
        shared_reader.resume_with(DebugAction::Quit);
    });

    // Block until the IDE sends `launch`.
    let lp = match launch_rx.recv() {
        Ok(p) => p,
        Err(_) => {
            tracing::warn!(target: "zshrs::dap", "no launch received before disconnect");
            return 1;
        }
    };
    tracing::info!(
        target: "zshrs::dap",
        program = %lp.program,
        cwd = ?lp.cwd,
        args = ?lp.args,
        stop_on_entry = lp.stop_on_entry,
        "launch received",
    );

    // Install global hooks for BUILTIN_SET_LINENO.
    *shared.program.lock().expect("program lock") = lp.program.clone();
    let _ = DAP_SHARED.set(shared.clone());
    let _ = DAP_BREAKPOINTS.set(bp_state.clone());
    if lp.stop_on_entry {
        // Force a pause at the first statement.
        shared.inner.lock().expect("dap lock").step_mode = true;
    }

    // Cosmetic events for IDE UI.
    let _ = shared.emit_event(
        "process",
        json!({
            "name": lp.program,
            "systemProcessId": std::process::id(),
            "isLocalProcess": true,
            "startMethod": "launch",
        }),
    );
    let _ = shared.emit_event("thread", json!({ "reason": "started", "threadId": 1 }));

    if let Some(cwd) = &lp.cwd {
        let _ = std::env::set_current_dir(cwd);
    }

    // Everything the script writes to fd 1 / fd 2 from here on becomes
    // a DAP `output` event (see `install_output_capture`). Installed
    // BEFORE the executor starts so the very first `echo` is captured.
    let capture = install_output_capture(&shared);

    // Run the script in-process. The BUILTIN_SET_LINENO hook will
    // block on `shared.pause()` whenever a breakpoint matches.
    tracing::info!(target: "zshrs::dap", "entering executor (in-process)");
    let mut exec = crate::vm_helper::ShellExecutor::new();
    // `launch.args` → the script's positional parameters. DAP spec:
    // "arguments: Command line arguments passed to the program". They
    // were parsed into `LaunchParams` but never handed to the executor,
    // so `$1` / `$@` were empty for every debugged script no matter
    // what the launch config said.
    exec.set_pparams(lp.args.clone());
    let exit_code = match exec.execute_script_file(&lp.program) {
        Ok(status) => status,
        Err(e) => {
            tracing::error!(target: "zshrs::dap", %e, "executor returned error");
            let _ = shared.emit_event(
                "output",
                json!({
                    "category": "stderr",
                    "output": format!("zshrs --dap: {}\n", e),
                }),
            );
            1
        }
    };
    // Restore fd 1 / fd 2 and drain the capture threads before the
    // `exited` / `terminated` events so the IDE has every line of the
    // program's output before it tears the session down.
    if let Some(c) = capture {
        c.restore();
    }
    tracing::info!(target: "zshrs::dap", exit_code, "executor exited");

    let _ = shared.emit_event("exited", json!({ "exitCode": exit_code }));
    let _ = shared.emit_event("terminated", json!({}));
    let _ = shared.emit_event("thread", json!({ "reason": "exited", "threadId": 1 }));
    // Brief grace period for the writer to drain before the process exits.
    thread::sleep(Duration::from_millis(50));
    exit_code
}

#[derive(Debug, Clone)]
struct LaunchParams {
    program: String,
    cwd: Option<String>,
    args: Vec<String>,
    stop_on_entry: bool,
}

/// Dispatch a single DAP request. `launch_tx` is used to hand the
/// program info back to the main thread; everything else (breakpoints,
/// continue, etc.) is handled here directly.
fn handle_request(
    shared: &Arc<DapShared>,
    bp_state: &Arc<Mutex<BreakpointState>>,
    launch_tx: &std::sync::mpsc::Sender<LaunchParams>,
    msg: Value,
) {
    let cmd = msg
        .get("command")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();
    let req_seq = msg.get("seq").and_then(|v| v.as_i64()).unwrap_or(0);
    let args = msg.get("arguments").cloned().unwrap_or(Value::Null);
    tracing::trace!(
        target: "zshrs::dap::recv",
        seq = req_seq,
        %cmd,
        "request",
    );

    match cmd.as_str() {
        "initialize" => {
            let _ = shared.emit_response(
                req_seq,
                &cmd,
                true,
                json!({
                    "supportsConfigurationDoneRequest": true,
                    "supportsEvaluateForHovers": true,
                    "supportsTerminateRequest": true,
                    "supportsStepBack": false,
                    "supportsSetVariable": false,
                    "supportsConditionalBreakpoints": false,
                    "supportsHitConditionalBreakpoints": false,
                    "supportsFunctionBreakpoints": false,
                    "supportsRestartFrame": false,
                    "supportsGotoTargetsRequest": false,
                    "supportsStepInTargetsRequest": false,
                    "supportsCompletionsRequest": false,
                    "supportsModulesRequest": false,
                    "supportsExceptionInfoRequest": false,
                }),
            );
            let _ = shared.emit_event("initialized", json!({}));
        }
        "setBreakpoints" => {
            let path = args["source"]["path"].as_str().unwrap_or("").to_string();
            // Canonicalize so `check_line` lookup hits regardless of
            // relative vs absolute path differences between IDE +
            // executor.
            let canon_path = std::fs::canonicalize(&path)
                .map(|p| p.to_string_lossy().into_owned())
                .unwrap_or_else(|_| path.clone());
            let mut lines: Vec<u32> = Vec::new();
            let mut verified = Vec::new();
            if let Some(arr) = args["breakpoints"].as_array() {
                for b in arr {
                    if let Some(l) = b["line"].as_u64() {
                        lines.push(l as u32);
                        verified.push(json!({ "verified": true, "line": l }));
                    }
                }
            }
            tracing::info!(
                target: "zshrs::dap::breakpoints",
                path = %canon_path,
                count = lines.len(),
                lines = ?lines,
                "registered",
            );
            if let Ok(mut bp) = bp_state.lock() {
                if !canon_path.is_empty() {
                    bp.line_breakpoints.insert(canon_path, lines);
                }
            }
            let _ = shared.emit_response(req_seq, &cmd, true, json!({ "breakpoints": verified }));
        }
        "setExceptionBreakpoints" => {
            let _ = shared.emit_response(req_seq, &cmd, true, json!({}));
        }
        "configurationDone" => {
            shared.configuration_done.store(true, Ordering::SeqCst);
            let _ = shared.emit_response(req_seq, &cmd, true, json!({}));
        }
        "launch" => {
            let program_raw = args["program"].as_str().unwrap_or("").to_string();
            // Canonicalize so it matches the canonicalized path the
            // setBreakpoints handler stored.
            let program = std::fs::canonicalize(&program_raw)
                .map(|p| p.to_string_lossy().into_owned())
                .unwrap_or(program_raw);
            let cwd = args["cwd"].as_str().map(|s| s.to_string());
            let lp_args: Vec<String> = args["args"]
                .as_array()
                .map(|a| {
                    a.iter()
                        .filter_map(|v| v.as_str().map(String::from))
                        .collect()
                })
                .unwrap_or_default();
            let stop_on_entry = args["stopOnEntry"].as_bool().unwrap_or(false);
            // Mark launched HERE, not on the executor thread: from the
            // moment this send happens the main thread is committed to
            // running the script and will never come back to
            // `launch_rx.recv()`, so a `disconnect` racing the executor
            // startup still needs the watchdog. Setting the flag next to
            // `execute_script_file` left a window (ShellExecutor::new,
            // fd capture install) in which disconnect saw `false` and
            // spawned nothing.
            shared.launched.store(true, Ordering::SeqCst);
            // The program path is recorded before the response goes out: a client
            // may send `stackTrace` the moment it sees the reply, and the main
            // thread has not necessarily taken the launch parameters yet, so
            // that request read an empty path.
            if let Ok(mut p) = shared.program.lock() {
                *p = program.clone();
            }
            let _ = shared.emit_response(req_seq, &cmd, true, json!({}));
            let _ = launch_tx.send(LaunchParams {
                program,
                cwd,
                args: lp_args,
                stop_on_entry,
            });
        }
        "threads" => {
            let _ = shared.emit_response(
                req_seq,
                &cmd,
                true,
                json!({ "threads": [{ "id": 1, "name": "main" }] }),
            );
        }
        "stackTrace" => {
            // v1: synthesize a single frame from the launched program +
            // current $LINENO. Multi-frame call-stack walk-back is
            // future work (needs funcstack reflection).
            let program = shared.program.lock().map(|g| g.clone()).unwrap_or_default();
            let line = current_lineno();
            let frames = vec![json!({
                "id": 1,
                "name": "main",
                "source": { "path": program, "name": file_name(&program) },
                "line": line,
                "column": 1,
            })];
            let _ = shared.emit_response(
                req_seq,
                &cmd,
                true,
                json!({ "stackFrames": frames, "totalFrames": 1 }),
            );
        }
        "scopes" => {
            // Three separate scopes so IntelliJ renders them as
            // collapsible groups in this fixed order — `Locals` (user
            // vars) first, then `Specials` (zsh PM_SPECIAL params),
            // then `Environment` (caps-name env vars). One big scope
            // gets alpha-sorted client-side, burying the user's vars
            // under 300 env entries. The IDE's sort toggle still works
            // WITHIN each scope but the groups themselves are stable.
            let _ = shared.emit_response(
                req_seq,
                &cmd,
                true,
                json!({
                    "scopes": [
                        { "name": "Locals",      "variablesReference": 1, "expensive": false, "presentationHint": "locals" },
                        { "name": "Specials",    "variablesReference": 2, "expensive": false },
                        { "name": "Environment", "variablesReference": 3, "expensive": false },
                    ],
                }),
            );
        }
        "variables" => {
            let r = args["variablesReference"].as_u64().unwrap_or(1);
            let vars = match r {
                1 => snapshot_user_vars(),
                2 => snapshot_special_vars(),
                3 => snapshot_env_vars(),
                _ => snapshot_user_vars(),
            };
            let _ = shared.emit_response(req_seq, &cmd, true, json!({ "variables": vars }));
        }
        "evaluate" => {
            let expr = args["expression"].as_str().unwrap_or("");
            let (result, ty) = evaluate_expression(expr);
            let _ = shared.emit_response(
                req_seq,
                &cmd,
                true,
                json!({
                    "result": result,
                    "type": ty,
                    "variablesReference": 0,
                }),
            );
        }
        "continue" => {
            let _ =
                shared.emit_response(req_seq, &cmd, true, json!({ "allThreadsContinued": true }));
            shared.resume_with(DebugAction::Continue);
        }
        "next" => {
            let _ =
                shared.emit_response(req_seq, &cmd, true, json!({ "allThreadsContinued": true }));
            shared.resume_with(DebugAction::StepOver);
        }
        "stepIn" => {
            let _ =
                shared.emit_response(req_seq, &cmd, true, json!({ "allThreadsContinued": true }));
            shared.resume_with(DebugAction::StepIn);
        }
        "stepOut" => {
            let _ =
                shared.emit_response(req_seq, &cmd, true, json!({ "allThreadsContinued": true }));
            shared.resume_with(DebugAction::StepOut);
        }
        "pause" => {
            let _ = shared.emit_response(req_seq, &cmd, true, json!({}));
            shared.request_pause();
        }
        "disconnect" | "terminate" => {
            let _ = shared.emit_response(req_seq, &cmd, true, json!({}));
            shared.disconnected.store(true, Ordering::SeqCst);
            shared.resume_with(DebugAction::Quit);
            // Before `launch`, that is enough: the reader loop breaks on
            // `disconnected`, drops `launch_tx`, and the main thread
            // falls out of `launch_rx.recv()` and returns.
            //
            // After `launch` the script owns the main thread. Setting
            // `errflag` only halts it at the next VM safe point, and the
            // executor may be parked in `waitpid` on an external command
            // (`sleep 30`) that will not return for a long time — so the
            // adapter would sit there long after the IDE let go, holding
            // the debuggee alive with it. DAP requires the adapter to
            // shut down promptly on `disconnect`, and with
            // `terminateDebuggee` (the default for `launch` sessions) to
            // take the debuggee with it. Hand that to a watchdog.
            if shared.launched.load(Ordering::SeqCst) {
                let terminate_debuggee = args["terminateDebuggee"].as_bool().unwrap_or(true);
                thread::spawn(move || terminate_debuggee_watchdog(terminate_debuggee));
            }
        }
        "source" => {
            let _ = shared.emit_response(req_seq, &cmd, true, json!({}));
        }
        _ => {
            tracing::debug!(target: "zshrs::dap", %cmd, "unsupported request");
            let _ = shared.emit_response(req_seq, &cmd, false, json!({ "error": "unsupported" }));
        }
    }
}

/// Grace period between `disconnect` and the hard exit. Long enough for
/// an executor parked at a `check_line` cv-wait to observe `Quit` and
/// unwind on its own (the tidy path, which still runs `always` blocks
/// and EXIT traps), short enough that an IDE never waits on us.
const DISCONNECT_GRACE: Duration = Duration::from_millis(300);

/// Watchdog spawned by `disconnect` / `terminate` once the script is
/// running. Waits [`DISCONNECT_GRACE`] for the executor to unwind by
/// itself; if the process is still here after that (the executor may be
/// blocked in a `waitpid` on a child, or anywhere else the `errflag`
/// signal cannot reach), takes the debuggee down and exits the adapter.
///
/// The takedown is two-part because the two kinds of child live in
/// different places: [`kill_live_jobs`] SIGKILLs the pids the shell's
/// job table knows about, and [`terminate_process_group`] SIGTERMs the
/// process group for the foreground children that never reach the job
/// table.
///
/// `terminate_debuggee` mirrors the request's `terminateDebuggee`
/// field: false leaves the children running (DAP "detach" semantics)
/// but still shuts the adapter down, because a disconnected adapter
/// has nobody to talk to either way.
#[cfg(unix)]
fn terminate_debuggee_watchdog(terminate_debuggee: bool) {
    thread::sleep(DISCONNECT_GRACE);
    if terminate_debuggee {
        kill_live_jobs();
        terminate_process_group();
    }
    tracing::info!(
        target: "zshrs::dap",
        terminate_debuggee,
        "disconnect watchdog: grace period elapsed, exiting adapter",
    );
    let _ = io::Write::flush(&mut io::stdout());
    let _ = io::Write::flush(&mut io::stderr());
    std::process::exit(0);
}

#[cfg(not(unix))]
fn terminate_debuggee_watchdog(_terminate_debuggee: bool) {
    thread::sleep(DISCONNECT_GRACE);
    std::process::exit(0);
}

/// True when [`claim_process_group`] succeeded, i.e. every process the
/// debuggee forks is in a process group this adapter leads and nothing
/// else is.
static OWN_PROCESS_GROUP: AtomicBool = AtomicBool::new(false);

/// Put the adapter (and therefore everything the debugged script forks)
/// into its own process group, so `terminateDebuggee` can take the whole
/// tree down with one `killpg`.
///
/// Needed because the shell's job table only records pids for jobs that
/// went through `addproc` — a foreground external command in a
/// non-interactive script does not, so `kill_live_jobs` alone cannot see
/// a `sleep 30` the script is sitting in, and the debuggee would outlive
/// the session.
///
/// Skipped when stdin is a terminal: moving out of the terminal's
/// foreground process group would make any read from it raise SIGTTIN.
/// Under an IDE (and in the tests) stdin is a pipe or /dev/null, so the
/// group is claimed; a human running `zshrs --dap` by hand at a terminal
/// keeps the old behaviour and falls back to the job-table walk.
#[cfg(unix)]
fn claim_process_group() {
    use std::io::IsTerminal;
    if io::stdin().is_terminal() {
        tracing::debug!(target: "zshrs::dap", "stdin is a tty; not claiming a process group");
        return;
    }
    // SAFETY: setpgid(0, 0) on ourselves; fails only with EPERM when we
    // are already a session leader, which is reported, not fatal.
    if unsafe { libc::setpgid(0, 0) } == 0 {
        OWN_PROCESS_GROUP.store(true, Ordering::SeqCst);
        tracing::info!(target: "zshrs::dap", pgid = std::process::id(), "own process group claimed");
    } else {
        tracing::warn!(
            target: "zshrs::dap",
            err = %io::Error::last_os_error(),
            "setpgid failed; terminateDebuggee limited to job-table pids",
        );
    }
}

/// SIGTERM the debuggee's process group.
///
/// Only runs when [`claim_process_group`] took ownership, so the group
/// contains exactly this adapter and the processes the debugged script
/// forked — never the IDE or the shell that launched us.
///
/// SIGTERM (not SIGKILL) because the signal necessarily reaches this
/// process too and SIGKILL cannot be ignored: the adapter ignores
/// SIGTERM for the instant it takes to deliver, and children — which
/// inherited the DEFAULT disposition when they were forked, before this
/// ignore was installed — die from it.
#[cfg(unix)]
fn terminate_process_group() {
    if !OWN_PROCESS_GROUP.load(Ordering::SeqCst) {
        return;
    }
    // SAFETY: signal(2) + killpg(2) on our own process group.
    unsafe {
        libc::signal(libc::SIGTERM, libc::SIG_IGN);
        let pgid = libc::getpgrp();
        tracing::info!(target: "zshrs::dap", pgid, "SIGTERM debuggee process group");
        libc::killpg(pgid, libc::SIGTERM);
    }
}

/// SIGKILL every process the shell's job table still lists as ours.
///
/// This is the DAP `terminateDebuggee` action for the in-process model:
/// the "debuggee" is this process plus whatever external commands the
/// script has forked, and the background ones among those are the pids
/// recorded in `ported::jobs::JOBTAB` (`Src/jobs.c` `addproc`). A job
/// already marked DONE is skipped, and our own pid is never signalled.
/// Foreground children never reach `addproc`, which is what
/// [`terminate_process_group`] covers.
///
/// SIGKILL rather than the SIGHUP that `signals::killrunjobs` sends:
/// that walk is gated on the HUP option and on STAT_LOCKED background
/// jobs, neither of which holds for the foreground child of a
/// non-interactive debug session.
#[cfg(unix)]
fn kill_live_jobs() {
    let my_pid = unsafe { libc::getpid() };
    let Some(tab) = crate::ported::jobs::JOBTAB.get() else {
        return;
    };
    let Ok(tab) = tab.lock() else {
        return;
    };
    for job in tab.iter() {
        if (job.stat & crate::ported::jobs::stat::DONE) != 0 {
            continue;
        }
        for proc_ in job.procs.iter().chain(job.auxprocs.iter()) {
            if proc_.pid > 0 && proc_.pid != my_pid {
                tracing::info!(target: "zshrs::dap", pid = proc_.pid, "SIGKILL debuggee child");
                unsafe { libc::kill(proc_.pid, libc::SIGKILL) };
            }
        }
    }
}

/// Redirect fd 1 / fd 2 into pipes and re-emit everything written to
/// them as DAP `output` events (`category: "stdout"` / `"stderr"`).
///
/// The adapter runs the script IN-PROCESS, so "the program's output" is
/// literally this process's stdout/stderr plus that of any command it
/// forks — and a DAP client has no other way to see it. Per the DAP
/// spec, `output` events are how a debug adapter reports program output
/// unless `runInTerminal` is used, and a client attached over TCP (VS
/// Code, or anything that is not also reading our pipes) shows nothing
/// at all without them.
///
/// fd-level (`dup2`) rather than capturing Rust's `io::stdout` so that
/// forked children — external commands the script runs — are captured
/// too: they inherit fd 1 / fd 2, not a Rust handle.
///
/// Returns `None` if the pipes cannot be created, in which case output
/// keeps going straight to the adapter's own stdout/stderr (the old
/// behaviour) rather than being lost.
#[cfg(unix)]
fn install_output_capture(shared: &Arc<DapShared>) -> Option<OutputCapture> {
    use std::os::unix::io::FromRawFd;

    fn make_pipe() -> Option<(i32, i32)> {
        let mut fds = [0i32; 2];
        // SAFETY: `fds` is a valid 2-element array, the only contract pipe(2) has.
        if unsafe { libc::pipe(fds.as_mut_ptr()) } != 0 {
            return None;
        }
        Some((fds[0], fds[1]))
    }

    let (out_r, out_w) = make_pipe()?;
    let (err_r, err_w) = make_pipe()?;

    // SAFETY: plain fd bookkeeping on fds we own.
    let (saved_out, saved_err) = unsafe {
        let saved_out = libc::dup(libc::STDOUT_FILENO);
        let saved_err = libc::dup(libc::STDERR_FILENO);
        libc::dup2(out_w, libc::STDOUT_FILENO);
        libc::dup2(err_w, libc::STDERR_FILENO);
        // fd 1 / fd 2 are now the only references to the write ends;
        // restore() closing them is what gives the readers their EOF.
        libc::close(out_w);
        libc::close(err_w);
        (saved_out, saved_err)
    };

    let (done_tx, done_rx) = std::sync::mpsc::channel::<()>();
    let mut pump = |fd: i32, category: &'static str| {
        let shared = Arc::clone(shared);
        let done = done_tx.clone();
        thread::spawn(move || {
            // SAFETY: `fd` is a pipe read end this function just created
            // and hands to exactly one thread.
            let mut file = unsafe { std::fs::File::from_raw_fd(fd) };
            let mut buf = [0u8; 8192];
            loop {
                match file.read(&mut buf) {
                    Ok(0) | Err(_) => break,
                    Ok(n) => {
                        let text = String::from_utf8_lossy(&buf[..n]).into_owned();
                        // `write_message`, NOT `emit_event`: emit_event
                        // logs through `tracing`, and if a subscriber is
                        // ever pointed at stderr that log line lands back
                        // in the pipe this thread is draining — one
                        // chunk in, one log line out, forever. Framing
                        // the event by hand keeps the pump free of any
                        // path that can write to fd 1 / fd 2.
                        let _ = shared.write_message(json!({
                            "seq": shared.next_seq(),
                            "type": "event",
                            "event": "output",
                            "body": { "category": category, "output": text },
                        }));
                    }
                }
            }
            let _ = done.send(());
        })
    };
    pump(out_r, "stdout");
    pump(err_r, "stderr");
    drop(done_tx);

    Some(OutputCapture {
        saved_out,
        saved_err,
        done_rx,
    })
}

#[cfg(not(unix))]
fn install_output_capture(_shared: &Arc<DapShared>) -> Option<OutputCapture> {
    None
}

/// Live fd-redirection installed by [`install_output_capture`].
struct OutputCapture {
    /// dup of the original fd 1, restored by [`OutputCapture::restore`].
    saved_out: i32,
    /// dup of the original fd 2.
    saved_err: i32,
    /// Both pump threads send on this when they hit EOF.
    done_rx: std::sync::mpsc::Receiver<()>,
}

impl OutputCapture {
    /// Put fd 1 / fd 2 back and wait for the pump threads to drain.
    ///
    /// Restoring closes the last write-end reference, which is what ends
    /// the pumps — unless a background child of the script still holds
    /// an inherited copy, so the drain wait is bounded and the threads
    /// are left detached if it expires.
    #[cfg(unix)]
    fn restore(self) {
        let _ = io::Write::flush(&mut io::stdout());
        let _ = io::Write::flush(&mut io::stderr());
        // SAFETY: both saved fds came from `dup` in install_output_capture.
        unsafe {
            libc::dup2(self.saved_out, libc::STDOUT_FILENO);
            libc::dup2(self.saved_err, libc::STDERR_FILENO);
            libc::close(self.saved_out);
            libc::close(self.saved_err);
        }
        for _ in 0..2 {
            if self
                .done_rx
                .recv_timeout(Duration::from_millis(250))
                .is_err()
            {
                break;
            }
        }
    }

    #[cfg(not(unix))]
    fn restore(self) {}
}

/// Read the current `$LINENO` via paramtab — same path BUILTIN_SET_LINENO
/// writes to. Default to 1 when unset.
fn current_lineno() -> u32 {
    crate::ported::params::paramtab()
        .read()
        .ok()
        .and_then(|t| t.get("LINENO").map(|pm| pm.u_val as u32))
        .unwrap_or(1)
}

/// Buckets all paramtab entries into (user, specials, env) by zsh
/// `PM_SPECIAL` flag + caps-name + process-env presence. Returned
/// tuples are sorted alpha within each bucket. Used by the three
/// `snapshot_*_vars` helpers so each scope returns just its bucket.
///
/// Splitting into separate scopes (instead of one big list) means
/// IntelliJ renders them as collapsible groups in fixed order, even
/// when the user has "Sort Values Alphabetically" enabled — that
/// toggle only sorts WITHIN a scope, not across scopes.
fn snapshot_bucketed() -> (
    Vec<(String, String)>,
    Vec<(String, String)>,
    Vec<(String, String)>,
) {
    let mut user: Vec<(String, String)> = Vec::new();
    let mut specials: Vec<(String, String)> = Vec::new();
    let mut env: Vec<(String, String)> = Vec::new();

    if let Ok(tab) = crate::ported::params::paramtab().read() {
        for (name, pm) in tab.iter() {
            if matches!(
                name.as_str(),
                "_" | "PIPESTATUS" | "pipestatus" | "ZSH_ARGZERO"
            ) {
                continue;
            }
            let value = if pm.u_val != 0
                && (pm.node.flags & crate::ported::zsh_h::PM_INTEGER as i32) != 0
            {
                pm.u_val.to_string()
            } else if let Some(s) = pm.u_str.as_ref() {
                s.clone()
            } else {
                String::new()
            };
            let is_special = (pm.node.flags & crate::ported::zsh_h::PM_SPECIAL as i32) != 0;
            // Environment FIRST — most users think of PATH/HOME/USER
            // as env vars even though zsh marks them PM_SPECIAL. The
            // process-env presence is what makes them env, not the
            // zsh flag.
            let in_process_env = std::env::var(name).is_ok();
            // Caps-only names not in env → zsh internal specials
            // bucket (CPUTYPE, MACHTYPE, OSTYPE, HOST, etc).
            let is_caps_only = !name.is_empty()
                && name
                    .chars()
                    .all(|c| c.is_ascii_uppercase() || c == '_' || c.is_ascii_digit())
                && name
                    .chars()
                    .next()
                    .map(|c| c.is_ascii_uppercase())
                    .unwrap_or(false);
            if in_process_env {
                env.push((name.clone(), value));
            } else if is_special || is_caps_only {
                specials.push((name.clone(), value));
            } else {
                user.push((name.clone(), value));
            }
        }
    }
    if user.is_empty() && specials.is_empty() && env.is_empty() {
        for (k, v) in std::env::vars() {
            env.push((k, v));
        }
    }
    user.sort_by(|a, b| a.0.cmp(&b.0));
    specials.sort_by(|a, b| a.0.cmp(&b.0));
    env.sort_by(|a, b| a.0.cmp(&b.0));
    (user, specials, env)
}

fn vars_to_json(vars: &[(String, String)]) -> Vec<Value> {
    vars.iter()
        .take(500)
        .map(|(n, v)| {
            json!({
                "name": n,
                "value": v,
                "type": "scalar",
                "variablesReference": 0,
            })
        })
        .collect()
}

fn snapshot_user_vars() -> Vec<Value> {
    let (user, _, _) = snapshot_bucketed();
    vars_to_json(&user)
}

fn snapshot_special_vars() -> Vec<Value> {
    let (_, specials, _) = snapshot_bucketed();
    vars_to_json(&specials)
}

fn snapshot_env_vars() -> Vec<Value> {
    let (_, _, env) = snapshot_bucketed();
    vars_to_json(&env)
}

fn evaluate_expression(expr: &str) -> (String, String) {
    // v1: run a fresh subshell to evaluate. Doesn't see the paused
    // executor's local scope, but works for `$VAR` / `$(cmd)` / `$((expr))`.
    let exe = std::env::current_exe().unwrap_or_else(|_| "zshrs".into());
    let mut cmd = Command::new(exe);
    cmd.arg("-c").arg(expr);
    cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
    match cmd.output() {
        Ok(o) => {
            if o.status.success() {
                let s = String::from_utf8_lossy(&o.stdout).trim_end().to_string();
                (s, "scalar".into())
            } else {
                let s = String::from_utf8_lossy(&o.stderr).trim_end().to_string();
                (s, "error".into())
            }
        }
        Err(e) => (format!("evaluate: {}", e), "error".into()),
    }
}
// ── Framing ──────────────────────────────────────────────────────────────

fn read_message<R: BufRead>(reader: &mut R) -> io::Result<Option<Value>> {
    let mut content_length: Option<usize> = None;
    loop {
        let mut line = String::new();
        let n = reader.read_line(&mut line)?;
        if n == 0 {
            return Ok(None);
        }
        if line == "\r\n" || line == "\n" {
            break;
        }
        if let Some(rest) = line.strip_prefix("Content-Length:") {
            content_length =
                Some(rest.trim().parse().map_err(|_| {
                    io::Error::new(io::ErrorKind::InvalidData, "bad Content-Length")
                })?);
        }
    }
    let len = content_length
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing Content-Length"))?;
    let mut buf = vec![0u8; len];
    reader.read_exact(&mut buf)?;
    let v: Value =
        serde_json::from_slice(&buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
    Ok(Some(v))
}

fn write_message<W: Write>(mut writer: W, msg: &Value) -> io::Result<()> {
    let body = serde_json::to_vec(msg)?;
    write!(writer, "Content-Length: {}\r\n\r\n", body.len())?;
    writer.write_all(&body)?;
    writer.flush()
}

fn file_name(path: &str) -> String {
    path.rsplit_once('/')
        .map(|x| x.1)
        .unwrap_or(path)
        .to_string()
}