clusterflux-public/crates/disasmer-dap/src/main.rs

use std::collections::BTreeMap;
use std::io::{self, BufRead, BufReader, Write};
use std::net::TcpStream;
use std::process::{Child, Command, Stdio};

use anyhow::{anyhow, Result};
use disasmer_core::{DebugRuntimeState, Digest, ProcessId, TaskId};
use serde_json::{json, Value};

const MAIN_THREAD: i64 = 1;
const LINUX_THREAD: i64 = 2;
const WINDOWS_THREAD: i64 = 3;
const PACKAGE_THREAD: i64 = 4;

#[derive(Clone)]
struct DapWriter {
    seq: i64,
}

impl DapWriter {
    fn new() -> Self {
        Self { seq: 1 }
    }

    fn response(&mut self, request: &Value, success: bool, body: Value) -> io::Result<()> {
        let command = request
            .get("command")
            .and_then(Value::as_str)
            .unwrap_or("<unknown>");
        let request_seq = request.get("seq").and_then(Value::as_i64).unwrap_or(0);
        let seq = self.next_seq();
        self.write(json!({
            "seq": seq,
            "type": "response",
            "request_seq": request_seq,
            "success": success,
            "command": command,
            "body": body,
        }))
    }

    fn error_response(&mut self, request: &Value, message: impl Into<String>) -> io::Result<()> {
        let command = request
            .get("command")
            .and_then(Value::as_str)
            .unwrap_or("<unknown>");
        let request_seq = request.get("seq").and_then(Value::as_i64).unwrap_or(0);
        let seq = self.next_seq();
        self.write(json!({
            "seq": seq,
            "type": "response",
            "request_seq": request_seq,
            "success": false,
            "command": command,
            "message": message.into(),
        }))
    }

    fn event(&mut self, event: &str, body: Value) -> io::Result<()> {
        let seq = self.next_seq();
        self.write(json!({
            "seq": seq,
            "type": "event",
            "event": event,
            "body": body,
        }))
    }

    fn output(&mut self, category: &str, output: impl Into<String>) -> io::Result<()> {
        self.event(
            "output",
            json!({
                "category": category,
                "output": output.into(),
            }),
        )
    }

    fn next_seq(&mut self) -> i64 {
        let seq = self.seq;
        self.seq += 1;
        seq
    }

    fn write(&mut self, message: Value) -> io::Result<()> {
        let payload = serde_json::to_vec(&message)?;
        let mut out = io::stdout().lock();
        write!(out, "Content-Length: {}\r\n\r\n", payload.len())?;
        out.write_all(&payload)?;
        out.flush()
    }
}

#[derive(Clone)]
struct VirtualThread {
    id: i64,
    frame_id: i64,
    args_ref: i64,
    runtime_ref: i64,
    output_ref: i64,
    task: TaskId,
    name: String,
    line: i64,
    state: DebugRuntimeState,
    freeze_supported: bool,
    recent_output: Vec<String>,
}

#[derive(Clone)]
struct AdapterState {
    process: ProcessId,
    epoch: u64,
    entry: String,
    project: String,
    runtime_backend: RuntimeBackend,
    command_status: String,
    last_task_failed: bool,
    runtime_artifact_path: Option<String>,
    runtime_event_count: usize,
    breakpoints: Vec<i64>,
    threads: BTreeMap<i64, VirtualThread>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum RuntimeBackend {
    Simulated,
    LocalServices,
}

impl Default for AdapterState {
    fn default() -> Self {
        let entry = "build".to_owned();
        let project = ".".to_owned();
        Self {
            process: process_id(&project, &entry),
            epoch: 0,
            threads: launch_threads(&entry),
            entry,
            project,
            runtime_backend: RuntimeBackend::Simulated,
            command_status: "waiting for launch".to_owned(),
            last_task_failed: false,
            runtime_artifact_path: None,
            runtime_event_count: 0,
            breakpoints: Vec::new(),
        }
    }
}

impl AdapterState {
    fn configure_launch(
        &mut self,
        project: String,
        entry: String,
        runtime_backend: RuntimeBackend,
    ) {
        self.process = process_id(&project, &entry);
        self.entry = entry;
        self.project = project;
        self.runtime_backend = runtime_backend;
        self.command_status = "waiting for configurationDone".to_owned();
        self.last_task_failed = false;
        self.runtime_artifact_path = None;
        self.runtime_event_count = 0;
        self.epoch = 0;
        self.breakpoints.clear();
        self.threads = launch_threads(&self.entry);
    }

    fn apply_runtime_record(&mut self, record: RuntimeLaunchRecord) {
        let task_failed = record.status_code != Some(0);
        self.command_status = format!(
            "{} through local services with status {:?}",
            if task_failed { "failed" } else { "completed" },
            record.status_code
        );
        self.last_task_failed = task_failed;
        self.runtime_artifact_path = record.artifact_path;
        self.runtime_event_count = record.event_count;
        if let Some(thread) = self.threads.get_mut(&LINUX_THREAD) {
            thread.recent_output.push(format!(
                "local node {} task with {} stdout bytes and {} stderr bytes",
                if task_failed { "failed" } else { "completed" },
                record.stdout_bytes,
                record.stderr_bytes
            ));
            thread.recent_output.push(format!(
                "coordinator recorded {} task event(s)",
                record.event_count
            ));
            if task_failed {
                thread.state = DebugRuntimeState::Failed(format!(
                    "local services task exited with status {:?}",
                    record.status_code
                ));
            }
        }
    }
}

#[derive(Clone, Debug)]
struct RuntimeLaunchRecord {
    status_code: Option<i32>,
    stdout_bytes: u64,
    stderr_bytes: u64,
    artifact_path: Option<String>,
    event_count: usize,
}

fn launch_threads(entry: &str) -> BTreeMap<i64, VirtualThread> {
    [
        thread(MAIN_THREAD, "main", &format!("{entry} virtual process"), 12),
        thread(LINUX_THREAD, "compile-linux", "compile linux", 42),
        thread(WINDOWS_THREAD, "compile-windows", "compile windows", 52),
        thread(PACKAGE_THREAD, "package", "package artifacts", 64),
    ]
    .into_iter()
    .map(|thread| (thread.id, thread))
    .collect()
}

fn process_id(project: &str, entry: &str) -> ProcessId {
    let digest = Digest::from_parts([
        b"dap-process:v1".as_slice(),
        project.as_bytes(),
        entry.as_bytes(),
    ]);
    let suffix = digest
        .as_str()
        .trim_start_matches("sha256:")
        .chars()
        .take(12)
        .collect::<String>();
    ProcessId::new(format!("vp-{suffix}"))
}

fn thread(id: i64, task: &str, name: &str, line: i64) -> VirtualThread {
    VirtualThread {
        id,
        frame_id: 1000 + id,
        args_ref: 2000 + id,
        runtime_ref: 3000 + id,
        output_ref: 4000 + id,
        task: TaskId::from(task),
        name: name.to_owned(),
        line,
        state: DebugRuntimeState::Running,
        freeze_supported: true,
        recent_output: vec![format!("{name}: waiting")],
    }
}

fn main() -> Result<()> {
    let mut writer = DapWriter::new();
    let mut reader = BufReader::new(io::stdin());
    let mut state = AdapterState::default();

    while let Some(request) = read_message(&mut reader)? {
        let command = request
            .get("command")
            .and_then(Value::as_str)
            .unwrap_or_default();
        match command {
            "initialize" => writer.response(&request, true, initialize_capabilities())?,
            "launch" => {
                let args = request
                    .get("arguments")
                    .cloned()
                    .unwrap_or_else(|| json!({}));
                state.entry = args
                    .get("entry")
                    .and_then(Value::as_str)
                    .unwrap_or("build")
                    .to_owned();
                let project = args
                    .get("project")
                    .and_then(Value::as_str)
                    .unwrap_or(".")
                    .to_owned();
                let runtime_backend = match args.get("runtimeBackend").and_then(Value::as_str) {
                    Some("local-services") => RuntimeBackend::LocalServices,
                    _ => RuntimeBackend::Simulated,
                };
                state.configure_launch(project, state.entry.clone(), runtime_backend);
                writer.response(&request, true, json!({}))?;
                writer.event("initialized", json!({}))?;
            }
            "setBreakpoints" => {
                state.breakpoints = request
                    .get("arguments")
                    .and_then(|value| value.get("breakpoints"))
                    .and_then(Value::as_array)
                    .map(|items| {
                        items
                            .iter()
                            .filter_map(|item| item.get("line").and_then(Value::as_i64))
                            .collect::<Vec<_>>()
                    })
                    .unwrap_or_default();
                let response = state
                    .breakpoints
                    .iter()
                    .enumerate()
                    .map(|(index, line)| {
                        json!({
                            "id": index + 1,
                            "verified": true,
                            "line": line,
                            "message": "Mapped to Disasmer virtual source location",
                        })
                    })
                    .collect::<Vec<_>>();
                writer.response(&request, true, json!({ "breakpoints": response }))?;
            }
            "setExceptionBreakpoints" => {
                writer.response(&request, true, json!({ "breakpoints": [] }))?;
            }
            "configurationDone" => {
                if state.runtime_backend == RuntimeBackend::LocalServices {
                    match run_local_services_runtime(&state) {
                        Ok(record) => {
                            state.apply_runtime_record(record);
                        }
                        Err(err) => {
                            writer.error_response(
                                &request,
                                format!("local services runtime launch failed: {err:#}"),
                            )?;
                            continue;
                        }
                    }
                } else {
                    state.command_status = "running under simulated debug adapter state".to_owned();
                }
                writer.response(&request, true, json!({}))?;
                writer.output(
                    "console",
                    format!(
                        "Disasmer bundle refreshed for entry `{}`; starting virtual process {} with {:?} runtime\n",
                        state.entry, state.process, state.runtime_backend
                    ),
                )?;
                if !state.breakpoints.is_empty() {
                    let stopped_thread = stopped_thread_for_breakpoint(&state);
                    match freeze_all(&mut state, stopped_thread, None) {
                        Ok(()) => {
                            writer.event(
                                "stopped",
                                json!({
                                    "reason": "breakpoint",
                                    "description": "Disasmer Debug Epoch all-stop",
                                    "threadId": stopped_thread,
                                    "allThreadsStopped": true,
                                }),
                            )?;
                        }
                        Err(failure) => {
                            let message = failure.message();
                            writer.output("stderr", format!("{message}\n"))?;
                            writer.error_response(&request, message)?;
                        }
                    }
                } else {
                    writer.event("terminated", json!({}))?;
                }
            }
            "threads" => {
                let threads = state
                    .threads
                    .values()
                    .map(|thread| {
                        json!({
                            "id": thread.id,
                            "name": format!("{}/{}", state.process, thread.name),
                        })
                    })
                    .collect::<Vec<_>>();
                writer.response(&request, true, json!({ "threads": threads }))?;
            }
            "stackTrace" => {
                let thread = request_thread(&request, &state);
                writer.response(
                    &request,
                    true,
                    json!({
                        "stackFrames": [{
                            "id": thread.frame_id,
                            "name": format!("{}::run", thread.name),
                            "line": thread.line,
                            "column": 1,
                            "source": {
                                "name": "build.rs",
                                "path": "src/build.rs"
                            }
                        }],
                        "totalFrames": 1
                    }),
                )?;
            }
            "scopes" => {
                let frame_id = request
                    .get("arguments")
                    .and_then(|value| value.get("frameId"))
                    .and_then(Value::as_i64)
                    .unwrap_or(1000 + LINUX_THREAD);
                let thread = state
                    .threads
                    .values()
                    .find(|thread| thread.frame_id == frame_id)
                    .cloned()
                    .unwrap_or_else(|| state.threads[&LINUX_THREAD].clone());
                writer.response(
                    &request,
                    true,
                    json!({
                        "scopes": [
                            {
                                "name": "Task Args and Handles",
                                "variablesReference": thread.args_ref,
                                "expensive": false,
                                "presentationHint": "locals"
                            },
                            {
                                "name": "Disasmer Runtime",
                                "variablesReference": thread.runtime_ref,
                                "expensive": false
                            },
                            {
                                "name": "Recent Output",
                                "variablesReference": thread.output_ref,
                                "expensive": false
                            }
                        ]
                    }),
                )?;
            }
            "variables" => {
                let reference = request
                    .get("arguments")
                    .and_then(|value| value.get("variablesReference"))
                    .and_then(Value::as_i64)
                    .unwrap_or(0);
                writer.response(&request, true, variables_response(&state, reference))?;
            }
            "evaluate" => {
                let expression = request
                    .get("arguments")
                    .and_then(|value| value.get("expression"))
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                let result = match expression {
                    "virtual_process_id" => state.process.to_string(),
                    "debug_epoch" => state.epoch.to_string(),
                    "command_status" => state.command_status.clone(),
                    _ => "not available".to_owned(),
                };
                writer.response(
                    &request,
                    true,
                    json!({ "result": result, "variablesReference": 0 }),
                )?;
            }
            "pause" => {
                match freeze_all(
                    &mut state,
                    LINUX_THREAD,
                    simulated_freeze_failure_thread(&request),
                ) {
                    Ok(()) => {
                        writer.response(&request, true, json!({}))?;
                        writer.event(
                            "stopped",
                            json!({
                                "reason": "pause",
                                "description": "Disasmer Debug Epoch pause",
                                "threadId": LINUX_THREAD,
                                "allThreadsStopped": true,
                            }),
                        )?;
                    }
                    Err(failure) => {
                        let message = failure.message();
                        writer.output("stderr", format!("{message}\n"))?;
                        writer.error_response(&request, message)?;
                    }
                }
            }
            "continue" => {
                for thread in state.threads.values_mut() {
                    if thread.state == DebugRuntimeState::Frozen {
                        thread.state = DebugRuntimeState::Running;
                    }
                }
                let thread_id = request_thread_id(&request).unwrap_or(LINUX_THREAD);
                writer.response(&request, true, json!({ "allThreadsContinued": true }))?;
                writer.event(
                    "continued",
                    json!({
                        "threadId": thread_id,
                        "allThreadsContinued": true,
                    }),
                )?;
            }
            "restart" | "restartFrame" => {
                let thread_id = request_thread_id(&request).unwrap_or(LINUX_THREAD);
                if restart_requires_whole_process(&request) {
                    let message = format!(
                        "Incompatible source edit requires whole virtual-process restart for {}",
                        state.process
                    );
                    writer.output("stderr", format!("{message}\n"))?;
                    writer.error_response(&request, message)?;
                    continue;
                }
                let restarting_failed_task = state
                    .threads
                    .get(&thread_id)
                    .is_some_and(|thread| matches!(thread.state, DebugRuntimeState::Failed(_)));
                if let Some(thread) = state.threads.get_mut(&thread_id) {
                    thread.state = DebugRuntimeState::Running;
                    thread
                        .recent_output
                        .push("task restarted from VFS checkpoint".to_owned());
                }
                if restarting_failed_task {
                    state.last_task_failed = false;
                }
                state.command_status = format!(
                    "{} task restarted from compatible VFS checkpoint",
                    if restarting_failed_task {
                        "failed"
                    } else {
                        "selected"
                    }
                );
                writer.response(&request, true, json!({}))?;
                writer.output(
                    "console",
                    format!(
                        "Restarted {} task from compatible VFS checkpoint on thread {thread_id}\n",
                        if restarting_failed_task {
                            "failed"
                        } else {
                            "selected"
                        }
                    ),
                )?;
            }
            "disconnect" | "terminate" => {
                writer.response(&request, true, json!({}))?;
                writer.event("terminated", json!({}))?;
                break;
            }
            _ => writer.error_response(&request, format!("unsupported DAP command: {command}"))?,
        }
    }

    Ok(())
}

fn request_thread<'a>(request: &Value, state: &'a AdapterState) -> &'a VirtualThread {
    let thread_id = request_thread_id(request).unwrap_or(LINUX_THREAD);
    state
        .threads
        .get(&thread_id)
        .unwrap_or_else(|| &state.threads[&LINUX_THREAD])
}

fn request_thread_id(request: &Value) -> Option<i64> {
    request
        .get("arguments")
        .and_then(|value| value.get("threadId"))
        .and_then(Value::as_i64)
}

fn restart_requires_whole_process(request: &Value) -> bool {
    let Some(arguments) = request.get("arguments") else {
        return false;
    };

    arguments
        .get("requiresWholeProcessRestart")
        .and_then(Value::as_bool)
        .unwrap_or(false)
        || [
            "compatibility",
            "sourceCompatibility",
            "sourceEditCompatibility",
            "taskCompatibility",
        ]
        .iter()
        .filter_map(|field| arguments.get(field).and_then(Value::as_str))
        .any(is_incompatible_restart)
        || arguments
            .get("sourceEdit")
            .and_then(|value| value.get("compatibility"))
            .and_then(Value::as_str)
            .is_some_and(is_incompatible_restart)
}

fn is_incompatible_restart(value: &str) -> bool {
    value.eq_ignore_ascii_case("incompatible")
        || value.eq_ignore_ascii_case("whole-process")
        || value.eq_ignore_ascii_case("whole_process")
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct FreezeFailure {
    thread_id: i64,
    task: TaskId,
}

impl FreezeFailure {
    fn message(&self) -> String {
        format!(
            "debug all-stop failed: participant `{}` on thread {} could not freeze",
            self.task, self.thread_id
        )
    }
}

fn simulated_freeze_failure_thread(request: &Value) -> Option<i64> {
    request
        .get("arguments")
        .and_then(|arguments| arguments.get("simulateFreezeFailure"))
        .and_then(Value::as_bool)
        .unwrap_or(false)
        .then_some(PACKAGE_THREAD)
}

fn freeze_all(
    state: &mut AdapterState,
    stopped_thread: i64,
    forced_failure_thread: Option<i64>,
) -> Result<(), FreezeFailure> {
    if let Some(failure) = state.threads.values().find(|thread| {
        thread.state == DebugRuntimeState::Running
            && (!thread.freeze_supported || forced_failure_thread == Some(thread.id))
    }) {
        return Err(FreezeFailure {
            thread_id: failure.id,
            task: failure.task.clone(),
        });
    }

    state.epoch += 1;
    for thread in state.threads.values_mut() {
        if thread.state == DebugRuntimeState::Running {
            thread.state = DebugRuntimeState::Frozen;
        }
    }
    if let Some(thread) = state.threads.get_mut(&stopped_thread) {
        if let Some(line) = state.breakpoints.first().copied() {
            thread.line = line;
        }
        thread
            .recent_output
            .push(format!("debug epoch {} all-stop", state.epoch));
    }
    Ok(())
}

fn stopped_thread_for_breakpoint(state: &AdapterState) -> i64 {
    state
        .breakpoints
        .first()
        .and_then(|line| {
            state
                .threads
                .values()
                .find(|thread| thread.line == *line)
                .map(|thread| thread.id)
        })
        .unwrap_or(LINUX_THREAD)
}

fn variables_response(state: &AdapterState, reference: i64) -> Value {
    let Some(thread) = state.threads.values().find(|thread| {
        thread.args_ref == reference
            || thread.runtime_ref == reference
            || thread.output_ref == reference
    }) else {
        return json!({ "variables": [] });
    };

    if thread.args_ref == reference {
        return json!({
            "variables": [
                {
                    "name": "target",
                    "value": thread.task.to_string(),
                    "variablesReference": 0
                },
                {
                    "name": "artifact",
                    "value": state.runtime_artifact_path.clone().unwrap_or_else(|| {
                        format!("artifact://{}/{}/app.tar.zst", state.process, thread.task)
                    }),
                    "variablesReference": 0
                },
                {
                    "name": "source_snapshot",
                    "value": format!("source://local-checkout/{}", project_snapshot_suffix(&state.project)),
                    "variablesReference": 0
                }
            ]
        });
    }

    if thread.runtime_ref == reference {
        return json!({
            "variables": [
                {
                    "name": "virtual_process_id",
                    "value": state.process.to_string(),
                    "variablesReference": 0
                },
                {
                    "name": "virtual_thread",
                    "value": thread.name,
                    "variablesReference": 0
                },
                {
                    "name": "debug_epoch",
                    "value": state.epoch,
                    "variablesReference": 0
                },
                {
                    "name": "state",
                    "value": format!("{:?}", thread.state),
                    "variablesReference": 0
                },
                {
                    "name": "runtime_backend",
                    "value": format!("{:?}", state.runtime_backend),
                    "variablesReference": 0
                },
                {
                    "name": "coordinator_task_events",
                    "value": state.runtime_event_count,
                    "variablesReference": 0
                },
                {
                    "name": "command_status",
                    "value": state.command_status,
                    "variablesReference": 0
                }
            ]
        });
    }

    json!({
        "variables": thread
            .recent_output
            .iter()
            .enumerate()
            .map(|(index, output)| {
                json!({
                    "name": format!("line_{index}"),
                    "value": output,
                    "variablesReference": 0
                })
            })
            .collect::<Vec<_>>()
    })
}

fn run_local_services_runtime(state: &AdapterState) -> Result<RuntimeLaunchRecord> {
    let repo = std::env::current_dir()?;
    let mut coordinator = Command::new("cargo")
        .args([
            "run",
            "-q",
            "-p",
            "disasmer-coordinator",
            "--bin",
            "disasmer-coordinator",
            "--",
            "--listen",
            "127.0.0.1:0",
        ])
        .current_dir(&repo)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    let result = run_with_coordinator(state, &repo, &mut coordinator);
    let _ = coordinator.kill();
    let _ = coordinator.wait();
    result
}

fn run_with_coordinator(
    state: &AdapterState,
    repo: &std::path::Path,
    coordinator: &mut Child,
) -> Result<RuntimeLaunchRecord> {
    let stdout = coordinator
        .stdout
        .take()
        .ok_or_else(|| anyhow!("coordinator stdout was not captured"))?;
    let mut ready_line = String::new();
    BufReader::new(stdout).read_line(&mut ready_line)?;
    let ready: Value = serde_json::from_str(&ready_line)?;
    let listen = ready
        .get("listen")
        .and_then(Value::as_str)
        .ok_or_else(|| anyhow!("coordinator did not report a listen address"))?;

    let node_output = Command::new("cargo")
        .args([
            "run",
            "-q",
            "-p",
            "disasmer-node",
            "--bin",
            "disasmer-node",
            "--",
            "--coordinator",
            listen,
            "--tenant",
            "tenant",
            "--project-id",
            "project",
            "--node",
            "dap-node",
            "--process",
            state.process.as_str(),
            "--task",
            "compile-linux",
            "--project",
            &state.project,
            "--artifact",
            "/vfs/artifacts/dap-output.txt",
        ])
        .current_dir(repo)
        .output()?;
    if !node_output.status.success() {
        return Err(anyhow!(
            "node runtime failed: {}",
            String::from_utf8_lossy(&node_output.stderr)
        ));
    }

    let stdout = String::from_utf8_lossy(&node_output.stdout);
    let report_line = stdout
        .lines()
        .last()
        .ok_or_else(|| anyhow!("node runtime did not emit JSON"))?;
    let report: Value = serde_json::from_str(report_line)?;
    let events = coordinator_request(
        listen,
        json!({
            "type": "list_task_events",
            "tenant": "tenant",
            "project": "project",
            "actor_user": "dap",
            "process": state.process.as_str(),
        }),
    )?;
    let event_count = events
        .get("events")
        .and_then(Value::as_array)
        .map(Vec::len)
        .unwrap_or(0);

    Ok(RuntimeLaunchRecord {
        status_code: report
            .get("status_code")
            .and_then(Value::as_i64)
            .map(|value| value as i32),
        stdout_bytes: report
            .get("stdout_bytes")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        stderr_bytes: report
            .get("stderr_bytes")
            .and_then(Value::as_u64)
            .unwrap_or(0),
        artifact_path: report
            .get("staged_artifact")
            .and_then(|artifact| artifact.get("path"))
            .and_then(Value::as_str)
            .map(str::to_owned),
        event_count,
    })
}

fn coordinator_request(addr: &str, request: Value) -> Result<Value> {
    let mut stream = TcpStream::connect(addr)?;
    serde_json::to_writer(&mut stream, &request)?;
    stream.write_all(b"\n")?;
    stream.flush()?;

    let mut line = String::new();
    BufReader::new(stream).read_line(&mut line)?;
    Ok(serde_json::from_str(&line)?)
}

fn project_snapshot_suffix(project: &str) -> String {
    Digest::from_parts([b"source-snapshot:v1".as_slice(), project.as_bytes()])
        .as_str()
        .trim_start_matches("sha256:")
        .chars()
        .take(12)
        .collect()
}

fn initialize_capabilities() -> Value {
    json!({
        "supportsConfigurationDoneRequest": true,
        "supportsTerminateRequest": true,
        "supportsRestartRequest": true,
        "supportsRestartFrame": true,
        "supportsEvaluateForHovers": true,
        "supportsStepBack": false,
    })
}

fn read_message<R: BufRead>(reader: &mut R) -> Result<Option<Value>> {
    let mut content_length = None;

    loop {
        let mut line = String::new();
        let bytes = reader.read_line(&mut line)?;
        if bytes == 0 {
            return Ok(None);
        }

        let line = line.trim_end_matches(['\r', '\n']);
        if line.is_empty() {
            break;
        }

        if let Some(value) = line.strip_prefix("Content-Length:") {
            content_length = Some(value.trim().parse::<usize>()?);
        }
    }

    let length = content_length.ok_or_else(|| anyhow!("DAP message missing Content-Length"))?;
    let mut payload = vec![0; length];
    reader.read_exact(&mut payload)?;
    Ok(Some(serde_json::from_slice(&payload)?))
}

#[cfg(test)]
mod tests {
    use std::io::Cursor;

    use serde_json::json;

    use super::*;

    #[test]
    fn reads_standard_dap_content_length_frame_from_generic_client() {
        let payload = json!({
            "seq": 7,
            "type": "request",
            "command": "threads"
        })
        .to_string();
        let frame = format!("Content-Length: {}\r\n\r\n{payload}", payload.len());
        let mut reader = Cursor::new(frame.into_bytes());

        let message = read_message(&mut reader)
            .unwrap()
            .expect("DAP frame should contain a request");

        assert_eq!(message["seq"], 7);
        assert_eq!(message["type"], "request");
        assert_eq!(message["command"], "threads");
    }

    #[test]
    fn initialize_capabilities_use_standard_dap_flags() {
        let capabilities = initialize_capabilities();
        let capabilities = capabilities
            .as_object()
            .expect("initialize capabilities should be an object");

        assert_eq!(
            capabilities.get("supportsConfigurationDoneRequest"),
            Some(&json!(true))
        );
        assert_eq!(capabilities.get("supportsRestartFrame"), Some(&json!(true)));
        assert_eq!(capabilities.get("supportsStepBack"), Some(&json!(false)));
        assert!(capabilities
            .keys()
            .all(|key| key.starts_with("supports") && !key.contains("disasmer")));
    }

    #[test]
    fn compatible_restart_frame_does_not_require_whole_process_restart() {
        let request = json!({
            "command": "restartFrame",
            "arguments": {
                "frameId": 1,
                "sourceEdit": {
                    "compatibility": "compatible"
                }
            }
        });

        assert!(!restart_requires_whole_process(&request));
    }

    #[test]
    fn incompatible_source_edit_requires_whole_process_restart() {
        for arguments in [
            json!({ "sourceCompatibility": "incompatible" }),
            json!({ "sourceEdit": { "compatibility": "whole-process" } }),
            json!({ "requiresWholeProcessRestart": true }),
        ] {
            let request = json!({
                "command": "restartFrame",
                "arguments": arguments,
            });

            assert!(restart_requires_whole_process(&request));
        }
    }

    #[test]
    fn nonzero_local_runtime_record_marks_linux_task_failed() {
        let mut state = AdapterState::default();

        state.apply_runtime_record(RuntimeLaunchRecord {
            status_code: Some(1),
            stdout_bytes: 0,
            stderr_bytes: 12,
            artifact_path: None,
            event_count: 1,
        });

        let linux = state
            .threads
            .get(&LINUX_THREAD)
            .expect("linux task should exist");

        assert!(state.last_task_failed);
        assert!(state
            .command_status
            .contains("failed through local services"));
        assert!(matches!(linux.state, DebugRuntimeState::Failed(_)));
        assert!(linux
            .recent_output
            .iter()
            .any(|line| line.contains("local node failed task")));
    }

    #[test]
    fn freeze_all_reports_failure_without_claiming_all_stop() {
        let mut state = AdapterState::default();
        state
            .threads
            .get_mut(&PACKAGE_THREAD)
            .expect("package thread should exist")
            .freeze_supported = false;

        let error = freeze_all(&mut state, LINUX_THREAD, None).unwrap_err();

        assert_eq!(error.thread_id, PACKAGE_THREAD);
        assert_eq!(error.task, TaskId::from("package"));
        assert!(error.message().contains("could not freeze"));
        assert_eq!(state.epoch, 0);
        assert!(state
            .threads
            .values()
            .all(|thread| thread.state == DebugRuntimeState::Running));
    }

    #[test]
    fn freeze_all_success_freezes_every_running_participant() {
        let mut state = AdapterState::default();

        freeze_all(&mut state, LINUX_THREAD, None).unwrap();

        assert_eq!(state.epoch, 1);
        assert!(state
            .threads
            .values()
            .all(|thread| thread.state == DebugRuntimeState::Frozen));
    }

    #[test]
    fn breakpoint_line_selects_main_or_spawned_virtual_thread() {
        let mut state = AdapterState::default();

        state.breakpoints = vec![12];
        assert_eq!(stopped_thread_for_breakpoint(&state), MAIN_THREAD);

        state.breakpoints = vec![42];
        assert_eq!(stopped_thread_for_breakpoint(&state), LINUX_THREAD);
    }

    #[test]
    fn launch_threads_include_windows_task_in_same_virtual_process() {
        let state = AdapterState::default();
        let windows = state
            .threads
            .get(&WINDOWS_THREAD)
            .expect("windows task should be debugger-visible");

        assert_eq!(state.threads.len(), 4);
        assert_eq!(windows.id, WINDOWS_THREAD);
        assert_eq!(windows.task, TaskId::from("compile-windows"));
        assert_eq!(windows.name, "compile windows");
        assert_eq!(windows.line, 52);
        assert_eq!(process_id(&state.project, &state.entry), state.process);
    }

    #[test]
    fn windows_thread_runtime_variables_share_virtual_process() {
        let state = AdapterState::default();
        let windows = state
            .threads
            .get(&WINDOWS_THREAD)
            .expect("windows task should be debugger-visible");

        let variables = variables_response(&state, windows.runtime_ref);
        let variables = variables["variables"]
            .as_array()
            .expect("variables response should be an array");

        assert!(variables.iter().any(|variable| {
            variable["name"] == "virtual_process_id"
                && variable["value"] == state.process.to_string()
        }));
        assert!(variables.iter().any(|variable| {
            variable["name"] == "virtual_thread" && variable["value"] == "compile windows"
        }));
    }
}